speed up register blob idx (#9581)

Author: ivanmorozov333

ydb/core/kqp/ut/olap/sys_view_ut.cpp

@@ -28,6 +28,7 @@ Y_UNIT_TEST_SUITE(KqpOlapSysView) {
         auto selectQuery = TString(R"(
             SELECT PathId, Kind, TabletId, Sum(Rows) as Rows
             FROM `/Root/olapStore/.sys/store_primary_index_portion_stats`
+            WHERE Activity == 1
             GROUP BY PathId, Kind, TabletId
             ORDER BY TabletId, Kind, PathId
         )");
@@ -61,13 +62,14 @@ Y_UNIT_TEST_SUITE(KqpOlapSysView) {
             WriteTestData(kikimr, "/Root/olapStore/olapTable_1", 0, 1000000 + i * 10000, 1000);
             WriteTestData(kikimr, "/Root/olapStore/olapTable_2", 0, 1000000 + i * 10000, 2000);
         }
-        csController->WaitCompactions(TDuration::Seconds(10));
+        csController->WaitCompactions(TDuration::Seconds(5));
 
         auto tableClient = kikimr.GetTableClient();
         {
             auto selectQuery = TString(R"(
                 SELECT PathId, Kind, TabletId
                 FROM `/Root/olapStore/olapTable_1/.sys/primary_index_stats`
+                WHERE Activity = 1
                 GROUP BY PathId, TabletId, Kind
                 ORDER BY PathId, TabletId, Kind
             )");
@@ -82,6 +84,7 @@ Y_UNIT_TEST_SUITE(KqpOlapSysView) {
             auto selectQuery = TString(R"(
                 SELECT PathId, Kind, TabletId
                 FROM `/Root/olapStore/olapTable_2/.sys/primary_index_stats`
+                WHERE Activity = 1
                 GROUP BY PathId, TabletId, Kind
                 ORDER BY PathId, TabletId, Kind
             )");
@@ -168,10 +171,10 @@ Y_UNIT_TEST_SUITE(KqpOlapSysView) {
             helper.ExecuteSchemeQuery("ALTER OBJECT `/Root/olapStore` (TYPE TABLESTORE) SET (ACTION=ALTER_COLUMN, NAME=field, `SERIALIZER.CLASS_NAME`=`ARROW_SERIALIZER`, `COMPRESSION.TYPE`=`zstd`);");
             csController->WaitCompactions(TDuration::Seconds(10));
         }
-        const ui64 rawBytesUnpack = rawBytesUnpack1PK - rawBytesPK1;
-        const ui64 bytesUnpack = bytesUnpack1PK - bytesPK1;
-        const ui64 rawBytesPack = rawBytesPackAndUnpack2PK - rawBytesUnpack1PK - rawBytesPK1;
-        const ui64 bytesPack = bytesPackAndUnpack2PK - bytesUnpack1PK - bytesPK1;
+        const i64 rawBytesUnpack = rawBytesUnpack1PK - rawBytesPK1;
+        const i64 bytesUnpack = bytesUnpack1PK - bytesPK1;
+        const i64 rawBytesPack = rawBytesPackAndUnpack2PK - rawBytesUnpack1PK - rawBytesPK1;
+        const i64 bytesPack = bytesPackAndUnpack2PK - bytesUnpack1PK - bytesPK1;
         TStringBuilder result;
         result << "unpacked data: " << rawBytesUnpack << " / " << bytesUnpack << Endl;
         result << "packed data: " << rawBytesPack << " / " << bytesPack << Endl;
@@ -292,6 +295,7 @@ Y_UNIT_TEST_SUITE(KqpOlapSysView) {
         ui64 rawBytes1;
         ui64 bytes1;
         auto csController = NYDBTest::TControllers::RegisterCSControllerGuard<NOlap::TWaitCompactionController>();
+        csController->SetSmallSizeDetector(Max<ui32>());
         auto settings = TKikimrSettings().SetWithSampleTables(false);
         TKikimrRunner kikimr(settings);
         Tests::NCommon::TLoggerInit(kikimr).Initialize();
@@ -308,6 +312,7 @@ Y_UNIT_TEST_SUITE(KqpOlapSysView) {
             helper.ExecuteSchemeQuery("ALTER OBJECT `/Root/olapStore` (TYPE TABLESTORE) SET (ACTION=UPSERT_INDEX, NAME=pk_int_max, TYPE=MAX, FEATURES=`{\"column_name\" : \"pk_int\"}`);");
             helper.ExecuteSchemeQuery("ALTER OBJECT `/Root/olapStore` (TYPE TABLESTORE) SET (ACTION=UPSERT_OPTIONS, SCHEME_NEED_ACTUALIZATION=`true`);");
             csController->WaitActualization(TDuration::Seconds(40));
+            csController->WaitCompactions(TDuration::Seconds(5));
             {
                 ui64 rawBytes2;
                 ui64 bytes2;
@@ -316,7 +321,7 @@ Y_UNIT_TEST_SUITE(KqpOlapSysView) {
                 AFL_VERIFY(bytes2 < bytes1 * 0.5)("f1", bytes1)("f2", bytes2);
                 std::vector<NJson::TJsonValue> stats;
                 helper.GetStats(stats, true);
-                AFL_VERIFY(stats.size() == 3);
+                AFL_VERIFY(stats.size() == 3)("count", stats.size());
                 for (auto&& i : stats) {
                     AFL_VERIFY(i.IsArray());
                     AFL_VERIFY(i.GetArraySafe().size() == 1);
@@ -396,6 +401,7 @@ Y_UNIT_TEST_SUITE(KqpOlapSysView) {
             auto selectQuery = TString(R"(
                 SELECT SUM(BlobRangeSize) as Bytes, SUM(Rows) as Rows, PathId, TabletId
                 FROM `/Root/olapStore/.sys/store_primary_index_stats`
+                WHERE Activity == 1
                 GROUP BY PathId, TabletId
                 ORDER BY Bytes
             )");
@@ -409,6 +415,7 @@ Y_UNIT_TEST_SUITE(KqpOlapSysView) {
             auto selectQuery = TString(R"(
                 SELECT Sum(Rows) as Rows, Kind, Sum(ColumnRawBytes) as RawBytes, PathId
                 FROM `/Root/olapStore/.sys/store_primary_index_portion_stats`
+                WHERE Activity == 1
                 GROUP BY Kind, PathId
                 ORDER BY PathId, Kind, Rows
             )");
@@ -529,6 +536,7 @@ Y_UNIT_TEST_SUITE(KqpOlapSysView) {
             auto selectQuery = TString(R"(
                 SELECT PathId, Kind, TabletId, Sum(BlobRangeSize) as Bytes
                 FROM `/Root/olapStore/.sys/store_primary_index_stats`
+                WHERE Activity == 1
                 GROUP BY PathId, Kind, TabletId
                 ORDER BY PathId, Kind, TabletId;
             )");
@@ -542,6 +550,7 @@ Y_UNIT_TEST_SUITE(KqpOlapSysView) {
             auto selectQuery = TString(R"(
                 SELECT PathId, Kind, TabletId, Sum(BlobRangeSize) as Bytes
                 FROM `/Root/olapStore/.sys/store_primary_index_stats`
+                WHERE Activity == 1
                 GROUP BY PathId, Kind, TabletId
                 ORDER BY PathId, Kind, TabletId;
             )");
@@ -569,6 +578,7 @@ Y_UNIT_TEST_SUITE(KqpOlapSysView) {
                 SELECT PathId, Kind, TabletId
                 FROM `/Root/olapStore/.sys/store_primary_index_stats`
                 WHERE Kind IN ('SPLIT_COMPACTED', 'INACTIVE', 'EVICTED', 'INSERTED')
+                AND Activity == 1
                 GROUP BY PathId, Kind, TabletId
                 ORDER BY PathId, Kind, TabletId;
             )");
@@ -700,6 +710,7 @@ Y_UNIT_TEST_SUITE(KqpOlapSysView) {
             auto selectQuery = TString(R"(
                 SELECT PathId, TabletId, Kind
                 FROM `/Root/olapStore/.sys/store_primary_index_stats`
+                WHERE Activity == 1
                 GROUP BY PathId, TabletId, Kind
             )");
 
@@ -715,6 +726,7 @@ Y_UNIT_TEST_SUITE(KqpOlapSysView) {
                     count(distinct(Kind)) as KindsCount,
                     count(distinct(TabletId)) as TabletsCount
                 FROM `/Root/olapStore/.sys/store_primary_index_stats`
+                WHERE Activity == 1
             )");
 
             auto rows = ExecuteScanQuery(tableClient, selectQuery);
@@ -727,6 +739,7 @@ Y_UNIT_TEST_SUITE(KqpOlapSysView) {
             auto selectQuery = TString(R"(
                 SELECT PathId, count(*), sum(Rows), sum(BlobRangeSize), sum(RawBytes)
                 FROM `/Root/olapStore/.sys/store_primary_index_stats`
+                WHERE Activity == 1
                 GROUP BY PathId
                 ORDER BY PathId
             )");

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;