feat: cost-based join order enumeration - #24456
Draft
Dandandan wants to merge 18 commits into
Draft
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>
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
A sort merge join has no mode that collects one side, but the model offered one and priced it as a free broadcast, so under `prefer_hash_join = false` the search chose orders it could not carry out. TPC-H q5 lost 31% at SF10 that way. It now broadcasts only for hash joins. A keyless join compares every pair, which the output cardinality the rest of the cost is built from does not say, so a disjunctive filter estimated to keep few rows looked cheap. TPC-DS q85 turned an equijoin into a nested loop join and ran 122x slower. Keyless joins are now charged for the pairs. With sort merge joins: TPC-H SF10 0.85x, SF1 0.91x, TPC-DS SF1 0.46x. The hash path is unchanged: TPC-H SF10 0.88x, TPC-DS SF1 0.64x. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012YiAABcW4WSqij31zz2P6c
Contributor
Author
|
run benchmarks run tpch tpcds tpch10 |
|
Benchmark for this request failed before finishing (Kubernetes reason: Benchmarks requested: Kubernetes messageFile an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/join-order-enumeration (cf466ca) to bb038a6 (merge-base) diff Run configurationrun benchmark tpch10
env:
PREFER_HASH_JOIN: "false"Results 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 (cf466ca) to bb038a6 (merge-base) diff Run configurationrun benchmark tpcds
env:
PREFER_HASH_JOIN: "false"Results 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 (cf466ca) to bb038a6 (merge-base) diff Run configurationrun benchmark tpch
env:
PREFER_HASH_JOIN: "false"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing perf/join-order-enumeration (cf466ca) to bb038a6 (merge-base) diff Run configurationrun benchmark tpch
env:
PREFER_HASH_JOIN: "false"CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
Resolves one conflict in physical-optimizer/src/join_selection.rs, where upstream's null-aware RightAnti support (apache#23957) added `can_swap_hash_join` at the same spot this branch added `keep_partitioning_needed_above` and its helpers. Both are kept; the four `can_swap_hash_join` call sites merged cleanly. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing perf/join-order-enumeration (cf466ca) to bb038a6 (merge-base) diff Run configurationrun benchmark tpch10
env:
PREFER_HASH_JOIN: "false"CPU 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 (cf466ca) to bb038a6 (merge-base) diff Run configurationrun benchmark tpcds
env:
PREFER_HASH_JOIN: "false"CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
Most were longer than what they explain. Cut to one or two plain lines each, dropped the benchmark anecdotes, and removed a stale doc line left on `crossing_classes` from an earlier edit. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
A hash join carries its own projection, so the rebuilt tree derives one per node from what that node's parents need. A sort merge join has none, so the rule emitted every column at every node and projected once on top, leaving each join to sort columns nothing above it reads. `ProjectionPushdown` cannot repair it afterwards: it only pushes through a join whose columns stay left-then-right, and reordering interleaves them. Each rebuilt sort merge join now drops what nothing above needs. Reordering alone is still left to the parent, which addresses columns by position, so no projection is added where none narrows. Measured against emitting every column, same run, alternating order: TPC-DS SF1 0.98x over the 39 queries whose plans change, TPC-H SF10 0.97x. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012YiAABcW4WSqij31zz2P6c
This was referenced Aug 20, 2026
alamb
reviewed
Aug 20, 2026
alamb
left a comment
alamb
left a comment
Contributor
There was a problem hiding this comment.
First of all, thank you for doing this @Dandandan -- I think this is actually quite close to something we could merge in today
My core concern of any join ordering algorithm is that they are so complicated and often system specific (what is the cost of broadcast?) that if we were to put something clever in the core, we would end up in a constant battle where it didn't quite work for someone, so they tweaked it, and then it broke someone else's usecase (this happened to us at Vertica a lot)
However, I think we can address the concern in this PR by making the code in this PR extensible / reusable (suggestions below) -- Join Ordering is quite a beast (both theoretically and practically) and the chance that any particular join order implementation will work for many people I think is quite low.
I actually think the core dynamic programming implementation in this module is pretty small (most of the code is the mechanics of creating JoinGraphs and then reconstructing them after they are done, along with the cost model), which I think is a good sign.
Thus I suggest structuring this as a few pieces
JoinGraphrepresentation (already present)- Code to convert to and from JoinGraph (aka
ExtractorandRebuilder, already present) - 2 traits (see below -- not yet present)
- Default implementation of those traits included in the core (would need a refactor of code that is present)
- APIs to override the default traits via SessionContext
I think that would address my core concern and make it much easier for people to build with DataFusion. The core crates would have the mechanics of reordering plan nodes, but not the system specific heuristics /models to do so
The traits would be 1) something that enumerates possible plans
trait Enumerator {
/// Search `graph` for a join tree, scored by `model`. `None` if this
/// enumerator can't produce one (too large, doesn't apply, etc.).
fn solve(&self, graph: &JoinGraph, model: &dyn JoinCostModel) -> Option<Solution>;
}And then the cost model https://github.com/apache/datafusion/pull/24456/changes#r3824265511. Something like
trait JoinCostModel {
/// Estimated rows from joining every relation in `mask`.
fn cardinality(&self, mask: RelSet) -> f64;
/// Whether `left` and `right` can be combined, and how (inner join, or one
/// side applying as a reducer).
fn combine(&self, left: RelSet, right: RelSet) -> Option<Combine>;
/// Cost of joining `left` (partitioned as `left_part`) with `right`
/// (`right_part`), for each way of exchanging their inputs.
fn exchanges(
&self,
left: RelSet,
right: RelSet,
left_part: PartSet,
right_part: PartSet,
collect_only: Option<RelSet>,
) -> Vec<(f64, PartSet, RelSet, PartitionMode)>;
}|
|
||
| /// A connected set of joins as relations plus the predicates between them. | ||
| #[derive(Debug)] | ||
| struct JoinGraph { |
Contributor
There was a problem hiding this comment.
we finally have a JoinGraph in DataFusion! (this will probably get crazy, so I recommend it put in its own module to start)
| required: RelSet, | ||
| } | ||
|
|
||
| /// A connected set of joins as relations plus the predicates between them. |
Contributor
There was a problem hiding this comment.
I think it might make sense to point out that this is an alternative representation of at a sub clause in a query -- basically a SELECT .... but with nodes representing tables and edges representing joins between them
It looks like this representation takes some sub tree of joins and then replaces the JoinExec (maybe?)
It wold help to describe with an example how an input ExecutionPlan goes into a JoinGraph
Contributor
There was a problem hiding this comment.
Here is an example that Claude came up with
example four tables, B goes through GROUP BY before the join:
HashJoin(a.x = b.x)
├── HashJoin(a.y = c.y)
│ ├── Scan A
│ └── Scan C
└── HashJoin(b.z = d.z)
├── AggregateExec(group by: b.x, b.z; cnt = count(*))
│ └── Scan B
└── Scan D
And the join representation
JoinGraph {
kind: Hash, reducers: {} (none), filters: []
edge: A.y = C.y
┌───────────┐ ───────────────────────────── ┌───────────┐
│ rel 0: A │ │ rel 1: C │
│ (Scan A) │ │ (Scan C) │
└─────┬─────┘ └───────────┘
│
│ edge: A.x = Agg(B).col0
│
┌─────┴───────────┐
│ rel 2: Agg(B) │
│ [OPAQUE] │
└────────┬─────────┘
│
│ edge: Agg(B).col1 = D.z
│
┌─────┴─────┐
│ rel 3: D │
│ (Scan D) │
└───────────┘
output: columns of A ++ C ++ Agg(B) ++ D (flattened left-to-right, depth-first)
}
| struct JoinGraph { | ||
| relations: Vec<Relation>, | ||
| edges: Vec<Edge>, | ||
| filters: Vec<Filter>, |
Contributor
There was a problem hiding this comment.
I normally think of a join graph where the edges themselves have filters. But maybe that is implicit in the fact that the edges and filters are parallell lists
| /// moves columns to other positions. | ||
| #[derive(Clone, Copy, PartialEq, Eq, Debug)] | ||
| struct ColRef { | ||
| rel: usize, |
Contributor
There was a problem hiding this comment.
would help to document what this is an index into
| /// lack row count statistics are left untouched. | ||
| pub join_enumeration: bool, default = true | ||
|
|
||
| /// How much cheaper an enumerated join order must be, in percent, before it |
Contributor
There was a problem hiding this comment.
this set of settings is exactly why I am hesitant to put something too complicated into the datafusion core -- the join ordering algorithms can get so out of hand complicated I think we need to find a way to keep the core (relatively) simple and leave API hooks to support the more complicated cases
I have more suggestions on how to do this below
Contributor
Author
There was a problem hiding this comment.
Yeah - my goal would be to mostly bring this down as well (i.e. remove or accept the regression).
| } | ||
|
|
||
| /// Cardinality and cost estimates over the subsets of a [`JoinGraph`]. | ||
| struct CostModel<'a> { |
Contributor
There was a problem hiding this comment.
Can we please make this a trait (so that other people can plug in their own cost models, based on whatever statistics they may have, or other knowledge they have about their data and the joins?
Something like this
trait JoinCostModel {
/// Estimated rows from joining every relation in `mask`.
fn cardinality(&self, mask: RelSet) -> f64;
/// Whether `left` and `right` can be combined, and how (inner join, or one
/// side applying as a reducer).
fn combine(&self, left: RelSet, right: RelSet) -> Option<Combine>;
/// Cost of joining `left` (partitioned as `left_part`) with `right`
/// (`right_part`), for each way of exchanging their inputs.
fn exchanges(
&self,
left: RelSet,
right: RelSet,
left_part: PartSet,
right_part: PartSet,
collect_only: Option<RelSet>,
) -> Vec<(f64, PartSet, RelSet, PartitionMode)>;
}
The search now asks a `JoinCostModel` for cardinalities, which pairs may be combined, and what each exchange costs, so anyone with better statistics than the plan carries -- or a different cost function -- can search under them via `JoinEnumeration::with_cost_model`. `DefaultJoinCostModel` is what the rule uses otherwise, unchanged in what it estimates. `reducer_side` and `tree_cost` are trait defaults, since both follow from the three required methods. `exchanges` returns a named `Exchange` rather than a tuple, and the graph the model works over is public: an external model needs to see the relations to key its own statistics off them. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
The graph, and the extraction that flattens a subtree into one, are what a plugged-in cost model has to read, so they now sit in `join_enumeration::graph` rather than among the rule's own internals. The rule, the cost model and the rebuild stay in `mod.rs`; the column-list helpers only the rebuild uses stay with it. Each type keeps a single public path, so `JoinGraph` and the set vocabulary are named through `graph` now. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
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
4 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 (DPsub-like).