feat: log sort shuffle spills at WARN, keep write completion at DEBUG by andygrove · Pull Request #2091 · apache/datafusion-ballista

andygrove · 2026-07-18T20:10:23Z

Which issue does this PR close?

Part of the observability work tracked in #2081. This extracts the sort
shuffle write/spill logging from the WIP PR #2084 into a small, standalone
change.

Rationale for this change

When a sort shuffle task spills to disk under memory pressure it produced no
log output at the default level, so spilling was effectively invisible.
Spilling is a key signal when diagnosing slow or memory-constrained TPC-H
stages, so it is worth surfacing even at the default log level.

Per reviewer feedback, the routine per-partition write completion message
stays at debug! to avoid adding day-to-day noise.

What changes are included in this PR?

When a task spilled to disk, emit a warn! that names the job, stage, and
input partition and reports the bytes, batches, and events spilled under
memory pressure.
Keep the per-partition shuffle write completion log at debug!, and add
the output row count to that message.

Are there any user-facing changes?

Executor logs now report sort shuffle spill activity at WARN. The routine
write completion message remains at DEBUG. There are no API or
configuration changes.

milenkovicm

thanks @andygrove, just one comment, but fell free to ignore it if you dont agree

milenkovicm · 2026-07-18T20:40:19Z

+                    elapsed_secs,
+                );
+            } else {
+                info!(


can info be debug in this case?
it might be important at this moment but in day-to-day operations probably would be just noise? wdyt ?

I agree this should probably be debug

Claude also wanted debug, so it's 3 against 1 at this point. Updated!

phillipleblanc · 2026-07-19T02:45:15Z

+                    elapsed_secs,
+                );
+            } else {
+                info!(


I agree this should probably be debug

Log sort shuffle spills at warn! with job, stage, and input partition context so memory pressure is visible at the default log level. Keep the per-partition write completion message at debug! and add the output row count to it.

andygrove · 2026-07-19T14:20:24Z

The timings in the log messages are very misleading because they include the time to execute the query that produces the shuffle data. Moving this to draft for now.

…apsed The previous log reported now.elapsed(), which spans the entire input stream consumption and therefore includes upstream query execution, not just shuffle write cost. Report the existing repart_time, spill_time, and write_time metrics instead, which isolate the partition, spill, and finalize phases.

…ite-logging # Conflicts: # ballista/core/src/execution_plans/sort_shuffle/writer.rs

Resolve conflicts from the merged sort-shuffle spill-cap work (apache#2091): adopt the optional proto field and Option<u64> serde handling from main, keep the per-task budget default at 0 (uncapped), and preserve this branch's INFO-level shuffle-write logging while folding in main's repart/spill/write timing breakdown. Take main's refreshed benchmarking.md results as the base for the pending SF1000 re-run.

Drop this branch's change of the sort-shuffle per-task spill-cap default to 0 (uncapped); restore the 256 MiB default from main. The serde plumbing for the value stays (it landed on main via apache#2091). This PR no longer alters the shipped spill-cap default.

* feat: make AQE respect broadcast_join_threshold_bytes The ballista.optimizer.broadcast_join_threshold_bytes config is only consumed by the static distributed planner (maybe_promote_to_broadcast). Under adaptive query planning, broadcast (CollectLeft) selection in DynamicJoinSelectionExec::to_actual_join instead used DataFusion's hash_join_single_partition_threshold (1 MiB default), so the Ballista key had no effect and the effective cutoff was silently a different value. Use broadcast_join_threshold_bytes as the byte threshold in the AQE join selection path, keeping DataFusion's row threshold as the absent-stats fallback. A value of 0 disables broadcast promotion, matching the static planner. This gives a single config key consistent behavior under both planners. Closes #2085 * feat: add Ballista broadcast_join_threshold_rows and use it in AQE Follow-up within the same change: the AQE join-selection path also used DataFusion's hash_join_single_partition_threshold_rows as the row-count fallback. There was no Ballista equivalent, so the row threshold still escaped the single-config goal. Add ballista.optimizer.broadcast_join_threshold_rows (default 128K, mirroring DataFusion's previous default) plus SessionConfigExt accessors, and use it in DynamicJoinSelectionExec::to_actual_join instead of the DataFusion key. AQE broadcast selection now depends only on Ballista config. Document both broadcast thresholds in the AQE tuning guide. * fix: default broadcast_join_threshold_rows to 1M to match existing behavior SessionConfig::new_with_ballista() already installs a 1,000,000 row threshold for DataFusion's hash_join_single_partition_threshold_rows. Default the new Ballista row key to the same value so consolidating AQE onto the Ballista keys does not silently lower the effective row-count broadcast cutoff. * refactor: derive new_with_ballista join thresholds from Ballista config defaults Instead of hard-coding 10 MiB / 1M for the DataFusion hash_join_single_partition_threshold[_rows] session settings, read them from BallistaConfig::default().broadcast_join_threshold_bytes()/_rows(). The Ballista broadcast-threshold defaults are now the single source of truth for both DataFusion's built-in JoinSelection and Ballista's AQE join selection. Values are unchanged. * feat: demote DataFusion CollectLeft joins over the Ballista threshold In the static planner, maybe_promote_to_broadcast trusted any broadcast-safe HashJoinExec(CollectLeft) that DataFusion's JoinSelection produced. DataFusion decides CollectLeft from its own session threshold, which can exceed a runtime override of ballista.optimizer.broadcast_join_threshold_bytes. Demote such a join back to Partitioned when its build side is not under the current Ballista threshold (or when broadcasts are disabled with threshold 0), so the Ballista key is authoritative in the static path too. Null-aware anti joins are never demoted since they require CollectLeft. * docs: apply prettier formatting to AQE tuning-guide table * test: disable AQE broadcast via Ballista threshold in repartition tests The join-selection repartition tests forced the repartition path by setting DataFusion's hash_join_single_partition_threshold[_rows] to 0. AQE join selection now reads the broadcast cutoff from the Ballista config (broadcast_join_threshold_bytes), so those DataFusion keys no longer gate CollectLeft promotion and the small test tables were promoted to broadcast. Set the Ballista broadcast byte threshold to 0 in the helper, which disables CollectLeft promotion and restores the repartitioned plans the snapshots assert. * fix: size the AQE broadcast decision by bytes, not row count `supports_collect_by_thresholds` compared a row count against `hash_join_single_partition_threshold_rows` whenever `total_byte_size` was unknown, so a build side of up to a million arbitrarily wide rows could be broadcast to every probe task without the byte threshold ever applying. Unknown `total_byte_size` is the common case, not an edge case: DataFusion discards it on every join, and rebuilding it in `Statistics::calculate_total_byte_size` only works when every column has a fixed width, so a single `Utf8` column loses it permanently. In TPC-H that covers most dimension-side join results. Estimate the size instead and hold it to the same byte threshold. Each column contributes its own `byte_size` statistic when present -- a total for that column's output, already scaled for filters and limits -- otherwise its fixed width times the row count, otherwise a default width mirroring Spark's `StringType`/`BinaryType` defaults. An overflowing estimate declines the broadcast rather than wrapping to a small number. The row threshold is retained as a ceiling, so this can only reject a broadcast the row rule would have allowed, never introduce a new one. Closes #2081. * test: cover broadcast thresholds with declared statistics The broadcast-vs-partitioned decision is a function of statistics, but the tests around it could only describe tables they were willing to materialise, so the sizes it actually turns on had no coverage: a build side of hundreds of thousands of rows, or one whose `total_byte_size` is unknown. The existing tests instead toggle the decision by zeroing the threshold, which shows the rule is self-consistent but not that the shipped thresholds behave. Add `StatsTable`, a table that declares its statistics and holds no rows, so a fixture can say "800,000 rows of unknown size" in one line. Its scan reports the declared figures and cannot be executed, which is enough for the planner tests, and it deliberately does not recompute `total_byte_size` on projection, since an unknown size is the case these fixtures exist to express. Add tests covering the decision at both edges -- wide rows of unknown size are not broadcast, while small dimensions, narrow rows, and known sizes under the threshold still are -- run under `SessionConfig::new_with_ballista` so they exercise the 10 MB / 1,000,000 row thresholds a deployment ships with rather than DataFusion's defaults, plus a test pinning those defaults directly. `wide_rows_of_unknown_size_are_not_broadcast` fails on the rule that preceded the previous commit and passes with it. The rest pass either way: they guard against the estimate rejecting broadcasts it should allow. Part of #2081. * feat: log AQE join decisions at INFO and shuffle spills at WARN Surface two previously debug-only diagnostics at levels visible under the default INFO filter, so join-strategy and memory-pressure behavior can be observed without enabling debug logging. - AQE dynamic join selection now logs each decision at INFO, naming the resolved action (CollectLeft/Hash/SortMerge/Repartition), the partition mode, and the size-aware inputs (per-side row/byte estimates and the byte/row broadcast thresholds). - Sort-shuffle write completion logs at INFO with row and spill counts; when a partition spills under memory pressure it logs at WARN with the spilled bytes, batches, and event count. Per-batch spill events stay at debug. * feat: default sort-shuffle spill cap to 0 (uncapped) and plumb it through serde The sort-shuffle writer's per-task buffered-bytes cap defaulted to 256 MB, far below the per-task memory-pool budget, so it spilled long before the pool was under pressure. Default the cap to 0, which disables it: spilling is then driven solely by memory-pool pressure. A non-zero value still adds a second spill trigger and is retained for tests and explicit tuning. The cap was also dropped during physical-plan serialization (the executor rebuilt the config with the default), so a configured value never reached the executor. Carry memory_limit_per_task_bytes through the SortShuffleWriterExecNode protobuf and apply it on decode, so an override via ballista.shuffle.sort_based.memory_limit_per_task_bytes takes effect on executors. * feat: add hash_join_max_build_partition_bytes config * feat: add max_per_partition_build_bytes helper for hash-join fit check Reads the actual materialized per-partition byte sizes off the resolved ExchangeExec feeding a Partitioned hash join's build side (the same source CoalescePartitionsRule reads) and returns the MAX rather than the average, since a single oversized partition is enough to OOM even when the average partition is small (the Q18 failure shape). * feat: fall back to SMJ when hash-join build exceeds per-slot budget * feat: log hash-join build-fit decision at INFO * refactor: hoist build-size call and tidy hash-join fit-check docs * docs: add hash-join + safety-fallback SF1000 suite results (2x16, p64) Record a full 22-query SF1000 run with prefer_hash_join=true and the AQE hash-join build-size safety fallback (hash_join_max_build_partition_bytes=64 MiB) on a 2 executor x 16 core cluster at target_partitions=64: all 22 queries complete with no OOM, where a pure hash-join run fails on Q18. Remove the AQE-off column pending a re-run at a matched core count. * docs: refresh SF1000 Ballista AQE-on results on #2084 build Re-ran the TPC-H SF1000 suite (AQE on, target_partitions=64, prefer_hash_join=false, 1 iteration) on the 2x16-core reference cluster against the PR build (becb376). Q1-Q17 from a full-suite run, Q19-Q22 as individual jobs; Q18 still OOMs (Partitioned build side, unchanged). Ballista total (excl. Q18) improves 4817.8 -> 4661.0s, led by the join-heavy queries (Q7 -88s, Q8 -191s, Q9 -169s). * revert: keep upstream 256 MiB sort-shuffle spill-cap default Drop this branch's change of the sort-shuffle per-task spill-cap default to 0 (uncapped); restore the 256 MiB default from main. The serde plumbing for the value stays (it landed on main via #2091). This PR no longer alters the shipped spill-cap default. * docs: note SF1000 results use uncapped sort-shuffle spill override Make explicit that the reference AQE-on numbers were produced with the sort-shuffle per-task spill cap overridden to 0 (uncapped); the shipped default is 256 MiB. Point the Ballista row at the current branch commit. * Update ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs Co-authored-by: Phillip LeBlanc <phillip@spice.ai> * refactor: log per-partition shuffle write at debug; fix log import Demote the per-partition shuffle-write completion log to debug (it fires once per shuffle partition; spills stay at WARN). Update the dynamic_join log import to match the debug! call. --------- Co-authored-by: Phillip LeBlanc <phillip@spice.ai>

andygrove requested a review from milenkovicm July 18, 2026 20:11

andygrove marked this pull request as ready for review July 18, 2026 20:11

andygrove requested a review from avantgardnerio July 18, 2026 20:11

andygrove mentioned this pull request Jul 18, 2026

feat: Improve TPC-H performance with AQE on #2084

Merged

milenkovicm approved these changes Jul 18, 2026

View reviewed changes

phillipleblanc approved these changes Jul 19, 2026

View reviewed changes

andygrove force-pushed the worktree-shuffle-write-logging branch from f1223b2 to 3a0741b Compare July 19, 2026 13:53

andygrove changed the title ~~feat: improve sort shuffle write and spill INFO logging~~ feat: log sort shuffle spills at WARN, keep write completion at DEBUG Jul 19, 2026

andygrove marked this pull request as draft July 19, 2026 14:19

andygrove marked this pull request as ready for review July 20, 2026 13:47

Merge remote-tracking branch 'upstream/main' into worktree-shuffle-wr…

28bfd60

…ite-logging # Conflicts: # ballista/core/src/execution_plans/sort_shuffle/writer.rs

phillipleblanc approved these changes Jul 20, 2026

View reviewed changes

andygrove merged commit 2d93837 into apache:main Jul 20, 2026
18 checks passed

andygrove deleted the worktree-shuffle-write-logging branch July 20, 2026 14:16

Navigation Menu

Uh oh!

feat: log sort shuffle spills at WARN, keep write completion at DEBUG - #2091

feat: log sort shuffle spills at WARN, keep write completion at DEBUG#2091
andygrove merged 3 commits into
apache:mainfrom
andygrove:worktree-shuffle-write-logging

andygrove commented Jul 18, 2026 •

edited

Loading

Uh oh!

milenkovicm left a comment

Uh oh!

milenkovicm Jul 18, 2026

Uh oh!

phillipleblanc Jul 19, 2026

Uh oh!

andygrove Jul 19, 2026

Uh oh!

phillipleblanc Jul 19, 2026

Uh oh!

andygrove commented Jul 19, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Uh oh!

Conversation

andygrove commented Jul 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are there any user-facing changes?

Uh oh!

milenkovicm left a comment

Choose a reason for hiding this comment

Uh oh!

milenkovicm Jul 18, 2026

Choose a reason for hiding this comment

Uh oh!

phillipleblanc Jul 19, 2026

Choose a reason for hiding this comment

Uh oh!

andygrove Jul 19, 2026

Choose a reason for hiding this comment

Uh oh!

phillipleblanc Jul 19, 2026

Choose a reason for hiding this comment

Uh oh!

andygrove commented Jul 19, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

andygrove commented Jul 18, 2026 •

edited

Loading