diff --git a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt
index 28809b0..835a2b7 100644
--- a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt
+++ b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt
@@ -799,3 +799,165 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess {
         return CreateTriggerResult(statement.triggerName)
     }
 
+    private fun validateTriggerBodyStatementTarget(statement: Statement) {
+        when (statement) {
+            is InsertStatement -> { catalog.table(statement.tableName); rejectReturning(statement.returning) }
+            is UpdateStatement -> { catalog.table(statement.tableName); rejectReturning(statement.returning) }
+            is DeleteStatement -> { catalog.table(statement.tableName); rejectReturning(statement.returning) }
+            else -> throw DiskExecutorException("trigger body only supports INSERT, UPDATE, or DELETE statements")
+        }
+    }
+
+    /** Ported 1:1 from [kestrel.engine.Executor.rejectReturning]: a trigger body statement's own [runTriggerBody] discards whatever [StatementResult] it returns, so a RETURNING clause there would compute output nobody could ever see. */
+    private fun rejectReturning(returning: Projection?) {
+        if (returning != null) {
+            throw DiskExecutorException("RETURNING is not allowed in a trigger body statement - its result would never be visible to anything")
+        }
+    }
+
+    /** Ported 1:1 from [kestrel.engine.Executor.validateTriggerRowReferences]. */
+    private fun validateTriggerRowReferences(expression: Expression, event: TriggerEvent, schema: TableSchema) {
+        when (expression) {
+            is ColumnExpression -> {
+                val qualifier = expression.qualifier
+                    ?: throw DiskExecutorException("column '${expression.name}' in trigger WHEN clause must be qualified with NEW or OLD")
+                val isNew = qualifier.equals("new", ignoreCase = true)
+                val isOld = qualifier.equals("old", ignoreCase = true)
+                if (!isNew && !isOld) throw DiskExecutorException("no such table or alias: $qualifier")
+                val availableForEvent = when (event) {
+                    TriggerEvent.Insert -> isNew
+                    TriggerEvent.Delete -> isOld
+                    TriggerEvent.Update -> true
+                }
+                if (!availableForEvent) {
+                    throw DiskExecutorException("${if (isNew) "NEW" else "OLD"} is not available in a trigger ON ${event.name.uppercase()}")
+                }
+                if (schema.columnIndex(expression.name) == null) {
+                    throw DiskExecutorException("no such column: ${expression.name}")
+                }
+            }
+            is BinaryExpression -> {
+                validateTriggerRowReferences(expression.left, event, schema)
+                validateTriggerRowReferences(expression.right, event, schema)
+            }
+            is UnaryExpression -> validateTriggerRowReferences(expression.operand, event, schema)
+            is CallExpression -> expression.arguments.forEach { validateTriggerRowReferences(it, event, schema) }
+            is IsNullExpression -> validateTriggerRowReferences(expression.operand, event, schema)
+            is InExpression -> {
+                validateTriggerRowReferences(expression.operand, event, schema)
+                expression.values.forEach { validateTriggerRowReferences(it, event, schema) }
+            }
+            is BetweenExpression -> {
+                validateTriggerRowReferences(expression.operand, event, schema)
+                validateTriggerRowReferences(expression.low, event, schema)
+                validateTriggerRowReferences(expression.high, event, schema)
+            }
+            is LikeExpression -> {
+                validateTriggerRowReferences(expression.operand, event, schema)
+                validateTriggerRowReferences(expression.pattern, event, schema)
+            }
+            is CaseExpression -> {
+                expression.operand?.let { validateTriggerRowReferences(it, event, schema) }
+                expression.branches.forEach {
+                    validateTriggerRowReferences(it.condition, event, schema)
+                    validateTriggerRowReferences(it.result, event, schema)
+                }
+                expression.elseExpression?.let { validateTriggerRowReferences(it, event, schema) }
+            }
+            is CastExpression -> validateTriggerRowReferences(expression.operand, event, schema)
+            is CollateExpression -> validateTriggerRowReferences(expression.operand, event, schema)
+            is LiteralExpression -> Unit
+            is FunctionExpression, is WindowExpression, is ScalarSubqueryExpression,
+            is ExistsExpression, is InSubqueryExpression,
+            is NextValueExpression, is CurrentValueExpression, is MergeActionExpression, is ParameterExpression ->
+                error("unreachable: banned by requireNoSubquery")
+        }
+    }
+
+    private fun executeDropTrigger(statement: DropTriggerStatement): StatementResult {
+        catalog.dropTrigger(statement.triggerName)
+        return DropTriggerResult(statement.triggerName)
+    }
+
+    private fun executeCreateTable(statement: CreateTableStatement): StatementResult {
+        if (statement.temporary) return executeCreateTemporaryTable(statement)
+        if (statement.columns.count { it.primaryKey } > 1 ||
+            (statement.compositePrimaryKey.isNotEmpty() && statement.columns.any { it.primaryKey })
+        ) {
+            throw DiskExecutorException("table ${statement.tableName} declares more than one primary key")
+        }
+        validateUniqueDefinitions(statement)
+        validateCheckDefinitions(statement)
+        val columns = applyCompositePrimaryKey(statement)
+        validateForeignKeyDefinitions(statement, columns)
+        for (column in columns) {
+            when (val default = column.default) {
+                is NextValueExpression -> validateNextValueDefault(default, column)
+                is CurrentValueExpression -> throw DiskExecutorException("DEFAULT expression cannot contain CURRVAL")
+                else -> Unit
+            }
+        }
+        validateGeneratedDefinitions(columns)
+        catalog.createTable(
+            statement.tableName, columns, statement.compositePrimaryKey, statement.uniqueConstraints,
+            statement.checks, statement.foreignKeys
+        )
+        return CreateTableResult(statement.tableName)
+    }
+
+    /**
+     * `CREATE TEMP TABLE`/`CREATE TEMPORARY TABLE` - the disk-backed executor's counterpart of
+     * [executeCreateTable], but building a [kestrel.storage.Table] via [DiskCatalog.createTemporaryTable]
+     * instead of a [DiskTable]: a real, live table for the rest of this process's lifetime (INSERT/
+     * SELECT/UPDATE/DELETE, WHERE/JOIN/ORDER BY/subqueries, NOT NULL/PRIMARY KEY/UNIQUE/CHECK/DEFAULT/
+     * GENERATED all enforced exactly like a permanent table - see [executeTemporaryInsert]/
+     * [executeTemporaryUpdate]/[executeTemporaryDelete] and [resolveFromSource]), simply never
+     * touching the pager/WAL and never surviving a reopen. FOREIGN KEY is rejected outright rather
+     * than silently ignored: [DiskForeignKeyEnforcer] and its cascade bookkeeping are wired to
+     * [DiskTable]/[DiskCatalog]'s directory, which a temp table deliberately has no entry in, so
+     * honoring the constraint would need a parallel enforcement path this session scoped out - a
+     * named, documented scope-cut rather than a constraint that looks declared but is never checked.
+     */
+    private fun executeCreateTemporaryTable(statement: CreateTableStatement): StatementResult {
+        if (statement.foreignKeys.isNotEmpty()) {
+            throw DiskExecutorException("FOREIGN KEY on a temporary table is not supported by the disk-backed executor yet")
+        }
+        if (statement.columns.count { it.primaryKey } > 1 ||
+            (statement.compositePrimaryKey.isNotEmpty() && statement.columns.any { it.primaryKey })
+        ) {
+            throw DiskExecutorException("table ${statement.tableName} declares more than one primary key")
+        }
+        validateUniqueDefinitions(statement)
+        validateCheckDefinitions(statement)
+        val columns = applyCompositePrimaryKey(statement)
+        for (column in columns) {
+            when (val default = column.default) {
+                is NextValueExpression -> validateNextValueDefault(default, column)
+                is CurrentValueExpression -> throw DiskExecutorException("DEFAULT expression cannot contain CURRVAL")
+                else -> Unit
+            }
+        }
+        validateGeneratedDefinitions(columns)
+        val schema = TableSchema(
+            statement.tableName, columns,
+            checks = statement.checks,
+            uniqueConstraints = statement.uniqueConstraints,
+            compositePrimaryKey = statement.compositePrimaryKey,
+            temporary = true
+        )
+        catalog.createTemporaryTable(schema)
+        return CreateTableResult(statement.tableName)
+    }
+
+    /**
+     * Ported 1:1 from [kestrel.engine.Executor.validateNextValueDefault]: checked statically rather
+     * than by evaluating the DEFAULT for real, since a bare `CREATE TABLE` must not itself consume a
+     * sequence value before a single row exists.
+     */
+    private fun validateNextValueDefault(default: NextValueExpression, column: Column) {
+        catalog.sequence(default.sequenceName)
+        if (column.type != ColumnType.Integer) {
+            throw DiskExecutorException("NEXTVAL can only be used as a DEFAULT for an INTEGER column, but ${column.name} is ${column.type.displayName}")
+        }
+    }
+