This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-ballista.git
The following commit(s) were added to refs/heads/main by this push:
new 2d938371a feat: log sort shuffle spills at WARN, keep write completion
at DEBUG (#2091)
2d938371a is described below
commit 2d938371a7a881bc65bc8e70e569fde8bd23e694
Author: Andy Grove <[email protected]>
AuthorDate: Mon Jul 20 08:16:27 2026 -0600
feat: log sort shuffle spills at WARN, keep write completion at DEBUG
(#2091)
* feat: improve sort shuffle spill logging and add row count to write log
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.
* feat: log shuffle write/repart/spill timings instead of wall-clock elapsed
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.
---
.../src/execution_plans/sort_shuffle/writer.rs | 53 ++++++++++++++++------
1 file changed, 38 insertions(+), 15 deletions(-)
diff --git a/ballista/core/src/execution_plans/sort_shuffle/writer.rs
b/ballista/core/src/execution_plans/sort_shuffle/writer.rs
index cb4689c68..5a41aeb5c 100644
--- a/ballista/core/src/execution_plans/sort_shuffle/writer.rs
+++ b/ballista/core/src/execution_plans/sort_shuffle/writer.rs
@@ -26,7 +26,7 @@ use std::fs::File;
use std::io::{BufWriter, Seek, Write};
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
-use std::time::Instant;
+use std::time::Duration;
use tokio::sync::oneshot;
use super::super::shuffle_writer::{result_schema, summaries_to_batch};
@@ -62,7 +62,7 @@ use datafusion::physical_plan::{
SendableRecordBatchStream, Statistics, displayable,
};
use futures::{StreamExt, TryStreamExt};
-use log::debug;
+use log::{debug, warn};
/// Result of finalizing shuffle output: (data_path, index_path,
partition_write_stats)
/// where partition_write_stats is (partition_id, num_batches, num_rows,
num_bytes)
@@ -407,7 +407,6 @@ impl SortShuffleWriterExec {
let file_id_usize = file_id as usize;
async move {
- let now = Instant::now();
let mut stream = plan.execute(input_partition, context.clone())?;
let schema = stream.schema();
let ballista_config =
Arc::new(context.session_config().ballista_config());
@@ -549,18 +548,42 @@ impl SortShuffleWriterExec {
// Reservation drops naturally; nothing left to free.
drop(reservation);
- debug!(
- "Sort shuffle write for partition {} completed in {} seconds. \
- Output: {:?}, Index: {:?}, Spill events: {}, Spill batches:
{}, \
- Spill bytes: {}",
- input_partition,
- now.elapsed().as_secs(),
- data_path,
- index_path,
- spill_events,
- total_spilled_batches,
- total_bytes_spilled
- );
+ let repart_time = Duration::from_nanos(metrics.repart_time.value()
as u64);
+ let spill_time = Duration::from_nanos(metrics.spill_time.value()
as u64);
+ let write_time = Duration::from_nanos(metrics.write_time.value()
as u64);
+ if total_bytes_spilled > 0 {
+ warn!(
+ "Sort shuffle spill: job={} stage={} input_partition={} \
+ spilled {} bytes in {} batches ({} events) under memory \
+ pressure; repart_time={:?} spill_time={:?}
write_time={:?}",
+ job_id,
+ stage_id,
+ input_partition,
+ total_bytes_spilled,
+ total_spilled_batches,
+ spill_events,
+ repart_time,
+ spill_time,
+ write_time,
+ );
+ } else {
+ debug!(
+ "Sort shuffle write for partition {} completed. \
+ Output: {:?}, Index: {:?}, Rows: {}, \
+ repart_time={:?} spill_time={:?} write_time={:?}, \
+ Spill events: {}, Spill batches: {}, Spill bytes: {}",
+ input_partition,
+ data_path,
+ index_path,
+ total_rows,
+ repart_time,
+ spill_time,
+ write_time,
+ spill_events,
+ total_spilled_batches,
+ total_bytes_spilled
+ );
+ }
let mut results = Vec::new();
for (part_id, num_batches, num_rows, num_bytes) in partition_stats
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
