stuhood opened a new issue, #25421:
URL: https://github.com/apache/datafusion/issues/25421
# [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#734), 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, 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 rather than losing range guarantees
across stages.
* 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)
- [ ] #25302
- [ ] #24947
- [ ] Align multi-key equi-join conditions to match Range ordering
- When equi-join conditions match the columns of an input's
lexicographical `RangePartitioning` but differ in sequence, align the join keys
to match the range ordering to avoid unnecessary shuffles.
- [ ] Preserve partitioning through co-partitioned Full Outer Joins
- In `symmetric_join_output_partitioning`, retain range or hash
partitioning for co-partitioned full outer joins rather than unconditionally
degrading to `UnknownPartitioning`.
* Partition elasticity and scaling
- [ ] #24712 (PR #24766)
* Partition-aware dynamic filtering
- [ ] #23817
* Benchmarks and documentation
- [ ] #24095
- [ ] End-to-end benchmarks for range joins and stage scaling (referencing
patterns in datafusion-contrib/datafusion-distributed#734)
- [ ] #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
* Companion epic on pre-partitioned data and aggregation: @gene-bordegaray
is planning to open a companion epic tracking optimizations over range- and
pre-partitioned data. Potential overlapping topics to align on boundaries:
* 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
- Prune non-overlapping partition slots at plan time when query filters
constrain declared `RangePartitioning` split points, avoiding executing tasks
over empty partitions.
* Runtime partition extrema for range-aware operators (#23089)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
