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