[KQP] Remove the CA nodes from ParticipantNodes in KqpDataExecuter (#20714)

Author: dahbka-lis

ydb/core/kqp/executer_actor/kqp_executer_impl.h

@@ -526,11 +526,6 @@ class TKqpExecuterBase : public TActor<TDerived> {
         auto& state = ev->Get()->Record;
         ui64 taskId = state.GetTaskId();
 
-        ParticipantNodes.emplace(computeActor.NodeId());
-        if (TxManager) {
-            TxManager->AddParticipantNode(computeActor.NodeId());
-        }
-
         bool populateChannels = HandleComputeStats(ev);
 
         switch (state.GetState()) {
@@ -2311,7 +2306,7 @@ class TKqpExecuterBase : public TActor<TDerived> {
 
     ui32 StatementResultIndex;
 
-    // Track which nodes has been involved during execution
+    // Track which nodes (by shards) have been involved during execution
     THashSet<ui32> ParticipantNodes;
 
     bool AlreadyReplied = false;

ydb/core/kqp/ut/query/kqp_stats_ut.cpp

@@ -718,9 +718,10 @@ Y_UNIT_TEST_TWIN(OneShardLocalExec, UseSink) {
     UNIT_ASSERT_VALUES_EQUAL(counters.NonLocalSingleNodeReqCount->Val(), 0);
 }
 
-Y_UNIT_TEST_TWIN(OneShardNonLocalExec, UseSink) {
+Y_UNIT_TEST_QUAD(OneShardNonLocalExec, UseSink, EnableParallelPointReadConsolidation) {
     NKikimrConfig::TAppConfig app;
     app.MutableTableServiceConfig()->SetEnableOltpSink(UseSink);
+    app.MutableTableServiceConfig()->SetEnableParallelPointReadConsolidation(EnableParallelPointReadConsolidation);
     TKikimrRunner kikimr(TKikimrSettings().SetNodeCount(2).SetAppConfig(app));
     auto db = kikimr.GetTableClient();
     auto session = db.CreateSession().GetValueSync().GetSession();
@@ -791,89 +792,50 @@ Y_UNIT_TEST_TWIN(OneShardNonLocalExec, UseSink) {
         )", TTxControl::BeginTx().CommitTx()).ExtractValueSync();
         UNIT_ASSERT(result.IsSuccess());
 
-        if (app.GetTableServiceConfig().GetEnableParallelPointReadConsolidation()) {
-            // Session is on node 1, read task is for node 2 -> TotalSingleNodeReqCount does not increase
-            UNIT_ASSERT_VALUES_EQUAL(counters.TotalSingleNodeReqCount->Val(), expectedTotalSingleNodeReqCount);
-        } else {
-            UNIT_ASSERT_VALUES_EQUAL(counters.TotalSingleNodeReqCount->Val(), ++expectedTotalSingleNodeReqCount);
-        }
+        UNIT_ASSERT_VALUES_EQUAL(counters.TotalSingleNodeReqCount->Val(), ++expectedTotalSingleNodeReqCount);
     }
     {
         auto result = session.ExecuteDataQuery(R"(
             UPSERT INTO `/Root/EightShard` (Key, Data) VALUES (1, 1);
         )", TTxControl::BeginTx().CommitTx()).ExtractValueSync();
         UNIT_ASSERT(result.IsSuccess());
 
-        if (app.GetTableServiceConfig().GetEnableParallelPointReadConsolidation()) {
-            // If UseSink is enabled, compute is on node 1, write is on node 2 -> TotalSingleNodeReqCount does not increase
-            // If UseSink is disabled, the execution is on node 2 -> TotalSingleNodeReqCount increases
-            UNIT_ASSERT_VALUES_EQUAL(counters.TotalSingleNodeReqCount->Val(), expectedTotalSingleNodeReqCount += (UseSink) ? 0 : 1);
-        } else {
-            UNIT_ASSERT_VALUES_EQUAL(counters.TotalSingleNodeReqCount->Val(), ++expectedTotalSingleNodeReqCount);
-        }
+        UNIT_ASSERT_VALUES_EQUAL(counters.TotalSingleNodeReqCount->Val(), ++expectedTotalSingleNodeReqCount);
     }
     {
         auto result = kikimr.GetQueryClient().ExecuteQuery(R"(
             SELECT * FROM `/Root/EightShard` WHERE Key = 1;
         )", NYdb::NQuery::TTxControl::BeginTx().CommitTx()).ExtractValueSync();
         UNIT_ASSERT(result.IsSuccess());
 
-        if (app.GetTableServiceConfig().GetEnableParallelPointReadConsolidation()) {
-            // Session is on node 1, read task is for node 2 -> TotalSingleNodeReqCount does not increase
-            UNIT_ASSERT_VALUES_EQUAL(counters.TotalSingleNodeReqCount->Val(), expectedTotalSingleNodeReqCount);
-        } else {
-            UNIT_ASSERT_VALUES_EQUAL(counters.TotalSingleNodeReqCount->Val(), ++expectedTotalSingleNodeReqCount);
-        }
+        UNIT_ASSERT_VALUES_EQUAL(counters.TotalSingleNodeReqCount->Val(), ++expectedTotalSingleNodeReqCount);
     }
     {
         auto result = kikimr.GetQueryClient().ExecuteQuery(R"(
             UPSERT INTO `/Root/EightShard` (Key, Data) VALUES (1, 1);
         )", NYdb::NQuery::TTxControl::BeginTx().CommitTx()).ExtractValueSync();
         UNIT_ASSERT(result.IsSuccess());
 
-        if (app.GetTableServiceConfig().GetEnableParallelPointReadConsolidation()) {
-            // If UseSink is enabled, compute is on node 1, write is on node 2 -> TotalSingleNodeReqCount does not increase
-            // If UseSink is disabled, the execution is on node 2 -> TotalSingleNodeReqCount increases
-            UNIT_ASSERT_VALUES_EQUAL(counters.TotalSingleNodeReqCount->Val(), expectedTotalSingleNodeReqCount += (UseSink) ? 0 : 1);
-        } else {
-            UNIT_ASSERT_VALUES_EQUAL(counters.TotalSingleNodeReqCount->Val(), ++expectedTotalSingleNodeReqCount);
-        }
+        UNIT_ASSERT_VALUES_EQUAL(counters.TotalSingleNodeReqCount->Val(), ++expectedTotalSingleNodeReqCount);
     }
     {
         auto result = session.ExecuteDataQuery(R"(
             UPDATE `/Root/EightShard` SET Data = 111 WHERE Key = 1;
         )", TTxControl::BeginTx().CommitTx()).ExtractValueSync();
         UNIT_ASSERT(result.IsSuccess());
 
-        if (app.GetTableServiceConfig().GetEnableParallelPointReadConsolidation()) {
-            // Read is on node 1, write is on node 2 -> TotalSingleNodeReqCount does not increase
-            UNIT_ASSERT_VALUES_EQUAL(counters.TotalSingleNodeReqCount->Val(), expectedTotalSingleNodeReqCount);
-        } else {
-            UNIT_ASSERT_VALUES_EQUAL(counters.TotalSingleNodeReqCount->Val(), ++expectedTotalSingleNodeReqCount);
-        }
+        UNIT_ASSERT_VALUES_EQUAL(counters.TotalSingleNodeReqCount->Val(), ++expectedTotalSingleNodeReqCount);
     }
     {
         auto result = kikimr.GetQueryClient().ExecuteQuery(R"(
             UPDATE `/Root/EightShard` SET Data = 111 WHERE Key = 1;
         )", NYdb::NQuery::TTxControl::BeginTx().CommitTx()).ExtractValueSync();
         UNIT_ASSERT(result.IsSuccess());
 
-        if (app.GetTableServiceConfig().GetEnableParallelPointReadConsolidation()) {
-            // Read is on node 1, write is on node 2 -> TotalSingleNodeReqCount does not increase
-            UNIT_ASSERT_VALUES_EQUAL(counters.TotalSingleNodeReqCount->Val(), expectedTotalSingleNodeReqCount);
-        } else {
-            UNIT_ASSERT_VALUES_EQUAL(counters.TotalSingleNodeReqCount->Val(), ++expectedTotalSingleNodeReqCount);
-        }
-    }
-
-    if (app.GetTableServiceConfig().GetEnableParallelPointReadConsolidation()) {
-        // If UseSink is enabled, all requests are local or with two nodes -> NonLocalSingleNodeReqCount does not increase
-        // If UseSink is disabled, UPSERT queries are non-local and for the single node -> NonLocalSingleNodeReqCount increases
-        expectedNonLocalSingleNodeReqCount += (UseSink) ? 0 : 2;
-    } else {
-        expectedNonLocalSingleNodeReqCount += 6;
+        UNIT_ASSERT_VALUES_EQUAL(counters.TotalSingleNodeReqCount->Val(), ++expectedTotalSingleNodeReqCount);
     }
 
+    expectedNonLocalSingleNodeReqCount += 6;
     UNIT_ASSERT_VALUES_EQUAL(counters.NonLocalSingleNodeReqCount->Val(), expectedNonLocalSingleNodeReqCount);
 
     // Now resume node 1 and move all tablets on the node1