From ec13738fca3701dcd2167fc56a8687fe5623e7e5 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Sun, 6 Sep 2026 21:55:51 +0200 Subject: [PATCH] feat(sql): add enforceTemporaryPrimaryKeyUniqueness --- .../kotlin/kestrel/store/sql/DiskExecutor.kt | 169 ++++++++++++++++++ 1 file changed, 169 insertions(+) diff --git a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt index 8bd64d7..048629b 100644 --- a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt +++ b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt @@ -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) { + val indices = schema.primaryKeyIndices + if (indices.isEmpty()) return + val seen = HashSet>() + 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) { + for (unique in schema.uniqueConstraints) { + val columnIndices = unique.columns.map { schema.columnIndex(it)!! } + val seen = HashSet>() + 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 { + 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) + } +