feat(engine): extend Executor with applyMergeDelete and applyMergeInsert

src/main/kotlin/kestrel/engine/Executor.kt

@@ -1805,6 +1805,179 @@ class Executor(private val catalog: Catalog) : SequenceAccess {
         return updatedRow
     }
 
+    /** Returns [targetRow] itself - the pre-deletion image, matching plain DELETE's own RETURNING (see [evaluateReturning]'s doc). */
+    private fun applyMergeDelete(schema: TableSchema, table: Table, targetRow: Row, hasTriggers: Boolean): Row {
+        fireRowTriggers(schema, TriggerTiming.Before, TriggerEvent.Delete, listOf(targetRow to null))
+        ForeignKeyEnforcer.cascadeDelete(catalog, schema.name, listOf(targetRow))
+        table.rows.removeAll { it === targetRow }
+        if (hasTriggers) catalog.reindexTable(schema.name)
+        fireRowTriggers(schema, TriggerTiming.After, TriggerEvent.Delete, listOf(targetRow to null))
+        return targetRow
+    }
+
+    private fun applyMergeInsert(
+        schema: TableSchema,
+        table: Table,
+        clause: MergeNotMatchedInsert,
+        resolver: ColumnResolver,
+        hasTriggers: Boolean
+    ): Row {
+        val targetIndices = resolveInsertTargets(schema, clause.columns)
+        val evaluator = Evaluator(resolver, subqueryRunner, this)
+        val row = buildInsertRow(schema, targetIndices, clause.values, evaluator)
+        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)
+        if (hasTriggers) catalog.reindexTable(schema.name)
+        fireRowTriggers(schema, TriggerTiming.After, TriggerEvent.Insert, listOf(null to row))
+        return row
+    }
+
+    /**
+     * Resolves a MERGE `ON`/`WHEN MATCHED` column reference against whichever of [targetRow]/
+     * [sourceRow] its qualifier names ([targetLabel]/[sourceLabel] - alias if given, bare name
+     * otherwise); an unqualified reference must exist on exactly one side, matching standard SQL's
+     * "ambiguous column" rule for a two-relation scope.
+     */
+    private fun mergeResolver(
+        targetSchema: TableSchema,
+        targetLabel: String,
+        targetRow: Row,
+        sourceSchema: TableSchema,
+        sourceLabel: String,
+        sourceRow: Row
+    ): ColumnResolver = object : ColumnResolver {
+        override fun resolve(expression: ColumnExpression): Value {
+            val qualifier = expression.qualifier
+            if (qualifier != null) {
+                return when {
+                    qualifier.equals(sourceLabel, ignoreCase = true) ->
+                        sourceRow.values[sourceSchema.columnIndex(expression.name) ?: throw KestrelException("no such column: ${expression.name}")]
+                    qualifier.equals(targetLabel, ignoreCase = true) ->
+                        targetRow.values[targetSchema.columnIndex(expression.name) ?: throw KestrelException("no such column: ${expression.name}")]
+                    else -> throw KestrelException("no such table or alias: $qualifier")
+                }
+            }
+            val targetIndex = targetSchema.columnIndex(expression.name)
+            val sourceIndex = sourceSchema.columnIndex(expression.name)
+            if (targetIndex != null && sourceIndex != null) {
+                throw KestrelException("column reference '${expression.name}' is ambiguous between $targetLabel and $sourceLabel")
+            }
+            targetIndex?.let { return targetRow.values[it] }
+            sourceIndex?.let { return sourceRow.values[it] }
+            throw KestrelException("no such column: ${expression.name}")
+        }
+
+        override fun declaredCollation(expression: ColumnExpression): Collation? {
+            val qualifier = expression.qualifier
+            if (qualifier != null) {
+                return when {
+                    qualifier.equals(sourceLabel, ignoreCase = true) -> sourceSchema.column(expression.name)?.collation
+                    qualifier.equals(targetLabel, ignoreCase = true) -> targetSchema.column(expression.name)?.collation
+                    else -> null
+                }
+            }
+            val targetColumn = targetSchema.column(expression.name)
+            val sourceColumn = sourceSchema.column(expression.name)
+            return if (targetColumn != null && sourceColumn == null) targetColumn.collation
+            else if (sourceColumn != null && targetColumn == null) sourceColumn.collation
+            else null
+        }
+    }
+
+    /** A `WHEN NOT MATCHED` clause has no target row to read - a qualifier naming the target throws rather than silently resolving to nothing. */
+    private fun notMatchedResolver(
+        sourceSchema: TableSchema,
+        sourceLabel: String,
+        sourceRow: Row,
+        targetLabel: String
+    ): ColumnResolver = object : ColumnResolver {
+        override fun resolve(expression: ColumnExpression): Value {
+            val qualifier = expression.qualifier
+            if (qualifier != null && qualifier.equals(targetLabel, ignoreCase = true)) {
+                throw KestrelException("no target row is available in a WHEN NOT MATCHED clause: $targetLabel.${expression.name}")
+            }
+            if (qualifier != null && !qualifier.equals(sourceLabel, ignoreCase = true)) {
+                throw KestrelException("no such table or alias: $qualifier")
+            }
+            val index = sourceSchema.columnIndex(expression.name) ?: throw KestrelException("no such column: ${expression.name}")
+            return sourceRow.values[index]
+        }
+
+        override fun declaredCollation(expression: ColumnExpression): Collation? = sourceSchema.column(expression.name)?.collation
+    }
+
+    /** Upfront validation for every clause of a MERGE statement, run even when the source produces zero rows - matching [evaluateReturning]'s own reasoning for why RETURNING is checked before knowing how many rows will actually match. */
+    private fun checkMergeColumns(
+        statement: MergeStatement,
+        targetSchema: TableSchema,
+        targetLabel: String,
+        sourceSchema: TableSchema,
+        sourceLabel: String
+    ) {
+        checkMergeExpressionColumns(statement.onCondition, targetSchema, targetLabel, sourceSchema, sourceLabel, allowTarget = true)
+        for (clause in statement.whenClauses) {
+            when (clause) {
+                is MergeMatchedUpdate -> {
+                    clause.condition?.let { checkMergeExpressionColumns(it, targetSchema, targetLabel, sourceSchema, sourceLabel, allowTarget = true) }
+                    for (assignment in clause.assignments) {
+                        if (targetSchema.columnIndex(assignment.columnName) == null) {
+                            throw KestrelException("no such column: ${assignment.columnName}")
+                        }
+                        assignment.value?.let { checkMergeExpressionColumns(it, targetSchema, targetLabel, sourceSchema, sourceLabel, allowTarget = true) }
+                    }
+                }
+                is MergeMatchedDelete ->
+                    clause.condition?.let { checkMergeExpressionColumns(it, targetSchema, targetLabel, sourceSchema, sourceLabel, allowTarget = true) }
+                is MergeNotMatchedInsert -> {
+                    clause.condition?.let { checkMergeExpressionColumns(it, targetSchema, targetLabel, sourceSchema, sourceLabel, allowTarget = false) }
+                    val targetIndices = resolveInsertTargets(targetSchema, clause.columns)
+                    if (clause.values.size != targetIndices.size) {
+                        throw KestrelException(
+                            "table ${targetSchema.name} expects ${targetIndices.size} values but ${clause.values.size} were supplied"
+                        )
+                    }
+                    clause.values.forEach { value ->
+                        value?.let { checkMergeExpressionColumns(it, targetSchema, targetLabel, sourceSchema, sourceLabel, allowTarget = false) }
+                    }
+                }
+            }
+        }
+    }
+
+    private fun checkMergeExpressionColumns(
+        expression: Expression,
+        targetSchema: TableSchema,
+        targetLabel: String,
+        sourceSchema: TableSchema,
+        sourceLabel: String,
+        allowTarget: Boolean
+    ) {
+        fun recurse(inner: Expression) =
+            checkMergeExpressionColumns(inner, targetSchema, targetLabel, sourceSchema, sourceLabel, allowTarget)
+        when (expression) {
+            is ColumnExpression -> checkMergeColumnReference(expression, targetSchema, targetLabel, sourceSchema, sourceLabel, allowTarget)
+            is BinaryExpression -> { recurse(expression.left); recurse(expression.right) }
+            is UnaryExpression -> recurse(expression.operand)
+            is FunctionExpression -> throw KestrelException("aggregate functions are not allowed here")
+            is WindowExpression -> throw KestrelException("window functions are not allowed here")
+            is LiteralExpression -> Unit
+            is CallExpression -> expression.arguments.forEach { recurse(it) }
+            is IsNullExpression -> recurse(expression.operand)
+            is InExpression -> {
+                recurse(expression.operand)
+                expression.values.forEach { recurse(it) }
+            }
+            is BetweenExpression -> {
+                recurse(expression.operand)
+                recurse(expression.low)
+                recurse(expression.high)
+    }
+
     private fun coerceForColumn(value: Value, column: Column): Value {
         if (value is Value.NullValue) return value
         return when (column.type) {