feat(sql): add evaluateCombinedRow and resolve to DiskExecutor
This commit is contained in:
dualfroz committed
1 parent
f6037d6772
commit
bb57f5ac91
1 file changed
+149
@@ -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<JoinSource>,
|
||||
outer: ColumnResolver?,
|
||||
subqueries: SubqueryRunner,
|
||||
expression: Expression,
|
||||
row: List<Value>
|
||||
): 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<JoinSource>,
|
||||
outer: ColumnResolver?,
|
||||
subqueries: SubqueryRunner,
|
||||
rows: List<List<Value>>,
|
||||
orderKeys: List<OrderKey>,
|
||||
windowExpressions: List<WindowExpression>,
|
||||
baseColumnCount: Int
|
||||
): List<List<Value>> {
|
||||
val keyed = rows.map { row ->
|
||||
row to orderKeys.map { key -> evaluateWindowAware(sources, outer, subqueries, windowExpressions, baseColumnCount, key.expression, row) }
|
||||
}
|
||||
val comparator = Comparator<List<Value>> { 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<JoinSource>,
|
||||
outer: ColumnResolver?,
|
||||
subqueries: SubqueryRunner,
|
||||
windowExpressions: List<WindowExpression>,
|
||||
baseColumnCount: Int,
|
||||
expression: Expression,
|
||||
row: List<Value>
|
||||
): 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<Value>,
|
||||
windowExpressions: List<WindowExpression>,
|
||||
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<JoinSource>,
|
||||
outer: ColumnResolver?,
|
||||
subqueries: SubqueryRunner,
|
||||
rows: List<List<Value>>,
|
||||
windowExpressions: List<WindowExpression>,
|
||||
baseColumnCount: Int
|
||||
): Pair<List<String>, List<List<Value>>> =
|
||||
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) }
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user