NoahKusaba commented on code in PR #2438: URL: https://github.com/apache/datafusion-ballista/pull/2438#discussion_r3994496340
########## docs/source/contributors-guide/architecture.md: ########## @@ -204,22 +220,64 @@ compares to the pipelined shuffle used by engines such as DataFusion Distributed [shufflewriterexec]: https://github.com/apache/datafusion-ballista/blob/main/ballista/core/src/execution_plans/shuffle_writer.rs [shufflereaderexec]: https://github.com/apache/datafusion-ballista/blob/main/ballista/core/src/execution_plans/shuffle_reader.rs +### Multi-partition tasks + +Ballista dispatches at the _slice_ level, not the partition level. Each executor advertises a fixed number of +virtual cores (`vcores`), and the scheduler packs up to that many of a stage's output partitions into a single +task. All partitions in the slice execute concurrently under one DataFusion plan invocation: scans and shuffle +readers are rewritten to see only the assigned partition ids, and DataFusion's per-partition `execute(N)` +contract fans the work across the executor's threads. Slice size is bounded by the executor's free vcore count +and by `ballista.scheduler.max_partitions_per_task` (`0` = unbounded — the default; fills each task up to the +executor's free vcore count; `1` = one task per partition, the pre-multi-partition-tasks model). + +Compared to Apache Spark, whose unit of dispatch is one task per partition, Ballista's unit is one task per +slice of partitions bound to a single executor. Spark achieves cluster-scale parallelism the same way — many +tasks running concurrently across cores — but each task is single-threaded and doesn't share state with its +neighbours. Ballista's slice model preserves cluster-scale parallelism _and_ adds intra-task shared-memory +parallelism: partitions inside one slice share DataFusion's per-task memory pool budget, share the collect-left +build side of a broadcast hash join (one hash table probed by every partition, instead of one materialization +per task), share segment-tree indices needed for degenerate window aggregates (non-invertible aggregates like +MIN/MAX, or wide/data-dependent frames where a sliding accumulator degrades to O(n × frame)), and can cooperate +on shared-memory algorithms like PSRS parallel sort that a shuffle-based system can't express within a stage. +It also unlocks pipelines whose intra-task state must be global-per-slice — e.g. +`SELECT sum(v2) OVER (ORDER BY v2 RANGE 3 PRECEDING) FROM large`, which today collapses onto a single-partition +sort+window and OOMs at h2o 10 GB scale; with the KLL-adaptive range-repartition rewrite that builds on this +model, one slice-task per executor holds the sketch, buffered input, and per-partition halo state inside a +single plan. + +#### Composition with in-flight DataFusion AQE Review Comment: I removed it and made the issue at https://github.com/apache/datafusion-ballista/issues/2445 -- 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]
