feat(sql): add addUniqueConstraint and addCheckConstraint to DiskCatalog
This commit is contained in:
dualfroz committed
1 parent
dafa4f8940
commit
d6d8dadb3d
1 file changed
+165
@@ -819,3 +819,168 @@ class DiskCatalog private constructor(
|
||||
}
|
||||
}
|
||||
|
||||
/** [addForeignKey]'s counterpart for `ADD CONSTRAINT name UNIQUE (...)` - see its own doc for the validate-before-install contract. */
|
||||
fun addUniqueConstraint(tableName: String, unique: UniqueConstraint) {
|
||||
val key = tableName.lowercase()
|
||||
val row = rowOrThrow(tableName)
|
||||
val schema = row.table.schema
|
||||
val newSchema = TableSchema(
|
||||
schema.name, schema.columns, schema.foreignKeys, schema.checks, schema.uniqueConstraints + unique, 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
|
||||
}
|
||||
}
|
||||
|
||||
/** [addForeignKey]'s counterpart for `ADD CONSTRAINT name CHECK (...)` - see its own doc for the validate-before-install contract. */
|
||||
fun addCheckConstraint(tableName: String, check: CheckConstraint) {
|
||||
val key = tableName.lowercase()
|
||||
val row = rowOrThrow(tableName)
|
||||
val schema = row.table.schema
|
||||
val newSchema = TableSchema(
|
||||
schema.name, schema.columns, schema.foreignKeys, schema.checks + check, 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
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* `ALTER TABLE t DROP CONSTRAINT name [RESTRICT | CASCADE]` - mirrors
|
||||
* [kestrel.catalog.Catalog.dropConstraint]: finds a named CHECK, UNIQUE, or FOREIGN KEY
|
||||
* constraint on [tableName] (case-insensitive; a PRIMARY KEY has no name to match against) and
|
||||
* removes it. Only a UNIQUE constraint has cross-table fallout - see [dropUniqueConstraint].
|
||||
*/
|
||||
fun dropConstraint(tableName: String, name: String, cascade: Boolean) {
|
||||
val key = tableName.lowercase()
|
||||
val row = rowOrThrow(tableName)
|
||||
val schema = row.table.schema
|
||||
val foreignKey = schema.foreignKeys.find { it.name?.equals(name, ignoreCase = true) == true }
|
||||
val check = schema.checks.find { it.name?.equals(name, ignoreCase = true) == true }
|
||||
val unique = schema.uniqueConstraints.find { it.name?.equals(name, ignoreCase = true) == true }
|
||||
when {
|
||||
foreignKey != null -> {
|
||||
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
|
||||
}
|
||||
}
|
||||
check != null -> {
|
||||
val newSchema = TableSchema(
|
||||
schema.name, schema.columns, schema.foreignKeys, schema.checks - check,
|
||||
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
|
||||
}
|
||||
}
|
||||
unique != null -> dropUniqueConstraint(tableName, unique, cascade)
|
||||
else -> throw DiskCatalogException("no such constraint: $name in table $tableName")
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Mirrors [kestrel.catalog.Catalog.dropUniqueConstraint]: a FOREIGN KEY's `referencedColumns`
|
||||
* must exactly match one of the parent's enforced-unique key groups to have been accepted in the
|
||||
* first place, so dropping a UNIQUE constraint can silently invalidate any FOREIGN KEY (on this
|
||||
* table itself, via a self-reference, or on any other table) that leans on exactly this column
|
||||
* set for its uniqueness guarantee, UNLESS another remaining group still covers it. RESTRICT
|
||||
* (default) blocks the drop when a real dependent exists; CASCADE strips that FOREIGN KEY right
|
||||
* along with it, mirroring [dropTable]'s own FOREIGN KEY handling.
|
||||
*/
|
||||
private fun dropUniqueConstraint(tableName: String, unique: UniqueConstraint, cascade: Boolean) {
|
||||
val key = tableName.lowercase()
|
||||
val schema = rowOrThrow(tableName).table.schema
|
||||
val remainingUniques = schema.uniqueConstraints - unique
|
||||
val targetSet = unique.columns.map { it.lowercase() }.toSet()
|
||||
val stillEnforced = TableSchema(
|
||||
schema.name, schema.columns, schema.foreignKeys, schema.checks, remainingUniques, schema.compositePrimaryKey
|
||||
).enforcedUniqueKeyGroups().any { group -> group.map { it.lowercase() }.toSet() == targetSet }
|
||||
|
||||
fun dependsOnTarget(fk: ForeignKeyConstraint) =
|
||||
fk.referencedTable.equals(tableName, ignoreCase = true) && fk.referencedColumns.map { it.lowercase() }.toSet() == targetSet
|
||||
|
||||
val dependents = if (stillEnforced) emptyList() else rowsByName.values.filter { other -> other.table.schema.foreignKeys.any(::dependsOnTarget) }
|
||||
if (!cascade && dependents.isNotEmpty()) {
|
||||
val names = dependents.map { it.table.schema.name }.distinct()
|
||||
throw DiskCatalogException(
|
||||
"cannot drop constraint ${unique.name} because table(s) ${names.joinToString(", ")} have a " +
|
||||
"FOREIGN KEY referencing it; use CASCADE"
|
||||
)
|
||||
}
|
||||
|
||||
val txn = engine.beginTransaction()
|
||||
try {
|
||||
for (other in dependents) {
|
||||
val otherKey = other.table.schema.name.lowercase()
|
||||
val strippedForeignKeys = other.table.schema.foreignKeys.filterNot(::dependsOnTarget)
|
||||
other.table.replaceSchemaMetadata(withForeignKeys(other.table.schema, strippedForeignKeys))
|
||||
persistEntry(otherKey, txn)
|
||||
}
|
||||
// Re-fetched: a self-referencing FOREIGN KEY on tableName itself may have just been
|
||||
// rewritten above (dependents is not restricted to OTHER tables).
|
||||
val current = rowOrThrow(tableName)
|
||||
val newSchema = TableSchema(
|
||||
current.table.schema.name, current.table.schema.columns, current.table.schema.foreignKeys,
|
||||
current.table.schema.checks, current.table.schema.uniqueConstraints - unique, current.table.schema.compositePrimaryKey
|
||||
)
|
||||
current.table.replaceSchemaMetadata(newSchema)
|
||||
persistEntry(key, txn)
|
||||
engine.commit(txn)
|
||||
} catch (failure: Exception) {
|
||||
engine.abort(txn)
|
||||
throw failure
|
||||
}
|
||||
}
|
||||
|
||||
private fun renameInList(names: List<String>, oldName: String, newName: String): List<String> =
|
||||
names.map { if (it.equals(oldName, ignoreCase = true)) newName else it }
|
||||
|
||||
/** Convenience for a single-column secondary index; delegates to the general, composite-capable overload below. */
|
||||
fun createIndex(tableName: String, indexName: String, columnName: String, unique: Boolean = false): SecondaryIndexHandle =
|
||||
createIndex(tableName, indexName, listOf(columnName), unique)
|
||||
|
||||
/**
|
||||
* Creates a secondary index over [columnNames] (more than one entry for a composite,
|
||||
* multi-column index) - see [SecondaryIndexHandle]/[DiskTable] for how a composite key is
|
||||
* encoded and how uniqueness is enforced.
|
||||
*/
|
||||
fun createIndex(tableName: String, indexName: String, columnNames: List<String>, unique: Boolean = false): SecondaryIndexHandle {
|
||||
require(columnNames.isNotEmpty()) { "index must cover at least one column" }
|
||||
val key = tableName.lowercase()
|
||||
val row = rowsByName[key] ?: throw DiskCatalogException("no such table: $tableName")
|
||||
if (row.table.secondaryIndexes.any { it.name.equals(indexName, ignoreCase = true) }) {
|
||||
throw DiskCatalogException("index '$indexName' already exists")
|
||||
}
|
||||
val columnIndices = columnNames.map { columnName ->
|
||||
row.table.schema.columnIndex(columnName)
|
||||
?: throw DiskCatalogException("no such column: $columnName in table $tableName")
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user