Merge 3ddc5ea into 81da9ab

Author: kunga

ydb/core/tablet_flat/flat_fwd_cache.h

@@ -1,89 +1,115 @@
 #pragma once
 
 #include "flat_part_iface.h"
-#include "flat_part_forward.h"
 #include "flat_fwd_iface.h"
 #include "flat_fwd_misc.h"
 #include "flat_fwd_page.h"
+#include "flat_part_index_iter.h"
+#include "flat_part_index_iter_iface.h"
+#include "flat_table_part.h"
+#include "flat_part_slice.h"
 
 namespace NKikimr {
 namespace NTable {
 namespace NFwd {
 
-    class TCache : public IPageLoadingLogic {
-        using TGroupId = NPage::TGroupId;
-
-        template<size_t Items>
-        struct TRound {
-            const TSharedData* Get(TPageId pageId) const
-            {
-                if (pageId < Edge) {
-                    const auto pred = [pageId](const NPageCollection::TLoadedPage &page) {
-                        return page.PageId == pageId;
-                    };
-
-                    auto it = std::find_if(Pages.begin(), Pages.end(), pred);
-
-                    if (it == Pages.end()) {
-                        Y_ABORT("Failed to locate page within forward trace");
+    template<size_t Capacity>
+    class TLoadedPagesCircularBuffer {
+    public:
+        const TSharedData* Get(TPageId pageId) const
+        {
+            if (pageId < FirstUnseenPageId) {
+                for (const auto& page : LoadedPages) {
+                    if (page.PageId == pageId) {
+                        return &page.Data;
                     }
-
-                    return &it->Data;
                 }
 
-                return nullptr;
+                Y_ABORT("Failed to locate page within forward trace");
             }
 
-            ui32 Emplace(TPage &page)
-            {
-                Y_ABORT_UNLESS(page, "Cannot push invalid page to trace cache");
+            // next pages may be requested, ignore them
+            return nullptr;
+        }
+
+        // returns released data size
+        ui64 Emplace(TPage &page)
+        {
+            Y_ABORT_UNLESS(page, "Cannot push invalid page to trace cache");
 
-                Offset = (Pages.size() + Offset - 1) % Pages.size();
+            Offset = (Offset + 1) % Capacity;
 
-                const ui32 was = Pages[Offset].Data.size();
+            const ui64 releasedDataSize = LoadedPages[Offset].Data.size();
 
-                Pages[Offset].Data = page.Release();
-                Pages[Offset].PageId = page.PageId;
-                Edge = Max(Edge, page.PageId + 1);
+            LoadedPages[Offset].Data = page.Release();
+            LoadedPages[Offset].PageId = page.PageId;
+            FirstUnseenPageId = Max(FirstUnseenPageId, page.PageId + 1);
 
-                return was;
-            }
+            return releasedDataSize;
+        }
 
-        private:
-            std::array<NPageCollection::TLoadedPage, Items> Pages;
-            TPageId Edge = 0;
-            ui32 Offset = 0;
-        };
+    private:
+        std::array<NPageCollection::TLoadedPage, Capacity> LoadedPages;
+        ui32 Offset = 0;
+        TPageId FirstUnseenPageId = 0;
+    };
 
+    class TCache : public IPageLoadingLogic {
     public:
+        using TGroupId = NPage::TGroupId;
+
         TCache() = delete;
 
         TCache(const TPart* part, IPages* env, TGroupId groupId, const TIntrusiveConstPtr<TSlices>& bounds = nullptr)
-            : Index(part, env, groupId, 1, bounds)
-        { }
+            : Index(MakeHolder<TPartIndexIt>(part, env, groupId)) // TODO: use CreateIndexIter(part, env, groupId)
+        { 
+            if (bounds && !bounds->empty()) {
+                BeginRowId = bounds->front().BeginRowId();
+                EndRowId = bounds->back().EndRowId();
+            } else {
+                BeginRowId = 0;
+                EndRowId = Index->GetEndRowId();
+            }
+        }
 
         ~TCache()
         {
-            for (auto &it: Pages) it.Release();
+            for (auto &it: Pages) {
+                it.Release();
+            }
         }
 
         TResult Handle(IPageLoadingQueue *head, TPageId pageId, ui64 lower) noexcept override
         {
-            Y_ABORT_UNLESS(pageId != Max<TPageId>(), "Invalid requested pageId");
+            Y_ABORT_UNLESS(pageId != Max<TPageId>(), "Requested page is invalid");
 
-            if (auto *page = Trace.Get(pageId))
+            if (auto *page = Trace.Get(pageId)) {
                 return { page, false, true };
+            }
 
-            Rewind(pageId).Shrink(); /* points Offset to pageId */
+            DropPagesBefore(pageId);
+            Shrink();
+
+            bool grow = OnHold + OnFetch <= lower;
 
-            bool more = Grow && (OnHold + OnFetch <= lower);
+            if (Offset == Pages.size()) { // isn't processed yet
+                SyncIndex(pageId);
+                AddToQueue(head, pageId);
+            }
 
-            return { Preload(head, 0).Touch(pageId, Stat), more, true };
+            grow &= Index->IsValid() && Index->GetRowId() < EndRowId;
+
+            return {Pages.at(Offset).Touch(pageId, Stat), grow, true};
         }
 
         void Forward(IPageLoadingQueue *head, ui64 upper) noexcept override
         {
-            Preload(head, upper);
+            Y_ABORT_UNLESS(Started, "Couldn't be called before Handle returns grow");
+
+            while (OnHold + OnFetch < upper && Index->IsValid() && Index->GetRowId() < EndRowId) {
+                AddToQueue(head, Index->GetPageId());
+                Y_ABORT_UNLESS(Index->Next() != EReady::Page);
+            }
         }
 
         void Apply(TArrayRef<NPageCollection::TLoadedPage> loaded) noexcept override
@@ -105,7 +131,7 @@ namespace NFwd {
 
                 Stat.Saved += one.Data.size();
                 OnFetch -= one.Data.size();
-                OnHold += it->Settle(one);
+                OnHold += it->Settle(one); // settle of a dropped page returns 0 and releases its data
 
                 ++it;
             }
@@ -114,65 +140,72 @@ namespace NFwd {
         }
 
     private:
-        TPage& Preload(IPageLoadingQueue *head, ui64 upper) noexcept
+        void DropPagesBefore(TPageId pageId) noexcept
         {
-            auto until = [this, upper]() {
-                return OnHold + OnFetch < upper ? Max<TPageId>() : 0;
-            };
+            while (Offset < Pages.size()) {
+                auto &page = Pages.at(Offset);
 
-            while (auto more = Index.More(until())) {
-                auto size = head->AddToQueue(more, EPage::DataPage);
+                if (page.PageId >= pageId) {
+                    break;
+                }
 
-                Stat.Fetch += size;
-                OnFetch += size;
+                if (page.Size == 0) {
+                    Y_ABORT("Dropping page that has not been touched");
+                } else if (page.Usage == EUsage::Keep && page) {
+                    OnHold -= Trace.Emplace(page);
+                } else {
+                    OnHold -= page.Release().size();
+                    *(page.Ready() ? &Stat.After : &Stat.Before) += page.Size;
+                }
 
-                Pages.emplace_back(more, size, 0, Max<TPageId>());
-                Pages.back().Fetch = EFetch::Wait;
+                // keep pending pages but increment offset
+                Offset++;
             }
-
-            Grow = Grow && Index.On(true) < Max<TPageId>();
-
-            return Pages.at(Offset);
         }
 
-        TCache& Rewind(TPageId pageId) noexcept
+        void Shrink() noexcept
         {
-            while (auto drop = Index.Clean(pageId)) {
-                auto &page = Pages.at(Offset);
-
-                if (!Pages || page.PageId != drop.PageId) {
-                    Y_ABORT("Dropping page that is not exist in cache");
-                } else if (page.Size == 0) {
-                    Y_ABORT("Dropping page that has not been touched");
-                } else if (page.Usage == EUsage::Keep) {
-                    OnHold -= Trace.Emplace(page);
-                } else if (auto size = page.Release().size()) {
-                    OnHold -= size;
+            for (; Offset && Pages[0].Ready(); Offset--) {
+                Pages.pop_front();
+            }
+        }
 
-                    *(page.Ready() ? &Stat.After : &Stat.Before) += size;
-                }
+        void SyncIndex(TPageId pageId) noexcept
+        {
+            if (!Started) {
+                Y_ABORT_UNLESS(Index->Seek(BeginRowId) == EReady::Data);
+                Y_ABORT_UNLESS(Index->GetPageId() <= pageId, "Requested page is out of slice bounds");
+                Started = true;
+            }
 
-                Offset++;
+            while (Index->IsValid() && Index->GetPageId() < pageId) {
+                Y_ABORT_UNLESS(Index->Next() == EReady::Data);
+                Y_ABORT_UNLESS(Index->GetRowId() < EndRowId, "Requested page is out of slice bounds");
             }
 
-            return *this;
+            Y_ABORT_UNLESS(Index->GetPageId() == pageId, "Requested page doesn't belong to the part");
+            Y_ABORT_UNLESS(Index->Next() != EReady::Page);
         }
 
-        TCache& Shrink() noexcept
+        void AddToQueue(IPageLoadingQueue *head, TPageId pageId) noexcept
         {
-            for (; Offset && Pages[0].Ready(); Offset--)
-                Pages.pop_front();
+            auto size = head->AddToQueue(pageId, EPage::DataPage);
+
+            Stat.Fetch += size;
+            OnFetch += size;
 
-            return *this;
+            Y_ABORT_UNLESS(!Pages || Pages.back().PageId < pageId);
+            Pages.emplace_back(pageId, size, 0, Max<TPageId>());
+            Pages.back().Fetch = EFetch::Wait;
         }
 
     private:
-        bool Grow = true;       /* Have some pages for Forward(...) */
-        TForward Index;
-        TRound<TPart::Trace> Trace;
+        THolder<IIndexIter> Index; /* Points on next to load page */
+        bool Started = false;
+        TRowId BeginRowId, EndRowId;
+        TLoadedPagesCircularBuffer<TPart::Trace> Trace;
 
         /*_ Forward cache line state */
-
         ui64 OnHold = 0;
         ui64 OnFetch = 0;
         ui32 Offset = 0;

ydb/core/tablet_flat/flat_fwd_env.h

@@ -10,10 +10,8 @@
 #include "flat_table_subset.h"
 #include "flat_part_iface.h"
 #include "flat_part_store.h"
-#include "flat_sausage_packet.h"
 #include "flat_sausage_fetch.h"
 #include "util_fmt_abort.h"
-#include "util_basics.h"
 #include <util/generic/deque.h>
 #include <util/random/random.h>
 #include <util/generic/intrlist.h>

ydb/core/tablet_flat/flat_fwd_misc.cpp

@@ -0,0 +1,5 @@
+#include "flat_fwd_misc.h"
+
+Y_DECLARE_OUT_SPEC(, NKikimr::NTable::NFwd::TStat, stream, value) {
+    value.Describe(stream);
+}

ydb/core/tablet_flat/flat_fwd_misc.h

@@ -18,11 +18,11 @@ namespace NFwd {
         {
             out
                 << "TFwd{"
-                << Fetch << "b"
-                << " > " << Saved << "b"
-                << " > " << Usage << "b"
-                << " +" << After << "b"
-                << " ~" << Before << "b"
+                << "fetch=" << Fetch
+                << ",saved=" << Saved
+                << ",usage=" << Usage
+                << ",after=" << After
+                << ",before=" << Before
                 << "}";
         }
 
@@ -37,6 +37,8 @@ namespace NFwd {
             return *this;
         }
 
+        auto operator<=>(const TStat&) const = default;
+
         ui64 Fetch = 0;     /* Requested to load by cache       */
         ui64 Saved = 0;     /* Obtained by cache with DoSave()  */
         ui64 Usage = 0;     /* Actually was used by client      */

ydb/core/tablet_flat/flat_fwd_page.h

@@ -75,7 +75,7 @@ namespace NFwd {
         const TSharedData* Touch(TPageId pageId, TStat &stat) noexcept
         {
             if (PageId != pageId || (!Data && Fetch == EFetch::Done)) {
-                Y_ABORT("Touching page thatd doesn't fits to this action");
+                Y_ABORT("Touching page that doesn't fit to this action");
             } else {
                 auto to = Fetch == EFetch::None ? EUsage::Seen : EUsage::Keep;
 

ydb/core/tablet_flat/flat_page_other.h

@@ -28,7 +28,7 @@ namespace NPage {
             Tags.resize(TagsCount + (TagsCount & 1), 0);
             Cook.reserve(tags);
 
-            Y_ABORT_UNLESS(tags <= ui32(-Min<i16>()), "Too many columnt tags");
+            Y_ABORT_UNLESS(tags <= ui32(-Min<i16>()), "Too many column tags");
         }
 
         void Put(TRowId row, ui16 tag, ui32 bytes) noexcept