data query uses different local compute tasks param (24-3) (#8633)

Author: gridnevvvit

ydb/core/kqp/executer_actor/kqp_planner.cpp

@@ -258,8 +258,14 @@ std::unique_ptr<IEventHandle> TKqpPlanner::AssignTasksToNodes() {
 
     auto placingOptions = ResourceManager_->GetPlacingOptions();
 
+    ui64 nonParallelLimit = placingOptions.MaxNonParallelTasksExecutionLimit;
+    if (MayRunTasksLocally) {
+        // not applied to column shards and external sources
+        nonParallelLimit = placingOptions.MaxNonParallelDataQueryTasksLimit;
+    }
+
     bool singleNodeExecutionMakeSence = (
-        ResourceEstimations.size() <= placingOptions.MaxNonParallelTasksExecutionLimit ||
+        ResourceEstimations.size() <= nonParallelLimit ||
         // all readers are located on the one node.
         TasksPerNode.size() == 1
     );

ydb/core/kqp/rm_service/kqp_rm_service.cpp

@@ -145,6 +145,7 @@ class TKqpResourceManager : public IKqpResourceManager {
     TPlannerPlacingOptions GetPlacingOptions() override {
         return TPlannerPlacingOptions{
             .MaxNonParallelTasksExecutionLimit = MaxNonParallelTasksExecutionLimit.load(),
+            .MaxNonParallelDataQueryTasksLimit = MaxNonParallelDataQueryTasksLimit.load(),
             .MaxNonParallelTopStageExecutionLimit = MaxNonParallelTopStageExecutionLimit.load(),
             .PreferLocalDatacenterExecution = PreferLocalDatacenterExecution.load(),
         };
@@ -425,6 +426,7 @@ class TKqpResourceManager : public IKqpResourceManager {
         MaxNonParallelTopStageExecutionLimit.store(config.GetMaxNonParallelTopStageExecutionLimit());
         MaxNonParallelTasksExecutionLimit.store(config.GetMaxNonParallelTasksExecutionLimit());
         PreferLocalDatacenterExecution.store(config.GetPreferLocalDatacenterExecution());
+        MaxNonParallelDataQueryTasksLimit.store(config.GetMaxNonParallelDataQueryTasksLimit());
     }
 
     ui32 GetNodeId() override {
@@ -474,6 +476,7 @@ class TKqpResourceManager : public IKqpResourceManager {
     std::atomic<ui64> MaxNonParallelTopStageExecutionLimit = 1;
     std::atomic<ui64> MaxNonParallelTasksExecutionLimit = 8;
     std::atomic<bool> PreferLocalDatacenterExecution = true;
+    std::atomic<ui64> MaxNonParallelDataQueryTasksLimit = 1000;
 
     // current state
     std::atomic<ui64> LastResourceBrokerTaskId = 0;

ydb/core/kqp/rm_service/kqp_rm_service.h

@@ -204,6 +204,7 @@ struct TKqpLocalNodeResources {
 
 struct TPlannerPlacingOptions {
     ui64 MaxNonParallelTasksExecutionLimit = 8;
+    ui64 MaxNonParallelDataQueryTasksLimit = 1000;
     ui64 MaxNonParallelTopStageExecutionLimit = 1;
     bool PreferLocalDatacenterExecution = true;
 };

ydb/core/protos/table_service_config.proto

@@ -50,6 +50,7 @@ message TTableServiceConfig {
         optional uint64 MaxNonParallelTasksExecutionLimit = 25 [default = 8];
         optional uint64 MaxNonParallelTopStageExecutionLimit = 26 [default = 1];
         optional bool PreferLocalDatacenterExecution = 27 [ default = true ];
+        optional uint64 MaxNonParallelDataQueryTasksLimit = 28 [default = 1000];
     }
 
     message TSpillingServiceConfig {
@@ -301,6 +302,6 @@ message TTableServiceConfig {
     optional uint64 QueryReplayCacheUploadTTLSec = 62 [default = 36000];
 
     optional bool EnableQueryServiceSpilling = 63 [ default = true ];
-    
+
     optional bool EnableImplicitQueryParameterTypes = 66 [ default = false ];
 };