add local exchange partition buffer size system config

Author: kagamiori

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

@@ -226,6 +226,14 @@ avoid exceeding memory limits for the query.
 When ``spill_enabled`` is ``true``, this determines whether Presto will try spilling memory to disk for order by to
 avoid exceeding memory limits for the query.
 
+``local-exchange.max-partition-buffer-size``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``integer``
+* **Default value:** ``65536`` (64KB)
+
+  Specifies the maximum size in bytes to accumulate for a single partition of a local exchange before flushing.
+
 
 ``shared-arbitrator.reserved-capacity``
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

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

@@ -59,7 +59,9 @@ void updateFromSystemConfigs(
           {core::QueryConfig::kRequestDataSizesMaxWaitSec,
           std::string(SystemConfig::kRequestDataSizesMaxWaitSec)},
           {core::QueryConfig::kMaxSplitPreloadPerDriver,
-          std::string(SystemConfig::kDriverMaxSplitPreload)}};
+          std::string(SystemConfig::kDriverMaxSplitPreload)},
+          {core::QueryConfig::kMaxLocalExchangePartitionBufferSize,
+          std::string(SystemConfig::kMaxLocalExchangePartitionBufferSize)}};
   for (const auto& configNameEntry : sessionSystemConfigMapping) {
     const auto& sessionName = configNameEntry.first;
     const auto& systemConfigName = configNameEntry.second;

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

@@ -264,6 +264,7 @@ SystemConfig::SystemConfig() {
           STR_PROP(kPluginDir, ""),
           NUM_PROP(kExchangeIoEvbViolationThresholdMs, 1000),
           NUM_PROP(kHttpSrvIoEvbViolationThresholdMs, 1000),
+          NUM_PROP(kMaxLocalExchangePartitionBufferSize, 65536),
       };
 }
 
@@ -914,6 +915,10 @@ int32_t SystemConfig::httpSrvIoEvbViolationThresholdMs() const {
       .value();
 }
 
+uint64_t SystemConfig::maxLocalExchangePartitionBufferSize() const {
+  return optionalProperty<uint64_t>(kMaxLocalExchangePartitionBufferSize).value();
+}
+
 NodeConfig::NodeConfig() {
   registeredProps_ =
       std::unordered_map<std::string, folly::Optional<std::string>>{

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

@@ -760,6 +760,9 @@ class SystemConfig : public ConfigBase {
   static constexpr std::string_view kHttpSrvIoEvbViolationThresholdMs{
       "http-server.io-evb-violation-threshold-ms"};
 
+  static constexpr std::string_view kMaxLocalExchangePartitionBufferSize{
+      "local-exchange.max-partition-buffer-size"};
+
   SystemConfig();
 
   virtual ~SystemConfig() = default;
@@ -1042,6 +1045,8 @@ class SystemConfig : public ConfigBase {
   int32_t exchangeIoEvbViolationThresholdMs() const;
 
   int32_t httpSrvIoEvbViolationThresholdMs() const;
+
+  uint64_t maxLocalExchangePartitionBufferSize() const;
 };
 
 /// Provides access to node properties defined in node.properties file.

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

@@ -129,6 +129,8 @@ TEST_F(QueryContextManagerTest, defaultSessionProperties) {
       queryConfig.requestDataSizesMaxWaitSec(), defaultQC->requestDataSizesMaxWaitSec());
   EXPECT_EQ(
       queryConfig.maxSplitPreloadPerDriver(), defaultQC->maxSplitPreloadPerDriver());
+  EXPECT_EQ(
+      queryConfig.maxLocalExchangePartitionBufferSize(), defaultQC->maxLocalExchangePartitionBufferSize());
 }
 
 TEST_F(QueryContextManagerTest, overridingSessionProperties) {