YDB-6486 Implement external blob precharge in read iterator (#5758)

Closes #6486

Author: SammyVimes

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

@@ -1288,7 +1288,6 @@ THolder<NSchemeCache::TSchemeCacheNavigate> Navigate(TTestActorRuntime& runtime,
 {
     auto &runtime = *server->GetRuntime();
     TAutoPtr<IEventHandle> handle;
-    TVector<ui64> shards;
 
     auto request = MakeHolder<TEvTxUserProxy::TEvNavigate>();
     request->Record.MutableDescribePath()->SetPath(path);

ydb/core/scheme/scheme_tablecell.cpp

@@ -349,13 +349,7 @@ size_t TOwnedCellVecBatch::Append(TConstArrayRef<TCell> cells) {
         return 0;
     }
 
-    size_t size = sizeof(TCell) * cellsSize;
-    for (auto& cell : cells) {
-        if (!cell.IsNull() && !cell.IsInline()) {
-            const size_t cellSize = cell.Size();
-            size += AlignUp(cellSize);
-        }
-    }
+    size_t size = EstimateSize(cells);
 
     char * allocatedBuffer = reinterpret_cast<char *>(Pool->Allocate(size));
 

ydb/core/scheme/scheme_tablecell.h

@@ -169,6 +169,20 @@ struct TCell {
 static_assert(sizeof(TCell) == 12, "TCell must be 12 bytes");
 using TCellsRef = TConstArrayRef<const TCell>;
 
+inline size_t EstimateSize(TCellsRef cells) {
+    size_t cellsSize = cells.size();
+
+    size_t size = sizeof(TCell) * cellsSize;
+    for (auto& cell : cells) {
+        if (!cell.IsNull() && !cell.IsInline()) {
+            const size_t cellSize = cell.Size();
+            size += AlignUp(cellSize);
+        }
+    }
+    
+    return size;
+}
+
 inline int CompareCellsAsByteString(const TCell& a, const TCell& b, bool isDescending) {
     const char* pa = (const char*)a.Data();
     const char* pb = (const char*)b.Data();

ydb/core/tablet_flat/flat_executor_tx_env.h

@@ -29,7 +29,13 @@ namespace NTabletFlatExecutor {
         {
             auto *partStore = CheckedCast<const NTable::TPartStore*>(part);
 
-            return { true, Lookup(partStore->Locate(lob, ref), ref) };
+            const TSharedData* page = Lookup(partStore->Locate(lob, ref), ref);
+
+            if (!page && ReadMissingReferences) {
+                MissingReferencesSize_ += Max<ui64>(1, part->GetPageSize(lob, ref));
+            }
+
+            return { !ReadMissingReferences, page };
         }
 
         const TSharedData* TryGetPage(const TPart* part, TPageId page, TGroupId groupId) override
@@ -39,6 +45,20 @@ namespace NTabletFlatExecutor {
             return Lookup(partStore->PageCollections.at(groupId.Index).Get(), page);
         }
 
+        void EnableReadMissingReferences() noexcept {
+            ReadMissingReferences = true;
+        }
+
+        void DisableReadMissingReferences() noexcept {
+            ReadMissingReferences = false;
+            MissingReferencesSize_ = 0;
+        }
+
+        ui64 MissingReferencesSize() const noexcept
+        { 
+            return MissingReferencesSize_;
+        }
+
     private:
         const TSharedData* Lookup(TPrivatePageCache::TInfo *info, TPageId pageId) noexcept
         {
@@ -47,6 +67,11 @@ namespace NTabletFlatExecutor {
 
     public:
         TPrivatePageCache& Cache;
+    
+    private:
+        bool ReadMissingReferences = false;
+
+        ui64 MissingReferencesSize_ = 0;
     };
 
     struct TPageCollectionTxEnv : public TPageCollectionReadEnv, public IExecuting {
@@ -187,6 +212,20 @@ namespace NTabletFlatExecutor {
             LoanConfirmation.insert(std::make_pair(bundle, TLoanConfirmation{borrow}));
         }
 
+        void EnableReadMissingReferences() noexcept override
+        {
+            TPageCollectionReadEnv::EnableReadMissingReferences();
+        }
+
+        void DisableReadMissingReferences() noexcept override
+        {
+            TPageCollectionReadEnv::DisableReadMissingReferences();
+        }
+
+        ui64 MissingReferencesSize() const noexcept override
+        {
+            return TPageCollectionReadEnv::MissingReferencesSize();
+        }
     protected:
         NTable::TDatabase& DB;
 

ydb/core/tablet_flat/flat_part_iter.h

@@ -1319,6 +1319,7 @@ namespace NTable {
 
                 if (ref >> (sizeof(ui32) * 8))
                     Y_ABORT("Upper bits of ELargeObj ref now isn't used");
+
                 if (auto blob = Env->Locate(Part, ref, op)) {
                     const auto got = NPage::TLabelWrapper().Read(**blob);
 
@@ -1332,13 +1333,12 @@ namespace NTable {
                 } else {
                     Y_ABORT_UNLESS(ref < (*Part->Blobs)->size(), "out of blobs catalog");
 
+                    op = TCellOp(blob.Need ? ECellOp::Null : ECellOp(op), ELargeObj::GlobId);
+
                     /* Have to preserve reference to memory with TGlobId until
                         of next iterator alteration method invocation. This is
                         why here direct array of TGlobId is used.
                     */
-
-                    op = TCellOp(blob.Need ? ECellOp::Null : ECellOp(op), ELargeObj::GlobId);
-
                     row.Set(pin.To, op, TCell::Make((**Part->Blobs)[ref]));
                 }
             } else {

ydb/core/tablet_flat/flat_part_store.h

@@ -79,6 +79,13 @@ class TPartStore : public TPart, public IBundle {
         return PageCollections[groupId.Index]->GetPageSize(pageId);
     }
 
+    ui64 GetPageSize(ELargeObj lob, ui64 ref) const override
+    {
+        auto* cache = Locate(lob, ref);
+
+        return cache->PageCollection->Page(ref).Size;
+    }
+
     NPage::EPage GetPageType(NPage::TPageId pageId, NPage::TGroupId groupId) const override
     {
         Y_ABORT_UNLESS(groupId.Index < PageCollections.size());

ydb/core/tablet_flat/flat_table_part.h

@@ -156,6 +156,7 @@ namespace NTable {
         virtual ui64 DataSize() const = 0;
         virtual ui64 BackingSize() const = 0;
         virtual ui64 GetPageSize(NPage::TPageId pageId, NPage::TGroupId groupId) const = 0;
+        virtual ui64 GetPageSize(ELargeObj lob, ui64 ref) const = 0;
         virtual NPage::EPage GetPageType(NPage::TPageId pageId, NPage::TGroupId groupId) const = 0;
         virtual ui8 GetGroupChannel(NPage::TGroupId groupId) const = 0;
         virtual ui8 GetPageChannel(ELargeObj lob, ui64 ref) const = 0;

ydb/core/tablet_flat/tablet_flat_executor.h

@@ -103,6 +103,9 @@ struct IExecuting {
     virtual void LoanTable(ui32 tableId, const TString &partsInfo) = 0; // attach table parts to table (called on part destination)
     virtual void CleanupLoan(const TLogoBlobID &bundleId, ui64 from) = 0; // mark loan completion (called on part source)
     virtual void ConfirmLoan(const TLogoBlobID &bundleId, const TLogoBlobID &borrowId) = 0; // confirm loan update delivery (called on part destination)
+    virtual void EnableReadMissingReferences() noexcept = 0;
+    virtual void DisableReadMissingReferences() noexcept = 0;
+    virtual ui64 MissingReferencesSize() const noexcept = 0;
 };
 
 class TTxMemoryProviderBase : TNonCopyable {

ydb/core/tablet_flat/test/libs/table/test_part.h

@@ -52,6 +52,13 @@ namespace NTest {
             return Store->GetPageSize(groupId.Index, pageId);
         }
 
+        ui64 GetPageSize(ELargeObj lob, ui64 ref) const override
+        {
+            Y_UNUSED(lob);
+            Y_UNUSED(ref);
+            return 0;
+        }
+
         NPage::EPage GetPageType(NPage::TPageId pageId, NPage::TGroupId groupId) const override
         {
             return Store->GetPageType(groupId.Index, pageId);