[YQL-16903] Change default value for EmitAggApply and dq.EnableDqReplicate if BlockEngine is enabled

Author: nepal

ydb/library/yql/core/common_opt/yql_co_pgselect.cpp

@@ -1858,7 +1858,8 @@ TExprNode::TPtr BuildAggregationTraits(TPositionHandle pos, bool onWindow, const
         extractor = ctx.NewLambda(pos, std::move(arguments), std::move(aggFuncArgs));
     }
 
-    if (optCtx.Types->PgEmitAggApply && !onWindow) {
+    const bool blockEngineEnabled = optCtx.Types->BlockEngineMode != EBlockEngineMode::Disable;
+    if (optCtx.Types->PgEmitAggApply.GetOrElse(blockEngineEnabled) && !onWindow) {
         return ctx.Builder(pos)
             .Callable("AggApply")
                 .Atom(0, TString("pg_") + func)

ydb/library/yql/core/yql_type_annotation.h

@@ -259,7 +259,7 @@ struct TTypeAnnotationContext: public TThrRefBase {
     ui32 FolderSubDirsLimit = 1000;
     bool UseBlocks = false;
     EBlockEngineMode BlockEngineMode = EBlockEngineMode::Disable;
-    bool PgEmitAggApply = false;
+    TMaybe<bool> PgEmitAggApply;
     IArrowResolver::TPtr ArrowResolver;
     ECostBasedOptimizerType CostBasedOptimizer = ECostBasedOptimizerType::Disable;
     bool MatchRecognize = false;

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

@@ -209,6 +209,11 @@ struct TDqSettings {
     bool IsSpillingEnabled() const {
         return SpillingEngine.Get().GetOrElse(TDqSettings::TDefault::SpillingEngine) != ESpillingEngine::Disable;
     }
+
+    bool IsDqReplicateEnabled(const TTypeAnnotationContext& typesCtx) const {
+        return EnableDqReplicate.Get().GetOrElse(
+            typesCtx.BlockEngineMode != EBlockEngineMode::Disable || TDqSettings::TDefault::EnableDqReplicate);
+    }
 };
 
 struct TDqConfiguration: public TDqSettings, public NCommon::TSettingDispatcher {

ydb/library/yql/providers/dq/opt/logical_optimize.cpp

@@ -101,7 +101,7 @@ class TDqsLogicalOptProposalTransformer : public TOptimizeTransformerBase {
     TMaybeNode<TExprBase> UnorderedOverDqReadWrap(TExprBase node, TExprContext& ctx, const TGetParents& getParents) const {
         const auto unordered = node.Cast<TCoUnorderedBase>();
         if (const auto maybeRead = unordered.Input().Maybe<TDqReadWrapBase>().Input()) {
-            if (Config->EnableDqReplicate.Get().GetOrElse(TDqSettings::TDefault::EnableDqReplicate)) {
+            if (Config->IsDqReplicateEnabled(TypesCtx)) {
                 const TParentsMap* parentsMap = getParents();
                 auto parentsIt = parentsMap->find(unordered.Input().Raw());
                 YQL_ENSURE(parentsIt != parentsMap->cend());
@@ -127,7 +127,7 @@ class TDqsLogicalOptProposalTransformer : public TOptimizeTransformerBase {
     TMaybeNode<TExprBase> ExtractMembersOverDqReadWrap(TExprBase node, TExprContext& ctx, const TGetParents& getParents) {
         auto extract = node.Cast<TCoExtractMembers>();
         if (const auto maybeRead = extract.Input().Maybe<TDqReadWrap>().Input()) {
-            if (Config->EnableDqReplicate.Get().GetOrElse(TDqSettings::TDefault::EnableDqReplicate)) {
+            if (Config->IsDqReplicateEnabled(TypesCtx)) {
                 const TParentsMap* parentsMap = getParents();
                 auto parentsIt = parentsMap->find(extract.Input().Raw());
                 YQL_ENSURE(parentsIt != parentsMap->cend());

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

@@ -36,7 +36,7 @@ class TDqDataProviderSink: public TDataProviderBase {
         , PhysicalFinalizingTransformer([] () { return CreateDqsFinalizingOptTransformer(); })
         , TypeAnnotationTransformer([state] () {
             return CreateDqsDataSinkTypeAnnotationTransformer(
-                state->TypeCtx, state->Settings->EnableDqReplicate.Get().GetOrElse(TDqSettings::TDefault::EnableDqReplicate));
+                state->TypeCtx, state->Settings->IsDqReplicateEnabled(*state->TypeCtx));
         })
         , ConstraintsTransformer([] () { return CreateDqDataSinkConstraintTransformer(); })
         , RecaptureTransformer([state] () { return CreateDqsRecaptureTransformer(state); })

ydb/library/yql/sql/v1/aggregation.cpp

@@ -28,6 +28,11 @@ namespace {
         }
         return false;
     }
+
+    bool ShouldEmitAggApply(const TContext& ctx) {
+        const bool blockEngineEnabled = ctx.BlockEngineEnable || ctx.BlockEngineForce;
+        return ctx.EmitAggApply.GetOrElse(blockEngineEnabled);
+    }
 }
 
 static const THashSet<TString> AggApplyFuncs = {
@@ -58,7 +63,7 @@ class TAggregationFactory : public IAggregation {
 
 protected:
     bool InitAggr(TContext& ctx, bool isFactory, ISource* src, TAstListNode& node, const TVector<TNodePtr>& exprs) override {
-        if (!ctx.EmitAggApply) {
+        if (!ShouldEmitAggApply(ctx)) {
             AggApplyName = "";
         }
 
@@ -1349,7 +1354,7 @@ class TPGFactoryAggregation final : public TAggregationFactory {
         Y_UNUSED(many);
         Y_UNUSED(ctx);
         Y_UNUSED(allowAggApply);
-        if (ctx.EmitAggApply && allowAggApply && AggMode != EAggregateMode::OverWindow) {
+        if (ShouldEmitAggApply(ctx) && allowAggApply && AggMode != EAggregateMode::OverWindow) {
             return Y("AggApply",
                 Q("pg_" + to_lower(PgFunc)), Y("ListItemType", type), Lambda);
         }

ydb/library/yql/sql/v1/context.cpp

@@ -55,7 +55,6 @@ THashMap<TStringBuf, TPragmaField> CTX_PRAGMA_FIELDS = {
     {"FlexibleTypes", &TContext::FlexibleTypes},
     {"AnsiCurrentRow", &TContext::AnsiCurrentRow},
     {"EmitStartsWith", &TContext::EmitStartsWith},
-    {"EmitAggApply", &TContext::EmitAggApply},
     {"AnsiLike", &TContext::AnsiLike},
     {"UseBlocks", &TContext::UseBlocks},
     {"BlockEngineEnable", &TContext::BlockEngineEnable},
@@ -67,6 +66,7 @@ typedef TMaybe<bool> TContext::*TPragmaMaybeField;
 THashMap<TStringBuf, TPragmaMaybeField> CTX_PRAGMA_MAYBE_FIELDS = {
     {"AnsiRankForNullableKeys", &TContext::AnsiRankForNullableKeys},
     {"AnsiInForEmptyOrNullableItemsCollections", &TContext::AnsiInForEmptyOrNullableItemsCollections},
+    {"EmitAggApply", &TContext::EmitAggApply},
     {"CompactGroupBy", &TContext::CompactGroupBy},
 };
 

ydb/library/yql/sql/v1/context.h

@@ -305,7 +305,7 @@ namespace NSQLTranslationV1 {
         NYql::TWarningPolicy WarningPolicy;
         TString PqReadByRtmrCluster;
         bool EmitStartsWith = true;
-        bool EmitAggApply = false;
+        TMaybe<bool> EmitAggApply;
         bool UseBlocks = false;
         bool AnsiLike = false;
         bool FeatureR010 = false; //Row pattern recognition: FROM clause