feat(sql): add executeSavepoint and executeReleaseSavepoint
This commit is contained in:
dualfroz committed
1 parent
73672c0184
commit
dff226d846
1 file changed
+169
@@ -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 <T> 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 <T> 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<Row>, 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<Pair<Row?, Row?>>, 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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user