Repository navigation
perf: semi-join small filtered dimensions below the joins in between - #2799
Merged
Merged
Conversation
github-actions
Bot
added
the
documentation
DataFusion decorrelates IN/EXISTS/NOT EXISTS subqueries into semi and anti joins placed above the FROM clause's join tree, so the subquery filters the joined result. Add a logical rule, like Spark's PushDownLeftSemiAntiJoin, that moves them below passthrough projections, inner joins and the preserved side of outer joins, onto the input that holds their keys. TPC-H q18 then filters orders down to the qualifying orders before joining lineitem. The rule stops at two places where the move is sound but slower: - A Filter. The semi join should see the filter's output: TPC-H q4's EXISTS probes 5.7M filtered orders instead of all 150M (3.3 s to 9.6 s at SF100 when it did not). - A join whose other input is filtered. That join can remove most rows itself, so a semi join below it probes rows it would have dropped. TPC-H q21's EXISTS and NOT EXISTS moved onto all 379M l1 rows at SF100, doubling its run time (32 s to 65 s). Controlled by ballista.optimizer.semi_join_pushdown (default true).
For an inner join against a small dimension filtered by =, IN or LIKE on literals (a nation, region or part type), also semi-join the dimension with the table the join key comes from, below the joins in between, so the filter removes rows before they are joined and shuffled. This is the effect of Spark's runtime filters. In TPC-H q7 supplier and customer are reduced to the two filtered nations before joining lineitem. Logical plans carry no statistics, so the scheduler runs the rule before physical planning: candidates are found by shape, planned on their own to read their row estimate, and used only under ballista.optimizer.broadcast_join_threshold_rows. A semi join that lands on a tiny input (under 100K estimated rows, such as nation reduced by region in q2/q5/q8) only adds a join, so it is removed again. Unlike SemiJoinPushdown, the reduction also sinks below joins with a filtered input: the semi join is the selective side there. Controlled by ballista.optimizer.semi_join_reduction (default true).
andygrove
force-pushed
the
perf-semi-join-reduction
branch
from
November 7, 2026 17:55
bbb0362 to
dbff8ac
Compare
milenkovicm
approved these changes
Nov 9, 2026
andygrove
marked this pull request as ready for review
November 10, 2026 00:18
andygrove
added a commit
to andygrove/datafusion-ballista
that referenced
this pull request
Nov 10, 2026
apache#2799 landed this branch's first commit (the SemiJoinPushdown rule, its config key and docs) as part of the stacked semi-join reduction work, and refactored the rule's traversal so SemiJoinReduction can share it. That is what conflicted in config.rs, optimizer.rs, semi_join_pushdown.rs (added on both sides) and configs.md. Each conflicted file takes main's version, and the second commit's changes are re-applied on top of it: the volatile key and filter guard in is_pushable_semi_join with its test, the whole-plan assertion in the session test, and the results test with the rule on and off. Against main the branch now differs only in optimizer.rs and semi_join_pushdown.rs.
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
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?
No single issue. Split out of #2792. Related: #2520 (performance review tracking), #1375 (dynamic filters).
Depends on the semi-join pushdown PR (#2793). It is built on that PR's commit, so review only the last commit; this PR will be rebased once the first one merges.
Rationale for this change
DataFusion applies a selective dimension filter (such as
nation) only at the join that names it. TPC-H q7 joins and shuffles 182Mlineitemrows before the nation filter drops 92% of them. Spark's runtime filters avoid this.What changes are included in this PR?
A rule,
SemiJoinReduction, controlled byballista.optimizer.semi_join_reduction(defaulttrue). For an inner join against a dimension filtered by=,INorLIKEon literals (anORneeds one of these in each branch), it adds a semi join against a copy of the dimension on the table the key comes from, below the joins in between. In q7,supplierandcustomerare reduced to the two filtered nations before they joinlineitem.Logical plans have no statistics, so the scheduler runs the rule through
apply_semi_join_reductioninTaskManager::submit_jobbefore physical planning:ballista.optimizer.broadcast_join_threshold_rowsare used; 0 disables the reduction.nationreduced byregionin q2/q5/q8) only adds a join, so it is removed again.stephelper inSemiJoinPushdown, but, unlike that rule, also through inner joins with a filtered input, since the semi join is the selective side there.Tests cover the shape checks and run
apply_semi_join_reductionend to end over in-memory tables: a semi join is added where it lands on a large table, removed where it lands on a small one, and not added when disabled.With every change in #2792 together, on one machine at TPC-H SF100 (2 executors x 8 cores, 16 partitions), q7 goes from 20.1 s to 5.6 s. All 22 queries pass
--verifyagainst DataFusion at SF10 on that branch.Known limits, so reviewers can weigh them:
DynamicFiltersin ballista #1375) would use measured sizes and need no size heuristic, but would only apply to AQE jobs. This rule applies to logical submissions in every execution mode.Are there any user-facing changes?
One new config key,
ballista.optimizer.semi_join_reduction, enabled by default. Plans for queries with filtered dimension joins change.docs/source/user-guide/configs.mdis regenerated.