You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by mb...@apache.org on 2026/01/16 14:31:31 UTC
(datafusion-comet) branch main updated: fix: native shuffle now reports spill metrics correctly (#3197)
This is an automated email from the ASF dual-hosted git repository.
mbutrovich pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 6c34c7a59 fix: native shuffle now reports spill metrics correctly (#3197)
6c34c7a59 is described below
commit 6c34c7a592cdea1ac80ba1a602f7b233840babf0
Author: Andy Grove <ag...@apache.org>
AuthorDate: Fri Jan 16 07:31:27 2026 -0700
fix: native shuffle now reports spill metrics correctly (#3197)
---
native/core/src/execution/shuffle/shuffle_writer.rs | 2 +-
.../sql/comet/execution/shuffle/CometNativeShuffleWriter.scala | 8 ++++++++
2 files changed, 9 insertions(+), 1 deletion(-)
diff --git a/native/core/src/execution/shuffle/shuffle_writer.rs b/native/core/src/execution/shuffle/shuffle_writer.rs
index c4dcddc54..f21cde2ba 100644
--- a/native/core/src/execution/shuffle/shuffle_writer.rs
+++ b/native/core/src/execution/shuffle/shuffle_writer.rs
@@ -700,7 +700,7 @@ impl MultiPartitionShuffleRepartitioner {
}
fn spill(&mut self) -> Result<()> {
- log::debug!(
+ log::info!(
"ShuffleRepartitioner spilling shuffle data of {} to disk while inserting ({} time(s) so far)",
self.used(),
self.spill_count()
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala
index 9fd101854..b5d15b41f 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala
@@ -132,6 +132,14 @@ class CometNativeShuffleWriter[K, V](
metricsReporter.incRecordsWritten(metricsOutputRows.value)
metricsReporter.incWriteTime(metricsWriteTime.value)
+ // Report spill metrics to Spark's task metrics so they appear in
+ // Spark UI task summaries (not just SQL metrics)
+ val spilledBytes = nativeSQLMetrics.get("spilled_bytes").map(_.value).getOrElse(0L)
+ if (spilledBytes > 0) {
+ context.taskMetrics().incMemoryBytesSpilled(spilledBytes)
+ context.taskMetrics().incDiskBytesSpilled(spilledBytes)
+ }
+
// commit
shuffleBlockResolver.writeMetadataFileAndCommit(
shuffleId,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org