add new metrics skippedPages / processedPages

zhztheplayer · zhztheplayer · commit b91845a5b3b5 · 2026-07-07T21:06:45.000+02:00
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
@@ -49,6 +49,8 @@ public class Metrics implements IMetrics {
   public long[] processedSplits;
   public long[] skippedStrides;
   public long[] processedStrides;
+  public long[] skippedPages;
+  public long[] processedPages;
   public long[] remainingFilterTime;
   public long[] ioWaitTime;
   public long[] storageReadBytes;
@@ -103,6 +105,8 @@ public Metrics(
       long[] processedSplits,
       long[] skippedStrides,
       long[] processedStrides,
+      long[] skippedPages,
+      long[] processedPages,
       long[] remainingFilterTime,
       long[] ioWaitTime,
       long[] storageReadBytes,
@@ -149,6 +153,8 @@ public Metrics(
     this.processedSplits = processedSplits;
     this.skippedStrides = skippedStrides;
     this.processedStrides = processedStrides;
+    this.skippedPages = skippedPages;
+    this.processedPages = processedPages;
     this.remainingFilterTime = remainingFilterTime;
     this.ioWaitTime = ioWaitTime;
     this.storageReadBytes = storageReadBytes;
@@ -203,6 +209,8 @@ public OperatorMetrics getOperatorMetrics(int index) {
         processedSplits[index],
         skippedStrides[index],
         processedStrides[index],
+        skippedPages[index],
+        processedPages[index],
         remainingFilterTime[index],
         ioWaitTime[index],
         storageReadBytes[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
@@ -47,6 +47,8 @@ public class OperatorMetrics implements IOperatorMetrics {
   public long processedSplits;
   public long skippedStrides;
   public long processedStrides;
+  public long skippedPages;
+  public long processedPages;
   public long remainingFilterTime;
   public long ioWaitTime;
   public long storageReadBytes;
@@ -96,6 +98,8 @@ public OperatorMetrics(
       long processedSplits,
       long skippedStrides,
       long processedStrides,
+      long skippedPages,
+      long processedPages,
       long remainingFilterTime,
       long ioWaitTime,
       long storageReadBytes,
@@ -140,6 +144,8 @@ public OperatorMetrics(
     this.processedSplits = processedSplits;
     this.skippedStrides = skippedStrides;
     this.processedStrides = processedStrides;
+    this.skippedPages = skippedPages;
+    this.processedPages = processedPages;
     this.remainingFilterTime = remainingFilterTime;
     this.ioWaitTime = ioWaitTime;
     this.storageReadBytes = storageReadBytes;
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
@@ -136,6 +136,8 @@ class VeloxMetricsApi extends MetricsApi with Logging {
         "data source read time"),
       "skippedStrides" -> SQLMetrics.createMetric(sparkContext, "number of skipped row groups"),
       "processedStrides" -> SQLMetrics.createMetric(sparkContext, "number of processed row groups"),
+      "skippedPages" -> SQLMetrics.createMetric(sparkContext, "number of skipped pages"),
+      "processedPages" -> SQLMetrics.createMetric(sparkContext, "number of processed pages"),
       "remainingFilterTime" -> SQLMetrics.createNanoTimingMetric(
         sparkContext,
         "remaining filter time"),
@@ -194,6 +196,8 @@ class VeloxMetricsApi extends MetricsApi with Logging {
         "data source read time"),
       "skippedStrides" -> SQLMetrics.createMetric(sparkContext, "number of skipped row groups"),
       "processedStrides" -> SQLMetrics.createMetric(sparkContext, "number of processed row groups"),
+      "skippedPages" -> SQLMetrics.createMetric(sparkContext, "number of skipped pages"),
+      "processedPages" -> SQLMetrics.createMetric(sparkContext, "number of processed pages"),
       "remainingFilterTime" -> SQLMetrics.createNanoTimingMetric(
         sparkContext,
         "remaining filter time"),
@@ -252,6 +256,8 @@ class VeloxMetricsApi extends MetricsApi with Logging {
         "data source read time"),
       "skippedStrides" -> SQLMetrics.createMetric(sparkContext, "number of skipped row groups"),
       "processedStrides" -> SQLMetrics.createMetric(sparkContext, "number of processed row groups"),
+      "skippedPages" -> SQLMetrics.createMetric(sparkContext, "number of skipped pages"),
+      "processedPages" -> SQLMetrics.createMetric(sparkContext, "number of processed pages"),
       "remainingFilterTime" -> SQLMetrics.createNanoTimingMetric(
         sparkContext,
         "remaining filter 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
@@ -46,6 +46,8 @@ class BatchScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric])
   private val processedSplits: Option[SQLMetric] = metric("processedSplits")
   private val skippedStrides: Option[SQLMetric] = metric("skippedStrides")
   private val processedStrides: Option[SQLMetric] = metric("processedStrides")
+  private val skippedPages: Option[SQLMetric] = metric("skippedPages")
+  private val processedPages: Option[SQLMetric] = metric("processedPages")
   private val remainingFilterTime: Option[SQLMetric] = metric("remainingFilterTime")
   private val ioWaitTime: Option[SQLMetric] = metric("ioWaitTime")
   private val storageReadBytes: Option[SQLMetric] = metric("storageReadBytes")
@@ -84,6 +86,8 @@ class BatchScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric])
       ScanMetricsUtil.inc(processedSplits, operatorMetrics.processedSplits)
       ScanMetricsUtil.inc(skippedStrides, operatorMetrics.skippedStrides)
       ScanMetricsUtil.inc(processedStrides, operatorMetrics.processedStrides)
+      ScanMetricsUtil.inc(skippedPages, operatorMetrics.skippedPages)
+      ScanMetricsUtil.inc(processedPages, operatorMetrics.processedPages)
       ScanMetricsUtil.inc(remainingFilterTime, operatorMetrics.remainingFilterTime)
       ScanMetricsUtil.inc(ioWaitTime, operatorMetrics.ioWaitTime)
       ScanMetricsUtil.inc(storageReadBytes, operatorMetrics.storageReadBytes)
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
@@ -44,6 +44,8 @@ class FileSourceScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric
   private val processedSplits: Option[SQLMetric] = metric("processedSplits")
   private val skippedStrides: Option[SQLMetric] = metric("skippedStrides")
   private val processedStrides: Option[SQLMetric] = metric("processedStrides")
+  private val skippedPages: Option[SQLMetric] = metric("skippedPages")
+  private val processedPages: Option[SQLMetric] = metric("processedPages")
   private val remainingFilterTime: Option[SQLMetric] = metric("remainingFilterTime")
   private val ioWaitTime: Option[SQLMetric] = metric("ioWaitTime")
   private val storageReadBytes: Option[SQLMetric] = metric("storageReadBytes")
@@ -79,6 +81,8 @@ class FileSourceScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric
       ScanMetricsUtil.inc(processedSplits, operatorMetrics.processedSplits)
       ScanMetricsUtil.inc(skippedStrides, operatorMetrics.skippedStrides)
       ScanMetricsUtil.inc(processedStrides, operatorMetrics.processedStrides)
+      ScanMetricsUtil.inc(skippedPages, operatorMetrics.skippedPages)
+      ScanMetricsUtil.inc(processedPages, operatorMetrics.processedPages)
       ScanMetricsUtil.inc(remainingFilterTime, operatorMetrics.remainingFilterTime)
       ScanMetricsUtil.inc(ioWaitTime, operatorMetrics.ioWaitTime)
       ScanMetricsUtil.inc(storageReadBytes, operatorMetrics.storageReadBytes)
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
@@ -40,6 +40,8 @@ class HiveTableScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric]
   private val processedSplits: Option[SQLMetric] = metric("processedSplits")
   private val skippedStrides: Option[SQLMetric] = metric("skippedStrides")
   private val processedStrides: Option[SQLMetric] = metric("processedStrides")
+  private val skippedPages: Option[SQLMetric] = metric("skippedPages")
+  private val processedPages: Option[SQLMetric] = metric("processedPages")
   private val remainingFilterTime: Option[SQLMetric] = metric("remainingFilterTime")
   private val ioWaitTime: Option[SQLMetric] = metric("ioWaitTime")
   private val storageReadBytes: Option[SQLMetric] = metric("storageReadBytes")
@@ -75,6 +77,8 @@ class HiveTableScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric]
       ScanMetricsUtil.inc(processedSplits, operatorMetrics.processedSplits)
       ScanMetricsUtil.inc(skippedStrides, operatorMetrics.skippedStrides)
       ScanMetricsUtil.inc(processedStrides, operatorMetrics.processedStrides)
+      ScanMetricsUtil.inc(skippedPages, operatorMetrics.skippedPages)
+      ScanMetricsUtil.inc(processedPages, operatorMetrics.processedPages)
       ScanMetricsUtil.inc(remainingFilterTime, operatorMetrics.remainingFilterTime)
       ScanMetricsUtil.inc(ioWaitTime, operatorMetrics.ioWaitTime)
       ScanMetricsUtil.inc(storageReadBytes, operatorMetrics.storageReadBytes)
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
@@ -129,6 +129,8 @@ object MetricsUtil extends Logging {
     var skippedSplits: Long = 0
     var processedSplits: Long = 0
     var skippedStrides: Long = 0
+    var skippedPages: Long = 0
+    var processedPages: Long = 0
     var processedStrides: Long = 0
     var remainingFilterTime: Long = 0
     var ioWaitTime: Long = 0
@@ -168,6 +170,8 @@ object MetricsUtil extends Logging {
       processedSplits += metrics.processedSplits
       skippedStrides += metrics.skippedStrides
       processedStrides += metrics.processedStrides
+      skippedPages += metrics.skippedPages
+      processedPages += metrics.processedPages
       remainingFilterTime += metrics.remainingFilterTime
       ioWaitTime += metrics.ioWaitTime
       storageReadBytes += metrics.storageReadBytes
@@ -213,6 +217,8 @@ object MetricsUtil extends Logging {
       processedSplits,
       skippedStrides,
       processedStrides,
+      skippedPages,
+      processedPages,
       remainingFilterTime,
       ioWaitTime,
       storageReadBytes,
diff --git a/cpp/core/jni/JniWrapper.cc b/cpp/core/jni/JniWrapper.cc
@@ -313,7 +313,7 @@ jint JNI_OnLoad(JavaVM* vm, void* reserved) {
       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[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[J[J[J[JLjava/lang/String;)V");
 
   nativeColumnarToRowInfoClass =
       createGlobalClassReferenceOrError(env, "Lorg/apache/gluten/vectorized/NativeColumnarToRowInfo;");
@@ -668,6 +668,8 @@ JNIEXPORT jobject JNICALL Java_org_apache_gluten_metrics_IteratorMetricsJniWrapp
       longArray[Metrics::kProcessedSplits],
       longArray[Metrics::kSkippedStrides],
       longArray[Metrics::kProcessedStrides],
+      longArray[Metrics::kSkippedPages],
+      longArray[Metrics::kProcessedPages],
       longArray[Metrics::kRemainingFilterTime],
       longArray[Metrics::kIoWaitTime],
       longArray[Metrics::kStorageReadBytes],
diff --git a/cpp/core/utils/Metrics.h b/cpp/core/utils/Metrics.h
@@ -77,6 +77,8 @@ struct Metrics {
     kProcessedSplits,
     kSkippedStrides,
     kProcessedStrides,
+    kSkippedPages,
+    kProcessedPages,
     kRemainingFilterTime,
     kIoWaitTime,
     kStorageReadBytes,
diff --git a/cpp/velox/compute/WholeStageResultIterator.cc b/cpp/velox/compute/WholeStageResultIterator.cc
@@ -61,6 +61,8 @@ const std::string kSkippedSplits = "skippedSplits";
 const std::string kProcessedSplits = "processedSplits";
 const std::string kSkippedStrides = "skippedStrides";
 const std::string kProcessedStrides = "processedStrides";
+const std::string kSkippedPages = "skippedPages";
+const std::string kProcessedPages = "processedPages";
 const std::string kRemainingFilterTime = "totalRemainingFilterWallNanos";
 const std::string kIoWaitTime = "ioWaitWallNanos";
 const std::string kStorageReadBytes = "storageReadBytes";
@@ -569,6 +571,10 @@ 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::kSkippedPages)[metricIndex] =
+          runtimeMetric("sum", second->customStats, kSkippedPages);
+      metrics_->get(Metrics::kProcessedPages)[metricIndex] =
+          runtimeMetric("sum", second->customStats, kProcessedPages);
       metrics_->get(Metrics::kRemainingFilterTime)[metricIndex] =
           runtimeMetric("sum", second->customStats, kRemainingFilterTime);
       metrics_->get(Metrics::kIoWaitTime)[metricIndex] = runtimeMetric("sum", second->customStats, kIoWaitTime);
diff --git a/ep/build-velox/src/get-velox.sh b/ep/build-velox/src/get-velox.sh
@@ -166,6 +166,17 @@ function apply_compilation_fixes {
   git add ${VELOX_HOME}/CMake/resolve_dependency_modules/arrow/modify_arrow.patch # to avoid the file from being deleted by git clean -dffx :/
 }
 
+function apply_skipped_processed_pages_patch {
+  echo "Applying skipped/processed pages patch..."
+  pushd $VELOX_HOME
+  git apply --check ${CURRENT_DIR}/skipped-processed-pages.patch && \
+    git apply ${CURRENT_DIR}/skipped-processed-pages.patch || {
+    echo "Failed to apply skipped/processed pages patch"
+    exit 1
+  }
+  popd
+}
+
 function setup_linux {
   local LINUX_DISTRIBUTION=$(. /etc/os-release && echo ${ID})
   local LINUX_VERSION_ID=$(. /etc/os-release && echo ${VERSION_ID})
@@ -248,4 +259,6 @@ apply_provided_velox_patch
 
 apply_compilation_fixes
 
+apply_bloomfilter_final_patch
+
 echo "Finished getting Velox code"
diff --git a/ep/build-velox/src/skipped-processed-pages.patch b/ep/build-velox/src/skipped-processed-pages.patch
@@ -0,0 +1,103 @@
+diff --git a/velox/dwio/common/Statistics.h b/velox/dwio/common/Statistics.h
+index 12483bb4c..1c496416a 100644
+--- a/velox/dwio/common/Statistics.h
++++ b/velox/dwio/common/Statistics.h
+@@ -682,6 +682,13 @@ struct ColumnReaderStatistics {
+   // Total time spent in loading pages, in nanoseconds.
+   io::IoCounter pageLoadTimeNs;
+ 
++  // Number of pages skipped (only page header read, compressed data never
++  // fetched or decompressed).
++  int64_t skippedPages{0};
++
++  // Number of pages processed (compressed data fetched and decompressed).
++  int64_t processedPages{0};
++
+   // Per-column decoding statistics. Only populated when decoding stats
+   // collection is enabled.
+   std::optional<DecodingStatsSet> decodingStatsSet;
+@@ -702,6 +709,8 @@ struct ColumnReaderStatistics {
+   void mergeFrom(const ColumnReaderStatistics& other) {
+     flattenStringDictionaryValues += other.flattenStringDictionaryValues;
+     pageLoadTimeNs.merge(other.pageLoadTimeNs);
++    skippedPages += other.skippedPages;
++    processedPages += other.processedPages;
+     if (other.decodingStatsSet) {
+       if (!decodingStatsSet) {
+         decodingStatsSet.emplace();
+@@ -728,6 +737,12 @@ struct ColumnReaderStatistics {
+               pageLoadTimeNs.max(),
+               RuntimeCounter::Unit::kNanos));
+     }
++    if (skippedPages > 0) {
++      result.emplace("skippedPages", RuntimeMetric(skippedPages));
++    }
++    if (processedPages > 0) {
++      result.emplace("processedPages", RuntimeMetric(processedPages));
++    }
+     if (decodingStatsSet) {
+       decodingStatsSet->toRuntimeMetrics(result);
+     }
+diff --git a/velox/dwio/parquet/reader/PageReader.cpp b/velox/dwio/parquet/reader/PageReader.cpp
+index 3e7065288..edfccb5c3 100644
+--- a/velox/dwio/parquet/reader/PageReader.cpp
++++ b/velox/dwio/parquet/reader/PageReader.cpp
+@@ -73,9 +73,11 @@ void PageReader::seekToPage(int64_t row) {
+               inputStream_.get(),
+               bufferStart_,
+               bufferEnd_);
++          ++stats_.skippedPages;
+           continue;
+         }
+         prepareDictionary(pageHeader);
++        ++stats_.processedPages;
+         continue;
+       default:
+         break; // ignore INDEX page type and any other custom extensions
+@@ -271,6 +273,7 @@ void PageReader::prepareDataPageV1(const PageHeader& pageHeader, int64_t row) {
+         bufferStart_,
+         bufferEnd_);
+ 
++    ++stats_.skippedPages;
+     return;
+   }
+   pageData_ = readBytes(*pageHeader.compressed_page_size(), pageBuffer_);
+@@ -336,6 +339,7 @@ void PageReader::prepareDataPageV1(const PageHeader& pageHeader, int64_t row) {
+ 
+   if (row != kRepDefOnly) {
+     makeDecoder();
++    ++stats_.processedPages;
+   }
+ }
+ 
+@@ -350,6 +354,7 @@ void PageReader::prepareDataPageV2(const PageHeader& pageHeader, int64_t row) {
+         inputStream_.get(),
+         bufferStart_,
+         bufferEnd_);
++    ++stats_.skippedPages;
+     return;
+   }
+ 
+@@ -415,6 +420,7 @@ void PageReader::prepareDataPageV2(const PageHeader& pageHeader, int64_t row) {
+   }
+   if (row != kRepDefOnly) {
+     makeDecoder();
++    ++stats_.processedPages;
+   }
+ }
+ 
+diff --git a/velox/dwio/parquet/reader/ParquetReader.cpp b/velox/dwio/parquet/reader/ParquetReader.cpp
+index 8ece28fed..c78cce994 100644
+--- a/velox/dwio/parquet/reader/ParquetReader.cpp
++++ b/velox/dwio/parquet/reader/ParquetReader.cpp
+@@ -1609,8 +1609,7 @@ class ParquetRowReader::Impl {
+     stats.skippedStrides += skippedStrides_;
+     stats.processedStrides += rowGroupIds_.size();
+     stats.parquetFooterEstimatedBytes += readerBase_->initialThriftSize();
+-    stats.columnReaderStats.pageLoadTimeNs.merge(
+-        columnReaderStats_.pageLoadTimeNs);
++    stats.columnReaderStats.mergeFrom(columnReaderStats_);
+   }
+ 
+   void resetFilterCaches() {
+
