misc: Use ioCounter for pageLoadTimeNs (#15407)

Summary:
This PR uses `ioCounter` for `pageLoadTimeNs` to collect its min, max, count stats.

Pull Request resolved: #15407

Reviewed By: mbasmanova

Differential Revision: D86894092

Pulled By: Yuhta

fbshipit-source-id: da80213f821aadb24124d16b912cf41d28bc87d7

Author: rui-mo

velox/common/io/IoStatistics.h

@@ -43,6 +43,24 @@ struct OperationCounters {
 
 class IoCounter {
  public:
+  IoCounter& operator=(const IoCounter& other) noexcept {
+    if (this != &other) {
+      count_.store(
+          other.count_.load(std::memory_order_relaxed),
+          std::memory_order_relaxed);
+      sum_.store(
+          other.sum_.load(std::memory_order_relaxed),
+          std::memory_order_relaxed);
+      min_.store(
+          other.min_.load(std::memory_order_relaxed),
+          std::memory_order_relaxed);
+      max_.store(
+          other.max_.load(std::memory_order_relaxed),
+          std::memory_order_relaxed);
+    }
+    return *this;
+  }
+
   uint64_t count() const {
     return count_;
   }

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,7 @@ struct ColumnReaderStatistics {
   int64_t flattenStringDictionaryValues{0};
 
   // Total time spent in loading pages, in nanoseconds.
-  uint64_t pageLoadTimeNs{0};
+  io::IoCounter pageLoadTimeNs;
 };
 
 struct RuntimeStatistics {
@@ -601,11 +602,14 @@ struct RuntimeStatistics {
           "flattenStringDictionaryValues",
           RuntimeMetric(columnReaderStatistics.flattenStringDictionaryValues));
     }
-    if (columnReaderStatistics.pageLoadTimeNs > 0) {
+    if (columnReaderStatistics.pageLoadTimeNs.sum() > 0) {
       result.emplace(
           "pageLoadTimeNs",
           RuntimeMetric(
-              columnReaderStatistics.pageLoadTimeNs,
+              columnReaderStatistics.pageLoadTimeNs.sum(),
+              columnReaderStatistics.pageLoadTimeNs.count(),
+              columnReaderStatistics.pageLoadTimeNs.min(),
+              columnReaderStatistics.pageLoadTimeNs.max(),
               RuntimeCounter::Unit::kNanos));
     }
     return result;

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.merge(
+        columnReaderStats_.pageLoadTimeNs);
   }
 
   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) {