stuhood opened a new issue, #24599: URL: https://github.com/apache/datafusion/issues/24599
### Is your feature request related to a problem or challenge? When joining a dataset (usually an intermediate relational plan: particularly the output of a previous join) with a table that is already partitioned (particularly a storage source exposing native `Partitioning::Range`), DataFusion's physical optimizer forces a two-sided repartitioning / network shuffle (`RepartitionExec: Hash` on both inputs). When multi-child operators (such as `HashJoinExec` or `SortMergeJoinExec`) declare co-partitioning requirements via [`InputDistributionRequirements`](https://github.com/apache/datafusion/blob/d5552342012888b7d1a3ab88d92e3d292fc0cde0/datafusion/physical-plan/src/distribution_requirements.rs#L58-L64): * During per-child requirement satisfaction in [`ensure_distribution`](https://github.com/apache/datafusion/blob/d5552342012888b7d1a3ab88d92e3d292fc0cde0/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs#L1285-L1552), any input that is unpartitioned or does not yet meet `Distribution::KeyPartitioned` is wrapped in `RepartitionExec: Hash(keys, target_partitions)`. An input that is already natively partitioned (e.g. `Partitioning::Range`) satisfies its own child requirement without modification. * When [`enforce_distribution_relationships`](https://github.com/apache/datafusion/blob/d5552342012888b7d1a3ab88d92e3d292fc0cde0/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs#L1078-L1271) checks the relationship between the inputs via [`InputDistributionRequirements::unsatisfied_co_partitioned_children`](https://github.com/apache/datafusion/blob/d5552342012888b7d1a3ab88d92e3d292fc0cde0/datafusion/physical-plan/src/distribution_requirements.rs#L190-L205), it detects a layout mismatch between the synthesized `Hash` and the native `Range`. * Because distribution enforcement only synthesizes creates `Hash` partitioning currently, it discards the native partitioning on the pre-partitioned relation and wraps both inputs in `RepartitionExec: Hash`. This forces a shuffle of _both_ inputs (rather than only one of them), and loses pre-existing partition layouts -- often on the largest table in the query, due to join ordering. ### Describe the solution you'd like DataFusion should avoid double-shuffles when joining against already partitioned datasets. When enforcing co-partitioning relationships in multi-child operators: - If one of the inputs already satisfies its partitioning requirement (e.g., native `Partitioning::Range` or `Partitioning::Hash`), the optimizer should recognize this existing partitioning as a candidate reference. - Rather than forcing all inputs into a new hash layout, the optimizer should attempt to adapt the unsatisfied peer inputs into a compatible partition layout matching the reference child (e.g., repartitioning the unsatisfied input into the reference child's exact range split points or hash partition count). - When multiple inputs already satisfy their requirements, the optimizer should prefer native source partitionings over synthetic exchange nodes (which are likely to have the most accurate statistics and balanced partitions), and use table size / row count statistics as a tie-breaker. - If the existing partitioning cannot be adapted across all peer inputs (e.g. incompatible data types or expression counts), the optimizer should fall back to standard two-sided hash repartitioning. ### Describe alternatives you've considered - Status quo: - Discarding pre-existing partitioning on all inputs whenever any peer input requires repartitioning, forcing full two-sided hash repartitioning on every co-partitioned join. - Manual repartition / custom rules: - Requiring custom external physical optimizer rules or manual plan transformations in downstream query engines to align partitionings before invoking DataFusion's distribution enforcement. ### Additional context - Epic: [#22395](https://github.com/apache/datafusion/issues/22395) (Implement Range Partitioning) -- 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]
