diff --git a/src/main/kotlin/kestrel/store/sql/DiskCatalog.kt b/src/main/kotlin/kestrel/store/sql/DiskCatalog.kt
index b562bca..b234dfc 100644
--- a/src/main/kotlin/kestrel/store/sql/DiskCatalog.kt
+++ b/src/main/kotlin/kestrel/store/sql/DiskCatalog.kt
@@ -491,3 +491,164 @@ class DiskCatalog private constructor(
         }
     }
 
+    /**
+     * Appends a new column (already validated by [DiskExecutor]) to every existing row - mirrors
+     * [kestrel.catalog.Catalog.addColumn]. [valueForRow] backfills every pre-existing row (each
+     * called with that row's OLD, pre-append values, then [DiskTable.rewriteRows]-transformed to
+     * carry the result) - a constant lambda for a plain DEFAULT, or a fresh per-row expression
+     * evaluation for GENERATED ALWAYS AS; a table with zero rows just gets the new schema with
+     * nothing to rewrite.
+     */
+    fun addColumn(tableName: String, column: Column, valueForRow: (Row) -> Value) {
+        val row = rowOrThrow(tableName)
+        val schema = row.table.schema
+        if (schema.columnIndex(column.name) != null) {
+            throw DiskCatalogException("duplicate column name: ${column.name}")
+        }
+        val newSchema = TableSchema(
+            schema.name, schema.columns + column, schema.foreignKeys, schema.checks, schema.uniqueConstraints, schema.compositePrimaryKey
+        )
+        val txn = engine.beginTransaction()
+        try {
+            row.table.rewriteRows(txn, newSchema) { existingRow -> Row(existingRow.values + valueForRow(existingRow)) }
+            persistEntry(tableName.lowercase(), txn)
+            engine.commit(txn)
+        } catch (failure: Exception) {
+            engine.abort(txn)
+            throw failure
+        }
+    }
+
+    /**
+     * `ALTER TABLE t ALTER COLUMN col TYPE ...` - mirrors [kestrel.catalog.Catalog.changeColumnType],
+     * built on [DiskTable.rewriteRows] exactly like [addColumn] rather than [addColumn]'s own
+     * `List<Value> -> Value` append shape, since this replaces an EXISTING position instead of
+     * appending a new one. Unlike the in-memory engine's [kestrel.engine.Executor], which validates
+     * every row up front and only ever calls its own `Catalog.changeColumnType` once every row is
+     * already known-good, [valueForRow] here is BOTH the value-computation AND the NOT NULL/CHECK
+     * validation step (see [DiskExecutor.executeChangeColumnType]) - a failure partway through
+     * throws out of [DiskTable.rewriteRows]'s loop, and the `catch` below aborts the transaction,
+     * unwinding every row already rewritten. Same "validate lazily, rely on abort to undo a partial
+     * per-row rewrite" contract [addColumn] already leans on for a NOT NULL `GENERATED ALWAYS AS`
+     * backfill.
+     */
+    fun changeColumnType(tableName: String, columnName: String, newColumn: Column, valueForRow: (Row) -> Value) {
+        val row = rowOrThrow(tableName)
+        val schema = row.table.schema
+        val columnIndex = schema.columnIndex(columnName)
+            ?: throw DiskCatalogException("no such column: $columnName in table $tableName")
+        val newColumns = schema.columns.toMutableList().also { it[columnIndex] = newColumn }
+        val newSchema = TableSchema(
+            schema.name, newColumns, schema.foreignKeys, schema.checks, schema.uniqueConstraints, schema.compositePrimaryKey
+        )
+        val txn = engine.beginTransaction()
+        try {
+            row.table.rewriteRows(txn, newSchema) { existingRow ->
+                val values = existingRow.values.toMutableList()
+                values[columnIndex] = valueForRow(existingRow)
+                Row(values)
+            }
+            persistEntry(tableName.lowercase(), txn)
+            engine.commit(txn)
+        } catch (failure: Exception) {
+            engine.abort(txn)
+            throw failure
+        }
+    }
+
+    /**
+     * RESTRICT (default) rejects the drop if [columnName] is used by another table's FOREIGN KEY,
+     * this table's own FOREIGN KEY, a UNIQUE constraint, a CHECK constraint, or a secondary index -
+     * CASCADE removes all of those alongside the column itself. Mirrors
+     * [kestrel.catalog.Catalog.dropColumn], including that a PRIMARY KEY column can never be
+     * dropped, with or without CASCADE.
+     */
+    fun dropColumn(tableName: String, columnName: String, cascade: Boolean) {
+        val key = tableName.lowercase()
+        val row = rowOrThrow(tableName)
+        val schema = row.table.schema
+        schema.columnIndex(columnName) ?: throw DiskCatalogException("no such column: $columnName in table $tableName")
+        if (schema.columns.size == 1) {
+            throw DiskCatalogException("cannot drop $columnName: table $tableName would be left with no columns")
+        }
+        if (schema.primaryKeyColumns.any { it.equals(columnName, ignoreCase = true) }) {
+            throw DiskCatalogException("cannot drop column $columnName because it is part of the PRIMARY KEY of $tableName")
+        }
+
+        // Mirrors kestrel.catalog.Catalog.dropColumn: a GENERATED column can only ever read this
+        // table's own non-GENERATED columns, so dropping columnName orphans any GENERATED column
+        // depending on it exactly like a CHECK constraint referencing it - CASCADE drops it too,
+        // RESTRICT blocks the whole drop instead. GENERATED can never be a PRIMARY KEY column, so
+        // there is no "cascaded into dropping the PRIMARY KEY" case to guard against here.
+        val ownGeneratedDependents = schema.columns.filter { column ->
+            !column.name.equals(columnName, ignoreCase = true) &&
+                column.generated?.let { CheckExpressionColumns.references(it, columnName) } == true
+        }
+        val droppedNames = (ownGeneratedDependents.map { it.name } + columnName).map { it.lowercase() }.toSet()
+        val droppedIndices = schema.columns.withIndex()
+            .filter { (_, column) -> column.name.lowercase() in droppedNames }
+            .map { it.index }
+            .toSet()
+
+        val referencingForeignKeys = rowsByName.entries.filter { (otherKey, other) ->
+            otherKey != key && other.table.schema.foreignKeys.any { fk ->
+                fk.referencedTable.equals(tableName, ignoreCase = true) && fk.referencedColumns.any { it.lowercase() in droppedNames }
+            }
+        }
+        val ownForeignKeys = schema.foreignKeys.filter { fk -> fk.columns.any { it.lowercase() in droppedNames } }
+        val ownUniqueConstraints = schema.uniqueConstraints.filter { unique -> unique.columns.any { it.lowercase() in droppedNames } }
+        val ownChecks = schema.checks.filter { check -> droppedNames.any { name -> CheckExpressionColumns.references(check.expression, name) } }
+        val dependentIndexes = row.table.secondaryIndexes.filter { it.columnNames.any { c -> c.lowercase() in droppedNames } }
+
+        if (!cascade) {
+            val reasons = mutableListOf<String>()
+            if (ownGeneratedDependents.isNotEmpty()) {
+                reasons.add(
+                    "column(s) ${ownGeneratedDependents.joinToString(", ") { it.name }} are GENERATED columns that depend on it"
+                )
+            }
+            if (referencingForeignKeys.isNotEmpty()) {
+                reasons.add(
+                    "table(s) ${referencingForeignKeys.joinToString(", ") { it.value.table.schema.name }} have a FOREIGN KEY referencing it"
+                )
+            }
+            if (ownForeignKeys.isNotEmpty()) reasons.add("it is used by one of its own FOREIGN KEY constraints")
+            if (ownUniqueConstraints.isNotEmpty()) reasons.add("it is used by a UNIQUE constraint")
+            if (ownChecks.isNotEmpty()) reasons.add("it is used by a CHECK constraint")
+            if (dependentIndexes.isNotEmpty()) {
+                reasons.add("index(es) ${dependentIndexes.joinToString(", ") { it.name }} are defined on it")
+            }
+            if (reasons.isNotEmpty()) {
+                throw DiskCatalogException(
+                    "cannot drop column $columnName from $tableName because ${reasons.joinToString("; ")}; use CASCADE"
+                )
+            }
+        }
+
+        val newColumns = schema.columns.filterNot { it.name.lowercase() in droppedNames }
+        val newForeignKeys = schema.foreignKeys - ownForeignKeys.toSet()
+        val newUniqueConstraints = schema.uniqueConstraints - ownUniqueConstraints.toSet()
+        val newChecks = schema.checks - ownChecks.toSet()
+        val newSchema = TableSchema(schema.name, newColumns, newForeignKeys, newChecks, newUniqueConstraints, schema.compositePrimaryKey)
+
+        val txn = engine.beginTransaction()
+        try {
+            for (indexName in dependentIndexes.map { it.name }) row.table.dropSecondaryIndexByName(indexName, txn)
+            row.table.rewriteRows(txn, newSchema) { existingRow ->
+                Row(existingRow.values.filterIndexed { valueIndex, _ -> valueIndex !in droppedIndices })
+            }
+            persistEntry(key, txn)
+            for ((otherKey, other) in referencingForeignKeys) {
+                val updatedForeignKeys = other.table.schema.foreignKeys.filterNot { fk ->
+                    fk.referencedTable.equals(tableName, ignoreCase = true) && fk.referencedColumns.any { it.lowercase() in droppedNames }
+                }
+                other.table.replaceSchemaMetadata(withForeignKeys(other.table.schema, updatedForeignKeys))
+                persistEntry(otherKey, txn)
+            }
+            engine.commit(txn)
+        } catch (failure: Exception) {
+            engine.abort(txn)
+            throw failure
+        }
+    }
+