feat(sql): add executeUpdate and executeTemporaryUpdate to DiskExecutor
This commit is contained in:
dualfroz committed
1 parent
441b6073ee
commit
aee0fc649f
1 file changed
+159
@@ -3831,3 +3831,162 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess {
|
||||
table.rangeBySecondary(txn, plan.indexName, plan.prefix, plan.lower, plan.lowerInclusive, plan.upper, plan.upperInclusive)
|
||||
}
|
||||
|
||||
private fun executeUpdate(statement: UpdateStatement, triggerContext: TriggerRowContext? = null): StatementResult {
|
||||
catalog.temporaryTableOrNull(statement.tableName)?.let { return executeTemporaryUpdate(statement, it) }
|
||||
val table = catalog.table(statement.tableName)
|
||||
val schema = table.schema
|
||||
val assignments = buildAssignments(schema, statement)
|
||||
return withStatementTransaction { txn ->
|
||||
withTriggerAtomicity(schema.name, txn, triggerContext) {
|
||||
executeUpdateInTxn(statement, assignments, table, schema, txn, triggerContext)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* UPDATE on a temp table's plain [Table.rows] list - ported from the same whole-batch-then-swap
|
||||
* shape as [executeUpdateInTxn]/[kestrel.engine.Executor.executeUpdateInner]: every row is
|
||||
* planned into [newRows] first (evaluating SET/DEFAULT/GENERATED and validating NOT NULL) before
|
||||
* [table.rows] is ever mutated, so a CHECK/UNIQUE/PRIMARY KEY failure on any one row leaves the
|
||||
* table completely untouched rather than half-updated - the in-memory list equivalent of
|
||||
* [DiskTable.update]'s own per-row validate-before-write ordering, just batched like the
|
||||
* in-memory engine does instead of row-by-row like [DiskTable].
|
||||
*/
|
||||
private fun executeTemporaryUpdate(statement: UpdateStatement, table: Table): StatementResult {
|
||||
val schema = table.schema
|
||||
val assignments = buildAssignments(schema, statement)
|
||||
return withStatementTransaction { txn ->
|
||||
val subqueries = subqueryRunnerFor(txn)
|
||||
val oldRows = table.rows.toList()
|
||||
val touchedRows = mutableListOf<Pair<Row, Row>>()
|
||||
val newRows = oldRows.map { row ->
|
||||
val matches = statement.where == null ||
|
||||
isTrue(Evaluator(composeResolver(rowResolver(schema, row), null), subqueries).evaluate(statement.where))
|
||||
if (!matches) return@map row
|
||||
val values = row.values.toMutableList()
|
||||
val evaluator = Evaluator(composeResolver(rowResolver(schema, row), null), subqueries, this)
|
||||
for ((index, expression) in assignments) {
|
||||
val column = schema.columns[index]
|
||||
values[index] = if (expression != null) {
|
||||
coerceForColumn(evaluator.evaluate(expression), column)
|
||||
} else {
|
||||
column.default?.let { coerceForColumn(evaluator.evaluate(it), column) } ?: Value.NullValue
|
||||
}
|
||||
}
|
||||
applyGeneratedColumns(schema, values, subqueries)
|
||||
validateTemporaryNotNull(schema, values)
|
||||
val newRow = Row(values)
|
||||
touchedRows.add(row to newRow)
|
||||
newRow
|
||||
}
|
||||
for (row in newRows) enforceChecks(schema, row, subqueries)
|
||||
enforceTemporaryPrimaryKeyUniqueness(schema, newRows)
|
||||
enforceTemporaryUniqueConstraints(schema, newRows)
|
||||
table.rows.clear()
|
||||
table.rows.addAll(newRows)
|
||||
val returning = statement.returning?.let { evaluateReturning(schema, touchedRows.map { it.second }, it, subqueries) }
|
||||
UpdateResult(touchedRows.size, returning)
|
||||
}
|
||||
}
|
||||
|
||||
private fun executeUpdateInTxn(
|
||||
statement: UpdateStatement,
|
||||
assignments: List<Pair<Int, Expression?>>,
|
||||
table: DiskTable,
|
||||
schema: TableSchema,
|
||||
txn: Long,
|
||||
triggerContext: TriggerRowContext?
|
||||
): StatementResult {
|
||||
val subqueries = subqueryRunnerFor(txn)
|
||||
val matches = candidateRows(table, txn, statement.where, alias = null).filter { (_, row) ->
|
||||
statement.where == null ||
|
||||
isTrue(Evaluator(composeResolver(rowResolver(schema, row), triggerContext), subqueries).evaluate(statement.where))
|
||||
}
|
||||
// Built as a whole batch first (not applied inline) for the same reason as executeInsert:
|
||||
// a self-referencing FK must see every row's final value for this statement, not just
|
||||
// rows already committed to the table.
|
||||
val planned = matches.map { (rid, row) ->
|
||||
val evaluator = Evaluator(composeResolver(rowResolver(schema, row), triggerContext), subqueries, this)
|
||||
val values = row.values.toMutableList()
|
||||
for ((index, expression) in assignments) {
|
||||
val column = schema.columns[index]
|
||||
values[index] = if (expression != null) {
|
||||
coerceForColumn(evaluator.evaluate(expression), column)
|
||||
} else {
|
||||
column.default?.let { coerceForColumn(evaluator.evaluate(it), column) } ?: Value.NullValue
|
||||
}
|
||||
}
|
||||
applyGeneratedColumns(schema, values, subqueries)
|
||||
Triple(rid, row, Row(values))
|
||||
}
|
||||
val newRows = planned.map { it.third }
|
||||
// Fires before any constraint below is checked, same as executeInsert's BEFORE trigger - a
|
||||
// BEFORE UPDATE trigger's body can reject the whole statement before this executor's own
|
||||
// validation ever runs.
|
||||
fireRowTriggers(schema, TriggerTiming.Before, TriggerEvent.Update, planned.map { it.second to it.third }, txn)
|
||||
for ((_, _, newRow) in planned) {
|
||||
DiskForeignKeyEnforcer.validateRow(catalog, txn, schema, newRow.values, uncommittedSiblings = newRows)
|
||||
enforceChecks(schema, newRow, subqueries)
|
||||
}
|
||||
// Every enforced-unique key group (the PRIMARY KEY, single-column or composite, plus each
|
||||
// UNIQUE constraint) is a valid FOREIGN KEY target, so each group needs its own
|
||||
// cascade/RESTRICT pass on UPDATE - not just the PRIMARY KEY. Run before the parent's own
|
||||
// rows are rewritten below, mirroring kestrel.engine.Executor.executeUpdate's own order.
|
||||
for (group in schema.enforcedUniqueKeyGroups()) {
|
||||
val columnIndices = group.map { schema.columnIndex(it)!! }
|
||||
val changedPairs = planned.map { (_, oldRow, newRow) ->
|
||||
columnIndices.map { oldRow.values[it] } to columnIndices.map { newRow.values[it] }
|
||||
}
|
||||
DiskForeignKeyEnforcer.cascadeUpdate(catalog, txn, schema.name, group, changedPairs)
|
||||
}
|
||||
for ((rid, _, newRow) in planned) table.update(txn, rid, newRow)
|
||||
fireRowTriggers(schema, TriggerTiming.After, TriggerEvent.Update, planned.map { it.second to it.third }, txn)
|
||||
val returning = statement.returning?.let { evaluateReturning(schema, planned.map { triple -> triple.third }, it, subqueries) }
|
||||
return UpdateResult(planned.size, returning)
|
||||
}
|
||||
|
||||
private fun executeDelete(statement: DeleteStatement, triggerContext: TriggerRowContext? = null): StatementResult {
|
||||
catalog.temporaryTableOrNull(statement.tableName)?.let { return executeTemporaryDelete(statement, it) }
|
||||
val table = catalog.table(statement.tableName)
|
||||
val schema = table.schema
|
||||
return withStatementTransaction { txn ->
|
||||
withTriggerAtomicity(schema.name, txn, triggerContext) {
|
||||
executeDeleteInTxn(statement, table, schema, txn, triggerContext)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun executeDeleteInTxn(
|
||||
statement: DeleteStatement, table: DiskTable, schema: TableSchema, txn: Long, triggerContext: TriggerRowContext?
|
||||
): StatementResult {
|
||||
val subqueries = subqueryRunnerFor(txn)
|
||||
val matches = candidateRows(table, txn, statement.where, alias = null).filter { (_, row) ->
|
||||
statement.where == null ||
|
||||
isTrue(Evaluator(composeResolver(rowResolver(schema, row), triggerContext), subqueries).evaluate(statement.where))
|
||||
}
|
||||
fireRowTriggers(schema, TriggerTiming.Before, TriggerEvent.Delete, matches.map { it.second to null }, txn)
|
||||
DiskForeignKeyEnforcer.cascadeDelete(catalog, txn, schema.name, matches.map { it.second })
|
||||
for ((rid, _) in matches) table.delete(txn, rid)
|
||||
fireRowTriggers(schema, TriggerTiming.After, TriggerEvent.Delete, matches.map { it.second to null }, txn)
|
||||
val returning = statement.returning?.let { evaluateReturning(schema, matches.map { it.second }, it, subqueries) }
|
||||
return DeleteResult(matches.size, returning)
|
||||
}
|
||||
|
||||
/** DELETE on a temp table's plain [Table.rows] list - ported from the same shape as [executeDeleteInTxn]/[kestrel.engine.Executor.executeDeleteInner], minus trigger firing/[DiskForeignKeyEnforcer] cascade (both rejected outright at `CREATE TEMP TABLE` time). */
|
||||
private fun executeTemporaryDelete(statement: DeleteStatement, table: Table): StatementResult {
|
||||
val schema = table.schema
|
||||
return withStatementTransaction { txn ->
|
||||
val subqueries = subqueryRunnerFor(txn)
|
||||
val deletedRows = if (statement.where == null) {
|
||||
table.rows.toList()
|
||||
} else {
|
||||
table.rows.filter { row ->
|
||||
isTrue(Evaluator(composeResolver(rowResolver(schema, row), null), subqueries).evaluate(statement.where))
|
||||
}
|
||||
}
|
||||
table.rows.removeAll(deletedRows)
|
||||
val returning = statement.returning?.let { evaluateReturning(schema, deletedRows, it, subqueries) }
|
||||
DeleteResult(deletedRows.size, returning)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user