diff --git a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt index c2b7dd0..c5e153c 100644 --- a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt +++ b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt @@ -314,3 +314,162 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess { val preparedStatementNames: List get() = preparedStatements.keys.toList() + /** The disk transaction opened by a `BEGIN` still awaiting `COMMIT`/`ROLLBACK`, or null outside one. */ + private var explicitTxn: Long? = null + + /** Whether an explicit `BEGIN` is currently open - mirrors [kestrel.engine.Executor.inTransaction]. */ + val inTransaction: Boolean get() = explicitTxn != null + + /** + * Named savepoints established inside the currently open [explicitTxn], oldest first, each paired + * with the storage engine LSN [DiskCatalog.savepoint] captured for it - mirrors + * [kestrel.engine.Executor]'s own `savepoints` stack (catalog snapshots there, WAL positions here) + * including reuse-shadows-the-older-one-of-the-same-name semantics via [indexOfLastSavepoint]. + */ + private val savepoints = mutableListOf>() + + /** + * A temp table's rows live in a plain [Table.rows] list outside [DiskCatalog]'s own WAL/2PL + * transaction (see [DiskCatalog]'s KDoc on [DiskCatalog.temporaryTableValues]), so `ROLLBACK`/ + * `ROLLBACK TO SAVEPOINT` cannot undo a temp table's INSERT/UPDATE/DELETE the way they undo a + * real [DiskTable]'s writes just by aborting/rewinding [explicitTxn] - this stack is what makes + * them do it anyway. Index 0 is the snapshot taken at `BEGIN`; index `i + 1` is the snapshot + * taken when `savepoints[i]` was established, so `ROLLBACK` restores index 0 and `ROLLBACK TO + * SAVEPOINT` restores `temporaryTableSnapshots[index + 1]` - kept in lockstep with [savepoints] + * by every method below. Only *row contents* ever need restoring, never which temp tables exist: + * `CREATE`/`DROP TEMPORARY TABLE` are DDL, already rejected inside an explicit transaction (see + * [isDdl]), so the set of temp tables is fixed for the whole lifetime of one [explicitTxn]. + */ + private val temporaryTableSnapshots = mutableListOf>>() + + private fun snapshotTemporaryTables(): Map> = + catalog.temporaryTableValues.associateWith { it.rows.toList() } + + private fun restoreTemporaryTables(snapshot: Map>) { + for ((table, rows) in snapshot) { + table.rows.clear() + table.rows.addAll(rows) + } + } + + fun execute(statement: Statement): StatementResult { + if (explicitTxn != null && isDdl(statement)) { + throw DiskExecutorException( + "${statement::class.simpleName} cannot run inside an explicit transaction on the " + + "disk-backed executor - COMMIT or ROLLBACK first. Unlike the in-memory engine's " + + "whole-catalog snapshot/restore, an explicit transaction here is the storage " + + "engine's own WAL/2PL transaction (see BEGIN/COMMIT/ROLLBACK below), which has no " + + "way to undo a schema change recorded by DiskCatalog outside of it." + ) + } + return when (statement) { + is CreateTableStatement -> executeCreateTable(statement) + is DropTableStatement -> executeDropTable(statement) + is TruncateTableStatement -> executeTruncate(statement) + is AlterTableStatement -> executeAlterTable(statement) + is CreateIndexStatement -> executeCreateIndex(statement) + is DropIndexStatement -> executeDropIndex(statement) + is CreateViewStatement -> executeCreateView(statement) + is DropViewStatement -> executeDropView(statement) + is CreateSequenceStatement -> executeCreateSequence(statement) + is DropSequenceStatement -> executeDropSequence(statement) + is AlterSequenceStatement -> executeAlterSequence(statement) + is CreateTriggerStatement -> executeCreateTrigger(statement) + is DropTriggerStatement -> executeDropTrigger(statement) + is InsertStatement -> executeInsert(statement) + is MergeStatement -> executeMerge(statement) + is QueryStatement -> executeQuery(statement) + is ExplainStatement -> executeExplain(statement) + is UpdateStatement -> executeUpdate(statement) + is DeleteStatement -> executeDelete(statement) + is BeginStatement -> executeBegin() + is CommitStatement -> executeCommit() + is RollbackStatement -> executeRollback() + is SavepointStatement -> executeSavepoint(statement) + is ReleaseSavepointStatement -> executeReleaseSavepoint(statement) + is RollbackToSavepointStatement -> executeRollbackToSavepoint(statement) + is PrepareStatement -> executePrepare(statement) + is ExecuteStatement -> execute(resolveExecute(statement)) + is DeallocateStatement -> executeDeallocate(statement) + else -> throw DiskExecutorException( + "${statement::class.simpleName} is not supported by the disk-backed executor yet" + ) + } + } + + private fun isDdl(statement: Statement): Boolean = when (statement) { + is CreateTableStatement, is DropTableStatement, is AlterTableStatement, + is CreateIndexStatement, is DropIndexStatement, is CreateViewStatement, is DropViewStatement, + is CreateSequenceStatement, is DropSequenceStatement, is AlterSequenceStatement, + is CreateTriggerStatement, is DropTriggerStatement -> true + else -> false + } + + private fun executePrepare(statement: PrepareStatement): StatementResult { + val key = statement.name.lowercase() + if (preparedStatements.containsKey(key)) { + throw DiskExecutorException("prepared statement '${statement.name}' already exists") + } + preparedStatements[key] = statement.target + return PrepareResult(statement.name) + } + + /** + * Ported 1:1 from [kestrel.engine.Executor.resolveExecute] - looks [ExecuteStatement.name] up, + * evaluates every argument as a CONSTANT (a bare [Evaluator] with no resolver/subquery/sequence + * access, exactly like a DEFAULT expression), and substitutes the results for every `$1`, `$2`, + * ... via the same [ParameterSubstitution] both engines share. The result is always a concrete + * [PreparableStatement] - never [ExecuteStatement] itself. + */ + fun resolveExecute(statement: ExecuteStatement): Statement { + val target = preparedStatements[statement.name.lowercase()] + ?: throw DiskExecutorException("no such prepared statement: ${statement.name}") + val required = ParameterSubstitution.parameterCount(target) + if (statement.arguments.size != required) { + throw DiskExecutorException( + "prepared statement '${statement.name}' expects $required parameter(s) but got ${statement.arguments.size}" + ) + } + val values = statement.arguments.map { Evaluator().evaluate(it) } + return ParameterSubstitution.substitute(target, values) + } + + private fun executeDeallocate(statement: DeallocateStatement): StatementResult { + if (statement.all) { + preparedStatements.clear() + return DeallocateResult(null, all = true) + } + val name = statement.name!! + if (preparedStatements.remove(name.lowercase()) == null) { + throw DiskExecutorException("no such prepared statement: $name") + } + return DeallocateResult(name, all = false) + } + + private fun executeBegin(): StatementResult { + if (explicitTxn != null) throw DiskExecutorException("a transaction is already in progress") + explicitTxn = catalog.beginTransaction() + temporaryTableSnapshots.clear() + temporaryTableSnapshots.add(snapshotTemporaryTables()) + return BeginResult + } + + private fun executeCommit(): StatementResult { + val txn = explicitTxn ?: throw DiskExecutorException("no transaction is in progress") + catalog.commit(txn) + explicitTxn = null + savepoints.clear() + temporaryTableSnapshots.clear() + return CommitResult + } + + private fun executeRollback(): StatementResult { + val txn = explicitTxn ?: throw DiskExecutorException("no transaction is in progress") + catalog.abort(txn) + explicitTxn = null + savepoints.clear() + restoreTemporaryTables(temporaryTableSnapshots[0]) + temporaryTableSnapshots.clear() + return RollbackResult + } +