feat(sql): add executeAddUniqueConstraint
This commit is contained in:
dualfroz committed
1 parent
ec13738fca
commit
faad477a32
1 file changed
+162
@@ -1454,3 +1454,165 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess {
|
|||||||
catalog.addCheckConstraint(tableName, check)
|
catalog.addCheckConstraint(tableName, check)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* `ALTER TABLE t ADD CONSTRAINT name UNIQUE (...)` - mirrors
|
||||||
|
* [kestrel.engine.Executor.executeAddUniqueConstraint]: every existing row is checked for a
|
||||||
|
* duplicate combination of [unique]'s columns (folded through [collationFoldedUniquenessKey], NULL
|
||||||
|
* exempt per standard SQL) under its own probe transaction before the constraint is installed.
|
||||||
|
*/
|
||||||
|
private fun executeAddUniqueConstraint(tableName: String, unique: UniqueConstraint) {
|
||||||
|
val table = catalog.table(tableName)
|
||||||
|
val schema = table.schema
|
||||||
|
unique.name?.let { requireUniqueConstraintName(schema, it) }
|
||||||
|
for (columnName in unique.columns) {
|
||||||
|
if (schema.columnIndex(columnName) == null) {
|
||||||
|
throw DiskExecutorException("UNIQUE constraint refers to unknown column: $columnName")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
val columnIndices = unique.columns.map { schema.columnIndex(it)!! }
|
||||||
|
val probeTxn = catalog.beginTransaction()
|
||||||
|
runInTransaction(probeTxn) {
|
||||||
|
val seen = HashSet<List<Value>>()
|
||||||
|
for ((_, row) in table.scan(probeTxn)) {
|
||||||
|
val rawKey = columnIndices.map { row.values[it] }
|
||||||
|
if (rawKey.any { it is Value.NullValue }) continue
|
||||||
|
if (!seen.add(collationFoldedUniquenessKey(schema, columnIndices, rawKey))) {
|
||||||
|
val label = unique.name?.let { " $it" } ?: ""
|
||||||
|
throw DiskExecutorException(
|
||||||
|
"UNIQUE constraint$label failed: an existing row combination in " +
|
||||||
|
"$tableName(${unique.columns.joinToString(", ")}) is already duplicated"
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
catalog.addUniqueConstraint(tableName, unique)
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* `ALTER TABLE t ADD CONSTRAINT name FOREIGN KEY (...) REFERENCES ...` - mirrors
|
||||||
|
* [kestrel.engine.Executor.executeAddForeignKeyConstraint]: validated the same way
|
||||||
|
* [validateForeignKeyDefinitions] validates one at CREATE TABLE time, PLUS every existing row of
|
||||||
|
* [tableName] is checked against it via [DiskForeignKeyEnforcer.validateRow] (under its own probe
|
||||||
|
* transaction) against a throwaway [TableSchema] carrying [foreignKey] alongside every FOREIGN KEY
|
||||||
|
* [tableName] already has, so a self-referencing FOREIGN KEY can find its own sibling rows.
|
||||||
|
*/
|
||||||
|
private fun executeAddForeignKeyConstraint(tableName: String, foreignKey: ForeignKeyConstraint) {
|
||||||
|
val table = catalog.table(tableName)
|
||||||
|
val schema = table.schema
|
||||||
|
foreignKey.name?.let { requireUniqueConstraintName(schema, it) }
|
||||||
|
if (foreignKey.columns.size != foreignKey.referencedColumns.size) {
|
||||||
|
throw DiskExecutorException(
|
||||||
|
"FOREIGN KEY (${foreignKey.columns.joinToString(", ")}) column count does not match " +
|
||||||
|
"REFERENCES ${foreignKey.referencedTable}(${foreignKey.referencedColumns.joinToString(", ")})"
|
||||||
|
)
|
||||||
|
}
|
||||||
|
for (columnName in foreignKey.columns) {
|
||||||
|
if (schema.columnIndex(columnName) == null) {
|
||||||
|
throw DiskExecutorException("FOREIGN KEY refers to unknown column: $columnName")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
val selfReference = foreignKey.referencedTable.equals(tableName, ignoreCase = true)
|
||||||
|
val referencedSchema = if (selfReference) {
|
||||||
|
schema
|
||||||
|
} else {
|
||||||
|
catalog.tableOrNull(foreignKey.referencedTable)?.schema
|
||||||
|
?: throw DiskExecutorException("FOREIGN KEY references unknown table: ${foreignKey.referencedTable}")
|
||||||
|
}
|
||||||
|
for (columnName in foreignKey.referencedColumns) {
|
||||||
|
if (referencedSchema.column(columnName) == null) {
|
||||||
|
throw DiskExecutorException("FOREIGN KEY references unknown column: ${foreignKey.referencedTable}.$columnName")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (!referencedSchema.hasEnforcedUniqueness(foreignKey.referencedColumns)) {
|
||||||
|
throw DiskExecutorException(
|
||||||
|
"FOREIGN KEY must reference a PRIMARY KEY or UNIQUE column(s), but " +
|
||||||
|
"${foreignKey.referencedTable}(${foreignKey.referencedColumns.joinToString(", ")}) is not one"
|
||||||
|
)
|
||||||
|
}
|
||||||
|
val tempSchema = TableSchema(
|
||||||
|
schema.name, schema.columns, schema.foreignKeys + foreignKey, schema.checks, schema.uniqueConstraints, schema.compositePrimaryKey
|
||||||
|
)
|
||||||
|
val probeTxn = catalog.beginTransaction()
|
||||||
|
runInTransaction(probeTxn) {
|
||||||
|
for ((_, row) in table.scan(probeTxn)) {
|
||||||
|
DiskForeignKeyEnforcer.validateRow(catalog, probeTxn, tempSchema, row.values)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
catalog.addForeignKey(tableName, foreignKey)
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Mirrors [kestrel.engine.Executor.executeChangeColumnType] - same RESTRICT-only dependency
|
||||||
|
* checks (PRIMARY KEY/FOREIGN KEY on either side/UNIQUE constraint/secondary index/GENERATED,
|
||||||
|
* all rejected outright rather than threading a type change through them) and the same DEFAULT
|
||||||
|
* compatibility pre-check. The one structural difference: [DiskCatalog.changeColumnType]'s
|
||||||
|
* `valueForRow` callback runs DURING the real per-row rewrite (via [DiskTable.rewriteRows]), not
|
||||||
|
* before it - see that method's own doc for why a NOT NULL/CHECK failure discovered mid-rewrite
|
||||||
|
* is still safe (transaction abort unwinds it), the same contract a NOT NULL `GENERATED ALWAYS
|
||||||
|
* AS` backfill already relies on.
|
||||||
|
*/
|
||||||
|
private fun executeChangeColumnType(tableName: String, action: ChangeColumnType) {
|
||||||
|
val table = catalog.table(tableName)
|
||||||
|
val schema = table.schema
|
||||||
|
val columnIndex = schema.columnIndex(action.columnName)
|
||||||
|
?: throw DiskExecutorException("no such column: ${action.columnName} in table $tableName")
|
||||||
|
val column = schema.columns[columnIndex]
|
||||||
|
if (column.generated != null) {
|
||||||
|
throw DiskExecutorException(
|
||||||
|
"cannot change the type of GENERATED column ${action.columnName} in $tableName; drop and re-add it instead"
|
||||||
|
)
|
||||||
|
}
|
||||||
|
if (schema.primaryKeyColumns.any { it.equals(action.columnName, ignoreCase = true) }) {
|
||||||
|
throw DiskExecutorException(
|
||||||
|
"cannot change the type of column ${action.columnName} because it is part of the PRIMARY KEY of $tableName"
|
||||||
|
)
|
||||||
|
}
|
||||||
|
if (schema.foreignKeys.any { fk -> fk.columns.any { it.equals(action.columnName, ignoreCase = true) } }) {
|
||||||
|
throw DiskExecutorException(
|
||||||
|
"cannot change the type of column ${action.columnName} because it is used by one of ${tableName}'s own " +
|
||||||
|
"FOREIGN KEY constraints; drop the constraint first"
|
||||||
|
)
|
||||||
|
}
|
||||||
|
val referencingTables = catalog.tableNames.filter { other ->
|
||||||
|
!other.equals(tableName, ignoreCase = true) &&
|
||||||
|
catalog.tableOrNull(other)?.schema?.foreignKeys?.any { fk ->
|
||||||
|
fk.referencedTable.equals(tableName, ignoreCase = true) &&
|
||||||
|
fk.referencedColumns.any { it.equals(action.columnName, ignoreCase = true) }
|
||||||
|
} == true
|
||||||
|
}
|
||||||
|
if (referencingTables.isNotEmpty()) {
|
||||||
|
throw DiskExecutorException(
|
||||||
|
"cannot change the type of column ${action.columnName} because table(s) ${referencingTables.joinToString(", ")} " +
|
||||||
|
"have a FOREIGN KEY referencing it; drop the constraint first"
|
||||||
|
)
|
||||||
|
}
|
||||||
|
if (schema.uniqueConstraints.any { unique -> unique.columns.any { it.equals(action.columnName, ignoreCase = true) } }) {
|
||||||
|
throw DiskExecutorException(
|
||||||
|
"cannot change the type of column ${action.columnName} because it is used by a UNIQUE constraint of $tableName; " +
|
||||||
|
"drop the constraint first"
|
||||||
|
)
|
||||||
|
}
|
||||||
|
val dependentIndexes = table.secondaryIndexes.filter { it.columnNames.any { c -> c.equals(action.columnName, ignoreCase = true) } }
|
||||||
|
if (dependentIndexes.isNotEmpty()) {
|
||||||
|
throw DiskExecutorException(
|
||||||
|
"cannot change the type of column ${action.columnName} because index(es) " +
|
||||||
|
"${dependentIndexes.joinToString(", ") { it.name }} are defined on it; drop the index first"
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
val newColumn = column.copy(type = action.newType, precision = action.precision, scale = action.scale)
|
||||||
|
newColumn.default?.let { default ->
|
||||||
|
try {
|
||||||
|
coerceForColumn(Evaluator().evaluate(default), newColumn)
|
||||||
|
} catch (e: KestrelException) {
|
||||||
|
throw DiskExecutorException(
|
||||||
|
"cannot change the type of column ${action.columnName}: its DEFAULT expression is not compatible " +
|
||||||
|
"with ${action.newType.displayWith(action.precision, action.scale)} (${e.message})"
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
val newSchema = TableSchema(
|
||||||
|
schema.name, schema.columns.toMutableList().also { it[columnIndex] = newColumn }, schema.foreignKeys,
|
||||||
|
schema.checks, schema.uniqueConstraints, schema.compositePrimaryKey
|
||||||
|
)
|
||||||
|
|
||||||
Reference in new issue
Block a user