initial

nepal · nepal · commit 8b7eb046f0d0 · 2023-12-21T00:27:29.000+03:00
diff --git a/ydb/core/kqp/host/kqp_host.cpp b/ydb/core/kqp/host/kqp_host.cpp
@@ -1558,6 +1558,7 @@ class TKqpHost : public IKqpHost {
                 || settingName == "DisableOrderedColumns"
                 || settingName == "Warning"
                 || settingName == "UseBlocks"
+                || settingName == "BlockEngine"
                 ;
         };
         auto configProvider = CreateConfigProvider(*TypesCtx, gatewaysConfig, {}, allowSettings);
diff --git a/ydb/core/kqp/opt/peephole/kqp_opt_peephole_wide_read.cpp b/ydb/core/kqp/opt/peephole/kqp_opt_peephole_wide_read.cpp
@@ -58,7 +58,7 @@ TExprBase KqpBuildWideReadTable(const TExprBase& node, TExprContext& ctx, TTypeA
     } else if (auto maybeRead = node.Maybe<TKqpReadOlapTableRanges>()) {
         auto read = maybeRead.Cast();
 
-        if (typesCtx.UseBlocks) {
+        if (typesCtx.IsBlockEngineEnabled()) {
             wideRead = Build<TCoWideFromBlocks>(ctx, node.Pos())
                 .Input<TKqpBlockReadOlapTableRanges>()
                     .Table(read.Table())
diff --git a/ydb/library/yql/core/peephole_opt/yql_opt_peephole_physical.cpp b/ydb/library/yql/core/peephole_opt/yql_opt_peephole_physical.cpp
@@ -7774,7 +7774,7 @@ THolder<IGraphTransformer> CreatePeepHoleFinalStageTransformer(TTypeAnnotationCo
     pipeline.Add(
         CreateFunctorTransformer(
             [&types](const TExprNode::TPtr& input, TExprNode::TPtr& output, TExprContext& ctx) -> IGraphTransformer::TStatus {
-                if (types.UseBlocks) {
+                if (types.IsBlockEngineEnabled()) {
                     const auto& extStageRules = TPeepHoleRules::Instance().BlockStageExtRules;
                     return PeepHoleBlockStage(input, output, ctx, types, extStageRules);
                 } else {
@@ -7789,7 +7789,7 @@ THolder<IGraphTransformer> CreatePeepHoleFinalStageTransformer(TTypeAnnotationCo
     pipeline.Add(
         CreateFunctorTransformer(
             [&types](const TExprNode::TPtr& input, TExprNode::TPtr& output, TExprContext& ctx) -> IGraphTransformer::TStatus {
-                if (types.UseBlocks) {
+                if (types.IsBlockEngineEnabled()) {
                     const auto& extStageRules = TPeepHoleRules::Instance().BlockStageExtFinalRules;
                     return PeepHoleBlockStage(input, output, ctx, types, extStageRules);
                 } else {
diff --git a/ydb/library/yql/core/type_ann/type_ann_list.cpp b/ydb/library/yql/core/type_ann/type_ann_list.cpp
@@ -4689,7 +4689,7 @@ namespace {
             return IGraphTransformer::TStatus::Repeat;
         }
 
-        if (isMany && ctx.Types.UseBlocks) {
+        if (isMany && ctx.Types.IsBlockEngineEnabled()) {
             auto streamIndex = inputStructType->FindItem("_yql_group_stream_index");
             if (streamIndex) {
                 const TTypeAnnotationNode* streamIndexType = inputStructType->GetItems()[*streamIndex]->GetItemType();
diff --git a/ydb/library/yql/core/yql_aggregate_expander.cpp b/ydb/library/yql/core/yql_aggregate_expander.cpp
@@ -25,7 +25,7 @@ TExprNode::TPtr TAggregateExpander::ExpandAggregate()
 
     HaveDistinct = AnyOf(AggregatedColumns->ChildrenList(),
         [](const auto& child) { return child->ChildrenSize() == 3; });
-    EffectiveCompact = (HaveDistinct && CompactForDistinct && !TypesCtx.UseBlocks) || ForceCompact || HasSetting(*settings, "compact");
+    EffectiveCompact = (HaveDistinct && CompactForDistinct && !TypesCtx.IsBlockEngineEnabled()) || ForceCompact || HasSetting(*settings, "compact");
     for (const auto& trait : Traits) {
         auto mergeLambda = trait->Child(5);
         if (mergeLambda->Tail().IsCallable("Void")) {
@@ -56,7 +56,7 @@ TExprNode::TPtr TAggregateExpander::ExpandAggregate()
         return GeneratePhases();
     }
 
-    if (TypesCtx.UseBlocks) {
+    if (TypesCtx.IsBlockEngineEnabled()) {
         if (Suffix == "Combine") {
             auto ret = TryGenerateBlockCombine();
             if (ret) {
@@ -2776,7 +2776,7 @@ TExprNode::TPtr TAggregateExpander::GeneratePhases() {
         streams.push_back(SerializeIdxSet(indicies));
     }
 
-    if (TypesCtx.UseBlocks) {
+    if (TypesCtx.IsBlockEngineEnabled()) {
         for (ui32 i = 0; i < unionAllInputs.size(); ++i) {
             unionAllInputs[i] = Ctx.Builder(Node->Pos())
                 .Callable("Map")
@@ -2797,7 +2797,7 @@ TExprNode::TPtr TAggregateExpander::GeneratePhases() {
     }
 
     auto settings = Node->ChildPtr(3);
-    if (TypesCtx.UseBlocks) {
+    if (TypesCtx.IsBlockEngineEnabled()) {
         settings = AddSetting(*settings, Node->Pos(), "many_streams", Ctx.NewList(Node->Pos(), std::move(streams)), Ctx);
     }
 
@@ -2830,7 +2830,7 @@ TExprNode::TPtr TAggregateExpander::TryGenerateBlockCombine() {
 }
 
 TExprNode::TPtr TAggregateExpander::TryGenerateBlockMergeFinalize() {
-    if (UsePartitionsByKeys || !TypesCtx.UseBlocks) {
+    if (UsePartitionsByKeys || !TypesCtx.IsBlockEngineEnabled()) {
         return nullptr;
     }
 
@@ -2919,13 +2919,13 @@ TExprNode::TPtr TAggregateExpander::TryGenerateBlockMergeFinalizeHashed() {
 TExprNode::TPtr ExpandAggregatePeephole(const TExprNode::TPtr& node, TExprContext& ctx, TTypeAnnotationContext& typesCtx) {
     if (NNodes::TCoAggregate::Match(node.Get())) {
         NNodes::TCoAggregate self(node);
-        auto ret = TAggregateExpander::CountAggregateRewrite(self, ctx, typesCtx.UseBlocks);
+        auto ret = TAggregateExpander::CountAggregateRewrite(self, ctx, typesCtx.IsBlockEngineEnabled());
         if (ret != node) {
             YQL_CLOG(DEBUG, Core) << "CountAggregateRewrite on peephole";
             return ret;
         }
     }
-    return ExpandAggregatePeepholeImpl(node, ctx, typesCtx, false, typesCtx.UseBlocks);
+    return ExpandAggregatePeepholeImpl(node, ctx, typesCtx, false, typesCtx.IsBlockEngineEnabled());
 }
 
 } // namespace NYql
diff --git a/ydb/library/yql/core/yql_type_annotation.h b/ydb/library/yql/core/yql_type_annotation.h
@@ -193,6 +193,12 @@ enum class EMatchRecognizeStreamingMode {
     Force,
 };
 
+enum class EBlockEngineMode {
+    Disable /* "disable" */,
+    Auto /* "auto" */,
+    Force /* "force" */,
+};
+
 struct TUdfCachedInfo {
     const TTypeAnnotationNode* FunctionType = nullptr;
     const TTypeAnnotationNode* RunConfigType = nullptr;
@@ -251,6 +257,7 @@ struct TTypeAnnotationContext: public TThrRefBase {
     bool YsonCastToString = true;
     ui32 FolderSubDirsLimit = 1000;
     bool UseBlocks = false;
+    EBlockEngineMode BlockEngineMode = EBlockEngineMode::Disable;
     bool PgEmitAggApply = false;
     IArrowResolver::TPtr ArrowResolver;
     ECostBasedOptimizerType CostBasedOptimizer = ECostBasedOptimizerType::Disable;
@@ -350,7 +357,10 @@ struct TTypeAnnotationContext: public TThrRefBase {
     void SetStats(const TExprNode* input, std::shared_ptr<TOptimizerStatistics> stats) {
         StatisticsMap[input] = stats;
     }
-    
+
+    bool IsBlockEngineEnabled() const {
+        return BlockEngineMode != EBlockEngineMode::Disable || UseBlocks;
+    }
 };
 
 template <> inline
diff --git a/ydb/library/yql/dq/opt/dq_opt_log.cpp b/ydb/library/yql/dq/opt/dq_opt_log.cpp
@@ -21,7 +21,7 @@ TExprBase DqRewriteAggregate(TExprBase node, TExprContext& ctx, TTypeAnnotationC
     if (!node.Maybe<TCoAggregateBase>()) {
         return node;
     }
-    TAggregateExpander aggExpander(true, !typesCtx.UseBlocks && !useFinalizeByKey, useFinalizeByKey, node.Ptr(), ctx, typesCtx, false, compactForDistinct, usePhases);
+    TAggregateExpander aggExpander(true, !typesCtx.IsBlockEngineEnabled() && !useFinalizeByKey, useFinalizeByKey, node.Ptr(), ctx, typesCtx, false, compactForDistinct, usePhases);
     auto result = aggExpander.ExpandAggregate();
     YQL_ENSURE(result);
 
diff --git a/ydb/library/yql/dq/opt/dq_opt_peephole.cpp b/ydb/library/yql/dq/opt/dq_opt_peephole.cpp
@@ -681,7 +681,7 @@ NNodes::TExprBase DqPeepholeRewriteLength(const NNodes::TExprBase& node, TExprCo
     }
 
     auto dqPhyLength = node.Cast<TDqPhyLength>();
-    if (typesCtx.UseBlocks) {
+    if (typesCtx.IsBlockEngineEnabled()) {
         return NNodes::TExprBase(ctx.Builder(node.Pos())
             .Callable("NarrowMap")
                 .Callable(0, "BlockCombineAll")
diff --git a/ydb/library/yql/providers/config/yql_config_provider.cpp b/ydb/library/yql/providers/config/yql_config_provider.cpp
@@ -888,6 +888,18 @@ namespace {
                     return false;
                 }
             }
+            else if (name == "BlockEngine") {
+                if (args.size() != 1) {
+                    ctx.AddError(TIssue(pos, TStringBuilder() << "Expected at most 1 argument, but got " << args.size()));
+                    return false;
+                }
+
+                auto arg = TString{args[0]};
+                if (!TryFromString(arg, Types.BlockEngineMode)) {
+                    ctx.AddError(TIssue(pos, TStringBuilder() << "Expected `disable|auto|force', but got: " << args[0]));
+                    return false;
+                }
+            }
             else {
                 ctx.AddError(TIssue(pos, TStringBuilder() << "Unsupported command: " << name));
                 return false;
diff --git a/ydb/library/yql/providers/dq/opt/logical_optimize.cpp b/ydb/library/yql/providers/dq/opt/logical_optimize.cpp
@@ -324,7 +324,7 @@ class TDqsLogicalOptProposalTransformer : public TOptimizeTransformerBase {
             auto input = aggregate.Input().Maybe<TDqConnection>();
 
             if (input) {
-                auto newNode = TAggregateExpander::CountAggregateRewrite(aggregate, ctx, TypesCtx.UseBlocks);
+                auto newNode = TAggregateExpander::CountAggregateRewrite(aggregate, ctx, TypesCtx.IsBlockEngineEnabled());
                 if (node.Ptr() != newNode) {
                     return TExprBase(newNode);
                 }
diff --git a/ydb/library/yql/providers/yt/provider/yql_yt_logical_optimize.cpp b/ydb/library/yql/providers/yt/provider/yql_yt_logical_optimize.cpp
@@ -2512,7 +2512,7 @@ class TYtLogicalOptProposalTransformer : public TOptimizeTransformerBase {
             return node;
         }
 
-        return TAggregateExpander::CountAggregateRewrite(aggregate, ctx, State_->Types->UseBlocks);
+        return TAggregateExpander::CountAggregateRewrite(aggregate, ctx, State_->Types->IsBlockEngineEnabled());
     }
 
     TMaybeNode<TExprBase> ZeroSampleToZeroLimit(TExprBase node, TExprContext& ctx) const {
diff --git a/ydb/library/yql/sql/pg/pg_sql.cpp b/ydb/library/yql/sql/pg/pg_sql.cpp
@@ -272,6 +272,7 @@ class TConverter : public IPGParseEvents {
         : AstParseResult(astParseResult)
         , Settings(settings)
         , DqEngineEnabled(Settings.DqDefaultAuto->Allow())
+        , BlockEngineEnabled(Settings.BlockDefaultAuto->Allow())
     {
         Positions.push_back({});
         ScanRows(query);
@@ -281,6 +282,10 @@ class TConverter : public IPGParseEvents {
                 DqEngineEnabled = true;
             } else if (flag == "DqEngineForce") {
                 DqEngineForce = true;
+            } else if (flag == "BlockEngineEnable") {
+                BlockEngineEnabled = true;
+            } else if (flag == "BlockEngineForce") {
+                BlockEngineForce = true;
             }
         }
 
@@ -320,6 +325,8 @@ class TConverter : public IPGParseEvents {
         Statements.push_back(L(A("let"), A("world"), L(A(TString(NYql::ConfigureName)), A("world"), configSource,
             QA("OrderedColumns"))));
 
+        ui32 blockEnginePgmPos = Statements.size();
+        Statements.push_back(configSource);
         ui32 costBasedOptimizerPos = Statements.size();
         Statements.push_back(configSource);
         ui32 dqEnginePgmPos = Statements.size();
@@ -359,6 +366,13 @@ class TConverter : public IPGParseEvents {
             Statements.erase(Statements.begin() + costBasedOptimizerPos);
         }
 
+        if (BlockEngineEnabled) {
+            Statements[blockEnginePgmPos] = L(A("let"), A("world"), L(A(TString(NYql::ConfigureName)), A("world"), configSource,
+                QA("BlockEngine"), QA(BlockEngineForce ? "force" : "auto")));
+        } else {
+            Statements.erase(Statements.begin() + blockEnginePgmPos);
+        }
+
         return VL(Statements.data(), Statements.size());
     }
 
@@ -2001,7 +2015,7 @@ class TConverter : public IPGParseEvents {
                 AddError(TStringBuilder() << "VariableSetStmt, expected string literal for " << value->name << " option");
                 return nullptr;
             }
-        } else if (name == "dqengine") {
+        } else if (name == "dqengine" || name == "blockengine") {
             if (ListLength(value->args) != 1) {
                 AddError(TStringBuilder() << "VariableSetStmt, expected 1 arg, but got: " << ListLength(value->args));
                 return nullptr;
@@ -2011,17 +2025,20 @@ class TConverter : public IPGParseEvents {
             if (NodeTag(arg) == T_A_Const && (NodeTag(CAST_NODE(A_Const, arg)->val) == T_String)) {
                 auto rawStr = StrVal(CAST_NODE(A_Const, arg)->val);
                 auto str = to_lower(TString(rawStr));
+                const bool isDqEngine = name == "dqengine";
+                auto& enable = isDqEngine ? DqEngineEnabled : BlockEngineEnabled;
+                auto& force =  isDqEngine ? DqEngineForce   : BlockEngineForce;
                 if (str == "auto") {
-                    DqEngineEnabled = true;
-                    DqEngineForce = false;
+                    enable = true;
+                    force = false;
                 } else if (str == "force") {
-                    DqEngineEnabled = true;
-                    DqEngineForce = true;
+                    enable = true;
+                    force = true;
                 } else if (str == "disable") {
-                    DqEngineEnabled = false;
-                    DqEngineForce = false;
+                    enable = false;
+                    force = false;
                 } else {
-                    AddError(TStringBuilder() << "VariableSetStmt, not supported DqEngine option value: " << rawStr);
+                    AddError(TStringBuilder() << "VariableSetStmt, not supported " << value->name << " option value: " << rawStr);
                     return nullptr;
                 }
             } else {
@@ -4247,6 +4264,8 @@ class TConverter : public IPGParseEvents {
     bool DqEngineEnabled = false;
     bool DqEngineForce = false;
     TString CostBasedOptimizer;
+    bool BlockEngineEnabled = false;
+    bool BlockEngineForce = false;
     TVector<TAstNode*> Statements;
     ui32 ReadIndex = 0;
     TViews Views;
diff --git a/ydb/library/yql/sql/settings/translation_settings.cpp b/ydb/library/yql/sql/settings/translation_settings.cpp
@@ -56,6 +56,7 @@ namespace NSQLTranslation {
         , WarnOnV0(true)
         , V0WarnAsError(ISqlFeaturePolicy::MakeAlwaysDisallow())
         , DqDefaultAuto(ISqlFeaturePolicy::MakeAlwaysDisallow())
+        , BlockDefaultAuto(ISqlFeaturePolicy::MakeAlwaysDisallow())
         , AssumeYdbOnClusterWithSlash(false)
     {}
 
diff --git a/ydb/library/yql/sql/settings/translation_settings.h b/ydb/library/yql/sql/settings/translation_settings.h
@@ -101,6 +101,7 @@ namespace NSQLTranslation {
         bool WarnOnV0;
         ISqlFeaturePolicy::TPtr V0WarnAsError;
         ISqlFeaturePolicy::TPtr DqDefaultAuto;
+        ISqlFeaturePolicy::TPtr BlockDefaultAuto;
         bool AssumeYdbOnClusterWithSlash;
         TString DynamicClusterProvider;
         TString FileAliasPrefix;
diff --git a/ydb/library/yql/sql/v1/context.cpp b/ydb/library/yql/sql/v1/context.cpp
@@ -58,6 +58,8 @@ THashMap<TStringBuf, TPragmaField> CTX_PRAGMA_FIELDS = {
     {"EmitAggApply", &TContext::EmitAggApply},
     {"AnsiLike", &TContext::AnsiLike},
     {"UseBlocks", &TContext::UseBlocks},
+    {"BlockEngineEnable", &TContext::BlockEngineEnable},
+    {"BlockEngineForce", &TContext::BlockEngineForce},
 };
 
 typedef TMaybe<bool> TContext::*TPragmaMaybeField;
@@ -84,6 +86,7 @@ TContext::TContext(const NSQLTranslation::TTranslationSettings& settings,
     , HasPendingErrors(false)
     , DqEngineEnable(Settings.DqDefaultAuto->Allow())
     , AnsiQuotedIdentifiers(settings.AnsiLexer)
+    , BlockEngineEnable(Settings.BlockDefaultAuto->Allow())
 {
     for (auto lib : settings.Libraries) {
         Libraries.emplace(lib, TLibraryStuff());
diff --git a/ydb/library/yql/sql/v1/context.h b/ydb/library/yql/sql/v1/context.h
@@ -308,6 +308,8 @@ namespace NSQLTranslationV1 {
         bool AnsiLike = false;
         bool FeatureR010 = false; //Row pattern recognition: FROM clause
         TMaybe<bool> CompactGroupBy;
+        bool BlockEngineEnable = false;
+        bool BlockEngineForce = false;
     };
 
     class TColumnRefScope {
diff --git a/ydb/library/yql/sql/v1/query.cpp b/ydb/library/yql/sql/v1/query.cpp
@@ -2691,6 +2691,12 @@ class TYqlProgramNode: public TAstListNode {
                 if (ctx.UseBlocks) {
                     Add(Y("let", "world", Y(TString(ConfigureName), "world", configSource, BuildQuotedAtom(Pos, "UseBlocks"))));
                 }
+
+                if (ctx.BlockEngineEnable) {
+                    TString mode = ctx.BlockEngineForce ? "force" : "auto";
+                    Add(Y("let", "world", Y(TString(ConfigureName), "world", configSource,
+                        BuildQuotedAtom(Pos, "BlockEngine"), BuildQuotedAtom(Pos, mode))));
+                }
             }
         }
 
diff --git a/ydb/library/yql/sql/v1/sql_query.cpp b/ydb/library/yql/sql/v1/sql_query.cpp
@@ -1847,7 +1847,7 @@ TNodePtr TSqlQuery::PragmaStatement(const TRule_pragma_stmt& stmt, bool& success
         } else if (normalizedPragma == "disableansiinforemptyornullableitemscollections") {
             Ctx.AnsiInForEmptyOrNullableItemsCollections = false;
             Ctx.IncrementMonCounter("sql_pragma", "DisableAnsiInForEmptyOrNullableItemsCollections");
-        } else if (normalizedPragma == "dqengine") {
+        } else if (normalizedPragma == "dqengine" || normalizedPragma == "blockengine") {
             Ctx.IncrementMonCounter("sql_pragma", "DqEngine");
             if (values.size() != 1 || !values[0].GetLiteral()
                 || ! (*values[0].GetLiteral() == "disable" || *values[0].GetLiteral() == "auto" || *values[0].GetLiteral() == "force"))
@@ -1856,15 +1856,18 @@ TNodePtr TSqlQuery::PragmaStatement(const TRule_pragma_stmt& stmt, bool& success
                 Ctx.IncrementMonCounter("sql_errors", "BadPragmaValue");
                 return {};
             }
+            const bool isDqEngine = normalizedPragma == "dqengine";
+            auto& enable = isDqEngine ? Ctx.DqEngineEnable : Ctx.BlockEngineEnable;
+            auto& force =  isDqEngine ? Ctx.DqEngineForce  : Ctx.BlockEngineForce;
             if (*values[0].GetLiteral() == "disable") {
-                Ctx.DqEngineEnable = false;
-                Ctx.DqEngineForce = false;
+                enable = false;
+                force = false;
             } else if (*values[0].GetLiteral() == "force") {
-                Ctx.DqEngineEnable = true;
-                Ctx.DqEngineForce = true;
+                enable = true;
+                force = true;
             } else if (*values[0].GetLiteral() == "auto") {
-                Ctx.DqEngineEnable = true;
-                Ctx.DqEngineForce = false;
+                enable = true;
+                force = false;
             }
         } else if (normalizedPragma == "ansirankfornullablekeys") {
             Ctx.AnsiRankForNullableKeys = true;

Original file line number	Diff line number	Diff line change
`@@ -1558,6 +1558,7 @@ class TKqpHost : public IKqpHost {`
`1558`	`1558`	`\|\| settingName == "DisableOrderedColumns"`
`1559`	`1559`	`\|\| settingName == "Warning"`
`1560`	`1560`	`\|\| settingName == "UseBlocks"`
	`1561`	`+ \|\| settingName == "BlockEngine"`
`1561`	`1562`	`;`
`1562`	`1563`	`};`
`1563`	`1564`	`auto configProvider = CreateConfigProvider(*TypesCtx, gatewaysConfig, {}, allowSettings);`
Original file line number	Diff line number	Diff line change
`@@ -4689,7 +4689,7 @@ namespace {`
`4689`	`4689`	`return IGraphTransformer::TStatus::Repeat;`
`4690`	`4690`	`}`
`4691`	`4691`
`4692`		`- if (isMany && ctx.Types.UseBlocks) {`
	`4692`	`+ if (isMany && ctx.Types.IsBlockEngineEnabled()) {`
`4693`	`4693`	`auto streamIndex = inputStructType->FindItem("_yql_group_stream_index");`
`4694`	`4694`	`if (streamIndex) {`
`4695`	`4695`	`const TTypeAnnotationNode* streamIndexType = inputStructType->GetItems()[*streamIndex]->GetItemType();`
Original file line number	Diff line number	Diff line change
`@@ -25,7 +25,7 @@ TExprNode::TPtr TAggregateExpander::ExpandAggregate()`
`25`	`25`
`26`	`26`	`HaveDistinct = AnyOf(AggregatedColumns->ChildrenList(),`
`27`	`27`	`[](const auto& child) { return child->ChildrenSize() == 3; });`
`28`		`- EffectiveCompact = (HaveDistinct && CompactForDistinct && !TypesCtx.UseBlocks) \|\| ForceCompact \|\| HasSetting(*settings, "compact");`
	`28`	`+ EffectiveCompact = (HaveDistinct && CompactForDistinct && !TypesCtx.IsBlockEngineEnabled()) \|\| ForceCompact \|\| HasSetting(*settings, "compact");`
`29`	`29`	`for (const auto& trait : Traits) {`
`30`	`30`	`auto mergeLambda = trait->Child(5);`
`31`	`31`	`if (mergeLambda->Tail().IsCallable("Void")) {`
`@@ -56,7 +56,7 @@ TExprNode::TPtr TAggregateExpander::ExpandAggregate()`
`56`	`56`	`return GeneratePhases();`
`57`	`57`	`}`
`58`	`58`
`59`		`- if (TypesCtx.UseBlocks) {`
	`59`	`+ if (TypesCtx.IsBlockEngineEnabled()) {`
`60`	`60`	`if (Suffix == "Combine") {`
`61`	`61`	`auto ret = TryGenerateBlockCombine();`
`62`	`62`	`if (ret) {`
`@@ -2776,7 +2776,7 @@ TExprNode::TPtr TAggregateExpander::GeneratePhases() {`
`2776`	`2776`	`streams.push_back(SerializeIdxSet(indicies));`
`2777`	`2777`	`}`
`2778`	`2778`
`2779`		`- if (TypesCtx.UseBlocks) {`
	`2779`	`+ if (TypesCtx.IsBlockEngineEnabled()) {`
`2780`	`2780`	`for (ui32 i = 0; i < unionAllInputs.size(); ++i) {`
`2781`	`2781`	`unionAllInputs[i] = Ctx.Builder(Node->Pos())`
`2782`	`2782`	`.Callable("Map")`
`@@ -2797,7 +2797,7 @@ TExprNode::TPtr TAggregateExpander::GeneratePhases() {`
`2797`	`2797`	`}`
`2798`	`2798`
`2799`	`2799`	`auto settings = Node->ChildPtr(3);`
`2800`		`- if (TypesCtx.UseBlocks) {`
	`2800`	`+ if (TypesCtx.IsBlockEngineEnabled()) {`
`2801`	`2801`	`settings = AddSetting(*settings, Node->Pos(), "many_streams", Ctx.NewList(Node->Pos(), std::move(streams)), Ctx);`
`2802`	`2802`	`}`
`2803`	`2803`
`@@ -2830,7 +2830,7 @@ TExprNode::TPtr TAggregateExpander::TryGenerateBlockCombine() {`
`2830`	`2830`	`}`
`2831`	`2831`
`2832`	`2832`	`TExprNode::TPtr TAggregateExpander::TryGenerateBlockMergeFinalize() {`
`2833`		`- if (UsePartitionsByKeys \|\| !TypesCtx.UseBlocks) {`
	`2833`	`+ if (UsePartitionsByKeys \|\| !TypesCtx.IsBlockEngineEnabled()) {`
`2834`	`2834`	`return nullptr;`
`2835`	`2835`	`}`
`2836`	`2836`
`@@ -2919,13 +2919,13 @@ TExprNode::TPtr TAggregateExpander::TryGenerateBlockMergeFinalizeHashed() {`
`2919`	`2919`	`TExprNode::TPtr ExpandAggregatePeephole(const TExprNode::TPtr& node, TExprContext& ctx, TTypeAnnotationContext& typesCtx) {`
`2920`	`2920`	`if (NNodes::TCoAggregate::Match(node.Get())) {`
`2921`	`2921`	`NNodes::TCoAggregate self(node);`
`2922`		`- auto ret = TAggregateExpander::CountAggregateRewrite(self, ctx, typesCtx.UseBlocks);`
	`2922`	`+ auto ret = TAggregateExpander::CountAggregateRewrite(self, ctx, typesCtx.IsBlockEngineEnabled());`
`2923`	`2923`	`if (ret != node) {`
`2924`	`2924`	`YQL_CLOG(DEBUG, Core) << "CountAggregateRewrite on peephole";`
`2925`	`2925`	`return ret;`
`2926`	`2926`	`}`
`2927`	`2927`	`}`
`2928`		`- return ExpandAggregatePeepholeImpl(node, ctx, typesCtx, false, typesCtx.UseBlocks);`
	`2928`	`+ return ExpandAggregatePeepholeImpl(node, ctx, typesCtx, false, typesCtx.IsBlockEngineEnabled());`
`2929`	`2929`	`}`
`2930`	`2930`
`2931`	`2931`	`} // namespace NYql`
Original file line number	Diff line number	Diff line change
`@@ -21,7 +21,7 @@ TExprBase DqRewriteAggregate(TExprBase node, TExprContext& ctx, TTypeAnnotationC`
`21`	`21`	`if (!node.Maybe<TCoAggregateBase>()) {`
`22`	`22`	`return node;`
`23`	`23`	`}`
`24`		`- TAggregateExpander aggExpander(true, !typesCtx.UseBlocks && !useFinalizeByKey, useFinalizeByKey, node.Ptr(), ctx, typesCtx, false, compactForDistinct, usePhases);`
	`24`	`+ TAggregateExpander aggExpander(true, !typesCtx.IsBlockEngineEnabled() && !useFinalizeByKey, useFinalizeByKey, node.Ptr(), ctx, typesCtx, false, compactForDistinct, usePhases);`
`25`	`25`	`auto result = aggExpander.ExpandAggregate();`
`26`	`26`	`YQL_ENSURE(result);`
`27`	`27`
Original file line number	Diff line number	Diff line change
`@@ -681,7 +681,7 @@ NNodes::TExprBase DqPeepholeRewriteLength(const NNodes::TExprBase& node, TExprCo`
`681`	`681`	`}`
`682`	`682`
`683`	`683`	`auto dqPhyLength = node.Cast<TDqPhyLength>();`
`684`		`- if (typesCtx.UseBlocks) {`
	`684`	`+ if (typesCtx.IsBlockEngineEnabled()) {`
`685`	`685`	`return NNodes::TExprBase(ctx.Builder(node.Pos())`
`686`	`686`	`.Callable("NarrowMap")`
`687`	`687`	`.Callable(0, "BlockCombineAll")`
Original file line number	Diff line number	Diff line change
`@@ -324,7 +324,7 @@ class TDqsLogicalOptProposalTransformer : public TOptimizeTransformerBase {`
`324`	`324`	`auto input = aggregate.Input().Maybe<TDqConnection>();`
`325`	`325`
`326`	`326`	`if (input) {`
`327`		`- auto newNode = TAggregateExpander::CountAggregateRewrite(aggregate, ctx, TypesCtx.UseBlocks);`
	`327`	`+ auto newNode = TAggregateExpander::CountAggregateRewrite(aggregate, ctx, TypesCtx.IsBlockEngineEnabled());`
`328`	`328`	`if (node.Ptr() != newNode) {`
`329`	`329`	`return TExprBase(newNode);`
`330`	`330`	`}`