Conversation
`JoinSelection` only made local decisions -- the build side and partition mode of one join at a time -- so the shape of the join tree stayed whatever the logical planner produced. For a query written as a flat list of relations that is a left-deep tree in `FROM`-clause order, which ignores how much each join reduces or inflates its inputs. Add a dynamic programming enumerator that flattens a subtree of reorderable joins into a graph of opaque relations plus the predicates between them, searches every connected order (bushy as well as left-deep) under a `C_out` cost model built from the same cardinality estimates the rest of the rule uses, and rebuilds the subtree only when the winner is strictly cheaper than the planner's order. Semi and anti joins take part as reducers: they are filters on their output side, so their quantified side becomes a relation that may be applied at any node covering the columns its keys reference. Non-equi join filters are re-attached at their lowest common ancestor, so a join carrying one no longer blocks reordering. Controlled by `datafusion.optimizer.join_enumeration` (default on) and `datafusion.optimizer.join_enumeration_limit`. TPC-H SF1, best of 5 interleaved runs: q18 0.67x, q7 0.72x, q2 0.87x, q21 0.87x, q8 0.94x, q9 0.94x, and nothing regressed beyond the noise floor measured on join-free control queries. All 22 queries return byte-identical results with the flag on and off. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Contributor
Author
|
run benchmarks |
Contributor
Author
|
run bencmark tpch10 |
github-actions
Bot
added
documentation
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/join-order-enumeration (a329cf3) to bb038a6 (merge-base) diff Run configurationrun benchmark clickbench_partitionedResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/join-order-enumeration (a329cf3) to bb038a6 (merge-base) diff Run configurationrun benchmark tpcdsResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/join-order-enumeration (a329cf3) to bb038a6 (merge-base) diff Run configurationrun benchmark tpchResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing perf/join-order-enumeration (a329cf3) to bb038a6 (merge-base) diff Run configurationrun benchmark tpchCPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing perf/join-order-enumeration (a329cf3) to bb038a6 (merge-base) diff Run configurationrun benchmark tpcdsCPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing perf/join-order-enumeration (a329cf3) to bb038a6 (merge-base) diff Run configurationrun benchmark clickbench_partitionedCPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
TPC-DS q6 got 37% slower. Its `date_dim` is filtered by a subquery and DataFusion estimates that subplan at 14,610 rows against 31 real ones, so no join in the query appears to reduce anything, every order ties, and the winner is picked essentially arbitrarily -- a coin flip that costs 37%. Only replace the planner's order when the enumerated one is cheaper by a clear margin, configured by `datafusion.optimizer.join_enumeration_min_improvement` (default 10%). Measured: q6's "gain" was under 1%, while every TPC-H win survives a margin of 10% or more. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
|
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 |
Interval analysis rejects a predicate outright if any part of it is out of reach, and `OR` is out of reach -- which an `IN` list becomes once the planner expands it. So a predicate that is mostly analyzable fell back to the flat default selectivity for all of it. TPC-DS q68 filters `date_dim` by `d_dom between 1 and 2 AND d_year IN (1999, 2000, 2001)`, which was estimated at 20% of the table, 14,610 rows, against 72 that survive. Split the predicate into top-level conjuncts, analyze the ones interval arithmetic supports, and estimate an `IN` list -- or the chain of `OR`ed equalities a short list expands into -- as the fraction of the column's values it selects, the same reasoning `col = literal` gets from `1 / NDV`. An unrecognized conjunct still contributes the default, once, as the whole predicate used to. The predicate is passed through untouched when nothing was split off it, because rebuilding the conjunction re-associates it and interval propagation is sensitive to the shape of the tree it walks. That `date_dim` filter is now estimated at 71 rows. Better estimates also mean better plans: TPC-DS q17 goes from 1.05x slower to 0.74x with join enumeration on, q6 to 0.99x and q68 to 1.02x, while every TPC-H win holds and q7 improves to 0.65x. The baseline gains too, independently of enumeration: q22 drops from 147ms to 116ms. One expectation moves: a cross join whose filtered side is now correctly estimated smaller swaps its inputs, which reorders the rows of a query that does not ask for an order. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Contributor
Author
|
run benchmarks |
Contributor
Author
|
run benchmark tpcds |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/join-order-enumeration (de80dd7) to bb038a6 (merge-base) diff Run configurationrun benchmark clickbench_partitionedResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/join-order-enumeration (de80dd7) to bb038a6 (merge-base) diff Run configurationrun benchmark tpcdsResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/join-order-enumeration (de80dd7) to bb038a6 (merge-base) diff Run configurationrun benchmark tpcdsResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/join-order-enumeration (de80dd7) to bb038a6 (merge-base) diff Run configurationrun benchmark tpchResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing perf/join-order-enumeration (de80dd7) to bb038a6 (merge-base) diff Run configurationrun benchmark tpchCPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing perf/join-order-enumeration (de80dd7) to bb038a6 (merge-base) diff Run configurationrun benchmark tpcdsCPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing perf/join-order-enumeration (de80dd7) to bb038a6 (merge-base) diff Run configurationrun benchmark tpcdsCPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing perf/join-order-enumeration (de80dd7) to bb038a6 (merge-base) diff Run configurationrun benchmark clickbench_partitionedCPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
Comments were carrying explanation that the code already states. Cut them back to the reasoning that is not evident from reading it: why reordering is sound, why one key per class per cut is enough, why the predicate is not re-associated before interval analysis. Drop `solve_greedy`. It was a second search strategy for graphs above the exhaustive limit that no benchmark reaches -- TPC-H and TPC-DS graphs are eight relations at most -- so it was untested surface. Graphs that large now keep the planner's order, which is what happened before enumeration existed. That also merges the two size bounds into one. Fold three end-to-end row-equality tests into one over four queries, and fix two lint failures: `HashSet<ScalarValue>` trips `mutable_key_type`, and two bitmask tests are `is_power_of_two`. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Member
|
The benchmark results are very promising and I am very excited to see work in this area! I haven't had time to look into the details of the code, but since this is still a draft I'd like to mention DpHyp for the join ordering problem. It has been integrated in CALCITE-6846, an implementation ported from Apache Doris with a few additional improvements. According to https://www.alibabacloud.com/blog/602899 it's also the join ordering implementation of both DuckDB and MySQL. DuckDB is especially relevant as it's often cited here in the community as a source of inspiration. DPHyp might be a good alternative over the in-house implementation, especially for dealing with many non inner-join cases, which are always tricky. Alternatively, it might at least provide ideas for some of the open questions we have. I feel that the algorithm itself wouldn't need customization points, as it's mostly cost-agnostic, what we would like to provide is a customization around statistics/cost-model. For the former #23651 will probably suffice (paired with the improved version of #21122 which is next on my list). For the latter, I think it can be done via a physical planning rule "costing" alternatives coming from the hypergraph, to pick the cheapest, as physical rules are already a known customization point. |
Contributor
Yes, this kind of observation is exactly why I think we need to have an extensible framework (to be able to swap different implementations in/out)
Yes, I think in general join optimization algorthms are all:
And there are 50 years of papers about different implementations of the above I think it is absolutely certain we will not be able to include any implementation in the core that will satisfy all use cases.
Great -- thank you. I'll try and look at them later
yes, I agree -- I will try and propose an API for this |
Contributor
|
@asolimando, perhaps once we have an API proposed a good test would be "could we implement DpHyp using it" |
Contributor
|
BTW I plan to finish up the DF 55 release blog first, but then I hope to spend some time wiht this PR making some API proposals |
Contributor
Author
I think the current approach is more close to On my machine with the join ordering, DataFusion beats DuckDB on TPC-H at least (didn't check TPC-DS compared to DuckDB). It would make a nice blog post if that's a robust result... I also agree with @alamb if we have a nice API that allows to register a different algorithm, one could just swap the default (although my goal would be to have a great default that works >90% of the time). |
Member
In case you want further coverage beyond TPC-* benchmarks, I can suggest the join-ordering-benchmark/JOB (original paper, implementation). |
Contributor
Author
Thanks for reminding - I think we also have it in here (or at least a version). https://github.com/apache/datafusion/tree/58150eac882b65c3d137e0dedb34b018e1605d3a/benchmarks/sql_benchmarks/imdb |
…ins with it The search now enumerates connected subgraph/complement pairs (Moerkotte and Neumann, SIGMOD 2008) instead of every split of every subset. It reaches the same pairs, so it picks the same order; a sparse graph just has far fewer connected pairs than it has splits. The every-split search stays in `dphyp.rs`'s tests as the oracle a randomized parity test checks against. Breaking ties on the relation sets, rather than on whichever pair the search reached first, keeps the plan from depending on the enumeration order at all -- without it, 20 of 99 TPC-DS plans differed between the two searches purely by visit order. What takes part in a reorder is wider: - Outer and mark joins join semi and anti joins as relations *applied* to the set their keys come from: a semi or anti join filters that side, an outer join extends it, a mark join marks it. They are emitted as the `Right*` join types with the applied side building. A predicate over a null-supplied or marked column still stops the rewrite, since where such a join lands decides which rows those columns are null or false for. - A join key may now be a cast of a column, and a projection between joins may compute one. Coercing a key's two sides to one type puts the cast in a projection below the join, which used to make that projection an opaque leaf and cut the join tree into fragments -- the whole Join Order Benchmark was invisible to this rule because its ids are `UInt32` and its foreign keys `Int32`. Predicates that share a column state one equality however many pairs write it. Counting a pair each collapsed the estimate for a redundant clique -- JOB 15d writes one `movie_id` equality ten ways between five relations -- so each equality now spends at most `k - 1` pair denominators. `join_enumeration_min_improvement` is gone: the rule takes any cheaper order. `join_enumeration_limit` defaults to 16, which the cheaper search affords. Planning cost was almost all allocation and hashing per visited pair: `exchanges` appends into a buffer the table reuses instead of returning a `Vec`, the per-subset partitioning maps became scanned vectors, `combine` is asked once per pair rather than twice, and both cardinality and the reusable classes are memoized. A plan is also no longer kept apart from the cheapest one once no join above can hash on its partitioning, which is what kept ten plans per subset alive: TPC-DS q64 now holds one. Measured here, against this rule turned off: JOB execution 70.5s -> 30.5s, TPC-DS SF1 8.53s -> 5.88s, TPC-H SF10 8.26s -> 7.23s, TPC-H SF1 0.80s -> 0.78s, with every JOB row count unchanged. Enumeration's share of planning is now below the noise of the rest of it; the worst single query went from +678ms to +8ms. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Pv9iaZfELchz2UeHRwQeVv
Contributor
Author
Thanks for the suggestion - current draft is implements DPHyp, and I benchmarked the implementation against JOB. It seems quite a bit faster in planning and now supports outer / mark joins as well, improving some queries that use them. Ill probably compare plans against DuckDB and or calcite and see if query plans are significantly different. |
…meration `default_filter_selectivity` is a `ConfigFilterSelectivity` now, so the cost model reads it through `get()`, and the distribution match in `keep_partitioning_needed_above` is a `let...else` now that upstream's `KeyPartitioned` gave it a second pattern and `manual_let_else` fires. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Pv9iaZfELchz2UeHRwQeVv
Contributor
Author
|
run benchmark tpch tpcds tpch10 |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/join-order-enumeration (b8acd17) to fc15e25 (merge-base) diff Run configurationrun benchmark tpchResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/join-order-enumeration (b8acd17) to fc15e25 (merge-base) diff Run configurationrun benchmark tpcdsResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/join-order-enumeration (b8acd17) to fc15e25 (merge-base) diff Run configurationrun benchmark tpch10Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing perf/join-order-enumeration (b8acd17) to fc15e25 (merge-base) diff Run configurationrun benchmark tpchCPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing perf/join-order-enumeration (b8acd17) to fc15e25 (merge-base) diff Run configurationrun benchmark tpch10CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing perf/join-order-enumeration (b8acd17) to fc15e25 (merge-base) diff Run configurationrun benchmark tpcdsCPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
Dandandan
added a commit
to Dandandan/arrow-datafusion
that referenced
this pull request
Aug 22, 2026
## Which issue does this PR close? None; split out of apache#24456 so the default change can be reviewed on its own. ## Rationale for this change We can increase the default value as generally a larger value performs better (avoiding a repartioning) ## What changes are included in this PR? - `hash_join_single_partition_threshold` defaults to 4MB instead of 1MB, and the generated `configs.md` row follows. - The two sqllogictest files that print or restore the value are updated. - The `join_selection` tests build their inputs from the threshold but asserted the byte sizes the 1MB default happened to produce (`8192` / `2097152`). They now derive those expectations from the configured threshold, so the assertions stay correct wherever the default sits. ``` TPC-H 10 │ Total Time (HEAD) │ 6500.41ms │ │ Total Time (perf_hash-join-single-partition-threshold-4mb) │ 6262.84ms │ TPC-DS │ Total Time (HEAD) │ 9919.35ms │ │ Total Time (perf_hash-join-single-partition-threshold-4mb) │ 9572.04ms │ ``` ## Are these changes tested? Yes, by existing coverage: the full sqllogictest suite (504 files) passes, as do the `datafusion`, `datafusion-common` and `datafusion-physical-optimizer` unit and integration tests. No new test is added -- the change is a default, and the tests that pin plan shapes around it already exist. ## Are there any user-facing changes? Yes: a changed default, so a query whose build side is between 1MB and 4MB now plans as `CollectLeft` where it previously repartitioned. `configs.md` documents the new value, and anyone relying on the old behaviour can set the option back to `1048576`. Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
Contributor
|
Hello folks, it's great to see the community getting involved in work related to join order. We at LakeSail have implemented a round-trip join reorder optimizer based on the DPHyp algorithm. https://github.com/lakehq/sail/tree/v0.7.1/crates/sail-physical-optimizer/src/join_reorder As @alamb mentioned, it would be fantastic if the upstream could expose the full set of relevant traits and abstractions. We are happy to participate in discussions and get involved from a downstream perspective. |
Contributor
Thank you 🙏 I am out at VLDB next week but then I hope to help try and get the Join traits / machinery sorted out. DataFusion has needed it for a long time |
Contributor
|
I just wanted to say the more I talk to people at VLDB the more excited I am about this PR -- I think it will finally unlock more sophisticated join ordering I am doing research on JoinGraph apis, etc -- and I expect to work on this next week |
Contributor
Fanscinating! 🚀 I'm curious what the final implementation/algorithm looks like in your mind. Any reading recommendations? |
14 tasks
Contributor
What I am most excited about is the APIs for:
I think with those primitives, we could implement pretty much any join order algorithm we want (and could allow others to plug their own in) My informal conversations suggested that there is no existing writeup on API best practices, but all optimizers have some version of these APIs. |
Contributor
I asked around and no one seemed to know of any good writeups on the API design (I asked Thomas Neumann, Goetz Graefe, and Peter Boncz, among others, and they did not know of any) |
Contributor
Author
|
Amazing! Btw I am on vacation so I am responding more slowly / less these days. |
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
7 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
Depends on:
hash_join_single_partition_thresholdto 4MB #24558Currently DataFusion only reorders joins locally, limiting the performance of deeply nested joins.
We can add a pass that optimizes thw order to make them generally run faster, especially at scale.
See benchmarks below:
Benchmark results:
Details
Benchmarks SMJ
Details
What changes are included in this PR?
A dynamic algorithm for join enumeration/optimization (up to some maximum depth).
Are these changes tested?
Yes, existing, new tests.
Are there any user-facing changes?
JoinEnumerationpass (DPHyp).