`enforce_distribution_relationships` should consider size of unsatisfied children before selecting a reference child in asymmetric joins

[stuhood]

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:

1. When satisfied_children.len() > 1: DataFusion inspects PlanSize across 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()))

2. 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_children contains 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 None for best_satisfied_child so 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).