feat(engine): extend SelectRunner with rowResolver and resolve

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

@@ -662,10 +662,11 @@ class SelectRunner(private val catalog: Catalog) {
                 statement.projection.items.forEach { requireGrouped(it.expression, groupColumns) }
             }
             statement.having?.let { requireGrouped(it, groupColumns) }
-            orderExpression?.let { requireGrouped(it, groupColumns) }
+            orderKeys.forEach { requireGrouped(it.expression, groupColumns) }
         }
 
         private fun requireGrouped(expression: Expression, groupColumns: Set<Int>) {
+            if (expression in statement.groupBy) return
             when (expression) {
                 is FunctionExpression -> Unit
                 is ColumnExpression -> if (resolveIndex(expression) !in groupColumns) {
@@ -679,14 +680,107 @@ class SelectRunner(private val catalog: Catalog) {
                 }
                 is UnaryExpression -> requireGrouped(expression.operand, groupColumns)
                 is LiteralExpression -> Unit
+                is CallExpression -> expression.arguments.forEach { requireGrouped(it, groupColumns) }
+                is IsNullExpression -> requireGrouped(expression.operand, groupColumns)
+                is InExpression -> {
+                    requireGrouped(expression.operand, groupColumns)
+                    expression.values.forEach { requireGrouped(it, groupColumns) }
+                }
+                is BetweenExpression -> {
+                    requireGrouped(expression.operand, groupColumns)
+                    requireGrouped(expression.low, groupColumns)
+                    requireGrouped(expression.high, groupColumns)
+                }
+                is LikeExpression -> {
+                    requireGrouped(expression.operand, groupColumns)
+                    requireGrouped(expression.pattern, groupColumns)
+                }
+                is CaseExpression -> {
+                    expression.operand?.let { requireGrouped(it, groupColumns) }
+                    expression.branches.forEach {
+                        requireGrouped(it.condition, groupColumns)
+                        requireGrouped(it.result, groupColumns)
+                    }
+                    expression.elseExpression?.let { requireGrouped(it, groupColumns) }
+                }
+                is ScalarSubqueryExpression -> Unit
+                is ExistsExpression -> Unit
+                is InSubqueryExpression -> requireGrouped(expression.operand, groupColumns)
+                is CastExpression -> requireGrouped(expression.operand, groupColumns)
+                is CollateExpression -> requireGrouped(expression.operand, groupColumns)
+                // Unreachable: validateWindows() rejects mixing window functions with GROUP BY/
+                // aggregates before requireGrouped ever runs, so this branch only keeps `when` exhaustive.
+                is WindowExpression -> Unit
+                // Unreachable in practice: NEXTVAL/CURRVAL/MERGE_ACTION() are never legal in a SELECT
+                // at all (MERGE's RETURNING is evaluated by Executor.evaluateMergeReturning, an
+                // entirely separate projection path from this one) - kept here only to keep `when`
+                // exhaustive.
+                is NextValueExpression, is CurrentValueExpression, is MergeActionExpression -> Unit
+                // A parameter reference behaves like any other constant here - `SELECT $1, SUM(x)
+                // FROM t GROUP BY y` needs $1 in GROUP BY no more than a literal would.
+                is ParameterExpression -> Unit
             }
         }
 
-        private fun rowResolver(row: List<Value>) =
+        private fun rowResolver(row: List<Value>) = object : kestrel.engine.ColumnResolver {
-            kestrel.engine.ColumnResolver { expression -> row[resolveIndex(expression)] }
+            override fun resolve(expression: ColumnExpression): Value {
+                val index = resolveIndexOrNull(expression)
+                return if (index != null) {
+                    row[index]
+                } else {
+                    (outer ?: throw KestrelException("no such column: ${describe(expression)}")).resolve(expression)
+                }
+            }
 
-        private fun evaluateRow(expression: Expression, row: List<Value>): Value =
+            override fun declaredCollation(expression: ColumnExpression): Collation? =
-            Evaluator(rowResolver(row)).evaluate(expression)
+                declaredCollationOrNull(expression) ?: outer?.declaredCollation(expression)
+        }
+
+        /**
+         * When [windowExpressions] is non-empty, a [row] handed in here may carry extra trailing
+         * values appended by [appendWindowColumns] (one per entry of [windowExpressions], in order,
+         * right after the real columns) - [substituteWindowValues] rewrites any window-function node
+         * in [expression] into a [LiteralExpression] reading straight from that trailing slot before
+         * the plain [Evaluator] ever sees it, the same trick [GroupAggregation] uses for aggregates.
+         */
+        private fun evaluateRow(expression: Expression, row: List<Value>): Value {
+            val prepared = if (windowExpressions.isEmpty()) expression else substituteWindowValues(expression, row)
+            return Evaluator(rowResolver(row), subqueryRunner).evaluate(prepared)
+        }
+
+        private fun substituteWindowValues(expression: Expression, row: List<Value>): Expression = when (expression) {
+            is WindowExpression -> LiteralExpression(row[columnNames.size + windowExpressions.indexOf(expression)])
+            is BinaryExpression -> expression.copy(
+                left = substituteWindowValues(expression.left, row),
+                right = substituteWindowValues(expression.right, row)
+            )
+            is UnaryExpression -> expression.copy(operand = substituteWindowValues(expression.operand, row))
+            is CallExpression -> expression.copy(arguments = expression.arguments.map { substituteWindowValues(it, row) })
+            is IsNullExpression -> expression.copy(operand = substituteWindowValues(expression.operand, row))
+            is InExpression -> expression.copy(
+                operand = substituteWindowValues(expression.operand, row),
+                values = expression.values.map { substituteWindowValues(it, row) }
+            )
+            is BetweenExpression -> expression.copy(
+                operand = substituteWindowValues(expression.operand, row),
+                low = substituteWindowValues(expression.low, row),
+                high = substituteWindowValues(expression.high, row)
+            )
+            is LikeExpression -> expression.copy(
+                operand = substituteWindowValues(expression.operand, row),
+                pattern = substituteWindowValues(expression.pattern, row)
+            )
+            is CaseExpression -> expression.copy(
+                operand = expression.operand?.let { substituteWindowValues(it, row) },
+                branches = expression.branches.map {
+                    it.copy(condition = substituteWindowValues(it.condition, row), result = substituteWindowValues(it.result, row))
+                },
+                elseExpression = expression.elseExpression?.let { substituteWindowValues(it, row) }
+            )
+            is CastExpression -> expression.copy(operand = substituteWindowValues(expression.operand, row))
+            is CollateExpression -> expression.copy(operand = substituteWindowValues(expression.operand, row))
+            else -> expression
+        }
 
         fun materialize(): List<List<Value>> {
             var combined = primaryRows().map { it.values }