From dff226d8460f5578c70d659429329db64fc91c56 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Sun, 6 Sep 2026 19:45:52 +0200 Subject: [PATCH] feat(sql): add executeSavepoint and executeReleaseSavepoint --- .../kotlin/kestrel/store/sql/DiskExecutor.kt | 169 ++++++++++++++++++ 1 file changed, 169 insertions(+) diff --git a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt index c5e153c..18a3c9b 100644 --- a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt +++ b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt @@ -473,3 +473,172 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess { return RollbackResult } + /** + * Establishing a savepoint costs nothing beyond capturing the storage engine's current WAL + * position and every open index's root pointer (see [DiskCatalog.savepoint]) plus a temp table + * row snapshot (see [temporaryTableSnapshots]) - unlike [kestrel.engine.Executor], there is no + * whole-CATALOG snapshot to take, since DDL (the only thing that mutates [catalog] itself rather + * than a table's data) is already rejected inside an explicit transaction on this executor. + */ + private fun executeSavepoint(statement: SavepointStatement): StatementResult { + val txn = explicitTxn ?: throw DiskExecutorException("SAVEPOINT can only be used inside a transaction") + savepoints.add(statement.name to catalog.savepoint(txn)) + temporaryTableSnapshots.add(snapshotTemporaryTables()) + return SavepointResult(statement.name) + } + + /** Forgets the named savepoint and every savepoint established after it, without undoing anything. */ + private fun executeReleaseSavepoint(statement: ReleaseSavepointStatement): StatementResult { + val txn = explicitTxn ?: throw DiskExecutorException("RELEASE SAVEPOINT can only be used inside a transaction") + val index = indexOfLastSavepoint(statement.name) + while (savepoints.size > index) { + val (_, lsn) = savepoints.removeAt(savepoints.size - 1) + catalog.discardSavepoint(txn, lsn) + temporaryTableSnapshots.removeAt(temporaryTableSnapshots.size - 1) + } + return ReleaseSavepointResult(statement.name) + } + + /** + * Undoes every write made since the named savepoint, but - matching [kestrel.engine.Executor] and + * real SQL savepoint semantics - keeps that savepoint itself usable for a later rollback, only + * discarding the ones established after it. + */ + private fun executeRollbackToSavepoint(statement: RollbackToSavepointStatement): StatementResult { + val txn = explicitTxn ?: throw DiskExecutorException("ROLLBACK TO SAVEPOINT can only be used inside a transaction") + val index = indexOfLastSavepoint(statement.name) + catalog.rollbackToSavepoint(txn, savepoints[index].second) + restoreTemporaryTables(temporaryTableSnapshots[index + 1]) + while (savepoints.size > index + 1) { + val (_, lsn) = savepoints.removeAt(savepoints.size - 1) + catalog.discardSavepoint(txn, lsn) + temporaryTableSnapshots.removeAt(temporaryTableSnapshots.size - 1) + } + return RollbackToSavepointResult(statement.name) + } + + private fun indexOfLastSavepoint(name: String): Int { + val index = savepoints.indexOfLast { it.first.equals(name, ignoreCase = true) } + if (index < 0) throw DiskExecutorException("no such savepoint: $name") + return index + } + + /** + * Runs [block] against the transaction opened by a `BEGIN` still in progress so several + * statements share one commit/abort boundary, or against a fresh transaction scoped to just this + * statement (committed on success, aborted on failure via [runInTransaction]) when there isn't + * one - the same per-statement auto-commit [kestrel.engine.Executor] gets for free from running + * directly against an in-memory [kestrel.catalog.Catalog] with no transaction concept at all. + * Locks taken by [block] are held by the shared transaction until that later COMMIT/ROLLBACK per + * strict two-phase locking (see [kestrel.store.txn.LockManager]), so a SELECT inside an open + * transaction can block a concurrent writer the way a real multi-statement transaction should. + */ + private fun withStatementTransaction(block: (Long) -> T): T { + val active = explicitTxn + if (active != null) return block(active) + val txn = catalog.beginTransaction() + return runInTransaction(txn) { block(txn) } + } + + /** + * Savepoint-based analog of [kestrel.engine.Executor.withStatementAtomicity]: 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. Only the OUTERMOST statement in a trigger cascade takes the savepoint + * ([triggerContext] is null exactly there, mirroring the in-memory engine's own `triggerContext == + * null` check): [catalog.savepoint]/[catalog.rollbackToSavepoint] are keyed purely by WAL LSN, + * entirely independent of - and safely nestable inside - the executor's own named `SAVEPOINT` + * stack ([savepoints]), so this composes with an explicit transaction the user is already inside + * without disturbing it. Skipped entirely when [tableName] has no trigger at all, so the + * overwhelmingly common trigger-free INSERT/UPDATE/DELETE pays no savepoint cost. + */ + private fun withTriggerAtomicity(tableName: String, txn: Long, triggerContext: TriggerRowContext?, block: () -> T): T { + if (triggerContext != null || !catalog.hasTriggers(tableName)) return block() + val savepointLsn = catalog.savepoint(txn) + val result = try { + block() + } catch (failure: Exception) { + catalog.rollbackToSavepoint(txn, savepointLsn) + catalog.discardSavepoint(txn, savepointLsn) + throw failure + } + catalog.discardSavepoint(txn, savepointLsn) + return result + } + + /** + * 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. Ported 1:1 from + * [kestrel.engine.Executor.composeResolver], just with [DiskExecutorException] for the fallthrough. + */ + 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 DiskExecutorException("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) + } + + /** Ported 1:1 from [kestrel.engine.Executor.rowResolver] - a [ColumnResolver] [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 DiskExecutorException("no such column: ${expression.name}") + return row.values[index] + } + + override fun declaredCollation(expression: ColumnExpression): Collation? = schema.column(expression.name)?.collation + } + + /** + * Ported 1:1 from [kestrel.engine.Executor.evaluateReturning]: evaluates an INSERT/UPDATE/ + * DELETE statement's `RETURNING` clause against the rows it actually affected - the freshly + * inserted/updated row for INSERT/UPDATE, the pre-deletion row for DELETE. [checkReturningColumns] + * validates every projected expression up front so a bad column name is reported even when zero + * rows matched, not just lazily the first time [rowResolver] happens to see one. + */ + private fun evaluateReturning(schema: TableSchema, rows: List, projection: Projection, subqueries: SubqueryRunner): SelectResult { + val columns = when (projection) { + is StarProjection -> schema.columns.map { it.name } + is ExpressionProjection -> projection.items.mapIndexed { index, item -> + checkReturningColumns(item.expression, schema) + columnLabel(item, index) + } + } + val resultRows = rows.map { row -> + when (projection) { + is StarProjection -> row.values.toList() + is ExpressionProjection -> { + val evaluator = Evaluator(rowResolver(schema, row), subqueries) + projection.items.map { evaluator.evaluate(it.expression) } + } + } + } + return SelectResult(columns, resultRows) + } + + /** + * Fires every trigger registered for ([schema], [timing], [event]) in creation order, once per + * (oldRow, newRow) pair in [rows] - mirrors [kestrel.engine.Executor.fireRowTriggers] exactly, + * including that a `WHEN` clause evaluating to false/NULL skips only that trigger for that row. + */ + private fun fireRowTriggers(schema: TableSchema, timing: TriggerTiming, event: TriggerEvent, rows: List>, txn: Long) { + val triggers = catalog.triggersOn(schema.name, timing, event) + if (triggers.isEmpty()) return + val subqueries = subqueryRunnerFor(txn) + 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), subqueries) + if (!isTrue(evaluator.evaluate(whenClause))) continue + } + runTriggerBody(trigger, context, txn) + } + } + } +