Return backward compatibility to YDB for spilling

Author: Darych

ydb/core/kqp/executer_actor/kqp_data_executer.cpp

@@ -2211,7 +2211,7 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
         const bool localComputeTasks = !((HasExternalSources || HasOlapTable || HasDatashardSourceScan) && DatashardTxs.size() == 0);
 
         Planner = CreateKqpPlanner(TasksGraph, TxId, SelfId(), GetSnapshot(),
-            Database, UserToken, Deadline.GetOrElse(TInstant::Zero()), Request.StatsMode, Nothing(),
+            Database, UserToken, Deadline.GetOrElse(TInstant::Zero()), Request.StatsMode, false, Nothing(),
             ExecuterSpan, std::move(ResourceSnapshot), ExecuterRetriesConfig, useDataQueryPool, localComputeTasks, Request.MkqlMemoryLimit, AsyncIoFactory, enableOptForTasks, GetUserRequestContext());
 
         auto err = Planner->PlanExecution();

ydb/core/kqp/executer_actor/kqp_planner.cpp

@@ -59,7 +59,7 @@ constexpr ui32 MAX_NON_PARALLEL_TASKS_EXECUTION_LIMIT = 4;
 
 TKqpPlanner::TKqpPlanner(TKqpTasksGraph& graph, ui64 txId, const TActorId& executer, const IKqpGateway::TKqpSnapshot& snapshot,
     const TString& database, const TIntrusiveConstPtr<NACLib::TUserToken>& userToken, TInstant deadline,
-    const Ydb::Table::QueryStatsCollection::Mode& statsMode,
+    const Ydb::Table::QueryStatsCollection::Mode& statsMode, bool withSpilling,
     const TMaybe<NKikimrKqp::TRlPath>& rlPath, NWilson::TSpan& executerSpan,
     TVector<NKikimrKqp::TKqpNodeResources>&& resourcesSnapshot,
     const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig,
@@ -72,6 +72,7 @@ TKqpPlanner::TKqpPlanner(TKqpTasksGraph& graph, ui64 txId, const TActorId& execu
     , UserToken(userToken)
     , Deadline(deadline)
     , StatsMode(statsMode)
+    , WithSpilling(withSpilling)
     , RlPath(rlPath)
     , ResourcesSnapshot(std::move(resourcesSnapshot))
     , ExecuterSpan(executerSpan)
@@ -186,6 +187,7 @@ std::unique_ptr<TEvKqpNode::TEvStartKqpTasksRequest> TKqpPlanner::SerializeReque
         request.MutableRuntimeSettings()->SetExecType(NYql::NDqProto::TComputeRuntimeSettings::DATA);
     } else {
         request.MutableRuntimeSettings()->SetExecType(NYql::NDqProto::TComputeRuntimeSettings::SCAN);
+        request.MutableRuntimeSettings()->SetUseSpilling(WithSpilling);
     }
 
     if (RlPath) {
@@ -332,6 +334,7 @@ void TKqpPlanner::ExecuteDataComputeTask(ui64 taskId, bool shareMailbox, bool op
     settings.ExtraMemoryAllocationPool = NRm::EKqpMemoryPool::DataQuery;
     settings.FailOnUndelivery = true;
     settings.StatsMode = GetDqStatsMode(StatsMode);
+    settings.UseSpilling = false;
 
     NYql::NDq::TComputeMemoryLimits limits;
     limits.ChannelBufferSize = 32_MB;  // Depends on NYql::NDq::TDqOutputChannelSettings::ChunkSizeLimit (now 48 MB) with a ratio of 1.5
@@ -515,14 +518,14 @@ ui32 TKqpPlanner::CalcSendMessageFlagsForNode(ui32 nodeId) {
 std::unique_ptr<TKqpPlanner> CreateKqpPlanner(TKqpTasksGraph& tasksGraph, ui64 txId, const TActorId& executer,
     const IKqpGateway::TKqpSnapshot& snapshot,
     const TString& database, const TIntrusiveConstPtr<NACLib::TUserToken>& userToken, TInstant deadline,
-    const Ydb::Table::QueryStatsCollection::Mode& statsMode,
+    const Ydb::Table::QueryStatsCollection::Mode& statsMode, bool withSpilling,
     const TMaybe<NKikimrKqp::TRlPath>& rlPath, NWilson::TSpan& executerSpan,
     TVector<NKikimrKqp::TKqpNodeResources>&& resourcesSnapshot, const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig,
     bool useDataQueryPool, bool localComputeTasks, ui64 mkqlMemoryLimit, NYql::NDq::IDqAsyncIoFactory::TPtr asyncIoFactory, bool doOptimization,
     const TIntrusivePtr<TUserRequestContext>& userRequestContext)
 {
     return std::make_unique<TKqpPlanner>(tasksGraph, txId, executer, snapshot,
-        database, userToken, deadline, statsMode, rlPath, executerSpan,
+        database, userToken, deadline, statsMode, withSpilling, rlPath, executerSpan,
         std::move(resourcesSnapshot), executerRetriesConfig, useDataQueryPool,
         localComputeTasks, mkqlMemoryLimit, asyncIoFactory, doOptimization, userRequestContext);
 }

ydb/core/kqp/executer_actor/kqp_planner.h

@@ -43,7 +43,7 @@ class TKqpPlanner {
 public:
     TKqpPlanner(TKqpTasksGraph& tasksGraph, ui64 txId, const TActorId& executer, const IKqpGateway::TKqpSnapshot& snapshot,
         const TString& database, const TIntrusiveConstPtr<NACLib::TUserToken>& userToken, TInstant deadline,
-        const Ydb::Table::QueryStatsCollection::Mode& statsMode,
+        const Ydb::Table::QueryStatsCollection::Mode& statsMode, bool withSpilling,
         const TMaybe<NKikimrKqp::TRlPath>& rlPath, NWilson::TSpan& ExecuterSpan,
         TVector<NKikimrKqp::TKqpNodeResources>&& resourcesSnapshot, const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig,
         bool useDataQueryPool, bool localComputeTasks, ui64 mkqlMemoryLimit, NYql::NDq::IDqAsyncIoFactory::TPtr asyncIoFactory, bool doOptimization,
@@ -84,6 +84,7 @@ class TKqpPlanner {
     const TIntrusiveConstPtr<NACLib::TUserToken> UserToken;
     const TInstant Deadline;
     const Ydb::Table::QueryStatsCollection::Mode StatsMode;
+    const bool WithSpilling;
     const TMaybe<NKikimrKqp::TRlPath> RlPath;
     THashSet<ui32> TrackingNodes;
     const TVector<NKikimrKqp::TKqpNodeResources> ResourcesSnapshot;
@@ -113,7 +114,7 @@ class TKqpPlanner {
 std::unique_ptr<TKqpPlanner> CreateKqpPlanner(TKqpTasksGraph& tasksGraph, ui64 txId, const TActorId& executer,
     const IKqpGateway::TKqpSnapshot& snapshot,
     const TString& database, const TIntrusiveConstPtr<NACLib::TUserToken>& userToken, TInstant deadline,
-    const Ydb::Table::QueryStatsCollection::Mode& statsMode,
+    const Ydb::Table::QueryStatsCollection::Mode& statsMode, bool withSpilling,
     const TMaybe<NKikimrKqp::TRlPath>& rlPath, NWilson::TSpan& executerSpan,
     TVector<NKikimrKqp::TKqpNodeResources>&& resourcesSnapshot,
     const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& ExecuterRetriesConfig,

ydb/core/kqp/executer_actor/kqp_scan_executer.cpp

@@ -332,7 +332,7 @@ class TKqpScanExecuter : public TKqpExecuterBase<TKqpScanExecuter, EExecType::Sc
     void ExecuteScanTx() {
 
         Planner = CreateKqpPlanner(TasksGraph, TxId, SelfId(), GetSnapshot(),
-            Database, UserToken, Deadline.GetOrElse(TInstant::Zero()), Request.StatsMode,
+            Database, UserToken, Deadline.GetOrElse(TInstant::Zero()), Request.StatsMode, AppData()->EnableKqpSpilling,
             Request.RlPath, ExecuterSpan, std::move(ResourcesSnapshot), ExecuterRetriesConfig, /* useDataQueryPool */ false, /* localComputeTasks */ false,
             Request.MkqlMemoryLimit, nullptr, false, GetUserRequestContext());
 

ydb/core/kqp/node_service/kqp_node_service.cpp

@@ -410,6 +410,7 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
         runtimeSettingsBase.FailOnUndelivery = msgRtSettings.GetExecType() != NYql::NDqProto::TComputeRuntimeSettings::SCAN;
 
         runtimeSettingsBase.StatsMode = msgRtSettings.GetStatsMode();
+        runtimeSettingsBase.UseSpilling = msgRtSettings.GetUseSpilling();
 
         if (msgRtSettings.HasRlPath()) {
             runtimeSettingsBase.RlPath = msgRtSettings.GetRlPath();

ydb/library/yql/dq/actors/compute/dq_compute_actor.h

@@ -242,6 +242,7 @@ struct TComputeRuntimeSettings {
     ui32 ExtraMemoryAllocationPool = 0;
 
     bool FailOnUndelivery = true;
+    bool UseSpilling = false;
 
     std::function<void(bool success, const TIssues& issues)> TerminateHandler;
     TMaybe<NDqProto::TRlPath> RlPath;