feat: Single join for correlated scalar subqueries - #24782
Closed
Dandandan wants to merge 3 commits into
Closed
Conversation
Correlated scalar subqueries must return at most one row per set of outer values. DataFusion enforced that syntactically: the analyzer rejected any correlated scalar subquery that did not have an aggregate on top, with "Correlated scalar subquery must be aggregated to return at most one row". So a lookup like select o_orderkey, (select c_name from customer where c_custkey = o_custkey) from orders did not plan at all, and had to be written with a redundant aggregate (min/max/any_value) whose only job was to prove the row count -- paying a full hash aggregation over the subquery side. This adds the "single join" of Neumann and Kemper's unnesting paper as JoinType::LeftSingle / JoinType::RightSingle: a left/right outer join that emits exactly one row per row of its preserved side and raises "Scalar subquery returned more than one row" when a second row matches. ScalarSubqueryToJoin now emits it for the subqueries whose shape does not already guarantee the property, so those queries decorrelate into a plain join with no aggregate at all. Subqueries that do guarantee it -- an aggregate grouped only by columns the correlated predicate fixes -- keep their plain LEFT JOIN, so every plan that worked before is byte-identical, including TPC-H q2/q17/q20 and TPC-DS q1/q6/q30/q32/q41/q81/q92. Implemented in HashJoinExec (both build- and probe-driven, so the join can still be swapped for build-side selection) and NestedLoopJoinExec. The matched bitmaps double as duplicate detectors, so the check costs one already-cached bit test per matched row and covers matches spread across probe batches and partitions. SortMergeJoinExec and PiecewiseMergeJoin reject single joins so the planner falls back to a join that implements them. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SgqwvctZdvR1ZCz2hbkEJC
Adds the single-join section of `subquery.slt` -- attribute lookup with and
without a match, the more-than-one-row error, the same in a filter rather than
a projection, the nested-loop path for a correlation with no equijoin key, and
a plan assertion that a provably-single-row subquery still gets a plain left
join -- plus `HashJoinExec` and `NestedLoopJoinExec` unit tests covering both
directions across partition modes and batch sizes.
Writing them found two gaps, both fixed here:
- `projection_pushdown` and `NestedLoopJoinExec::build_unmatched_batch` hit
catch-all arms for the new join types, panicking on a correlation with no
equijoin key.
- `NestedLoopJoinStream::update_matched_bitmap`, the path taken when the
right batch is large relative to the batch size, set the matched bitmaps
without checking them, so a second match went unreported.
Decorrelation still cannot pull a correlated predicate through a `LIMIT`, and
those subqueries no longer hit the "must be aggregated" check that used to
report them. They now say so directly instead of reaching the physical planner
as a leftover `ScalarSubquery` expression.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SgqwvctZdvR1ZCz2hbkEJC
… the keys
Preferring `LEFT JOIN` over a single join matters for more than the runtime
check. A left join lets the optimizer turn the join inner when a predicate
above it rejects nulls, fold a comparison against an outer column into a second
join key, and finish with a semi join. None of that is sound for a single join,
because all three change how many rows match -- which is the one thing a single
join has to observe. So a correlated scalar subquery in a `WHERE` clause was
about 1.6x slower than the aggregated form it replaces.
`correlated_scalar_subquery_yields_single_row` only looks for an aggregate the
rule can see from the top of the subquery. This also asks the decorrelated
subquery's functional dependencies whether it is already unique on the columns
the join equates with the outer plan, which covers uniqueness it inherits
rather than declares:
- a declared key, when the correlated predicate carries a further condition
and so does not reduce to the `Filter::is_scalar` shape `max_rows` looks
for;
- a `GROUP BY` on the correlated column with nothing to aggregate;
- uniqueness carried up through the subquery's own joins and projections.
Only equality conjuncts count as join keys: uniqueness on `(a, b)` says nothing
about how many rows `sub.a = outer.a AND sub.b > outer.b` matches.
On TPC-H SF10 with `c_custkey` declared a primary key, `where (select
c_mktsegment from customer where c_custkey = o_custkey) = 'BUILDING'` goes from
1.5x slower than the `min()` form to 1.6x faster (87 ms -> 55 ms), and the
non-pushable variant from 1.55x slower to 1.15x faster.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SgqwvctZdvR1ZCz2hbkEJC
github-actions
Bot
added
sql
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #24782 +/- ##
========================================
Coverage 81.52% 81.53%
========================================
Files 1123 1123
Lines 405970 406200 +230
Branches 405970 406200 +230
========================================
+ Hits 330978 331177 +199
- Misses 55627 55657 +30
- Partials 19365 19366 +1 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
Thank you for opening this pull request! Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details |
Contributor
Author
|
Superseded by #24784, which has the same change with the history cleaned up (an unrelated file had been committed by mistake). |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Rationale for this change
A correlated scalar subquery must return at most one row per set of outer
values. DataFusion enforced this in the analyzer by requiring an aggregate on
top of the subquery. Without one, planning failed with
So a plain attribute lookup did not plan at all:
To run it you had to wrap the column in
min,maxorany_value. Thataggregate does no useful work; it only proves a row count that the data already
guarantees, and it costs a full hash aggregation over the subquery side.
The single join from Neumann and Kemper's unnesting paper removes the need for
it. It is an outer join that emits one row per row of its preserved side and
raises an error when a second row matches, so the subquery decorrelates into a
join with no aggregate at all.
What changes are included in this PR?
JoinType::LeftSingleandJoinType::RightSingle. They behave likeLeftand
Rightexcept that a second match for a row of the preserved side failswith
Scalar subquery returned more than one row, the errorScalarSubqueryExecalready raises for uncorrelated scalar subqueries.ScalarSubqueryToJoinuses a single join only for subqueries not alreadyknown to return at most one row. Subqueries that are known to, from an
aggregate grouped on correlated columns or from functional dependencies
showing the subquery unique on the join keys, keep their plain
LEFT JOIN.The analyzer no longer rejects the rest.
HashJoinExecimplements both directions, soJoinSelectioncan still swapthe inputs to choose a build side.
NestedLoopJoinExecimplements both forcorrelations with no equijoin key. Both reuse the matched bitmaps as the
duplicate detector, so the check costs one bit test per matched row and
covers matches spread over batches and partitions.
SortMergeJoinExecandPiecewiseMergeJoinreject single joins, and thephysical planner routes them to an operator that implements them.
unsupported, since Substrait has no equivalent.
No query in TPC-H (22) or TPC-DS (99) produces a single join, so no plan in
either suite changes. Every correlated scalar subquery in them aggregates a
real value rather than proving a row count.
Benchmarks
TPC-H SF10, comparing the form you had to write before against the form that
now plans. Median of 7 runs, interleaved and repeated.
In a projection the aggregate is pure overhead, and removing it saves the share
of the work the subquery side represents.
In a filter it is not that simple. A plain
LEFT JOINlets the optimizer use apredicate on the subquery output:
eliminate_outer_joinmakes the join innerbecause the predicate rejects nulls,
extract_equijoin_predicatefolds acomparison against an outer column into a second join key, and
eliminate_jointurns the result into a semi join. None of these are sound for a single join,
because all three change how many rows match, which is what the single join
must observe.
So the rule tries harder to avoid needing a single join. Besides the aggregate
it can see at the top of the subquery, it asks the decorrelated subquery's
functional dependencies whether it is already unique on the columns the join
equates with the outer plan. That covers uniqueness the subquery inherits
rather than declares: a key that survives an extra condition in the correlated
predicate, a
GROUP BYon the correlated column with nothing to aggregate, anduniqueness carried up through the subquery's own joins.
With
c_custkeydeclared a primary key, the filter cases end up faster thanthe form they replace, because they get both the aggregate removal and the
rewrites the left join enables:
On a table with no declared key and no way to infer uniqueness, the filter case
keeps its 1.5x cost. That is the price of the stricter semantics, and it is
opt-in: the aggregated query still plans exactly as before.
Are these changes tested?
Yes.
subquery.sltgains a single join section: attribute lookup with and withouta match, the more-than-one-row error, the same in a filter rather than a
projection, the nested loop path for a correlation with no equijoin key, and
plan assertions for each way a subquery can show it needs no single join (an
aggregate, a declared key surviving an extra predicate, a bare
GROUP BY),plus one showing a key proves nothing under an inequality. Four existing
cases changed from rejected by the analyzer to running.
HashJoinExecandNestedLoopJoinExecunit tests cover both directionsacross partition modes and batch sizes, the accepted and the rejected case,
and the all-NULL build key path.
Are there any user-facing changes?
data has a second matching row they fail at run time with
Scalar subquery returned more than one rowinstead of failing to plan.GROUP BYin a correlated scalar subquery may now name columns thecorrelated predicate does not fix.
LIMITabove the correlated predicate still cannot be decorrelated. Thatcase now reports
Correlated scalar subquery with a LIMIT must be limited to a single rowinstead of reporting a missing aggregate.JoinTypegains two variants, so exhaustive matches on it in downstream codeneed a new arm. This is an API change.