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 9b7cd7c8d7f..1bcfd67392b 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 @@ -51,6 +51,7 @@ public class Metrics implements IMetrics { public long[] remainingFilterTime; public long[] ioWaitTime; public long[] storageReadBytes; + public long[] storageReads; public long[] localReadBytes; public long[] ramReadBytes; public long[] preloadSplits; @@ -103,6 +104,7 @@ public Metrics( long[] remainingFilterTime, long[] ioWaitTime, long[] storageReadBytes, + long[] storageReads, long[] localReadBytes, long[] ramReadBytes, long[] preloadSplits, @@ -147,6 +149,7 @@ public Metrics( this.remainingFilterTime = remainingFilterTime; this.ioWaitTime = ioWaitTime; this.storageReadBytes = storageReadBytes; + this.storageReads = storageReads; this.localReadBytes = localReadBytes; this.ramReadBytes = ramReadBytes; this.preloadSplits = preloadSplits; @@ -199,6 +202,7 @@ public OperatorMetrics getOperatorMetrics(int index) { remainingFilterTime[index], ioWaitTime[index], storageReadBytes[index], + storageReads[index], localReadBytes[index], ramReadBytes[index], preloadSplits[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 0726fd1b766..d52245a3340 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 @@ -49,6 +49,7 @@ public class OperatorMetrics implements IOperatorMetrics { public long remainingFilterTime; public long ioWaitTime; public long storageReadBytes; + public long storageReads; public long localReadBytes; public long ramReadBytes; public long preloadSplits; @@ -96,6 +97,7 @@ public OperatorMetrics( long remainingFilterTime, long ioWaitTime, long storageReadBytes, + long storageReads, long localReadBytes, long ramReadBytes, long preloadSplits, @@ -138,6 +140,7 @@ public OperatorMetrics( this.remainingFilterTime = remainingFilterTime; this.ioWaitTime = ioWaitTime; this.storageReadBytes = storageReadBytes; + this.storageReads = storageReads; this.localReadBytes = localReadBytes; this.ramReadBytes = ramReadBytes; this.preloadSplits = preloadSplits; 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 9643f4902a0..69ed2f09a99 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 @@ -135,6 +135,7 @@ class VeloxMetricsApi extends MetricsApi with Logging { "remaining filter time"), "ioWaitTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "io wait time"), "storageReadBytes" -> SQLMetrics.createSizeMetric(sparkContext, "storage read bytes"), + "storageReads" -> SQLMetrics.createMetric(sparkContext, "number of storage reads"), "localReadBytes" -> SQLMetrics.createSizeMetric(sparkContext, "local ssd read bytes"), "ramReadBytes" -> SQLMetrics.createSizeMetric(sparkContext, "ram read bytes"), "loadLazyVectorTime" -> SQLMetrics.createNanoTimingMetric( @@ -186,6 +187,7 @@ class VeloxMetricsApi extends MetricsApi with Logging { "remaining filter time"), "ioWaitTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "io wait time"), "storageReadBytes" -> SQLMetrics.createSizeMetric(sparkContext, "storage read bytes"), + "storageReads" -> SQLMetrics.createMetric(sparkContext, "number of storage reads"), "localReadBytes" -> SQLMetrics.createSizeMetric(sparkContext, "local ssd read bytes"), "ramReadBytes" -> SQLMetrics.createSizeMetric(sparkContext, "ram read bytes"), "loadLazyVectorTime" -> SQLMetrics.createNanoTimingMetric( @@ -237,6 +239,7 @@ class VeloxMetricsApi extends MetricsApi with Logging { "remaining filter time"), "ioWaitTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "io wait time"), "storageReadBytes" -> SQLMetrics.createSizeMetric(sparkContext, "storage read bytes"), + "storageReads" -> SQLMetrics.createMetric(sparkContext, "number of storage reads"), "localReadBytes" -> SQLMetrics.createSizeMetric(sparkContext, "local ssd read bytes"), "ramReadBytes" -> SQLMetrics.createSizeMetric(sparkContext, "ram read bytes"), "loadLazyVectorTime" -> SQLMetrics.createNanoTimingMetric( 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 dc404328fa8..9e2b1d7c920 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 @@ -51,6 +51,7 @@ class BatchScanMetricsUpdater(val metrics: Map[String, SQLMetric]) extends Metri metrics("remainingFilterTime") += operatorMetrics.remainingFilterTime metrics("ioWaitTime") += operatorMetrics.ioWaitTime metrics("storageReadBytes") += operatorMetrics.storageReadBytes + metrics("storageReads") += operatorMetrics.storageReads metrics("localReadBytes") += operatorMetrics.localReadBytes metrics("ramReadBytes") += operatorMetrics.ramReadBytes metrics("preloadSplits") += operatorMetrics.preloadSplits 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 1e21fcc00e7..527f8d7bc5f 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 @@ -50,6 +50,7 @@ class FileSourceScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric val remainingFilterTime: SQLMetric = metrics("remainingFilterTime") val ioWaitTime: SQLMetric = metrics("ioWaitTime") val storageReadBytes: SQLMetric = metrics("storageReadBytes") + val storageReads: SQLMetric = metrics("storageReads") val localReadBytes: SQLMetric = metrics("localReadBytes") val ramReadBytes: SQLMetric = metrics("ramReadBytes") val loadLazyVectorTime: SQLMetric = metrics("loadLazyVectorTime") @@ -81,6 +82,7 @@ class FileSourceScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric remainingFilterTime += operatorMetrics.remainingFilterTime ioWaitTime += operatorMetrics.ioWaitTime storageReadBytes += operatorMetrics.storageReadBytes + storageReads += operatorMetrics.storageReads localReadBytes += operatorMetrics.localReadBytes ramReadBytes += operatorMetrics.ramReadBytes preloadSplits += operatorMetrics.preloadSplits 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 a8732b709ae..3330f8b15ae 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 @@ -45,6 +45,7 @@ class HiveTableScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric] val remainingFilterTime: SQLMetric = metrics("remainingFilterTime") val ioWaitTime: SQLMetric = metrics("ioWaitTime") val storageReadBytes: SQLMetric = metrics("storageReadBytes") + val storageReads: SQLMetric = metrics("storageReads") val localReadBytes: SQLMetric = metrics("localReadBytes") val ramReadBytes: SQLMetric = metrics("ramReadBytes") val loadLazyVectorTime: SQLMetric = metrics("loadLazyVectorTime") @@ -76,6 +77,7 @@ class HiveTableScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric] remainingFilterTime += operatorMetrics.remainingFilterTime ioWaitTime += operatorMetrics.ioWaitTime storageReadBytes += operatorMetrics.storageReadBytes + storageReads += operatorMetrics.storageReads localReadBytes += operatorMetrics.localReadBytes ramReadBytes += operatorMetrics.ramReadBytes preloadSplits += operatorMetrics.preloadSplits 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 0567ccc8ceb..f71b2dc342f 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 @@ -132,6 +132,7 @@ object MetricsUtil extends Logging { var remainingFilterTime: Long = 0 var ioWaitTime: Long = 0 var storageReadBytes: Long = 0 + var storageReads: Long = 0 var localReadBytes: Long = 0 var ramReadBytes: Long = 0 var preloadSplits: Long = 0 @@ -168,6 +169,7 @@ object MetricsUtil extends Logging { remainingFilterTime += metrics.remainingFilterTime ioWaitTime += metrics.ioWaitTime storageReadBytes += metrics.storageReadBytes + storageReads += metrics.storageReads localReadBytes += metrics.localReadBytes ramReadBytes += metrics.ramReadBytes preloadSplits += metrics.preloadSplits @@ -211,6 +213,7 @@ object MetricsUtil extends Logging { remainingFilterTime, ioWaitTime, storageReadBytes, + storageReads, localReadBytes, ramReadBytes, preloadSplits, diff --git a/cpp/core/jni/JniWrapper.cc b/cpp/core/jni/JniWrapper.cc index cfa06abfcb8..a08e55b741e 100644 --- a/cpp/core/jni/JniWrapper.cc +++ b/cpp/core/jni/JniWrapper.cc @@ -273,7 +273,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[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[J[J[J[JLjava/lang/String;)V"); nativeColumnarToRowInfoClass = createGlobalClassReferenceOrError(env, "Lorg/apache/gluten/vectorized/NativeColumnarToRowInfo;"); @@ -601,6 +601,7 @@ JNIEXPORT jobject JNICALL Java_org_apache_gluten_metrics_IteratorMetricsJniWrapp longArray[Metrics::kRemainingFilterTime], longArray[Metrics::kIoWaitTime], longArray[Metrics::kStorageReadBytes], + longArray[Metrics::kStorageReads], longArray[Metrics::kLocalReadBytes], longArray[Metrics::kRamReadBytes], longArray[Metrics::kPreloadSplits], diff --git a/cpp/core/utils/Metrics.h b/cpp/core/utils/Metrics.h index 6123f4a7ac5..55f2bb4ef50 100644 --- a/cpp/core/utils/Metrics.h +++ b/cpp/core/utils/Metrics.h @@ -79,6 +79,7 @@ struct Metrics { kRemainingFilterTime, kIoWaitTime, kStorageReadBytes, + kStorageReads, kLocalReadBytes, kRamReadBytes, kPreloadSplits, diff --git a/cpp/velox/compute/WholeStageResultIterator.cc b/cpp/velox/compute/WholeStageResultIterator.cc index 3f85dddb83f..dafca3fd519 100644 --- a/cpp/velox/compute/WholeStageResultIterator.cc +++ b/cpp/velox/compute/WholeStageResultIterator.cc @@ -510,6 +510,8 @@ void WholeStageResultIterator::collectMetrics() { metrics_->get(Metrics::kIoWaitTime)[metricIndex] = runtimeMetric("sum", second->customStats, kIoWaitTime); metrics_->get(Metrics::kStorageReadBytes)[metricIndex] = runtimeMetric("sum", second->customStats, kStorageReadBytes); + metrics_->get(Metrics::kStorageReads)[metricIndex] = + runtimeMetric("count", second->customStats, kStorageReadBytes); metrics_->get(Metrics::kLocalReadBytes)[metricIndex] = runtimeMetric("sum", second->customStats, kLocalReadBytes); metrics_->get(Metrics::kRamReadBytes)[metricIndex] = runtimeMetric("sum", second->customStats, kRamReadBytes); metrics_->get(Metrics::kPreloadSplits)[metricIndex] =