fix: native shuffle now reports spill metrics correctly (apache#3197)

Author: andygrove

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()

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,