diff --git a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt
index 7d77980..5f0eab6 100644
--- a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt
+++ b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt
@@ -1616,3 +1616,147 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess {
             schema.checks, schema.uniqueConstraints, schema.compositePrimaryKey
         )
 
+        val using = action.using ?: ColumnExpression(action.columnName)
+        catalog.changeColumnType(tableName, action.columnName, newColumn) { row ->
+            val value = coerceForColumn(Evaluator(schema, row).evaluate(using), newColumn)
+            if (newColumn.notNull && value is Value.NullValue) {
+                throw DiskExecutorException("NOT NULL constraint failed: $tableName.${action.columnName}")
+            }
+            val newRow = Row(row.values.toMutableList().also { it[columnIndex] = value })
+            for (check in newSchema.checks) {
+                val result = Evaluator(newSchema, newRow).evaluate(check.expression)
+                if (result is Value.BooleanValue && !result.value) {
+                    val label = check.name?.let { " $it" } ?: ""
+                    throw DiskExecutorException("CHECK constraint$label failed: $tableName")
+                }
+            }
+            value
+        }
+    }
+
+    /** Mirrors [kestrel.engine.Executor.requireUniqueConstraintName] - see that method's own doc. */
+    private fun requireUniqueConstraintName(schema: TableSchema, name: String) {
+        val taken = schema.foreignKeys.any { it.name?.equals(name, ignoreCase = true) == true } ||
+            schema.checks.any { it.name?.equals(name, ignoreCase = true) == true } ||
+            schema.uniqueConstraints.any { it.name?.equals(name, ignoreCase = true) == true }
+        if (taken) throw DiskExecutorException("constraint '$name' already exists on table ${schema.name}")
+    }
+
+    /**
+     * A NOT NULL column added to a table that already has rows must carry a DEFAULT (there is no
+     * value to backfill existing rows with otherwise) - an empty table has no such requirement.
+     * Mirrors [kestrel.engine.Executor.executeAddColumn] (same reasoning as [coerceForColumn]):
+     * duplicated rather than shared, since this executor's [DiskTable] has no equivalent of the
+     * in-memory engine's `Table.rows` to check emptiness against without a transaction of its own.
+     * GENERATED ALWAYS AS sidesteps the NOT NULL/DEFAULT requirement entirely - see
+     * [addGeneratedColumnValue].
+     */
+    private fun executeAddColumn(tableName: String, column: Column) {
+        val table = catalog.table(tableName)
+        if (column.generated != null) {
+            validateAddGeneratedColumn(table.schema, column)
+            catalog.addColumn(tableName, column) { row -> addGeneratedColumnValue(table.schema, column, row) }
+            return
+        }
+        column.default?.let { validateConstantDefault(it) }
+        val probeTxn = catalog.beginTransaction()
+        val hasRows = runInTransaction(probeTxn) { table.scan(probeTxn).isNotEmpty() }
+        if (column.notNull && column.default == null && hasRows) {
+            throw DiskExecutorException(
+                "cannot add NOT NULL column ${column.name} to table $tableName with existing rows without a DEFAULT value"
+            )
+        }
+        val defaultValue = column.default?.let { coerceForColumn(Evaluator().evaluate(it), column) } ?: Value.NullValue
+        catalog.addColumn(tableName, column) { defaultValue }
+    }
+
+    /** Mirrors [kestrel.engine.Executor.validateAddGeneratedColumn] - see that method's own doc. */
+    private fun validateAddGeneratedColumn(schema: TableSchema, column: Column) {
+        val generated = column.generated!!
+        checkColumns(generated, schema)
+        requireNoSubquery(generated)
+        val generatedNames = schema.columns.filter { it.generated != null }.map { it.name.lowercase() }.toSet()
+        requireNoGeneratedColumnReference(generated, generatedNames, column.name)
+    }
+
+    /** Mirrors [kestrel.engine.Executor.addGeneratedColumnValue] - see that method's own doc. */
+    private fun addGeneratedColumnValue(schema: TableSchema, column: Column, row: Row): Value {
+        val value = coerceForColumn(Evaluator(schema, row).evaluate(column.generated!!), column)
+        if (column.notNull && value is Value.NullValue) {
+            throw DiskExecutorException("NOT NULL constraint failed: ${schema.name}.${column.name}")
+        }
+        return value
+    }
+
+    /**
+     * A DEFAULT must be a constant expression: no column reference (there is no row to read one
+     * from when backfilling `ALTER TABLE ... ADD COLUMN`), no subquery, and no aggregate. Ported
+     * 1:1 from [kestrel.engine.Executor.validateConstantDefault] - see [checkColumns] for why this
+     * executor duplicates rather than shares this kind of validation.
+     */
+    private fun validateConstantDefault(expression: Expression) {
+        when (expression) {
+            is ColumnExpression -> throw DiskExecutorException("DEFAULT expression cannot reference a column")
+            is ScalarSubqueryExpression, is ExistsExpression, is InSubqueryExpression ->
+                throw DiskExecutorException("DEFAULT expression cannot contain a subquery")
+            is FunctionExpression -> throw DiskExecutorException("DEFAULT expression cannot contain an aggregate function")
+            is BinaryExpression -> {
+                validateConstantDefault(expression.left)
+                validateConstantDefault(expression.right)
+            }
+            is UnaryExpression -> validateConstantDefault(expression.operand)
+            is CallExpression -> expression.arguments.forEach { validateConstantDefault(it) }
+            is IsNullExpression -> validateConstantDefault(expression.operand)
+            is InExpression -> {
+                validateConstantDefault(expression.operand)
+                expression.values.forEach { validateConstantDefault(it) }
+            }
+            is BetweenExpression -> {
+                validateConstantDefault(expression.operand)
+                validateConstantDefault(expression.low)
+                validateConstantDefault(expression.high)
+            }
+            is LikeExpression -> {
+                validateConstantDefault(expression.operand)
+                validateConstantDefault(expression.pattern)
+            }
+            is CaseExpression -> {
+                expression.operand?.let { validateConstantDefault(it) }
+                expression.branches.forEach {
+                    validateConstantDefault(it.condition)
+                    validateConstantDefault(it.result)
+                }
+                expression.elseExpression?.let { validateConstantDefault(it) }
+            }
+            is CastExpression -> validateConstantDefault(expression.operand)
+            is CollateExpression -> validateConstantDefault(expression.operand)
+            is WindowExpression -> throw DiskExecutorException("DEFAULT expression cannot contain a window function")
+            is NextValueExpression ->
+                throw DiskExecutorException(
+                    "NEXTVAL cannot be used as an ADD COLUMN DEFAULT - it would give every existing row " +
+                        "the exact same value instead of a growing sequence, silently defeating the one thing NEXTVAL is for"
+                )
+            is CurrentValueExpression -> throw DiskExecutorException("DEFAULT expression cannot contain CURRVAL")
+            is MergeActionExpression -> throw DiskExecutorException("DEFAULT expression cannot contain MERGE_ACTION()")
+            is ParameterExpression -> throw DiskExecutorException("DEFAULT expression cannot contain a parameter reference")
+            is LiteralExpression -> Unit
+        }
+    }
+
+    /**
+     * Builds every row's final values first (in [newRows], none written yet) and only then validates
+     * and inserts them one at a time - not inline in a single pass - so that a self-referencing FK
+     * can see sibling rows from the same multi-row INSERT regardless of their order in the statement,
+     * exactly like [kestrel.engine.Executor.executeInsert]'s own `uncommittedSiblings = newRows`.
+     */
+    private fun executeInsert(statement: InsertStatement, triggerContext: TriggerRowContext? = null): StatementResult {
+        catalog.temporaryTableOrNull(statement.tableName)?.let { return executeTemporaryInsert(statement, it) }
+        val table = catalog.table(statement.tableName)
+        val schema = table.schema
+        return withStatementTransaction { txn ->
+            withTriggerAtomicity(schema.name, txn, triggerContext) {
+                executeInsertInTxn(statement, table, schema, txn, triggerContext)
+            }
+        }
+    }
+