feat(sql): add renameColumn and renameTable to DiskCatalog
This commit is contained in:
dualfroz committed
1 parent
3972e07d37
commit
dafa4f8940
1 file changed
+167
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user