feat(engine): extend Executor with checkMergeColumnReference

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

@@ -1977,6 +1977,61 @@ class Executor(private val catalog: Catalog) : SequenceAccess {
                 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 (no column reference to check) - none of ON/WHEN's
+            // AND-condition/assignments/INSERT values are where MERGE_ACTION() is ever legal (only
+            // this statement's own RETURNING is, see evaluateMergeReturning), and it is never handed
+            // a bound action there, so evaluation itself throws - same division of labor as
+            // checkColumns's own NEXTVAL/CURRVAL branch. A parameter reference IS legitimately allowed
+            // here (MERGE is one of PreparableStatement's own kinds) - by the time this runs it has
+            // already been substituted for a literal by resolveExecute, so this branch only ever sees
+            // one directly typed outside PREPARE, which is just as harmless structurally.
+            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 KestrelException("no target row is available in a WHEN NOT MATCHED clause: $targetLabel.${expression.name}")
+            }
+            if (!matchesSource && !matchesTargetLabel) {
+                throw KestrelException("no such table or alias: $qualifier")
+            }
+            val schema = if (matchesSource) sourceSchema else targetSchema
+            if (schema.columnIndex(expression.name) == null) throw KestrelException("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 KestrelException("column reference '${expression.name}' is ambiguous between $targetLabel and $sourceLabel")
+        }
+        if (!inTarget && !inSource) throw KestrelException("no such column: ${expression.name}")
+    }
 
     private fun coerceForColumn(value: Value, column: Column): Value {
         if (value is Value.NullValue) return value
@@ -1989,6 +2044,16 @@ class Executor(private val catalog: Catalog) : SequenceAccess {
                 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)
         }
     }
 
@@ -2004,28 +2069,130 @@ class Executor(private val catalog: Catalog) : SequenceAccess {
     }
 
     private fun enforcePrimaryKeyUniqueness(schema: TableSchema, rows: List<Row>) {
-        val primaryKeyIndex = schema.primaryKeyIndex ?: return
+        val indices = schema.primaryKeyIndices
-        val seen = HashSet<Value>()
+        if (indices.isEmpty()) return
+        val seen = HashSet<List<Value>>()
         for (row in rows) {
-            val key = row.values[primaryKeyIndex]
+            val key = collationFoldedUniquenessKey(schema, indices, indices.map { row.values[it] })
             if (!seen.add(key)) {
-                throw KestrelException("UNIQUE constraint failed: ${schema.name}.${schema.columns[primaryKeyIndex].name}")
+                val columnList = indices.joinToString(", ") { schema.columns[it].name }
+                throw KestrelException("UNIQUE constraint failed: ${schema.name}($columnList)")
             }
         }
     }
 
-    private fun checkColumns(expression: Expression, schema: TableSchema) {
+    /**
+     * A row where any constrained column is NULL is exempt (standard SQL: NULLs never collide
+     * with each other), so such rows are simply skipped rather than added to [seen].
+     */
+    private fun enforceUniqueConstraints(schema: TableSchema, rows: List<Row>) {
+        for (unique in schema.uniqueConstraints) {
+            val columnIndices = unique.columns.map { schema.columnIndex(it)!! }
+            val seen = HashSet<List<Value>>()
+            for (row in rows) {
+                val rawKey = columnIndices.map { row.values[it] }
+                if (rawKey.any { it is Value.NullValue }) continue
+                if (!seen.add(collationFoldedUniquenessKey(schema, columnIndices, rawKey))) {
+                    val label = unique.name?.let { " $it" } ?: ""
+                    val columnList = unique.columns.joinToString(", ")
+                    throw KestrelException("UNIQUE constraint$label failed: ${schema.name}($columnList)")
+                }
+            }
+        }
+    }
+
+    /**
+     * A `CREATE UNIQUE INDEX` is a way to attach a uniqueness guarantee to a table AFTER it
+     * already exists (the engine has no ALTER TABLE), so it is checked the same way as a
+     * declarative UNIQUE constraint - against the whole candidate row set, before any row is
+     * written - rather than relying on `SortedIndex.add` to reject a duplicate during the
+     * post-write `catalog.reindexTable` rebuild, which would leave the table already mutated.
+     * This key is folded through [collationFoldedUniquenessKey] for the same reason
+     * [enforcePrimaryKeyUniqueness]/[enforceUniqueConstraints] are: since `SortedIndex`'s own
+     * comparator is *also* collation-aware (see [kestrel.catalog.IndexKeyOrdering]), skipping the
+     * fold here would not merely miss the duplicate - `table.rows.addAll`/`.add` would already have
+     * run by the time `reindexTable` calls `SortedIndex.add` and throws on the same duplicate, and
+     * for a trigger-free table [withStatementAtomicity] takes no snapshot, so that later throw would
+     * leave the just-inserted duplicate permanently stuck in `table.rows`, indexed nowhere - a real,
+     * reproduced-during-development corruption mode, not a hypothetical one.
+     */
+    private fun enforceUniqueIndexes(schema: TableSchema, rows: List<Row>) {
+        for (definition in catalog.indexDefinitions()) {
+            if (!definition.unique || !definition.tableName.equals(schema.name, ignoreCase = true)) continue
+            val columnIndices = definition.columnNames.mapNotNull { schema.columnIndex(it) }
+            if (columnIndices.size != definition.columnNames.size) continue
+            val seen = HashSet<List<Value>>()
+            for (row in rows) {
+                val rawKey = columnIndices.map { row.values[it] }
+                if (rawKey.any { it is Value.NullValue }) continue
+                if (!seen.add(collationFoldedUniquenessKey(schema, columnIndices, rawKey))) {
+                    throw KestrelException("UNIQUE constraint failed: ${schema.name}(${definition.columnDisplayName})")
+                }
+            }
+        }
+    }
+
+    /**
+     * `triggerContext` lets a `NEW.col`/`OLD.col` reference inside a trigger body statement pass
+     * validation against the trigger's OWN table (via [TriggerRowContext.checkColumnExists])
+     * instead of being checked - wrongly - against [schema], the body statement's own target
+     * table, which is often a different table entirely.
+     */
+    private fun checkColumns(expression: Expression, schema: TableSchema, triggerContext: TriggerRowContext? = null) {
         when (expression) {
-            is ColumnExpression -> if (schema.columnIndex(expression.name) == null) {
+            is ColumnExpression -> if (triggerContext != null && triggerContext.matches(expression)) {
+                triggerContext.checkColumnExists(expression)
+            } else if (schema.columnIndex(expression.name) == null) {
                 throw KestrelException("no such column: ${expression.name}")
             }
             is BinaryExpression -> {
-                checkColumns(expression.left, schema)
+                checkColumns(expression.left, schema, triggerContext)
-                checkColumns(expression.right, schema)
+                checkColumns(expression.right, schema, triggerContext)
             }
-            is UnaryExpression -> checkColumns(expression.operand, schema)
+            is UnaryExpression -> checkColumns(expression.operand, schema, triggerContext)
             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 { checkColumns(it, schema, triggerContext) }
+            is IsNullExpression -> checkColumns(expression.operand, schema, triggerContext)
+            is InExpression -> {
+                checkColumns(expression.operand, schema, triggerContext)
+                expression.values.forEach { checkColumns(it, schema, triggerContext) }
+            }
+            is BetweenExpression -> {
+                checkColumns(expression.operand, schema, triggerContext)
+                checkColumns(expression.low, schema, triggerContext)
+                checkColumns(expression.high, schema, triggerContext)
+            }
+            is LikeExpression -> {
+                checkColumns(expression.operand, schema, triggerContext)
+                checkColumns(expression.pattern, schema, triggerContext)
+            }
+            is CaseExpression -> {
+                expression.operand?.let { checkColumns(it, schema, triggerContext) }
+                expression.branches.forEach {
+                    checkColumns(it.condition, schema, triggerContext)
+                    checkColumns(it.result, schema, triggerContext)
+                }
+                expression.elseExpression?.let { checkColumns(it, schema, triggerContext) }
+            }
+            is ScalarSubqueryExpression -> Unit
+            is ExistsExpression -> Unit
+            is InSubqueryExpression -> checkColumns(expression.operand, schema, triggerContext)
+            is CastExpression -> checkColumns(expression.operand, schema, triggerContext)
+            is CollateExpression -> checkColumns(expression.operand, schema, triggerContext)
+            // Structurally harmless either way (no column reference to check) - whether NEXTVAL/
+            // CURRVAL is actually legal at this particular call site (SET assignments: yes; WHERE:
+            // no, see Evaluator's SequenceAccess doc) is enforced by whether the Evaluator this
+            // feeds into was actually given a SequenceAccess, not by this column-existence pass.
+            // MERGE_ACTION() is the same story: this generic checkColumns also backs plain INSERT/
+            // UPDATE/DELETE's own RETURNING (evaluateReturning), where it is never legal - but that is
+            // enforced by evaluateReturning's Evaluator never being handed a mergeAction, exactly like
+            // evaluateMergeReturning is the one call site that does. A parameter reference is
+            // legitimately allowed anywhere this generic pass runs (INSERT/UPDATE/DELETE/SELECT are
+            // all PreparableStatement kinds) - already substituted for a literal by the time a real
+            // EXECUTE reaches here.
+            is NextValueExpression, is CurrentValueExpression, is MergeActionExpression, is ParameterExpression -> Unit
         }
     }