Repository navigation
enforce_distribution_relationships should consider size of unsatisfied children before selecting a reference child in asymmetric joins #25302
Description
Is your feature request related to a problem or challenge?
In PR #24600 ("Adapt existing partitioning for unsatisfied inputs in co-partitioned joins"), enforce_distribution_relationships was introduced to adapt unsatisfied children in co-partitioned joins to an existing satisfied reference partitioning.
When selecting best_satisfied_child:
datafusion/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs
Lines 1181 to 1229 in f40d978
| let best_satisfied_child: Option<(usize, Partitioning)> = match satisfied_children | |
| .len() | |
| { | |
| 0 => None, | |
| 1 => satisfied_children | |
| .into_iter() | |
| .next() | |
| .map(|(i, p, _)| (i, p)), | |
| _ => { | |
| // Prefer native partitioned children over newly repartitioned ones | |
| let native_children: Vec<_> = satisfied_children | |
| .iter() | |
| .filter(|(_, _, is_native)| *is_native) | |
| .collect(); | |
| if native_children.len() == 1 { | |
| let (i, p, _) = native_children[0]; | |
| Some((*i, p.clone())) | |
| } else { | |
| let pool = if !native_children.is_empty() { | |
| native_children | |
| } else { | |
| satisfied_children.iter().collect() | |
| }; | |
| let candidates: Vec<_> = pool | |
| .into_iter() | |
| .map(|(idx, part, _)| { | |
| let size = | |
| PlanSize::from_plan(children[*idx].context.plan.as_ref()); | |
| (size, *idx, part.clone()) | |
| }) | |
| .collect(); | |
| // Only select a reference candidate if there is a unique, strictly | |
| // larger winner (`size_a > size_b`). If candidates have equal or | |
| // incomparable sizes (e.g. non-overlapping metrics), return None | |
| // so the optimizer avoids arbitrary tie-breaking and falls back to | |
| // standard distribution. | |
| candidates | |
| .iter() | |
| .find(|(size_a, idx_a, _)| { | |
| size_a.is_known() | |
| && candidates.iter().all(|(size_b, idx_b, _)| { | |
| idx_a == idx_b || size_a > size_b | |
| }) | |
| }) | |
| .map(|(_, idx, part)| (*idx, part.clone())) | |
| } | |
| } | |
| }; |
Notice the current selection logic:
- When
satisfied_children.len() > 1: DataFusion inspectsPlanSizeacross the candidate satisfied children and only selects a reference candidate if there is a unique, strictly larger winner (size_a > size_b), avoiding arbitrary tie-breaking:
datafusion/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs
Lines 1213 to 1226 in f40d978
// Only select a reference candidate if there is a unique, strictly // larger winner (`size_a > size_b`). If candidates have equal or // incomparable sizes (e.g. non-overlapping metrics), return None // so the optimizer avoids arbitrary tie-breaking and falls back to // standard distribution. candidates .iter() .find(|(size_a, idx_a, _)| { size_a.is_known() && candidates.iter().all(|(size_b, idx_b, _)| { idx_a == idx_b || size_a > size_b }) }) .map(|(_, idx, part)| (*idx, part.clone())) - But when
satisfied_children.len() == 1: DataFusion unconditionally selects that single satisfied child as the reference partitioning without checking its size relative to the unsatisfied children that will be adapted to it:
datafusion/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs
Lines 1185 to 1188 in f40d978
1 => satisfied_children .into_iter() .next() .map(|(i, p, _)| (i, p)),
The Problem in Asymmetric Joins:
In an asymmetric join where only one table has a declared partition key matching the join condition (for example, a small dimension table with 100 rows is range-partitioned, but a large fact table with 100M rows is unpartitioned on that join key):
satisfied_childrencontains only the small dimension table (len == 1).- DataFusion chooses the 100-row dimension table as the reference child.
- The 100M-row fact table is forced to repartition (via
ref_partitioning.adapt(...)) to match the small table's partition boundaries. - In distributed environments, this causes the 100M-row table to be shuffled across the network to match the 100-row table, rather than keeping the large table local and shuffling the small table (or falling back to symmetric hash partitioning / broadcast).
Describe the solution you'd like
When satisfied_children.len() == 1 (or before selecting any reference child), compare the PlanSize of the candidate satisfied reference child against the PlanSize of the unsatisfied children that would be adapted to it.
If an unsatisfied child is strictly larger than the satisfied reference child (unsatisfied_size > reference_size):
- Do not select the smaller child as the reference.
- Return
Noneforbest_satisfied_childso that the join falls back to standard symmetric distribution (e.g. symmetric hash repartitioning or broadcast), preventing an asymmetric 1-sided shuffle of the larger relation.
Describe alternatives you've considered
Downstream query planners currently have to defend against this by artificially withholding / refusing to stamp partition metadata on smaller tables in asymmetric joins to prevent DataFusion from shuffling the larger partner to match the smaller table.
Having DataFusion check PlanSize before adapting larger unsatisfied children to a smaller satisfied child would make enforce_distribution_relationships robust against asymmetric joins out of the box.
Additional context
- Introduced in PR Adapt existing partitioning for unsatisfied inputs in co-partitioned joins #24600:
d109f1b - Relevant code in
enforce_distribution.rs:
datafusion/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs
Lines 1181 to 1229 in f40d978
let best_satisfied_child: Option<(usize, Partitioning)> = match satisfied_children .len() { 0 => None, 1 => satisfied_children .into_iter() .next() .map(|(i, p, _)| (i, p)), _ => { // Prefer native partitioned children over newly repartitioned ones let native_children: Vec<_> = satisfied_children .iter() .filter(|(_, _, is_native)| *is_native) .collect(); if native_children.len() == 1 { let (i, p, _) = native_children[0]; Some((*i, p.clone())) } else { let pool = if !native_children.is_empty() { native_children } else { satisfied_children.iter().collect() }; let candidates: Vec<_> = pool .into_iter() .map(|(idx, part, _)| { let size = PlanSize::from_plan(children[*idx].context.plan.as_ref()); (size, *idx, part.clone()) }) .collect(); // Only select a reference candidate if there is a unique, strictly // larger winner (`size_a > size_b`). If candidates have equal or // incomparable sizes (e.g. non-overlapping metrics), return None // so the optimizer avoids arbitrary tie-breaking and falls back to // standard distribution. candidates .iter() .find(|(size_a, idx_a, _)| { size_a.is_known() && candidates.iter().all(|(size_b, idx_b, _)| { idx_a == idx_b || size_a > size_b }) }) .map(|(_, idx, part)| (*idx, part.clone())) } } }; - Related PR: PR feat: allow scaling RangePartitioning #24766 (Range partitioning scaling and preservation).
Activity
[-]enforce_distribution_relationships should consider size of unsatisfied children before selecting a reference child in asymmetric joins[/-][+]`enforce_distribution_relationships` should consider size of unsatisfied children before selecting a reference child in asymmetric joins[/+]perf: Optimize range partitioning across multi-key and asymmetric joi…
ShayanGho commentedon Sep 23, 2026
Hello @stuhood , I’d like to take this on next. Are you already working on it or planning to? If so, I’m happy to pick another issue from the epic.
ShayanGho commentedon Sep 23, 2026
take
stuhood commentedon Sep 23, 2026
Great, thank you!