From 408fa22423a1f293c996f2700bd4e535d79dce7d Mon Sep 17 00:00:00 2001 From: Rui Mo Date: Wed, 30 Jul 2025 17:31:55 +0100 Subject: [PATCH] Add metric --- .../src/main/java/org/apache/gluten/metrics/Metrics.java | 4 ++++ .../main/java/org/apache/gluten/metrics/OperatorMetrics.java | 3 +++ .../org/apache/gluten/backendsapi/velox/VeloxMetricsApi.scala | 3 +++ .../org/apache/gluten/metrics/BatchScanMetricsUpdater.scala | 1 + .../apache/gluten/metrics/FileSourceScanMetricsUpdater.scala | 2 ++ .../apache/gluten/metrics/HiveTableScanMetricsUpdater.scala | 2 ++ .../main/scala/org/apache/gluten/metrics/MetricsUtil.scala | 3 +++ cpp/core/jni/JniWrapper.cc | 3 ++- cpp/core/utils/Metrics.h | 1 + cpp/velox/compute/WholeStageResultIterator.cc | 2 ++ 10 files changed, 23 insertions(+), 1 deletion(-) diff --git a/backends-velox/src/main/java/org/apache/gluten/metrics/Metrics.java b/backends-velox/src/main/java/org/apache/gluten/metrics/Metrics.java index 5515dd9300a..940daf977a1 100644 --- a/backends-velox/src/main/java/org/apache/gluten/metrics/Metrics.java +++ b/backends-velox/src/main/java/org/apache/gluten/metrics/Metrics.java @@ -52,6 +52,7 @@ public class Metrics implements IMetrics { public long[] localReadBytes; public long[] ramReadBytes; public long[] preloadSplits; + public long[] pageLoadTime; public long[] dataSourceAddSplitTime; public long[] dataSourceReadTime; @@ -101,6 +102,7 @@ public Metrics( long[] localReadBytes, long[] ramReadBytes, long[] preloadSplits, + long[] pageLoadTime, long[] dataSourceAddSplitTime, long[] dataSourceReadTime, long[] physicalWrittenBytes, @@ -142,6 +144,7 @@ public Metrics( this.localReadBytes = localReadBytes; this.ramReadBytes = ramReadBytes; this.preloadSplits = preloadSplits; + this.pageLoadTime = pageLoadTime; this.dataSourceAddSplitTime = dataSourceAddSplitTime; this.dataSourceReadTime = dataSourceReadTime; @@ -191,6 +194,7 @@ public OperatorMetrics getOperatorMetrics(int index) { localReadBytes[index], ramReadBytes[index], preloadSplits[index], + pageLoadTime[index], dataSourceAddSplitTime[index], dataSourceReadTime[index], physicalWrittenBytes[index], diff --git a/backends-velox/src/main/java/org/apache/gluten/metrics/OperatorMetrics.java b/backends-velox/src/main/java/org/apache/gluten/metrics/OperatorMetrics.java index 9292caebeb4..0f4ab96cbc3 100644 --- a/backends-velox/src/main/java/org/apache/gluten/metrics/OperatorMetrics.java +++ b/backends-velox/src/main/java/org/apache/gluten/metrics/OperatorMetrics.java @@ -50,6 +50,7 @@ public class OperatorMetrics implements IOperatorMetrics { public long localReadBytes; public long ramReadBytes; public long preloadSplits; + public long pageLoadTime; public long dataSourceAddSplitTime; public long dataSourceReadTime; @@ -94,6 +95,7 @@ public OperatorMetrics( long localReadBytes, long ramReadBytes, long preloadSplits, + long pageLoadTime, long dataSourceAddSplitTime, long dataSourceReadTime, long physicalWrittenBytes, @@ -133,6 +135,7 @@ public OperatorMetrics( this.localReadBytes = localReadBytes; this.ramReadBytes = ramReadBytes; this.preloadSplits = preloadSplits; + this.pageLoadTime = pageLoadTime; this.dataSourceAddSplitTime = dataSourceAddSplitTime; this.dataSourceReadTime = dataSourceReadTime; this.physicalWrittenBytes = physicalWrittenBytes; diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxMetricsApi.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxMetricsApi.scala index 03579e307b6..70597442daa 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxMetricsApi.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxMetricsApi.scala @@ -108,6 +108,7 @@ class VeloxMetricsApi extends MetricsApi with Logging { "skippedSplits" -> SQLMetrics.createMetric(sparkContext, "number of skipped splits"), "processedSplits" -> SQLMetrics.createMetric(sparkContext, "number of processed splits"), "preloadSplits" -> SQLMetrics.createMetric(sparkContext, "number of preloaded splits"), + "pageLoadTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "page load time"), "dataSourceAddSplitTime" -> SQLMetrics.createNanoTimingMetric( sparkContext, "data source add split time"), @@ -158,6 +159,7 @@ class VeloxMetricsApi extends MetricsApi with Logging { "skippedSplits" -> SQLMetrics.createMetric(sparkContext, "number of skipped splits"), "processedSplits" -> SQLMetrics.createMetric(sparkContext, "number of processed splits"), "preloadSplits" -> SQLMetrics.createMetric(sparkContext, "number of preloaded splits"), + "pageLoadTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "page load time"), "dataSourceAddSplitTime" -> SQLMetrics.createNanoTimingMetric( sparkContext, "data source add split time"), @@ -208,6 +210,7 @@ class VeloxMetricsApi extends MetricsApi with Logging { "skippedSplits" -> SQLMetrics.createMetric(sparkContext, "number of skipped splits"), "processedSplits" -> SQLMetrics.createMetric(sparkContext, "number of processed splits"), "preloadSplits" -> SQLMetrics.createMetric(sparkContext, "number of preloaded splits"), + "pageLoadTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "page load time"), "dataSourceAddSplitTime" -> SQLMetrics.createNanoTimingMetric( sparkContext, "data source add split time"), diff --git a/backends-velox/src/main/scala/org/apache/gluten/metrics/BatchScanMetricsUpdater.scala b/backends-velox/src/main/scala/org/apache/gluten/metrics/BatchScanMetricsUpdater.scala index 07b5c301070..dc404328fa8 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/metrics/BatchScanMetricsUpdater.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/metrics/BatchScanMetricsUpdater.scala @@ -54,6 +54,7 @@ class BatchScanMetricsUpdater(val metrics: Map[String, SQLMetric]) extends Metri metrics("localReadBytes") += operatorMetrics.localReadBytes metrics("ramReadBytes") += operatorMetrics.ramReadBytes metrics("preloadSplits") += operatorMetrics.preloadSplits + metrics("pageLoadTime") += operatorMetrics.pageLoadTime metrics("dataSourceAddSplitTime") += operatorMetrics.dataSourceAddSplitTime metrics("dataSourceReadTime") += operatorMetrics.dataSourceReadTime metrics("loadLazyVectorTime") += operatorMetrics.loadLazyVectorTime diff --git a/backends-velox/src/main/scala/org/apache/gluten/metrics/FileSourceScanMetricsUpdater.scala b/backends-velox/src/main/scala/org/apache/gluten/metrics/FileSourceScanMetricsUpdater.scala index 4ba0195802d..1e21fcc00e7 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/metrics/FileSourceScanMetricsUpdater.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/metrics/FileSourceScanMetricsUpdater.scala @@ -42,6 +42,7 @@ class FileSourceScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric val skippedSplits: SQLMetric = metrics("skippedSplits") val processedSplits: SQLMetric = metrics("processedSplits") val preloadSplits: SQLMetric = metrics("preloadSplits") + val pageLoadTime: SQLMetric = metrics("pageLoadTime") val dataSourceAddSplitTime: SQLMetric = metrics("dataSourceAddSplitTime") val dataSourceReadTime: SQLMetric = metrics("dataSourceReadTime") val skippedStrides: SQLMetric = metrics("skippedStrides") @@ -83,6 +84,7 @@ class FileSourceScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric localReadBytes += operatorMetrics.localReadBytes ramReadBytes += operatorMetrics.ramReadBytes preloadSplits += operatorMetrics.preloadSplits + pageLoadTime += operatorMetrics.pageLoadTime dataSourceAddSplitTime += operatorMetrics.dataSourceAddSplitTime dataSourceReadTime += operatorMetrics.dataSourceReadTime loadLazyVectorTime += operatorMetrics.loadLazyVectorTime diff --git a/backends-velox/src/main/scala/org/apache/gluten/metrics/HiveTableScanMetricsUpdater.scala b/backends-velox/src/main/scala/org/apache/gluten/metrics/HiveTableScanMetricsUpdater.scala index c061540ab61..a8732b709ae 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/metrics/HiveTableScanMetricsUpdater.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/metrics/HiveTableScanMetricsUpdater.scala @@ -37,6 +37,7 @@ class HiveTableScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric] val skippedSplits: SQLMetric = metrics("skippedSplits") val processedSplits: SQLMetric = metrics("processedSplits") val preloadSplits: SQLMetric = metrics("preloadSplits") + val pageLoadTime: SQLMetric = metrics("pageLoadTime") val dataSourceAddSplitTime: SQLMetric = metrics("dataSourceAddSplitTime") val dataSourceReadTime: SQLMetric = metrics("dataSourceReadTime") val skippedStrides: SQLMetric = metrics("skippedStrides") @@ -78,6 +79,7 @@ class HiveTableScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric] localReadBytes += operatorMetrics.localReadBytes ramReadBytes += operatorMetrics.ramReadBytes preloadSplits += operatorMetrics.preloadSplits + pageLoadTime += operatorMetrics.pageLoadTime dataSourceAddSplitTime += operatorMetrics.dataSourceAddSplitTime dataSourceReadTime += operatorMetrics.dataSourceReadTime loadLazyVectorTime += operatorMetrics.loadLazyVectorTime diff --git a/backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala b/backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala index 3fd44456b1d..0624fd1f2db 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala @@ -133,6 +133,7 @@ object MetricsUtil extends Logging { var localReadBytes: Long = 0 var ramReadBytes: Long = 0 var preloadSplits: Long = 0 + var pageLoadTime: Long = 0 var dataSourceAddSplitTime: Long = 0 var dataSourceReadTime: Long = 0 var numWrittenFiles: Long = 0 @@ -166,6 +167,7 @@ object MetricsUtil extends Logging { localReadBytes += metrics.localReadBytes ramReadBytes += metrics.ramReadBytes preloadSplits += metrics.preloadSplits + pageLoadTime += metrics.pageLoadTime dataSourceAddSplitTime += metrics.dataSourceAddSplitTime dataSourceReadTime += metrics.dataSourceReadTime numWrittenFiles += metrics.numWrittenFiles @@ -206,6 +208,7 @@ object MetricsUtil extends Logging { localReadBytes, ramReadBytes, preloadSplits, + pageLoadTime, dataSourceAddSplitTime, dataSourceReadTime, physicalWrittenBytes, diff --git a/cpp/core/jni/JniWrapper.cc b/cpp/core/jni/JniWrapper.cc index fe69708192e..0420f5a6c35 100644 --- a/cpp/core/jni/JniWrapper.cc +++ b/cpp/core/jni/JniWrapper.cc @@ -259,7 +259,7 @@ jint JNI_OnLoad(JavaVM* vm, void* reserved) { env, metricsBuilderClass, "", - "([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[J[J[J[J[J[J[J[J[J[J[J[J[JLjava/lang/String;)V"); + "([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[J[J[J[J[J[J[J[J[J[J[J[J[J[JLjava/lang/String;)V"); nativeColumnarToRowInfoClass = createGlobalClassReferenceOrError(env, "Lorg/apache/gluten/vectorized/NativeColumnarToRowInfo;"); @@ -585,6 +585,7 @@ JNIEXPORT jobject JNICALL Java_org_apache_gluten_metrics_IteratorMetricsJniWrapp longArray[Metrics::kLocalReadBytes], longArray[Metrics::kRamReadBytes], longArray[Metrics::kPreloadSplits], + longArray[Metrics::kPageLoadTime], longArray[Metrics::kDataSourceAddSplitWallNanos], longArray[Metrics::kDataSourceReadWallNanos], longArray[Metrics::kPhysicalWrittenBytes], diff --git a/cpp/core/utils/Metrics.h b/cpp/core/utils/Metrics.h index 01d25182cd6..c271a28a4f8 100644 --- a/cpp/core/utils/Metrics.h +++ b/cpp/core/utils/Metrics.h @@ -80,6 +80,7 @@ struct Metrics { kLocalReadBytes, kRamReadBytes, kPreloadSplits, + kPageLoadTime, kDataSourceAddSplitWallNanos, kDataSourceReadWallNanos, diff --git a/cpp/velox/compute/WholeStageResultIterator.cc b/cpp/velox/compute/WholeStageResultIterator.cc index 4aacc43b7f7..a0b5b978504 100644 --- a/cpp/velox/compute/WholeStageResultIterator.cc +++ b/cpp/velox/compute/WholeStageResultIterator.cc @@ -52,6 +52,7 @@ const std::string kStorageReadBytes = "storageReadBytes"; const std::string kLocalReadBytes = "localReadBytes"; const std::string kRamReadBytes = "ramReadBytes"; const std::string kPreloadSplits = "readyPreloadedSplits"; +const std::string kPageLoadTime = "pageLoadTimeNs"; const std::string kDataSourceAddSplitWallNanos = "dataSourceAddSplitWallNanos"; const std::string kDataSourceReadWallNanos = "dataSourceReadWallNanos"; const std::string kNumWrittenFiles = "numWrittenFiles"; @@ -496,6 +497,7 @@ void WholeStageResultIterator::collectMetrics() { metrics_->get(Metrics::kRamReadBytes)[metricIndex] = runtimeMetric("sum", second->customStats, kRamReadBytes); metrics_->get(Metrics::kPreloadSplits)[metricIndex] = runtimeMetric("sum", entry.second->customStats, kPreloadSplits); + metrics_->get(Metrics::kPageLoadTime)[metricIndex] = runtimeMetric("sum", second->customStats, kPageLoadTime); metrics_->get(Metrics::kDataSourceAddSplitWallNanos)[metricIndex] = runtimeMetric("sum", second->customStats, kDataSourceAddSplitWallNanos); metrics_->get(Metrics::kDataSourceReadWallNanos)[metricIndex] =