From 3a0741b29c63297255a31b64e84d1ea026439293 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Sat, 18 Jul 2026 14:10:01 -0600 Subject: [PATCH 1/2] 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. --- .../execution_plans/sort_shuffle/writer.rs | 43 +++++++++++++------ 1 file changed, 30 insertions(+), 13 deletions(-) diff --git a/ballista/core/src/execution_plans/sort_shuffle/writer.rs b/ballista/core/src/execution_plans/sort_shuffle/writer.rs index 8612acdb33..d85150b17a 100644 --- a/ballista/core/src/execution_plans/sort_shuffle/writer.rs +++ b/ballista/core/src/execution_plans/sort_shuffle/writer.rs @@ -65,7 +65,7 @@ use datafusion::physical_plan::{ SendableRecordBatchStream, Statistics, displayable, }; use futures::{StreamExt, TryFutureExt, TryStreamExt}; -use log::debug; +use log::{debug, warn}; use crate::serde::scheduler::PartitionStats; @@ -350,18 +350,35 @@ 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 elapsed_secs = now.elapsed().as_secs(); + if total_bytes_spilled > 0 { + warn!( + "Sort shuffle spill: job={} stage={} input_partition={} \ + spilled {} bytes in {} batches ({} events) under memory \ + pressure; write completed in {} seconds", + job_id, + stage_id, + input_partition, + total_bytes_spilled, + total_spilled_batches, + spill_events, + elapsed_secs, + ); + } else { + debug!( + "Sort shuffle write for partition {} completed in {} seconds. \ + Output: {:?}, Index: {:?}, Rows: {}, Spill events: {}, \ + Spill batches: {}, Spill bytes: {}", + input_partition, + elapsed_secs, + data_path, + index_path, + total_rows, + 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 { From 47d632dccdc021e98c4e372c3eca5b3c65fecf4d Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Sun, 19 Jul 2026 08:26:49 -0600 Subject: [PATCH 2/2] 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. --- .../execution_plans/sort_shuffle/writer.rs | 24 ++++++++++++------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/ballista/core/src/execution_plans/sort_shuffle/writer.rs b/ballista/core/src/execution_plans/sort_shuffle/writer.rs index d85150b17a..bd7a966570 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::future::Future; use std::io::{BufWriter, Seek, Write}; use std::path::PathBuf; use std::sync::Arc; -use std::time::Instant; +use std::time::Duration; use super::super::shuffle_writer_trait::ShuffleWriter; use super::buffer::BufferedBatches; @@ -213,7 +213,6 @@ impl SortShuffleWriterExec { let partitioning = self.shuffle_output_partitioning.clone(); 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()); @@ -350,30 +349,37 @@ impl SortShuffleWriterExec { // Reservation drops naturally; nothing left to free. drop(reservation); - let elapsed_secs = now.elapsed().as_secs(); + 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; write completed in {} seconds", + pressure; repart_time={:?} spill_time={:?} write_time={:?}", job_id, stage_id, input_partition, total_bytes_spilled, total_spilled_batches, spill_events, - elapsed_secs, + repart_time, + spill_time, + write_time, ); } else { debug!( - "Sort shuffle write for partition {} completed in {} seconds. \ - Output: {:?}, Index: {:?}, Rows: {}, Spill events: {}, \ - Spill batches: {}, Spill bytes: {}", + "Sort shuffle write for partition {} completed. \ + Output: {:?}, Index: {:?}, Rows: {}, \ + repart_time={:?} spill_time={:?} write_time={:?}, \ + Spill events: {}, Spill batches: {}, Spill bytes: {}", input_partition, - elapsed_secs, data_path, index_path, total_rows, + repart_time, + spill_time, + write_time, spill_events, total_spilled_batches, total_bytes_spilled