From 68f82b7815ed7ea6ade0e4a89d23a9c0eda3ad4b Mon Sep 17 00:00:00 2001 From: DualFroz Date: Tue, 1 Sep 2026 20:12:45 +0200 Subject: [PATCH] feat(engine): extend Executor with executeMerge and executeMergeInner --- src/main/kotlin/kestrel/engine/Executor.kt | 174 ++++++++++++++++++++- 1 file changed, 173 insertions(+), 1 deletion(-) diff --git a/src/main/kotlin/kestrel/engine/Executor.kt b/src/main/kotlin/kestrel/engine/Executor.kt index c28e565..0e454a4 100644 --- a/src/main/kotlin/kestrel/engine/Executor.kt +++ b/src/main/kotlin/kestrel/engine/Executor.kt @@ -1630,7 +1630,179 @@ class Executor(private val catalog: Catalog) : SequenceAccess { table.rows.clear() table.rows.addAll(newRows) catalog.reindexTable(schema.name) - return UpdateResult(updated) + fireRowTriggers(schema, TriggerTiming.After, TriggerEvent.Update, touchedRows) + val returning = statement.returning?.let { evaluateReturning(schema, touchedRows.map { pair -> pair.second }, it) } + return UpdateResult(updated, returning) + } + + private fun executeMerge(statement: MergeStatement): StatementResult = + withStatementAtomicity(statement.targetTable, null) { executeMergeInner(statement) } + + /** + * The `ON` join is matched against [originalTargetRows] - a snapshot of the target table taken + * once before any mutation - not the live, mutating `table.rows`: this is what makes "a target + * row matched by two different source rows" a well-defined, detectable condition at all (see + * below), and matches how a real MERGE's source/target join reads one consistent view of the + * target rather than re-joining against its own in-flight writes. A row this MERGE inserts is + * therefore never itself matched by a later source row (it never enters the snapshot), and a + * row this MERGE updates/deletes keeps its ORIGINAL identity for join purposes even after the + * live table's copy of it changes underneath - [applyMergeUpdate]/[applyMergeDelete] are handed + * that same original [Row] object and use `===` against it to find their spot in the live table. + * + * For each source row, every target row (from the snapshot) satisfying [MergeStatement.onCondition] + * is a MATCHED pair: the first [MergeMatchedUpdate]/[MergeMatchedDelete] clause (in + * [MergeStatement.whenClauses] written order - MATCHED and NOT MATCHED clauses can be + * interleaved, but a given pair only ever considers its own category) whose own `AND` condition + * also holds runs; if none holds, that pair is left untouched, matching Postgres. A target row + * already affected earlier in this same statement by a different matched pair throws rather than + * silently applying a second action - the same "MERGE command cannot affect a row a second time" + * guarantee Postgres gives, needed here because this engine has no snapshot isolation to make + * that outcome well-defined otherwise. A source row matched by zero target rows is NOT MATCHED: + * the first [MergeNotMatchedInsert] whose own condition holds (evaluated against the source row + * alone - there is no target row to reference) inserts one new row; [checkColumns]-equivalent + * validation for every clause happens up front via [checkMergeColumns], even when zero source + * rows exist, matching this engine's usual eager-validation bar for every other schema-shaped + * clause. + */ + private fun executeMergeInner(statement: MergeStatement): StatementResult { + val table = catalog.table(statement.targetTable) + val schema = table.schema + val targetLabel = statement.targetAlias ?: statement.targetTable + val (sourceSchema, sourceRows) = selectRunner.resolveMergeSource(statement.source) + val sourceLabel = selectRunner.mergeSourceLabel(statement.source) + if (targetLabel.equals(sourceLabel, ignoreCase = true)) { + throw KestrelException("MERGE target and source cannot share the same name or alias: $targetLabel") + } + checkMergeColumns(statement, schema, targetLabel, sourceSchema, sourceLabel) + + val originalTargetRows = table.rows.toList() + val hasTriggers = catalog.hasTriggers(schema.name) + val affectedTargets = mutableListOf() + val returnedRows = mutableListOf>() + var inserted = 0 + var updated = 0 + var deleted = 0 + + for (sourceRow in sourceRows) { + val matchedTargets = originalTargetRows.filter { targetRow -> + val resolver = mergeResolver(schema, targetLabel, targetRow, sourceSchema, sourceLabel, sourceRow) + isTrue(Evaluator(resolver, subqueryRunner).evaluate(statement.onCondition)) + } + if (matchedTargets.isNotEmpty()) { + for (targetRow in matchedTargets) { + val resolver = mergeResolver(schema, targetLabel, targetRow, sourceSchema, sourceLabel, sourceRow) + val clause = statement.whenClauses.firstOrNull { candidate -> + (candidate is MergeMatchedUpdate || candidate is MergeMatchedDelete) && + (candidate.condition == null || isTrue(Evaluator(resolver, subqueryRunner).evaluate(candidate.condition!!))) + } ?: continue + if (affectedTargets.any { it === targetRow }) { + throw KestrelException("MERGE command cannot affect a target row a second time") + } + affectedTargets.add(targetRow) + when (clause) { + is MergeMatchedUpdate -> { + val updatedRow = applyMergeUpdate(schema, table, targetRow, clause, resolver, hasTriggers) + returnedRows.add(updatedRow to "UPDATE") + updated++ + } + is MergeMatchedDelete -> { + val deletedRow = applyMergeDelete(schema, table, targetRow, hasTriggers) + returnedRows.add(deletedRow to "DELETE") + deleted++ + } + else -> error("unreachable: filtered to MATCHED clauses above") + } + } + continue + } + val resolver = notMatchedResolver(sourceSchema, sourceLabel, sourceRow, targetLabel) + val clause = statement.whenClauses.firstOrNull { candidate -> + candidate is MergeNotMatchedInsert && + (candidate.condition == null || isTrue(Evaluator(resolver, subqueryRunner).evaluate(candidate.condition!!))) + } as? MergeNotMatchedInsert ?: continue + val insertedRow = applyMergeInsert(schema, table, clause, resolver, hasTriggers) + returnedRows.add(insertedRow to "INSERT") + inserted++ + } + if (!hasTriggers) catalog.reindexTable(schema.name) + val returning = statement.returning?.let { evaluateMergeReturning(schema, returnedRows, it) } + return MergeResult(inserted, updated, deleted, returning) + } + + /** + * Evaluates a MERGE statement's RETURNING clause against every row an action of this MERGE + * actually touched - the post-image for an inserted/updated row, the pre-deletion image for a + * deleted one (same "nothing else to show" reasoning as [evaluateReturning]) - in the exact order + * [executeMergeInner] applied those actions in (source-row order, and within a single source row's + * multiple matched target rows, the order [executeMergeInner]'s own inner loop visits them). + * [actionedRow]'s second element is `"INSERT"`/`"UPDATE"`/`"DELETE"`, bound to that row's + * [Evaluator] as [MergeActionExpression]'s value so `MERGE_ACTION()` can appear in the projection - + * the one context where it is actually legal. Column references resolve against [schema] (the + * MERGE's target table) alone, same as [evaluateReturning] - RETURNING never sees the source row, + * only what ended up true of the target. + */ + private fun evaluateMergeReturning(schema: TableSchema, actionedRows: List>, projection: Projection): SelectResult { + val columns = when (projection) { + is StarProjection -> schema.columns.map { it.name } + is ExpressionProjection -> projection.items.mapIndexed { index, item -> + checkColumns(item.expression, schema) + returningColumnLabel(item, index) + } + } + val resultRows = actionedRows.map { (row, action) -> + when (projection) { + is StarProjection -> row.values.toList() + is ExpressionProjection -> { + val evaluator = Evaluator(rowResolver(schema, row), subqueryRunner, mergeAction = action) + projection.items.map { evaluator.evaluate(it.expression) } + } + } + } + return SelectResult(columns, resultRows) + } + + private fun applyMergeUpdate( + schema: TableSchema, + table: Table, + targetRow: Row, + clause: MergeMatchedUpdate, + resolver: ColumnResolver, + hasTriggers: Boolean + ): Row { + val assignmentEvaluator = Evaluator(resolver, subqueryRunner, this) + val values = targetRow.values.toMutableList() + for (assignment in clause.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 updatedRow = Row(values) + fireRowTriggers(schema, TriggerTiming.Before, TriggerEvent.Update, listOf(targetRow to updatedRow)) + ForeignKeyEnforcer.validateRow(catalog, schema, updatedRow.values, uncommittedSiblings = emptyList()) + enforceChecks(schema, updatedRow) + val others = table.rows.filter { it !== targetRow } + enforcePrimaryKeyUniqueness(schema, others + updatedRow) + enforceUniqueConstraints(schema, others + updatedRow) + enforceUniqueIndexes(schema, others + updatedRow) + for (group in schema.enforcedUniqueKeyGroups()) { + val columnIndices = group.map { schema.columnIndex(it)!! } + val oldKey = columnIndices.map { targetRow.values[it] } + val newKey = columnIndices.map { updatedRow.values[it] } + ForeignKeyEnforcer.cascadeUpdate(catalog, schema.name, group, listOf(oldKey to newKey)) + } + val position = table.rows.indexOfFirst { it === targetRow } + table.rows[position] = updatedRow + if (hasTriggers) catalog.reindexTable(schema.name) + fireRowTriggers(schema, TriggerTiming.After, TriggerEvent.Update, listOf(targetRow to updatedRow)) + return updatedRow } private fun coerceForColumn(value: Value, column: Column): Value {