andygrove opened a new issue, #2680: URL: https://github.com/apache/datafusion-ballista/issues/2680
**Is your feature request related to a problem or challenge? Please describe what you are trying to do.** DataFusion 55 added `Partitioning::Range` with split points, and `RepartitionExec` can produce it. When a submitted physical plan contains one, the static planner removes it: `DefaultDistributedPlanner` keeps only hash repartitions as stage boundaries and drops every other `RepartitionExec` ([planner.rs#L298-L301](https://github.com/apache/datafusion-ballista/blob/98d724b3bd9b603211f444d992b0bcca13c75bca/ballista/scheduler/src/planner.rs#L298-L301)). Dropping a round-robin repartition only costs parallelism, but a range repartition is part of what a plan means. Consider a plan that range-partitions and then sorts each partition, which is a distributed global sort. It comes back as partitions that are each sorted but not range-disjoint, and there is no error. Ballista already has range shuffles (`RangeShuffleWriterExec` and `RangeShuffleReaderExec`), but only the AQE path builds them ([aqe/adapter.rs#L95](https://github.com/apache/datafusion-ballista/blob/98d724b3bd9b603211f444d992b0bcca13c75bca/ballista/scheduler/src/state/aqe/adapter.rs#L95)), from cuts it measures itself. Task slicing has a TODO to map `Partitioning::Range` onto them ([task_builder.rs#L274](https://github.com/apache/datafusion-ballista/blob/98d724b3bd9b603211f444d992b0bcca13c75bca/ballista/scheduler/src/state/task_builder.rs#L274)). Comet Native Executors (apache/datafusion-comet#7288) need this for Spark's range exchanges, as in `ORDER BY`. The driver can supply the bounds, as Spark computes them by sampling, but for now the prototype (apache/datafusion-comet#7289) has to sort in a single task. **Describe the solution you'd like** 1. In the static planner, plan a `RepartitionExec` with `Partitioning::Range` as a range shuffle stage that uses the plan's split points, so that partition `i` of the consuming stage reads range `i`. 2. Restrict range shuffle readers per task by slicing their split points along with their partitions, as the TODO describes. 3. Until the first two land, return an error for a range repartition rather than dropping it. **Describe alternatives you've considered** - Collapsing to one partition and sorting there, which is what the Comet prototype does. It's correct, but the whole sort runs in one task. - AQE's range repartitioning, which computes cuts at runtime. Physical-plan submissions never go through AQE (#2678). **Additional context** Part of apache/datafusion-comet#7288. -- 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]
