add spilling nodes and pragma (#6769)

Author: gridnevvvit

ydb/core/kqp/provider/yql_kikimr_exec.cpp

@@ -815,10 +815,11 @@ class TKiSourceCallableExecutionTransformer : public TAsyncCallbackTransformer<T
 
         TVector<TExprBase> fakeReads;
         auto paramsType = NDq::CollectParameters(programLambda, ctx);
+        NDq::TSpillingSettings spillingSettings{SessionCtx->Config().GetEnabledSpillingNodes()};
         lambda = NDq::BuildProgram(
             programLambda, *paramsType, compiler, SessionCtx->Query().QueryData->GetAllocState()->TypeEnv,
                 *SessionCtx->Query().QueryData->GetAllocState()->HolderFactory.GetFunctionRegistry(),
-                ctx, fakeReads, {});
+                ctx, fakeReads, spillingSettings);
 
         NKikimr::NMiniKQL::TProgramBuilder programBuilder(SessionCtx->Query().QueryData->GetAllocState()->TypeEnv,
             *SessionCtx->Query().QueryData->GetAllocState()->HolderFactory.GetFunctionRegistry());

ydb/core/kqp/provider/yql_kikimr_settings.cpp

@@ -3,6 +3,8 @@
 #include <ydb/core/protos/config.pb.h>
 #include <ydb/core/protos/table_service_config.pb.h>
 #include <util/generic/size_literals.h>
+#include <util/string/split.h>
+#include <ydb/library/yql/providers/dq/common/yql_dq_settings.h>
 
 namespace NYql {
 
@@ -73,6 +75,21 @@ TKikimrConfiguration::TKikimrConfiguration() {
 
     REGISTER_SETTING(*this, OptUseFinalizeByKey);
     REGISTER_SETTING(*this, CostBasedOptimizationLevel);
+    REGISTER_SETTING(*this, EnableSpillingNodes)
+        .Parser([](const TString& v) {
+            ui64 res = 0;
+            TVector<TString> vec;
+            StringSplitter(v).SplitBySet(",;| ").AddTo(&vec);
+            for (auto& s: vec) {
+                if (s.empty()) {
+                    throw yexception() << "Empty value item";
+                }
+                auto value = FromStringWithDefault<NYql::TDqSettings::EEnabledSpillingNodes>(
+                    s, NYql::TDqSettings::EEnabledSpillingNodes::None);
+                res |= ui64(value);
+            }
+            return res;
+        });
 
     REGISTER_SETTING(*this, MaxDPccpDPTableSize);
 
@@ -129,6 +146,10 @@ bool TKikimrSettings::HasOptUseFinalizeByKey() const {
     return GetOptionalFlagValue(OptUseFinalizeByKey.Get()) != EOptionalFlag::Disabled;
 }
 
+ui64 TKikimrSettings::GetEnabledSpillingNodes() const {
+    return EnableSpillingNodes.Get().GetOrElse(0);
+}
+
 
 EOptionalFlag TKikimrSettings::GetOptPredicateExtract() const {
     return GetOptionalFlagValue(OptEnablePredicateExtract.Get());

ydb/core/kqp/provider/yql_kikimr_settings.h

@@ -58,6 +58,7 @@ struct TKikimrSettings {
     NCommon::TConfSetting<TString, false> OptCardinalityHints;
     NCommon::TConfSetting<TString, false> OptJoinAlgoHints;
     NCommon::TConfSetting<TString, false> OptJoinOrderHints;
+    NCommon::TConfSetting<TString, false> OverrideStatistics;
 
     /* Disable optimizer rules */
     NCommon::TConfSetting<bool, false> OptDisableTopSort;
@@ -89,6 +90,7 @@ struct TKikimrSettings {
     bool HasOptEnableOlapPushdown() const;
     bool HasOptEnableOlapProvideComputeSharding() const;
     bool HasOptUseFinalizeByKey() const;
+    ui64 GetEnabledSpillingNodes() const;
 
     EOptionalFlag GetOptPredicateExtract() const;
     EOptionalFlag GetUseLlvm() const;

ydb/core/kqp/query_compiler/kqp_query_compiler.cpp

@@ -777,8 +777,9 @@ class TKqpQueryCompiler : public IKqpQueryCompiler {
         stageProto.SetIsEffectsStage(hasEffects || hasTxTableSink);
 
         auto paramsType = CollectParameters(stage, ctx);
+        NDq::TSpillingSettings spillingSettings{Config->GetEnabledSpillingNodes()};
         auto programBytecode = NDq::BuildProgram(stage.Program(), *paramsType, *KqlCompiler, TypeEnv, FuncRegistry,
-            ctx, {}, {});
+            ctx, {}, spillingSettings);
 
         auto& programProto = *stageProto.MutableProgram();
         programProto.SetRuntimeVersion(NYql::NDqProto::ERuntimeVersion::RUNTIME_VERSION_YQL_1_0);

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

@@ -110,7 +110,7 @@ TDqConfiguration::TDqConfiguration() {
                 if (s.empty()) {
                     throw yexception() << "Empty value item";
                 }
-                auto value = FromString<EEnabledSpillingNodes>(s);
+                auto value = FromStringWithDefault<EEnabledSpillingNodes>(s, EEnabledSpillingNodes::None);
                 res |= ui64(value);
             }
             return res;

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

@@ -29,6 +29,7 @@ struct TDqSettings {
     };
 
     enum class EEnabledSpillingNodes : ui64 {
+        None        = 0ULL      /* None */,
         GraceJoin   = 1ULL      /* "GraceJoin" */,
         All         = ~0ULL     /* "All" */,
     };