diff --git a/src/main/kotlin/kestrel/store/sql/DiskCatalog.kt b/src/main/kotlin/kestrel/store/sql/DiskCatalog.kt
index b234dfc..50b708a 100644
--- a/src/main/kotlin/kestrel/store/sql/DiskCatalog.kt
+++ b/src/main/kotlin/kestrel/store/sql/DiskCatalog.kt
@@ -652,3 +652,170 @@ class DiskCatalog private constructor(
         }
     }
 
+    /**
+     * Always fully propagates rather than being rejected the way [dropColumn] can be - mirrors
+     * [kestrel.catalog.Catalog.renameColumn]: this table's own PRIMARY KEY/FOREIGN KEY/UNIQUE/CHECK
+     * definitions, any other table's FOREIGN KEY referencing this column, and any secondary index
+     * defined on it (see [DiskTable.renameColumnAndIndexes]) all get the new name.
+     */
+    fun renameColumn(tableName: String, oldName: String, newName: String) {
+        val key = tableName.lowercase()
+        val row = rowOrThrow(tableName)
+        val schema = row.table.schema
+        if (schema.columnIndex(oldName) == null) {
+            throw DiskCatalogException("no such column: $oldName in table $tableName")
+        }
+        if (!oldName.equals(newName, ignoreCase = true) && schema.columnIndex(newName) != null) {
+            throw DiskCatalogException("column $newName already exists in table $tableName")
+        }
+
+        val newColumns = schema.columns.map { column ->
+            val renamed = if (column.name.equals(oldName, ignoreCase = true)) column.copy(name = newName) else column
+            // A GENERATED column can only ever reference this table's own columns, so renaming one
+            // of them must rewrite that reference too - mirrors kestrel.catalog.Catalog.renameColumn.
+            renamed.generated?.let { renamed.copy(generated = CheckExpressionColumns.rename(it, oldName, newName)) } ?: renamed
+        }
+        val newForeignKeys = schema.foreignKeys.map { fk ->
+            val selfReference = fk.referencedTable.equals(tableName, ignoreCase = true)
+            fk.copy(
+                columns = renameInList(fk.columns, oldName, newName),
+                referencedColumns = if (selfReference) renameInList(fk.referencedColumns, oldName, newName) else fk.referencedColumns
+            )
+        }
+        val newUniqueConstraints = schema.uniqueConstraints.map { it.copy(columns = renameInList(it.columns, oldName, newName)) }
+        val newChecks = schema.checks.map { CheckConstraint(it.name, CheckExpressionColumns.rename(it.expression, oldName, newName)) }
+        val newCompositePrimaryKey = renameInList(schema.compositePrimaryKey, oldName, newName)
+        val newSchema = TableSchema(schema.name, newColumns, newForeignKeys, newChecks, newUniqueConstraints, newCompositePrimaryKey)
+
+        val affectedOthers = rowsByName.entries.filter { (otherKey, other) ->
+            otherKey != key && other.table.schema.foreignKeys.any { fk ->
+                fk.referencedTable.equals(tableName, ignoreCase = true) && fk.referencedColumns.any { it.equals(oldName, ignoreCase = true) }
+            }
+        }
+
+        val txn = engine.beginTransaction()
+        try {
+            row.table.renameColumnAndIndexes(newSchema, oldName, newName)
+            persistEntry(key, txn)
+            for ((otherKey, other) in affectedOthers) {
+                val updatedForeignKeys = other.table.schema.foreignKeys.map { fk ->
+                    if (fk.referencedTable.equals(tableName, ignoreCase = true)) {
+                        fk.copy(referencedColumns = renameInList(fk.referencedColumns, oldName, newName))
+                    } else {
+                        fk
+                    }
+                }
+                other.table.replaceSchemaMetadata(withForeignKeys(other.table.schema, updatedForeignKeys))
+                persistEntry(otherKey, txn)
+            }
+            engine.commit(txn)
+        } catch (failure: Exception) {
+            engine.abort(txn)
+            throw failure
+        }
+    }
+
+    /**
+     * Renames a table and propagates the identifier everywhere it is used by name - mirrors
+     * [kestrel.catalog.Catalog.renameTable]: this table's own FOREIGN KEY(s) that self-reference it,
+     * any other table's FOREIGN KEY pointing at it, every view whose defining query mentions it
+     * anywhere (see [RelationRename]), and every trigger defined ON it (just the [TriggerDefinition.
+     * tableName] field, re-rendered and re-persisted - trigger body statements are NOT rewritten,
+     * same documented scope-cut as [kestrel.catalog.Catalog.renameTable]). Unlike the in-memory
+     * engine there is no separate global index map to fix up - every secondary index lives inside its
+     * own [DiskTable], keyed by column name, which a table rename never touches.
+     */
+    fun renameTable(oldName: String, newName: String) {
+        val oldKey = oldName.lowercase()
+        val row = rowOrThrow(oldName)
+        val newKey = newName.lowercase()
+        if (!oldName.equals(newName, ignoreCase = true)) {
+            if (rowsByName.containsKey(newKey)) throw DiskCatalogException("table '$newName' already exists")
+            if (viewsByName.containsKey(newKey)) throw DiskCatalogException("a view named '$newName' already exists")
+        }
+
+        val schema = row.table.schema
+        val selfReferencing = schema.foreignKeys.any { it.referencedTable.equals(oldName, ignoreCase = true) }
+        val ownNewForeignKeys = if (selfReferencing) {
+            schema.foreignKeys.map { fk -> if (fk.referencedTable.equals(oldName, ignoreCase = true)) fk.copy(referencedTable = newName) else fk }
+        } else {
+            schema.foreignKeys
+        }
+        val newSchema = TableSchema(newName, schema.columns, ownNewForeignKeys, schema.checks, schema.uniqueConstraints, schema.compositePrimaryKey)
+
+        val affectedOthers = rowsByName.entries.filter { (otherKey, other) ->
+            otherKey != oldKey && other.table.schema.foreignKeys.any { it.referencedTable.equals(oldName, ignoreCase = true) }
+        }
+        val affectedViews = viewsByName.entries.filter { (_, view) -> DependencyAnalysis.referencesRelation(view.query, oldName) }
+        val affectedTriggers = triggersByName.entries.filter { (_, trigger) -> trigger.definition.tableName.equals(oldName, ignoreCase = true) }
+
+        val txn = engine.beginTransaction()
+        try {
+            row.table.replaceSchemaMetadata(newSchema)
+            if (oldKey != newKey) {
+                rowsByName.remove(oldKey)
+                row.keyBox.value = newKey
+                rowsByName[newKey] = row
+            }
+            persistEntry(newKey, txn)
+
+            for ((otherKey, other) in affectedOthers) {
+                val updatedForeignKeys = other.table.schema.foreignKeys.map { fk ->
+                    if (fk.referencedTable.equals(oldName, ignoreCase = true)) fk.copy(referencedTable = newName) else fk
+                }
+                other.table.replaceSchemaMetadata(withForeignKeys(other.table.schema, updatedForeignKeys))
+                persistEntry(otherKey, txn)
+            }
+
+            for ((viewKey, viewRow) in affectedViews) {
+                val renamedQuery = RelationRename.rename(viewRow.query, oldName, newName)
+                val newEntry = ViewDirectoryEntry(
+                    viewRow.entry.viewName, SqlWriter.write(CreateViewStatement(viewRow.entry.viewName, renamedQuery, orReplace = false))
+                )
+                val rid = directory.update(txn, viewRow.rid, taggedView(newEntry))
+                viewsByName[viewKey] = ViewRow(rid, newEntry, renamedQuery)
+            }
+
+            for ((triggerKey, triggerRow) in affectedTriggers) {
+                val renamedDefinition = triggerRow.definition.copy(tableName = newName)
+                val newEntry = TriggerDirectoryEntry(triggerRow.entry.triggerName, SqlWriter.triggerText(renamedDefinition))
+                val rid = directory.update(txn, triggerRow.rid, taggedTrigger(newEntry))
+                triggersByName[triggerKey] = TriggerRow(rid, newEntry, renamedDefinition)
+            }
+
+            engine.commit(txn)
+        } catch (failure: Exception) {
+            engine.abort(txn)
+            throw failure
+        }
+    }
+
+    private fun withForeignKeys(schema: TableSchema, foreignKeys: List<ForeignKeyConstraint>): TableSchema = TableSchema(
+        schema.name, schema.columns, foreignKeys, schema.checks, schema.uniqueConstraints, schema.compositePrimaryKey
+    )
+
+    /**
+     * `ALTER TABLE t ADD CONSTRAINT name FOREIGN KEY (...) REFERENCES ...` - mirrors
+     * [kestrel.catalog.Catalog.addForeignKey]: [kestrel.store.sql.DiskExecutor.executeAddConstraint]
+     * has already validated every existing row against [foreignKey] before calling this, so installing
+     * it is pure schema metadata - [DiskTable.replaceSchemaMetadata] plus a re-persisted directory
+     * entry, no row rewrite (row shape is unchanged) and no index to build.
+     */
+    fun addForeignKey(tableName: String, foreignKey: ForeignKeyConstraint) {
+        val key = tableName.lowercase()
+        val row = rowOrThrow(tableName)
+        val schema = row.table.schema
+        val newSchema = TableSchema(
+            schema.name, schema.columns, schema.foreignKeys + foreignKey, schema.checks, schema.uniqueConstraints, schema.compositePrimaryKey
+        )
+        val txn = engine.beginTransaction()
+        try {
+            row.table.replaceSchemaMetadata(newSchema)
+            persistEntry(key, txn)
+            engine.commit(txn)
+        } catch (failure: Exception) {
+            engine.abort(txn)
+            throw failure
+        }
+    }
+