Keep read-path tombstone conversion from deleting ingested files an older iterator cannot see (#15461)

Summary:
Pull Request resolved: https://github.com/facebook/rocksdb/pull/15461

**What:** With `min_tombstones_for_range_conversion` set, an iterator whose view predates an ingestion can delete the ingested keys for good.

**Why it matters:** Silent, permanent data loss. `Get` returns the ingested key right after ingestion. After one scan from the older iterator, `Get` returns NotFound, also after flush and reopen. Every MyRocks bulk load with the option set takes this path: MyRocks ingests with `allow_global_seqno=false` and `snapshot_consistency=false` (seqno 0).

**Root cause:**
- The older iterator does not see the ingested file. It sees the deleted keys on both sides of the file as one run, and inserts a range tombstone at its read sequence number. The tombstone covers the ingested keys.
- `ingest_seqno_barrier_` (D102044512) refuses a conversion only when the iterator's sequence number is below the ingested seqno. That misses two cases:
  1. The file gets no new seqno: no overlap (seqno 0), `ingest_behind`, or `allow_db_generated_files`. Ingestion does not move the barrier at all (`assigned > 0`).
  2. `NewIterator()` references the SuperVersion before it reads the sequence number. If an ingestion and a write happen in between, the iterator reads at a sequence number above the ingested file, with a SuperVersion that does not have it. This also loses files ingested with a new seqno.
- Ingestions that flush the memtable are safe: the older iterator's memtable becomes immutable.

**Fix:** Barrier on the Version, not only on the sequence number.
- After `LogAndApply()`, ingestion records the new Version number on the active memtable (`ingest_version_barrier_`), while it still holds the per-CF `ingest_sst_lock`.
- `DBIter` passes the number of the Version it reads. `AddLogicallyRedundantRangeTombstone()` refuses the conversion if that number is below the barrier.
- Version numbers only go up, and every later Version of the column family has the ingested files. So the check is true exactly when the iterator cannot see an ingestion into this memtable. Iterators created after the ingestion convert as before.
- The seqno barrier stays: it still covers old explicit snapshots that read a newer Version.

**Related:** D126796014 (`Disable tombstone conversion for kMemtableTier iterators`) fixes a different trigger of the same feature.

Reviewed By: joshkang97, archang19

Differential Revision: D126875720

fbshipit-source-id: 5767b866c6174d791f9ddae78311816300ac49f3

db/db_flush_test.cc

@@ -4433,7 +4433,8 @@ TEST_F(DBFlushTest, FlushAfterReadPathRangeTombstoneInsertion) {
         // already immutable. Without the fix, this succeeds and creates
         // an entry count mismatch.
         cfh->cfd()->mem()->AddLogicallyRedundantRangeTombstone(
-            1 /* seq */, "b", "f", cfh->cfd()->GetIngestSstLock());
+            1 /* seq */, "b", "f", 0 /* version_number */,
+            cfh->cfd()->GetIngestSstLock());
       });
   SyncPoint::GetInstance()->EnableProcessing();
 

db/db_impl/db_impl.cc

@@ -7891,6 +7891,20 @@ Status DBImpl::CommitFileIngestionHandles(
           versions_->SetLastSequence(max_assigned_seqno);
         }
       }
+      // An iterator whose Version predates the ingestion cannot see the
+      // ingested files, whatever sequence number it reads at: files can get
+      // no new seqno (seqno zero, ingest_behind, DB-generated files), and an
+      // iterator can pair an older SuperVersion with a newer sequence number.
+      // Such iterators must not convert tombstones over the ingested files.
+      // We still hold the per-CF ingest_sst_lock as a ReadLock, so no
+      // conversion runs until the barrier is in place.
+      if (status.ok()) {
+        for (auto* job : ingestion_jobs) {
+          auto* cfd = job->GetColumnFamilyData();
+          cfd->mem()->BumpIngestVersionBarrier(
+              cfd->current()->GetVersionNumber());
+        }
+      }
     }
 
     for (auto* job : ingestion_jobs) {

db/db_iter.cc

@@ -96,6 +96,7 @@ DBIter::DBIter(Env* _env, const ReadOptions& read_options,
       trace_cf_id_(cfd != nullptr ? cfd->GetID() : 0),
       has_trace_state_(db_impl != nullptr && cfd != nullptr),
       ingest_sst_lock_(cfd != nullptr ? &cfd->GetIngestSstLock() : nullptr),
+      version_number_(version != nullptr ? version->GetVersionNumber() : 0),
       timestamp_ub_(read_options.timestamp),
       timestamp_lb_(read_options.iter_start_ts),
       timestamp_size_(timestamp_ub_ ? timestamp_ub_->size() : 0),
@@ -1833,7 +1834,7 @@ void DBIter::MaybeInsertRangeTombstone(const Slice& end_key) {
   assert(ingest_sst_lock_ != nullptr);
   if (active_mem_->AddLogicallyRedundantRangeTombstone(
           insert_seq, range_tomb_first_key_.GetUserKey(), end_key,
-          *ingest_sst_lock_)) {
+          version_number_, *ingest_sst_lock_)) {
     RecordTick(statistics_, READ_PATH_RANGE_TOMBSTONES_INSERTED);
     ROCKS_LOG_DEBUG(logger_,
                     "Inserted range tombstone [%s, %s) @ seq %" PRIu64

db/db_iter.h

@@ -713,6 +713,8 @@ class DBIter final : public Iterator {
   uint32_t trace_cf_id_;
   bool has_trace_state_;
   port::RWMutex* ingest_sst_lock_;
+  // Number of the Version this iterator reads, or 0 if unknown.
+  const uint64_t version_number_;
   const Slice* const timestamp_ub_;
   const Slice* const timestamp_lb_;
   const size_t timestamp_size_;

db/db_iterator_test.cc

@@ -8414,6 +8414,361 @@ TEST_P(ReadPathRangeTombstoneTest,
   db_->ReleaseSnapshot(snap);
 }
 
+// An ingested file that gets no new sequence number is invisible to an
+// iterator created before the ingestion. That iterator must not convert the
+// deleted keys around the file into a range tombstone that covers it.
+TEST_P(ReadPathRangeTombstoneTest, IngestedFileSurvivesOlderIterator) {
+  enum class IngestMode { kSeqnoZero, kIngestBehind, kDbGenerated };
+  for (IngestMode mode : {IngestMode::kSeqnoZero, IngestMode::kIngestBehind,
+                          IngestMode::kDbGenerated}) {
+    SCOPED_TRACE("ingest mode " + std::to_string(static_cast<int>(mode)));
+    Options options = CurrentOptions();
+    options.min_tombstones_for_range_conversion = 2;
+    options.disable_auto_compactions = true;
+    options.cf_allow_ingest_behind = mode == IngestMode::kIngestBehind;
+    DestroyAndReopen(options);
+
+    std::string file = dbname_ + "/ingest_c.sst";
+    SequenceNumber expected_file_seqno = 0;
+    if (mode == IngestMode::kDbGenerated) {
+      const std::string src_path = dbname_ + "_src";
+      ASSERT_OK(DestroyDB(src_path, options));
+      std::unique_ptr<DB> src;
+      ASSERT_OK(DB::Open(options, src_path, &src));
+      ASSERT_OK(src->Put(WriteOptions(), "c", "vc"));
+      ASSERT_OK(src->Flush(FlushOptions()));
+      std::vector<LiveFileMetaData> src_files;
+      src->GetLiveFilesMetaData(&src_files);
+      ASSERT_EQ(src_files.size(), 1);
+      expected_file_seqno = src_files[0].largest_seqno;
+      ASSERT_GT(expected_file_seqno, 0);
+      file = src_files[0].directory + "/" + src_files[0].relative_filename;
+      ASSERT_OK(src->Close());
+    } else {
+      SstFileWriter writer(EnvOptions(), options);
+      ASSERT_OK(writer.Open(file));
+      ASSERT_OK(writer.Put("c", "vc"));
+      ASSERT_OK(writer.Finish());
+    }
+
+    for (const char* key : {"a", "b", "d", "e", "f", "g"}) {
+      ASSERT_OK(Put(key, std::string("v") + key));
+    }
+    ASSERT_OK(Flush());
+    ASSERT_OK(Delete("b"));
+    ASSERT_OK(Delete("d"));
+    // The iterator sequence number must be above the DB-generated file's.
+    ASSERT_OK(Delete("f"));
+    ASSERT_GT(db_->GetLatestSequenceNumber(), expected_file_seqno);
+
+    std::unique_ptr<Iterator> iter(db_->NewIterator(ReadOptions()));
+
+    // "c" falls between the deleted keys and overlaps no key in the DB.
+    IngestExternalFileOptions ifo;
+    ifo.ingest_behind = mode == IngestMode::kIngestBehind;
+    ifo.allow_db_generated_files = mode == IngestMode::kDbGenerated;
+    ASSERT_OK(db_->IngestExternalFile({file}, ifo));
+    std::vector<LiveFileMetaData> files;
+    db_->GetLiveFilesMetaData(&files);
+    bool found = false;
+    for (const auto& f : files) {
+      if (f.smallestkey == "c") {
+        found = true;
+        ASSERT_EQ(f.level, options.num_levels - 1);
+        ASSERT_EQ(f.largest_seqno, expected_file_seqno);
+      }
+    }
+    ASSERT_TRUE(found);
+    ASSERT_EQ(Get("c"), "vc");
+
+    // The iterator predates the ingestion, so it sees b and d as one run of
+    // deleted keys.
+    attempted_insert_ranges_.clear();
+    if (Forward()) {
+      iter->SeekToFirst();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_EQ(iter->key(), "a");
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_EQ(iter->key(), "e");
+    } else {
+      iter->SeekToLast();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_EQ(iter->key(), "g");
+      iter->Prev();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_EQ(iter->key(), "e");
+      iter->Prev();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_EQ(iter->key(), "a");
+    }
+    ASSERT_OK(iter->status());
+    iter.reset();
+    ASSERT_EQ(attempted_insert_ranges_.size(), 1);
+    AssertRange(0, "b", "e");
+
+    EXPECT_EQ(Get("c"), "vc");
+    VerifyIteration({"a", "c", "e", "g"});
+    ASSERT_OK(Flush());
+    Reopen(options);
+    EXPECT_EQ(Get("c"), "vc");
+  }
+}
+
+// NewIterator() references the SuperVersion before it reads the sequence
+// number. An ingestion plus a write in between gives the iterator a sequence
+// number above the ingested files' while its SuperVersion lacks them. It must
+// not convert the deleted keys around those files into a range tombstone.
+TEST_P(ReadPathRangeTombstoneTest,
+       IngestedFileSurvivesIteratorWithOlderSuperVersion) {
+  for (bool new_seqno : {false, true}) {
+    SCOPED_TRACE(new_seqno ? "ingested with a new seqno"
+                           : "ingested with seqno zero");
+    Options options = CurrentOptions();
+    options.min_tombstones_for_range_conversion = 2;
+    options.disable_auto_compactions = true;
+    DestroyAndReopen(options);
+
+    for (const char* key : {"a", "b", "d", "e"}) {
+      ASSERT_OK(Put(key, std::string("v") + key));
+    }
+    ASSERT_OK(Flush());
+    ASSERT_OK(Delete("b"));
+    ASSERT_OK(Delete("d"));
+    // The memtable stays empty, so the ingestion does not switch it.
+    ASSERT_OK(Flush());
+
+    // "c" alone overlaps no key, so it gets seqno zero. Adding "b" overlaps
+    // the deleted "b" in L0, so the file gets a new seqno.
+    const std::string file = dbname_ + "/ingest.sst";
+    SstFileWriter writer(EnvOptions(), options);
+    ASSERT_OK(writer.Open(file));
+    if (new_seqno) {
+      ASSERT_OK(writer.Put("b", "vb2"));
+    }
+    ASSERT_OK(writer.Put("c", "vc"));
+    ASSERT_OK(writer.Finish());
+
+    SyncPoint::GetInstance()->LoadDependency(
+        {{"DBImpl::NewIterator:1",
+          "ReadPathRangeTombstoneTest::OlderSuperVersion:Ingest"},
+         {"ReadPathRangeTombstoneTest::OlderSuperVersion:Written",
+          "DBImpl::NewIterator:2"}});
+    std::unique_ptr<Iterator> iter;
+    port::Thread reader([&]() { iter.reset(db_->NewIterator(ReadOptions())); });
+    TEST_SYNC_POINT("ReadPathRangeTombstoneTest::OlderSuperVersion:Ingest");
+    ASSERT_OK(db_->IngestExternalFile({file}, IngestExternalFileOptions()));
+    ASSERT_OK(Put("z", "vz"));
+    TEST_SYNC_POINT("ReadPathRangeTombstoneTest::OlderSuperVersion:Written");
+    reader.join();
+    SyncPoint::GetInstance()->ClearTrace();
+    SyncPoint::GetInstance()->LoadDependency({});
+
+    std::vector<LiveFileMetaData> files;
+    db_->GetLiveFilesMetaData(&files);
+    bool found = false;
+    for (const auto& f : files) {
+      if (f.largestkey == "c") {
+        found = true;
+        ASSERT_EQ(f.largest_seqno > 0, new_seqno);
+      }
+    }
+    ASSERT_TRUE(found);
+
+    // The iterator reads past the write of "z" but not the ingested file.
+    attempted_insert_ranges_.clear();
+    std::vector<std::string> seen;
+    if (Forward()) {
+      for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
+        seen.push_back(iter->key().ToString());
+      }
+    } else {
+      for (iter->SeekToLast(); iter->Valid(); iter->Prev()) {
+        seen.insert(seen.begin(), iter->key().ToString());
+      }
+    }
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(seen, std::vector<std::string>({"a", "e", "z"}));
+    iter.reset();
+    ASSERT_EQ(attempted_insert_ranges_.size(), 1);
+    AssertRange(0, "b", "e");
+
+    EXPECT_EQ(Get("c"), "vc");
+    EXPECT_EQ(Get("b"), new_seqno ? "vb2" : "NOT_FOUND");
+    ASSERT_OK(Flush());
+    Reopen(options);
+    EXPECT_EQ(Get("c"), "vc");
+  }
+}
+
+// One IngestExternalFiles() call installs files in two column families. Each
+// column family's older iterator must refuse its conversion.
+TEST_P(ReadPathRangeTombstoneTest, AtomicIngestionSurvivesOlderIterators) {
+  Options options = CurrentOptions();
+  options.min_tombstones_for_range_conversion = 2;
+  options.disable_auto_compactions = true;
+  DestroyAndReopen(options);
+  CreateAndReopenWithCF({"pikachu"}, options);
+  std::vector<std::string> files;
+  for (int cf = 0; cf < 2; cf++) {
+    files.push_back(dbname_ + "/ingest_c" + std::to_string(cf) + ".sst");
+    SstFileWriter writer(EnvOptions(), options);
+    ASSERT_OK(writer.Open(files.back()));
+    ASSERT_OK(writer.Put("c", "vc"));
+    ASSERT_OK(writer.Finish());
+    for (const char* key : {"a", "b", "d", "e"}) {
+      ASSERT_OK(Put(cf, key, std::string("v") + key));
+    }
+    ASSERT_OK(Flush(cf));
+    ASSERT_OK(Delete(cf, "b"));
+    ASSERT_OK(Delete(cf, "d"));
+  }
+  std::unique_ptr<Iterator> it0(db_->NewIterator(ReadOptions(), handles_[0]));
+  std::unique_ptr<Iterator> it1(db_->NewIterator(ReadOptions(), handles_[1]));
+  std::vector<IngestExternalFileArg> args(2);
+  for (int cf = 0; cf < 2; cf++) {
+    args[cf].column_family = handles_[cf];
+    args[cf].external_files = {files[cf]};
+  }
+  ASSERT_OK(db_->IngestExternalFiles(args));
+
+  attempted_insert_ranges_.clear();
+  for (auto* it : {it0.get(), it1.get()}) {
+    if (Forward()) {
+      for (it->SeekToFirst(); it->Valid(); it->Next()) {
+      }
+    } else {
+      for (it->SeekToLast(); it->Valid(); it->Prev()) {
+      }
+    }
+    ASSERT_OK(it->status());
+  }
+  it0.reset();
+  it1.reset();
+  EXPECT_EQ(attempted_insert_ranges_.size(), 2);
+  EXPECT_EQ(Get(0, "c"), "vc");
+  EXPECT_EQ(Get(1, "c"), "vc");
+}
+
+// An iterator created after an ingestion sees the ingested files, so the
+// barrier must still let it convert.
+TEST_P(ReadPathRangeTombstoneTest, IteratorCreatedAfterIngestionConverts) {
+  for (bool seqno_zero : {true, false}) {
+    SCOPED_TRACE(seqno_zero ? "ingested with seqno zero"
+                            : "ingested with a new seqno");
+    Options options = CurrentOptions();
+    options.min_tombstones_for_range_conversion = 2;
+    options.disable_auto_compactions = true;
+    options.statistics = CreateDBStatistics();
+    DestroyAndReopen(options);
+
+    // "x" alone overlaps no key, so it gets seqno zero. Adding "e" overlaps
+    // the flushed "e", so the file gets a new seqno.
+    const std::string file = dbname_ + "/ingest_x.sst";
+    SstFileWriter writer(EnvOptions(), options);
+    ASSERT_OK(writer.Open(file));
+    if (!seqno_zero) {
+      ASSERT_OK(writer.Put("e", "ve2"));
+    }
+    ASSERT_OK(writer.Put("x", "vx"));
+    ASSERT_OK(writer.Finish());
+    for (const char* key : {"a", "b", "c", "d", "e"}) {
+      ASSERT_OK(Put(key, std::string("v") + key));
+    }
+    ASSERT_OK(Flush());
+    ASSERT_OK(Delete("b"));
+    ASSERT_OK(Delete("c"));
+    ASSERT_OK(Delete("d"));
+    IngestExternalFileOptions ifo;
+    ifo.allow_blocking_flush = false;
+    ASSERT_OK(db_->IngestExternalFile({file}, ifo));
+
+    {
+      std::unique_ptr<Iterator> it(db_->NewIterator(ReadOptions()));
+      for (it->SeekToFirst(); it->Valid(); it->Next()) {
+      }
+      ASSERT_OK(it->status());
+    }
+    EXPECT_EQ(TestGetTickerCount(options, READ_PATH_RANGE_TOMBSTONES_INSERTED),
+              1);
+    VerifyIteration({"a", "e", "x"});
+    EXPECT_EQ(Get("e"), seqno_zero ? "ve" : "ve2");
+  }
+}
+
+// A conversion that runs before an ingestion leaves a range tombstone in the
+// memtable over the gap where the file lands. The ingestion must either fail
+// or make the file visible, whatever the ingestion options.
+TEST_P(ReadPathRangeTombstoneTest, IngestionAfterConversionIsNeverHidden) {
+  enum class IngestMode {
+    kNoGlobalSeqnoNoFlush,
+    kNoGlobalSeqno,
+    kGlobalSeqno,
+    kDbGenerated
+  };
+  for (IngestMode mode :
+       {IngestMode::kNoGlobalSeqnoNoFlush, IngestMode::kNoGlobalSeqno,
+        IngestMode::kGlobalSeqno, IngestMode::kDbGenerated}) {
+    SCOPED_TRACE("ingest mode " + std::to_string(static_cast<int>(mode)));
+    Options options = CurrentOptions();
+    options.min_tombstones_for_range_conversion = 2;
+    options.disable_auto_compactions = true;
+    options.statistics = CreateDBStatistics();
+    DestroyAndReopen(options);
+
+    std::string file = dbname_ + "/ingest_c.sst";
+    if (mode == IngestMode::kDbGenerated) {
+      const std::string src_path = dbname_ + "_src";
+      ASSERT_OK(DestroyDB(src_path, options));
+      std::unique_ptr<DB> src;
+      ASSERT_OK(DB::Open(options, src_path, &src));
+      ASSERT_OK(src->Put(WriteOptions(), "c", "vc"));
+      ASSERT_OK(src->Flush(FlushOptions()));
+      std::vector<LiveFileMetaData> src_files;
+      src->GetLiveFilesMetaData(&src_files);
+      ASSERT_EQ(src_files.size(), 1);
+      file = src_files[0].directory + "/" + src_files[0].relative_filename;
+      ASSERT_OK(src->Close());
+    } else {
+      SstFileWriter writer(EnvOptions(), options);
+      ASSERT_OK(writer.Open(file));
+      ASSERT_OK(writer.Put("c", "vc"));
+      ASSERT_OK(writer.Finish());
+    }
+
+    for (const char* key : {"a", "b", "d", "e"}) {
+      ASSERT_OK(Put(key, std::string("v") + key));
+    }
+    ASSERT_OK(Flush());
+    ASSERT_OK(Delete("b"));
+    ASSERT_OK(Delete("d"));
+    attempted_insert_ranges_.clear();
+    VerifyIteration({"a", "e"});
+    ASSERT_EQ(TestGetTickerCount(options, READ_PATH_RANGE_TOMBSTONES_INSERTED),
+              1);
+    AssertRange(0, "b", "e");
+
+    IngestExternalFileOptions ifo;
+    ifo.allow_global_seqno = mode == IngestMode::kGlobalSeqno;
+    ifo.allow_blocking_flush = mode != IngestMode::kNoGlobalSeqnoNoFlush;
+    ifo.allow_db_generated_files = mode == IngestMode::kDbGenerated;
+    Status s = db_->IngestExternalFile({file}, ifo);
+    if (mode == IngestMode::kGlobalSeqno) {
+      ASSERT_OK(s);
+    } else {
+      ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
+    }
+
+    const std::string expected_c = s.ok() ? "vc" : "NOT_FOUND";
+    EXPECT_EQ(Get("c"), expected_c);
+    VerifyIteration(s.ok() ? std::vector<std::string>{"a", "c", "e"}
+                           : std::vector<std::string>{"a", "e"});
+    ASSERT_OK(Flush());
+    Reopen(options);
+    EXPECT_EQ(Get("c"), expected_c);
+  }
+}
+
 // SeekForPrev uses the reverse path: PrevInternal -> FindValueForCurrentKey.
 // When a key has more versions than max_sequential_skip_in_iterations,
 // FindValueForCurrentKey delegates to FindValueForCurrentKeyUsingSeek which

db/memtable.cc

@@ -972,7 +972,7 @@ void MemTable::ConstructFragmentedRangeTombstones() {
 
 bool MemTable::AddLogicallyRedundantRangeTombstone(
     SequenceNumber seq, const Slice& start_key, const Slice& end_key,
-    port::RWMutex& ingest_sst_lock) {
+    uint64_t version_number, port::RWMutex& ingest_sst_lock) {
   // Fast path: skip if already immutable.
   if (is_immutable_.LoadRelaxed()) {
     return false;
@@ -1016,6 +1016,11 @@ bool MemTable::AddLogicallyRedundantRangeTombstone(
   if (seq < ingest_seqno_barrier_.LoadRelaxed()) {
     return false;
   }
+  // The iterator's Version predates an ingestion, so the deleted run it saw
+  // can surround ingested keys that it cannot see.
+  if (version_number < ingest_version_barrier_.LoadRelaxed()) {
+    return false;
+  }
 
   MemTablePostProcessInfo post_process_info;
   Status s = Add(seq, kTypeRangeDeletion, start_key, end_key,
@@ -1034,6 +1039,12 @@ void MemTable::BumpIngestSeqnoBarrier(SequenceNumber y) {
   }
 }
 
+void MemTable::BumpIngestVersionBarrier(uint64_t version_number) {
+  if (ingest_version_barrier_.LoadRelaxed() < version_number) {
+    ingest_version_barrier_.StoreRelaxed(version_number);
+  }
+}
+
 port::RWMutex* MemTable::GetLock(const Slice& key) {
   return &locks_[GetSliceRangedNPHash(key, locks_.size())];
 }

db/memtable.h

@@ -337,10 +337,14 @@ class ReadOnlyMemTable {
   // - the per-memtable ingest seqno barrier already exceeds `seq` (an
   //   ingestion has committed an L0 file at a seq that this converted
   //   tombstone would shadow) or an ingestion is in progress.
+  // - `version_number`, the number of the Version the converting iterator
+  //   reads, is below the per-memtable ingest version barrier (the iterator
+  //   cannot see a file ingested while this memtable was active).
   // Returns true if the range tombstone was inserted, false if skipped.
   virtual bool AddLogicallyRedundantRangeTombstone(
       SequenceNumber /*seq*/, const Slice& /*start_key*/,
-      const Slice& /*end_key*/, port::RWMutex& /*ingest_sst_lock*/) {
+      const Slice& /*end_key*/, uint64_t /*version_number*/,
+      port::RWMutex& /*ingest_sst_lock*/) {
     return false;
   }
 
@@ -899,7 +903,7 @@ class MemTable final : public ReadOnlyMemTable {
 
   bool AddLogicallyRedundantRangeTombstone(
       SequenceNumber seq, const Slice& start_key, const Slice& end_key,
-      port::RWMutex& ingest_sst_lock) override;
+      uint64_t version_number, port::RWMutex& ingest_sst_lock) override;
 
   // Monotonically raises ingest_seqno_barrier_ to `y` (no-op if `y` is not
   // greater than the current value). The conversion's barrier check
@@ -911,6 +915,13 @@ class MemTable final : public ReadOnlyMemTable {
   // a CAS loop. Only IngestExternalFiles calls this.
   void BumpIngestSeqnoBarrier(SequenceNumber y);
 
+  // Monotonically raises ingest_version_barrier_ to `version_number`, the
+  // number of the Version that installed an ingestion.
+  //
+  // REQUIRES: DB mutex held by the caller. Only IngestExternalFiles calls
+  // this.
+  void BumpIngestVersionBarrier(uint64_t version_number);
+
   bool IsFragmentedRangeTombstonesConstructed() const override {
     return fragmented_range_tombstone_list_.get() != nullptr ||
            is_range_del_table_empty_.LoadRelaxed();
@@ -975,6 +986,10 @@ class MemTable final : public ReadOnlyMemTable {
   // ColumnFamilyData::ingest_sst_lock_.
   RelaxedAtomic<SequenceNumber> ingest_seqno_barrier_{0};
 
+  // Number of the latest Version that installed an ingestion. See also
+  // ColumnFamilyData::ingest_sst_lock_.
+  RelaxedAtomic<uint64_t> ingest_version_barrier_{0};
+
   SequenceNumber creation_seq_;
 
   // the earliest log containing a prepared section

include/rocksdb/advanced_options.h

@@ -1440,7 +1440,8 @@ struct AdvancedColumnFamilyOptions {
   //     and the snapshot seq is at or above its min uncommitted seq.
   //   * An IngestExternalFile call is currently in flight on this column
   //     family OR the inserted range tombstone seqno would be lower than the
-  //     ingested file seqno.
+  //     ingested file seqno OR the iterator's Version predates a file
+  //     ingested while the active memtable was current.
   //
   // Read-write iterators using ReadOptions::table_filter are rejected while
   // this option is enabled, see more details in ReadOptions::table_filter

unreleased_history/bug_fixes/range_conversion_ingest_older_iterator.md

@@ -0,0 +1 @@
+Fixed data loss with `min_tombstones_for_range_conversion` and external file ingestion. An iterator whose view predated the ingestion could convert the deleted keys around the ingested keys into a range tombstone that deleted them. This happened when the file got no new sequence number (sequence number zero, `ingest_behind`, or `allow_db_generated_files`), and when an iterator was created during the ingestion.