Include mvcc data in lsm stats

2 files changed, 53 insertions, 6 deletions

diff --git a/ydb/core/tablet_flat/flat_stat_part.h b/ydb/core/tablet_flat/flat_stat_part.h
index 6c01b3be570..14a07477cec 100644
--- a/ydb/core/tablet_flat/flat_stat_part.h
+++ b/ydb/core/tablet_flat/flat_stat_part.h
@@ -107,6 +107,9 @@ public:
         for (ui32 group : xrange(size_t(1), Part->Scheme->Groups.size())) {
             AltGroups.emplace_back(Part.Get(), NPage::TGroupId(group));
         }
+        for (ui32 group : xrange(Part->HistoricIndexes.size())) {
+            HistoryGroups.emplace_back(Part.Get(), NPage::TGroupId(group, true));
+        }
         FillKey();
     }
 
@@ -130,6 +133,26 @@ public:
                 ++g.Pos;
             }
         }
+        // Include mvcc data
+        if (!HistoryGroups.empty()) {
+            auto& h = HistoryGroups[0];
+            const auto& hscheme = Part->Scheme->HistoryGroup;
+            Y_VERIFY_DEBUG(hscheme.ColsKeyIdx.size() == 3);
+            while (h.Pos && h.Pos->Cell(hscheme.ColsKeyIdx[0]).AsValue<TRowId>() < nextRowId) {
+                // eagerly include all history up to the next row id
+                CurrentSize += GetPageSize(h.Pos->GetPageId(), h.GroupId);
+                ++h.Pos;
+            }
+            TRowId nextHistoryRowId = h.Pos ? h.Pos->GetRowId() : Max<TRowId>();
+            for (size_t index = 1; index < HistoryGroups.size(); ++index) {
+                auto& g = HistoryGroups[index];
+                while (g.Pos && g.Pos->GetRowId() < nextHistoryRowId) {
+                    // eagerly include all data up to the next row id
+                    CurrentSize += GetPageSize(g.Pos->GetPageId(), g.GroupId);
+                    ++g.Pos;
+                }
+            }
+        }
         FillKey();
     }
 
@@ -205,6 +228,7 @@ private:
     ui64 LastRowId = 0;
     ui64 LastSize = 0;
     TSmallVec<TGroupState> AltGroups;
+    TSmallVec<TGroupState> HistoryGroups;
 };
 
 // This iterator skipps pages that are screened. Currently the logic is simple:
diff --git a/ydb/core/tablet_flat/flat_table_part_ut.cpp b/ydb/core/tablet_flat/flat_table_part_ut.cpp
index 185265a95ef..4b83632d383 100644
--- a/ydb/core/tablet_flat/flat_table_part_ut.cpp
+++ b/ydb/core/tablet_flat/flat_table_part_ut.cpp
@@ -47,33 +47,55 @@ Y_UNIT_TEST_SUITE(TLegacy) {
             .Col(0, 0,  NScheme::NTypeIds::Uint64)
             .Col(0, 1,  NScheme::NTypeIds::Uint32)
             .Col(0, 2,  NScheme::NTypeIds::Uint32)
-            .Key({ 0, 1});
+            .Key({ 0, 1 });
 
         TPartCook cook(lay, { true, 4096 });
+        TPartCook vcook(lay, { true, 4096 });
 
         const ui64 X1 = 0, X2 = 3000;
 
         for (ui64 key1 = X1; key1 <= X2; key1++) {
-            for (ui32 key2 = 0; key2 < 1 + key1/1000; key2++)
+            for (ui32 key2 = 0; key2 < 1 + key1/1000; key2++) {
                 cook.AddN(key1, key2, key2);
+                for (int i = 0; i < 10; ++i) {
+                    vcook.Ver(TRowVersion(1000, 1000 - i)).AddN(key1, key2, key2 + i);
+                }
+            }
         }
 
         TPartEggs eggs = cook.Finish();
         UNIT_ASSERT_C(eggs.Parts.size() == 1,
             "Unexpected " << eggs.Parts.size() << " results");
-
-        auto fnIterate = [&dbgOut, &typeRegistry] (TIntrusiveConstPtr<TPartStore> part, TIntrusiveConstPtr<TRowScheme> scheme) {
+        TPartEggs veggs = vcook.Finish();
+        UNIT_ASSERT_C(veggs.Parts.size() == 1,
+            "Unexpected " << veggs.Parts.size() << " results");
+
+        auto fnIterate = [&dbgOut, &typeRegistry] (
+                TIntrusiveConstPtr<TPartStore> part,
+                TIntrusiveConstPtr<TRowScheme> scheme,
+                std::vector<ui64>& sizes)
+        {
             TPartIndexIterator idxIter(part, scheme->Keys);
+            sizes.clear();
 
             while (idxIter.IsValid()) {
                 TDbTupleRef key = idxIter.GetCurrentKey();
                 dbgOut << DbgPrintTuple(key, typeRegistry) << " " << idxIter.GetCurrentRowId() << " " << idxIter.GetCurrentDataSize() << Endl;
+                sizes.push_back(idxIter.GetCurrentDataSize());
                 idxIter.Next();
             }
         };
 
         dbgOut << "Iterate with the matching row scheme" << Endl;
-        fnIterate(eggs.At(0), eggs.Scheme);
+        std::vector<ui64> sizes;
+        fnIterate(eggs.At(0), eggs.Scheme, sizes);
+
+        dbgOut << "Iterate same data with versions" << Endl;
+        std::vector<ui64> vsizes;
+        fnIterate(veggs.At(0), veggs.Scheme, vsizes);
+
+        UNIT_ASSERT_C(vsizes.back() / sizes.back() >= 5,
+            "Expected to have 5-15x more bytes in versioned " << vsizes.back() << " vs unversioned " << sizes.back() << " part");
 
         // Add a column with default value to the key
         ui32 def10 = 121212;
@@ -86,7 +108,8 @@ Y_UNIT_TEST_SUITE(TLegacy) {
             .Key({ 0, 1, 10});
 
         dbgOut << "Iterate with added key column with default value" << Endl;
-        fnIterate(eggs.At(0), newLay.RowScheme());
+        std::vector<ui64> sizesWithDefaults;
+        fnIterate(eggs.At(0), newLay.RowScheme(), sizesWithDefaults);
     }
 
     Y_UNIT_TEST(ScreenedIndexIter) {