[native] Add a metric to track expected reduction of memory and minor fixes

Summary:
- Add a metric `presto_cpp.memory_pushback_expected_reduction_bytes`
to track expected reduction in memory after a pushback attempt as
this can be different from actual reduction due to other threads
allocating simultaneously
- Fix registeration of a pushback metric
- Make periodicCb() API non-const to allow it to update internal
state if required

Author: bikramSingh91

presto-native-execution/presto_cpp/main/PeriodicMemoryChecker.cpp

@@ -211,6 +211,8 @@ void PeriodicMemoryChecker::pushbackMemory() {
       kCounterMemoryPushbackLatencyMs, latencyUs / 1000);
   const auto actualFreedBytes = std::max<int64_t>(
       0, static_cast<int64_t>(currentMemBytes) - systemUsedMemoryBytes());
+  RECORD_HISTOGRAM_METRIC_VALUE(
+      kCounterMemoryPushbackExpectedReductionBytes, freedBytes);
   RECORD_HISTOGRAM_METRIC_VALUE(
       kCounterMemoryPushbackReductionBytes, actualFreedBytes);
   LOG(INFO) << "Memory pushback shrunk " << velox::succinctBytes(freedBytes)

presto-native-execution/presto_cpp/main/PeriodicMemoryChecker.h

@@ -87,7 +87,7 @@ class PeriodicMemoryChecker {
 
   /// Callback function that is invoked by 'PeriodicMemoryChecker' periodically.
   /// Light operations such as stats reporting can be done in this call back.
-  virtual void periodicCb() const = 0;
+  virtual void periodicCb() = 0;
 
   /// Callback function that performs a heap dump. Returns true if dump is
   /// successful.

presto-native-execution/presto_cpp/main/common/Counters.cpp

@@ -103,7 +103,16 @@ void registerPrestoMetrics() {
   DEFINE_HISTOGRAM_METRIC(
       kCounterMemoryPushbackLatencyMs, 10'000, 0, 100'000, 50, 90, 99, 100);
   DEFINE_HISTOGRAM_METRIC(
-      kCounterMemoryPushbackLatencyMs,
+      kCounterMemoryPushbackReductionBytes,
+      100l * 1024 * 1024, // 100MB
+      0,
+      15l * 1024 * 1024 * 1024, // 15GB
+      50,
+      90,
+      99,
+      100);
+  DEFINE_HISTOGRAM_METRIC(
+      kCounterMemoryPushbackExpectedReductionBytes,
       100l * 1024 * 1024, // 100MB
       0,
       15l * 1024 * 1024 * 1024, // 15GB

presto-native-execution/presto_cpp/main/common/Counters.h

@@ -166,9 +166,16 @@ constexpr folly::StringPiece kCounterMemoryPushbackCount{
 /// reports P50, P90, P99, and P100.
 constexpr folly::StringPiece kCounterMemoryPushbackLatencyMs{
     "presto_cpp.memory_pushback_latency_ms"};
-/// Distribution of reduction in memory usage achieved by each memory pushback
-/// attempt. This is to gauge its effectiveness. In range of [0, 15GB] with 150
-/// buckets and reports P50, P90, P99, and P100.
+/// Distribution of actual reduction in memory usage achieved by each memory
+/// pushback attempt. This is to gauge its effectiveness. In range of [0, 15GB]
+/// with 150 buckets and reports P50, P90, P99, and P100.
 constexpr folly::StringPiece kCounterMemoryPushbackReductionBytes{
     "presto_cpp.memory_pushback_reduction_bytes"};
+/// Distribution of expected reduction in memory usage achieved by each memory
+/// pushback attempt. This is to gauge its effectiveness. In range of [0, 15GB]
+/// with 150 buckets and reports P50, P90, P99, and P100. The expected reduction
+/// can be different as other threads might have allocated memory in the
+/// meantime.
+constexpr folly::StringPiece kCounterMemoryPushbackExpectedReductionBytes{
+    "presto_cpp.memory_pushback_expected_reduction_bytes"};
 } // namespace facebook::presto

presto-native-execution/presto_cpp/main/tests/PeriodicMemoryCheckerTest.cpp

@@ -54,7 +54,7 @@ class PeriodicMemoryCheckerTest : public testing::Test {
       return mallocBytes_;
     }
 
-    void periodicCb() const override {
+    void periodicCb() override {
       if (periodicCb_) {
         periodicCb_();
       }