andygrove commented on PR #2038: URL: https://github.com/apache/datafusion-ballista/pull/2038#issuecomment-5006035504
Thanks for this β the slice-per-task direction is compelling. I ran this branch on our k3s TPC-H **SF1000** cluster to see how it moves the needle, and wanted to share the numbers plus a root-cause read while the run was going, since a clear pattern showed up early. I stopped the run after Q7 (single iteration) once the signal was consistent, so this is a partial suite β but it's enough to surface a regression on the join-heavy queries that's worth discussing. ### Setup (identical to our published SF1000 baseline, only the image differs) | | | |---|---| | Branch | `avantgardnerio/arrow-ballista@308f1495` (this PR) | | Baseline | `apache/main` (the published [benchmarking-guide](https://datafusion.apache.org/ballista/contributors-guide/benchmarking.html) SF1000 AQE-off numbers, same cluster) | | Scale | SF1000, node-local Parquet | | Cluster | 2 executors, one per node, 8 cores / 56 GiB, `--memory-pool-size=48GB` | | Config | `target_partitions=32`, `prefer_hash_join=false`, `enable_dynamic_filter_pushdown=false`, **AQE off** (`ballista.planner.adaptive.enabled=false`) | | Iterations | 1 | ### Results (Q1βQ7, seconds; ratio = baseline / PR, >1 = PR faster) | Query | PR #2038 | main (AQE off) | ratio | | |---|---|---|---|---| | 1 | 41.3 | 70.8 | **1.71Γ** | π’ | | 2 | 183.2 | 155.5 | 0.85Γ | π΄ | | 3 | 256.5 | 206.2 | 0.80Γ | π΄ | | 4 | 67.2 | 76.8 | **1.14Γ** | π’ | | 5 | 692.9 | 542.7 | 0.78Γ | π΄ | | 6 | 15.0 | 18.3 | **1.22Γ** | π’ | | 7 | 748.7 | 575.5 | 0.77Γ | π΄ | | **Ξ£ 1β7** | **2004.8** | **1645.8** | **0.82Γ** | | The direction is consistent: **join-free queries improve (Q1, Q4, Q6), join-heavy queries regress ~20β25% (Q2, Q3, Q5, Q7)**. ### Root cause: coarser scheduling granularity β straggler tails on skewed join stages The `DIAG bind_one` scheduler traces make the mechanism visible. On a light query, each stage binds cleanly as a few fat slices with no leftover single-partition tasks and a sub-second spread. On a join-heavy query, the same 32-partition stages fragment and dispatch over a long window: | stage kind | tasks | 1-partition fragments | firstβlast bind spread | |---|---|---|---| | light query, stage | 4 | 0 | <1 s | | join query, stage A | 6 | 2 | 100 s | | join query, stage B | 9 | 5 | **178 s** | | join query, worst | 18 | 15 | β | A slice-task holds **all** of its vcore slots until its **slowest** partition finishes. On join stages, per-partition runtime is very uneven (join-key skew + spilling `SortExec` β one partition spilled 22 GB / 455 M rows in an SMJ with `join_timeβ413 s`). So the binder places a couple of fat slices, then has to dribble the remaining partitions out one at a time as slots free (the `frag1` counts), and a whole stage's partitions end up spread across 100β180 s. During each slice's tail, the sibling vcores sit idle and the next partitions can't dispatch. The pre-PR 1-task-per-partition model freed each slot independently, so the 32 partitions load-balanced finely across the 16 vcores. Uniform stages don't hit the tail and actually gain from the lower per-task overhead β hence the wins on Q1/Q4/Q6. A second-order cost: fragmented stages re-encode a restricted plan per task, so a 15-fragment stage pays ~15Γ the plan-serialization of a clean 4-task stage. ### What this does *not* appear to be: memory starvation I initially suspected the shared per-task pool, but `memory_pool_policy` scales it correctly β an N-vcore slice gets `N Γ per_vcore`, so an 8-vcore task gets the full 48 GB `FairSpillPool` split across its 8 pipelines (β6 GB each, same as the old per-task budget, and better at the tail as siblings deregister). The heavy spilling looks inherent to a 455 M-row SMJ sort at ~6 GB/partition, not amplified by this PR. So this reads as a scheduling/utilization regression rather than a memory one. ### Possible directions (just ideas) - Cap or shrink slice size on stages whose child is a join / sort so partition granularity stays fine where runtimes are skewed. - Let a slice release vcores as individual partitions complete, rather than holding all slots until the slowest finishes. Happy to re-run the full 22-query suite (and with AQE on) on any follow-up, or share the raw scheduler/executor logs. Really nice work opening up the parallel-sort / parallel-join path β this is about tuning the scheduling granularity, not the idea. --- *Benchmarking, log analysis, and this write-up were done with LLM assistance (Claude).* -- 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]
