Add request-data-sizes-max-wait-sec to session properties (#24918)

Summary: As title

Differential Revision: D73019179

Author: HeidiHan0000

presto-docs/src/main/sphinx/presto_cpp/properties-session.rst

@@ -117,7 +117,7 @@ This property can only be enabled with native execution.
 * **Default value:** ``false``
 
 Temporary flag to control whether selective Nimble reader should be used in this
-query or not.  
+query or not.
 
 ``native_join_spill_enabled``
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -197,7 +197,7 @@ Native Execution only. Enable simplified path in expression evaluation.
 * **Type:** ``integer``
 * **Default value:** ``100000``
 
-Native Execution only. The `reduce <https://prestodb.io/docs/current/functions/array.html#reduce-array-T-initialState-S-inputFunction-S-T-S-outputFunction-S-R-R>`_ 
+Native Execution only. The `reduce <https://prestodb.io/docs/current/functions/array.html#reduce-array-T-initialState-S-inputFunction-S-T-S-outputFunction-S-R-R>`_
 function will throw an error if it encounters an array of size greater than this value.
 
 ``native_expression_max_compiled_regexes``
@@ -459,6 +459,14 @@ while a lower value limits the amount of memory used.
 * **Type:** ``integer``
 * **Default value:** ``0``
 
-In streaming aggregation, wait until there are enough output rows 
-to produce a batch of the size specified by this property. If set to ``0``, then 
-``Operator::outputBatchRows`` is used as the minimum number of output batch rows.
+In streaming aggregation, wait until there are enough output rows
+to produce a batch of the size specified by this property. If set to ``0``, then
+``Operator::outputBatchRows`` is used as the minimum number of output batch rows.
+
+``native_request_data_sizes_max_wait_sec``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``integer``
+* **Default value:** ``10``
+
+Maximum wait time for exchange long poll requests in seconds.

presto-main-base/src/main/java/com/facebook/presto/sessionpropertyproviders/NativeWorkerSessionPropertyProvider.java

@@ -77,6 +77,7 @@ public class NativeWorkerSessionPropertyProvider
     public static final String NATIVE_TABLE_SCAN_SCALED_PROCESSING_ENABLED = "native_table_scan_scaled_processing_enabled";
     public static final String NATIVE_TABLE_SCAN_SCALE_UP_MEMORY_USAGE_RATIO = "native_table_scan_scale_up_memory_usage_ratio";
     public static final String NATIVE_STREAMING_AGGREGATION_MIN_OUTPUT_BATCH_ROWS = "native_streaming_aggregation_min_output_batch_rows";
+    public static final String NATIVE_REQUEST_DATA_SIZES_MAX_WAIT_SEC = "native_request_data_sizes_max_wait_sec";
     private final List<PropertyMetadata<?>> sessionProperties;
 
     @Inject
@@ -344,6 +345,11 @@ public NativeWorkerSessionPropertyProvider(FeaturesConfig featuresConfig)
                                 "to produce a batch of size specified by this. If set to 0, then " +
                                 "Operator::outputBatchRows will be used as the min output batch rows.",
                         0,
+                        !nativeExecution),
+                integerProperty(
+                        NATIVE_REQUEST_DATA_SIZES_MAX_WAIT_SEC,
+                        "Maximum wait time for exchange long poll requests in seconds.",
+                        10,
                         !nativeExecution));
     }
 

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

@@ -481,6 +481,14 @@ SessionProperties::SessionProperties() {
       false,
       QueryConfig::kStreamingAggregationMinOutputBatchRows,
       std::to_string(c.streamingAggregationMinOutputBatchRows()));
+
+  addSessionProperty(
+      kRequestDataSizesMaxWaitSec,
+      "Maximum wait time for exchange long poll requests in seconds.",
+      INTEGER(),
+      10,
+      QueryConfig::kRequestDataSizesMaxWaitSec,
+      std::to_string(c.requestDataSizesMaxWaitSec()));
 }
 
 const std::unordered_map<std::string, std::shared_ptr<SessionProperty>>&

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

@@ -309,6 +309,10 @@ class SessionProperties {
   static constexpr const char* kStreamingAggregationMinOutputBatchRows =
       "native_streaming_aggregation_min_output_batch_rows";
 
+  /// Maximum wait time for exchange long poll requests in seconds.
+  static constexpr const char* kRequestDataSizesMaxWaitSec = 
+      "native_request_data_sizes_max_wait_sec";
+
   SessionProperties();
 
   const std::unordered_map<std::string, std::shared_ptr<SessionProperty>>&

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

@@ -34,7 +34,8 @@ TEST_F(SessionPropertiesTest, validateMapping) {
       SessionProperties::kScaleWriterMinProcessedBytesRebalanceThreshold,
       SessionProperties::kTableScanScaledProcessingEnabled,
       SessionProperties::kTableScanScaleUpMemoryUsageRatio,
-      SessionProperties::kStreamingAggregationMinOutputBatchRows};
+      SessionProperties::kStreamingAggregationMinOutputBatchRows,
+      SessionProperties::kRequestDataSizesMaxWaitSec};
   const std::vector<std::string> veloxConfigNames = {
       core::QueryConfig::kAdjustTimestampToTimezone,
       core::QueryConfig::kDriverCpuTimeSliceLimitMs,
@@ -46,7 +47,8 @@ TEST_F(SessionPropertiesTest, validateMapping) {
       core::QueryConfig::kScaleWriterMinProcessedBytesRebalanceThreshold,
       core::QueryConfig::kTableScanScaledProcessingEnabled,
       core::QueryConfig::kTableScanScaleUpMemoryUsageRatio,
-      core::QueryConfig::kStreamingAggregationMinOutputBatchRows};
+      core::QueryConfig::kStreamingAggregationMinOutputBatchRows,
+      core::QueryConfig::kRequestDataSizesMaxWaitSec};
   auto sessionProperties = SessionProperties().getSessionProperties();
   const auto len = names.size();
   for (auto i = 0; i < len; i++) {