diff --git a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt index 0e55298..9d2368e 100644 --- a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt +++ b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt @@ -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>() + 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>, + 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) + } + } +