Use ioCounter for pageLoadTimeNs

Author: rui-mo

velox/common/io/IoStatistics.h

@@ -73,6 +73,13 @@ class IoCounter {
     casLoop(max_, other.max_, std::less());
   }
 
+  void clear() {
+    count_ = 0;
+    sum_ = 0;
+    min_ = std::numeric_limits<uint64_t>::max();
+    max_ = 0;
+  }
+
  private:
   template <typename Compare>
   static void

velox/connectors/hive/iceberg/tests/IcebergSplitReaderBenchmark.cpp

@@ -233,7 +233,7 @@ int IcebergSplitReaderBenchmark::read(
     const RowTypePtr& rowType,
     uint32_t nextSize,
     std::unique_ptr<IcebergSplitReader> icebergSplitReader) {
-  runtimeStats_ = RuntimeStatistics();
+  runtimeStats_.clear();
   icebergSplitReader->resetFilterCaches();
   int resultSize = 0;
   auto result = BaseVector::create(rowType, 0, leafPool_.get());

velox/dwio/common/Statistics.h

@@ -22,6 +22,7 @@
 
 #include "velox/common/base/Exceptions.h"
 #include "velox/common/base/RuntimeMetrics.h"
+#include "velox/common/io/IoStatistics.h"
 #include "velox/dwio/common/exception/Exception.h"
 
 namespace facebook::velox::dwio::common {
@@ -540,7 +541,12 @@ struct ColumnReaderStatistics {
   int64_t flattenStringDictionaryValues{0};
 
   // Total time spent in loading pages, in nanoseconds.
-  uint64_t pageLoadTimeNs{0};
+  io::IoCounter pageLoadTimeNs;
+
+  void clear() {
+    flattenStringDictionaryValues = 0;
+    pageLoadTimeNs.clear();
+  }
 };
 
 struct RuntimeStatistics {
@@ -566,6 +572,18 @@ struct RuntimeStatistics {
   UnitLoaderStats unitLoaderStats;
   ColumnReaderStatistics columnReaderStatistics;
 
+  void clear() {
+    skippedSplits = 0;
+    processedSplits = 0;
+    skippedSplitBytes = 0;
+    skippedStrides = 0;
+    processedStrides = 0;
+    footerBufferOverread = 0;
+    numStripes = 0;
+    unitLoaderStats = UnitLoaderStats();
+    columnReaderStatistics.clear();
+  }
+
   std::unordered_map<std::string, RuntimeMetric> toRuntimeMetricMap() {
     std::unordered_map<std::string, RuntimeMetric> result;
     for (const auto& [name, metric] : unitLoaderStats.stats()) {
@@ -601,10 +619,10 @@ struct RuntimeStatistics {
           "flattenStringDictionaryValues",
           RuntimeMetric(columnReaderStatistics.flattenStringDictionaryValues));
     }
-    if (columnReaderStatistics.pageLoadTimeNs > 0) {
+    if (columnReaderStatistics.pageLoadTimeNs.sum() > 0) {
       result.emplace(
           "pageLoadTimeNs",
-          RuntimeMetric(columnReaderStatistics.pageLoadTimeNs));
+          RuntimeMetric(columnReaderStatistics.pageLoadTimeNs.sum()));
     }
     return result;
   }

velox/dwio/common/tests/utils/E2EFilterTestBase.cpp

@@ -163,7 +163,7 @@ void E2EFilterTestBase::readWithFilter(
   }
   OwnershipChecker ownershipChecker;
   auto rowReader = reader->createRowReader(rowReaderOpts);
-  runtimeStats_ = dwio::common::RuntimeStatistics();
+  runtimeStats_.clear();
   auto rowIndex = 0;
   auto resultBatch = BaseVector::create(rowType_, 1, leafPool_.get());
   resetReadBatchSizes();

velox/dwio/dwrf/test/ReaderTest.cpp

@@ -2585,7 +2585,7 @@ TEST_F(TestReader, readStringDictionaryAsFlat) {
   ASSERT_EQ(rowReader->next(20, actual), 20);
   ASSERT_EQ(actual->size(), 1);
   ASSERT_TRUE(actual->as<RowVector>()->childAt(0)->isFlatEncoding());
-  stats = {};
+  stats.clear();
   rowReader->updateRuntimeStats(stats);
   ASSERT_EQ(stats.columnReaderStatistics.flattenStringDictionaryValues, 1);
 }

velox/dwio/parquet/reader/PageReader.cpp

@@ -92,7 +92,7 @@ PageHeader PageReader::readPageHeader() {
       MicrosecondTimer timer(&readUs);
       inputStream_->Next(&buffer, &size);
     }
-    stats_.pageLoadTimeNs += readUs * 1'000;
+    stats_.pageLoadTimeNs.increment(readUs * 1'000);
     bufferStart_ = reinterpret_cast<const char*>(buffer);
     bufferEnd_ = bufferStart_ + size;
   }
@@ -135,7 +135,7 @@ const char* PageReader::readBytes(int32_t size, BufferPtr& copy) {
         bufferStart_,
         bufferEnd_);
   }
-  stats_.pageLoadTimeNs += readUs * 1'000;
+  stats_.pageLoadTimeNs.increment(readUs * 1'000);
   return copy->as<char>();
 }
 
@@ -388,7 +388,7 @@ void PageReader::prepareDictionary(const PageHeader& pageHeader) {
               bufferStart_,
               bufferEnd_);
         }
-        stats_.pageLoadTimeNs += readUs * 1'000;
+        stats_.pageLoadTimeNs.increment(readUs * 1'000);
       }
       if (type_->type()->isShortDecimal() &&
           parquetType == thrift::Type::INT32) {
@@ -428,7 +428,7 @@ void PageReader::prepareDictionary(const PageHeader& pageHeader) {
               bufferStart_,
               bufferEnd_);
         }
-        stats_.pageLoadTimeNs += readUs * 1'000;
+        stats_.pageLoadTimeNs.increment(readUs * 1'000);
       }
       // Expand the Parquet type length values to Velox type length.
       // We start from the end to allow in-place expansion.
@@ -461,7 +461,7 @@ void PageReader::prepareDictionary(const PageHeader& pageHeader) {
           dwio::common::readBytes(
               numBytes, inputStream_.get(), strings, bufferStart_, bufferEnd_);
         }
-        stats_.pageLoadTimeNs += readUs * 1'000;
+        stats_.pageLoadTimeNs.increment(readUs * 1'000);
       }
       auto header = strings;
       for (auto i = 0; i < dictionary_.numValues; ++i) {
@@ -493,7 +493,7 @@ void PageReader::prepareDictionary(const PageHeader& pageHeader) {
               bufferStart_,
               bufferEnd_);
         }
-        stats_.pageLoadTimeNs += readUs * 1'000;
+        stats_.pageLoadTimeNs.increment(readUs * 1'000);
       }
       if (type_->type()->isShortDecimal()) {
         // Parquet decimal values have a fixed typeLength_ and are in big-endian

velox/dwio/parquet/reader/ParquetReader.cpp

@@ -1336,8 +1336,8 @@ class ParquetRowReader::Impl {
   void updateRuntimeStats(dwio::common::RuntimeStatistics& stats) const {
     stats.skippedStrides += skippedStrides_;
     stats.processedStrides += rowGroupIds_.size();
-    stats.columnReaderStatistics.pageLoadTimeNs +=
-        columnReaderStats_.pageLoadTimeNs;
+    stats.columnReaderStatistics.pageLoadTimeNs.increment(
+        columnReaderStats_.pageLoadTimeNs.sum());
   }
 
   void resetFilterCaches() {

velox/dwio/parquet/tests/reader/ParquetPageReaderTest.cpp

@@ -52,7 +52,7 @@ TEST_F(ParquetPageReaderTest, smallPage) {
   auto maxValue = header.data_page_header.statistics.max_value;
   EXPECT_EQ(minValue, expectedMinValue);
   EXPECT_EQ(maxValue, expectedMaxValue);
-  EXPECT_GT(stats.pageLoadTimeNs, 0);
+  EXPECT_GT(stats.pageLoadTimeNs.sum(), 0);
 }
 
 TEST_F(ParquetPageReaderTest, largePage) {
@@ -84,7 +84,7 @@ TEST_F(ParquetPageReaderTest, largePage) {
   auto maxValue = header.data_page_header.statistics.max_value;
   EXPECT_EQ(minValue, expectedMinValue);
   EXPECT_EQ(maxValue, expectedMaxValue);
-  EXPECT_GT(stats.pageLoadTimeNs, 0);
+  EXPECT_GT(stats.pageLoadTimeNs.sum(), 0);
 }
 
 TEST_F(ParquetPageReaderTest, corruptedPageHeader) {

velox/dwio/parquet/tests/reader/ParquetReaderBenchmark.cpp

@@ -124,7 +124,7 @@ int ParquetReaderBenchmark::read(
     std::shared_ptr<ScanSpec> scanSpec,
     uint32_t nextSize) {
   auto rowReader = createReader(scanSpec, rowType);
-  runtimeStats_ = dwio::common::RuntimeStatistics();
+  runtimeStats_.clear();
 
   rowReader->resetFilterCaches();
   auto result = BaseVector::create(rowType, 1, leafPool_.get());