From f6037d6772e637453af9eaddcf0818d8d77271bc Mon Sep 17 00:00:00 2001 From: DualFroz Date: Mon, 7 Sep 2026 13:51:27 +0200 Subject: [PATCH] feat(sql): add initialRows and applyJoin to DiskExecutor --- .../kotlin/kestrel/store/sql/DiskExecutor.kt | 150 ++++++++++++++++++ 1 file changed, 150 insertions(+) diff --git a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt index fb7a2ba..daf858f 100644 --- a/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt +++ b/src/main/kotlin/kestrel/store/sql/DiskExecutor.kt @@ -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> = + 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>, + rightRows: List>, + sourcesSoFar: List, + outer: ColumnResolver?, + subqueries: SubqueryRunner, + rightSource: JoinSource + ): List> { + 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>() + 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>, + lateralSelect: QueryStatement, + sourcesSoFar: List, + rightSource: JoinSource, + outer: ColumnResolver?, + subqueries: SubqueryRunner, + txn: Long + ): List> { + val nullRight = List(rightSource.schema.columns.size) { Value.NullValue } + val allSources = sourcesSoFar + rightSource + val condition = join.on + val result = mutableListOf>() + 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, leftRow: List): 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, expression: ColumnExpression): Int? { + val qualifier = expression.qualifier + val matches = mutableListOf() + 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, 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, expression: ColumnExpression): Collation? { + val qualifier = expression.qualifier + val matches = mutableListOf() + 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() + } +