[GLUTTEN-3851][VL] Add remaining filter time metric (#3852)

[VL] Add remaining filter time metric
apache · Nov 28, 2023 · e0f197f · e0f197f
1 parent 8518680
commit e0f197f
Show file tree

Hide file tree

Showing 10 changed files with 39 additions and 10 deletions.
diff --git a/backends-velox/src/main/scala/io/glutenproject/backendsapi/velox/MetricsApiImpl.scala b/backends-velox/src/main/scala/io/glutenproject/backendsapi/velox/MetricsApiImpl.scala
@@ -60,7 +60,10 @@ class MetricsApiImpl extends MetricsApi with Logging {
       "skippedSplits" -> SQLMetrics.createMetric(sparkContext, "number of skipped splits"),
       "processedSplits" -> SQLMetrics.createMetric(sparkContext, "number of processed splits"),
       "skippedStrides" -> SQLMetrics.createMetric(sparkContext, "number of skipped row groups"),
-      "processedStrides" -> SQLMetrics.createMetric(sparkContext, "number of processed row groups")
+      "processedStrides" -> SQLMetrics.createMetric(sparkContext, "number of processed row groups"),
+      "remainingFilterTime" -> SQLMetrics.createNanoTimingMetric(
+        sparkContext,
+        "remaining filter time")
     )
 
   override def genBatchScanTransformerMetricsUpdater(
@@ -96,7 +99,10 @@ class MetricsApiImpl extends MetricsApi with Logging {
       "skippedSplits" -> SQLMetrics.createMetric(sparkContext, "number of skipped splits"),
       "processedSplits" -> SQLMetrics.createMetric(sparkContext, "number of processed splits"),
       "skippedStrides" -> SQLMetrics.createMetric(sparkContext, "number of skipped row groups"),
-      "processedStrides" -> SQLMetrics.createMetric(sparkContext, "number of processed row groups")
+      "processedStrides" -> SQLMetrics.createMetric(sparkContext, "number of processed row groups"),
+      "remainingFilterTime" -> SQLMetrics.createNanoTimingMetric(
+        sparkContext,
+        "remaining filter time")
     )
 
   override def genHiveTableScanTransformerMetricsUpdater(
@@ -132,7 +138,10 @@ class MetricsApiImpl extends MetricsApi with Logging {
       "skippedSplits" -> SQLMetrics.createMetric(sparkContext, "number of skipped splits"),
       "processedSplits" -> SQLMetrics.createMetric(sparkContext, "number of processed splits"),
       "skippedStrides" -> SQLMetrics.createMetric(sparkContext, "number of skipped row groups"),
-      "processedStrides" -> SQLMetrics.createMetric(sparkContext, "number of processed row groups")
+      "processedStrides" -> SQLMetrics.createMetric(sparkContext, "number of processed row groups"),
+      "remainingFilterTime" -> SQLMetrics.createNanoTimingMetric(
+        sparkContext,
+        "remaining filter time")
     )
 
   override def genFileSourceScanTransformerMetricsUpdater(

diff --git a/cpp/core/jni/JniWrapper.cc b/cpp/core/jni/JniWrapper.cc
@@ -253,8 +253,8 @@ jint JNI_OnLoad(JavaVM* vm, void* reserved) {
 
   metricsBuilderClass = createGlobalClassReferenceOrError(env, "Lio/glutenproject/metrics/Metrics;");
 
-  metricsBuilderConstructor =
-      getMethodIdOrError(env, metricsBuilderClass, "<init>", "([J[J[J[J[J[J[J[J[J[JJ[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J)V");
+  metricsBuilderConstructor = getMethodIdOrError(
+      env, metricsBuilderClass, "<init>", "([J[J[J[J[J[J[J[J[J[JJ[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J)V");
 
   serializedColumnarBatchIteratorClass =
       createGlobalClassReferenceOrError(env, "Lio/glutenproject/vectorized/ColumnarBatchInIterator;");
@@ -485,7 +485,8 @@ JNIEXPORT jobject JNICALL Java_io_glutenproject_vectorized_ColumnarBatchOutItera
       longArray[Metrics::kSkippedSplits],
       longArray[Metrics::kProcessedSplits],
       longArray[Metrics::kSkippedStrides],
-      longArray[Metrics::kProcessedStrides]);
+      longArray[Metrics::kProcessedStrides],
+      longArray[Metrics::kRemainingFilterTime]);
 
   JNI_METHOD_END(nullptr)
 }

diff --git a/cpp/core/utils/metrics.h b/cpp/core/utils/metrics.h
@@ -69,6 +69,7 @@ struct Metrics {
     kProcessedSplits,
     kSkippedStrides,
     kProcessedStrides,
+    kRemainingFilterTime,
 
     // The end of enum items.
     kEnd,

diff --git a/cpp/velox/compute/WholeStageResultIterator.cc b/cpp/velox/compute/WholeStageResultIterator.cc
@@ -76,6 +76,7 @@ const std::string kSkippedSplits = "skippedSplits";
 const std::string kProcessedSplits = "processedSplits";
 const std::string kSkippedStrides = "skippedStrides";
 const std::string kProcessedStrides = "processedStrides";
+const std::string kRemainingFilterTime = "totalRemainingFilterTime";
 
 // others
 const std::string kHiveDefaultPartition = "__HIVE_DEFAULT_PARTITION__";
@@ -284,6 +285,8 @@ void WholeStageResultIterator::collectMetrics() {
       metrics_->get(Metrics::kSkippedStrides)[metricIndex] = runtimeMetric("sum", second->customStats, kSkippedStrides);
       metrics_->get(Metrics::kProcessedStrides)[metricIndex] =
           runtimeMetric("sum", second->customStats, kProcessedStrides);
+      metrics_->get(Metrics::kRemainingFilterTime)[metricIndex] =
+          runtimeMetric("sum", second->customStats, kRemainingFilterTime);
       metricIndex += 1;
     }
   }

diff --git a/gluten-data/src/main/java/io/glutenproject/metrics/Metrics.java b/gluten-data/src/main/java/io/glutenproject/metrics/Metrics.java
@@ -44,6 +44,7 @@ public class Metrics implements IMetrics {
   public long[] processedSplits;
   public long[] skippedStrides;
   public long[] processedStrides;
+  public long[] remainingFilterTime;
   public SingleMetric singleMetric = new SingleMetric();
 
   /** Create an instance for native metrics. */
@@ -73,7 +74,8 @@ public Metrics(
       long[] skippedSplits,
       long[] processedSplits,
       long[] skippedStrides,
-      long[] processedStrides) {
+      long[] processedStrides,
+      long[] remainingFilterTime) {
     this.inputRows = inputRows;
     this.inputVectors = inputVectors;
     this.inputBytes = inputBytes;
@@ -100,6 +102,7 @@ public Metrics(
     this.processedSplits = processedSplits;
     this.skippedStrides = skippedStrides;
     this.processedStrides = processedStrides;
+    this.remainingFilterTime = remainingFilterTime;
   }
 
   public OperatorMetrics getOperatorMetrics(int index) {
@@ -132,7 +135,8 @@ public OperatorMetrics getOperatorMetrics(int index) {
         skippedSplits[index],
         processedSplits[index],
         skippedStrides[index],
-        processedStrides[index]);
+        processedStrides[index],
+        remainingFilterTime[index]);
   }
 
   public SingleMetric getSingleMetrics() {

diff --git a/gluten-data/src/main/java/io/glutenproject/metrics/OperatorMetrics.java b/gluten-data/src/main/java/io/glutenproject/metrics/OperatorMetrics.java
@@ -42,6 +42,7 @@ public class OperatorMetrics implements IOperatorMetrics {
   public long processedSplits;
   public long skippedStrides;
   public long processedStrides;
+  public long remainingFilterTime;
 
   /** Create an instance for operator metrics. */
   public OperatorMetrics(
@@ -69,7 +70,8 @@ public OperatorMetrics(
       long skippedSplits,
       long processedSplits,
       long skippedStrides,
-      long processedStrides) {
+      long processedStrides,
+      long remainingFilterTime) {
     this.inputRows = inputRows;
     this.inputVectors = inputVectors;
     this.inputBytes = inputBytes;
@@ -95,5 +97,6 @@ public OperatorMetrics(
     this.processedSplits = processedSplits;
     this.skippedStrides = skippedStrides;
     this.processedStrides = processedStrides;
+    this.remainingFilterTime = remainingFilterTime;
   }
 }
diff --git a/gluten-data/src/main/scala/io/glutenproject/metrics/BatchScanMetricsUpdater.scala b/gluten-data/src/main/scala/io/glutenproject/metrics/BatchScanMetricsUpdater.scala
@@ -48,6 +48,7 @@ class BatchScanMetricsUpdater(val metrics: Map[String, SQLMetric]) extends Metri
       metrics("processedSplits") += operatorMetrics.processedSplits
       metrics("skippedStrides") += operatorMetrics.skippedStrides
       metrics("processedStrides") += operatorMetrics.processedStrides
+      metrics("remainingFilterTime") += operatorMetrics.remainingFilterTime
     }
   }
 }
diff --git a/gluten-data/src/main/scala/io/glutenproject/metrics/FileSourceScanMetricsUpdater.scala b/gluten-data/src/main/scala/io/glutenproject/metrics/FileSourceScanMetricsUpdater.scala
@@ -43,6 +43,7 @@ class FileSourceScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric
   val processedSplits: SQLMetric = metrics("processedSplits")
   val skippedStrides: SQLMetric = metrics("skippedStrides")
   val processedStrides: SQLMetric = metrics("processedStrides")
+  val remainingFilterTime: SQLMetric = metrics("remainingFilterTime")
 
   override def updateInputMetrics(inputMetrics: InputMetricsWrapper): Unit = {
     inputMetrics.bridgeIncBytesRead(rawInputBytes.value)
@@ -68,6 +69,7 @@ class FileSourceScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric
       processedSplits += operatorMetrics.processedSplits
       skippedStrides += operatorMetrics.skippedStrides
       processedStrides += operatorMetrics.processedStrides
+      remainingFilterTime += operatorMetrics.remainingFilterTime
     }
   }
 }
diff --git a/gluten-data/src/main/scala/io/glutenproject/metrics/HiveTableScanMetricsUpdater.scala b/gluten-data/src/main/scala/io/glutenproject/metrics/HiveTableScanMetricsUpdater.scala
@@ -38,6 +38,7 @@ class HiveTableScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric]
   val processedSplits: SQLMetric = metrics("processedSplits")
   val skippedStrides: SQLMetric = metrics("skippedStrides")
   val processedStrides: SQLMetric = metrics("processedStrides")
+  val remainingFilterTime: SQLMetric = metrics("remainingFilterTime")
 
   override def updateInputMetrics(inputMetrics: InputMetricsWrapper): Unit = {
     inputMetrics.bridgeIncBytesRead(rawInputBytes.value)
@@ -63,6 +64,7 @@ class HiveTableScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric]
       processedSplits += operatorMetrics.processedSplits
       skippedStrides += operatorMetrics.skippedStrides
       processedStrides += operatorMetrics.processedStrides
+      remainingFilterTime += operatorMetrics.remainingFilterTime
     }
   }
 }
diff --git a/gluten-data/src/main/scala/io/glutenproject/metrics/MetricsUtil.scala b/gluten-data/src/main/scala/io/glutenproject/metrics/MetricsUtil.scala
@@ -109,6 +109,7 @@ object MetricsUtil extends Logging {
     var processedSplits: Long = 0
     var skippedStrides: Long = 0
     var processedStrides: Long = 0
+    var remainingFilterTime: Long = 0
 
     val metricsIterator = operatorMetrics.iterator()
     while (metricsIterator.hasNext) {
@@ -130,6 +131,7 @@ object MetricsUtil extends Logging {
       processedSplits += metrics.processedSplits
       skippedStrides += metrics.skippedStrides
       processedStrides += metrics.processedStrides
+      remainingFilterTime += metrics.remainingFilterTime
     }
 
     new OperatorMetrics(
@@ -157,7 +159,8 @@ object MetricsUtil extends Logging {
       skippedSplits,
       processedSplits,
       skippedStrides,
-      processedStrides
+      processedStrides,
+      remainingFilterTime
     )
   }