diff --git a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt index 048629b..7d77980 100644 --- a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt +++ b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt @@ -1454,3 +1454,165 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess { 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>() + 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 + ) +