andygrove opened a new issue, #2757: URL: https://github.com/apache/datafusion-ballista/issues/2757
> **This is a research task.** Spark AQE is the reference point for the problem, not for the solution. Ballista has different constraints than Spark (DataFusion operators, stage-at-a-time replanning, no external shuffle service, partition-count-sensitive operators), so we may not want to copy how Spark does this. The first deliverable is a write-up of whether and how to do it here, ideally with a measurement showing the gain, before any implementation. ## Problem Partition coalescing is off by default (`ballista.planner.coalesce.enabled=false`) and, when on, has a single policy: bin-pack toward `target_partition_bytes` (64 MiB). The module docs describe off as roughly Spark's `parallelismFirst=true`, but that is not quite right. Spark with `parallelismFirst=true` (its default) still coalesces very small partitions: it sizes the target from the available parallelism, bounded below by `minPartitionSize` (1 MiB) and above by the advisory size. It protects parallelism without giving up the tiny-partition win. Ballista's off setting gives up the win entirely, and its on setting gives up the parallelism protection. Spark also exposes `initialPartitionNum`, which lets a job start with many more shuffle partitions than the final count and coalesce down. That is how Spark users get safe behaviour on both small and very large inputs from one setting. ## Questions to answer - What coalescing policy fits Ballista? Options include: a parallelism-aware target (`max(total / available_slots, min_size)` capped by the advisory size), a minimum partition size only, or the current advisory-only mode. - Should the rule be on by default once the policy is parallelism-aware? What do TPC-H/TPC-DS at several scale factors say about the regression risk? (Query-level wins and losses, not just total time.) - Is over-partitioning then coalescing viable here? Hash shuffle write cost grows with partition count and each open writer holds buffers, so the cost of a large initial `M` may outweigh the benefit. The sort-based and hybrid shuffle work (#2578) changes that trade-off and should be considered. - Should sizing consider row counts or the executor's slot count rather than bytes only? - Is a different mechanism better than coalescing, such as choosing the partition count from the stage's measured input before the stage is planned? ## Related - #1987, #2166, #2167, #2220, #2728, #2578 -- 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]
