feat(sql): add applyMergeInsert and resolveMergeSource to DiskExecutor

src/main/kotlin/kestrel/store/sql/DiskExecutor.kt

@@ -4148,3 +4148,157 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess {
         return targetRow
     }
 
+    private fun applyMergeInsert(
+        schema: TableSchema,
+        table: DiskTable,
+        txn: Long,
+        clause: MergeNotMatchedInsert,
+        resolver: ColumnResolver,
+        subqueries: SubqueryRunner
+    ): Row {
+        val targetIndices = resolveInsertTargets(schema, clause.columns)
+        val evaluator = Evaluator(resolver, subqueries, this)
+        val values = MutableList<Value>(schema.columns.size) { Value.NullValue }
+        val touched = BooleanArray(schema.columns.size)
+        for ((position, expression) in clause.values.withIndex()) {
+            val columnIndex = targetIndices[position]
+            if (expression != null) {
+                rejectGeneratedAssignment(schema.columns[columnIndex], explicitValueGiven = true)
+                values[columnIndex] = coerceForColumn(evaluator.evaluate(expression), schema.columns[columnIndex])
+                touched[columnIndex] = true
+            }
+        }
+        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, subqueries)
+        val row = Row(values)
+        fireRowTriggers(schema, TriggerTiming.Before, TriggerEvent.Insert, listOf(null to row), txn)
+        DiskForeignKeyEnforcer.validateRow(catalog, txn, schema, row.values, uncommittedSiblings = emptyList())
+        enforceChecks(schema, row, subqueries)
+        table.insert(txn, row)
+        fireRowTriggers(schema, TriggerTiming.After, TriggerEvent.Insert, listOf(null to row), txn)
+        return row
+    }
+
+    /** A snapshot (schema, rows) for a MERGE `USING` source - ported 1:1 from [kestrel.engine.SelectRunner.resolveMergeSource], just via [resolveFromItem] rather than a separate in-memory `resolveSource`. */
+    private fun resolveMergeSource(item: FromItem, txn: Long): Pair<TableSchema, List<Row>> {
+        val source = resolveFromItem(item, txn)
+        return source.schema to source.scan(txn).map { Row(it) }
+    }
+
+    /**
+     * 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. Ported 1:1 from
+     * [kestrel.engine.Executor.mergeResolver].
+     */
+    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 DiskExecutorException("no such column: ${expression.name}")]
+                    qualifier.equals(targetLabel, ignoreCase = true) ->
+                        targetRow.values[targetSchema.columnIndex(expression.name) ?: throw DiskExecutorException("no such column: ${expression.name}")]
+                    else -> throw DiskExecutorException("no such table or alias: $qualifier")
+                }
+            }
+            val targetIndex = targetSchema.columnIndex(expression.name)
+            val sourceIndex = sourceSchema.columnIndex(expression.name)
+            if (targetIndex != null && sourceIndex != null) {
+                throw DiskExecutorException("column reference '${expression.name}' is ambiguous between $targetLabel and $sourceLabel")
+            }
+            targetIndex?.let { return targetRow.values[it] }
+            sourceIndex?.let { return sourceRow.values[it] }
+            throw DiskExecutorException("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. Ported 1:1 from [kestrel.engine.Executor.notMatchedResolver]. */
+    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 DiskExecutorException("no target row is available in a WHEN NOT MATCHED clause: $targetLabel.${expression.name}")
+            }
+            if (qualifier != null && !qualifier.equals(sourceLabel, ignoreCase = true)) {
+                throw DiskExecutorException("no such table or alias: $qualifier")
+            }
+            val index = sourceSchema.columnIndex(expression.name) ?: throw DiskExecutorException("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. Ported 1:1 from [kestrel.engine.Executor.checkMergeColumns]. */
+    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 DiskExecutorException("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 DiskExecutorException(
+                            "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) }
+                    }
+                }
+            }
+        }
+    }
+