feat(sql): add enforceTemporaryPrimaryKeyUniqueness
This commit is contained in:
dualfroz committed
1 parent
775d8dcb2b
commit
ec13738fca
1 file changed
+169
@@ -1285,3 +1285,172 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* PRIMARY KEY uniqueness for a temp table, checked by full scan over [rows] (the whole candidate
|
||||
* set post-mutation, not just the newly touched rows - same reasoning as
|
||||
* [kestrel.engine.Executor.enforcePrimaryKeyUniqueness]) since a temp table has no B+Tree to
|
||||
* enforce it structurally the way [DiskTable.rejectDuplicatePrimaryKey] does. Ported 1:1 from
|
||||
* that same in-memory method, [DiskExecutorException] aside.
|
||||
*/
|
||||
private fun enforceTemporaryPrimaryKeyUniqueness(schema: TableSchema, rows: List<Row>) {
|
||||
val indices = schema.primaryKeyIndices
|
||||
if (indices.isEmpty()) return
|
||||
val seen = HashSet<List<Value>>()
|
||||
for (row in rows) {
|
||||
val key = collationFoldedUniquenessKey(schema, indices, indices.map { row.values[it] })
|
||||
if (!seen.add(key)) {
|
||||
val columnList = indices.joinToString(", ") { schema.columns[it].name }
|
||||
throw DiskExecutorException("UNIQUE constraint failed: ${schema.name}($columnList)")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** UNIQUE-constraint counterpart of [enforceTemporaryPrimaryKeyUniqueness] - ported 1:1 from [kestrel.engine.Executor.enforceUniqueConstraints]. */
|
||||
private fun enforceTemporaryUniqueConstraints(schema: TableSchema, rows: List<Row>) {
|
||||
for (unique in schema.uniqueConstraints) {
|
||||
val columnIndices = unique.columns.map { schema.columnIndex(it)!! }
|
||||
val seen = HashSet<List<Value>>()
|
||||
for (row in rows) {
|
||||
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" } ?: ""
|
||||
val columnList = unique.columns.joinToString(", ")
|
||||
throw DiskExecutorException("UNIQUE constraint$label failed: ${schema.name}($columnList)")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Deliberately duplicated from [kestrel.engine.Executor.validateUniqueDefinitions] (same
|
||||
* reasoning as [coerceForColumn]): only checks that every referenced column exists, mirroring
|
||||
* the in-memory engine exactly - it does not reject a constraint naming the same column twice
|
||||
* (`UNIQUE (a, a)`), because the in-memory engine does not either.
|
||||
*/
|
||||
private fun validateUniqueDefinitions(statement: CreateTableStatement) {
|
||||
for (unique in statement.uniqueConstraints) {
|
||||
for (columnName in unique.columns) {
|
||||
if (statement.columns.none { it.name.equals(columnName, ignoreCase = true) }) {
|
||||
throw DiskExecutorException("UNIQUE constraint refers to unknown column: $columnName")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A table-level `PRIMARY KEY (a, b, ...)` implies NOT NULL on every listed column, exactly
|
||||
* like the inline `col TYPE PRIMARY KEY` spelling already does. Deliberately duplicated from
|
||||
* [kestrel.engine.Executor.applyCompositePrimaryKey] (same reasoning as [coerceForColumn]):
|
||||
* this executor reuses the in-memory engine's AST/types but must stay free-standing so a
|
||||
* change to the in-memory engine's error handling can never silently affect the disk-backed
|
||||
* path or vice versa.
|
||||
*/
|
||||
private fun applyCompositePrimaryKey(statement: CreateTableStatement): List<Column> {
|
||||
if (statement.compositePrimaryKey.isEmpty()) return statement.columns
|
||||
val declared = statement.compositePrimaryKey.map { it.lowercase() }.toSet()
|
||||
if (declared.size != statement.compositePrimaryKey.size) {
|
||||
throw DiskExecutorException("table '${statement.tableName}' declares a duplicate column in its PRIMARY KEY")
|
||||
}
|
||||
for (name in statement.compositePrimaryKey) {
|
||||
if (statement.columns.none { it.name.equals(name, ignoreCase = true) }) {
|
||||
throw DiskExecutorException("PRIMARY KEY refers to unknown column: $name")
|
||||
}
|
||||
}
|
||||
return statement.columns.map { column ->
|
||||
if (column.name.lowercase() in declared) column.copy(notNull = true) else column
|
||||
}
|
||||
}
|
||||
|
||||
private fun executeDropTable(statement: DropTableStatement): StatementResult {
|
||||
if (catalog.temporaryTableOrNull(statement.tableName) != null) {
|
||||
catalog.dropTemporaryTable(statement.tableName)
|
||||
} else {
|
||||
catalog.dropTable(statement.tableName, cascade = statement.mode == DropMode.Cascade)
|
||||
}
|
||||
return DropTableResult(statement.tableName)
|
||||
}
|
||||
|
||||
/**
|
||||
* Deliberately NOT in [isDdl] - like DELETE, TRUNCATE's row-clearing is fully covered by the
|
||||
* storage engine's own WAL/2PL transaction, so it may run inside an explicit `BEGIN` and roll
|
||||
* back cleanly with everything else that transaction did. `RESTART IDENTITY` is the one
|
||||
* exception: [DiskCatalog.alterSequence] always opens and commits its own transaction (mirroring
|
||||
* [DiskCatalog.nextValue]'s already-non-transactional sequence position - see that method's own
|
||||
* doc), so a sequence restarted here stays restarted even if the enclosing statement transaction
|
||||
* later rolls back. This is a deliberate, bounded asymmetry with [kestrel.engine.Executor]'s
|
||||
* in-memory engine, where RESTART IDENTITY *is* undone by ROLLBACK because ALTER SEQUENCE there
|
||||
* mutates a map entry covered by the whole-catalog snapshot/restore (see
|
||||
* [kestrel.catalog.Catalog.truncateTables]).
|
||||
*/
|
||||
private fun executeTruncate(statement: TruncateTableStatement): StatementResult {
|
||||
val truncated = withStatementTransaction { txn ->
|
||||
catalog.truncateTables(txn, statement.tableNames, statement.mode == DropMode.Cascade)
|
||||
}
|
||||
if (statement.restartIdentity) {
|
||||
for (name in truncated) restartOwnedSequences(catalog.table(name).schema)
|
||||
}
|
||||
return TruncateTableResult(truncated)
|
||||
}
|
||||
|
||||
/** Ported 1:1 from [kestrel.catalog.Catalog.restartOwnedSequences] - see that method's own doc for why a NEXTVAL DEFAULT is this engine's only signal for "this column is identity-backed". */
|
||||
private fun restartOwnedSequences(schema: TableSchema) {
|
||||
for (column in schema.columns) {
|
||||
val default = column.default
|
||||
if (default is NextValueExpression) {
|
||||
val sequence = catalog.sequenceOrNull(default.sequenceName) ?: continue
|
||||
catalog.alterSequence(default.sequenceName, sequence.definition, restartAt = sequence.definition.startValue)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun executeAlterTable(statement: AlterTableStatement): StatementResult {
|
||||
when (val action = statement.action) {
|
||||
is AddColumn -> executeAddColumn(statement.tableName, action.column)
|
||||
is DropColumn -> catalog.dropColumn(statement.tableName, action.columnName, cascade = action.mode == DropMode.Cascade)
|
||||
is RenameColumn -> catalog.renameColumn(statement.tableName, action.oldName, action.newName)
|
||||
is RenameTable -> catalog.renameTable(statement.tableName, action.newName)
|
||||
is AddConstraint -> executeAddConstraint(statement.tableName, action.constraint)
|
||||
is DropConstraint -> catalog.dropConstraint(statement.tableName, action.name, cascade = action.mode == DropMode.Cascade)
|
||||
is ChangeColumnType -> executeChangeColumnType(statement.tableName, action)
|
||||
}
|
||||
val resultName = (statement.action as? RenameTable)?.newName ?: statement.tableName
|
||||
return AlterTableResult(resultName)
|
||||
}
|
||||
|
||||
/** Mirrors [kestrel.engine.Executor.executeAddConstraint] - see that method's own doc. */
|
||||
private fun executeAddConstraint(tableName: String, added: AddedConstraint) {
|
||||
when (added) {
|
||||
is AddedCheck -> executeAddCheckConstraint(tableName, added.check)
|
||||
is AddedUnique -> executeAddUniqueConstraint(tableName, added.unique)
|
||||
is AddedForeignKey -> executeAddForeignKeyConstraint(tableName, added.foreignKey)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* `ALTER TABLE t ADD CONSTRAINT name CHECK (...)` - mirrors
|
||||
* [kestrel.engine.Executor.executeAddCheckConstraint]: validated exactly like a CREATE TABLE-time
|
||||
* CHECK ([validateCheckDefinitions]: known columns only, no subquery), PLUS every existing row is
|
||||
* evaluated against it up front (under its own probe transaction, mirroring [executeAddColumn]'s
|
||||
* `probeTxn`) before the constraint is installed via [DiskCatalog.addCheckConstraint].
|
||||
*/
|
||||
private fun executeAddCheckConstraint(tableName: String, check: CheckConstraint) {
|
||||
val table = catalog.table(tableName)
|
||||
val schema = table.schema
|
||||
check.name?.let { requireUniqueConstraintName(schema, it) }
|
||||
checkColumns(check.expression, schema)
|
||||
requireNoSubquery(check.expression)
|
||||
val probeTxn = catalog.beginTransaction()
|
||||
runInTransaction(probeTxn) {
|
||||
val subqueries = subqueryRunnerFor(probeTxn)
|
||||
for ((_, row) in table.scan(probeTxn)) {
|
||||
val result = Evaluator(schema, row, subqueries).evaluate(check.expression)
|
||||
if (result is Value.BooleanValue && !result.value) {
|
||||
val label = check.name?.let { " $it" } ?: ""
|
||||
throw DiskExecutorException("CHECK constraint$label failed: an existing row in $tableName violates it")
|
||||
}
|
||||
}
|
||||
}
|
||||
catalog.addCheckConstraint(tableName, check)
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user