feat(engine): extend Executor with executeUpdate and executeUpdateInner

src/main/kotlin/kestrel/engine/Executor.kt

@@ -1542,53 +1542,91 @@ class Executor(private val catalog: Catalog) : SequenceAccess {
     private fun executeDeleteInner(statement: DeleteStatement, triggerContext: TriggerRowContext?): StatementResult {
         val table = catalog.table(statement.tableName)
         val schema = table.schema
-        if (statement.where == null) {
-            val removed = table.rows.size
-            table.rows.clear()
-            catalog.reindexTable(schema.name)
-            return DeleteResult(removed)
-        }
-        checkColumns(statement.where, schema)
-        val kept = mutableListOf<Row>()
-        var deleted = 0
-        for (row in table.rows) {
-            if (isTrue(Evaluator(schema, row).evaluate(statement.where))) {
-                deleted++
-            } else {
-                kept.add(row)
+        val deletedRows = if (statement.where == null) {
+            table.rows.toList()
+        } else {
+            checkColumns(statement.where, schema, triggerContext)
+            table.rows.filter {
+                isTrue(Evaluator(composeResolver(rowResolver(schema, it), triggerContext), subqueryRunner).evaluate(statement.where))
             }
         }
-        table.rows.clear()
-        table.rows.addAll(kept)
+        fireRowTriggers(schema, TriggerTiming.Before, TriggerEvent.Delete, deletedRows.map { it to null })
+        ForeignKeyEnforcer.cascadeDelete(catalog, schema.name, deletedRows)
+        table.rows.removeAll(deletedRows)
         catalog.reindexTable(schema.name)
-        return DeleteResult(deleted)
+        fireRowTriggers(schema, TriggerTiming.After, TriggerEvent.Delete, deletedRows.map { it to null })
+        val returning = statement.returning?.let { evaluateReturning(schema, deletedRows, it) }
+        return DeleteResult(deletedRows.size, returning)
     }
 
-    private fun executeUpdate(statement: UpdateStatement): StatementResult {
+    private fun executeUpdate(statement: UpdateStatement, triggerContext: TriggerRowContext? = null): StatementResult =
+        withStatementAtomicity(statement.tableName, triggerContext) { executeUpdateInner(statement, triggerContext) }
+
+    private fun executeUpdateInner(statement: UpdateStatement, triggerContext: TriggerRowContext?): StatementResult {
         val table = catalog.table(statement.tableName)
         val schema = table.schema
         val assignments = statement.assignments.map { assignment ->
             val index = schema.columnIndex(assignment.columnName)
                 ?: throw KestrelException("no such column: ${assignment.columnName}")
-            checkColumns(assignment.value, schema)
+            assignment.value?.let { checkColumns(it, schema, triggerContext) }
+            rejectGeneratedAssignment(schema.columns[index], explicitValueGiven = assignment.value != null)
             index to assignment.value
         }
-        statement.where?.let { checkColumns(it, schema) }
+        statement.where?.let { checkColumns(it, schema, triggerContext) }
 
         var updated = 0
-        val newRows = table.rows.map { row ->
-            val matches = statement.where == null || isTrue(Evaluator(schema, row).evaluate(statement.where))
+        val oldRows = table.rows.toList()
+        // Every row actually matched by WHERE, paired old-to-new - fires an UPDATE trigger for
+        // every row WHERE selects, regardless of whether any assigned value actually differs from
+        // what was already there (matching standard SQL: WHERE selects the affected rows, not the
+        // changed ones).
+        val touchedRows = mutableListOf<Pair<Row, Row>>()
+        val newRows = oldRows.map { row ->
+            val matches = statement.where == null ||
+                isTrue(Evaluator(composeResolver(rowResolver(schema, row), triggerContext), subqueryRunner).evaluate(statement.where))
             if (!matches) return@map row
             val values = row.values.toMutableList()
-            val evaluator = Evaluator(schema, row)
+            // NEXTVAL/CURRVAL are allowed in SET's assignment values (below), not in WHERE (above) -
+            // a separate Evaluator per clause, same reasoning as executeInsertWithConflictHandling.
+            val evaluator = Evaluator(composeResolver(rowResolver(schema, row), triggerContext), subqueryRunner, this)
             for ((index, expression) in assignments) {
-                values[index] = coerceForColumn(evaluator.evaluate(expression), schema.columns[index])
+                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)
             validateNotNull(schema, values)
             updated++
-            Row(values)
+            val newRow = Row(values)
+            touchedRows.add(row to newRow)
+            newRow
+        }
+        // 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 engine's own
+        // validation ever runs.
+        fireRowTriggers(schema, TriggerTiming.Before, TriggerEvent.Update, touchedRows)
+        // Same whole-batch validation as executeInsert: a self-referencing FK must see every row's
+        // final value for this statement, not just rows already committed to the table.
+        for (row in newRows) {
+            ForeignKeyEnforcer.validateRow(catalog, schema, row.values, uncommittedSiblings = newRows)
+            enforceChecks(schema, row)
         }
         enforcePrimaryKeyUniqueness(schema, newRows)
+        enforceUniqueConstraints(schema, newRows)
+        enforceUniqueIndexes(schema, newRows)
+        // 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.
+        for (group in schema.enforcedUniqueKeyGroups()) {
+            val columnIndices = group.map { schema.columnIndex(it)!! }
+            val changedPairs = oldRows.indices.map { i ->
+                columnIndices.map { oldRows[i].values[it] } to columnIndices.map { newRows[i].values[it] }
+            }
+            ForeignKeyEnforcer.cascadeUpdate(catalog, schema.name, group, changedPairs)
+        }
         table.rows.clear()
         table.rows.addAll(newRows)
         catalog.reindexTable(schema.name)