From 66cc015636ed9521f12f0102dc97320b15b00f81 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Tue, 1 Sep 2026 19:40:04 +0200 Subject: [PATCH] feat(engine): extend Executor with UniqueKey and uniqueKeysOf --- src/main/kotlin/kestrel/engine/Executor.kt | 196 ++++++++++++++++++++- 1 file changed, 191 insertions(+), 5 deletions(-) diff --git a/src/main/kotlin/kestrel/engine/Executor.kt b/src/main/kotlin/kestrel/engine/Executor.kt index 5c566d2..8fa79ac 100644 --- a/src/main/kotlin/kestrel/engine/Executor.kt +++ b/src/main/kotlin/kestrel/engine/Executor.kt @@ -1339,10 +1339,193 @@ class Executor(private val catalog: Catalog) : SequenceAccess { touched[columnIndex] = true } } - enforcePrimaryKeyUniqueness(schema, table.rows + newRows) - table.rows.addAll(newRows) - catalog.reindexTable(schema.name) - return InsertResult(newRows.size) + // A column omitted from an explicit INSERT column list falls back to its DEFAULT + // (evaluated fresh per row); one that was supplied - even an explicit NULL - never + // does, matching standard SQL DEFAULT semantics. + 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, val columnNames: List) + + private fun uniqueKeysOf(schema: TableSchema): List { + val keys = mutableListOf() + 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, keys: List): List { + 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, candidate: List, keys: List): 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, + 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() + 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?): List { @@ -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 schema = table.schema if (statement.where == null) {