Added test with page faults KIKIMR-14366

ref:dac449e3cea8b89e705ea5d787a6f483676d8eb6

9 files changed, 96 insertions, 45 deletions

diff --git a/ydb/core/testlib/test_client.cpp b/ydb/core/testlib/test_client.cpp
index d4907c26f1e..99cb332737f 100644
--- a/ydb/core/testlib/test_client.cpp
+++ b/ydb/core/testlib/test_client.cpp
@@ -204,7 +204,8 @@ namespace Tests {
         Runtime->SetupMonitoring();
         Runtime->SetLogBackend(Settings->LogBackend);
 
-        SetupTabletServices(*Runtime, &app, (StaticNodes() + DynamicNodes()) == 1 && Settings->EnableMockOnSingleNode, Settings->CustomDiskParams);
+        const bool mockDisk = (StaticNodes() + DynamicNodes()) == 1 && Settings->EnableMockOnSingleNode;
+        SetupTabletServices(*Runtime, &app, mockDisk, Settings->CustomDiskParams, Settings->CacheParams);
 
         CreateBootstrapTablets();
         SetupStorage();
diff --git a/ydb/core/testlib/test_client.h b/ydb/core/testlib/test_client.h
index 2064752ab45..0f1bfa7a46b 100644
--- a/ydb/core/testlib/test_client.h
+++ b/ydb/core/testlib/test_client.h
@@ -100,6 +100,7 @@ namespace Tests {
         ui32 NodeCount = 1;
         ui32 DynamicNodeCount = 0;
         NFake::TStorage CustomDiskParams;
+        NFake::TCaches CacheParams;
         TControls Controls;
         TAppPrepare::TFnReg FrFactory = &DefaultFrFactory;
         TIntrusivePtr<TFormatFactory> Formats;
@@ -136,6 +137,7 @@ namespace Tests {
         TServerSettings& SetNodeCount(ui32 value) { NodeCount = value; return *this; }
         TServerSettings& SetDynamicNodeCount(ui32 value) { DynamicNodeCount = value; return *this; }
         TServerSettings& SetCustomDiskParams(const NFake::TStorage& value) { CustomDiskParams = value; return *this; }
+        TServerSettings& SetCacheParams(const NFake::TCaches& value) { CacheParams = value; return *this; }
         TServerSettings& SetControls(const TControls& value) { Controls = value; return *this; }
         TServerSettings& SetFrFactory(const TAppPrepare::TFnReg& value) { FrFactory = value; return *this; }
         TServerSettings& SetEnableMockOnSingleNode(bool value) { EnableMockOnSingleNode = value; return *this; }
diff --git a/ydb/core/tx/datashard/datashard.h b/ydb/core/tx/datashard/datashard.h
index 5a93c035026..bc85865cd68 100644
--- a/ydb/core/tx/datashard/datashard.h
+++ b/ydb/core/tx/datashard/datashard.h
@@ -1377,6 +1377,9 @@ struct TEvDataShard {
             Record.MutablePathId()->SetOwnerId(ownerId);
             Record.MutablePathId()->SetLocalId(localId);
         }
+        TEvCompactTable(const TPathId& pathId)
+            : TEvCompactTable(pathId.OwnerId, pathId.LocalPathId)
+        { }
     };
 
     struct TEvCompactTableResult : public TEventPB<TEvCompactTableResult, NKikimrTxDataShard::TEvCompactTableResult,
diff --git a/ydb/core/tx/datashard/datashard_ut_background_compaction.cpp b/ydb/core/tx/datashard/datashard_ut_background_compaction.cpp
index 4872979bdf6..758508bd29e 100644
--- a/ydb/core/tx/datashard/datashard_ut_background_compaction.cpp
+++ b/ydb/core/tx/datashard/datashard_ut_background_compaction.cpp
@@ -20,15 +20,7 @@ NKikimrTxDataShard::TEvCompactTableResult CompactTable(
         ui64 tabletId,
         ui64 ownerId)
 {
-    auto &runtime = *server->GetRuntime();
-
-    auto sender = runtime.AllocateEdgeActor();
-    auto request = MakeHolder<TEvDataShard::TEvCompactTable>(ownerId, userTable.GetPathId());
-    runtime.SendToPipe(tabletId, sender, request.Release(), 0, GetPipeConfigWithRetries());
-
-    TAutoPtr<IEventHandle> handle;
-    auto response = runtime.GrabEdgeEventRethrow<TEvDataShard::TEvCompactTableResult>(handle);
-    return response->Record;
+    return CompactTable(*server->GetRuntime(), tabletId, TTableId(ownerId, userTable.GetPathId()));
 }
 
 } // namespace
diff --git a/ydb/core/tx/datashard/datashard_ut_change_collector.cpp b/ydb/core/tx/datashard/datashard_ut_change_collector.cpp
index cdce49ecdb0..5f3f8524b97 100644
--- a/ydb/core/tx/datashard/datashard_ut_change_collector.cpp
+++ b/ydb/core/tx/datashard/datashard_ut_change_collector.cpp
@@ -527,7 +527,8 @@ Y_UNIT_TEST_SUITE(AsyncIndexChangeCollector) {
             }},
         });
     }
-}
+
+} // AsyncIndexChangeCollector
 
 Y_UNIT_TEST_SUITE(CdcStreamChangeCollector) {
     using TCdcStream = TShardedTableOptions::TCdcStream;
@@ -538,7 +539,8 @@ Y_UNIT_TEST_SUITE(CdcStreamChangeCollector) {
         return pqConfig;
     }
 
-    void Run(const TString& path, const TShardedTableOptions& opts, const TVector<TCdcStream>& streams,
+    void Run(const NFake::TCaches& cacheParams, const TString& path,
+            const TShardedTableOptions& opts, const TVector<TCdcStream>& streams,
             const TVector<TString>& queries, const TStructRecords& expectedRecords)
     {
         const auto pathParts = SplitPath(path);
@@ -554,7 +556,8 @@ Y_UNIT_TEST_SUITE(CdcStreamChangeCollector) {
             .SetDomainName(domainName)
             .SetUseRealThreads(false)
             .SetEnableDataColumnForIndexTable(true)
-            .SetEnableAsyncIndexes(true);
+            .SetEnableAsyncIndexes(true)
+            .SetCacheParams(cacheParams);
 
         TServer::TPtr server = new TServer(serverSettings);
         auto& runtime = *server->GetRuntime();
@@ -579,13 +582,22 @@ Y_UNIT_TEST_SUITE(CdcStreamChangeCollector) {
             WaitTxNotification(server, sender, AsyncAlterAddStream(server, workingDir, tableName, stream));
         }
 
-        for (const auto& query : queries) {
-            ExecSQL(server, sender, query);
-        }
-
         auto desc = Navigate(runtime, sender, path);
         const auto& entry = desc->ResultSet.at(0);
 
+        const auto tabletIds = GetTableShards(server, sender, path);
+        UNIT_ASSERT_VALUES_EQUAL(tabletIds.size(), 1);
+
+        for (const auto& query : queries) {
+            if (query.StartsWith("COMPACT TABLE")) {
+                const auto result = CompactTable(runtime, tabletIds[0], entry.TableId);
+                UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), NKikimrTxDataShard::TEvCompactTableResult::OK);
+                RebootTablet(runtime, tabletIds[0], sender);
+            } else {
+                ExecSQL(server, sender, query);
+            }
+        }
+
         THashMap<NTable::TTag, TString> tagToName;
         for (const auto& [tag, column] : entry.Columns) {
             tagToName.emplace(tag, column.Name);
@@ -605,9 +617,6 @@ Y_UNIT_TEST_SUITE(CdcStreamChangeCollector) {
             streamPathIdToName.emplace(pathId, name);
         }
 
-        const auto tabletIds = GetTableShards(server, sender, path);
-        UNIT_ASSERT_VALUES_EQUAL(tabletIds.size(), 1);
-
         for (const auto& [pathId, actual] : GetChangeRecordsWithDetails(runtime, sender, tabletIds[0])) {
             TString name;
             if (streamPathIdToName.contains(pathId)) {
@@ -628,6 +637,22 @@ Y_UNIT_TEST_SUITE(CdcStreamChangeCollector) {
         }
     }
 
+    NFake::TCaches DefaultCacheParams() {
+        return {};
+    }
+
+    NFake::TCaches TinyCacheParams() {
+        auto params = DefaultCacheParams();
+        params.Shared = 1; // byte
+        return params;
+    }
+
+    void Run(const TString& path, const TShardedTableOptions& opts, const TVector<TCdcStream>& streams,
+            const TVector<TString>& queries, const TStructRecords& expectedRecords)
+    {
+        Run(DefaultCacheParams(), path, opts, streams, queries, expectedRecords);
+    }
+
     void Run(const TString& path, const TShardedTableOptions& opts, const TCdcStream& stream,
             const TString& query, const TStructRecords& expectedRecords)
     {
@@ -757,7 +782,30 @@ Y_UNIT_TEST_SUITE(CdcStreamChangeCollector) {
             {"updates_stream", {TStructRecord(NTable::ERowOp::Upsert, {{"pkey", 1}}, {{"ikey", 10}})}},
         });
     }
-}
+
+    TShardedTableOptions TinyCacheTable() {
+        return TShardedTableOptions()
+            .ExecutorCacheSize(1)
+            .Columns({
+                {"key", "Uint32", true, false},
+                {"value", "Uint32", false, false},
+            });
+    }
+
+    Y_UNIT_TEST(PageFaults) {
+        Run(TinyCacheParams(), "/Root/path", TinyCacheTable(), TVector<TCdcStream>{KeysOnly()}, TVector<TString>{
+            "UPSERT INTO `/Root/path` (key, value) VALUES (1, 10);",
+            "COMPACT TABLE `/Root/path`;",
+            "UPDATE `/Root/path` SET value = 20 WHERE key = 1;",
+        }, {
+            {"keys_stream", {
+                TStructRecord(NTable::ERowOp::Upsert, {{"key", 1}}),
+                TStructRecord(NTable::ERowOp::Upsert, {{"key", 1}}),
+            }},
+        });
+    }
+
+} // CdcStreamChangeCollector
 
 } // NKikimr
 
diff --git a/ydb/core/tx/datashard/datashard_ut_common.cpp b/ydb/core/tx/datashard/datashard_ut_common.cpp
index d754f7ad0e2..e40e8eee887 100644
--- a/ydb/core/tx/datashard/datashard_ut_common.cpp
+++ b/ydb/core/tx/datashard/datashard_ut_common.cpp
@@ -1168,6 +1168,10 @@ void CreateShardedTable(Tests::TServer::TPtr server,
             family.SetStorage(NKikimrSchemeOp::ColumnStorageTest_1_2_1k);
         }
 
+        if (opts.ExecutorCacheSize_) {
+            desc->MutablePartitionConfig()->SetExecutorCacheSize(*opts.ExecutorCacheSize_);
+        }
+
         runtime.Send(new IEventHandle(MakeTxProxyID(), sender, request.Release()));
         auto reply = runtime.GrabEdgeEventRethrow<TEvTxUserProxy::TEvProposeTransactionStatus>(handle);
         UNIT_ASSERT_VALUES_EQUAL(reply->Record.GetStatus(), TEvTxUserProxy::TEvProposeTransactionStatus::EStatus::ExecInProgress);
@@ -1228,6 +1232,18 @@ TVector<ui64> GetTableShards(Tests::TServer::TPtr server,
     return shards;
 }
 
+NKikimrTxDataShard::TEvCompactTableResult CompactTable(
+    TTestActorRuntime& runtime, ui64 shardId, const TTableId& tableId, bool compactBorrowed)
+{
+    auto sender = runtime.AllocateEdgeActor();
+    auto request = MakeHolder<TEvDataShard::TEvCompactTable>(tableId.PathId);
+    request->Record.SetCompactBorrowed(compactBorrowed);
+    runtime.SendToPipe(shardId, sender, request.Release(), 0, GetPipeConfigWithRetries());
+
+    auto ev = runtime.GrabEdgeEventRethrow<TEvDataShard::TEvCompactTableResult>(sender);
+    return ev->Get()->Record;
+}
+
 std::pair<TTableInfoMap, ui64> GetTables(
     Tests::TServer::TPtr server,
     ui64 tabletId)
diff --git a/ydb/core/tx/datashard/datashard_ut_common.h b/ydb/core/tx/datashard/datashard_ut_common.h
index 4b5d7fe4ea0..99ec4bc9799 100644
--- a/ydb/core/tx/datashard/datashard_ut_common.h
+++ b/ydb/core/tx/datashard/datashard_ut_common.h
@@ -417,6 +417,7 @@ struct TShardedTableOptions {
     TABLE_OPTION(ui64, Followers, 0);
     TABLE_OPTION(bool, FollowerPromotion, false);
     TABLE_OPTION(bool, ExternalStorage, false);
+    TABLE_OPTION(std::optional<ui64>, ExecutorCacheSize, std::nullopt);
 
 #undef TABLE_OPTION
 #undef TABLE_OPTION_IMPL
@@ -470,6 +471,9 @@ TVector<ui64> GetTableShards(Tests::TServer::TPtr server,
                              TActorId sender,
                              const TString &path);
 
+NKikimrTxDataShard::TEvCompactTableResult CompactTable(
+    TTestActorRuntime& runtime, ui64 shardId, const TTableId& tableId, bool compactBorrowed = false);
+
 using TTableInfoMap = THashMap<TString, NKikimrTxDataShard::TEvGetInfoResponse::TUserTable>;
 
 std::pair<TTableInfoMap, ui64> GetTables(Tests::TServer::TPtr server,
diff --git a/ydb/core/tx/datashard/datashard_ut_replication.cpp b/ydb/core/tx/datashard/datashard_ut_replication.cpp
index 8950e1a080e..54bae39cb55 100644
--- a/ydb/core/tx/datashard/datashard_ut_replication.cpp
+++ b/ydb/core/tx/datashard/datashard_ut_replication.cpp
@@ -82,17 +82,6 @@ Y_UNIT_TEST_SUITE(DataShardReplication) {
         }
     }
 
-    void CompactTable(TTestActorRuntime& runtime, ui64 shardId, const TTableId& tableId) {
-        auto sender = runtime.AllocateEdgeActor();
-        auto request = MakeHolder<TEvDataShard::TEvCompactTable>(tableId.PathId.OwnerId, tableId.PathId.LocalPathId);
-        request->Record.SetCompactBorrowed(true);
-        runtime.SendToPipe(shardId, sender, request.Release(), 0, GetPipeConfigWithRetries());
-
-        auto ev = runtime.GrabEdgeEventRethrow<TEvDataShard::TEvCompactTableResult>(sender);
-        // auto& result = ev->Get()->Record;
-        // UNIT_ASSERT(result.GetStatus() == NKikimrTxDataShard::TEvCompactTableResult::OK);
-    }
-
     void DoSplitMergeChanges(bool withMvcc, bool withReboots) {
         TPortManager pm;
         TServerSettings serverSettings(pm.GetPort(2134));
@@ -140,7 +129,7 @@ Y_UNIT_TEST_SUITE(DataShardReplication) {
 
         // Compact tables so we can merge them later
         for (ui64 shardId : shards1) {
-            CompactTable(runtime, shardId, tableId1);
+            CompactTable(runtime, shardId, tableId1, true);
         }
 
         if (withReboots) {
diff --git a/ydb/core/tx/datashard/datashard_ut_snapshot.cpp b/ydb/core/tx/datashard/datashard_ut_snapshot.cpp
index ca3bc867fa6..1182c04b338 100644
--- a/ydb/core/tx/datashard/datashard_ut_snapshot.cpp
+++ b/ydb/core/tx/datashard/datashard_ut_snapshot.cpp
@@ -885,16 +885,6 @@ Y_UNIT_TEST_SUITE(DataShardSnapshots) {
         }
     }
 
-    void CompactTable(TTestActorRuntime& runtime, ui64 shardId, const TTableId& tableId) {
-        auto sender = runtime.AllocateEdgeActor();
-        auto request = MakeHolder<TEvDataShard::TEvCompactTable>(tableId.PathId.OwnerId, tableId.PathId.LocalPathId);
-        runtime.SendToPipe(shardId, sender, request.Release(), 0, GetPipeConfigWithRetries());
-
-        auto ev = runtime.GrabEdgeEventRethrow<TEvDataShard::TEvCompactTableResult>(sender);
-        auto& result = ev->Get()->Record;
-        UNIT_ASSERT(result.GetStatus() == NKikimrTxDataShard::TEvCompactTableResult::OK);
-    }
-
     Y_UNIT_TEST(SwitchMvccSnapshots) {
         TPortManager pm;
         TServerSettings serverSettings(pm.GetPort(2134));
@@ -940,8 +930,14 @@ Y_UNIT_TEST_SUITE(DataShardSnapshots) {
 
         auto snapshot3 = CreateVolatileSnapshot(server, { "/Root/table-1", "/Root/table-2" }, TDuration::MilliSeconds(30000));
 
-        CompactTable(runtime, shards1.at(0), tableId1);
-        CompactTable(runtime, shards2.at(0), tableId2);
+        {
+            const auto result = CompactTable(runtime, shards1.at(0), tableId1);
+            UNIT_ASSERT(result.GetStatus() == NKikimrTxDataShard::TEvCompactTableResult::OK);
+        }
+        {
+            const auto result = CompactTable(runtime, shards2.at(0), tableId2);
+            UNIT_ASSERT(result.GetStatus() == NKikimrTxDataShard::TEvCompactTableResult::OK);
+        }
 
         // None of created snapshots should be removed
         auto removed1 = GetRemovedRowVersions(server, shards1.at(0));