[EPIC] Optimize Range Partitioning for Joins and Distributed Pipelines

Background and Motivation

Range partitioning landed in DataFusion 55 (#22395), establishing the core representation, physical planning, and execution primitives for range-partitioned data. With the foundation in place, the next step is continuing to optimize planner decisions and core execution nodes—particularly around joins, distribution enforcement, partition scaling, and runtime execution primitives like dynamic filters.

As Range partitioning is integrated into distributed execution engines and multi-partition queries (such as datafusion-contrib/datafusion-distributed#730), several opportunities have emerged to preserve partition alignment, eliminate unnecessary shuffles and broadcasts, support flexible partition scaling across stages, and streamline runtime dynamic filtering across partitioned operators.

Scenarios to Optimize

• Symmetric pre-partitioned joins
  • When two relations are already range-partitioned on join keys with matching or compatible split points (e.g. finer or coarser partition granularities), executing task-local joins with zero network shuffle and zero broadcast overhead, and targeting dynamic filter bounds directly to matching partitions.
• Asymmetric stream adaptation
  • When joining a range-partitioned relation against an unpartitioned stream, adapting the unpartitioned stream to match the reference layout while ensuring large tables are not shuffled to match smaller dimension tables.
• Concurrency mismatch and partition scaling
  • When target worker concurrency differs from the number of declared physical split points (under-parallelism or over-parallelism), scaling partition counts via sample down-sampling and identifying range layout relationships to group adjacent partitions or route sparse channels without full all-to-all repartitions.
• Multi-way and hybrid pipelines
  • Preserving range metadata through multi-way join chains and hybrid pipelines where broadcast dimension joins coexist with range-partitioned streams.

Focus Areas and Priorities

High Priority: Core Optimizer Decisions, Execution Primitives, and Benchmarks

Because all consumers rely on the optimizer and core execution nodes (HashJoinExec, DynamicFilterPhysicalExpr), improving planner decisions in datafusion/physical-optimizer and execution primitives in datafusion/physical-plan—and pairing them with benchmarks—is the highest initial priority. This work also unblocks publishing the dedicated Range partitioning blog post.

• Join planning and distribution enforcement

  • #25301 (PR #25304)
  • #25483
  • #25302
  • #25435
  • #25436
  • #24947

• Partition elasticity and scaling

  • #24712 (PR #24766)

• Partition-aware dynamic filtering

  • #23817

• Benchmarks and documentation

  • #25439
  • #24095
  • #24667

Secondary Priority: Providers, Ecosystem, and Code Organization

While optimizer rules and execution nodes benefit all consumers, table providers, ecosystem serialization, and code modularity can be progressed in parallel:

• Scan and table provider integration

  • #23436 (PR #25279)
  • #20195

• Ecosystem serialization and code organization

  • #24245
  • #22788

Related Work and References

• Implementation epic: #22395
• Dynamic filters on range joins: #23376, PR #23854
• Integration testing and benchmark patterns: datafusion-contrib/datafusion-distributed#734
• Pre-partitioned data and aggregation (collaborating with @gene-bordegaray):
  • Monotonic function satisfaction (date_bin / date_trunc: #23569, #24644, #25344, #25345)
  • Streaming aggregation with partition-disjoint keys (#24438 / PR #24440)
  • Linear merge and early termination over sorted Range partitions (#10316)
  • Plan-time static partition pruning for Range partitioning (#25437)
  • Runtime partition extrema for range-aware operators (#23089)