fix

Author: pashandor789

ydb/core/kqp/compute_actor/kqp_compute_actor_factory.cpp

@@ -211,7 +211,7 @@ class TKqpCaFactory : public IKqpNodeComputeActorFactory {
 
         if (tableKind == ETableKind::Datashard || tableKind == ETableKind::Olap) {
             YQL_ENSURE(args.ComputesByStages);
-            auto& info = args.ComputesByStages->UpsertTaskWithScan(*args.Task, meta, !AppData()->FeatureFlags.GetEnableSeparationComputeActorsFromRead());
+            auto& info = args.ComputesByStages->UpsertTaskWithScan(*args.Task, meta);
             IActor* computeActor = CreateKqpScanComputeActor(
                 args.ExecuterId, args.TxId,
                 args.Task, AsyncIoFactory, runtimeSettings, memoryLimits,

ydb/core/kqp/compute_actor/kqp_compute_actor_factory.h

@@ -44,9 +44,9 @@ class TComputeStageInfo {
         return true;
     }
 
-    TMetaScan& MergeMetaReads(const NYql::NDqProto::TDqTask& task, const NKikimrTxDataShard::TKqpTransaction::TScanTaskMeta& meta, const bool forceOneToMany) {
+    TMetaScan& MergeMetaReads(const NYql::NDqProto::TDqTask& task, const NKikimrTxDataShard::TKqpTransaction::TScanTaskMeta& meta) {
         YQL_ENSURE(meta.ReadsSize(), "unexpected merge with no reads");
-        if (forceOneToMany || !task.HasMetaId()) {
+        if (!task.HasMetaId()) {
             MetaInfo.emplace_back(TMetaScan(meta));
             return MetaInfo.back();
         } else {
@@ -85,12 +85,12 @@ class TComputeStagesWithScan {
         }
     }
 
-    TMetaScan& UpsertTaskWithScan(const NYql::NDqProto::TDqTask& dqTask, const NKikimrTxDataShard::TKqpTransaction::TScanTaskMeta& meta, const bool forceOneToMany) {
+    TMetaScan& UpsertTaskWithScan(const NYql::NDqProto::TDqTask& dqTask, const NKikimrTxDataShard::TKqpTransaction::TScanTaskMeta& meta) {
         auto it = Stages.find(dqTask.GetStageId());
         if (it == Stages.end()) {
             it = Stages.emplace(dqTask.GetStageId(), TComputeStageInfo()).first;
         }
-        return it->second.MergeMetaReads(dqTask, meta, forceOneToMany);
+        return it->second.MergeMetaReads(dqTask, meta);
     }
 };
 

ydb/core/kqp/executer_actor/kqp_executer_impl.h

@@ -1733,7 +1733,8 @@ class TKqpExecuterBase : public TActor<TDerived> {
         auto& stage = stageInfo.Meta.GetStage(stageInfo.Id);
 
         auto& columnShardHashV1Params = stageInfo.Meta.ColumnShardHashV1Params;
-        if (enableShuffleElimination && stage.GetIsShuffleEliminated() && stageInfo.Meta.ColumnTableInfoPtr) {
+        bool shuffleEliminated = enableShuffleElimination && stage.GetIsShuffleEliminated();
+        if (shuffleEliminated && stageInfo.Meta.ColumnTableInfoPtr) {
             const auto& tableDesc = stageInfo.Meta.ColumnTableInfoPtr->Description;
             columnShardHashV1Params.SourceShardCount = tableDesc.GetColumnShardCount();
             columnShardHashV1Params.SourceTableKeyColumnTypes = std::make_shared<TVector<NScheme::TTypeInfo>>();
@@ -1779,7 +1780,7 @@ class TKqpExecuterBase : public TActor<TDerived> {
                 }
             }
 
-            if (!AppData()->FeatureFlags.GetEnableSeparationComputeActorsFromRead() || (!isOlapScan && readSettings.IsSorted())) {
+            if (!AppData()->FeatureFlags.GetEnableSeparationComputeActorsFromRead() && !shuffleEliminated || (!isOlapScan && readSettings.IsSorted())) {
                 for (auto&& pair : nodeShards) {
                     auto& shardsInfo = pair.second;
                     for (auto&& shardInfo : shardsInfo) {
@@ -1798,7 +1799,7 @@ class TKqpExecuterBase : public TActor<TDerived> {
                     }
                 }
 
-            } else if (enableShuffleElimination && stage.GetIsShuffleEliminated() /* save partitioning for shuffle elimination */) {
+            } else if (shuffleEliminated /* save partitioning for shuffle elimination */) {
                 std::size_t stageInternalTaskId = 0;
                 columnShardHashV1Params.TaskIndexByHash = std::make_shared<TVector<ui64>>();
                 columnShardHashV1Params.TaskIndexByHash->resize(columnShardHashV1Params.SourceShardCount);

ydb/core/kqp/ut/join/kqp_join_order_ut.cpp

@@ -92,7 +92,17 @@ static void CreateSampleTable(NYdb::NQuery::TSession session, bool useColumnStor
     CreateView(session, "view/tpch_random_join_view.sql");
 }
 
-static TKikimrRunner GetKikimrWithJoinSettings(bool useStreamLookupJoin = false, TString stats = "", bool useCBO = true){
+struct TExecuteParams {
+    bool RemoveLimitOperator = false;
+    bool EnableSeparationComputeActorsFromRead = true;
+};
+
+static TKikimrRunner GetKikimrWithJoinSettings(
+    bool useStreamLookupJoin = false,
+    TString stats = "",
+    bool useCBO = true,
+    const TExecuteParams& params = {}
+){
     TVector<NKikimrKqp::TKqpSetting> settings;
 
     NKikimrKqp::TKqpSetting setting;
@@ -116,6 +126,7 @@ static TKikimrRunner GetKikimrWithJoinSettings(bool useStreamLookupJoin = false,
     appConfig.MutableTableServiceConfig()->SetDefaultEnableShuffleElimination(true);
 
     auto serverSettings = TKikimrSettings().SetAppConfig(appConfig);
+    serverSettings.FeatureFlags.SetEnableSeparationComputeActorsFromRead(params.EnableSeparationComputeActorsFromRead);
     serverSettings.SetKqpSettings(settings);
     return TKikimrRunner(serverSettings);
 }
@@ -365,8 +376,20 @@ Y_UNIT_TEST_SUITE(KqpJoinOrder) {
         TChainTester(65).Test();
     }
 
+<<<<<<< HEAD
     TString ExecuteJoinOrderTestGenericQueryWithStats(const TString& queryPath, const TString& statsPath, bool useStreamLookupJoin, bool useColumnStore, bool useCBO = true) {
         auto kikimr = GetKikimrWithJoinSettings(useStreamLookupJoin, GetStatic(statsPath), useCBO);
+=======
+    std::pair<TString, std::vector<NYdb::TResultSet>> ExecuteJoinOrderTestGenericQueryWithStats(
+        const TString& queryPath,
+        const TString& statsPath,
+        bool useStreamLookupJoin,
+        bool useColumnStore,
+        bool useCBO = true,
+        const TExecuteParams& params = {}
+    ) {
+        auto kikimr = GetKikimrWithJoinSettings(useStreamLookupJoin, GetStatic(statsPath), useCBO, params);
+>>>>>>> 62f01135b35 ([KQP] Add ShuffleEliminated flag (#20175))
         kikimr.GetTestServer().GetRuntime()->GetAppData(0).FeatureFlags.SetEnableViews(true);
         auto db = kikimr.GetQueryClient();
         auto session = db.GetSession().GetValueSync().GetSession();
@@ -692,17 +715,27 @@ Y_UNIT_TEST_SUITE(KqpJoinOrder) {
         TVector<TString> RequestedLabels;
     };
 
+<<<<<<< HEAD
     Y_UNIT_TEST(ShuffleEliminationOneJoin) {
         auto plan = ExecuteJoinOrderTestGenericQueryWithStats("queries/shuffle_elimination_one_join.sql", "stats/tpch1000s.json", false, true, true);
+=======
+    Y_UNIT_TEST_TWIN(ShuffleEliminationOneJoin, EnableSeparationComputeActorsFromRead) {
+        auto [plan, _] = ExecuteJoinOrderTestGenericQueryWithStats("queries/shuffle_elimination_one_join.sql", "stats/tpch1000s.json", false, true, true, {.EnableSeparationComputeActorsFromRead = EnableSeparationComputeActorsFromRead});
+>>>>>>> 62f01135b35 ([KQP] Add ShuffleEliminated flag (#20175))
         auto joinFinder = TFindJoinWithLabels(plan);
         auto join = joinFinder.Find({"customer", "orders"});
         UNIT_ASSERT_C(join.Join == "InnerJoin (Grace)", join.Join);
         UNIT_ASSERT(!join.LhsShuffled);
         UNIT_ASSERT(join.RhsShuffled);
     }
 
+<<<<<<< HEAD
     Y_UNIT_TEST(ShuffleEliminationReuseShuffleTwoJoins) {
         auto plan = ExecuteJoinOrderTestGenericQueryWithStats("queries/shuffle_elimination_reuse_shuffle_two_joins.sql", "stats/tpch1000s.json", false, true, true);
+=======
+    Y_UNIT_TEST_TWIN(ShuffleEliminationReuseShuffleTwoJoins, EnableSeparationComputeActorsFromRead) {
+        auto [plan, _] = ExecuteJoinOrderTestGenericQueryWithStats("queries/shuffle_elimination_reuse_shuffle_two_joins.sql", "stats/tpch1000s.json", false, true, true, {.EnableSeparationComputeActorsFromRead = EnableSeparationComputeActorsFromRead});
+>>>>>>> 62f01135b35 ([KQP] Add ShuffleEliminated flag (#20175))
         auto joinFinder = TFindJoinWithLabels(plan);
 
         {