feat(engine): extend Executor with fireRowTriggers and runTriggerBody

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

@@ -1128,15 +1128,180 @@ class Executor(private val catalog: Catalog) : SequenceAccess {
             else -> existing.maxValue
         }
         if (minValue > maxValue) {
+            throw KestrelException("sequence '${statement.sequenceName}' MINVALUE ($minValue) must not exceed MAXVALUE ($maxValue)")
+        }
+        val cycle = statement.cycle ?: existing.cycle
+        val newDefinition = SequenceDefinition(statement.sequenceName, incrementBy, minValue, maxValue, existing.startValue, cycle)
+        val restartAt = if (statement.restart) (statement.restartWith ?: existing.startValue) else null
+        if (restartAt != null && (restartAt < minValue || restartAt > maxValue)) {
+            throw KestrelException("RESTART value ($restartAt) must be between MINVALUE ($minValue) and MAXVALUE ($maxValue)")
+        }
+        catalog.alterSequence(statement.sequenceName, newDefinition, restartAt)
+        return AlterSequenceResult(statement.sequenceName)
+    }
+
+    /**
+     * Fires every trigger registered for ([schema], [timing], [event]) in creation order, once per
+     * (oldRow, newRow) pair in [rows] - INSERT passes `null` for oldRow, DELETE passes `null` for
+     * newRow, UPDATE passes both. A `WHEN` clause that evaluates to false or NULL skips only that
+     * trigger for that row, exactly like a plain WHERE clause would; it never skips other triggers
+     * on the same event, and never affects the firing statement itself.
+     */
+    private fun fireRowTriggers(schema: TableSchema, timing: TriggerTiming, event: TriggerEvent, rows: List<Pair<Row?, Row?>>) {
+        val triggers = catalog.triggersOn(schema.name, timing, event)
+        if (triggers.isEmpty()) return
+        for ((oldRow, newRow) in rows) {
+            val context = TriggerRowContext(schema, newRow, oldRow)
+            for (trigger in triggers) {
+                val whenClause = trigger.whenClause
+                if (whenClause != null) {
+                    val evaluator = Evaluator(composeResolver(null, context), subqueryRunner)
+                    if (!isTrue(evaluator.evaluate(whenClause))) continue
+                }
+                runTriggerBody(trigger, context)
+            }
+        }
+    }
+
+    private fun runTriggerBody(trigger: TriggerDefinition, context: TriggerRowContext) {
+        if (triggerDepth >= MAX_TRIGGER_DEPTH) {
+            throw KestrelException(
+                "trigger recursion limit ($MAX_TRIGGER_DEPTH) exceeded while firing '${trigger.name}' - " +
+                    "check for a trigger whose body ends up re-triggering itself"
+            )
+        }
+        triggerDepth++
+        try {
+            for (bodyStatement in trigger.body) {
+                when (bodyStatement) {
+                    is InsertStatement -> executeInsert(bodyStatement, context)
+                    is UpdateStatement -> executeUpdate(bodyStatement, context)
+                    is DeleteStatement -> executeDelete(bodyStatement, context)
+                    else -> error("unreachable: trigger bodies only ever contain INSERT/UPDATE/DELETE")
+                }
+            }
+        } finally {
+            triggerDepth--
+        }
+    }
+
+    /**
+     * Resolves `NEW.col`/`OLD.col` against [trigger] first (when both the qualifier and that side
+     * of the row are present); everything else - including a bare column with no qualifier at
+     * all - falls through to [base] exactly as it would with no trigger in play, so passing
+     * `trigger = null` (or a [TriggerRowContext] that never matches) reproduces today's behavior
+     * without change.
+     */
+    private fun composeResolver(base: ColumnResolver?, trigger: TriggerRowContext?): ColumnResolver = object : ColumnResolver {
+        override fun resolve(expression: ColumnExpression): Value =
+            if (trigger != null && trigger.matches(expression)) trigger.resolve(expression)
+            else base?.resolve(expression) ?: throw KestrelException("column '${expression.name}' is not allowed here")
+
+        override fun declaredCollation(expression: ColumnExpression): Collation? =
+            if (trigger != null && trigger.matches(expression)) trigger.declaredCollation(expression)
+            else base?.declaredCollation(expression)
+    }
+
+    /** Reproduces the (schema, row) convenience constructor on [Evaluator], as a [ColumnResolver] that [composeResolver] can layer a [TriggerRowContext] on top of. */
+    private fun rowResolver(schema: TableSchema, row: Row): ColumnResolver = object : ColumnResolver {
+        override fun resolve(expression: ColumnExpression): Value {
+            val index = schema.columnIndex(expression.name)
+                ?: throw KestrelException("no such column: ${expression.name}")
+            return row.values[index]
+        }
+
+        override fun declaredCollation(expression: ColumnExpression): Collation? = schema.column(expression.name)?.collation
+    }
+
+    /**
+     * Evaluates an INSERT/UPDATE/DELETE statement's `RETURNING` clause against the rows it actually
+     * affected. [rows] is always the image RETURNING should see for that statement kind: the freshly
+     * inserted or updated row for INSERT/UPDATE (there is nothing else to show), the pre-deletion row
+     * for DELETE (it no longer exists in the table afterward). Column references resolve against
+     * [schema] alone - never NEW/OLD, since [rejectReturning] already keeps RETURNING out of every
+     * trigger body statement, so this is always the outermost, non-nested call. [checkColumns] runs
+     * over every projected expression up front (not just lazily via [rowResolver] while evaluating
+     * row by row) so a bad column name or a stray aggregate is reported even when zero rows matched -
+     * matching how every other schema-shaped clause in this engine is validated regardless of data.
+     */
+    private fun evaluateReturning(schema: TableSchema, rows: List<Row>, projection: Projection): SelectResult {
+        val columns = when (projection) {
+            is StarProjection -> schema.columns.map { it.name }
+            is ExpressionProjection -> projection.items.mapIndexed { index, item ->
+                checkColumns(item.expression, schema)
+                returningColumnLabel(item, index)
+            }
+        }
+        val resultRows = rows.map { row ->
+            when (projection) {
+                is StarProjection -> row.values.toList()
+                is ExpressionProjection -> {
+                    val evaluator = Evaluator(rowResolver(schema, row), subqueryRunner)
+                    projection.items.map { evaluator.evaluate(it.expression) }
+                }
+            }
+        }
+        return SelectResult(columns, resultRows)
+    }
+
+    private fun returningColumnLabel(item: SelectItem, index: Int): String =
+        item.alias ?: when (val expression = item.expression) {
+            is ColumnExpression -> expression.name
+            is FunctionExpression -> expression.name.lowercase()
+            is CallExpression -> expression.name.lowercase()
+            is WindowExpression -> expression.name.lowercase()
+            else -> "column${index + 1}"
+        }
+
+    /**
+     * Without this, a BEFORE trigger's own INSERT/UPDATE/DELETE would survive even when the
+     * statement that fired it goes on to fail its own constraint check (or an AFTER trigger fails
+     * after the row already landed) - a real database treats a whole statement plus everything its
+     * triggers do as one atomic unit, and this reuses the exact snapshot/restore machinery
+     * [executeRollback]/[executeRollbackToSavepoint] already rely on to do the same. Only the
+     * OUTERMOST statement in a trigger cascade takes the snapshot ([triggerContext] is null exactly
+     * there): [catalog.snapshot] captures the WHOLE catalog, not just [tableName], so restoring it
+     * once at the top undoes every table any nested trigger body touched, not only this one - a
+     * nested call skips wrapping itself again, and skips it entirely when [tableName] has no
+     * trigger at all, so the overwhelmingly common trigger-free INSERT/UPDATE/DELETE pays no
+     * snapshot cost.
+     */
+    private fun <T> withStatementAtomicity(tableName: String, triggerContext: TriggerRowContext?, block: () -> T): T {
+        if (triggerContext != null || !catalog.hasTriggers(tableName)) return block()
+        val snapshot = catalog.snapshot()
+        try {
+            return block()
+        } catch (exception: Throwable) {
+            catalog.restore(snapshot)
+            throw exception
+        }
+    }
+
+    private companion object {
+        const val MAX_TRIGGER_DEPTH = 100
+    }
+
+    private fun executeInsert(statement: InsertStatement, triggerContext: TriggerRowContext? = null): StatementResult =
+        withStatementAtomicity(statement.tableName, triggerContext) { executeInsertInner(statement, triggerContext) }
+
+    private fun executeInsertInner(statement: InsertStatement, triggerContext: TriggerRowContext?): StatementResult {
         val table = catalog.table(statement.tableName)
         val schema = table.schema
         val targetIndices = resolveInsertTargets(schema, statement.columns)
-        val evaluator = Evaluator()
+        val evaluator = Evaluator(composeResolver(null, triggerContext), subqueryRunner, this)
-        val newRows = statement.rows.map { rawRow ->
+        val newRows = statement.rows.map { rawRow -> buildInsertRow(schema, targetIndices, rawRow, evaluator) }
-            if (rawRow.size != targetIndices.size) {
+        val onConflict = statement.onConflict
-                throw KestrelException(
+        if (onConflict == null) {
-                    "table ${schema.name} expects ${targetIndices.size} values but ${rawRow.size} were supplied"
+            // Fires before any constraint is even checked, matching SQLite/Postgres: a BEFORE
-                )
+            // trigger can reject the whole statement (by having its body throw) before this
+            // engine's own NOT NULL/CHECK/FOREIGN KEY/UNIQUE validation ever runs.
+            fireRowTriggers(schema, TriggerTiming.Before, TriggerEvent.Insert, newRows.map { null to it })
+            // Validated as a whole batch (not inline while building newRows above) so that a
+            // self-referencing FK can see sibling rows from the same multi-row INSERT regardless
+            // of their order, exactly like enforcePrimaryKeyUniqueness below already does.
+            for (row in newRows) {
+                ForeignKeyEnforcer.validateRow(catalog, schema, row.values, uncommittedSiblings = newRows)
+                enforceChecks(schema, row)
             }
             val values = MutableList<Value>(schema.columns.size) { Value.NullValue }
             for ((position, expression) in rawRow.withIndex()) {