Merge aa8bba0 into 34e573c

Author: ivanmorozov333

ydb/core/tx/columnshard/engines/portions/constructor.cpp

@@ -31,6 +31,45 @@ TPortionInfo TPortionInfoConstructor::Build(const bool needChunksNormalization)
     NActors::TLogContextGuard lGuard = NActors::TLogContextBuilder::Build()("portion_id", GetPortionIdVerified());
     FullValidation();
 
+    if (BlobIdxs.size()) {
+        auto itRecord = Records.begin();
+        auto itIndex = Indexes.begin();
+        auto itBlobIdx = BlobIdxs.begin();
+        while (itRecord != Records.end() && itIndex != Indexes.end() && itBlobIdx != BlobIdxs.end()) {
+            if (itRecord->GetAddress() < itIndex->GetAddress()) {
+                AFL_VERIFY(itRecord->GetAddress() == itBlobIdx->GetAddress());
+                itRecord->RegisterBlobIdx(itBlobIdx->GetBlobIdx());
+                ++itRecord;
+                ++itBlobIdx;
+            } else if (itIndex->GetAddress() < itRecord->GetAddress()) {
+                if (itIndex->HasBlobData()) {
+                    ++itIndex;
+                    continue;
+                }
+                AFL_VERIFY(itIndex->GetAddress() == itBlobIdx->GetAddress());
+                itIndex->RegisterBlobIdx(itBlobIdx->GetBlobIdx());
+                ++itIndex;
+                ++itBlobIdx;
+            } else {
+                AFL_VERIFY(false);
+            }
+        }
+        for (; itRecord != Records.end() && itBlobIdx != BlobIdxs.end(); ++itRecord, ++itBlobIdx) {
+            AFL_VERIFY(itRecord->GetAddress() == itBlobIdx->GetAddress());
+            itRecord->RegisterBlobIdx(itBlobIdx->GetBlobIdx());
+        }
+        for (; itIndex != Indexes.end() && itBlobIdx != BlobIdxs.end(); ++itIndex) {
+            if (itIndex->HasBlobData()) {
+                continue;
+            }
+            AFL_VERIFY(itIndex->GetAddress() == itBlobIdx->GetAddress());
+            itIndex->RegisterBlobIdx(itBlobIdx->GetBlobIdx());
+            ++itBlobIdx;
+        }
+        AFL_VERIFY(itRecord == Records.end());
+        AFL_VERIFY(itBlobIdx == BlobIdxs.end());
+    }
+
     result.Indexes = Indexes;
     result.Records = Records;
     result.BlobIds = BlobIds;

ydb/core/tx/columnshard/engines/portions/constructor.h

@@ -29,6 +29,26 @@ class TPortionInfoConstructor {
     YDB_ACCESSOR_DEF(std::vector<TColumnRecord>, Records);
     std::vector<TUnifiedBlobId> BlobIds;
 
+    class TAddressBlobId {
+    private:
+        TChunkAddress Address;
+        YDB_READONLY(TBlobRangeLink16::TLinkId, BlobIdx, 0);
+
+    public:
+        const TChunkAddress& GetAddress() const {
+            return Address;
+        }
+
+        TAddressBlobId(const TChunkAddress& address, const TBlobRangeLink16::TLinkId blobIdx)
+            : Address(address)
+            , BlobIdx(blobIdx)
+        {
+
+        }
+    };
+    std::vector<TAddressBlobId> BlobIdxs;
+    bool NeedBlobIdxsSort = false;
+
 public:
     void SetPortionId(const ui64 value) {
         AFL_VERIFY(value);
@@ -216,6 +236,16 @@ class TPortionInfoConstructor {
         return linkRange.RestoreRange(GetBlobId(linkRange.GetBlobIdxVerified()));
     }
 
+    const TBlobRange RestoreBlobRangeSlow(const TBlobRangeLink16& linkRange, const TChunkAddress& address) const {
+        for (auto&& i : BlobIdxs) {
+            if (i.GetAddress() == address) {
+                return linkRange.RestoreRange(GetBlobId(i.GetBlobIdx()));
+            }
+        }
+        AFL_VERIFY(false);
+        return TBlobRange();
+    }
+
     const TUnifiedBlobId& GetBlobId(const TBlobRangeLink16::TLinkId linkId) const {
         AFL_VERIFY(linkId < BlobIds.size());
         return BlobIds[linkId];
@@ -226,19 +256,10 @@ class TPortionInfoConstructor {
     }
 
     void RegisterBlobIdx(const TChunkAddress& address, const TBlobRangeLink16::TLinkId blobIdx) {
-        for (auto&& i : Records) {
-            if (i.GetColumnId() == address.GetEntityId() && i.GetChunkIdx() == address.GetChunkIdx()) {
-                i.RegisterBlobIdx(blobIdx);
-                return;
-            }
+        if (BlobIdxs.size() && address < BlobIdxs.back().GetAddress()) {
+            NeedBlobIdxsSort = true;
         }
-        for (auto&& i : Indexes) {
-            if (i.GetIndexId() == address.GetEntityId() && i.GetChunkIdx() == address.GetChunkIdx()) {
-                i.RegisterBlobIdx(blobIdx);
-                return;
-            }
-        }
-        AFL_VERIFY(false)("problem", "portion haven't address for blob registration")("address", address.DebugString());
+        BlobIdxs.emplace_back(address, blobIdx);
     }
 
     TString DebugString() const {
@@ -265,26 +286,37 @@ class TPortionInfoConstructor {
             std::sort(Indexes.begin(), Indexes.end(), pred);
             CheckChunksOrder(Indexes);
         }
+        if (NeedBlobIdxsSort) {
+            auto pred = [](const TAddressBlobId& l, const TAddressBlobId& r) {
+                return l.GetAddress() < r.GetAddress();
+            };
+            std::sort(BlobIdxs.begin(), BlobIdxs.end(), pred);
+        }
     }
 
     void FullValidation() const {
         AFL_VERIFY(Records.size());
         CheckChunksOrder(Records);
         CheckChunksOrder(Indexes);
-        std::set<ui32> blobIdxs;
-        for (auto&& i : Records) {
-            blobIdxs.emplace(i.GetBlobRange().GetBlobIdxVerified());
-        }
-        for (auto&& i : Indexes) {
-            if (i.HasBlobRange()) {
-                blobIdxs.emplace(i.GetBlobRangeVerified().GetBlobIdxVerified());
-            }
-        }
-        if (BlobIds.size()) {
-            AFL_VERIFY(BlobIds.size() == blobIdxs.size());
-            AFL_VERIFY(BlobIds.size() == *blobIdxs.rbegin() + 1);
+        if (BlobIdxs.size()) {
+            AFL_VERIFY(BlobIdxs.size() <= Records.size() + Indexes.size())("blobs", BlobIdxs.size())("records", Records.size())(
+                                                               "indexes", Indexes.size());
         } else {
-            AFL_VERIFY(blobIdxs.empty());
+            std::set<ui32> blobIdxs;
+            for (auto&& i : Records) {
+                blobIdxs.emplace(i.GetBlobRange().GetBlobIdxVerified());
+            }
+            for (auto&& i : Indexes) {
+                if (i.HasBlobRange()) {
+                    blobIdxs.emplace(i.GetBlobRangeVerified().GetBlobIdxVerified());
+                }
+            }
+            if (BlobIds.size()) {
+                AFL_VERIFY(BlobIds.size() == blobIdxs.size());
+                AFL_VERIFY(BlobIds.size() == *blobIdxs.rbegin() + 1);
+            } else {
+                AFL_VERIFY(blobIdxs.empty());
+            }
         }
     }
 

ydb/core/tx/columnshard/engines/ut/ut_logs_engine.cpp

@@ -284,7 +284,7 @@ void AddIdsToBlobs(std::vector<TWritePortionInfoWithBlobsResult>& portions, NBlo
             blobsData.emplace(blobId, b.GetResultBlob());
         }
         for (auto&& rec : portion.GetPortionConstructor().GetRecords()) {
-            auto range = portion.GetPortionConstructor().RestoreBlobRange(rec.BlobRange);
+            auto range = portion.GetPortionConstructor().RestoreBlobRangeSlow(rec.BlobRange, rec.GetAddress());
             auto it = blobsData.find(range.BlobId);
             AFL_VERIFY(it != blobsData.end());
             const TString& data = it->second;