KIKIMR-21403: add StaleRO followers usage test

Author: raydzast

ydb/core/kqp/ut/common/kqp_ut_common.cpp

@@ -1472,6 +1472,36 @@ void WaitForZeroReadIterators(Tests::TServer& server, const TString& path) {
     UNIT_ASSERT_C(iterators == 0, "Unable to wait for proper read iterator count, it looks like cancelation doesn`t work (" << iterators << ")");
 }
 
+int GetCumulativeCounterValue(Tests::TServer& server, const TString& path, const TString& counterName) {
+    int result = 0;
+
+    TTestActorRuntime* runtime = server.GetRuntime();
+    auto sender = runtime->AllocateEdgeActor();
+    auto shards = GetTableShards(&server, sender, path);
+    UNIT_ASSERT_C(shards.size() > 0, "Table: " << path << " has no shards");
+
+    for (auto x : shards) {
+        runtime->SendToPipe(
+            x,
+            sender,
+            new TEvTablet::TEvGetCounters,
+            0,
+            GetPipeConfigWithRetries());
+
+        auto ev = runtime->GrabEdgeEvent<TEvTablet::TEvGetCountersResponse>(sender);
+        UNIT_ASSERT(ev);
+
+        const NKikimrTabletBase::TEvGetCountersResponse& resp = ev->Get()->Record;
+        for (const auto& counter : resp.GetTabletCounters().GetAppCounters().GetCumulativeCounters()) {
+            if (counter.GetName() == counterName) {
+                result += counter.GetValue();
+            }
+        }
+    }
+
+    return result;
+}
+
 NJson::TJsonValue SimplifyPlan(NJson::TJsonValue& opt, const TGetPlanParams& params) {
     const auto& [_, nodeType] = *opt.GetMapSafe().find("Node Type");
     bool isShuffle = nodeType.GetStringSafe().find("HashShuffle") != TString::npos;

ydb/core/kqp/ut/common/kqp_ut_common.h

@@ -382,6 +382,7 @@ TVector<ui64> GetColumnTableShards(Tests::TServer* server, TActorId sender, cons
 
 void WaitForZeroSessions(const NKqp::TKqpCounters& counters);
 void WaitForZeroReadIterators(Tests::TServer& server, const TString& path);
+int GetCumulativeCounterValue(Tests::TServer& server, const TString& path, const TString& counterName);
 
 bool JoinOrderAndAlgosMatch(const TString& optimized, const TString& reference);
 

ydb/core/kqp/ut/opt/kqp_ne_ut.cpp

@@ -2307,6 +2307,11 @@ Y_UNIT_TEST_SUITE(KqpNewEngine) {
             SELECT * FROM FollowersKv WHERE Key = 21;
         )", TTxControl::BeginTx(TTxSettings::StaleRO()).CommitTx()).ExtractValueSync();
         AssertSuccessResult(result);
+        UNIT_ASSERT_UNEQUAL(0, GetCumulativeCounterValue(
+            kikimr.GetTestServer(),
+            "/Root/FollowersKv",
+            "DataShard/TxUpdateFollowerReadEdge/ExecuteCPUTime"
+        ));
 
         CompareYson(R"(
             [
@@ -2320,6 +2325,11 @@ Y_UNIT_TEST_SUITE(KqpNewEngine) {
             SELECT * FROM FollowersKv WHERE Value != "One" ORDER BY Key;
         )", TTxControl::BeginTx(TTxSettings::StaleRO()).CommitTx()).ExtractValueSync();
         AssertSuccessResult(result);
+        UNIT_ASSERT_UNEQUAL(0, GetCumulativeCounterValue(
+            kikimr.GetTestServer(),
+            "/Root/FollowersKv",
+            "DataShard/TxUpdateFollowerReadEdge/ExecuteCPUTime"
+        ));
 
         CompareYson(R"(
             [
@@ -2335,6 +2345,11 @@ Y_UNIT_TEST_SUITE(KqpNewEngine) {
             SELECT * FROM TwoShard WHERE Key = 2;
         )", TTxControl::BeginTx(TTxSettings::StaleRO()).CommitTx()).ExtractValueSync();
         AssertSuccessResult(result);
+        UNIT_ASSERT_EQUAL(0, GetCumulativeCounterValue(
+            kikimr.GetTestServer(),
+            "/Root/TwoShard",
+            "DataShard/TxUpdateFollowerReadEdge/ExecuteCPUTime"
+        ));
 
         CompareYson(R"(
             [
@@ -2355,6 +2370,11 @@ Y_UNIT_TEST_SUITE(KqpNewEngine) {
                 [[4000000001u];["BigOne"];[-1]]
             ]
         )", FormatResultSetYson(result.GetResultSet(0)));
+        UNIT_ASSERT_EQUAL(0, GetCumulativeCounterValue(
+            kikimr.GetTestServer(),
+            "/Root/TwoShard",
+            "DataShard/TxUpdateFollowerReadEdge/ExecuteCPUTime"
+        ));
     }
 
     Y_UNIT_TEST(StaleRO_Immediate) {
@@ -2379,6 +2399,66 @@ Y_UNIT_TEST_SUITE(KqpNewEngine) {
         )", FormatResultSetYson(result.GetResultSet(0)));
     }
 
+    Y_UNIT_TEST_TWIN(StaleRO_IndexFollowers, EnableFollowers) {
+        auto kikimr = DefaultKikimrRunner();
+        auto db = kikimr.GetTableClient();
+        auto session = db.CreateSession().GetValueSync().GetSession();
+
+        AssertSuccessResult(session.ExecuteSchemeQuery(R"(
+            --!syntax_v1
+            CREATE TABLE `KeySubkey` (
+                Key Uint64,
+                Subkey Uint64,
+                Value String,
+                Order Uint32,
+                PRIMARY KEY (Key, Subkey)
+            );
+
+            ALTER TABLE `KeySubkey` ADD INDEX `idx` GLOBAL SYNC ON (`Key`, `Order`) COVER (`Value`);
+        )").GetValueSync());
+
+        if constexpr (EnableFollowers) {
+            AssertSuccessResult(session.ExecuteSchemeQuery(R"(
+                --!syntax_v1
+                ALTER TABLE `KeySubkey` ALTER INDEX `idx` SET READ_REPLICAS_SETTINGS "PER_AZ:1";
+            )").GetValueSync());
+        }
+
+        AssertSuccessResult(session.ExecuteDataQuery(R"(
+            --!syntax_v1
+
+            REPLACE INTO `KeySubkey` (`Key`, `Subkey`, `Value`, `Order`) VALUES
+                (1u, 2u, "One", 7u),
+                (1u, 3u, "Two", 4u),
+                (21u, 8u, "Three", 1u),
+                (31u, 0u, "Four", 8u);
+        )", TTxControl::BeginTx().CommitTx()).GetValueSync());
+
+        auto result = session.ExecuteDataQuery(R"(
+            --!syntax_v1
+            SELECT * FROM `KeySubkey` VIEW `idx` WHERE Key = 1 ORDER BY `Order`;
+        )", TTxControl::BeginTx(TTxSettings::StaleRO()).CommitTx()).ExtractValueSync();
+        AssertSuccessResult(result);
+
+        const auto FollowerCpuTime = GetCumulativeCounterValue(
+            kikimr.GetTestServer(),
+            "/Root/KeySubkey/idx/indexImplTable",
+            "DataShard/TxUpdateFollowerReadEdge/ExecuteCPUTime"
+        );
+        if constexpr (EnableFollowers) {
+            UNIT_ASSERT_UNEQUAL(0, FollowerCpuTime);
+        } else {
+            UNIT_ASSERT_EQUAL(0, FollowerCpuTime);
+        }
+
+        CompareYson(R"(
+            [
+                [[1u];[4u];[3u];["Two"]];
+                [[1u];[7u];[2u];["One"]];
+            ]
+        )", FormatResultSetYson(result.GetResultSet(0)));
+    }
+
     Y_UNIT_TEST(ReadRangeWithParams) {
         auto kikimr = DefaultKikimrRunner();
         auto db = kikimr.GetTableClient();