[Native] Add driver.max-split-preload system config

Author: kewang1024

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

@@ -38,6 +38,14 @@ Presto C++ workers.
 These Presto coordinator configuration properties are described here, in
 alphabetical order.
 
+``driver.max-split-preload``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+* **Type:** ``integer``
+* **Default value:** ``2``
+
+  Maximum number of splits to preload per driver.
+  Set to 0 to disable preloading.
+
 ``driver.cancel-tasks-with-stuck-operators-threshold-ms``
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 * **Type:** ``string``

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

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

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

@@ -152,6 +152,7 @@ SystemConfig::SystemConfig() {
           NONE_PROP(kTaskPartitionedWriterCount),
           NUM_PROP(kConcurrentLifespansPerTask, 1),
           STR_PROP(kTaskMaxPartialAggregationMemory, "16MB"),
+          NUM_PROP(kDriverMaxSplitPreload, 2),
           NUM_PROP(kHttpServerNumIoThreadsHwMultiplier, 1.0),
           NUM_PROP(kHttpServerNumCpuThreadsHwMultiplier, 1.0),
           NONE_PROP(kHttpServerHttpsPort),
@@ -402,6 +403,10 @@ int32_t SystemConfig::maxDriversPerTask() const {
   return optionalProperty<int32_t>(kMaxDriversPerTask).value();
 }
 
+int32_t SystemConfig::driverMaxSplitPreload() const {
+  return optionalProperty<int32_t>(kDriverMaxSplitPreload).value();
+}
+
 folly::Optional<int32_t> SystemConfig::taskWriterCount() const {
   return optionalProperty<int32_t>(kTaskWriterCount);
 }

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

@@ -230,6 +230,11 @@ class SystemConfig : public ConfigBase {
   static constexpr std::string_view kConnectorNumIoThreadsHwMultiplier{
       "connector.num-io-threads-hw-multiplier"};
 
+  /// Maximum number of splits to preload per driver.
+  /// Set to 0 to disable preloading.
+  static constexpr std::string_view kDriverMaxSplitPreload{
+      "driver.max-split-preload"};
+
   /// Floating point number used in calculating how many threads we would use
   /// for Driver CPU executor: hw_concurrency x multiplier. 4.0 is default.
   static constexpr std::string_view kDriverNumCpuThreadsHwMultiplier{
@@ -812,6 +817,8 @@ class SystemConfig : public ConfigBase {
 
   int32_t maxDriversPerTask() const;
 
+  int32_t driverMaxSplitPreload() const;
+
   folly::Optional<int32_t> taskWriterCount() const;
 
   folly::Optional<int32_t> taskPartitionedWriterCount() const;

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

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