From daf6ce803fcc1c6f591e6d9ab71c2d112855a7eb Mon Sep 17 00:00:00 2001 From: DualFroz Date: Sun, 6 Sep 2026 19:54:11 +0200 Subject: [PATCH] feat(sql): add runTriggerBody and buildAssignments to DiskExecutor --- .../kotlin/kestrel/store/sql/DiskExecutor.kt | 157 ++++++++++++++++++ 1 file changed, 157 insertions(+) diff --git a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt index 18a3c9b..28809b0 100644 --- a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt +++ b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt @@ -642,3 +642,160 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess { } } + /** + * Runs a fired trigger's body statements in order, directly against [txn] - NOT through + * [executeInsert]/[executeUpdate]/[executeDelete] (which would re-select a transaction via + * [withStatementTransaction] and re-take a trigger-atomicity savepoint via + * [withTriggerAtomicity]) - so the whole cascade shares exactly one disk transaction and exactly + * one savepoint, taken once by the outermost caller. Mirrors [kestrel.engine.Executor. + * runTriggerBody], including [MAX_TRIGGER_DEPTH]. + */ + private fun runTriggerBody(trigger: TriggerDefinition, context: TriggerRowContext, txn: Long) { + if (triggerDepth >= MAX_TRIGGER_DEPTH) { + throw DiskExecutorException( + "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 -> { + val table = catalog.table(bodyStatement.tableName) + executeInsertInTxn(bodyStatement, table, table.schema, txn, context) + } + is UpdateStatement -> { + val table = catalog.table(bodyStatement.tableName) + executeUpdateInTxn(bodyStatement, buildAssignments(table.schema, bodyStatement), table, table.schema, txn, context) + } + is DeleteStatement -> { + val table = catalog.table(bodyStatement.tableName) + executeDeleteInTxn(bodyStatement, table, table.schema, txn, context) + } + else -> error("unreachable: trigger bodies only ever contain INSERT/UPDATE/DELETE") + } + } + } finally { + triggerDepth-- + } + } + + private fun buildAssignments(schema: TableSchema, statement: UpdateStatement): List> = + statement.assignments.map { assignment -> + val index = schema.columnIndex(assignment.columnName) + ?: throw DiskExecutorException("no such column: ${assignment.columnName} in table ${schema.name}") + rejectGeneratedAssignment(schema.columns[index], explicitValueGiven = assignment.value != null) + index to assignment.value + } + + private fun executeCreateIndex(statement: CreateIndexStatement): StatementResult { + catalog.createIndex(statement.tableName, statement.indexName, statement.columnNames, statement.unique) + return CreateIndexResult(statement.indexName) + } + + private fun executeDropIndex(statement: DropIndexStatement): StatementResult { + catalog.dropIndex(statement.indexName) + return DropIndexResult(statement.indexName) + } + + /** + * Ported 1:1 from [kestrel.engine.Executor.executeCreateSequence] - see that method's own doc + * for the ascending/descending default resolution and the one thing [SequenceDefinition]'s own + * `init` cannot check itself (START WITH falling inside [MINVALUE, MAXVALUE]). + */ + private fun executeCreateSequence(statement: CreateSequenceStatement): StatementResult { + val incrementBy = statement.incrementBy ?: 1L + if (incrementBy == 0L) throw DiskExecutorException("sequence '${statement.sequenceName}' INCREMENT BY cannot be zero") + val ascending = incrementBy > 0 + val minValue = statement.minValue ?: if (ascending) 1L else Long.MIN_VALUE + val maxValue = statement.maxValue ?: if (ascending) Long.MAX_VALUE else -1L + if (minValue > maxValue) { + throw DiskExecutorException("sequence '${statement.sequenceName}' MINVALUE ($minValue) must not exceed MAXVALUE ($maxValue)") + } + val startValue = statement.startValue ?: if (ascending) minValue else maxValue + if (startValue < minValue || startValue > maxValue) { + throw DiskExecutorException( + "START WITH ($startValue) must be between MINVALUE ($minValue) and MAXVALUE ($maxValue)" + ) + } + catalog.createSequence(SequenceDefinition(statement.sequenceName, incrementBy, minValue, maxValue, startValue, statement.cycle)) + return CreateSequenceResult(statement.sequenceName) + } + + private fun executeDropSequence(statement: DropSequenceStatement): StatementResult { + catalog.dropSequence(statement.sequenceName) + return DropSequenceResult(statement.sequenceName) + } + + /** Ported 1:1 from [kestrel.engine.Executor.executeAlterSequence] - see that method's own doc. */ + private fun executeAlterSequence(statement: AlterSequenceStatement): StatementResult { + val existing = catalog.sequence(statement.sequenceName).definition + val incrementBy = statement.incrementBy ?: existing.incrementBy + if (incrementBy == 0L) throw DiskExecutorException("sequence '${statement.sequenceName}' INCREMENT BY cannot be zero") + val ascending = incrementBy > 0 + val minValue = when { + statement.clearMinValue -> if (ascending) 1L else Long.MIN_VALUE + statement.minValue != null -> statement.minValue + else -> existing.minValue + } + val maxValue = when { + statement.clearMaxValue -> if (ascending) Long.MAX_VALUE else -1L + statement.maxValue != null -> statement.maxValue + else -> existing.maxValue + } + if (minValue > maxValue) { + throw DiskExecutorException("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 DiskExecutorException("RESTART value ($restartAt) must be between MINVALUE ($minValue) and MAXVALUE ($maxValue)") + } + catalog.alterSequence(statement.sequenceName, newDefinition, restartAt) + return AlterSequenceResult(statement.sequenceName) + } + + /** + * Runs [CreateViewStatement.query] once, discarding the result, purely to fail fast on a bad + * column/table reference before anything is persisted - mirrors [kestrel.engine.Executor. + * executeCreateView] calling `selectRunner.explain` for the same reason, just via a real + * execution rather than a plan-only walk (this executor has no `explain`). + */ + private fun executeCreateView(statement: CreateViewStatement): StatementResult { + val txn = catalog.beginTransaction() + runInTransaction(txn) { runQuery(statement.query, txn, outer = null) } + catalog.createView(statement.viewName, statement.query, statement.orReplace) + return CreateViewResult(statement.viewName) + } + + private fun executeDropView(statement: DropViewStatement): StatementResult { + catalog.dropView(statement.viewName, cascade = statement.mode == DropMode.Cascade) + return DropViewResult(statement.viewName) + } + + /** + * Ported 1:1 from [kestrel.engine.Executor.executeCreateTrigger]: body statements are + * deliberately NOT deep-validated here (no eager check that every column referenced in a body + * statement's WHERE/SET/VALUES actually exists) - a trigger body can target a different table per + * statement, and this executor already defers exactly this kind of check to the moment the + * trigger actually fires, same as SQLite. What IS checked eagerly: the trigger's own table and + * every body statement's target table both exist right now, and the WHEN clause - evaluated + * directly by this executor rather than through a nested statement run, so nothing else will ever + * validate it - contains no subquery/aggregate and only ever references NEW/OLD columns that both + * exist on this table and are actually available for this trigger's event. + */ + private fun executeCreateTrigger(statement: CreateTriggerStatement): StatementResult { + val schema = catalog.table(statement.tableName).schema + statement.whenClause?.let { whenClause -> + requireNoSubquery(whenClause) + validateTriggerRowReferences(whenClause, statement.event, schema) + } + for (bodyStatement in statement.body) validateTriggerBodyStatementTarget(bodyStatement) + catalog.createTrigger( + TriggerDefinition(statement.triggerName, statement.timing, statement.event, statement.tableName, statement.whenClause, statement.body) + ) + return CreateTriggerResult(statement.triggerName) + } +