Adapt existing partitioning for unsatisfied inputs in co-partitioned joins

[stuhood]

Which issue does this PR close?

• Closes Avoid unnecessary repartitioning when joining against pre-partitioned relations #24599.

Rationale for this change

When multi-child operators (such as HashJoinExec or SortMergeJoinExec) declare co-partitioning requirements via InputDistributionRequirements, all children must share a compatible partitioning scheme on their respective join keys.

Prior to this change, joining an unpartitioned stream with an input that was already partitioned (such as Partitioning::Range or Partitioning::Hash from a pre-sorted/partitioned data source scan):

1. During per-child requirement enforcement in ensure_distribution, the unpartitioned input failed its Distribution::KeyPartitioned requirement and was wrapped in RepartitionExec: Hash(keys, target_partitions).
2. When enforce_distribution_relationships inspected both children, it saw a layout mismatch between Hash and Range. The code would only create Hash partitionings, and so it discarded the existing native Range partitioning on the second child and wrapped both children in RepartitionExec: Hash, forcing a double shuffle.

What changes are included in this PR?

This PR adjusts enforce_distribution_relationships to recognize when one of the children in a co-partitioned group already satisfies its distribution requirement, selecting it as a reference partitioning and adapting unsatisfied peer inputs to match:

• Candidate Identification & Tie-Breaking:
  • Inspect co-partitioned children to find those that satisfy input_distributions.child_satisfaction().
  • Prefer native partitioned inputs (e.g. data source scans) over newly injected RepartitionExec exchange nodes (since their statistics and partition balance are more likely to be accurate).
  • For tie-breaking among multiple satisfied candidates, rank candidates using PlanSize (total_byte_size, then num_rows computed via StatisticsContext) and select the strictly largest input.
• Partitioning Adaptation:
  • RangePartitioning: If the satisfied reference child has Partitioning::Range(ref_range), adapt ref_range's split_points and sort options to the unsatisfied child's join expressions after verifying that expression data types match split point scalar values via split_points_match_expr_types.
  • HashPartitioning: If the reference child has Partitioning::Hash(_, count), adapt to Partitioning::Hash(exprs, count).
  • Validate that the reference candidate can be adapted across all unsatisfied children in the co-partitioned group; if any child is incompatible, safely fall back to standard two-sided hash repartitioning.
• In-Place Repartition Rewiring:
  • If the unsatisfied child was already wrapped in an eager RepartitionExec during Phase 1, replace its input directly rather than nesting redundant RepartitionExec nodes.

Are these changes tested?

Yes, integration tests in datafusion/core/tests/physical_optimizer/enforce_distribution.rs:

• range_hash_join_repartitions_unsatisfied_side_to_match_range: verifies unsatisfied range input adapts to peer reference range partitioning.
• range_hash_join_repartitions_unpartitioned_side_to_match_range: verifies unpartitioned scan input adapts directly to peer range partitioning without modifying the partitioned side.
• range_hash_join_rehashes_incompatible_data_type: verifies type mismatch (Int32 vs Int64) safely falls back to two-sided hash repartitioning.

Are there any user-facing changes?

No API changes.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 83.07692% with 44 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.50%. Comparing base (7e4188e) to head (24ad329).
⚠️ Report is 65 commits behind head on main.

Files with missing lines	Patch %	Lines
...er/src/ensure_requirements/enforce_distribution.rs	76.81%	30 Missing and 2 partials ⚠️
datafusion/physical-expr/src/partitioning.rs	92.43%	7 Missing and 2 partials ⚠️
...ion/physical-plan/src/distribution_requirements.rs	0.00%	3 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24600      +/-   ##
==========================================
+ Coverage   81.38%   81.50%   +0.11%     
==========================================
  Files        1116     1123       +7     
  Lines      397973   405013    +7040     
  Branches   397973   405013    +7040     
==========================================
+ Hits       323893   330092    +6199     
- Misses      55118    55602     +484     
- Partials    18962    19319     +357     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[stuhood]

cc @gene-bordegaray

[kosiew]

@stuhood,

Thanks for working on this. The approach looks good overall, especially preserving an existing compatible partitioning instead of falling back to repartitioning both sides.

I have one non-blocking test suggestion below.

[kosiew]

🚀
@stuhood
Thank you for your contribution.

[gene-bordegaray]

Have been super busy still will give this guy a review after merged. Thank you @stuhood and @kosiew 🙇