feat(sql): add snapshotTemporaryTables and restoreTemporaryTables

src/main/kotlin/kestrel/store/sql/DiskExecutor.kt

@@ -314,3 +314,162 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess {
 
     val preparedStatementNames: List<String> 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<Pair<String, Long>>()
+
+    /**
+     * 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<Map<Table, List<Row>>>()
+
+    private fun snapshotTemporaryTables(): Map<Table, List<Row>> =
+        catalog.temporaryTableValues.associateWith { it.rows.toList() }
+
+    private fun restoreTemporaryTables(snapshot: Map<Table, List<Row>>) {
+        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
+    }
+