Fix caching of incorrect erased ranges,

4 files changed, 102 insertions, 16 deletions

diff --git a/ydb/core/tablet_flat/flat_iterator.h b/ydb/core/tablet_flat/flat_iterator.h
index 4a4e44917cd..1ecf30ce5dd 100644
--- a/ydb/core/tablet_flat/flat_iterator.h
+++ b/ydb/core/tablet_flat/flat_iterator.h
@@ -258,7 +258,6 @@ public:
     {
         TEraseCachingState eraseCache(this);
 
-        bool isHead = true;
         for (Ready = EReady::Data; Ready == EReady::Data; ) {
             if (Stage == EStage::Seek) {
                 Ready = Start();
@@ -269,32 +268,31 @@ public:
                 Ready = Turn();
             } else if (Stage == EStage::Snap) {
                 if (mode != ENext::Uncommitted) {
-                    ui64 skipsBefore = Stats.InvisibleRowSkips;
                     Ready = Snap();
-                    isHead = skipsBefore == Stats.InvisibleRowSkips;
+                    if (ErasedKeysCache && mode == ENext::Data &&
+                        (Stats.InvisibleRowSkips != SnapInvisibleRowSkips || Stage != EStage::Fill))
+                    {
+                        // Interrupt range when key is not at a head version, or skipped entirely
+                        eraseCache.Flush();
+                    }
                 } else {
                     Y_VERIFY_DEBUG(Active != Inactive);
                     Stage = EStage::Fill;
-                    isHead = false;
                 }
             } else if ((Ready = Apply()) != EReady::Data) {
 
-            } else if (mode == ENext::All || mode == ENext::Uncommitted || State.GetRowState() != ERowOp::Erase) {
+            } else if (mode != ENext::Data || State.GetRowState() != ERowOp::Erase) {
                 break;
             } else {
                 ++Stats.DeletedRowSkips; /* skip internal technical row states w/o data */
-                if (ErasedKeysCache) {
-                    if (isHead) {
-                        eraseCache.OnEraseKey(GetKey().Cells(), GetRowVersion());
-                    } else {
-                        eraseCache.Flush();
-                        isHead = true;
-                    }
+                if (ErasedKeysCache && Stats.InvisibleRowSkips == SnapInvisibleRowSkips) {
+                    // Try to cache erases that are at a head version
+                    eraseCache.OnEraseKey(GetKey().Cells(), GetRowVersion());
                 }
             }
         }
 
-        if (ErasedKeysCache && mode != ENext::All) {
+        if (ErasedKeysCache && mode == ENext::Data) {
             eraseCache.Flush();
         }
 
@@ -415,6 +413,7 @@ private:
     TIterators Iterators;
     TForwardIter Active;
     TForwardIter Inactive;
+    ui64 SnapInvisibleRowSkips = 0;
     ui64 DeltaTxId = 0;
     TRowVersion DeltaVersion;
     bool Delta = false;
@@ -584,6 +583,7 @@ inline EReady TTableItBase<TIteratorOps>::Start() noexcept
     }
 
     Stage = EStage::Snap;
+    SnapInvisibleRowSkips = Stats.InvisibleRowSkips;
     Inactive = Iterators.end();
     return EReady::Data;
 }
diff --git a/ydb/core/tablet_flat/test/libs/table/test_dbase.h b/ydb/core/tablet_flat/test/libs/table/test_dbase.h
index b0bd726c6cc..bb846d1057a 100644
--- a/ydb/core/tablet_flat/test/libs/table/test_dbase.h
+++ b/ydb/core/tablet_flat/test/libs/table/test_dbase.h
@@ -145,6 +145,15 @@ namespace NTest {
             return check.To(CurrentStep()), check;
         }
 
+        TCheckIter IterData(ui32 table) noexcept
+        {
+            DoBegin(false), RowSchemeFor(table);
+
+            TCheckIter check{ *Base, { nullptr, 0, true }, table, Scheme, ReadVersion, ENext::Data };
+
+            return check.To(CurrentStep()), check;
+        }
+
         TCheckSelect Select(ui32 table, bool erased = true) noexcept
         {
             DoBegin(false), RowSchemeFor(table);
diff --git a/ydb/core/tablet_flat/test/libs/table/wrap_dbase.h b/ydb/core/tablet_flat/test/libs/table/wrap_dbase.h
index aa162bf0f19..f800e51c253 100644
--- a/ydb/core/tablet_flat/test/libs/table/wrap_dbase.h
+++ b/ydb/core/tablet_flat/test/libs/table/wrap_dbase.h
@@ -10,11 +10,13 @@ namespace NTest {
     struct TWrapDbIterImpl {
 
         TWrapDbIterImpl(TDatabase &base, ui32 table, TIntrusiveConstPtr<TRowScheme> scheme,
-                TRowVersion snapshot = TRowVersion::Max())
+                TRowVersion snapshot = TRowVersion::Max(),
+                ENext mode = ENext::All)
             : Scheme(std::move(scheme))
             , Base(base)
             , Table(table)
             , Snapshot(snapshot)
+            , Mode(mode)
         {
 
         }
@@ -68,12 +70,12 @@ namespace NTest {
 
             Iter = Base.IterateRangeGeneric<TIter>(Table, range, Scheme->Tags(), Snapshot);
 
-            return Iter->Next(ENext::All);
+            return Iter->Next(Mode);
         }
 
         EReady Next() noexcept
         {
-            return Iter->Next(ENext::All);
+            return Iter->Next(Mode);
         }
 
         const TRowState& Apply() noexcept
@@ -88,6 +90,7 @@ namespace NTest {
     private:
         const ui32 Table = Max<ui32>();
         const TRowVersion Snapshot;
+        const ENext Mode;
         TAutoPtr<TIter> Iter;
     };
 
diff --git a/ydb/core/tablet_flat/ut/ut_db_iface.cpp b/ydb/core/tablet_flat/ut/ut_db_iface.cpp
index 8d508394cf8..1e2c439af2e 100644
--- a/ydb/core/tablet_flat/ut/ut_db_iface.cpp
+++ b/ydb/core/tablet_flat/ut/ut_db_iface.cpp
@@ -734,6 +734,80 @@ Y_UNIT_TEST_SUITE(DBase) {
         RunVersionChecks(true, true);
     }
 
+    // Regression test for KIKIMR-15506
+    Y_UNIT_TEST(KIKIMR_15506_MissingSnapshotKeys) {
+        TDbExec me;
+
+        const ui32 table = 1;
+        me.To(10)
+            .Begin()
+            .Apply(*TAlter()
+                .AddTable("me_1", table)
+                .AddColumn(table, "key", 1, ETypes::Uint64, false)
+                .AddColumn(table, "val", 2, ETypes::Uint64, false, Cimple(0_u64))
+                .AddColumnToKey(table, 1)
+                .SetEraseCache(table, true, 2, 8192))
+            .Commit();
+
+        auto dumpCache = [&]() -> TString {
+            if (auto* cache = me->DebugGetTableErasedKeysCache(table)) {
+                TStringStream stream;
+                stream << cache->DumpRanges();
+                return stream.Str();
+            } else {
+                return nullptr;
+            }
+        };
+
+        // Write a bunch of rows at v1/50
+        me.To(20).Begin();
+        for (ui64 i = 1; i <= 18; ++i) {
+            if (i != 9) {
+                me.WriteVer({1, 50}).Put(table, *me.SchemedCookRow(table).Col(i, i));
+            }
+        }
+        me.Commit();
+
+        // Erase a bunch of rows at v2/50
+        me.To(21).Begin();
+        for (ui64 i = 1; i <= 16; ++i) {
+            if (i != 9) {
+                me.WriteVer({2, 50}).Add(table, *me.SchemedCookRow(table).Col(i), ERowOp::Erase);
+            }
+        }
+        me.Commit();
+
+        // Verify we can only see 2 last rows at v3/50 (all other are deleted)
+        me.To(22).ReadVer({3, 50}).IterData(table)
+            .Seek({ }, ESeek::Lower).Is(*me.SchemedCookRow(table).Col(17_u64, 17_u64))
+            .Next().Is(*me.SchemedCookRow(table).Col(18_u64, 18_u64))
+            .Next().Is(EReady::Gone);
+
+        UNIT_ASSERT_VALUES_EQUAL(dumpCache(), "TKeyRangeCache{ [{1}, {16}] }");
+
+        // Add a new row at v4/50 (it's expected to invalidate the cached range)
+        me.To(23).Begin();
+        me.WriteVer({4, 50}).Put(table, *me.SchemedCookRow(table).Col(9_u64, 9_u64));
+        me.Commit();
+
+        UNIT_ASSERT_VALUES_EQUAL(dumpCache(), "TKeyRangeCache{ }");
+
+        // Verify we can only see 2 last rows at v3/50 (erased range shouldn't be cached incorrectly)
+        me.To(24).ReadVer({3, 50}).IterData(table)
+            .Seek({ }, ESeek::Lower).Is(*me.SchemedCookRow(table).Col(17_u64, 17_u64))
+            .Next().Is(*me.SchemedCookRow(table).Col(18_u64, 18_u64))
+            .Next().Is(EReady::Gone);
+
+        UNIT_ASSERT_VALUES_EQUAL(dumpCache(), "TKeyRangeCache{ [{1}, {8}], [{10}, {16}] }");
+
+        // Verify we can see all 3 rows at v5/50 (bug would cause as to skip over the key 9)
+        me.To(25).ReadVer({5, 50}).IterData(table)
+            .Seek({ }, ESeek::Lower).Is(*me.SchemedCookRow(table).Col(9_u64, 9_u64))
+            .Next().Is(*me.SchemedCookRow(table).Col(17_u64, 17_u64))
+            .Next().Is(*me.SchemedCookRow(table).Col(18_u64, 18_u64))
+            .Next().Is(EReady::Gone);
+    }
+
 }
 
 }