From bb57f5ac910d9ae43a2de530f1432f9cfa182d2e Mon Sep 17 00:00:00 2001 From: DualFroz Date: Mon, 7 Sep 2026 14:35:36 +0200 Subject: [PATCH] feat(sql): add evaluateCombinedRow and resolve to DiskExecutor --- .../kotlin/kestrel/store/sql/DiskExecutor.kt | 149 ++++++++++++++++++ 1 file changed, 149 insertions(+) diff --git a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt index daf858f..e8a8ab2 100644 --- a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt +++ b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt @@ -3038,3 +3038,152 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess { return matches.singleOrNull() } + /** + * Resolves [expression]'s columns against [sources] first, falling back to [outer] (the + * enclosing statement's row, when this call is itself part of a correlated subquery) exactly + * once a name cannot be found locally - mirroring [kestrel.engine.SelectRunner]'s own + * `rowResolver`. The resulting [Evaluator] carries [subqueries] so any [ScalarSubqueryExpression]/ + * [ExistsExpression]/[InSubqueryExpression] nested inside [expression] can itself run, correlated + * back to this very row through that same resolver. + */ + private fun evaluateCombinedRow( + sources: List, + outer: ColumnResolver?, + subqueries: SubqueryRunner, + expression: Expression, + row: List + ): Value { + val resolver = object : ColumnResolver { + override fun resolve(columnExpression: ColumnExpression): Value { + val index = resolveIndexOrNull(sources, columnExpression) + return if (index != null) { + row[index] + } else { + (outer ?: throw DiskExecutorException("no such column: ${columnExpression.name}")).resolve(columnExpression) + } + } + + override fun declaredCollation(columnExpression: ColumnExpression): Collation? = + declaredCollationOrNull(sources, columnExpression) ?: outer?.declaredCollation(columnExpression) + } + return Evaluator(resolver, subqueries).evaluate(expression) + } + + private fun sortCombinedRows( + sources: List, + outer: ColumnResolver?, + subqueries: SubqueryRunner, + rows: List>, + orderKeys: List, + windowExpressions: List, + baseColumnCount: Int + ): List> { + val keyed = rows.map { row -> + row to orderKeys.map { key -> evaluateWindowAware(sources, outer, subqueries, windowExpressions, baseColumnCount, key.expression, row) } + } + val comparator = Comparator> { left, right -> + for ((index, key) in orderKeys.withIndex()) { + val comparison = compareOrderKeyValues(left[index], right[index], key) { expression -> + declaredCollationOrNull(sources, expression) + } + if (comparison != 0) return@Comparator comparison + } + 0 + } + return keyed.sortedWith(compareBy(comparator) { it.second }).map { it.first } + } + + /** + * When [windowExpressions] is non-empty, [row] may carry extra trailing values appended to + * [combined][runSelect] (one per entry, right after the real columns) - rewrites any + * [WindowExpression] node in [expression] into a [LiteralExpression] reading straight from + * that trailing slot before [evaluateCombinedRow] ever sees it, mirroring + * [kestrel.engine.SelectRunner]'s own `evaluateRow`/`substituteWindowValues`. + */ + private fun evaluateWindowAware( + sources: List, + outer: ColumnResolver?, + subqueries: SubqueryRunner, + windowExpressions: List, + baseColumnCount: Int, + expression: Expression, + row: List + ): Value { + val prepared = if (windowExpressions.isEmpty()) expression else substituteWindowValues(expression, row, windowExpressions, baseColumnCount) + return evaluateCombinedRow(sources, outer, subqueries, prepared, row) + } + + private fun substituteWindowValues( + expression: Expression, + row: List, + windowExpressions: List, + baseColumnCount: Int + ): Expression { + fun recurse(inner: Expression) = substituteWindowValues(inner, row, windowExpressions, baseColumnCount) + return when (expression) { + is WindowExpression -> LiteralExpression(row[baseColumnCount + windowExpressions.indexOf(expression)]) + is BinaryExpression -> expression.copy(left = recurse(expression.left), right = recurse(expression.right)) + is UnaryExpression -> expression.copy(operand = recurse(expression.operand)) + is CallExpression -> expression.copy(arguments = expression.arguments.map(::recurse)) + is IsNullExpression -> expression.copy(operand = recurse(expression.operand)) + is InExpression -> expression.copy(operand = recurse(expression.operand), values = expression.values.map(::recurse)) + is BetweenExpression -> expression.copy(operand = recurse(expression.operand), low = recurse(expression.low), high = recurse(expression.high)) + is LikeExpression -> expression.copy(operand = recurse(expression.operand), pattern = recurse(expression.pattern)) + is CaseExpression -> expression.copy( + operand = expression.operand?.let(::recurse), + branches = expression.branches.map { it.copy(condition = recurse(it.condition), result = recurse(it.result)) }, + elseExpression = expression.elseExpression?.let(::recurse) + ) + is CastExpression -> expression.copy(operand = recurse(expression.operand)) + is CollateExpression -> expression.copy(operand = recurse(expression.operand)) + else -> expression + } + } + + private fun projectCombined( + projection: Projection, + sources: List, + outer: ColumnResolver?, + subqueries: SubqueryRunner, + rows: List>, + windowExpressions: List, + baseColumnCount: Int + ): Pair, List>> = + when (projection) { + // rows may carry trailing window-function values appended in runSelect (only reachable + // here if ORDER BY names one directly, since SELECT * cannot spell an OVER clause + // itself) - trimmed back off so the row width matches the schema's own columns. + StarProjection -> sources.flatMap { it.schema.columnNames } to rows.map { it.subList(0, baseColumnCount) } + is ExpressionProjection -> { + val columns = projection.items.mapIndexed { index, item -> columnLabel(item, index) } + val values = rows.map { row -> + projection.items.map { item -> + evaluateWindowAware(sources, outer, subqueries, windowExpressions, baseColumnCount, item.expression, row) + } + } + columns to values + } + } + + private fun columnLabel(item: SelectItem, index: Int): String = + item.alias ?: when (val expression = item.expression) { + is ColumnExpression -> expression.name + is FunctionExpression -> expression.name.lowercase() + is CallExpression -> expression.name.lowercase() + else -> "column${index + 1}" + } + + /** + * Deliberately duplicated from [kestrel.engine.SelectRunner]'s own `containsAggregate`/ + * `requireGrouped` (same reasoning as [coerceForColumn]): this executor must stay free-standing + * from the in-memory engine's own validation, so a change to one can never silently affect the + * other. `else -> false` covers every [Expression] variant this executor gives no + * aggregation-aware meaning to (a literal, a window function - rejected whenever it is actually + * evaluated, by [Evaluator] itself - or a subquery, whose own aggregates belong to that nested + * [SelectStatement]'s independent [runSelect] call, never to this one). + */ + private fun containsAggregate(projection: Projection): Boolean = when (projection) { + StarProjection -> false + is ExpressionProjection -> projection.items.any { containsAggregate(it.expression) } + } +