feat(sql): add initialRows and applyJoin to DiskExecutor
This commit is contained in:
dualfroz committed
1 parent
b2e64dc4f5
commit
f6037d6772
1 file changed
+150
@@ -2888,3 +2888,153 @@ class DiskExecutor(private val catalog: DiskCatalog) : SequenceAccess {
|
||||
is SubquerySource -> item.alias
|
||||
}
|
||||
|
||||
/**
|
||||
* The candidate rows a query's FIRST source starts from: an indexed lookup narrowed by WHERE
|
||||
* via [candidateRows] when there is no JOIN and the source is a real disk table (preserving the
|
||||
* same pushdown a plain single-table SELECT already gets), or a full scan otherwise - mirroring
|
||||
* [kestrel.engine.SelectRunner]'s own choice to fall back to a full scan for the driving table
|
||||
* as soon as more than one source is involved, since WHERE may combine columns from sources
|
||||
* that are not open yet at this point in the plan. A CTE-backed source has no index to push
|
||||
* down into regardless of JOIN, so it always scans its already-materialized rows; either way
|
||||
* [statement.where] is re-applied in full immediately after by the caller, so skipping pushdown
|
||||
* here only costs performance, never correctness.
|
||||
*/
|
||||
private fun initialRows(statement: SelectStatement, txn: Long, item: FromItem, fromSource: DiskFromSource): List<List<Value>> =
|
||||
if (statement.joins.isEmpty() && fromSource is DiskTableSource) {
|
||||
candidateRows(fromSource.table, txn, statement.where, (item as TableSource).alias).map { it.second.values }
|
||||
} else {
|
||||
fromSource.scan(txn)
|
||||
}
|
||||
|
||||
/**
|
||||
* Nested-loop join: for every row already combined from earlier sources, scans the whole of
|
||||
* [rightRows] evaluating [JoinClause.on] against each pairing. A LEFT/FULL JOIN row left
|
||||
* unmatched still contributes one row, padded with NULLs for [rightSource]'s columns; a RIGHT/
|
||||
* FULL JOIN additionally emits every [rightRows] entry that never matched any [left] row, padded
|
||||
* with NULLs for everything [rightSource.offset] counts as coming before it. Deliberately no
|
||||
* hash-join optimization for an equality condition (unlike [kestrel.engine.SelectRunner]'s
|
||||
* `hashJoinKeys`) - a documented scope cut: this JOIN exists to be correct first, in a solo
|
||||
* session without peer review, and every disk-backed table already pays a full heap scan for
|
||||
* any predicate an index cannot serve, so nested loop is not a new order-of-magnitude cost here.
|
||||
*/
|
||||
private fun applyJoin(
|
||||
join: JoinClause,
|
||||
left: List<List<Value>>,
|
||||
rightRows: List<List<Value>>,
|
||||
sourcesSoFar: List<JoinSource>,
|
||||
outer: ColumnResolver?,
|
||||
subqueries: SubqueryRunner,
|
||||
rightSource: JoinSource
|
||||
): List<List<Value>> {
|
||||
val nullRight = List(rightSource.schema.columns.size) { Value.NullValue }
|
||||
val nullLeft = List(rightSource.offset) { Value.NullValue }
|
||||
val outerLeft = join.type == JoinType.Left || join.type == JoinType.Full
|
||||
val outerRight = join.type == JoinType.Right || join.type == JoinType.Full
|
||||
val condition = join.on
|
||||
val result = mutableListOf<List<Value>>()
|
||||
val matchedRight = BooleanArray(rightRows.size)
|
||||
for (leftRow in left) {
|
||||
var matched = false
|
||||
rightRows.forEachIndexed { index, rightRow ->
|
||||
val candidate = leftRow + rightRow
|
||||
if (condition == null || isTrue(evaluateCombinedRow(sourcesSoFar, outer, subqueries, condition, candidate))) {
|
||||
result.add(candidate)
|
||||
matched = true
|
||||
matchedRight[index] = true
|
||||
}
|
||||
}
|
||||
if (!matched && outerLeft) result.add(leftRow + nullRight)
|
||||
}
|
||||
if (outerRight) {
|
||||
rightRows.forEachIndexed { index, rightRow -> if (!matchedRight[index]) result.add(nullLeft + rightRow) }
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
private fun lateralSelectOf(item: FromItem): QueryStatement? = (item as? SubquerySource)?.takeIf { it.lateral }?.select
|
||||
|
||||
/**
|
||||
* Unlike every other join, a LATERAL derived table has no fixed row set: it is re-evaluated
|
||||
* once per row already accumulated on the left, correlated to that row via [lateralOuterResolver]
|
||||
* exactly like a correlated subquery in WHERE/SELECT (same [SubqueryRunner]/[Evaluator] plumbing,
|
||||
* just invoked from FROM instead) - mirrors [kestrel.engine.SelectRunner.executeLateralJoin] 1:1.
|
||||
* Hash-joining is meaningless here (there is no static right-hand row set to build buckets from),
|
||||
* so this is always a nested loop, matching how real engines execute LATERAL.
|
||||
*/
|
||||
private fun executeLateralJoin(
|
||||
join: JoinClause,
|
||||
left: List<List<Value>>,
|
||||
lateralSelect: QueryStatement,
|
||||
sourcesSoFar: List<JoinSource>,
|
||||
rightSource: JoinSource,
|
||||
outer: ColumnResolver?,
|
||||
subqueries: SubqueryRunner,
|
||||
txn: Long
|
||||
): List<List<Value>> {
|
||||
val nullRight = List(rightSource.schema.columns.size) { Value.NullValue }
|
||||
val allSources = sourcesSoFar + rightSource
|
||||
val condition = join.on
|
||||
val result = mutableListOf<List<Value>>()
|
||||
for (leftRow in left) {
|
||||
val rightRows = runQuery(lateralSelect, txn, lateralOuterResolver(sourcesSoFar, leftRow)).rows
|
||||
var matched = false
|
||||
for (rightRow in rightRows) {
|
||||
val candidate = leftRow + rightRow
|
||||
if (condition == null || isTrue(evaluateCombinedRow(allSources, outer, subqueries, condition, candidate))) {
|
||||
result.add(candidate)
|
||||
matched = true
|
||||
}
|
||||
}
|
||||
if (!matched && join.type == JoinType.Left) result.add(leftRow + nullRight)
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolves a column against [sourcesSoFar] - everything to the left of a LATERAL subquery in the
|
||||
* FROM clause - and nothing else: a name that does not resolve there means the expression names a
|
||||
* source not joined yet (or the LATERAL item itself), which SQL forbids, reported with a clear
|
||||
* LATERAL-specific message rather than the generic "no such column".
|
||||
*/
|
||||
private fun lateralOuterResolver(sourcesSoFar: List<JoinSource>, leftRow: List<Value>): ColumnResolver = object : ColumnResolver {
|
||||
override fun resolve(expression: ColumnExpression): Value =
|
||||
resolveIndexOrNull(sourcesSoFar, expression)?.let { leftRow[it] }
|
||||
?: throw DiskExecutorException(
|
||||
"LATERAL subquery cannot reference ${expression.name}: not joined yet at this point in the FROM clause"
|
||||
)
|
||||
|
||||
override fun declaredCollation(expression: ColumnExpression): Collation? =
|
||||
resolveIndexOrNull(sourcesSoFar, expression)?.let { declaredCollationOrNull(sourcesSoFar, expression) }
|
||||
}
|
||||
|
||||
private fun resolveIndexOrNull(sources: List<JoinSource>, expression: ColumnExpression): Int? {
|
||||
val qualifier = expression.qualifier
|
||||
val matches = mutableListOf<Int>()
|
||||
for (source in sources) {
|
||||
if (qualifier != null && !source.alias.equals(qualifier, ignoreCase = true)) continue
|
||||
val local = source.schema.columnIndex(expression.name) ?: continue
|
||||
matches.add(source.offset + local)
|
||||
}
|
||||
return when {
|
||||
matches.isEmpty() -> null
|
||||
matches.size > 1 -> throw DiskExecutorException("ambiguous column reference: ${expression.name}")
|
||||
else -> matches.single()
|
||||
}
|
||||
}
|
||||
|
||||
private fun resolveIndex(sources: List<JoinSource>, expression: ColumnExpression): Int =
|
||||
resolveIndexOrNull(sources, expression)
|
||||
?: throw DiskExecutorException("no such column: ${expression.name}")
|
||||
|
||||
/** [expression]'s declared `COLLATE` from whichever [sources] entry it resolves against - mirrors [resolveIndexOrNull]'s own matching, `null` on no match or an ambiguous one. */
|
||||
private fun declaredCollationOrNull(sources: List<JoinSource>, expression: ColumnExpression): Collation? {
|
||||
val qualifier = expression.qualifier
|
||||
val matches = mutableListOf<Collation>()
|
||||
for (source in sources) {
|
||||
if (qualifier != null && !source.alias.equals(qualifier, ignoreCase = true)) continue
|
||||
val column = source.schema.column(expression.name) ?: continue
|
||||
matches.add(column.collation)
|
||||
}
|
||||
return matches.singleOrNull()
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user