feat(native): Separate exchange get-data-size vs get-data counters for better tracking

Exchange has two phase protocol
1. First request is a get-data-size request, it uses long-poll protocol and wait until
exchange source has data or timeout
2. Second request is get-data request, it will be be non-blocking call to transmit data

Differentiating those two would be helpful to differentiate the time waiting on server
and time for transmitting data

Author: kewang1024

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

@@ -246,7 +246,8 @@ void PrestoExchangeSource::handleDataResponse(
       } else if (response->hasError()) {
         processDataError(httpRequestPath, maxBytes, maxWait, response->error());
       } else {
-        processDataResponse(std::move(response));
+        const bool isGetDataSizeRequest = (maxBytes == 0);
+        processDataResponse(std::move(response), isGetDataSizeRequest);
       }
     } catch (const std::exception& e) {
       processDataError(httpRequestPath, maxBytes, maxWait, e.what());
@@ -255,11 +256,20 @@ void PrestoExchangeSource::handleDataResponse(
 }
 
 void PrestoExchangeSource::processDataResponse(
-    std::unique_ptr<http::HttpResponse> response) {
-  RECORD_HISTOGRAM_METRIC_VALUE(
-      kCounterExchangeRequestDuration, dataRequestRetryState_.durationMs());
-  RECORD_HISTOGRAM_METRIC_VALUE(
-      kCounterExchangeRequestNumTries, dataRequestRetryState_.numTries());
+    std::unique_ptr<http::HttpResponse> response, bool isGetDataSizeRequest) {
+  if (isGetDataSizeRequest) {
+    RECORD_HISTOGRAM_METRIC_VALUE(
+        kCounterExchangeGetDataSizeDuration,
+        dataRequestRetryState_.durationMs());
+    RECORD_HISTOGRAM_METRIC_VALUE(
+        kCounterExchangeGetDataSizeNumTries,
+        dataRequestRetryState_.numTries());
+  } else {
+    RECORD_HISTOGRAM_METRIC_VALUE(
+        kCounterExchangeRequestDuration, dataRequestRetryState_.durationMs());
+    RECORD_HISTOGRAM_METRIC_VALUE(
+        kCounterExchangeRequestNumTries, dataRequestRetryState_.numTries());
+  }
   if (closed_.load()) {
     // If PrestoExchangeSource is already closed, just free all buffers
     // allocated without doing any processing. This can happen when a super slow
@@ -332,6 +342,12 @@ void PrestoExchangeSource::processDataResponse(
     }
     PrestoExchangeSource::updateMemoryUsage(totalBytes);
 
+    // Record page size counter when not a get-data-size request
+    if (!isGetDataSizeRequest) {
+      RECORD_HISTOGRAM_METRIC_VALUE(
+          kCounterExchangeRequestPageSize, totalBytes);
+    }
+
     if (enableBufferCopy_) {
       page = std::make_unique<exec::SerializedPage>(
           std::move(singleChain), [pool = pool_](folly::IOBuf& iobuf) {

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

@@ -200,7 +200,7 @@ class PrestoExchangeSource : public velox::exec::ExchangeSource {
   // response without an end marker. Sends delete-results if received an end
   // marker. The sequence of operations is: add data or end marker to the
   // queue; complete the future, send ack or delete-results.
-  void processDataResponse(std::unique_ptr<http::HttpResponse> response);
+  void processDataResponse(std::unique_ptr<http::HttpResponse> response, bool getDataSize);
 
   // If 'retry' is true, then retry the http request failure until reaches the
   // retry limit, otherwise just set exchange source error without retry. As

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

@@ -117,28 +117,49 @@ void registerPrestoMetrics() {
       99,
       100);
 
-  // Tracks exchange request duration in range of [0, 300s] with
-  // 300 buckets and reports P50, P90, P99, and P100.
+  // Tracks exchange request duration in range of [0, 10s] with
+  // 500 buckets and reports P50, P90, P99, and P100.
   DEFINE_HISTOGRAM_METRIC(
       kCounterExchangeRequestDuration,
-      1'000,
+      20, // 20ms bucket size
       0,
-      300'000,
+      10'000, // 10s max
       50,
       90,
       99,
       100);
-  // Tracks exchange request num of retris in range of [0, 20] with
+  // Tracks exchange request num of tries in range of [0, 20] with
   // 20 buckets and reports P50, P90, P99, and P100.
   DEFINE_HISTOGRAM_METRIC(
-      kCounterExchangeRequestNumTries,
-      1,
+      kCounterExchangeRequestNumTries, 1, 0, 20, 50, 90, 99, 100);
+  // Tracks exchange request page size in range of [0, 20MB] with
+  // 20K buckets and reports P50, P90, P99, and P100.
+  DEFINE_HISTOGRAM_METRIC(
+      kCounterExchangeRequestPageSize,
+      10 * 1024, // 10KB bucket size
       0,
-      20,
+      20 * 1024 * 1024, // 20MB max
       50,
       90,
       99,
       100);
+
+  // Tracks exchange get-data-size request duration in range of [0, 300s] with
+  // 300 buckets and reports P50, P90, P99, and P100.
+  DEFINE_HISTOGRAM_METRIC(
+      kCounterExchangeGetDataSizeDuration,
+      1'000,
+      0,
+      300'000,
+      50,
+      90,
+      99,
+      100);
+  // Tracks exchange get-data-size request num of tries in range of [0, 20] with
+  // 20 buckets and reports P50, P90, P99, and P100.
+  DEFINE_HISTOGRAM_METRIC(
+      kCounterExchangeGetDataSizeNumTries, 1, 0, 20, 50, 90, 99, 100);
+
   DEFINE_METRIC(kCounterMemoryPushbackCount, facebook::velox::StatType::COUNT);
   DEFINE_HISTOGRAM_METRIC(
       kCounterMemoryPushbackLatencyMs, 10'000, 0, 100'000, 50, 90, 99, 100);

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

@@ -51,6 +51,14 @@ constexpr folly::StringPiece kCounterExchangeRequestDuration{
     "presto_cpp.exchange.request.duration"};
 constexpr folly::StringPiece kCounterExchangeRequestNumTries{
     "presto_cpp.exchange.request.num_tries"};
+constexpr folly::StringPiece kCounterExchangeRequestPageSize{
+  "presto_cpp.exchange.request.page_size"};
+
+
+constexpr folly::StringPiece kCounterExchangeGetDataSizeDuration{
+  "presto_cpp.exchange.get_data_size.duration"};
+constexpr folly::StringPiece kCounterExchangeGetDataSizeNumTries{
+  "presto_cpp.exchange.get_data_size.num_tries"};
 
 constexpr folly::StringPiece kCounterNumQueryContexts{
     "presto_cpp.num_query_contexts"};