Add tests with reboots & fix bug on Retry() KIKIMR-18893

7 files changed, 229 insertions, 0 deletions

diff --git a/ydb/core/tx/datashard/change_sender_async_index.cpp b/ydb/core/tx/datashard/change_sender_async_index.cpp
index 08f75bc61bb..947160f0020 100644
--- a/ydb/core/tx/datashard/change_sender_async_index.cpp
+++ b/ydb/core/tx/datashard/change_sender_async_index.cpp
@@ -202,6 +202,10 @@ class TAsyncIndexChangeSenderShard: public TActorBootstrapped<TAsyncIndexChangeS
     }
 
     bool CanRetry() const {
+        if (CurrentStateFunc() != static_cast<TReceiveFunc>(&TThis::StateHandshake)) {
+            return false;
+        }
+
         return Attempt < MaxAttempts;
     }
 
diff --git a/ydb/core/tx/schemeshard/ut_index/CMakeLists.darwin-x86_64.txt b/ydb/core/tx/schemeshard/ut_index/CMakeLists.darwin-x86_64.txt
index 73b5067b407..ee0caeaea6d 100644
--- a/ydb/core/tx/schemeshard/ut_index/CMakeLists.darwin-x86_64.txt
+++ b/ydb/core/tx/schemeshard/ut_index/CMakeLists.darwin-x86_64.txt
@@ -20,8 +20,11 @@ target_link_libraries(ydb-core-tx-schemeshard-ut_index PUBLIC
   library-cpp-cpuid_check
   cpp-testing-unittest_main
   core-tx-schemeshard
+  ydb-core-base
+  ydb-core-scheme
   core-testlib-default
   tx-schemeshard-ut_helpers
+  lib-deprecated-kicli
 )
 target_link_options(ydb-core-tx-schemeshard-ut_index PRIVATE
   -Wl,-platform_version,macos,11.0,11.0
diff --git a/ydb/core/tx/schemeshard/ut_index/CMakeLists.linux-aarch64.txt b/ydb/core/tx/schemeshard/ut_index/CMakeLists.linux-aarch64.txt
index 6db7752a6aa..ea1ca422485 100644
--- a/ydb/core/tx/schemeshard/ut_index/CMakeLists.linux-aarch64.txt
+++ b/ydb/core/tx/schemeshard/ut_index/CMakeLists.linux-aarch64.txt
@@ -20,8 +20,11 @@ target_link_libraries(ydb-core-tx-schemeshard-ut_index PUBLIC
   yutil
   cpp-testing-unittest_main
   core-tx-schemeshard
+  ydb-core-base
+  ydb-core-scheme
   core-testlib-default
   tx-schemeshard-ut_helpers
+  lib-deprecated-kicli
 )
 target_link_options(ydb-core-tx-schemeshard-ut_index PRIVATE
   -ldl
diff --git a/ydb/core/tx/schemeshard/ut_index/CMakeLists.linux-x86_64.txt b/ydb/core/tx/schemeshard/ut_index/CMakeLists.linux-x86_64.txt
index b419d0a219d..4d107c9b580 100644
--- a/ydb/core/tx/schemeshard/ut_index/CMakeLists.linux-x86_64.txt
+++ b/ydb/core/tx/schemeshard/ut_index/CMakeLists.linux-x86_64.txt
@@ -21,8 +21,11 @@ target_link_libraries(ydb-core-tx-schemeshard-ut_index PUBLIC
   library-cpp-cpuid_check
   cpp-testing-unittest_main
   core-tx-schemeshard
+  ydb-core-base
+  ydb-core-scheme
   core-testlib-default
   tx-schemeshard-ut_helpers
+  lib-deprecated-kicli
 )
 target_link_options(ydb-core-tx-schemeshard-ut_index PRIVATE
   -ldl
diff --git a/ydb/core/tx/schemeshard/ut_index/CMakeLists.windows-x86_64.txt b/ydb/core/tx/schemeshard/ut_index/CMakeLists.windows-x86_64.txt
index 715928efb8e..d3e9a40103f 100644
--- a/ydb/core/tx/schemeshard/ut_index/CMakeLists.windows-x86_64.txt
+++ b/ydb/core/tx/schemeshard/ut_index/CMakeLists.windows-x86_64.txt
@@ -20,8 +20,11 @@ target_link_libraries(ydb-core-tx-schemeshard-ut_index PUBLIC
   library-cpp-cpuid_check
   cpp-testing-unittest_main
   core-tx-schemeshard
+  ydb-core-base
+  ydb-core-scheme
   core-testlib-default
   tx-schemeshard-ut_helpers
+  lib-deprecated-kicli
 )
 target_sources(ydb-core-tx-schemeshard-ut_index PRIVATE
   ${CMAKE_SOURCE_DIR}/ydb/core/tx/schemeshard/ut_index/ut_async_index.cpp
diff --git a/ydb/core/tx/schemeshard/ut_index/ut_async_index.cpp b/ydb/core/tx/schemeshard/ut_index/ut_async_index.cpp
index 7c22b595cad..e2e6bc130b8 100644
--- a/ydb/core/tx/schemeshard/ut_index/ut_async_index.cpp
+++ b/ydb/core/tx/schemeshard/ut_index/ut_async_index.cpp
@@ -1,5 +1,8 @@
+#include <ydb/core/base/path.h>
+#include <ydb/core/scheme/scheme_tablecell.h>
 #include <ydb/core/tx/schemeshard/ut_helpers/helpers.h>
 #include <ydb/core/testlib/tablet_helpers.h>
+#include <ydb/public/lib/deprecated/kicli/kicli.h>
 
 using namespace NKikimr;
 using namespace NSchemeShard;
@@ -51,4 +54,211 @@ Y_UNIT_TEST_SUITE(TAsyncIndexTests) {
         });
         env.TestWaitNotification(runtime, txId);
     }
+
+    THolder<TEvDataShard::TEvUploadRowsRequest> MakeUploadRows(ui64 tableId,
+            const TVector<ui32>& keyTags, const TVector<ui32>& valueTags,
+            TVector<std::pair<TString, TString>>&& serializedRows)
+    {
+        auto ev = MakeHolder<TEvDataShard::TEvUploadRowsRequest>();
+        ev->Record.SetTableId(tableId);
+
+        auto& scheme = *ev->Record.MutableRowScheme();
+        for (ui32 tag : keyTags) {
+            scheme.AddKeyColumnIds(tag);
+        }
+        for (ui32 tag : valueTags) {
+            scheme.AddValueColumnIds(tag);
+        }
+
+        for (const auto& [k, v] : serializedRows) {
+            auto& row = *ev->Record.AddRows();
+            row.SetKeyColumns(k);
+            row.SetValueColumns(v);
+        }
+
+        return ev;
+    }
+
+    TVector<ui64> Prepare(TTestActorRuntime& runtime, const TString& mainTablePath) {
+        ui64 mainTableId = 0;
+        TVector<ui64> mainTabletIds;
+        TVector<std::pair<TString, TString>> rows;
+
+        {
+            auto tableDesc = DescribePath(runtime, mainTablePath, true, true);
+            const auto& tablePartitions = tableDesc.GetPathDescription().GetTablePartitions();
+
+            mainTableId = tableDesc.GetPathId();
+            for (const auto& partition : tablePartitions) {
+                mainTabletIds.push_back(partition.GetDatashardId());
+            }
+        }
+
+        for (ui32 i = 0; i < 3; ++i) {
+            auto key = TVector<TCell>{TCell::Make(1 << i)};
+            auto value = TVector<TCell>{TCell::Make(i)};
+            rows.emplace_back(TSerializedCellVec::Serialize(key), TSerializedCellVec::Serialize(value));
+        }
+
+        auto ev = MakeUploadRows(mainTableId, {1}, {2}, std::move(rows));
+        ForwardToTablet(runtime, mainTabletIds[0], runtime.AllocateEdgeActor(), ev.Release());
+
+        return mainTabletIds;
+    }
+
+    NKikimrMiniKQL::TResult ReadTable(TTestActorRuntime& runtime, ui64 tabletId,
+            const TString& table, const TVector<TString>& pk, const TVector<TString>& columns)
+    {
+        TStringBuilder keyFmt;
+        for (const auto& k : pk) {
+            keyFmt << "'('" << k << " (Null) (Void)) ";
+        }
+        const auto columnsFmt = "'" + JoinSeq(" '", columns);
+
+        NKikimrMiniKQL::TResult result;
+        TString error;
+        NKikimrProto::EReplyStatus status = LocalMiniKQL(runtime, tabletId, Sprintf(R"((
+            (let range '(%s))
+            (let columns '(%s))
+            (let result (SelectRange '__user__%s range columns '()))
+            (return (AsList (SetResult 'Result result) ))
+        ))", keyFmt.data(), columnsFmt.data(), table.data()), result, error);
+        UNIT_ASSERT_VALUES_EQUAL_C(status, NKikimrProto::EReplyStatus::OK, error);
+
+        return result;
+    }
+
+    struct TTableTraits {
+        TString Path;
+        TVector<TString> Key;
+        TVector<TString> Columns;
+    };
+
+    void CheckWrittenToIndex(TTestActorRuntime& runtime, const TTableTraits& mainTable, const TTableTraits& indexTable) {
+        bool writtenToMainTable = false;
+        {
+            auto tableDesc = DescribePath(runtime, mainTable.Path, true, true);
+            const auto& tablePartitions = tableDesc.GetPathDescription().GetTablePartitions();
+            UNIT_ASSERT(!tablePartitions.empty());
+
+            auto result = ReadTable(runtime, tablePartitions[0].GetDatashardId(), SplitPath(mainTable.Path).back(),
+                mainTable.Key, mainTable.Columns);
+            auto value = NClient::TValue::Create(result);
+            writtenToMainTable = value["Result"]["List"].Size() == 3;
+        }
+
+        if (writtenToMainTable) {
+            auto tableDesc = DescribePrivatePath(runtime, indexTable.Path, true, true);
+            const auto& tablePartitions = tableDesc.GetPathDescription().GetTablePartitions();
+            UNIT_ASSERT(!tablePartitions.empty());
+
+            int i = 0;
+            while (++i < 10) {
+                runtime.SimulateSleep(TDuration::Seconds(1));
+
+                auto result = ReadTable(runtime, tablePartitions[0].GetDatashardId(), SplitPath(indexTable.Path).back(),
+                    indexTable.Key, indexTable.Columns);
+                auto value = NClient::TValue::Create(result);
+                if (value["Result"]["List"].Size() == 3) {
+                    break;
+                }
+            }
+
+            UNIT_ASSERT(i < 10);
+        }
+    }
+
+    Y_UNIT_TEST(SplitWithReboots) {
+        TTestWithReboots t;
+        t.Run([&](TTestActorRuntime& runtime, bool& activeZone) {
+            TVector<ui64> mainTabletIds;
+
+            {
+                TInactiveZone inactive(activeZone);
+
+                TestCreateIndexedTable(runtime, ++t.TxId, "/MyRoot", R"(
+                    TableDescription {
+                      Name: "Table"
+                      Columns { Name: "key" Type: "Uint32" }
+                      Columns { Name: "indexed" Type: "Uint32" }
+                      KeyColumnNames: ["key"]
+                    }
+                    IndexDescription {
+                      Name: "UserDefinedIndex"
+                      KeyColumnNames: ["indexed"]
+                      Type: EIndexTypeGlobalAsync
+                    }
+                )");
+                t.TestEnv->TestWaitNotification(runtime, t.TxId);
+
+                mainTabletIds = Prepare(runtime, "/MyRoot/Table");
+                UNIT_ASSERT_VALUES_EQUAL(mainTabletIds.size(), 1);
+            }
+
+            TestSplitTable(runtime, ++t.TxId, "/MyRoot/Table", Sprintf(R"(
+                SourceTabletId: %lu
+                SplitBoundary {
+                    KeyPrefix {
+                        Tuple { Optional { Uint32: 100 } }
+                    }
+                }
+            )", mainTabletIds[0]));
+            t.TestEnv->TestWaitNotification(runtime, t.TxId);
+
+            {
+                TInactiveZone inactive(activeZone);
+                CheckWrittenToIndex(runtime,
+                    {"/MyRoot/Table", {"key"}, {"key", "indexed"}},
+                    {"/MyRoot/Table/UserDefinedIndex/indexImplTable", {"indexed", "key"}, {"key", "indexed"}});
+            }
+        });
+    }
+
+    Y_UNIT_TEST(MergeWithReboots) {
+        TTestWithReboots t;
+        t.Run([&](TTestActorRuntime& runtime, bool& activeZone) {
+            TVector<ui64> mainTabletIds;
+
+            {
+                TInactiveZone inactive(activeZone);
+
+                TestCreateIndexedTable(runtime, ++t.TxId, "/MyRoot", R"(
+                    TableDescription {
+                      Name: "Table"
+                      Columns { Name: "key" Type: "Uint32" }
+                      Columns { Name: "indexed" Type: "Uint32" }
+                      KeyColumnNames: ["key"]
+                      UniformPartitionsCount: 2
+                      PartitionConfig {
+                        PartitioningPolicy {
+                          MinPartitionsCount: 1
+                        }
+                      }
+                    }
+                    IndexDescription {
+                      Name: "UserDefinedIndex"
+                      KeyColumnNames: ["indexed"]
+                      Type: EIndexTypeGlobalAsync
+                    }
+                )");
+                t.TestEnv->TestWaitNotification(runtime, t.TxId);
+
+                mainTabletIds = Prepare(runtime, "/MyRoot/Table");
+                UNIT_ASSERT_VALUES_EQUAL(mainTabletIds.size(), 2);
+            }
+
+            TestSplitTable(runtime, ++t.TxId, "/MyRoot/Table", Sprintf(R"(
+                SourceTabletId: %lu
+                SourceTabletId: %lu
+            )", mainTabletIds[0], mainTabletIds[1]));
+            t.TestEnv->TestWaitNotification(runtime, t.TxId);
+
+            {
+                TInactiveZone inactive(activeZone);
+                CheckWrittenToIndex(runtime,
+                    {"/MyRoot/Table", {"key"}, {"key", "indexed"}},
+                    {"/MyRoot/Table/UserDefinedIndex/indexImplTable", {"indexed", "key"}, {"key", "indexed"}});
+            }
+        });
+    }
 }
diff --git a/ydb/core/tx/schemeshard/ut_index/ya.make b/ydb/core/tx/schemeshard/ut_index/ya.make
index 45fa9a52ac0..c8e0c4f3ff9 100644
--- a/ydb/core/tx/schemeshard/ut_index/ya.make
+++ b/ydb/core/tx/schemeshard/ut_index/ya.make
@@ -14,8 +14,11 @@ ELSE()
 ENDIF()
 
 PEERDIR(
+    ydb/core/base
+    ydb/core/scheme
     ydb/core/testlib/default
     ydb/core/tx/schemeshard/ut_helpers
+    ydb/public/lib/deprecated/kicli
 )
 
 SRCS(