(incubator-gluten) branch main updated: [GLUTEN-6695][CH] Introduce shuffleWallTime in CHMetricsApi to calculate the overall shuffle write time (#6696)

[ch...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

changchen pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-gluten.git


The following commit(s) were added to refs/heads/main by this push:
     new 944c92623 [GLUTEN-6695][CH] Introduce shuffleWallTime in CHMetricsApi to calculate the overall shuffle write time (#6696)
944c92623 is described below

commit 944c92623ae07430d48b7569ae52ec14eefcfb0a
Author: Nicholas Jiang <pr...@163.com>
AuthorDate: Sun Aug 4 14:37:03 2024 +0800

    [GLUTEN-6695][CH] Introduce shuffleWallTime in CHMetricsApi to calculate the overall shuffle write time (#6696)
---
 .../scala/org/apache/gluten/backendsapi/clickhouse/CHMetricsApi.scala | 1 +
 .../org/apache/spark/shuffle/CHCelebornColumnarShuffleWriter.scala    | 4 ++++
 2 files changed, 5 insertions(+)

diff --git a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHMetricsApi.scala b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHMetricsApi.scala
index 85b298fa4..0ff53e1c5 100644
--- a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHMetricsApi.scala
+++ b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHMetricsApi.scala
@@ -222,6 +222,7 @@ class CHMetricsApi extends MetricsApi with Logging with LogLevelUtil {
       "spillTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "time to spill"),
       "compressTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "time to compress"),
       "prepareTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "time to prepare"),
+      "shuffleWallTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "shuffle wall time"),
       "avgReadBatchNumRows" -> SQLMetrics
         .createAverageMetric(sparkContext, "avg read batch num rows"),
       "numInputRows" -> SQLMetrics.createMetric(sparkContext, "number of input rows"),
diff --git a/gluten-celeborn/clickhouse/src/main/scala/org/apache/spark/shuffle/CHCelebornColumnarShuffleWriter.scala b/gluten-celeborn/clickhouse/src/main/scala/org/apache/spark/shuffle/CHCelebornColumnarShuffleWriter.scala
index ae22a0890..c7d7957c1 100644
--- a/gluten-celeborn/clickhouse/src/main/scala/org/apache/spark/shuffle/CHCelebornColumnarShuffleWriter.scala
+++ b/gluten-celeborn/clickhouse/src/main/scala/org/apache/spark/shuffle/CHCelebornColumnarShuffleWriter.scala
@@ -63,7 +63,9 @@ class CHCelebornColumnarShuffleWriter[K, V](
       } else {
         initShuffleWriter(cb)
         val col = cb.column(0).asInstanceOf[CHColumnVector]
+        val startTime = System.nanoTime()
         jniWrapper.split(nativeShuffleWriter, col.getBlockAddress)
+        dep.metrics("shuffleWallTime").add(System.nanoTime() - startTime)
         dep.metrics("numInputRows").add(cb.numRows)
         dep.metrics("inputBatches").add(1)
         // This metric is important, AQE use it to decide if EliminateLimit
@@ -77,8 +79,10 @@ class CHCelebornColumnarShuffleWriter[K, V](
       return
     }
 
+    val startTime = System.nanoTime()
     splitResult = jniWrapper.stop(nativeShuffleWriter)
 
+    dep.metrics("shuffleWallTime").add(System.nanoTime() - startTime)
     dep.metrics("splitTime").add(splitResult.getSplitTime)
     dep.metrics("IOTime").add(splitResult.getDiskWriteTime)
     dep.metrics("serializeTime").add(splitResult.getSerializationTime)


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org