feat(engine): extend Executor with UniqueKey and uniqueKeysOf
This commit is contained in:
dualfroz committed
1 parent
2a0c532de3
commit
66cc015636
1 file changed
+191
-5
@@ -1339,10 +1339,193 @@ class Executor(private val catalog: Catalog) : SequenceAccess {
|
|||||||
touched[columnIndex] = true
|
touched[columnIndex] = true
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
enforcePrimaryKeyUniqueness(schema, table.rows + newRows)
|
// A column omitted from an explicit INSERT column list falls back to its DEFAULT
|
||||||
table.rows.addAll(newRows)
|
// (evaluated fresh per row); one that was supplied - even an explicit NULL - never
|
||||||
catalog.reindexTable(schema.name)
|
// does, matching standard SQL DEFAULT semantics.
|
||||||
return InsertResult(newRows.size)
|
for ((index, column) in schema.columns.withIndex()) {
|
||||||
|
val default = column.default
|
||||||
|
if (!touched[index] && default != null) {
|
||||||
|
values[index] = coerceForColumn(evaluator.evaluate(default), column)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
applyGeneratedColumns(schema, values)
|
||||||
|
validateNotNull(schema, values)
|
||||||
|
return Row(values)
|
||||||
|
}
|
||||||
|
|
||||||
|
private class UniqueKey(val columnIndices: List<Int>, val columnNames: List<String>)
|
||||||
|
|
||||||
|
private fun uniqueKeysOf(schema: TableSchema): List<UniqueKey> {
|
||||||
|
val keys = mutableListOf<UniqueKey>()
|
||||||
|
val pkIndices = schema.primaryKeyIndices
|
||||||
|
if (pkIndices.isNotEmpty()) keys.add(UniqueKey(pkIndices, pkIndices.map { schema.columns[it].name }))
|
||||||
|
for (unique in schema.uniqueConstraints) {
|
||||||
|
val indices = unique.columns.map { schema.columnIndex(it)!! }
|
||||||
|
keys.add(UniqueKey(indices, unique.columns))
|
||||||
|
}
|
||||||
|
for (definition in catalog.indexDefinitions()) {
|
||||||
|
if (!definition.unique || !definition.tableName.equals(schema.name, ignoreCase = true)) continue
|
||||||
|
val indices = definition.columnNames.mapNotNull { schema.columnIndex(it) }
|
||||||
|
if (indices.size != definition.columnNames.size) continue
|
||||||
|
keys.add(UniqueKey(indices, definition.columnNames))
|
||||||
|
}
|
||||||
|
return keys
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Empty [targetColumns] (bare `ON CONFLICT DO ...`) matches against any of [keys]. */
|
||||||
|
private fun resolveConflictTargetKeys(schema: TableSchema, targetColumns: List<String>, keys: List<UniqueKey>): List<UniqueKey> {
|
||||||
|
if (targetColumns.isEmpty()) return keys
|
||||||
|
val targetIndices = targetColumns.map { name ->
|
||||||
|
schema.columnIndex(name) ?: throw KestrelException("no such column: $name")
|
||||||
|
}
|
||||||
|
val targetSet = targetIndices.toSet()
|
||||||
|
val matched = keys.firstOrNull { it.columnIndices.toSet() == targetSet }
|
||||||
|
?: throw KestrelException(
|
||||||
|
"no unique or primary key constraint matching ON CONFLICT specification: " +
|
||||||
|
targetColumns.joinToString(", ")
|
||||||
|
)
|
||||||
|
return listOf(matched)
|
||||||
|
}
|
||||||
|
|
||||||
|
/** NULLs never collide with each other (standard SQL), mirroring enforceUniqueConstraints/enforceUniqueIndexes above. */
|
||||||
|
private fun findConflictingRow(existingRows: List<Row>, candidate: List<Value>, keys: List<UniqueKey>): Row? {
|
||||||
|
for (existing in existingRows) {
|
||||||
|
for (key in keys) {
|
||||||
|
val candidateKey = key.columnIndices.map { candidate[it] }
|
||||||
|
if (candidateKey.any { it is Value.NullValue }) continue
|
||||||
|
val existingKey = key.columnIndices.map { existing.values[it] }
|
||||||
|
if (candidateKey == existingKey) return existing
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return null
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Processes each candidate row as its own micro-insert against the table's CURRENT state
|
||||||
|
* (rather than batch-validating the whole VALUES list like the no-conflict path above), so
|
||||||
|
* a later row in the same multi-row INSERT correctly sees an earlier row's insert or update -
|
||||||
|
* exactly how Postgres/SQLite define ON CONFLICT for multi-row VALUES. A row landing with no
|
||||||
|
* conflict fires an INSERT trigger pair, exactly like the no-conflict path in [executeInsertInner];
|
||||||
|
* a row resolved by DO UPDATE fires an UPDATE trigger pair instead, matching that the row's own
|
||||||
|
* data ends up changed by an UPDATE rather than a fresh INSERT (mirroring Postgres: the proposed
|
||||||
|
* row that lost the race is folded into an update of the pre-existing one). DO NOTHING fires no
|
||||||
|
* trigger at all, since nothing about the table actually changed for that row. When [schema] has
|
||||||
|
* at least one trigger, [catalog.reindexTable] runs right after each row's own mutation - before
|
||||||
|
* that row's AFTER trigger, rather than once for the whole statement like the batch paths do - so a
|
||||||
|
* trigger body's own query against this table never observes a stale index for a row this very
|
||||||
|
* statement already landed; a trigger-free table keeps the original single-reindex-at-the-end cost.
|
||||||
|
*/
|
||||||
|
private fun executeInsertWithConflictHandling(
|
||||||
|
schema: TableSchema,
|
||||||
|
table: Table,
|
||||||
|
candidateRows: List<Row>,
|
||||||
|
onConflict: OnConflictClause,
|
||||||
|
returning: Projection? = null,
|
||||||
|
triggerContext: TriggerRowContext? = null
|
||||||
|
): StatementResult {
|
||||||
|
val keys = uniqueKeysOf(schema)
|
||||||
|
val targetKeys = resolveConflictTargetKeys(schema, onConflict.targetColumns, keys)
|
||||||
|
// Reindexing per row (rather than once for the whole statement like the batch INSERT/UPDATE
|
||||||
|
// paths) is only needed so a trigger body's own query against this table never sees a stale
|
||||||
|
// index for a row this very statement already landed - skipped entirely, preserving the
|
||||||
|
// original single-reindex-at-the-end cost, when this table has no trigger to fire at all.
|
||||||
|
val hasTriggers = catalog.hasTriggers(schema.name)
|
||||||
|
var affected = 0
|
||||||
|
// DO NOTHING contributes no row (nothing about the table changed for it); a DO UPDATE row
|
||||||
|
// whose WHERE guard fails `continue`s before ever reaching the append below, same as DO
|
||||||
|
// NOTHING - both are absent from RETURNING exactly like Postgres.
|
||||||
|
val returnedRows = mutableListOf<Row>()
|
||||||
|
for (row in candidateRows) {
|
||||||
|
val conflicting = findConflictingRow(table.rows, row.values, targetKeys)
|
||||||
|
if (conflicting == null) {
|
||||||
|
fireRowTriggers(schema, TriggerTiming.Before, TriggerEvent.Insert, listOf(null to row))
|
||||||
|
ForeignKeyEnforcer.validateRow(catalog, schema, row.values, uncommittedSiblings = emptyList())
|
||||||
|
enforceChecks(schema, row)
|
||||||
|
enforcePrimaryKeyUniqueness(schema, table.rows + row)
|
||||||
|
enforceUniqueConstraints(schema, table.rows + row)
|
||||||
|
enforceUniqueIndexes(schema, table.rows + row)
|
||||||
|
table.rows.add(row)
|
||||||
|
affected++
|
||||||
|
returnedRows.add(row)
|
||||||
|
if (hasTriggers) catalog.reindexTable(schema.name)
|
||||||
|
fireRowTriggers(schema, TriggerTiming.After, TriggerEvent.Insert, listOf(null to row))
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
when (val action = onConflict.action) {
|
||||||
|
ConflictDoNothing -> Unit
|
||||||
|
is ConflictDoUpdate -> {
|
||||||
|
val resolver = composeResolver(excludedAwareResolver(schema, conflicting, row), triggerContext)
|
||||||
|
// NEXTVAL/CURRVAL are deliberately usable in the assignment values below but not
|
||||||
|
// in this WHERE-shaped guard - same "not in WHERE" rule as everywhere else, kept
|
||||||
|
// airtight here by using a second Evaluator instance rather than one shared one.
|
||||||
|
val conditionEvaluator = Evaluator(resolver, subqueryRunner)
|
||||||
|
if (action.where != null && !isTrue(conditionEvaluator.evaluate(action.where))) continue
|
||||||
|
val assignmentEvaluator = Evaluator(resolver, subqueryRunner, this)
|
||||||
|
val values = conflicting.values.toMutableList()
|
||||||
|
for (assignment in action.assignments) {
|
||||||
|
val index = schema.columnIndex(assignment.columnName)
|
||||||
|
?: throw KestrelException("no such column: ${assignment.columnName}")
|
||||||
|
val column = schema.columns[index]
|
||||||
|
rejectGeneratedAssignment(column, explicitValueGiven = assignment.value != null)
|
||||||
|
values[index] = if (assignment.value != null) {
|
||||||
|
coerceForColumn(assignmentEvaluator.evaluate(assignment.value), column)
|
||||||
|
} else {
|
||||||
|
column.default?.let { coerceForColumn(assignmentEvaluator.evaluate(it), column) } ?: Value.NullValue
|
||||||
|
}
|
||||||
|
}
|
||||||
|
applyGeneratedColumns(schema, values)
|
||||||
|
validateNotNull(schema, values)
|
||||||
|
val updated = Row(values)
|
||||||
|
// Fires before any constraint below is checked, same as executeUpdate's own BEFORE
|
||||||
|
// trigger - a BEFORE UPDATE trigger's body can reject this row's update before this
|
||||||
|
// engine's own validation ever runs.
|
||||||
|
fireRowTriggers(schema, TriggerTiming.Before, TriggerEvent.Update, listOf(conflicting to updated))
|
||||||
|
ForeignKeyEnforcer.validateRow(catalog, schema, updated.values, uncommittedSiblings = emptyList())
|
||||||
|
enforceChecks(schema, updated)
|
||||||
|
val others = table.rows.filter { it !== conflicting }
|
||||||
|
enforcePrimaryKeyUniqueness(schema, others + updated)
|
||||||
|
enforceUniqueConstraints(schema, others + updated)
|
||||||
|
enforceUniqueIndexes(schema, others + updated)
|
||||||
|
// Same as executeUpdate: any table with a FK pointing at one of this table's
|
||||||
|
// enforced-unique key groups needs a cascade/RESTRICT pass when that group's
|
||||||
|
// value actually changes, not just the PRIMARY KEY.
|
||||||
|
for (group in schema.enforcedUniqueKeyGroups()) {
|
||||||
|
val columnIndices = group.map { schema.columnIndex(it)!! }
|
||||||
|
val oldKey = columnIndices.map { conflicting.values[it] }
|
||||||
|
val newKey = columnIndices.map { updated.values[it] }
|
||||||
|
ForeignKeyEnforcer.cascadeUpdate(catalog, schema.name, group, listOf(oldKey to newKey))
|
||||||
|
}
|
||||||
|
val position = table.rows.indexOfFirst { it === conflicting }
|
||||||
|
table.rows[position] = updated
|
||||||
|
affected++
|
||||||
|
returnedRows.add(updated)
|
||||||
|
if (hasTriggers) catalog.reindexTable(schema.name)
|
||||||
|
fireRowTriggers(schema, TriggerTiming.After, TriggerEvent.Update, listOf(conflicting to updated))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (!hasTriggers) catalog.reindexTable(schema.name)
|
||||||
|
return InsertResult(affected, returning?.let { evaluateReturning(schema, returnedRows, it) })
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Bare column names resolve against [target] (the existing conflicting row, matching plain
|
||||||
|
* UPDATE semantics), while `EXCLUDED.column` resolves against [excluded] (the row proposed by
|
||||||
|
* this INSERT that failed to land) - the same pseudo-table Postgres exposes in DO UPDATE.
|
||||||
|
*/
|
||||||
|
private fun excludedAwareResolver(schema: TableSchema, target: Row, excluded: Row): ColumnResolver = object : ColumnResolver {
|
||||||
|
override fun resolve(expression: ColumnExpression): Value {
|
||||||
|
val qualifier = expression.qualifier
|
||||||
|
val useExcluded = qualifier != null && qualifier.equals("excluded", ignoreCase = true)
|
||||||
|
if (qualifier != null && !useExcluded && !qualifier.equals(schema.name, ignoreCase = true)) {
|
||||||
|
throw KestrelException("no such table or alias: $qualifier")
|
||||||
|
}
|
||||||
|
val index = schema.columnIndex(expression.name)
|
||||||
|
?: throw KestrelException("no such column: ${expression.name}")
|
||||||
|
return if (useExcluded) excluded.values[index] else target.values[index]
|
||||||
|
}
|
||||||
|
|
||||||
|
override fun declaredCollation(expression: ColumnExpression): Collation? = schema.column(expression.name)?.collation
|
||||||
}
|
}
|
||||||
|
|
||||||
private fun resolveInsertTargets(schema: TableSchema, columns: List<String>?): List<Int> {
|
private fun resolveInsertTargets(schema: TableSchema, columns: List<String>?): List<Int> {
|
||||||
@@ -1353,7 +1536,10 @@ class Executor(private val catalog: Catalog) : SequenceAccess {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private fun executeDelete(statement: DeleteStatement): StatementResult {
|
private fun executeDelete(statement: DeleteStatement, triggerContext: TriggerRowContext? = null): StatementResult =
|
||||||
|
withStatementAtomicity(statement.tableName, triggerContext) { executeDeleteInner(statement, triggerContext) }
|
||||||
|
|
||||||
|
private fun executeDeleteInner(statement: DeleteStatement, triggerContext: TriggerRowContext?): StatementResult {
|
||||||
val table = catalog.table(statement.tableName)
|
val table = catalog.table(statement.tableName)
|
||||||
val schema = table.schema
|
val schema = table.schema
|
||||||
if (statement.where == null) {
|
if (statement.where == null) {
|
||||||
|
|||||||
Reference in new issue
Block a user