split SpillingEngine and channels spilling (#6406)

Author: lll-phill-lll

ydb/library/yql/providers/dq/common/yql_dq_settings.cpp

@@ -80,10 +80,12 @@ TDqConfiguration::TDqConfiguration() {
     REGISTER_SETTING(*this, SpillingEngine)
         .Parser([](const TString& v) {
             return FromString<TDqSettings::ESpillingEngine>(v);
-        })
-        .ValueSetter([this](const TString&, TDqSettings::ESpillingEngine value) {
-            SpillingEngine = value;
-            if (value != TDqSettings::ESpillingEngine::Disable) {
+        });
+
+    REGISTER_SETTING(*this, EnableSpillingInChannels)
+        .ValueSetter([this](const TString&, bool value) {
+            EnableSpillingInChannels = value;
+            if (value) {
                 SplitStageOnDqReplicate = false;
                 EnableDqReplicate = true;
             }

ydb/library/yql/providers/dq/common/yql_dq_settings.h

@@ -66,6 +66,7 @@ struct TDqSettings {
         static constexpr ui64 MaxAttachmentsSize = 2_GB;
         static constexpr bool SplitStageOnDqReplicate = true;
         static constexpr ui64 EnableSpillingNodes = 0;
+        static constexpr bool EnableSpillingInChannels = false;
     };
 
     using TPtr = std::shared_ptr<TDqSettings>;
@@ -138,6 +139,7 @@ struct TDqSettings {
     NCommon::TConfSetting<bool, false> SplitStageOnDqReplicate;
 
     NCommon::TConfSetting<ui64, false> EnableSpillingNodes;
+    NCommon::TConfSetting<bool, false> EnableSpillingInChannels;
 
     NCommon::TConfSetting<ui64, false> _MaxAttachmentsSize;
     NCommon::TConfSetting<bool, false> DisableCheckpoints;
@@ -193,6 +195,7 @@ struct TDqSettings {
         SAVE_SETTING(ExportStats);
         SAVE_SETTING(TaskRunnerStats);
         SAVE_SETTING(SpillingEngine);
+        SAVE_SETTING(EnableSpillingInChannels);
         SAVE_SETTING(DisableCheckpoints);
 #undef SAVE_SETTING
     }
@@ -219,13 +222,18 @@ struct TDqSettings {
         }
     }
 
-    bool IsSpillingEnabled() const {
+    bool IsSpillingEngineEnabled() const {
         return SpillingEngine.Get().GetOrElse(TDqSettings::TDefault::SpillingEngine) != ESpillingEngine::Disable;
     }
 
+    bool IsSpillingInChannelsEnabled() const {
+        if (!IsSpillingEngineEnabled()) return false;
+        return EnableSpillingInChannels.Get().GetOrElse(TDqSettings::TDefault::EnableSpillingInChannels) != false;
+    }
+
     ui64 GetEnabledSpillingNodes() const {
-        if (!IsSpillingEnabled()) return 0;
-        return EnableSpillingNodes.Get().GetOrElse(0);
+        if (!IsSpillingEngineEnabled()) return 0;
+        return EnableSpillingNodes.Get().GetOrElse(TDqSettings::TDefault::EnableSpillingNodes);
     }
 
     bool IsDqReplicateEnabled(const TTypeAnnotationContext& typesCtx) const {

ydb/library/yql/providers/dq/planner/execution_planner.cpp

@@ -433,7 +433,7 @@ namespace NYql::NDqs {
 
             bool enableSpilling = false;
             if (task.Outputs.size() > 1) {
-                enableSpilling = Settings->IsSpillingEnabled();
+                enableSpilling = Settings->IsSpillingInChannelsEnabled();
             }
             for (auto& output : task.Outputs) {
                 FillOutputDesc(*taskDesc.AddOutputs(), output, enableSpilling);
@@ -448,7 +448,7 @@ namespace NYql::NDqs {
             taskMeta.SetStageId(publicId);
             taskDesc.MutableMeta()->PackFrom(taskMeta);
             taskDesc.SetStageId(stageId);
-            taskDesc.SetEnableSpilling(Settings->IsSpillingEnabled());
+            taskDesc.SetEnableSpilling(Settings->GetEnabledSpillingNodes());
 
             if (Settings->DisableLLVMForBlockStages.Get().GetOrElse(true)) {
                 auto& stage = TasksGraph.GetStageInfo(task.StageId).Meta.Stage;

ydb/library/yql/providers/dq/provider/yql_dq_validate.cpp

@@ -147,7 +147,7 @@ class TDqExecutionValidator {
         , State_(state)
         , CheckSelfJoin_(!TypeCtx_.ForceDq
             && !State_->Settings->SplitStageOnDqReplicate.Get().GetOrElse(TDqSettings::TDefault::SplitStageOnDqReplicate)
-            && !State_->Settings->IsSpillingEnabled())
+            && !State_->Settings->IsSpillingInChannelsEnabled())
     {}
 
     bool ValidateDqExecution(const TExprNode& node) {