JoinSelection should check if inputs already satisfy co-partitioning requirements before choosing CollectLeft (broadcast)

[stuhood]

Is your feature request related to a problem or challenge?

When executing a hash join where both input tables are already physically co-partitioned on the join keys (for example, via matching Partitioning::Range split points or compatible Partitioning::Hash), the optimal execution mode is PartitionMode::Partitioned because it achieves a 0-shuffle, task-local join.

However, in JoinSelection::try_collect_left, JoinSelection only consults size thresholds (supports_collect_by_thresholds against hash_join_single_partition_threshold / hash_join_single_partition_threshold_rows). If the build side is smaller than the threshold (default: 131,072 rows / 1 MB), JoinSelection unconditionally selects PartitionMode::CollectLeft (broadcast):

datafusion/datafusion/physical-optimizer/src/join_selection.rs

Lines 301 to 312 in f40d978

	let transformed = if let Some(hash_join) = plan.downcast_ref::<HashJoinExec>() {
	match hash_join.partition_mode() {
	PartitionMode::Auto => try_collect_left(hash_join, false, context)?
	.map_or_else(
	|| partitioned_hash_join(hash_join, context).map(Some),
	|v| Ok(Some(v)),
	)?,
	PartitionMode::CollectLeft => try_collect_left(hash_join, true, context)?
	.map_or_else(
	|| partitioned_hash_join(hash_join, context).map(Some),
	|v| Ok(Some(v)),
	)?,

In distributed engines (such as DataFusion-Distributed or downstream engines like ParadeDB), CollectLeft inserts a CoalescePartitions / network broadcast across all worker tasks, serializing and transmitting the table across the network. If both sides were already co-partitioned with identical split points / partition boundaries, broadcasting is strictly worse than a local partition-aligned join:

1. A co-partitioned Partitioned join requires 0 network shuffles and 0 broadcast overhead.
2. Degrading to CollectLeft introduces unnecessary network broadcast, data duplication, and memory amplification.

Describe the solution you'd like

Before selecting PartitionMode::CollectLeft in JoinSelection (or within try_collect_left), check whether the join inputs already satisfy the join's distribution requirements for PartitionMode::Partitioned on the join keys.

If the inputs already satisfy the partitioned join requirement (i.e. they are already co-partitioned):

• Preserve PartitionMode::Partitioned rather than degrading to CollectLeft.
• Only fall back to CollectLeft when the inputs are not co-partitioned (where broadcasting the smaller side avoids a 2-sided repartition shuffle).

Describe alternatives you've considered

Currently, downstream systems must either:

1. Globally disable broadcast joins by setting hash_join_single_partition_threshold_rows = 0 (which harms queries joining against small, unpartitioned dimension tables where broadcast is genuinely optimal), or
2. Implement a custom downstream physical optimizer rule that inspects CollectLeft joins, verifies whether both children share compatible Partitioning on the join keys, and mutates the join mode back to PartitionMode::Partitioned before EnsureRequirements runs.

Handling this directly within JoinSelection would benefit all users and downstream engines querying partitioned or co-partitioned tables.

Additional context

• Relevant lines in JoinSelection:
  • try_collect_left
  • statistical_join_selection_subrule
• Related PRs extending range partitioning and distribution satisfaction:
  • PR Adapt existing partitioning for unsatisfied inputs in co-partitioned joins #24600 / PR feat: allow scaling RangePartitioning #24766 (Preserve and adapt range partitioning in EnforceDistribution).

[stuhood]

/take