[native] Propagate table writer concurrency settings

Author: arhimondr

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

@@ -139,6 +139,8 @@ SystemConfig::SystemConfig() {
           BOOL_PROP(kHttpServerBindToNodeInternalAddressOnlyEnabled, false),
           NONE_PROP(kDiscoveryUri),
           NUM_PROP(kMaxDriversPerTask, 16),
+          NONE_PROP(kTaskWriterCount),
+          NONE_PROP(kTaskPartitionedWriterCount),
           NUM_PROP(kConcurrentLifespansPerTask, 1),
           STR_PROP(kTaskMaxPartialAggregationMemory, "16MB"),
           NUM_PROP(kHttpServerNumIoThreadsHwMultiplier, 1.0),
@@ -341,6 +343,14 @@ int32_t SystemConfig::maxDriversPerTask() const {
   return optionalProperty<int32_t>(kMaxDriversPerTask).value();
 }
 
+folly::Optional<int32_t> SystemConfig::taskWriterCount() const {
+  return optionalProperty<int32_t>(kTaskWriterCount);
+}
+
+folly::Optional<int32_t> SystemConfig::taskPartitionedWriterCount() const {
+  return optionalProperty<int32_t>(kTaskPartitionedWriterCount);
+}
+
 int32_t SystemConfig::concurrentLifespansPerTask() const {
   return optionalProperty<int32_t>(kConcurrentLifespansPerTask).value();
 }
@@ -876,7 +886,7 @@ void BaseVeloxQueryConfig::updateLoadedValues(
   auto systemConfig = SystemConfig::instance();
 
   using namespace velox::core;
-  const std::unordered_map<std::string, std::string> updatedValues{
+  std::unordered_map<std::string, std::string> updatedValues{
       {QueryConfig::kPrestoArrayAggIgnoreNulls,
        bool2String(systemConfig->useLegacyArrayAgg())},
       {QueryConfig::kMaxOutputBufferSize,
@@ -890,6 +900,17 @@ void BaseVeloxQueryConfig::updateLoadedValues(
            SystemConfig::kTaskMaxPartialAggregationMemory)},
   };
 
+  auto taskWriterCount = systemConfig->taskWriterCount();
+  if (taskWriterCount.has_value()) {
+    updatedValues[QueryConfig::kTaskWriterCount] =
+        std::to_string(taskWriterCount.value());
+  }
+  auto taskPartitionedWriterCount = systemConfig->taskPartitionedWriterCount();
+  if (taskPartitionedWriterCount.has_value()) {
+    updatedValues[QueryConfig::kTaskPartitionedWriterCount] =
+        std::to_string(taskPartitionedWriterCount.value());
+  }
+
   std::stringstream updated;
   for (const auto& pair : updatedValues) {
     updated << "  " << pair.first << "=" << pair.second << "\n";

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

@@ -179,6 +179,9 @@ class SystemConfig : public ConfigBase {
   static constexpr std::string_view kDiscoveryUri{"discovery.uri"};
   static constexpr std::string_view kMaxDriversPerTask{
       "task.max-drivers-per-task"};
+  static constexpr std::string_view kTaskWriterCount{"task.writer-count"};
+  static constexpr std::string_view kTaskPartitionedWriterCount{
+      "task.partitioned-writer-count"};
   static constexpr std::string_view kConcurrentLifespansPerTask{
       "task.concurrent-lifespans-per-task"};
   static constexpr std::string_view kTaskMaxPartialAggregationMemory{
@@ -695,6 +698,10 @@ class SystemConfig : public ConfigBase {
 
   int32_t maxDriversPerTask() const;
 
+  folly::Optional<int32_t> taskWriterCount() const;
+
+  folly::Optional<int32_t> taskPartitionedWriterCount() const;
+
   int32_t concurrentLifespansPerTask() const;
 
   double httpServerNumIoThreadsHwMultiplier() const;