feat(sql): add checkMergeExpressionColumns and recurse to DiskExecutor

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

@@ -4302,3 +4302,130 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess {
         }
     }
 
+    /** Fifth near-parallel recursive [Expression] walker in this file (alongside [requireNoSubquery]/[checkColumns]/[checkReturningColumns]/[validateTriggerRowReferences]) - same deliberate non-abstraction as [kestrel.engine.Executor.checkMergeExpressionColumns], since this one needs two relations plus ambiguity detection none of the others do. */
+    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 DiskExecutorException("aggregate functions are not allowed here")
+            is WindowExpression -> throw DiskExecutorException("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)
+            }
+            is LikeExpression -> {
+                recurse(expression.operand)
+                recurse(expression.pattern)
+            }
+            is CaseExpression -> {
+                expression.operand?.let { recurse(it) }
+                expression.branches.forEach { recurse(it.condition); recurse(it.result) }
+                expression.elseExpression?.let { recurse(it) }
+            }
+            is ScalarSubqueryExpression -> Unit
+            is ExistsExpression -> Unit
+            is InSubqueryExpression -> recurse(expression.operand)
+            is CastExpression -> recurse(expression.operand)
+            is CollateExpression -> recurse(expression.operand)
+            // Structurally harmless here too - see checkReturningColumns's own MERGE_ACTION() branch.
+            is NextValueExpression, is CurrentValueExpression, is MergeActionExpression, is ParameterExpression -> Unit
+        }
+    }
+
+    private fun checkMergeColumnReference(
+        expression: ColumnExpression,
+        targetSchema: TableSchema,
+        targetLabel: String,
+        sourceSchema: TableSchema,
+        sourceLabel: String,
+        allowTarget: Boolean
+    ) {
+        val qualifier = expression.qualifier
+        if (qualifier != null) {
+            val matchesSource = qualifier.equals(sourceLabel, ignoreCase = true)
+            val matchesTargetLabel = qualifier.equals(targetLabel, ignoreCase = true)
+            if (matchesTargetLabel && !allowTarget) {
+                throw DiskExecutorException("no target row is available in a WHEN NOT MATCHED clause: $targetLabel.${expression.name}")
+            }
+            if (!matchesSource && !matchesTargetLabel) {
+                throw DiskExecutorException("no such table or alias: $qualifier")
+            }
+            val schema = if (matchesSource) sourceSchema else targetSchema
+            if (schema.columnIndex(expression.name) == null) throw DiskExecutorException("no such column: ${expression.name}")
+            return
+        }
+        val inTarget = allowTarget && targetSchema.columnIndex(expression.name) != null
+        val inSource = sourceSchema.columnIndex(expression.name) != null
+        if (inTarget && inSource) {
+            throw DiskExecutorException("column reference '${expression.name}' is ambiguous between $targetLabel and $sourceLabel")
+        }
+        if (!inTarget && !inSource) throw DiskExecutorException("no such column: ${expression.name}")
+    }
+
+    /** Commits [txnId] if [block] returns normally, aborts it (then rethrows) if [block] throws. */
+    private inline fun <T> runInTransaction(txnId: Long, block: () -> T): T {
+        val result = try {
+            block()
+        } catch (failure: Exception) {
+            catalog.abort(txnId)
+            throw failure
+        }
+        catalog.commit(txnId)
+        return result
+    }
+
+    private fun isTrue(value: Value): Boolean = value is Value.BooleanValue && value.value
+
+    private fun coerceForColumn(value: Value, column: Column): Value {
+        if (value is Value.NullValue) return value
+        return when (column.type) {
+            ColumnType.Integer -> value as? Value.IntegerValue ?: throw typeMismatch(column, value)
+            ColumnType.Text -> value as? Value.TextValue ?: throw typeMismatch(column, value)
+            ColumnType.Boolean -> value as? Value.BooleanValue ?: throw typeMismatch(column, value)
+            ColumnType.Real -> when (value) {
+                is Value.RealValue -> value
+                is Value.IntegerValue -> Value.RealValue(value.value.toDouble())
+                else -> throw typeMismatch(column, value)
+            }
+            ColumnType.Decimal -> {
+                val raw = when (value) {
+                    is Value.DecimalValue -> value.value
+                    is Value.IntegerValue -> BigDecimal(value.value)
+                    is Value.RealValue -> BigDecimal.valueOf(value.value)
+                    else -> throw typeMismatch(column, value)
+                }
+                Value.DecimalValue(DecimalColumns.normalize(raw, column.precision, column.scale, "column ${column.name}"))
+            }
+            ColumnType.Date -> value as? Value.DateValue ?: throw typeMismatch(column, value)
+        }
+    }
+
+    private fun typeMismatch(column: Column, value: Value): DiskExecutorException =
+        DiskExecutorException("type mismatch: column ${column.name} expects ${column.type.displayName} but got ${value.typeName}")
+
+    companion object {
+        private const val MAX_RECURSION_ITERATIONS = 10_000
+        private const val MAX_TRIGGER_DEPTH = 100
+
+        /** Kinds whose result actually depends on which rows fall inside the frame. */
+        private val FRAME_CAPABLE_KINDS = setOf(WindowFunctionKind.Aggregate, WindowFunctionKind.FirstValue, WindowFunctionKind.LastValue)
+    }
+}