BTreeIndex Test Charge Iter consistency (#1869)

7 files changed, 246 insertions, 280 deletions

diff --git a/ydb/core/tablet_flat/benchmark/b_charge.cpp b/ydb/core/tablet_flat/benchmark/b_charge.cpp
deleted file mode 100644
index da1c06dbbd5..00000000000
--- a/ydb/core/tablet_flat/benchmark/b_charge.cpp
+++ /dev/null
@@ -1,153 +0,0 @@
-#include <benchmark/benchmark.h>
-
-#include <ydb/core/tablet_flat/flat_row_celled.h>
-#include <ydb/core/tablet_flat/flat_part_charge_range.h>
-#include <ydb/core/tablet_flat/flat_part_charge_create.h>
-#include <ydb/core/tablet_flat/test/libs/rows/cook.h>
-#include <ydb/core/tablet_flat/test/libs/rows/tool.h>
-#include <ydb/core/tablet_flat/test/libs/table/model/large.h>
-#include <ydb/core/tablet_flat/test/libs/table/test_writer.h>
-#include <ydb/core/tablet_flat/test/libs/table/test_envs.h>
-#include <ydb/core/tablet_flat/test/libs/table/wrap_part.h>
-#include <ydb/core/tablet_flat/test/libs/table/test_steps.h>
-
-#include <library/cpp/testing/unittest/registar.h>
-
-namespace NKikimr {
-namespace NTable {
-
-namespace {
-    const NTest::TMass Mass(new NTest::TModelStd(true), 2*1000);
-
-    struct TTouchEnv : public NTest::TTestEnv {
-        TTouchEnv(bool fail) : Fail(fail) { }
-
-        const TSharedData* TryGetPage(const TPart *part, TPageId id, TGroupId groupId) override
-        {
-            TouchedCount++;
-            return Fail ? nullptr : NTest::TTestEnv::TryGetPage(part, id, groupId);
-        }
-
-        const bool Fail = false;
-        ui64 TouchedCount = 0;
-    };
-
-    struct TPrechargeFixture : public benchmark::Fixture {
-        using TGroupId = NPage::TGroupId;
-
-        TPrechargeFixture()
-            : Tool(*Mass.Model->Scheme)
-        {
-            Y_ABORT_UNLESS(NTest::IndexTools::CountMainPages(*Eggs.Lone()) > 120);
-        }
-
-        static NTest::TPartEggs MakeEggs() noexcept
-        {
-            NPage::TConf conf{ true, 8192 };
-
-            auto groups = Mass.Model->Scheme->Families.size();
-            for (size_t group : xrange(groups)) {
-                conf.Group(group).PageRows = 10;
-            }
-            conf.Group(1).PageRows = 5;
-            conf.Group(2).PageRows = 2;
-
-            NTest::TPartCook cook(Mass.Model->Scheme, conf);
-
-            for (auto seq: xrange(Mass.Saved.Size())) {
-                // fill with random keys
-                if (seq % 3 != 0) cook.Add(Mass.Saved[seq]);
-            }
-
-            return cook.Finish();
-        }
-
-        void SetUp(const ::benchmark::State& state) 
-        {
-            bool fail = state.range(1);
-            ui32 groups = state.range(2);
-
-            Env = MakeHolder<TTouchEnv>(fail);
-
-            const auto &keyDefaults = *Tool.Scheme.Keys;
-            
-            Run = MakeHolder<TRun>(keyDefaults);
-
-            auto part = Eggs.Lone();
-            for (auto& slice : *part->Slices) {
-                Run->Insert(part, slice);
-            }
-
-            Tags = TVector<TTag>();
-            for (auto c : Mass.Model->Scheme->Cols) {
-                if (c.Group <= groups) {
-                    Tags.push_back(c.Tag);
-                }
-            }
-        }
-
-        void TearDown(const ::benchmark::State& state) {
-            (void)state;
-            Run.Reset();
-            Env.Reset();
-        }
-
-        const NTest::TRowTool Tool;
-        const NTest::TPartEggs Eggs = MakeEggs();
-        THolder<TTouchEnv> Env;
-        THolder<TRun> Run;
-        TVector<TTag> Tags;
-    };
-}
-
-BENCHMARK_DEFINE_F(TPrechargeFixture, PrechargeByKeys)(benchmark::State& state) {
-    ui64 items = state.range(0);
-
-    const auto &keyDefaults = *Tool.Scheme.Keys;
-
-    ui64 it = 0;
-    for (auto _ : state) {
-        ui32 lower = ++it % 50;
-        ui32 upper = lower + items;
-
-        const auto from = Tool.KeyCells(Mass.Saved[lower]);
-        const auto to = Tool.KeyCells(Mass.Saved[upper]);
-
-        ChargeRange(Env.Get(), from, to, *Run.Get(), keyDefaults, Tags, items, Max<ui64>());
-    }
-
-    state.counters["Touched"] = benchmark::Counter(Env->TouchedCount, benchmark::Counter::kAvgIterations);
-}
-
-BENCHMARK_DEFINE_F(TPrechargeFixture, PrechargeByRows)(benchmark::State& state) {
-    ui64 items = state.range(0);
-
-    const auto &keyDefaults = *Tool.Scheme.Keys;
-
-    ui64 it = 0;
-    for (auto _ : state) {
-        ui32 lower = ++it % 50;
-        ui32 upper = lower + items;
-
-        CreateCharge(Env.Get(), *(Run.Get())->begin()->Part, Tags, false)->Do(lower, upper, keyDefaults, items, Max<ui64>());
-    }
-
-    state.counters["Touched"] = Env->TouchedCount / it;
-}
-
-BENCHMARK_REGISTER_F(TPrechargeFixture, PrechargeByKeys)
-    ->ArgsProduct({
-        /* items: */ {0, 100, 1000}, 
-        /* fail: */ {0, 1}, 
-        /* groups: */ {0, 1, 2}})
-    ->Unit(benchmark::kMicrosecond);
-
-BENCHMARK_REGISTER_F(TPrechargeFixture, PrechargeByRows)
-    ->ArgsProduct({
-        /* items: */ {0, 100, 1000}, 
-        /* fail: */{0, 1}, 
-        /* groups: */ {0, 1, 2}})
-    ->Unit(benchmark::kMicrosecond);
-
-}
-}
diff --git a/ydb/core/tablet_flat/benchmark/b_part_index.cpp b/ydb/core/tablet_flat/benchmark/b_part.cpp
index f7421c6af3b..7eea2900321 100644
--- a/ydb/core/tablet_flat/benchmark/b_part_index.cpp
+++ b/ydb/core/tablet_flat/benchmark/b_part.cpp
@@ -41,7 +41,7 @@ namespace {
         return conf;
     }
 
-    struct TPartIndexSeekFixture : public benchmark::Fixture {
+    struct TPartEggsFixture : public benchmark::Fixture {
         using TGroupId = NPage::TGroupId;
 
         void SetUp(const ::benchmark::State& state) 
@@ -84,7 +84,7 @@ namespace {
         TGroupId GroupId;
     };
 
-    struct TPartIndexIteratorFixture : public benchmark::Fixture {
+    struct TPartSubsetFixture : public benchmark::Fixture {
         using TGroupId = NPage::TGroupId;
 
         void SetUp(const ::benchmark::State& state) 
@@ -96,6 +96,14 @@ namespace {
             Mass = new NTest::TMass(new NTest::TModelStd(groups), history ? 1000000 : 300000);
             Subset = TMake(*Mass, PageConf(Mass->Model->Scheme->Families.size(), useBTree)).Mixed(0, 1, TMixerOne{ }, history ? 0.7 : 0);
             
+            for (const auto& part : Subset->Flatten) {
+                Cerr << "DataBytes = " << part->Stat.Bytes << " DataPages = " << IndexTools::CountMainPages(*part) << Endl;
+                Cerr << "FlatIndexBytes = " << part->GetPageSize(part->IndexPages.Groups[groups ? 1 : 0], {}) << " BTreeIndexBytes = " << (useBTree ? part->IndexPages.BTreeGroups[groups ? 1 : 0].IndexSize : 0) << Endl;
+                if (useBTree) {
+                    Cerr << "Levels = " << part->IndexPages.BTreeGroups[groups ? 1 : 0].LevelCount << Endl;
+                }
+            }
+
             if (history) {
                 Checker = new TCheckIt(*Subset, {new TTestEnv()}, TRowVersion(0, 8));
                 CheckerReverse = new TCheckReverseIt(*Subset, {new TTestEnv()}, TRowVersion(0, 8));
@@ -110,10 +118,11 @@ namespace {
         TAutoPtr<TSubset> Subset;
         TAutoPtr<TCheckIt> Checker;
         TAutoPtr<TCheckReverseIt> CheckerReverse;
+        TTestEnv Env;
     };
 }
 
-BENCHMARK_DEFINE_F(TPartIndexSeekFixture, SeekRowId)(benchmark::State& state) {
+BENCHMARK_DEFINE_F(TPartEggsFixture, SeekRowId)(benchmark::State& state) {
     const bool useBTree = state.range(0);
 
     for (auto _ : state) {
@@ -129,7 +138,7 @@ BENCHMARK_DEFINE_F(TPartIndexSeekFixture, SeekRowId)(benchmark::State& state) {
     }
 }
 
-BENCHMARK_DEFINE_F(TPartIndexSeekFixture, Next)(benchmark::State& state) {
+BENCHMARK_DEFINE_F(TPartEggsFixture, Next)(benchmark::State& state) {
     const bool useBTree = state.range(0);
 
     THolder<IIndexIter> iter;
@@ -150,7 +159,7 @@ BENCHMARK_DEFINE_F(TPartIndexSeekFixture, Next)(benchmark::State& state) {
     }
 }
 
-BENCHMARK_DEFINE_F(TPartIndexSeekFixture, Prev)(benchmark::State& state) {
+BENCHMARK_DEFINE_F(TPartEggsFixture, Prev)(benchmark::State& state) {
     const bool useBTree = state.range(0);
 
     THolder<IIndexIter> iter;
@@ -171,7 +180,7 @@ BENCHMARK_DEFINE_F(TPartIndexSeekFixture, Prev)(benchmark::State& state) {
     }
 }
 
-BENCHMARK_DEFINE_F(TPartIndexSeekFixture, SeekKey)(benchmark::State& state) {
+BENCHMARK_DEFINE_F(TPartEggsFixture, SeekKey)(benchmark::State& state) {
     const bool useBTree = state.range(0);
     const ESeek seek = ESeek(state.range(2));
 
@@ -190,7 +199,7 @@ BENCHMARK_DEFINE_F(TPartIndexSeekFixture, SeekKey)(benchmark::State& state) {
     }
 }
 
-BENCHMARK_DEFINE_F(TPartIndexIteratorFixture, DoReads)(benchmark::State& state) {
+BENCHMARK_DEFINE_F(TPartSubsetFixture, DoReads)(benchmark::State& state) {
     const bool reverse = state.range(3);
     const ESeek seek = static_cast<ESeek>(state.range(4));
     const ui32 items = state.range(5);
@@ -212,39 +221,72 @@ BENCHMARK_DEFINE_F(TPartIndexIteratorFixture, DoReads)(benchmark::State& state)
     }
 }
 
-BENCHMARK_REGISTER_F(TPartIndexSeekFixture, SeekRowId)
+BENCHMARK_DEFINE_F(TPartSubsetFixture, DoCharge)(benchmark::State& state) {
+    const bool reverse = state.range(3);
+    const ui32 items = state.range(4);
+
+    auto tags = TVector<TTag>();
+    for (auto c : Subset->Scheme->Cols) {
+        tags.push_back(c.Tag);
+    }
+    TRun run(*Subset->Scheme->Keys);
+    NTest::TRowTool tool(*Subset->Scheme);
+
+    for (auto _ : state) {
+        auto row1 = Rnd.Uniform(Mass->Saved.Size());
+        auto row2 = Min(row1 + items, Mass->Saved.Size() - 1);
+        auto key1 = tool.KeyCells(Mass->Saved[row1]);
+        auto key2 = tool.KeyCells(Mass->Saved[row2]);
+        if (reverse) {
+            ChargeRangeReverse(&Env, key1, key2, run, *Subset->Scheme->Keys, tags, items, 0);
+        } else {
+            ChargeRange(&Env, key1, key2, run, *Subset->Scheme->Keys, tags, items, 0);
+        }
+    }
+}
+
+BENCHMARK_REGISTER_F(TPartEggsFixture, SeekRowId)
     ->ArgsProduct({
         /* b-tree */ {0, 1},
         /* groups: */ {0, 1}})
     ->Unit(benchmark::kMicrosecond);
 
-BENCHMARK_REGISTER_F(TPartIndexSeekFixture, Next)
+BENCHMARK_REGISTER_F(TPartEggsFixture, Next)
     ->ArgsProduct({
         /* b-tree */ {0, 1},
         /* groups: */ {0, 1}})
     ->Unit(benchmark::kMicrosecond);
 
-BENCHMARK_REGISTER_F(TPartIndexSeekFixture, Prev)
+BENCHMARK_REGISTER_F(TPartEggsFixture, Prev)
     ->ArgsProduct({
         /* b-tree */ {0, 1},
         /* groups: */ {0, 1}})
     ->Unit(benchmark::kMicrosecond);
 
-BENCHMARK_REGISTER_F(TPartIndexSeekFixture, SeekKey)
+BENCHMARK_REGISTER_F(TPartEggsFixture, SeekKey)
     ->ArgsProduct({
         /* b-tree */ {0, 1},
         /* groups: */ {0, 1},
-        /* ESeek: */ {0, 1, 2}})
+        /* ESeek: */ {1}})
     ->Unit(benchmark::kMicrosecond);
 
-BENCHMARK_REGISTER_F(TPartIndexIteratorFixture, DoReads)
+BENCHMARK_REGISTER_F(TPartSubsetFixture, DoReads)
     ->ArgsProduct({
         /* b-tree */ {0, 1},
         /* groups: */ {1},
         /* history: */ {1},
         /* reverse: */ {0},
         /* ESeek: */ {1},
-        /* items */ {1, 10, 100}})
+        /* items */ {1, 50, 1000}})
+    ->Unit(benchmark::kMicrosecond);
+
+BENCHMARK_REGISTER_F(TPartSubsetFixture, DoCharge)
+    ->ArgsProduct({
+        /* b-tree */ {0, 1},
+        /* groups: */ {1},
+        /* history: */ {1},
+        /* reverse: */ {0},
+        /* items */ {1, 50, 1000}})
     ->Unit(benchmark::kMicrosecond);
 
 }
diff --git a/ydb/core/tablet_flat/benchmark/ya.make b/ydb/core/tablet_flat/benchmark/ya.make
index 1ec2a13c43a..07dbfc3b251 100644
--- a/ydb/core/tablet_flat/benchmark/ya.make
+++ b/ydb/core/tablet_flat/benchmark/ya.make
@@ -5,8 +5,7 @@ SIZE(LARGE)
 TIMEOUT(1200)
 
 SRCS(
-    b_charge.cpp
-    b_part_index.cpp
+    b_part.cpp
 )
 
 PEERDIR(
diff --git a/ydb/core/tablet_flat/flat_part_charge_range.cpp b/ydb/core/tablet_flat/flat_part_charge_range.cpp
index 2906bcd8abe..bc7328c184f 100644
--- a/ydb/core/tablet_flat/flat_part_charge_range.cpp
+++ b/ydb/core/tablet_flat/flat_part_charge_range.cpp
@@ -40,8 +40,8 @@ bool ChargeRange(IPages *env, const TCells key1, const TCells key2,
             if (r.Overshot && ++pos != run.end()) {
                 // Unfortunately first key > key2 might be at the start of the next slice
                 TRowId firstRow = pos->Slice.BeginRowId();
-                // Precharge the first row on the next slice
-                ready &= CreateCharge(env, *pos->Part, tags, includeHistory)->Do(firstRow, firstRow, keyDefaults, items, bytes);
+                // Precharge the first row main key on the next slice
+                ready &= CreateCharge(env, *pos->Part, { }, false)->Do(firstRow, firstRow, keyDefaults, items, bytes);
             }
 
             break;
@@ -98,8 +98,8 @@ bool ChargeRangeReverse(IPages *env, const TCells key1, const TCells key2,
                 --pos;
                 // Unfortunately first key <= key2 might be at the end of the previous slice
                 TRowId lastRow = pos->Slice.EndRowId() - 1;
-                // Precharge the last row on the previous slice
-                ready &= CreateCharge(env, *pos->Part, tags, includeHistory)->DoReverse(lastRow, lastRow, keyDefaults, items, bytes);
+                // Precharge the last row main key on the previous slice
+                ready &= CreateCharge(env, *pos->Part, { }, false)->DoReverse(lastRow, lastRow, keyDefaults, items, bytes);
             }
 
             break;
diff --git a/ydb/core/tablet_flat/test/libs/table/test_part.h b/ydb/core/tablet_flat/test/libs/table/test_part.h
index e2607cd711c..3a133b1af84 100644
--- a/ydb/core/tablet_flat/test/libs/table/test_part.h
+++ b/ydb/core/tablet_flat/test/libs/table/test_part.h
@@ -148,7 +148,7 @@ namespace NTest {
     namespace IndexTools {
         using TGroupId = NPage::TGroupId;
 
-        inline size_t CountMainPages(const TPartStore& part) {
+        inline size_t CountMainPages(const TPart& part) {
             size_t result = 0;
 
             TTestEnv env;
@@ -165,20 +165,20 @@ namespace NTest {
             return result;
         }
 
-        inline TRowId GetEndRowId(const TPartStore& part) {
+        inline TRowId GetEndRowId(const TPart& part) {
             TTestEnv env;
             TPartIndexIt index(&part, &env, { });
             return index.GetEndRowId();
         }
 
-        inline const TPartIndexIt::TRecord * GetLastRecord(const TPartStore& part) {
+        inline const TPartIndexIt::TRecord * GetLastRecord(const TPart& part) {
             TTestEnv env;
             TPartIndexIt index(&part, &env, { });
             Y_ABORT_UNLESS(index.SeekLast() == EReady::Data);
             return index.GetLastRecord();
         }
 
-        inline const TPartIndexIt::TRecord * GetRecord(const TPartStore& part, ui32 pageIndex) {
+        inline const TPartIndexIt::TRecord * GetRecord(const TPart& part, ui32 pageIndex) {
             TTestEnv env;
             TPartIndexIt index(&part, &env, { });
 
@@ -190,14 +190,14 @@ namespace NTest {
             return index.GetRecord();
         }
 
-        inline TPageId GetFirstPageId(const TPartStore& part, TGroupId groupId) {
+        inline TPageId GetFirstPageId(const TPart& part, TGroupId groupId) {
             TTestEnv env;
             TPartIndexIt index(&part, &env, groupId);
             index.Seek(0);
             return index.GetPageId();
         }
 
-        inline TPageId GetLastPageId(const TPartStore& part, TGroupId groupId) {
+        inline TPageId GetLastPageId(const TPart& part, TGroupId groupId) {
             TTestEnv env;
             TPartIndexIt index(&part, &env, groupId);
             index.Seek(index.GetEndRowId() - 1);
diff --git a/ydb/core/tablet_flat/test/libs/table/wrap_part.h b/ydb/core/tablet_flat/test/libs/table/wrap_part.h
index 98505bb0e81..6c138dd92e2 100644
--- a/ydb/core/tablet_flat/test/libs/table/wrap_part.h
+++ b/ydb/core/tablet_flat/test/libs/table/wrap_part.h
@@ -15,14 +15,25 @@ namespace NTest {
 
     template<EDirection Direction>
     struct TWrapPartImpl {
-        TWrapPartImpl(const TPartEggs &eggs, TIntrusiveConstPtr<TSlices> slices = nullptr,
-                    bool defaults = true)
+        TWrapPartImpl(const TPartEggs &eggs, TRun& run, bool defaults = true)
             : Eggs(eggs)
             , Scheme(eggs.Scheme)
             , Remap_(TRemap::Full(*Scheme))
             , Defaults(defaults)
             , State(Remap_.Size())
-            , Run(*Scheme->Keys)
+            , Run_(*Scheme->Keys) // unused
+            , Run(run)
+        {
+        }
+        
+        TWrapPartImpl(const TPartEggs &eggs, TIntrusiveConstPtr<TSlices> slices = nullptr, bool defaults = true)
+            : Eggs(eggs)
+            , Scheme(eggs.Scheme)
+            , Remap_(TRemap::Full(*Scheme))
+            , Defaults(defaults)
+            , State(Remap_.Size())
+            , Run_(*Scheme->Keys)
+            , Run(Run_)
         {
             if (slices || Eggs.Parts.size() == 1) {
                 /* Allowed to override part slice only for lone eggs */
@@ -45,6 +56,8 @@ namespace NTest {
         }
 
     public:
+        using TCells = TArrayRef<const TCell>;
+
         explicit operator bool() const noexcept
         {
             return Iter && Iter->IsValid() && Ready == EReady::Data;
@@ -69,7 +82,11 @@ namespace NTest {
         EReady Seek(TRawVals key_, ESeek seek) noexcept
         {
             const TCelled key(key_, *Scheme->Keys, false);
+            return Seek(key, seek);
+        }
 
+        EReady Seek(const TCells key, ESeek seek) noexcept
+        {
             if constexpr (Direction == EDirection::Reverse) {
                 Ready = Iter->SeekReverse(key, seek);
             } else {
@@ -104,15 +121,6 @@ namespace NTest {
             return Iter->GetRowVersion();
         }
 
-        EReady DoIterNext() noexcept
-        {
-            if constexpr (Direction == EDirection::Reverse) {
-                return Iter->Prev();
-            } else {
-                return Iter->Next();
-            }
-        }
-
         void StopAfter(TArrayRef<const TCell> key) {
             StopKey = TOwnedCellVec::Make(key);
         }
@@ -147,7 +155,11 @@ namespace NTest {
             TDbTupleRef key = Iter->GetKey();
 
             if (StopKey) {
-                auto cmp = CompareTypedCellVectors(key.Cells().data(), StopKey.data(), Scheme->Keys->Types.data(), StopKey.size());
+                auto cmp = CompareTypedCellVectors(key.Cells().data(), StopKey.data(), Scheme->Keys->Types.data(), Min(key.Cells().size(), StopKey.size()));
+                if (cmp == 0 && key.Cells().size() != StopKey.size()) {
+                    // smaller key is filled with +inf => always bigger
+                    cmp = key.Cells().size() < StopKey.size() ? +1 : -1;
+                }
                 if (Direction == EDirection::Forward && cmp > 0 || Direction == EDirection::Reverse && cmp < 0) {
                    return EReady::Gone;
                 }
@@ -167,10 +179,20 @@ namespace NTest {
         const bool Defaults = true;
 
     private:
+        EReady DoIterNext() noexcept
+        {
+            if constexpr (Direction == EDirection::Reverse) {
+                return Iter->Prev();
+            } else {
+                return Iter->Next();
+            }
+        }
+
         EReady Ready = EReady::Gone;
         bool NoBlobs = false;
         TRowState State;
-        TRun Run;
+        TRun Run_;
+        TRun& Run;
         THolder<TRunIt> Iter;
         TOwnedCellVec StopKey;
     };
diff --git a/ydb/core/tablet_flat/ut/ut_btree_index_iter_charge.cpp b/ydb/core/tablet_flat/ut/ut_btree_index_iter_charge.cpp
index 4bb71c2a938..53fefde5063 100644
--- a/ydb/core/tablet_flat/ut/ut_btree_index_iter_charge.cpp
+++ b/ydb/core/tablet_flat/ut/ut_btree_index_iter_charge.cpp
@@ -3,8 +3,8 @@
 #include "flat_part_charge.h"
 #include "flat_part_charge_btree_index.h"
 #include "flat_part_charge_range.h"
-#include "flat_part_iter_multi.h"
 #include "test/libs/table/test_writer.h"
+#include "test/libs/table/wrap_part.h"
 #include <ydb/core/tablet_flat/test/libs/rows/layout.h>
 #include <library/cpp/testing/unittest/registar.h>
 
@@ -228,14 +228,25 @@ namespace {
     }
 
     EReady Retry(std::function<EReady()> action, TTouchEnv& env, const TString& message, ui32 failsAllowed = 10) {
-        while (true) {
+        for (ui32 attempt = 0; attempt <= failsAllowed; attempt++) {
+            env.LoadTouched();
             if (auto ready = action(); ready != EReady::Page) {
                 return ready;
             }
-            env.LoadTouched();
-            UNIT_ASSERT_C(failsAllowed--, "Too many fails " + message);
         }
-        Y_UNREACHABLE();
+
+        TStringBuilder error;
+        error << "Too many fails (" << failsAllowed + 1 << ") " << message << Endl << "Requests ";
+        for (const auto& [groupId, pages] : env.Touched) {
+            for (auto pageId : pages) {
+                if (!env.Loaded[groupId].contains(pageId)) {
+                    error << groupId << "#" << pageId << " ";
+                }
+            }
+        }
+
+        UNIT_ASSERT_C(false,  error);
+        return EReady::Page;
     }
 }
 
@@ -426,32 +437,25 @@ Y_UNIT_TEST_SUITE(TChargeBTreeIndex) {
 
     void DoChargeRowId(ICharge& charge, TTouchEnv& env, const TRowId row1, const TRowId row2, ui64 itemsLimit, ui64 bytesLimit,
             bool reverse, const TKeyCellDefaults &keyDefaults, const TString& message, ui32 failsAllowed = 15) {
-        while (true) {
+        Retry([&]() {
             bool ready = reverse
                 ? charge.DoReverse(row2, row1, keyDefaults, itemsLimit, bytesLimit)
                 : charge.Do(row1, row2, keyDefaults, itemsLimit, bytesLimit);
-            if (ready) {
-                return;
-            }
-            env.LoadTouched();
-            UNIT_ASSERT_C(failsAllowed--, "Too many fails " + message);
-        }
-        Y_UNREACHABLE();
+            return ready ? EReady::Data : EReady::Page;
+        }, env, message, failsAllowed);
     }
 
     bool DoChargeKeys(const TPartStore& part, ICharge& charge, TTouchEnv& env, const TCells key1, const TCells key2, ui64 itemsLimit, ui64 bytesLimit,
             bool reverse, const TKeyCellDefaults &keyDefaults, const TString& message, ui32 failsAllowed = 15) {
-        while (true) {
+        bool overshot = false;
+        Retry([&]() {
             auto result = reverse
                 ? charge.DoReverse(key1, key2, part.Stat.Rows - 1, 0, keyDefaults, itemsLimit, bytesLimit)
                 : charge.Do(key1, key2, 0, part.Stat.Rows - 1, keyDefaults, itemsLimit, bytesLimit);
-            if (result.Ready) {
-                return result.Overshot;
-            }
-            env.LoadTouched();
-            UNIT_ASSERT_C(failsAllowed--, "Too many fails " + message);
-        }
-        Y_UNREACHABLE();
+            overshot = result.Overshot;
+            return result.Ready ? EReady::Data : EReady::Page;
+        }, env, message, failsAllowed);
+        return overshot;
     }
 
     void CheckChargeRowId(TTestParams params, const TPartStore& part, TTagsRef tags, const TKeyCellDefaults *keyDefaults) {
@@ -675,40 +679,82 @@ Y_UNIT_TEST_SUITE(TPartBtreeIndexIteration) {
         }
     }
 
-    void AssertEqual(const TRunIt& bTree, EReady bTreeReady, const TRunIt& flat, EReady flatReady, const TString& message) {
+    ui32 GetFailsAllowed(TTestParams params) {
+        ui32 result = (params.Levels + 1) * 2;
+        if (params.History) {
+            result *= 2;
+        }
+        if (params.Groups) {
+            result *= 2;
+        }
+        return result;
+    }
+
+    template<EDirection Direction>
+    void AssertEqual(const TWrapPartImpl<Direction>& bTree, EReady bTreeReady, const TWrapPartImpl<Direction>& flat, EReady flatReady, const TString& message) {
         UNIT_ASSERT_VALUES_EQUAL_C(bTreeReady, flatReady, message);
-        UNIT_ASSERT_VALUES_EQUAL_C(bTree.IsValid(), flat.IsValid(), message);
-        UNIT_ASSERT_VALUES_EQUAL_C(bTree.GetRowId(), flat.GetRowId(), message);
+        UNIT_ASSERT_VALUES_EQUAL_C(bTree.Get()->IsValid(), flat.Get()->IsValid(), message);
+        UNIT_ASSERT_VALUES_EQUAL_C(bTree.Get()->GetRowId(), flat.Get()->GetRowId(), message);
     }
 
-    EReady Seek(TRunIt& iter, TTouchEnv& env, ESeek seek, bool reverse, TCells key, const TString& message, ui32 failsAllowed = 10) {
+    template<EDirection Direction>
+    EReady Seek(TWrapPartImpl<Direction>& wrap, TTouchEnv& env, const TCells key1, ESeek seek, const TString& message, ui32 failsAllowed) {
         return Retry([&]() {
-            return reverse ? iter.SeekReverse(key, seek) : iter.Seek(key, seek);
+            return wrap.Seek(key1, seek);
         }, env, message, failsAllowed);
     }
 
-    EReady Next(TRunIt& iter, TTouchEnv& env, bool reverse, const TString& message, ui32 failsAllowed = 10) {
+    template<EDirection Direction>
+    EReady Next(TWrapPartImpl<Direction>& wrap, TTouchEnv& env, const TString& message, ui32 failsAllowed) {
         return Retry([&]() {
-            return reverse ? iter.Prev() : iter.Next();
+            return wrap.Next();
+        }, env, message, failsAllowed);
+    }
+
+    template<EDirection Direction>
+    EReady SkipToRowVersion(TWrapPartImpl<Direction>& wrap, TTouchEnv& env, TRowVersion rowVersion, const TString& message, ui32 failsAllowed) {
+        return Retry([&]() {
+            return wrap.SkipToRowVersion(rowVersion);
         }, env, message, failsAllowed);
     }
 
     void Charge(const TRun &run, const TVector<TTag> tags, TTouchEnv& env, const TCells key1, const TCells key2, ui64 itemsLimit, ui64 bytesLimit,
-            bool reverse, const TKeyCellDefaults &keyDefaults, const TString& message, ui32 failsAllowed = 15) {
-        while (true) {
-            auto result = reverse
+            bool reverse, const TKeyCellDefaults &keyDefaults, const TString& message, ui32 failsAllowed) {
+        Retry([&]() {
+            auto ready = reverse
                 ? ChargeRangeReverse(&env, key1, key2, run, keyDefaults, tags, itemsLimit, bytesLimit, true)
                 : ChargeRange(&env, key1, key2, run, keyDefaults, tags, itemsLimit, bytesLimit, true);
-            if (result) {
+            return ready ? EReady::Data : EReady::Page;
+        }, env, message, failsAllowed);
+    }
+
+    template<EDirection Direction>
+    void Iterate(const TPartEggs& eggs, TRun& run, TTouchEnv& env, const TCells key1, const TCells key2, ESeek seek, ui64 itemsLimit, bool history, const TString& message, ui32 failsAllowed) {
+        TWrapPartImpl<Direction> wrap(eggs, run);
+        wrap.StopAfter(key2);
+        wrap.Make(&env);
+        
+        if (Seek(wrap, env, key1, seek, message + " Seek", failsAllowed) != EReady::Data) {
+            return;
+        }
+        if (history) {
+            UNIT_ASSERT_VALUES_EQUAL(SkipToRowVersion(wrap, env, {0, 1}, message + " Ver", failsAllowed), EReady::Data);
+        }
+
+        for (ui32 itemIndex = 1; itemsLimit == 0 || itemIndex < itemsLimit; itemIndex++) {
+            if (Next(wrap, env, message + " Next " + std::to_string(itemIndex), failsAllowed) != EReady::Data) {
                 return;
             }
-            env.LoadTouched();
-            UNIT_ASSERT_C(failsAllowed--, "Too many fails " + message);
+            if (history) {
+                UNIT_ASSERT_VALUES_EQUAL(SkipToRowVersion(wrap, env, {0, 1}, message + " Ver", failsAllowed), EReady::Data);
+            }
         }
-        Y_UNREACHABLE();
     }
 
-    void CheckIterate(const TPartEggs& eggs) {
+    template<EDirection Direction>
+    void CheckIterate(TTestParams params, const TPartEggs& eggs) {
+        constexpr bool reverse = Direction == EDirection::Reverse;
+        const ui32 failsAllowed = GetFailsAllowed(params);
         const auto part = *eggs.Lone();
 
         TRun btreeRun(*eggs.Scheme->Keys), flatRun(*eggs.Scheme->Keys);
@@ -719,45 +765,48 @@ Y_UNIT_TEST_SUITE(TPartBtreeIndexIteration) {
             tags.push_back(c.Tag);
         }
 
-        for (bool reverse : {false, true}) {
-            for (ESeek seek : {ESeek::Exact, ESeek::Lower, ESeek::Upper}) {
-                for (ui32 firstCell : xrange<ui32>(0, part.Stat.Rows / 7 + 1)) {
-                    for (ui32 secondCell : xrange<ui32>(0, 14)) {
-                        TVector<TCell> key = MakeKey(firstCell, secondCell);
+        for (ESeek seek : {ESeek::Exact, ESeek::Lower, ESeek::Upper}) {
+            for (ui32 firstCell : xrange<ui32>(0, part.Stat.Rows / 7 + 1)) {
+                for (ui32 secondCell : xrange<ui32>(0, 14)) {
+                    TVector<TCell> key = MakeKey(firstCell, secondCell);
 
-                        TTouchEnv bTreeEnv, flatEnv;
-                        TRunIt flat(flatRun, tags, eggs.Scheme->Keys, &flatEnv);
-                        TRunIt bTree(btreeRun, tags, eggs.Scheme->Keys, &bTreeEnv);
+                    TTouchEnv bTreeEnv, flatEnv;
+                    TWrapPartImpl<Direction> bTree(eggs, btreeRun);
+                    TWrapPartImpl<Direction> flat(eggs, flatRun);
+                    bTree.Make(&bTreeEnv);
+                    flat.Make(&flatEnv);
 
-                        {
-                            TStringBuilder message = TStringBuilder() << (reverse ?  "IterateReverse" : "Iterate") << "(" << seek << ") ";
-                            for (auto c : key) {
-                                message << c.AsValue<ui32>() << " ";
-                            }
-                            EReady bTreeReady = Seek(bTree, bTreeEnv, seek, reverse, key, message);
-                            EReady flatReady = Seek(flat, flatEnv, seek, reverse, key, message);
-                            AssertEqual(bTree, bTreeReady, flat, flatReady, message);
-                            AssertLoadedTheSame(part, bTreeEnv, flatEnv, message);
+                    {
+                        TStringBuilder message = TStringBuilder() << (reverse ?  "IterateReverse" : "Iterate") << "(" << seek << ") ";
+                        for (auto c : key) {
+                            message << c.AsValue<ui32>() << " ";
                         }
+                        EReady bTreeReady = Seek(bTree, bTreeEnv, key, seek, message, failsAllowed);
+                        EReady flatReady = Seek(flat, flatEnv, key, seek, message, failsAllowed);
+                        AssertEqual(bTree, bTreeReady, flat, flatReady, message);
+                        AssertLoadedTheSame(part, bTreeEnv, flatEnv, message);
+                    }
 
-                        for (ui32 steps = 1; steps <= 10; steps++) {
-                            TStringBuilder message = TStringBuilder() << (reverse ?  "IterateReverse" : "Iterate") << "(" << seek << ") ";
-                            for (auto c : key) {
-                                message << c.AsValue<ui32>() << " ";
-                            }
-                            message << " --> " << steps << " steps ";
-                            EReady bTreeReady = Next(bTree, bTreeEnv, reverse, message);
-                            EReady flatReady = Next(flat, flatEnv, reverse, message);
-                            AssertEqual(bTree, bTreeReady, flat, flatReady, message);
-                            AssertLoadedTheSame(part, bTreeEnv, flatEnv, message);
+                    for (ui32 steps = 1; steps <= 10; steps++) {
+                        TStringBuilder message = TStringBuilder() << (reverse ?  "IterateReverse" : "Iterate") << "(" << seek << ") ";
+                        for (auto c : key) {
+                            message << c.AsValue<ui32>() << " ";
                         }
+                        message << " --> " << steps << " steps ";
+                        EReady bTreeReady = Next(bTree, bTreeEnv, message, failsAllowed);
+                        EReady flatReady = Next(flat, flatEnv, message, failsAllowed);
+                        AssertEqual(bTree, bTreeReady, flat, flatReady, message);
+                        AssertLoadedTheSame(part, bTreeEnv, flatEnv, message);
                     }
                 }
             }
         }
     }
 
-    void CheckCharge(const TPartEggs& eggs) {
+    template<EDirection Direction>
+    void CheckCharge(TTestParams params, const TPartEggs& eggs) {
+        constexpr bool reverse = Direction == EDirection::Reverse;
+        const ui32 failsAllowed = GetFailsAllowed(params);
         const auto part = *eggs.Lone();
 
         TRun btreeRun(*eggs.Scheme->Keys), flatRun(*eggs.Scheme->Keys);
@@ -768,33 +817,38 @@ Y_UNIT_TEST_SUITE(TPartBtreeIndexIteration) {
             tags.push_back(c.Tag);
         }
 
-        for (bool reverse : {false, true}) {
-            for (ui64 itemsLimit : part.Slices->size() > 1 ? TVector<ui64>{0} : TVector<ui64>{0, 1, 2, 5, 13, 19, part.Stat.Rows - 2, part.Stat.Rows - 1}) {
-                for (ui32 firstCellKey1 : xrange<ui32>(0, part.Stat.Rows / 7 + 1)) {
-                    for (ui32 secondCellKey1 : xrange<ui32>(0, 14)) {
-                        for (ui32 firstCellKey2 : xrange<ui32>(0, part.Stat.Rows / 7 + 1)) {
-                            for (ui32 secondCellKey2 : xrange<ui32>(0, 14)) {
-                                TVector<TCell> key1 = MakeKey(firstCellKey1, secondCellKey1);
-                                TVector<TCell> key2 = MakeKey(firstCellKey2, secondCellKey2);
+        for (ui64 itemsLimit : part.Slices->size() > 1 ? TVector<ui64>{0, 1, 2, 5} : TVector<ui64>{0, 1, 2, 5, 13, 19, part.Stat.Rows - 2, part.Stat.Rows - 1}) {
+            for (ui32 firstCellKey1 : xrange<ui32>(0, part.Stat.Rows / 7 + 1)) {
+                for (ui32 secondCellKey1 : xrange<ui32>(0, 14)) {
+                    for (ui32 firstCellKey2 : xrange<ui32>(0, part.Stat.Rows / 7 + 1)) {
+                        for (ui32 secondCellKey2 : xrange<ui32>(0, 14)) {
+                            TVector<TCell> key1 = MakeKey(firstCellKey1, secondCellKey1);
+                            TVector<TCell> key2 = MakeKey(firstCellKey2, secondCellKey2);
 
-                                TTouchEnv bTreeEnv, flatEnv;
-                                
-                                TStringBuilder message = TStringBuilder() << (reverse ? "ChargeReverse " : "Charge ") << "(";
-                                for (auto c : key1) {
-                                    message << c.AsValue<ui32>() << " ";
-                                }
-                                message << ") (";
-                                for (auto c : key2) {
-                                    message << c.AsValue<ui32>() << " ";
-                                }
-                                message << ") items " << itemsLimit;
+                            TTouchEnv bTreeEnv, flatEnv;
+                            
+                            TStringBuilder message = TStringBuilder() << (reverse ? "ChargeReverse " : "Charge ") << "(";
+                            for (auto c : key1) {
+                                message << c.AsValue<ui32>() << " ";
+                            }
+                            message << ") (";
+                            for (auto c : key2) {
+                                message << c.AsValue<ui32>() << " ";
+                            }
+                            message << ") items " << itemsLimit;
 
-                                Charge(btreeRun, tags, bTreeEnv, key1, key2, itemsLimit, 0, reverse, *eggs.Scheme->Keys, message);
-                                Charge(flatRun, tags, flatEnv, key1, key2, itemsLimit, 0, reverse, *eggs.Scheme->Keys, message);
+                            Charge(btreeRun, tags, bTreeEnv, key1, key2, itemsLimit, 0, reverse, *eggs.Scheme->Keys, message, failsAllowed);
+                            Charge(flatRun, tags, flatEnv, key1, key2, itemsLimit, 0, reverse, *eggs.Scheme->Keys, message, failsAllowed);
 
+                            if (!itemsLimit || part.Slices->size() == 1) {
                                 AssertLoadedTheSame(part, bTreeEnv, flatEnv, message,
                                     false, reverse && itemsLimit, !reverse && itemsLimit);
                             }
+
+                            for (ESeek seek : {ESeek::Exact, ESeek::Lower, ESeek::Upper}) {
+                                Iterate<Direction>(eggs, btreeRun, bTreeEnv, key1, key2, seek, itemsLimit, params.History, message, 0);
+                                Iterate<Direction>(eggs, flatRun, flatEnv, key1, key2, seek, itemsLimit, params.History, message, 0);
+                            }
                         }
                     }
                 }
@@ -806,8 +860,10 @@ Y_UNIT_TEST_SUITE(TPartBtreeIndexIteration) {
         TPartEggs eggs = MakePart(params);
         const auto part = *eggs.Lone();
 
-        CheckIterate(eggs);
-        CheckCharge(eggs);
+        CheckIterate<EDirection::Forward>(params, eggs);
+        CheckIterate<EDirection::Reverse>(params, eggs);
+        CheckCharge<EDirection::Forward>(params, eggs);
+        CheckCharge<EDirection::Reverse>(params, eggs);
     }
 
     Y_UNIT_TEST(NoNodes) {