Fix prefix_same_as_start iterator correctness gaps

Summary:
Why the change is required:
The crash test found prefix_same_as_start iterators returning wrong keys in
two ways, both caused by child iterators using prefix filters (SST prefix
filter, memtable prefix bloom, LevelIterator prefix exhaustion) for a key in a
different prefix than the one the caller iterates:
1. SeekToFirst() without iterate_lower_bound and SeekToLast() without
   iterate_upper_bound have no seek key to take the prefix from. With a range
   tombstone that crosses a prefix boundary, MergingIterator reseeks older
   children to the tombstone boundary key, a prefix filter can drop a child,
   and the scan can return a wrong first key. Read-path tombstone conversion
   (min_tombstones_for_range_conversion, EXPERIMENTAL) trusts these scans, so
   a skipped live key could be covered by a converted tombstone and deleted
   for good.
2. With prefix_same_as_start and total_order_seek both set, a direction change
   (Next() after Prev()) can skip keys inside the seek prefix. Under
   DBOptions::prefix_seek_opt_in_only, total_order_seek is forced on, so every
   prefix_same_as_start iterator is exposed.

Previous behavior:
1. Unbounded SeekToFirst()/SeekToLast() in prefix seek mode were never
   promised to match total order (the option docs call keys outside the seek
   prefix undefined, and the Prefix-Seek wiki says these calls are not well
   supported), but they ran and could return wrong results.
2. Since prefix_seek_opt_in_only was added (9.7.0), prefix_same_as_start and
   auto_prefix_mode enable prefix filtering even with total_order_seek=true,
   so for iterators total_order_seek only turns off the implicit prefix seek
   of default ReadOptions. This was stated in that release note, but not in
   options.h, where total_order_seek still read as "always total order". Each
   iterator layer read total_order_seek on its own, and DBIter still
   treated total_order_seek=true as "the inner iterator is total order". So
   DBIter::ReverseToForward() skipped its reseek to the current key, and
   MergingIterator::SwitchToForward() reseeked the other children to the
   previous key, which can have another prefix. A child without that prefix
   was dropped by its prefix filter, losing its keys in the seek prefix.

New behavior:
1. With DBOptions::prefix_seek_opt_in_only, a prefix_same_as_start iterator
   leaves itself !Valid() with Status::NotSupported on SeekToFirst() without
   iterate_lower_bound and SeekToLast() without iterate_upper_bound. The
   bounded forms still become Seek()/SeekForPrev() and work. Without
   prefix_seek_opt_in_only, iteration results of these calls are unchanged,
   but a scan they start under prefix_same_as_start no longer creates
   read-path range tombstones, until the next Seek()/SeekForPrev().
2. prefix_same_as_start and auto_prefix_mode take precedence over
   total_order_seek in every iterator layer. A prefix_same_as_start iterator
   behaves the same with or without total_order_seek, and so with or without
   prefix_seek_opt_in_only for Seek()-started scans.
Total order and auto_prefix_mode iteration results are not affected.

Implementation:
- GetIteratorPrefixMode() in options_helper.h decides once how an iterator
  uses the prefix extractor (kTotalOrder, kAuto, kPrefix), with the order
  prefix_same_as_start > auto_prefix_mode > total_order_seek > implicit prefix
  seek. DBIter, the MergingIterator prefix flag, LevelIterator, MemTable, the
  block-based table filter and hash index, and PlainTable use it instead of
  reading total_order_seek themselves. Only DBIter's total-order assumption is
  a correctness fix; for the others the result is unchanged or the
  prefix_same_as_start + total_order_seek combination gets the same prefix
  optimizations as plain prefix_same_as_start.
- DBIter returns NotSupported from unbounded SeekToFirst()/SeekToLast() for
  prefix_same_as_start under prefix_seek_opt_in_only. It also remembers when
  the current scan was started by an unbounded SeekToFirst()/SeekToLast()
  under prefix_same_as_start, and MaybeInsertRangeTombstone(), the only place
  that writes a converted tombstone, does nothing then.
- options.h: ReadOptions::total_order_seek now starts with its iterator
  meaning (turns off the implicit prefix seek; prefix_same_as_start and
  auto_prefix_mode take precedence; forced on by prefix_seek_opt_in_only),
  auto_prefix_mode notes that prefix_same_as_start takes precedence, and
  prefix_seek_opt_in_only notes that prefix_same_as_start iterators still use
  prefix seek. prefix_seek_opt_in_only and prefix_same_as_start document the
  unbounded SeekToFirst()/SeekToLast() behavior.
- Release notes: a behavior change for the NotSupported and conversion change,
  and a bug fix for the direction change, which also states that
  prefix_same_as_start now consistently takes precedence over total_order_seek
  and auto_prefix_mode.
- db_stress: new prefix_seek_opt_in_only flag, randomized by db_crashtest.py.
  TestIterate sometimes sets total_order_seek together with
  prefix_same_as_start, and VerifyIterator uses GetIteratorPrefixMode(). Under
  prefix_seek_opt_in_only it checks that unbounded SeekToFirst()/SeekToLast()
  return NotSupported. Otherwise the prefix_same_as_start SeekToFirst() check
  only runs when no range tombstone can exist (no DeleteRange, no standalone
  range deletion ingestion, no read-path conversion), since it would keep
  hitting the known limitation.

Safety, compatibility, and limitations:
- No API or format change. Users with prefix_seek_opt_in_only who call
  unbounded SeekToFirst()/SeekToLast() on a prefix_same_as_start iterator now
  get NotSupported; they can set a bound, or use total order or
  auto_prefix_mode.
- prefix_same_as_start with total_order_seek or prefix_seek_opt_in_only now
  reseeks on a direction change, the cost plain prefix_same_as_start already
  pays. prefix_same_as_start with auto_prefix_mode now uses the SST prefix
  filter for the seek prefix, as the memtable already did.
- Without prefix_seek_opt_in_only, unbounded SeekToFirst()/SeekToLast() under
  prefix seek can still skip live keys when range tombstones cross prefix
  boundaries, and unbounded legacy prefix-mode scans have other known holes.
  These stay documented limitations.

Differential Revision: D129901533

Author: xingbowang

db/db_impl/db_impl.cc

@@ -3288,9 +3288,8 @@ InternalIterator* DBImpl::NewInternalIterator(
   // Need to create internal iterator from the arena.
   MergeIteratorBuilder merge_iter_builder(
       &cfd->internal_comparator(), arena,
-      // FIXME? It's not clear what interpretation of prefix seek is needed
-      // here, and no unit test cares about the value provided here.
-      !read_options.total_order_seek && prefix_extractor != nullptr,
+      GetIteratorPrefixMode(read_options, prefix_extractor != nullptr) ==
+          IteratorPrefixMode::kPrefix,
       read_options.iterate_upper_bound);
   Status s;
   const Comparator* user_comparator = cfd->user_comparator();

db/db_iter.cc

@@ -114,10 +114,13 @@ DBIter::DBIter(Env* _env, const ReadOptions& read_options,
       is_key_seqnum_zero_(false),
       prefix_same_as_start_(
           prefix_extractor_ ? read_options.prefix_same_as_start : false),
+      unbounded_scan_not_supported_(prefix_same_as_start_ &&
+                                    ioptions.prefix_seek_opt_in_only),
+      unbounded_prefix_scan_(false),
       pin_thru_lifetime_(read_options.pin_data),
-      expect_total_order_inner_iter_(prefix_extractor_ == nullptr ||
-                                     read_options.total_order_seek ||
-                                     read_options.auto_prefix_mode),
+      expect_total_order_inner_iter_(
+          GetIteratorPrefixMode(read_options, prefix_extractor_ != nullptr) !=
+          IteratorPrefixMode::kPrefix),
       // Read-path range conversion assumes the scan can observe all interior
       // live keys. table_filter can hide whole SSTs, read tiers other than
       // kReadAllTier can skip SSTs or uncached blocks, and timestamp filtering
@@ -1779,7 +1782,8 @@ void DBIter::FlushPendingTombstoneRun(const Slice& end_key,
 }
 
 void DBIter::MaybeInsertRangeTombstone(const Slice& end_key) {
-  if (contiguous_tombstone_count_ < min_tombstones_for_range_conversion_) {
+  if (contiguous_tombstone_count_ < min_tombstones_for_range_conversion_ ||
+      unbounded_prefix_scan_) {
     return;
   }
 
@@ -2031,6 +2035,7 @@ void DBIter::Seek(const Slice& target) {
   PERF_COUNTER_ADD(iter_seek_count, 1);
   PERF_CPU_TIMER_GUARD(iter_seek_cpu_nanos, clock_);
   StopWatch sw(clock_, statistics_, DB_SEEK);
+  unbounded_prefix_scan_ = false;
 
   if (scan_opts_.has_value()) {
     if (scan_opts_->reverse) {
@@ -2147,6 +2152,7 @@ void DBIter::SeekForPrev(const Slice& target) {
   PERF_COUNTER_ADD(iter_seek_count, 1);
   PERF_CPU_TIMER_GUARD(iter_seek_cpu_nanos, clock_);
   StopWatch sw(clock_, statistics_, DB_SEEK);
+  unbounded_prefix_scan_ = false;
 
   if (scan_opts_.has_value()) {
     if (!scan_opts_->reverse) {
@@ -2272,6 +2278,14 @@ void DBIter::SeekToFirst() {
     Seek(*iterate_lower_bound_);
     return;
   }
+  if (unbounded_scan_not_supported_) {
+    valid_ = false;
+    status_ = Status::NotSupported(
+        "SeekToFirst() without iterate_lower_bound is not supported with "
+        "prefix_same_as_start under prefix_seek_opt_in_only");
+    return;
+  }
+  unbounded_prefix_scan_ = prefix_same_as_start_;
   PERF_COUNTER_ADD(iter_seek_count, 1);
   PERF_CPU_TIMER_GUARD(iter_seek_cpu_nanos, clock_);
   // Don't use iter_::Seek() if we set a prefix extractor
@@ -2336,6 +2350,14 @@ void DBIter::SeekToLast() {
 #endif
     return;
   }
+  if (unbounded_scan_not_supported_) {
+    valid_ = false;
+    status_ = Status::NotSupported(
+        "SeekToLast() without iterate_upper_bound is not supported with "
+        "prefix_same_as_start under prefix_seek_opt_in_only");
+    return;
+  }
+  unbounded_prefix_scan_ = prefix_same_as_start_;
 
   PERF_COUNTER_ADD(iter_seek_count, 1);
   PERF_CPU_TIMER_GUARD(iter_seek_cpu_nanos, clock_);

db/db_iter.h

@@ -738,6 +738,18 @@ class DBIter final : public Iterator {
   // user key.
   bool is_key_seqnum_zero_;
   const bool prefix_same_as_start_;
+  // prefix_same_as_start_ under DBOptions::prefix_seek_opt_in_only.
+  // SeekToFirst() without iterate_lower_bound and SeekToLast() without
+  // iterate_upper_bound set NotSupported: they have no seek key to take the
+  // prefix from.
+  const bool unbounded_scan_not_supported_;
+  // True after SeekToFirst() without iterate_lower_bound or SeekToLast()
+  // without iterate_upper_bound under prefix_same_as_start, until the next
+  // Seek() or SeekForPrev(). Such a scan is not limited to one prefix while it
+  // looks for the first key, but prefix seek in the child iterators can skip
+  // live keys outside the seek prefix. Read-path range tombstone conversion is
+  // disabled so that a skipped live key cannot be covered by a new tombstone.
+  bool unbounded_prefix_scan_;
   // Means that we will pin all data blocks we read as long the Iterator
   // is not deleted, will be true if ReadOptions::pin_data is true
   const bool pin_thru_lifetime_;

db/db_iterator_test.cc

@@ -87,6 +87,147 @@ class DBIteratorBaseTest : public DBTestBase {
 };
 
 #ifdef OS_LINUX
+TEST_F(DBIteratorBaseTest, PrefixSameAsStartDirectionChange) {
+  // prefix_same_as_start uses prefix seek even with total_order_seek=true (as
+  // forced by prefix_seek_opt_in_only) or auto_prefix_mode. On Next() after
+  // Prev(), the inner iterator stands on a key with another prefix (bbbb1).
+  // DBIter must reseek it to the current key; moving it without a reseek
+  // seeks the other children to bbbb1, where a prefix filter drops the child
+  // holding kkkk1 and kkkk3.
+  struct ReadCase {
+    const char* name;
+    bool opt_in;
+    bool total_order_seek;
+    bool auto_prefix_mode;
+  };
+  const std::vector<ReadCase> read_cases = {
+      {"prefix_same_as_start", false, false, false},
+      {"+total_order_seek", false, true, false},
+      {"+auto_prefix_mode", false, false, true},
+      {"+prefix_seek_opt_in_only", true, false, false},
+  };
+  for (bool memtable_bloom : {false, true}) {
+    for (const ReadCase& rc : read_cases) {
+      SCOPED_TRACE(std::string(rc.name) +
+                   (memtable_bloom ? " memtable_bloom" : " sst_filter"));
+      Options options = CurrentOptions();
+      options.disable_auto_compactions = true;
+      options.prefix_extractor.reset(NewFixedPrefixTransform(4));
+      options.prefix_seek_opt_in_only = rc.opt_in;
+      BlockBasedTableOptions table_options;
+      if (memtable_bloom) {
+        options.memtable_prefix_bloom_size_ratio = 0.1;
+      } else {
+        table_options.filter_policy.reset(NewBloomFilterPolicy(10, false));
+        table_options.whole_key_filtering = false;
+      }
+      options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+      DestroyAndReopen(options);
+      // The child that gets filtered has prefix kkkk only: the SST for the
+      // SST filter case, the memtable for the memtable bloom case.
+      const std::vector<std::string> kkkk_only = {"kkkk1", "kkkk3"};
+      const std::vector<std::string> mixed = {"bbbb1", "kkkk2"};
+      for (const std::string& key : memtable_bloom ? mixed : kkkk_only) {
+        ASSERT_OK(Put(key, "v"));
+      }
+      ASSERT_OK(Flush());
+      for (const std::string& key : memtable_bloom ? kkkk_only : mixed) {
+        ASSERT_OK(Put(key, "v"));
+      }
+
+      ReadOptions ro;
+      ro.prefix_same_as_start = true;
+      ro.total_order_seek = rc.total_order_seek;
+      ro.auto_prefix_mode = rc.auto_prefix_mode;
+      std::unique_ptr<Iterator> iter(db_->NewIterator(ro));
+      iter->Seek("kkkk1");
+      std::vector<std::string> keys;
+      // Seek, Next, Prev, then Next until the end of the prefix.
+      for (int step = 0; iter->Valid(); ++step) {
+        keys.push_back(iter->key().ToString());
+        if (step == 1) {
+          iter->Prev();
+        } else {
+          iter->Next();
+        }
+      }
+      ASSERT_OK(iter->status());
+      const std::vector<std::string> expected = {"kkkk1", "kkkk2", "kkkk1",
+                                                 "kkkk2", "kkkk3"};
+      EXPECT_EQ(expected, keys);
+    }
+  }
+}
+
+TEST_F(DBIteratorBaseTest, PrefixSameAsStartWithExplicitBounds) {
+  // prefix_same_as_start implies a bound at the end of the seek prefix. With
+  // or without auto_prefix_mode, the iterator returns only keys within both
+  // that implied bound and the explicit iterate_lower_bound /
+  // iterate_upper_bound, whichever is narrower.
+  Options options = CurrentOptions();
+  options.disable_auto_compactions = true;
+  options.prefix_extractor.reset(NewFixedPrefixTransform(4));
+  BlockBasedTableOptions table_options;
+  table_options.filter_policy.reset(NewBloomFilterPolicy(10, false));
+  table_options.whole_key_filtering = false;
+  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+  DestroyAndReopen(options);
+  for (const char* key : {"kkkj9", "kkkk1", "kkkk5"}) {
+    ASSERT_OK(Put(key, "v"));
+  }
+  ASSERT_OK(Flush());
+  for (const char* key : {"kkkk3", "kkkl1"}) {
+    ASSERT_OK(Put(key, "v"));
+  }
+
+  struct BoundCase {
+    const char* name;
+    const char* bound;
+    std::vector<std::string> expected;
+  };
+  const std::vector<BoundCase> forward_cases = {
+      {"upper bound inside prefix", "kkkk4", {"kkkk1", "kkkk3"}},
+      {"upper bound past prefix", "kkkm", {"kkkk1", "kkkk3", "kkkk5"}},
+  };
+  const std::vector<BoundCase> backward_cases = {
+      {"lower bound inside prefix", "kkkk2", {"kkkk5", "kkkk3"}},
+      {"lower bound before prefix", "kkkj", {"kkkk5", "kkkk3", "kkkk1"}},
+  };
+  for (bool auto_prefix_mode : {false, true}) {
+    ReadOptions ro;
+    ro.prefix_same_as_start = true;
+    ro.auto_prefix_mode = auto_prefix_mode;
+    for (const BoundCase& bc : forward_cases) {
+      SCOPED_TRACE(std::string(bc.name) +
+                   (auto_prefix_mode ? " auto_prefix_mode" : ""));
+      Slice upper_bound(bc.bound);
+      ro.iterate_upper_bound = &upper_bound;
+      std::unique_ptr<Iterator> iter(db_->NewIterator(ro));
+      std::vector<std::string> keys;
+      for (iter->Seek("kkkk1"); iter->Valid(); iter->Next()) {
+        keys.push_back(iter->key().ToString());
+      }
+      ASSERT_OK(iter->status());
+      EXPECT_EQ(bc.expected, keys);
+    }
+    ro.iterate_upper_bound = nullptr;
+    for (const BoundCase& bc : backward_cases) {
+      SCOPED_TRACE(std::string(bc.name) +
+                   (auto_prefix_mode ? " auto_prefix_mode" : ""));
+      Slice lower_bound(bc.bound);
+      ro.iterate_lower_bound = &lower_bound;
+      std::unique_ptr<Iterator> iter(db_->NewIterator(ro));
+      std::vector<std::string> keys;
+      for (iter->SeekForPrev("kkkk5"); iter->Valid(); iter->Prev()) {
+        keys.push_back(iter->key().ToString());
+      }
+      ASSERT_OK(iter->status());
+      EXPECT_EQ(bc.expected, keys);
+    }
+    ro.iterate_lower_bound = nullptr;
+  }
+}
+
 TEST_F(DBIteratorBaseTest, PrevChecksQueryAbort) {
   ASSERT_OK(Put("a", "va"));
   ASSERT_OK(Put("b", "vb"));

db/db_range_del_test.cc

@@ -3165,6 +3165,146 @@ TEST_F(DBRangeDelTest, PrefixSentinelKey) {
   delete iter;
 }
 
+namespace {
+Options PrefixFilterOptions(const Options& base) {
+  Options options = base;
+  options.disable_auto_compactions = true;
+  options.prefix_extractor.reset(NewFixedPrefixTransform(4));
+  BlockBasedTableOptions table_options;
+  table_options.filter_policy.reset(NewBloomFilterPolicy(10, false));
+  table_options.whole_key_filtering = false;
+  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+  return options;
+}
+
+// Scans the whole DB from SeekToFirst() (or SeekToLast() if !forward).
+std::vector<std::string> ScanAllKeys(DB* db, const ReadOptions& ro,
+                                     bool forward) {
+  std::vector<std::string> keys;
+  std::unique_ptr<Iterator> iter(db->NewIterator(ro));
+  if (forward) {
+    for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
+      keys.push_back(iter->key().ToString());
+    }
+  } else {
+    for (iter->SeekToLast(); iter->Valid(); iter->Prev()) {
+      keys.push_back(iter->key().ToString());
+    }
+  }
+  EXPECT_OK(iter->status());
+  return keys;
+}
+}  // namespace
+
+// Older data with prefixes aaaa, kkkk and xxxx only; newer tombstones whose
+// boundary keys have prefixes cccc and pppp, which the older data lacks.
+class DBRangeDelCrossPrefixTest : public DBRangeDelTest {
+ protected:
+  void Populate(const Options& options, int older_level) {
+    DestroyAndReopen(options);
+    ASSERT_OK(Put("aaaa1", "v"));
+    ASSERT_OK(Put("kkkk1", "v"));
+    ASSERT_OK(Put("kkkk5", "v"));
+    ASSERT_OK(Put("xxxx1", "v"));
+    ASSERT_OK(Flush());
+    if (older_level > 0) {
+      MoveFilesToLevel(older_level);
+      ASSERT_EQ(1, NumTableFilesAtLevel(older_level));
+    }
+    ASSERT_OK(db_->DeleteRange(WriteOptions(), db_->DefaultColumnFamily(),
+                               "aaaa", "cccc2"));
+    ASSERT_OK(db_->DeleteRange(WriteOptions(), db_->DefaultColumnFamily(),
+                               "pppp", "zzzz"));
+  }
+  const std::vector<std::string> kLiveKeys = {"kkkk1", "kkkk5"};
+  const std::vector<std::string> kLiveKeysReversed = {"kkkk5", "kkkk1"};
+};
+
+TEST_F(DBRangeDelCrossPrefixTest, PrefixSameAsStartUnboundedScanNotSupported) {
+  Options options = PrefixFilterOptions(CurrentOptions());
+  options.prefix_seek_opt_in_only = true;
+  Populate(options, /*older_level=*/1);
+
+  ReadOptions ro;
+  ro.prefix_same_as_start = true;
+  {
+    std::unique_ptr<Iterator> iter(db_->NewIterator(ro));
+    iter->SeekToFirst();
+    ASSERT_FALSE(iter->Valid());
+    ASSERT_TRUE(iter->status().IsNotSupported());
+    iter->SeekToLast();
+    ASSERT_FALSE(iter->Valid());
+    ASSERT_TRUE(iter->status().IsNotSupported());
+    // Seek() still works on the same iterator.
+    iter->Seek("kkkk");
+    ASSERT_OK(iter->status());
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_EQ("kkkk1", iter->key().ToString());
+  }
+  // With bounds, SeekToFirst() and SeekToLast() become Seek(lower bound) and
+  // SeekForPrev(upper bound).
+  const Slice lower_bound("kkkk");
+  const Slice upper_bound("kkkk9");
+  ro.iterate_lower_bound = &lower_bound;
+  ro.iterate_upper_bound = &upper_bound;
+  ASSERT_EQ(kLiveKeys, ScanAllKeys(db_.get(), ro, /*forward=*/true));
+  ASSERT_EQ(kLiveKeysReversed, ScanAllKeys(db_.get(), ro, /*forward=*/false));
+
+  // Total order and auto_prefix_mode iterators are not affected.
+  ReadOptions total_order;
+  ASSERT_EQ(kLiveKeys, ScanAllKeys(db_.get(), total_order, /*forward=*/true));
+  ReadOptions auto_prefix;
+  auto_prefix.auto_prefix_mode = true;
+  ASSERT_EQ(kLiveKeys, ScanAllKeys(db_.get(), auto_prefix, /*forward=*/true));
+}
+
+TEST_F(DBRangeDelCrossPrefixTest, AutoPrefixModeFullScanCrossPrefixTombstone) {
+  ReadOptions auto_prefix;
+  auto_prefix.auto_prefix_mode = true;
+  for (int older_level : {0, 1}) {
+    SCOPED_TRACE("older_level=" + std::to_string(older_level));
+    Populate(PrefixFilterOptions(CurrentOptions()), older_level);
+    ASSERT_EQ(kLiveKeys, ScanAllKeys(db_.get(), auto_prefix, /*forward=*/true));
+    ASSERT_EQ(kLiveKeysReversed,
+              ScanAllKeys(db_.get(), auto_prefix, /*forward=*/false));
+  }
+}
+
+TEST_F(DBRangeDelTest, PrefixSameAsStartUnboundedScanNoRangeConversion) {
+  // An unbounded SeekToFirst() under prefix_same_as_start can skip live keys,
+  // so it must not convert point deletes into a range tombstone.
+  Options options = PrefixFilterOptions(CurrentOptions());
+  options.min_tombstones_for_range_conversion = 2;
+  options.statistics = CreateDBStatistics();
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("kkkk1", "v"));
+  ASSERT_OK(Flush());
+  ASSERT_OK(Delete("aaaa1"));
+  ASSERT_OK(Delete("aaaa2"));
+  ASSERT_OK(Delete("aaaa3"));
+
+  ReadOptions ro;
+  ro.prefix_same_as_start = true;
+  {
+    std::unique_ptr<Iterator> iter(db_->NewIterator(ro));
+    iter->SeekToFirst();
+    ASSERT_OK(iter->status());
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_EQ("kkkk1", iter->key().ToString());
+  }
+  ASSERT_EQ(0, options.statistics->getTickerCount(
+                   READ_PATH_RANGE_TOMBSTONES_INSERTED));
+  // A scan started by Seek() still converts the same deletes.
+  {
+    std::unique_ptr<Iterator> iter(db_->NewIterator(ro));
+    iter->Seek("aaaa");
+    ASSERT_OK(iter->status());
+    ASSERT_FALSE(iter->Valid());
+  }
+  ASSERT_EQ(1, options.statistics->getTickerCount(
+                   READ_PATH_RANGE_TOMBSTONES_INSERTED));
+}
+
 TEST_F(DBRangeDelTest, RefreshMemtableIter) {
   Options options = CurrentOptions();
   options.disable_auto_compactions = true;