fix: Change count metric from signed to unsigned (int64_t -> uint64_t)

velox/common/base/RuntimeMetrics.cpp

@@ -30,7 +30,7 @@
 }
 
 void RuntimeMetric::aggregate() {
-  count = std::min(count, static_cast<int64_t>(1));
+  count = std::min(count, static_cast<uint64_t>(1));
   min = max = sum;
 }
 

velox/common/base/RuntimeMetrics.h

@@ -23,6 +23,13 @@
 
 namespace facebook::velox {
 
+/// Converts unsigned bigint to signed, capping at int64_t max if overflow
+/// happens. Could be replaced by 'std::saturate_cast' since C++26.
+inline int64_t saturateCast(uint64_t value) {
+  return static_cast<int64_t>(std::min(
+      value, static_cast<uint64_t>(std::numeric_limits<int64_t>::max())));
+}
+
 struct RuntimeCounter {
   enum class Unit { kNone, kNanos, kBytes };
   int64_t value;
@@ -36,7 +43,7 @@ struct RuntimeMetric {
   // Sum, min, max have the same unit, count has kNone.
   RuntimeCounter::Unit unit;
   int64_t sum{0};
-  int64_t count{0};
+  uint64_t count{0};
   int64_t min{std::numeric_limits<int64_t>::max()};
   int64_t max{std::numeric_limits<int64_t>::min()};
 
@@ -51,7 +58,7 @@ struct RuntimeMetric {
 
   explicit RuntimeMetric(
       int64_t _sum,
-      int64_t _count,
+      uint64_t _count,
       int64_t _min,
       int64_t _max,
       RuntimeCounter::Unit _unit = RuntimeCounter::Unit::kNone)

velox/common/base/tests/RuntimeMetricsTest.cpp

@@ -24,7 +24,7 @@
   static void testMetric(
       const RuntimeMetric& rm1,
       int64_t expectedSum,
-      int64_t expectedCount,
+      uint64_t expectedCount,
       int64_t expectedMin = std::numeric_limits<int64_t>::max(),
       int64_t expectedMax = std::numeric_limits<int64_t>::min()) {
     EXPECT_EQ(expectedSum, rm1.sum);
@@ -84,4 +84,19 @@
       "sum:2.00us, count:1, min:2.00us, max:2.00us, avg: 2.00us");
 }
 
+TEST_F(RuntimeMetricsTest, saturateCast) {
+  auto maxUint64 = std::numeric_limits<uint64_t>::max();
+  RuntimeMetric rm{
+      saturateCast(maxUint64),
+      maxUint64,
+      saturateCast(maxUint64),
+      saturateCast(maxUint64)};
+
+  auto maxInt64 = std::numeric_limits<int64_t>::max();
+  EXPECT_EQ(rm.sum, maxInt64);
+  EXPECT_EQ(rm.count, maxUint64);
+  EXPECT_EQ(rm.min, maxInt64);
+  EXPECT_EQ(rm.max, maxInt64);
+}
+
 } // namespace facebook::velox

velox/connectors/hive/HiveDataSink.cpp

@@ -1533,7 +1533,8 @@
       ioStats_->rawBytesWritten() - writtenBytesBeforeReclaim;
   addThreadLocalRuntimeStat(
       kEarlyFlushedRawBytes,
-      RuntimeCounter(earlyFlushedRawBytes, RuntimeCounter::Unit::kBytes));
+      RuntimeCounter(
+          saturateCast(earlyFlushedRawBytes), RuntimeCounter::Unit::kBytes));
   if (earlyFlushedRawBytes > 0) {
     RECORD_METRIC_VALUE(
         kMetricFileWriterEarlyFlushedRawBytes, earlyFlushedRawBytes);

velox/connectors/hive/HiveDataSource.cpp

@@ -429,94 +429,104 @@ HiveDataSource::getRuntimeStats() {
   res.insert(
       {std::string(Connector::kIoWaitWallNanos),
        RuntimeMetric(
-           ioStatistics_->queryThreadIoLatencyUs().sum() * 1'000,
+           saturateCast(ioStatistics_->queryThreadIoLatencyUs().sum() * 1'000),
            ioStatistics_->queryThreadIoLatencyUs().count(),
-           ioStatistics_->queryThreadIoLatencyUs().min() * 1'000,
-           ioStatistics_->queryThreadIoLatencyUs().max() * 1'000,
+           saturateCast(ioStatistics_->queryThreadIoLatencyUs().min() * 1'000),
+           saturateCast(ioStatistics_->queryThreadIoLatencyUs().max() * 1'000),
            RuntimeCounter::Unit::kNanos)});
   // Breakdown of ioWaitWallNanos by I/O type
   if (ioStatistics_->storageReadLatencyUs().count() > 0) {
     res.insert(
         {std::string(Connector::kStorageReadWallNanos),
          RuntimeMetric(
-             ioStatistics_->storageReadLatencyUs().sum() * 1'000,
+             saturateCast(ioStatistics_->storageReadLatencyUs().sum() * 1'000),
              ioStatistics_->storageReadLatencyUs().count(),
-             ioStatistics_->storageReadLatencyUs().min() * 1'000,
-             ioStatistics_->storageReadLatencyUs().max() * 1'000,
+             saturateCast(ioStatistics_->storageReadLatencyUs().min() * 1'000),
+             saturateCast(ioStatistics_->storageReadLatencyUs().max() * 1'000),
              RuntimeCounter::Unit::kNanos)});
   }
   if (ioStatistics_->ssdCacheReadLatencyUs().count() > 0) {
     res.insert(
         {std::string(Connector::kSsdCacheReadWallNanos),
          RuntimeMetric(
-             ioStatistics_->ssdCacheReadLatencyUs().sum() * 1'000,
+             saturateCast(ioStatistics_->ssdCacheReadLatencyUs().sum() * 1'000),
              ioStatistics_->ssdCacheReadLatencyUs().count(),
-             ioStatistics_->ssdCacheReadLatencyUs().min() * 1'000,
-             ioStatistics_->ssdCacheReadLatencyUs().max() * 1'000,
+             saturateCast(ioStatistics_->ssdCacheReadLatencyUs().min() * 1'000),
+             saturateCast(ioStatistics_->ssdCacheReadLatencyUs().max() * 1'000),
              RuntimeCounter::Unit::kNanos)});
   }
   if (ioStatistics_->cacheWaitLatencyUs().count() > 0) {
     res.insert(
         {std::string(Connector::kCacheWaitWallNanos),
          RuntimeMetric(
-             ioStatistics_->cacheWaitLatencyUs().sum() * 1'000,
+             saturateCast(ioStatistics_->cacheWaitLatencyUs().sum() * 1'000),
              ioStatistics_->cacheWaitLatencyUs().count(),
-             ioStatistics_->cacheWaitLatencyUs().min() * 1'000,
-             ioStatistics_->cacheWaitLatencyUs().max() * 1'000,
+             saturateCast(ioStatistics_->cacheWaitLatencyUs().min() * 1'000),
+             saturateCast(ioStatistics_->cacheWaitLatencyUs().max() * 1'000),
              RuntimeCounter::Unit::kNanos)});
   }
   if (ioStatistics_->coalescedSsdLoadLatencyUs().count() > 0) {
     res.insert(
         {std::string(Connector::kCoalescedSsdLoadWallNanos),
          RuntimeMetric(
-             ioStatistics_->coalescedSsdLoadLatencyUs().sum() * 1'000,
+             saturateCast(
+                 ioStatistics_->coalescedSsdLoadLatencyUs().sum() * 1'000),
              ioStatistics_->coalescedSsdLoadLatencyUs().count(),
-             ioStatistics_->coalescedSsdLoadLatencyUs().min() * 1'000,
-             ioStatistics_->coalescedSsdLoadLatencyUs().max() * 1'000,
+             saturateCast(
+                 ioStatistics_->coalescedSsdLoadLatencyUs().min() * 1'000),
+             saturateCast(
+                 ioStatistics_->coalescedSsdLoadLatencyUs().max() * 1'000),
              RuntimeCounter::Unit::kNanos)});
   }
   if (ioStatistics_->coalescedStorageLoadLatencyUs().count() > 0) {
     res.insert(
         {std::string(Connector::kCoalescedStorageLoadWallNanos),
          RuntimeMetric(
-             ioStatistics_->coalescedStorageLoadLatencyUs().sum() * 1'000,
+             saturateCast(
+                 ioStatistics_->coalescedStorageLoadLatencyUs().sum() * 1'000),
              ioStatistics_->coalescedStorageLoadLatencyUs().count(),
-             ioStatistics_->coalescedStorageLoadLatencyUs().min() * 1'000,
-             ioStatistics_->coalescedStorageLoadLatencyUs().max() * 1'000,
+             saturateCast(
+                 ioStatistics_->coalescedStorageLoadLatencyUs().min() * 1'000),
+             saturateCast(
+                 ioStatistics_->coalescedStorageLoadLatencyUs().max() * 1'000),
              RuntimeCounter::Unit::kNanos)});
   }
   res.insert(
       {{std::string(kNumPrefetch),
         RuntimeMetric(ioStatistics_->prefetch().count())},
        {std::string(kPrefetchBytes),
         RuntimeMetric(
-            ioStatistics_->prefetch().sum(),
+            saturateCast(ioStatistics_->prefetch().sum()),
             ioStatistics_->prefetch().count(),
-            ioStatistics_->prefetch().min(),
-            ioStatistics_->prefetch().max(),
+            saturateCast(ioStatistics_->prefetch().min()),
+            saturateCast(ioStatistics_->prefetch().max()),
             RuntimeCounter::Unit::kBytes)},
        {std::string(kTotalScanTime),
         RuntimeMetric(
-            ioStatistics_->totalScanTime(), RuntimeCounter::Unit::kNanos)},
+            saturateCast(ioStatistics_->totalScanTime()),
+            RuntimeCounter::Unit::kNanos)},
        {std::string(Connector::kTotalRemainingFilterTime),
         RuntimeMetric(
-            totalRemainingFilterTime_.load(std::memory_order_relaxed),
+            saturateCast(
+                totalRemainingFilterTime_.load(std::memory_order_relaxed)),
             RuntimeCounter::Unit::kNanos)},
        {Connector::kTotalRemainingFilterCpuTime,
         RuntimeMetric(
-            totalRemainingFilterCpuTime_.load(std::memory_order_relaxed),
+            saturateCast(
+                totalRemainingFilterCpuTime_.load(std::memory_order_relaxed)),
             RuntimeCounter::Unit::kNanos)},
        {std::string(kOverreadBytes),
         RuntimeMetric(
-            ioStatistics_->rawOverreadBytes(), RuntimeCounter::Unit::kBytes)}});
+            saturateCast(ioStatistics_->rawOverreadBytes()),
+            RuntimeCounter::Unit::kBytes)}});
   if (ioStatistics_->read().count() > 0) {
     res.insert(
         {std::string(kStorageReadBytes),
          RuntimeMetric(
-             ioStatistics_->read().sum(),
+             saturateCast(ioStatistics_->read().sum()),
              ioStatistics_->read().count(),
-             ioStatistics_->read().min(),
-             ioStatistics_->read().max(),
+             saturateCast(ioStatistics_->read().min()),
+             saturateCast(ioStatistics_->read().max()),
              RuntimeCounter::Unit::kBytes)});
   }
   if (ioStatistics_->ssdRead().count() > 0) {
@@ -526,10 +536,10 @@ HiveDataSource::getRuntimeStats() {
     res.insert(
         {std::string(kLocalReadBytes),
          RuntimeMetric(
-             ioStatistics_->ssdRead().sum(),
+             saturateCast(ioStatistics_->ssdRead().sum()),
              ioStatistics_->ssdRead().count(),
-             ioStatistics_->ssdRead().min(),
-             ioStatistics_->ssdRead().max(),
+             saturateCast(ioStatistics_->ssdRead().min()),
+             saturateCast(ioStatistics_->ssdRead().max()),
              RuntimeCounter::Unit::kBytes)});
   }
   if (ioStatistics_->ramHit().count() > 0) {
@@ -539,10 +549,10 @@ HiveDataSource::getRuntimeStats() {
     res.insert(
         {std::string(kRamReadBytes),
          RuntimeMetric(
-             ioStatistics_->ramHit().sum(),
+             saturateCast(ioStatistics_->ramHit().sum()),
              ioStatistics_->ramHit().count(),
-             ioStatistics_->ramHit().min(),
-             ioStatistics_->ramHit().max(),
+             saturateCast(ioStatistics_->ramHit().min()),
+             saturateCast(ioStatistics_->ramHit().max()),
              RuntimeCounter::Unit::kBytes)});
   }
   if (numBucketConversion_ > 0) {

velox/dwio/common/OnDemandUnitLoader.cpp

@@ -83,10 +83,7 @@ class OnDemandUnitLoader : public UnitLoader {
     stats.addCounter(
         "unitLoadNanos",
         RuntimeCounter(
-            unitLoadNanos_ > std::numeric_limits<int64_t>::max()
-                ? std::numeric_limits<int64_t>::max()
-                : unitLoadNanos_,
-            RuntimeCounter::Unit::kNanos));
+            saturateCast(unitLoadNanos_), RuntimeCounter::Unit::kNanos));
     return stats;
   }
 

velox/dwio/common/ParallelUnitLoader.cpp

@@ -129,9 +129,7 @@ class ParallelUnitLoader : public UnitLoader {
     stats.addCounter(
         "waitForUnitReadyNanos",
         RuntimeCounter(
-            waitForUnitReadyNanos_ > std::numeric_limits<int64_t>::max()
-                ? std::numeric_limits<int64_t>::max()
-                : waitForUnitReadyNanos_,
+            saturateCast(waitForUnitReadyNanos_),
             RuntimeCounter::Unit::kNanos));
     return stats;
   }

velox/dwio/common/Statistics.h

@@ -634,10 +634,10 @@ struct ColumnMetricsSet {
                 nodeId,
                 TypeKindName::toName(stats->typeKind)),
             RuntimeMetric{
-                static_cast<int64_t>(decompressCounter.sum()),
-                static_cast<int64_t>(decompressCounter.count()),
-                static_cast<int64_t>(decompressCounter.min()),
-                static_cast<int64_t>(decompressCounter.max()),
+                saturateCast(decompressCounter.sum()),
+                decompressCounter.count(),
+                saturateCast(decompressCounter.min()),
+                saturateCast(decompressCounter.max()),
                 RuntimeCounter::Unit::kNanos});
       }
       // Export decode timing.
@@ -649,10 +649,10 @@ struct ColumnMetricsSet {
                 nodeId,
                 TypeKindName::toName(stats->typeKind)),
             RuntimeMetric{
-                static_cast<int64_t>(decodeCounter.sum()),
-                static_cast<int64_t>(decodeCounter.count()),
-                static_cast<int64_t>(decodeCounter.min()),
-                static_cast<int64_t>(decodeCounter.max()),
+                saturateCast(decodeCounter.sum()),
+                decodeCounter.count(),
+                saturateCast(decodeCounter.min()),
+                saturateCast(decodeCounter.max()),
                 RuntimeCounter::Unit::kNanos});
       }
     }

velox/dwio/dwrf/test/TestReadFile.h

@@ -64,8 +64,7 @@ class TestReadFile : public velox::ReadFile {
     if (context.ioStats) {
       context.ioStats->addCounter(
           "read",
-          RuntimeCounter(
-              static_cast<int64_t>(res), RuntimeCounter::Unit::kBytes));
+          RuntimeCounter(saturateCast(res), RuntimeCounter::Unit::kBytes));
     }
     ++numIos_;
     return res;

velox/exec/HashAggregation.cpp

@@ -285,7 +285,7 @@ void HashAggregation::resetPartialOutputIfNeed() {
         std::string(HashAggregation::kFlushTimes), RuntimeCounter(1));
     lockedStats->addRuntimeStat(
         std::string(HashAggregation::kPartialAggregationPct),
-        RuntimeCounter(static_cast<int64_t>(aggregationPct)));
+        RuntimeCounter(saturateCast(aggregationPct)));
   }
   groupingSet_->resetTable(/*freeTable=*/false);
   partialFull_ = false;