Pass compacted pages directly to Shared Cache (#11779)

Author: kunga

ydb/core/base/appdata.cpp

@@ -12,6 +12,7 @@
 
 #include <ydb/core/control/immediate_control_board_impl.h>
 #include <ydb/core/grpc_services/grpc_helper.h>
+#include <ydb/core/tablet_flat/shared_cache_pages.h>
 #include <ydb/core/protos/auth.pb.h>
 #include <ydb/core/protos/bootstrap.pb.h>
 #include <ydb/core/protos/blobstorage.pb.h>
@@ -93,6 +94,7 @@ TAppData::TAppData(
     , Mon(nullptr)
     , Icb(new TControlBoard())
     , InFlightLimiterRegistry(new NGRpcService::TInFlightLimiterRegistry(Icb))
+    , SharedCachePages(new NSharedCache::TSharedCachePages())
     , StreamingConfig(Impl->StreamingConfig)
     , PQConfig(Impl->PQConfig)
     , PQClusterDiscoveryConfig(Impl->PQClusterDiscoveryConfig)

ydb/core/base/appdata_fwd.h

@@ -19,6 +19,9 @@ namespace NKikimr {
     namespace NGRpcService {
         class TInFlightLimiterRegistry;
     }
+    namespace NSharedCache {
+        class TSharedCachePages;
+    }
 }
 
 namespace NKikimrCms {
@@ -189,6 +192,7 @@ struct TAppData {
     ::NMonitoring::TDynamicCounterPtr Counters;
     TIntrusivePtr<NKikimr::TControlBoard> Icb;
     TIntrusivePtr<NGRpcService::TInFlightLimiterRegistry> InFlightLimiterRegistry;
+    TIntrusivePtr<NSharedCache::TSharedCachePages> SharedCachePages;
 
     TIntrusivePtr<NInterconnect::TPollerThreads> PollerThreads;
 

ydb/core/tablet_flat/flat_executor.cpp

@@ -214,7 +214,9 @@ void TExecutor::RecreatePageCollectionsCache() noexcept
     for (const auto &it : Database->GetScheme().Tables) {
         auto subset = Database->Subset(it.first, NTable::TEpoch::Max(), { }, { });
 
-        for (auto &partView : subset->Flatten) AddCachesOfBundle(partView);
+        for (auto &partView : subset->Flatten) {
+            AddCachesOfBundle(partView);
+        }
     }
 
     if (TransactionWaitPads) {

ydb/core/tablet_flat/flat_ops_compact.h

@@ -13,6 +13,8 @@
 #include "flat_comp.h"
 #include "flat_executor_misc.h"
 #include "flat_bio_stats.h"
+#include "shared_cache_pages.h"
+#include "shared_sausagecache.h"
 #include "util_channel.h"
 
 #include <ydb/core/base/blobstorage.h>
@@ -102,6 +104,7 @@ namespace NTabletFlatExecutor {
 
             Spent = new TSpent(TAppData::TimeProvider.Get());
             Registry = AppData()->TypeRegistry;
+            SharedCachePages = AppData()->SharedCachePages.Get();
             Scheme = std::move(scheme);
             Driver = driver;
 
@@ -308,11 +311,39 @@ namespace NTabletFlatExecutor {
             auto *prod = new TProdCompact(!fail, Mask.Step(), std::move(Conf->Params),
                     std::move(YellowMoveChannels), std::move(YellowStopChannels));
 
+            if (fail) {
+                Results.clear(); /* shouldn't sent w/o fixation in bs */
+            }
+
             for (auto &result : Results) {
                 Y_ABORT_UNLESS(result.PageCollections, "Compaction produced a part without page collections");
+                TVector<TIntrusivePtr<NTable::TLoader::TCache>> pageCollections;
+                for (auto& pageCollection : result.PageCollections) {
+                    auto cache = MakeIntrusive<NTable::TLoader::TCache>(pageCollection.PageCollection);
+                    auto saveCompactedPages = MakeHolder<NSharedCache::TEvSaveCompactedPages>(pageCollection.PageCollection);
+                    auto gcList = SharedCachePages->GCList;
+                    auto addPage = [&saveCompactedPages, &pageCollection, &cache, &gcList](NPageCollection::TLoadedPage& loadedPage, bool sticky) {
+                        auto pageId = loadedPage.PageId;
+                        auto pageSize = pageCollection.PageCollection->Page(pageId).Size;
+                        auto sharedPage = MakeIntrusive<TPage>(pageId, pageSize, nullptr);
+                        sharedPage->Initialize(std::move(loadedPage.Data));
+                        saveCompactedPages->Pages.push_back(sharedPage);
+                        cache->Fill(pageId, TSharedPageRef::MakeUsed(std::move(sharedPage), gcList), sticky);
+                    };
+                    for (auto &page : pageCollection.StickyPages) {
+                        addPage(page, true);
+                    }
+                    for (auto &page : pageCollection.RegularPages) {
+                        addPage(page, false);
+                    }
+
+                    Send(MakeSharedPageCacheId(), saveCompactedPages.Release());
+
+                    pageCollections.push_back(std::move(cache));
+                }
 
                 NTable::TLoader loader(
-                    std::move(result.PageCollections),
+                    std::move(pageCollections),
                     { },
                     std::move(result.Overlay));
 
@@ -363,7 +394,7 @@ namespace NTabletFlatExecutor {
             }
 
             if (fail) {
-                prod->Results.clear(); /* shouldn't sent w/o fixation in bs */
+                Y_ABORT_IF(prod->Results); /* shouldn't sent w/o fixation in bs */
             } else if (bool(prod->Results) != bool(WriteStats.Rows > 0)) {
                 Y_ABORT("Unexpected rows production result after compaction");
             } else if ((bool(prod->Results) || bool(prod->TxStatus)) != bool(Blobs > 0)) {
@@ -525,6 +556,7 @@ namespace NTabletFlatExecutor {
         TVector<TBundle::TResult> Results;
         TVector<TIntrusiveConstPtr<NTable::TTxStatusPart>> TxStatus;
         const NScheme::TTypeRegistry * Registry = nullptr;
+        NSharedCache::TSharedCachePages * SharedCachePages;
 
         bool Finished = false;
         bool Failed = false;/* Failed to write blobs    */

ydb/core/tablet_flat/flat_sausage_writer.h

@@ -35,7 +35,9 @@ namespace NPageCollection {
                 Buffer.append(chunk.data(), chunk.size());
                 offset += piece;
 
-                if (Buffer.size() >= MaxBlobSize) Flush();
+                if (Buffer.size() >= MaxBlobSize) {
+                    Flush();
+                }
             }
 
             return Record.Push(type, body);

ydb/core/tablet_flat/flat_sausagecache.h

@@ -138,6 +138,10 @@ class TPrivatePageCache {
             EnsurePage(loaded.PageId)->Fill(std::move(loaded.Page), sticky);
         }
 
+        void Fill(ui32 pageId, TSharedPageRef page, bool sticky) noexcept {
+            EnsurePage(pageId)->Fill(std::move(page), sticky);
+        }
+
         const TLogoBlobID Id;
         const TIntrusiveConstPtr<NPageCollection::IPageCollection> PageCollection;
         TPageMap<THolder<TPage>> PageMap;

ydb/core/tablet_flat/flat_writer_blocks.h

@@ -18,14 +18,19 @@ namespace NWriter {
         using TPageId = NTable::NPage::TPageId;
         using TCache = TPrivatePageCache::TInfo;
 
+        struct TResult : TMoveOnly {
+            TIntrusiveConstPtr<NPageCollection::IPageCollection> PageCollection;
+            TVector<NPageCollection::TLoadedPage> RegularPages;
+            TVector<NPageCollection::TLoadedPage> StickyPages;
+        };
+
         TBlocks(ICone *cone, ui8 channel, ECache cache, ui32 block, bool stickyFlatIndex)
             : Cone(cone)
             , Channel(channel)
             , Cache(cache)
             , StickyFlatIndex(stickyFlatIndex)
             , Writer(Cone->CookieRange(1), Channel, block)
         {
-
         }
 
         ~TBlocks()
@@ -35,39 +40,39 @@ namespace NWriter {
 
         explicit operator bool() const noexcept
         {
-            return Writer || Regular || Sticky;
+            return Writer || Result.RegularPages || Result.StickyPages;
         }
 
-        TIntrusivePtr<TCache> Finish() noexcept
+        TResult Finish() noexcept
         {
-            TIntrusivePtr<TCache> pageCollection;
-
             if (auto meta = Writer.Finish(false /* omit empty page collection */)) {
-                for (auto &glob : Writer.Grab())
+                for (auto &glob : Writer.Grab()) {
                     Cone->Put(std::move(glob));
+                }
 
-                pageCollection = MakePageCollection(std::move(meta));
+                auto largeGlobId = CutToChunks(meta);
+                Result.PageCollection = MakeIntrusiveConst<NPageCollection::TPageCollection>(largeGlobId, std::move(meta));
             }
 
             Y_ABORT_UNLESS(!Writer, "Block writer is not empty after Finish");
-            Y_ABORT_UNLESS(!Regular && !Sticky, "Unexpected non-empty page lists");
 
-            return pageCollection;
+            return std::exchange(Result, {});
         }
 
         TPageId Write(TSharedData raw, EPage type)
         {
             auto pageId = Writer.AddPage(raw, (ui32)type);
 
-            for (auto &glob : Writer.Grab())
+            for (auto &glob : Writer.Grab()) {
                 Cone->Put(std::move(glob));
+            }
 
-            if (NTable::TLoader::NeedIn(type) || StickyFlatIndex && type == EPage::FlatIndex) {
+            if (NTable::TLoader::NeedIn(type) || Cache == ECache::Ever || StickyFlatIndex && type == EPage::FlatIndex) {
                 // Note: we mark flat index pages sticky after we load them
-                Sticky.emplace_back(pageId, std::move(raw));
+                Result.StickyPages.emplace_back(pageId, std::move(raw));
             } else if (bool(Cache) && type == EPage::DataPage || type == EPage::BTreeIndex) {
-                // Note: we save b-tree index pages to shared cache regardless of a cache mode  
-                Regular.emplace_back(pageId, std::move(raw));
+                // Note: we save b-tree index pages to shared cache regardless of a cache mode
+                Result.RegularPages.emplace_back(pageId, std::move(raw));
             }
 
             return pageId;
@@ -79,25 +84,6 @@ namespace NWriter {
         }
 
     private:
-        TIntrusivePtr<TCache> MakePageCollection(TSharedData body) noexcept
-        {
-            auto largeGlobId = CutToChunks(body);
-
-            auto *pack = new NPageCollection::TPageCollection(largeGlobId, std::move(body));
-
-            TIntrusivePtr<TCache> cache = new TCache(pack);
-
-            const bool sticky = (Cache == ECache::Ever);
-
-            for (auto &paged : Sticky) cache->Fill(paged, true);
-            for (auto &paged : Regular) cache->Fill(paged, sticky);
-
-            Sticky.clear();
-            Regular.clear();
-
-            return cache;
-        }
-
         NPageCollection::TLargeGlobId CutToChunks(TArrayRef<const char> body)
         {
             return Cone->Put(0, Channel, body, Writer.MaxBlobSize);
@@ -110,8 +96,7 @@ namespace NWriter {
         const bool StickyFlatIndex;
 
         NPageCollection::TWriter Writer;
-        TVector<NPageCollection::TLoadedPage> Regular;
-        TVector<NPageCollection::TLoadedPage> Sticky;
+        TResult Result;
     };
 }
 }

ydb/core/tablet_flat/flat_writer_bundle.h

@@ -14,9 +14,7 @@ namespace NWriter {
     class TBundle : public NTable::IPageWriter, protected ICone {
     public:
         struct TResult {
-            using TCache = TPrivatePageCache::TInfo;
-
-            TVector<TIntrusivePtr<TCache>> PageCollections;
+            TVector<TBlocks::TResult> PageCollections;
             TDeque<NTable::TScreen::THole> Growth;
             TString Overlay;
         };
@@ -100,8 +98,8 @@ namespace NWriter {
             auto &result = Results_.emplace_back();
 
             for (auto num : xrange(Blocks.size())) {
-                if (auto cache = Blocks[num]->Finish()) {
-                    result.PageCollections.emplace_back(std::move(cache));
+                if (auto written = Blocks[num]->Finish(); written.PageCollection) {
+                    result.PageCollections.emplace_back(std::move(written));
                 } else if (num < Blocks.size() - 1) {
                     Y_ABORT("Finish produced an empty main page collection");
                 }

ydb/core/tablet_flat/shared_cache_events.h

@@ -3,6 +3,7 @@
 #include "defs.h"
 #include "flat_bio_events.h"
 #include "shared_handle.h"
+#include "shared_page.h"
 #include <ydb/core/protos/shared_cache.pb.h>
 
 #include <util/generic/map.h>
@@ -20,6 +21,7 @@ namespace NKikimr::NSharedCache {
         EvUnregister,
         EvInvalidate,
         EvAttach,
+        EvSaveCompactedPages,
         EvRequest,
         EvResult,
         EvUpdated,
@@ -62,6 +64,19 @@ namespace NKikimr::NSharedCache {
         }
     };
 
+    // Note: compacted pages do not have an owner yet
+    // at first they should be accepted by an executor
+    // and it will send TEvAttach itself when it have happened
+    struct TEvSaveCompactedPages : public TEventLocal<TEvSaveCompactedPages, EvSaveCompactedPages> {
+        TIntrusiveConstPtr<NPageCollection::IPageCollection> PageCollection;
+        TVector<TIntrusivePtr<TPage>> Pages;
+
+        TEvSaveCompactedPages(TIntrusiveConstPtr<NPageCollection::IPageCollection> pageCollection)
+            : PageCollection(std::move(pageCollection))
+        {
+        }
+    };
+
     struct TEvRequest : public TEventLocal<TEvRequest, EvRequest> {
         const EPriority Priority;
         TAutoPtr<NPageCollection::TFetch> Fetch;

ydb/core/tablet_flat/shared_cache_pages.h

@@ -0,0 +1,12 @@
+#pragma once
+#include "defs.h"
+#include "shared_handle.h"
+
+namespace NKikimr::NSharedCache {
+
+class TSharedCachePages : public TThrRefBase {
+public:
+    TIntrusivePtr<TSharedPageGCList> GCList = new TSharedPageGCList;
+};
+
+}

ydb/core/tablet_flat/shared_handle.h

@@ -19,7 +19,7 @@ class TSharedPageHandle : public TThrRefBase {
     /**
      * Returns true if handle is initialized
      *
-     * Unitialized handle may not be shared with other threads.
+     * Uninitialized handle may not be shared with other threads.
      */
     bool IsInitialized() const noexcept {
         return Flags.load(std::memory_order_relaxed) != FlagsUninitialized;