update HISTORY.md and version.h for 7.7.8

Summary:
Currently, `iterate_upper_bound` is not checked for range tombstone keys in MergingIterator. This may impact performance when there is a large number of range tombstones right after `iterate_upper_bound`. This PR fixes this issue by checking `iterate_upper_bound` in MergingIterator for range tombstone keys.

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

Test Plan:
- added unit test
- stress test: `python3 tools/db_crashtest.py whitebox --simple --verify_iterator_with_expected_state_one_in=5 --delrangepercent=5 --prefixpercent=18 --writepercent=48 --readpercen=15 --duration=36000 --range_deletion_width=100`
- ran different stress tests over sandcastle
- Falcon team ran some test traffic and saw reduced CPU usage on processing range tombstones.

HISTORY.md

@@ -1,4 +1,30 @@
 # Rocksdb Change Log
+## 7.7.8 (11/27/2022)
+### Bug Fixes
+* Fix failed memtable flush retry bug that could cause wrongly ordered updates, which would surface to writers as `Status::Corruption` in case of `force_consistency_checks=true` (default). It affects use cases that enable both parallel flush (`max_background_flushes > 1` or `max_background_jobs >= 8`) and non-default memtable count (`max_write_buffer_number > 2`).
+* Tiered Storage: fixed excessive keys written to penultimate level in non-debug builds.
+* Fixed a regression in iterator where range tombstones after `iterate_upper_bound` is processed.
+
+## 7.7.7 (11/15/2022)
+### Bug Fixes
+* Fixed a regression in scan for async_io. During seek, valid buffers were getting cleared causing a regression.
+
+## 7.7.6 (11/03/2022)
+### Bug Fixes
+* Fix memory corruption error in scans if async_io is enabled. Memory corruption happened if there is IOError while reading the data leading to empty buffer and other buffer already in progress of async read goes again for reading.
+
+## 7.7.5 (10/28/2022)
+### Bug Fixes
+* Fixed an iterator performance regression for delete range users when scanning through a consecutive sequence of range tombstones (#10877).
+
+## 7.7.4 (10/28/2022)
+### Bug Fixes
+* Fixed a case of calling malloc_usable_size on result of operator new[].
+
+## 7.7.3 (10/11/2022)
+### Bug Fixes
+* Fixed a memory safety bug in experimental HyperClockCache (#10768)
+
 ## 7.7.2 (10/05/2022)
 ### Bug Fixes
 * Fixed a bug in iterator refresh that was not freeing up SuperVersion, which could cause excessive resource pinniung (#10770).

cache/clock_cache.cc

@@ -23,6 +23,12 @@ namespace ROCKSDB_NAMESPACE {
 
 namespace hyper_clock_cache {
 
+inline uint64_t GetRefcount(uint64_t meta) {
+  return ((meta >> ClockHandle::kAcquireCounterShift) -
+          (meta >> ClockHandle::kReleaseCounterShift)) &
+         ClockHandle::kCounterMask;
+}
+
 static_assert(sizeof(ClockHandle) == 64U,
               "Expecting size / alignment with common cache line size");
 
@@ -49,6 +55,7 @@ ClockHandleTable::~ClockHandleTable() {
         break;
       case ClockHandle::kStateInvisible:  // rare but possible
       case ClockHandle::kStateVisible:
+        assert(GetRefcount(h.meta) == 0);
         h.FreeData();
 #ifndef NDEBUG
         Rollback(h.hash, &h);
@@ -562,10 +569,7 @@ bool ClockHandleTable::Release(ClockHandle* h, bool useful,
     }
     // Take ownership if no refs
     do {
-      uint64_t refcount = ((old_meta >> ClockHandle::kAcquireCounterShift) -
-                           (old_meta >> ClockHandle::kReleaseCounterShift)) &
-                          ClockHandle::kCounterMask;
-      if (refcount != 0) {
+      if (GetRefcount(old_meta) != 0) {
         // Not last ref at some point in time during this Release call
         // Correct for possible (but rare) overflow
         CorrectNearOverflow(old_meta, h->meta);
@@ -622,6 +626,8 @@ void ClockHandleTable::Ref(ClockHandle& h) {
 
   assert((old_meta >> ClockHandle::kStateShift) &
          ClockHandle::kStateShareableBit);
+  // Must have already had a reference
+  assert(GetRefcount(old_meta) > 0);
   (void)old_meta;
 }
 
@@ -671,10 +677,7 @@ void ClockHandleTable::Erase(const CacheKeyBytes& key, uint32_t hash) {
             old_meta &= ~(uint64_t{ClockHandle::kStateVisibleBit}
                           << ClockHandle::kStateShift);
             for (;;) {
-              uint64_t refcount =
-                  ((old_meta >> ClockHandle::kAcquireCounterShift) -
-                   (old_meta >> ClockHandle::kReleaseCounterShift)) &
-                  ClockHandle::kCounterMask;
+              uint64_t refcount = GetRefcount(old_meta);
               assert(refcount > 0);
               if (refcount > 1) {
                 // Not last ref at some point in time during this Erase call
@@ -683,8 +686,10 @@ void ClockHandleTable::Erase(const CacheKeyBytes& key, uint32_t hash) {
                                   std::memory_order_release);
                 break;
               } else if (h->meta.compare_exchange_weak(
-                             old_meta, uint64_t{ClockHandle::kStateConstruction}
-                                           << ClockHandle::kStateShift)) {
+                             old_meta,
+                             uint64_t{ClockHandle::kStateConstruction}
+                                 << ClockHandle::kStateShift,
+                             std::memory_order_acq_rel)) {
                 // Took ownership
                 assert(hash == h->hash);
                 // TODO? Delay freeing?
@@ -740,20 +745,32 @@ void ClockHandleTable::ConstApplyToEntriesRange(
   for (uint32_t i = index_begin; i < index_end; i++) {
     ClockHandle& h = array_[i];
 
+    // Note: to avoid using compare_exchange, we have to be extra careful.
     uint64_t old_meta = h.meta.load(std::memory_order_relaxed);
     // Check if it's an entry visible to lookups
     if ((old_meta >> ClockHandle::kStateShift) & check_state_mask) {
-      // Increment acquire counter
+      // Increment acquire counter. Note: it's possible that the entry has
+      // completely changed since we loaded old_meta, but incrementing acquire
+      // count is always safe. (Similar to optimistic Lookup here.)
       old_meta = h.meta.fetch_add(ClockHandle::kAcquireIncrement,
                                   std::memory_order_acquire);
-      // Double-check
-      if ((old_meta >> ClockHandle::kStateShift) & check_state_mask) {
-        func(h);
+      // Check whether we actually acquired a reference.
+      if ((old_meta >> ClockHandle::kStateShift) &
+          ClockHandle::kStateShareableBit) {
+        // Apply func if appropriate
+        if ((old_meta >> ClockHandle::kStateShift) & check_state_mask) {
+          func(h);
+        }
+        // Pretend we never took the reference
+        h.meta.fetch_sub(ClockHandle::kAcquireIncrement,
+                         std::memory_order_release);
+        // No net change, so don't need to check for overflow
+      } else {
+        // For other states, incrementing the acquire counter has no effect
+        // so we don't need to undo it. Furthermore, we cannot safely undo
+        // it because we did not acquire a read reference to lock the
+        // entry in a Shareable state.
       }
-      // Pretend we never took the reference
-      h.meta.fetch_sub(ClockHandle::kAcquireIncrement,
-                       std::memory_order_release);
-      // No net change, so don't need to check for overflow
     }
   }
 }
@@ -763,12 +780,9 @@ void ClockHandleTable::EraseUnRefEntries() {
     ClockHandle& h = array_[i];
 
     uint64_t old_meta = h.meta.load(std::memory_order_relaxed);
-    uint64_t refcount = ((old_meta >> ClockHandle::kAcquireCounterShift) -
-                         (old_meta >> ClockHandle::kReleaseCounterShift)) &
-                        ClockHandle::kCounterMask;
     if (old_meta & (uint64_t{ClockHandle::kStateShareableBit}
                     << ClockHandle::kStateShift) &&
-        refcount == 0 &&
+        GetRefcount(old_meta) == 0 &&
         h.meta.compare_exchange_strong(old_meta,
                                        uint64_t{ClockHandle::kStateConstruction}
                                            << ClockHandle::kStateShift,
@@ -877,13 +891,12 @@ void ClockHandleTable::Evict(size_t requested_charge, size_t* freed_charge,
         // Only clock update entries with no outstanding refs
         continue;
       }
-      if (!(meta >> ClockHandle::kStateShift &
+      if (!((meta >> ClockHandle::kStateShift) &
             ClockHandle::kStateShareableBit)) {
         // Only clock update Shareable entries
         continue;
       }
-      // ModTableSize(old_clock_pointer + i));
-      if (meta >> ClockHandle::kStateShift == ClockHandle::kStateVisible &&
+      if ((meta >> ClockHandle::kStateShift == ClockHandle::kStateVisible) &&
           acquire_count > 0) {
         // Decrement clock
         uint64_t new_count = std::min(acquire_count - 1,
@@ -1101,9 +1114,7 @@ size_t ClockCacheShard::GetPinnedUsage() const {
   table_.ConstApplyToEntriesRange(
       [&table_pinned_usage, charge_metadata](const ClockHandle& h) {
         uint64_t meta = h.meta.load(std::memory_order_relaxed);
-        uint64_t refcount = ((meta >> ClockHandle::kAcquireCounterShift) -
-                             (meta >> ClockHandle::kReleaseCounterShift)) &
-                            ClockHandle::kCounterMask;
+        uint64_t refcount = GetRefcount(meta);
         // Holding one ref for ConstApplyToEntriesRange
         assert(refcount > 0);
         if (refcount > 1) {

cache/lru_cache.cc

@@ -382,7 +382,7 @@ Status LRUCacheShard::InsertItem(LRUHandle* e, Cache::Handle** handle,
         last_reference_list.push_back(e);
       } else {
         if (free_handle_on_fail) {
-          delete[] reinterpret_cast<char*>(e);
+          free(e);
           *handle = nullptr;
         }
         s = Status::MemoryLimit("Insert failed due to LRU cache being full.");
@@ -561,8 +561,7 @@ Cache::Handle* LRUCacheShard::Lookup(
         secondary_cache_->Lookup(key, create_cb, wait, found_dummy_entry,
                                  is_in_sec_cache);
     if (secondary_handle != nullptr) {
-      e = reinterpret_cast<LRUHandle*>(
-          new char[sizeof(LRUHandle) - 1 + key.size()]);
+      e = static_cast<LRUHandle*>(malloc(sizeof(LRUHandle) - 1 + key.size()));
 
       e->flags = 0;
       e->SetSecondaryCacheCompatible(true);
@@ -684,8 +683,8 @@ Status LRUCacheShard::Insert(const Slice& key, uint32_t hash, void* value,
   // Allocate the memory here outside of the mutex.
   // If the cache is full, we'll have to release it.
   // It shouldn't happen very often though.
-  LRUHandle* e = reinterpret_cast<LRUHandle*>(
-      new char[sizeof(LRUHandle) - 1 + key.size()]);
+  LRUHandle* e =
+      static_cast<LRUHandle*>(malloc(sizeof(LRUHandle) - 1 + key.size()));
 
   e->value = value;
   e->flags = 0;

cache/lru_cache.h

@@ -241,7 +241,8 @@ struct LRUHandle {
         (*info_.helper->del_cb)(key(), value);
       }
     }
-    delete[] reinterpret_cast<char*>(this);
+
+    free(this);
   }
 
   inline size_t CalcuMetaCharge(

db/compaction/compaction_iterator.cc

@@ -1086,7 +1086,9 @@ void CompactionIterator::DecideOutputLevel() {
   TEST_SYNC_POINT_CALLBACK("CompactionIterator::PrepareOutput.context",
                            &context);
   output_to_penultimate_level_ = context.output_to_penultimate_level;
-#endif /* !NDEBUG */
+#else
+  output_to_penultimate_level_ = false;
+#endif  // NDEBUG
 
   // if the key is newer than the cutoff sequence or within the earliest
   // snapshot, it should output to the penultimate level.

db/db_impl/db_impl.cc

@@ -1802,7 +1802,8 @@ InternalIterator* DBImpl::NewInternalIterator(
   MergeIteratorBuilder merge_iter_builder(
       &cfd->internal_comparator(), arena,
       !read_options.total_order_seek &&
-          super_version->mutable_cf_options.prefix_extractor != nullptr);
+          super_version->mutable_cf_options.prefix_extractor != nullptr,
+      read_options.iterate_upper_bound);
   // Collect iterator for mutable memtable
   auto mem_iter = super_version->mem->NewIterator(read_options, arena);
   Status s;

db/db_range_del_test.cc

@@ -2704,6 +2704,46 @@ TEST_F(DBRangeDelTest, RefreshMemtableIter) {
   ASSERT_OK(iter->Refresh());
 }
 
+TEST_F(DBRangeDelTest, RangeTombstoneRespectIterateUpperBound) {
+  // Memtable: a, [b, bz)
+  // Do a Seek on `a` with iterate_upper_bound being az
+  // range tombstone [b, bz) should not be processed (added to and
+  // popped from the min_heap in MergingIterator).
+  Options options = CurrentOptions();
+  options.disable_auto_compactions = true;
+  DestroyAndReopen(options);
+
+  ASSERT_OK(Put("a", "bar"));
+  ASSERT_OK(
+      db_->DeleteRange(WriteOptions(), db_->DefaultColumnFamily(), "b", "bz"));
+
+  // I could not find a cleaner way to test this without relying on
+  // implementation detail. Tried to test the value of
+  // `internal_range_del_reseek_count` but that did not work
+  // since BlockBasedTable iterator becomes !Valid() when point key
+  // is out of bound and that reseek only happens when a point key
+  // is covered by some range tombstone.
+  SyncPoint::GetInstance()->SetCallBack("MergeIterator::PopDeleteRangeStart",
+                                        [](void*) {
+                                          // there should not be any range
+                                          // tombstone in the heap.
+                                          FAIL();
+                                        });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  ReadOptions read_opts;
+  std::string upper_bound = "az";
+  Slice upper_bound_slice = upper_bound;
+  read_opts.iterate_upper_bound = &upper_bound_slice;
+  std::unique_ptr<Iterator> iter{db_->NewIterator(read_opts)};
+  iter->Seek("a");
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_EQ(iter->key(), "a");
+  iter->Next();
+  ASSERT_FALSE(iter->Valid());
+  ASSERT_OK(iter->status());
+}
+
 #endif  // ROCKSDB_LITE
 
 }  // namespace ROCKSDB_NAMESPACE

db/memtable_list.cc

@@ -419,6 +419,13 @@ void MemTableList::PickMemtablesToFlush(uint64_t max_memtable_id,
             std::max(m->GetNextLogNumber(), *max_next_log_number);
       }
       ret->push_back(m);
+    } else if (!ret->empty()) {
+      // This `break` is necessary to prevent picking non-consecutive memtables
+      // in case `memlist` has one or more entries with
+      // `flush_in_progress_ == true` sandwiched between entries with
+      // `flush_in_progress_ == false`. This could happen after parallel flushes
+      // are picked and the one flushing older memtables is rolled back.
+      break;
     }
   }
   if (!atomic_flush || num_flush_not_started_ == 0) {

db/memtable_list_test.cc

@@ -742,28 +742,40 @@ TEST_F(MemTableListTest, FlushPendingTest) {
   // Pick tables to flush
   list.PickMemtablesToFlush(
       std::numeric_limits<uint64_t>::max() /* memtable_id */, &to_flush);
-  // Should pick 4 of 5 since 1 table has been picked in to_flush2
-  ASSERT_EQ(4, to_flush.size());
+  // Picks three oldest memtables. The fourth oldest is picked in `to_flush2` so
+  // must be excluded. The newest (fifth oldest) is non-consecutive with the
+  // three oldest due to omitting the fourth oldest so must not be picked.
+  ASSERT_EQ(3, to_flush.size());
   ASSERT_EQ(5, list.NumNotFlushed());
   ASSERT_FALSE(list.IsFlushPending());
-  ASSERT_FALSE(list.imm_flush_needed.load(std::memory_order_acquire));
+  ASSERT_TRUE(list.imm_flush_needed.load(std::memory_order_acquire));
 
   // Pick tables to flush again
   autovector<MemTable*> to_flush3;
   list.PickMemtablesToFlush(
       std::numeric_limits<uint64_t>::max() /* memtable_id */, &to_flush3);
-  ASSERT_EQ(0, to_flush3.size());  // nothing not in progress of being flushed
+  // Picks newest (fifth oldest)
+  ASSERT_EQ(1, to_flush3.size());
   ASSERT_EQ(5, list.NumNotFlushed());
   ASSERT_FALSE(list.IsFlushPending());
   ASSERT_FALSE(list.imm_flush_needed.load(std::memory_order_acquire));
 
-  // Flush the 4 memtables that were picked in to_flush
+  // Nothing left to flush
+  autovector<MemTable*> to_flush4;
+  list.PickMemtablesToFlush(
+      std::numeric_limits<uint64_t>::max() /* memtable_id */, &to_flush4);
+  ASSERT_EQ(0, to_flush4.size());
+  ASSERT_EQ(5, list.NumNotFlushed());
+  ASSERT_FALSE(list.IsFlushPending());
+  ASSERT_FALSE(list.imm_flush_needed.load(std::memory_order_acquire));
+
+  // Flush the 3 memtables that were picked in to_flush
   s = Mock_InstallMemtableFlushResults(&list, mutable_cf_options, to_flush,
                                        &to_delete);
   ASSERT_OK(s);
 
-  // Note:  now to_flush contains tables[0,1,2,4].  to_flush2 contains
-  // tables[3].
+  // Note:  now to_flush contains tables[0,1,2].  to_flush2 contains
+  // tables[3]. to_flush3 contains tables[4].
   // Current implementation will only commit memtables in the order they were
   // created. So TryInstallMemtableFlushResults will install the first 3 tables
   // in to_flush and stop when it encounters a table not yet flushed.
@@ -779,7 +791,17 @@ TEST_F(MemTableListTest, FlushPendingTest) {
   ASSERT_FALSE(list.IsFlushPending());
   ASSERT_FALSE(list.imm_flush_needed.load(std::memory_order_acquire));
 
-  // Flush the 1 memtable that was picked in to_flush2
+  // Flush the 1 memtable (tables[4]) that was picked in to_flush3
+  s = MemTableListTest::Mock_InstallMemtableFlushResults(
+      &list, mutable_cf_options, to_flush3, &to_delete);
+  ASSERT_OK(s);
+
+  // This will install 0 tables since tables[4] flushed while tables[3] has not
+  // yet flushed.
+  ASSERT_EQ(2, list.NumNotFlushed());
+  ASSERT_EQ(0, to_delete.size());
+
+  // Flush the 1 memtable (tables[3]) that was picked in to_flush2
   s = MemTableListTest::Mock_InstallMemtableFlushResults(
       &list, mutable_cf_options, to_flush2, &to_delete);
   ASSERT_OK(s);
@@ -809,11 +831,11 @@ TEST_F(MemTableListTest, FlushPendingTest) {
   memtable_id = 4;
   // Pick tables to flush. The tables to pick must have ID smaller than or
   // equal to 4. Therefore, no table will be selected in this case.
-  autovector<MemTable*> to_flush4;
+  autovector<MemTable*> to_flush5;
   list.FlushRequested();
   ASSERT_TRUE(list.HasFlushRequested());
-  list.PickMemtablesToFlush(memtable_id, &to_flush4);
-  ASSERT_TRUE(to_flush4.empty());
+  list.PickMemtablesToFlush(memtable_id, &to_flush5);
+  ASSERT_TRUE(to_flush5.empty());
   ASSERT_EQ(1, list.NumNotFlushed());
   ASSERT_TRUE(list.imm_flush_needed.load(std::memory_order_acquire));
   ASSERT_FALSE(list.IsFlushPending());
@@ -823,8 +845,8 @@ TEST_F(MemTableListTest, FlushPendingTest) {
   // equal to 5. Therefore, only tables[5] will be selected.
   memtable_id = 5;
   list.FlushRequested();
-  list.PickMemtablesToFlush(memtable_id, &to_flush4);
-  ASSERT_EQ(1, static_cast<int>(to_flush4.size()));
+  list.PickMemtablesToFlush(memtable_id, &to_flush5);
+  ASSERT_EQ(1, static_cast<int>(to_flush5.size()));
   ASSERT_EQ(1, list.NumNotFlushed());
   ASSERT_FALSE(list.imm_flush_needed.load(std::memory_order_acquire));
   ASSERT_FALSE(list.IsFlushPending());

db_stress_tool/db_stress_common.h

@@ -173,6 +173,7 @@ DECLARE_bool(mock_direct_io);
 DECLARE_bool(statistics);
 DECLARE_bool(sync);
 DECLARE_bool(use_fsync);
+DECLARE_uint64(stats_dump_period_sec);
 DECLARE_uint64(bytes_per_sync);
 DECLARE_uint64(wal_bytes_per_sync);
 DECLARE_int32(kill_random_test);

db_stress_tool/db_stress_gflags.cc

@@ -1052,4 +1052,8 @@ DEFINE_bool(
     "be preserved similarly under `FLAGS_expected_values_dir/unverified` when "
     "`--expected_values_dir` is nonempty.");
 
+DEFINE_uint64(stats_dump_period_sec,
+              ROCKSDB_NAMESPACE::Options().stats_dump_period_sec,
+              "Gap between printing stats to log in seconds");
+
 #endif  // GFLAGS

db_stress_tool/db_stress_test_base.cc

@@ -3051,6 +3051,8 @@ void InitializeOptionsFromFlags(
   options.experimental_mempurge_threshold =
       FLAGS_experimental_mempurge_threshold;
   options.periodic_compaction_seconds = FLAGS_periodic_compaction_seconds;
+  options.stats_dump_period_sec =
+      static_cast<unsigned int>(FLAGS_stats_dump_period_sec);
   options.ttl = FLAGS_compaction_ttl;
   options.enable_pipelined_write = FLAGS_enable_pipelined_write;
   options.enable_write_thread_adaptive_yield =

file/file_prefetch_buffer.cc

@@ -225,10 +225,8 @@ void FilePrefetchBuffer::AbortIOIfNeeded(uint64_t offset) {
     bufs_[second].async_read_in_progress_ = false;
   }
 
-  if (bufs_[curr_].io_handle_ == nullptr &&
-      bufs_[curr_].async_read_in_progress_) {
+  if (bufs_[curr_].io_handle_ == nullptr) {
     bufs_[curr_].async_read_in_progress_ = false;
-    curr_ = curr_ ^ 1;
   }
 }
 
@@ -268,16 +266,36 @@ void FilePrefetchBuffer::UpdateBuffersIfNeeded(uint64_t offset) {
     bufs_[second].buffer_.Clear();
   }
 
-  // If data starts from second buffer, make it curr_. Second buffer can be
-  // either partial filled or full.
-  if (!bufs_[second].async_read_in_progress_ && DoesBufferContainData(second) &&
-      IsOffsetInBuffer(offset, second)) {
-    // Clear the curr_ as buffers have been swapped and curr_ contains the
-    // outdated data and switch the buffers.
-    if (!bufs_[curr_].async_read_in_progress_) {
-      bufs_[curr_].buffer_.Clear();
+  {
+    // In case buffers do not align, reset second buffer. This can happen in
+    // case readahead_size is set.
+    if (!bufs_[second].async_read_in_progress_ &&
+        !bufs_[curr_].async_read_in_progress_) {
+      if (DoesBufferContainData(curr_)) {
+        if (bufs_[curr_].offset_ + bufs_[curr_].buffer_.CurrentSize() !=
+            bufs_[second].offset_) {
+          bufs_[second].buffer_.Clear();
+        }
+      } else {
+        if (!IsOffsetInBuffer(offset, second)) {
+          bufs_[second].buffer_.Clear();
+        }
+      }
+    }
+  }
+
+  // If data starts from second buffer, make it curr_. Second buffer can be
+  // either partial filled, full or async read is in progress.
+  if (bufs_[second].async_read_in_progress_) {
+    if (IsOffsetInBufferWithAsyncProgress(offset, second)) {
+      curr_ = curr_ ^ 1;
+    }
+  } else {
+    if (DoesBufferContainData(second) && IsOffsetInBuffer(offset, second)) {
+      assert(bufs_[curr_].async_read_in_progress_ ||
+             bufs_[curr_].buffer_.CurrentSize() == 0);
+      curr_ = curr_ ^ 1;
     }
-    curr_ = curr_ ^ 1;
   }
 }
 
@@ -300,53 +318,16 @@ void FilePrefetchBuffer::PollAndUpdateBuffersIfNeeded(uint64_t offset) {
   UpdateBuffersIfNeeded(offset);
 }
 
-// If async_io is enabled in case of sequential reads, PrefetchAsyncInternal is
-// called. When buffers are switched, we clear the curr_ buffer as we assume the
-// data has been consumed because of sequential reads.
-// Data in buffers will always be sequential with curr_ following second and
-// not vice versa.
-//
-// Scenarios for prefetching asynchronously:
-// Case1: If both buffers are empty, prefetch n + readahead_size_/2 bytes
-//        synchronously in curr_ and prefetch readahead_size_/2 async in second
-//        buffer.
-// Case2: If second buffer has partial or full data, make it current and
-//        prefetch readahead_size_/2 async in second buffer. In case of
-//        partial data, prefetch remaining bytes from size n synchronously to
-//        fulfill the requested bytes request.
-// Case3: If curr_ has partial data, prefetch remaining bytes from size n
-//        synchronously in curr_ to fulfill the requested bytes request and
-//        prefetch readahead_size_/2 bytes async in second buffer.
-// Case4: (Special case) If data is in both buffers, copy requested data from
-//        curr_, send async request on curr_, wait for poll to fill second
-//        buffer (if any), and copy remaining data from second buffer to third
-//        buffer.
-Status FilePrefetchBuffer::PrefetchAsyncInternal(
+Status FilePrefetchBuffer::HandleOverlappingData(
     const IOOptions& opts, RandomAccessFileReader* reader, uint64_t offset,
-    size_t length, size_t readahead_size, Env::IOPriority rate_limiter_priority,
-    bool& copy_to_third_buffer) {
-  if (!enable_) {
-    return Status::OK();
-  }
-
-  TEST_SYNC_POINT("FilePrefetchBuffer::PrefetchAsyncInternal:Start");
-
-  size_t alignment = reader->file()->GetRequiredBufferAlignment();
+    size_t length, size_t readahead_size,
+    Env::IOPriority /*rate_limiter_priority*/, bool& copy_to_third_buffer,
+    uint64_t& tmp_offset, size_t& tmp_length) {
   Status s;
-  uint64_t tmp_offset = offset;
-  size_t tmp_length = length;
-
-  // 1. Abort IO and swap buffers if needed to point curr_ to first buffer with
-  // data.
-  {
-    if (!explicit_prefetch_submitted_) {
-      AbortIOIfNeeded(offset);
-    }
-    UpdateBuffersIfNeeded(offset);
-  }
+  size_t alignment = reader->file()->GetRequiredBufferAlignment();
   uint32_t second = curr_ ^ 1;
 
-  // 2. If data is overlapping over two buffers, copy the data from curr_ and
+  // If data is overlapping over two buffers, copy the data from curr_ and
   // call ReadAsync on curr_.
   if (!bufs_[curr_].async_read_in_progress_ && DoesBufferContainData(curr_) &&
       IsOffsetInBuffer(offset, curr_) &&
@@ -391,21 +372,80 @@ Status FilePrefetchBuffer::PrefetchAsyncInternal(
     }
     curr_ = curr_ ^ 1;
   }
+  return s;
+}
+// If async_io is enabled in case of sequential reads, PrefetchAsyncInternal is
+// called. When buffers are switched, we clear the curr_ buffer as we assume the
+// data has been consumed because of sequential reads.
+// Data in buffers will always be sequential with curr_ following second and
+// not vice versa.
+//
+// Scenarios for prefetching asynchronously:
+// Case1: If both buffers are empty, prefetch n + readahead_size_/2 bytes
+//        synchronously in curr_ and prefetch readahead_size_/2 async in second
+//        buffer.
+// Case2: If second buffer has partial or full data, make it current and
+//        prefetch readahead_size_/2 async in second buffer. In case of
+//        partial data, prefetch remaining bytes from size n synchronously to
+//        fulfill the requested bytes request.
+// Case3: If curr_ has partial data, prefetch remaining bytes from size n
+//        synchronously in curr_ to fulfill the requested bytes request and
+//        prefetch readahead_size_/2 bytes async in second buffer.
+// Case4: (Special case) If data is in both buffers, copy requested data from
+//        curr_, send async request on curr_, wait for poll to fill second
+//        buffer (if any), and copy remaining data from second buffer to third
+//        buffer.
+Status FilePrefetchBuffer::PrefetchAsyncInternal(
+    const IOOptions& opts, RandomAccessFileReader* reader, uint64_t offset,
+    size_t length, size_t readahead_size, Env::IOPriority rate_limiter_priority,
+    bool& copy_to_third_buffer) {
+  if (!enable_) {
+    return Status::OK();
+  }
+
+  TEST_SYNC_POINT("FilePrefetchBuffer::PrefetchAsyncInternal:Start");
+
+  size_t alignment = reader->file()->GetRequiredBufferAlignment();
+  Status s;
+  uint64_t tmp_offset = offset;
+  size_t tmp_length = length;
+
+  // 1. Abort IO and swap buffers if needed to point curr_ to first buffer with
+  // data.
+  if (!explicit_prefetch_submitted_) {
+    AbortIOIfNeeded(offset);
+  }
+  UpdateBuffersIfNeeded(offset);
+
+  // 2. Handle overlapping data over two buffers. If data is overlapping then
+  //    during this call:
+  //   - data from curr_ is copied into third buffer,
+  //   - curr_ is send for async prefetching of further data if second buffer
+  //     contains remaining requested data or in progress for async prefetch,
+  //   - switch buffers and curr_ now points to second buffer to copy remaining
+  //     data.
+  s = HandleOverlappingData(opts, reader, offset, length, readahead_size,
+                            rate_limiter_priority, copy_to_third_buffer,
+                            tmp_offset, tmp_length);
+  if (!s.ok()) {
+    return s;
+  }
 
   // 3. Call Poll only if data is needed for the second buffer.
-  //    - Return if whole data is in curr_ and second buffer in progress.
+  //    - Return if whole data is in curr_ and second buffer is in progress or
+  //      already full.
   //    - If second buffer is empty, it will go for ReadAsync for second buffer.
   if (!bufs_[curr_].async_read_in_progress_ && DoesBufferContainData(curr_) &&
       IsDataBlockInBuffer(offset, length, curr_)) {
     // Whole data is in curr_.
     UpdateBuffersIfNeeded(offset);
-    second = curr_ ^ 1;
-    if (bufs_[second].async_read_in_progress_) {
+    if (!IsSecondBuffEligibleForPrefetching()) {
       return s;
     }
   } else {
+    // After poll request, curr_ might be empty because of IOError in
+    // callback while reading or may contain required data.
     PollAndUpdateBuffersIfNeeded(offset);
-    second = curr_ ^ 1;
   }
 
   if (copy_to_third_buffer) {
@@ -427,19 +467,42 @@ Status FilePrefetchBuffer::PrefetchAsyncInternal(
     if (explicit_prefetch_submitted_) {
       return s;
     }
+    if (!IsSecondBuffEligibleForPrefetching()) {
+      return s;
+    }
+  }
+
+  uint32_t second = curr_ ^ 1;
+  assert(!bufs_[curr_].async_read_in_progress_);
+
+  // In case because of some IOError curr_ got empty, abort IO for second as
+  // well. Otherwise data might not align if more data needs to be read in curr_
+  // which might overlap with second buffer.
+  if (!DoesBufferContainData(curr_) && bufs_[second].async_read_in_progress_) {
+    if (bufs_[second].io_handle_ != nullptr) {
+      std::vector<void*> handles;
+      handles.emplace_back(bufs_[second].io_handle_);
+      {
+        StopWatch sw(clock_, stats_, ASYNC_PREFETCH_ABORT_MICROS);
+        Status status = fs_->AbortIO(handles);
+        assert(status.ok());
+      }
+    }
+    DestroyAndClearIOHandle(second);
+    bufs_[second].buffer_.Clear();
   }
 
   // 5. Data is overlapping i.e. some of the data has been copied to third
-  // buffer
-  // and remaining will be updated below.
-  if (copy_to_third_buffer) {
+  // buffer and remaining will be updated below.
+  if (copy_to_third_buffer && DoesBufferContainData(curr_)) {
     CopyDataToBuffer(curr_, offset, length);
 
     // Length == 0: All the requested data has been copied to third buffer and
     // it has already gone for async prefetching. It can return without doing
     // anything further.
-    // Length > 0: More data needs to be consumed so it will continue async and
-    // sync prefetching and copy the remaining data to third buffer in the end.
+    // Length > 0: More data needs to be consumed so it will continue async
+    // and sync prefetching and copy the remaining data to third buffer in the
+    // end.
     if (length == 0) {
       return s;
     }
@@ -458,6 +521,9 @@ Status FilePrefetchBuffer::PrefetchAsyncInternal(
   uint64_t chunk_len1 = 0;
   uint64_t read_len1 = 0;
 
+  assert(!bufs_[second].async_read_in_progress_ &&
+         !DoesBufferContainData(second));
+
   // For length == 0, skip the synchronous prefetching. read_len1 will be 0.
   if (length > 0) {
     CalculateOffsetAndLen(alignment, offset, roundup_len1, curr_,
@@ -594,8 +660,11 @@ bool FilePrefetchBuffer::TryReadFromCacheAsync(
   }
 
   if (explicit_prefetch_submitted_) {
+    // explicit_prefetch_submitted_ is special case where it expects request
+    // submitted in PrefetchAsync should match with this request. Otherwise
+    // buffers will be outdated.
+    // Random offset called. So abort the IOs.
     if (prev_offset_ != offset) {
-      // Random offset called. So abort the IOs.
       AbortAllIOs();
       bufs_[curr_].buffer_.Clear();
       bufs_[curr_ ^ 1].buffer_.Clear();

file/file_prefetch_buffer.h

@@ -8,6 +8,7 @@
 // found in the LICENSE file. See the AUTHORS file for names of contributors.
 
 #pragma once
+
 #include <algorithm>
 #include <atomic>
 #include <sstream>
@@ -236,6 +237,7 @@ class FilePrefetchBuffer {
     }
     prev_offset_ = offset;
     prev_len_ = len;
+    explicit_prefetch_submitted_ = false;
   }
 
   void GetReadaheadState(ReadaheadFileInfo::ReadaheadInfo* readahead_info) {
@@ -363,6 +365,27 @@ class FilePrefetchBuffer {
             bufs_[index].io_handle_ != nullptr &&
             offset >= bufs_[index].offset_ + bufs_[index].async_req_len_);
   }
+  bool IsOffsetInBufferWithAsyncProgress(uint64_t offset, uint32_t index) {
+    return (bufs_[index].async_read_in_progress_ &&
+            offset >= bufs_[index].offset_ &&
+            offset < bufs_[index].offset_ + bufs_[index].async_req_len_);
+  }
+
+  bool IsSecondBuffEligibleForPrefetching() {
+    uint32_t second = curr_ ^ 1;
+    if (bufs_[second].async_read_in_progress_) {
+      return false;
+    }
+    assert(!bufs_[curr_].async_read_in_progress_);
+
+    if (DoesBufferContainData(curr_) && DoesBufferContainData(second) &&
+        (bufs_[curr_].offset_ + bufs_[curr_].buffer_.CurrentSize() ==
+         bufs_[second].offset_)) {
+      return false;
+    }
+    bufs_[second].buffer_.Clear();
+    return true;
+  }
 
   void DestroyAndClearIOHandle(uint32_t index) {
     if (bufs_[index].io_handle_ != nullptr && bufs_[index].del_fn_ != nullptr) {
@@ -373,6 +396,13 @@ class FilePrefetchBuffer {
     bufs_[index].async_read_in_progress_ = false;
   }
 
+  Status HandleOverlappingData(const IOOptions& opts,
+                               RandomAccessFileReader* reader, uint64_t offset,
+                               size_t length, size_t readahead_size,
+                               Env::IOPriority rate_limiter_priority,
+                               bool& copy_to_third_buffer, uint64_t& tmp_offset,
+                               size_t& tmp_length);
+
   std::vector<BufferInfo> bufs_;
   // curr_ represents the index for bufs_ indicating which buffer is being
   // consumed currently.

file/random_access_file_reader.cc

@@ -471,6 +471,7 @@ IOStatus RandomAccessFileReader::ReadAsync(
                     (uintptr_t(req.scratch) & (alignment - 1)) == 0;
   read_async_info->is_aligned_ = is_aligned;
 
+  uint64_t elapsed = 0;
   if (use_direct_io() && is_aligned == false) {
     FSReadRequest aligned_req = Align(req, alignment);
     aligned_req.status.PermitUncheckedError();
@@ -491,12 +492,17 @@ IOStatus RandomAccessFileReader::ReadAsync(
 
     assert(read_async_info->buf_.CurrentSize() == 0);
 
+    StopWatch sw(clock_, nullptr /*stats*/, 0 /*hist_type*/, &elapsed,
+                 true /*overwrite*/, true /*delay_enabled*/);
     s = file_->ReadAsync(aligned_req, opts, read_async_callback,
                          read_async_info, io_handle, del_fn, nullptr /*dbg*/);
   } else {
+    StopWatch sw(clock_, nullptr /*stats*/, 0 /*hist_type*/, &elapsed,
+                 true /*overwrite*/, true /*delay_enabled*/);
     s = file_->ReadAsync(req, opts, read_async_callback, read_async_info,
                          io_handle, del_fn, nullptr /*dbg*/);
   }
+  RecordTick(stats_, READ_ASYNC_MICROS, elapsed);
 
 // Suppress false positive clang analyzer warnings.
 // Memory is not released if file_->ReadAsync returns !s.ok(), because
@@ -575,6 +581,8 @@ void RandomAccessFileReader::ReadAsyncCallback(const FSReadRequest& req,
   }
   if (req.status.ok()) {
     RecordInHistogram(stats_, ASYNC_READ_BYTES, req.result.size());
+  } else if (!req.status.IsAborted()) {
+    RecordTick(stats_, ASYNC_READ_ERROR_COUNT, 1);
   }
 #ifndef ROCKSDB_LITE
   if (ShouldNotifyListeners()) {

include/rocksdb/options.h

@@ -1480,7 +1480,7 @@ struct ReadOptions {
   const Slice* iterate_lower_bound;
 
   // "iterate_upper_bound" defines the extent up to which the forward iterator
-  // can returns entries. Once the bound is reached, Valid() will be false.
+  // can return entries. Once the bound is reached, Valid() will be false.
   // "iterate_upper_bound" is exclusive ie the bound value is
   // not a valid entry. If prefix_extractor is not null:
   // 1. If options.auto_prefix_mode = true, iterate_upper_bound will be used

include/rocksdb/statistics.h

@@ -448,6 +448,11 @@ enum Tickers : uint32_t {
   // # of bytes written into blob cache.
   BLOB_DB_CACHE_BYTES_WRITE,
 
+  // Time spent in the ReadAsync file system call
+  READ_ASYNC_MICROS,
+  // Number of errors returned to the async read callback
+  ASYNC_READ_ERROR_COUNT,
+
   TICKER_ENUM_MAX
 };
 

include/rocksdb/version.h

@@ -13,7 +13,7 @@
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 7
 #define ROCKSDB_MINOR 7
-#define ROCKSDB_PATCH 2
+#define ROCKSDB_PATCH 8
 
 // Do not use these. We made the mistake of declaring macros starting with
 // double underscore. Now we have to live with our choice. We'll deprecate these

java/rocksjni/portal.h

@@ -5104,6 +5104,10 @@ class TickerTypeJni {
         return -0x33;
       case ROCKSDB_NAMESPACE::Tickers::BLOB_DB_CACHE_BYTES_WRITE:
         return -0x34;
+      case ROCKSDB_NAMESPACE::Tickers::READ_ASYNC_MICROS:
+        return -0x35;
+      case ROCKSDB_NAMESPACE::Tickers::ASYNC_READ_ERROR_COUNT:
+        return -0x36;
       case ROCKSDB_NAMESPACE::Tickers::TICKER_ENUM_MAX:
         // 0x5F was the max value in the initial copy of tickers to Java.
         // Since these values are exposed directly to Java clients, we keep
@@ -5487,6 +5491,10 @@ class TickerTypeJni {
         return ROCKSDB_NAMESPACE::Tickers::BLOB_DB_CACHE_BYTES_READ;
       case -0x34:
         return ROCKSDB_NAMESPACE::Tickers::BLOB_DB_CACHE_BYTES_WRITE;
+      case -0x35:
+        return ROCKSDB_NAMESPACE::Tickers::READ_ASYNC_MICROS;
+      case -0x36:
+        return ROCKSDB_NAMESPACE::Tickers::ASYNC_READ_ERROR_COUNT;
       case 0x5F:
         // 0x5F was the max value in the initial copy of tickers to Java.
         // Since these values are exposed directly to Java clients, we keep

monitoring/statistics.cc

@@ -233,7 +233,9 @@ const std::vector<std::pair<Tickers, std::string>> TickersNameMap = {
     {BLOB_DB_CACHE_ADD, "rocksdb.blobdb.cache.add"},
     {BLOB_DB_CACHE_ADD_FAILURES, "rocksdb.blobdb.cache.add.failures"},
     {BLOB_DB_CACHE_BYTES_READ, "rocksdb.blobdb.cache.bytes.read"},
-    {BLOB_DB_CACHE_BYTES_WRITE, "rocksdb.blobdb.cache.bytes.write"}};
+    {BLOB_DB_CACHE_BYTES_WRITE, "rocksdb.blobdb.cache.bytes.write"},
+    {READ_ASYNC_MICROS, "rocksdb.read.async.micros"},
+    {ASYNC_READ_ERROR_COUNT, "rocksdb.async.read.error.count"}};
 
 const std::vector<std::pair<Histograms, std::string>> HistogramsNameMap = {
     {DB_GET, "rocksdb.db.get.micros"},

table/merging_iterator.cc

@@ -117,14 +117,16 @@ class MergingIterator : public InternalIterator {
  public:
   MergingIterator(const InternalKeyComparator* comparator,
                   InternalIterator** children, int n, bool is_arena_mode,
-                  bool prefix_seek_mode)
+                  bool prefix_seek_mode,
+                  const Slice* iterate_upper_bound = nullptr)
       : is_arena_mode_(is_arena_mode),
         prefix_seek_mode_(prefix_seek_mode),
         direction_(kForward),
         comparator_(comparator),
         current_(nullptr),
         minHeap_(comparator_),
-        pinned_iters_mgr_(nullptr) {
+        pinned_iters_mgr_(nullptr),
+        iterate_upper_bound_(iterate_upper_bound) {
     children_.resize(n);
     for (int i = 0; i < n; i++) {
       children_[i].level = i;
@@ -197,27 +199,48 @@ class MergingIterator : public InternalIterator {
   // Add range_tombstone_iters_[level] into min heap.
   // Updates active_ if the end key of a range tombstone is inserted.
   // @param start_key specifies which end point of the range tombstone to add.
-  void InsertRangeTombstoneToMinHeap(size_t level, bool start_key = true) {
+  void InsertRangeTombstoneToMinHeap(size_t level, bool start_key = true,
+                                     bool replace_top = false) {
     assert(!range_tombstone_iters_.empty() &&
            range_tombstone_iters_[level]->Valid());
     if (start_key) {
-      pinned_heap_item_[level].SetTombstoneKey(
-          range_tombstone_iters_[level]->start_key());
+      ParsedInternalKey pik = range_tombstone_iters_[level]->start_key();
+      // iterate_upper_bound does not have timestamp
+      if (iterate_upper_bound_ &&
+          comparator_->user_comparator()->CompareWithoutTimestamp(
+              pik.user_key, true /* a_has_ts */, *iterate_upper_bound_,
+              false /* b_has_ts */) >= 0) {
+        if (replace_top) {
+          // replace_top implies this range tombstone iterator is still in
+          // minHeap_ and at the top.
+          minHeap_.pop();
+        }
+        return;
+      }
+      pinned_heap_item_[level].SetTombstoneKey(std::move(pik));
       pinned_heap_item_[level].type = HeapItem::DELETE_RANGE_START;
       assert(active_.count(level) == 0);
     } else {
+      // allow end key to go over upper bound (if present) since start key is
+      // before upper bound and the range tombstone could still cover a
+      // range before upper bound.
       pinned_heap_item_[level].SetTombstoneKey(
           range_tombstone_iters_[level]->end_key());
       pinned_heap_item_[level].type = HeapItem::DELETE_RANGE_END;
       active_.insert(level);
     }
-    minHeap_.push(&pinned_heap_item_[level]);
+    if (replace_top) {
+      minHeap_.replace_top(&pinned_heap_item_[level]);
+    } else {
+      minHeap_.push(&pinned_heap_item_[level]);
+    }
   }
 
   // Add range_tombstone_iters_[level] into max heap.
   // Updates active_ if the start key of a range tombstone is inserted.
   // @param end_key specifies which end point of the range tombstone to add.
-  void InsertRangeTombstoneToMaxHeap(size_t level, bool end_key = true) {
+  void InsertRangeTombstoneToMaxHeap(size_t level, bool end_key = true,
+                                     bool replace_top = false) {
     assert(!range_tombstone_iters_.empty() &&
            range_tombstone_iters_[level]->Valid());
     if (end_key) {
@@ -231,7 +254,11 @@ class MergingIterator : public InternalIterator {
       pinned_heap_item_[level].type = HeapItem::DELETE_RANGE_START;
       active_.insert(level);
     }
-    maxHeap_->push(&pinned_heap_item_[level]);
+    if (replace_top) {
+      maxHeap_->replace_top(&pinned_heap_item_[level]);
+    } else {
+      maxHeap_->push(&pinned_heap_item_[level]);
+    }
   }
 
   // Remove HeapItems from top of minHeap_ that are of type DELETE_RANGE_START
@@ -241,10 +268,10 @@ class MergingIterator : public InternalIterator {
   void PopDeleteRangeStart() {
     while (!minHeap_.empty() &&
            minHeap_.top()->type == HeapItem::DELETE_RANGE_START) {
-      auto level = minHeap_.top()->level;
-      minHeap_.pop();
+      TEST_SYNC_POINT_CALLBACK("MergeIterator::PopDeleteRangeStart", nullptr);
       // insert end key of this range tombstone and updates active_
-      InsertRangeTombstoneToMinHeap(level, false /* start_key */);
+      InsertRangeTombstoneToMinHeap(
+          minHeap_.top()->level, false /* start_key */, true /* replace_top */);
     }
   }
 
@@ -255,10 +282,9 @@ class MergingIterator : public InternalIterator {
   void PopDeleteRangeEnd() {
     while (!maxHeap_->empty() &&
            maxHeap_->top()->type == HeapItem::DELETE_RANGE_END) {
-      auto level = maxHeap_->top()->level;
-      maxHeap_->pop();
       // insert start key of this range tombstone and updates active_
-      InsertRangeTombstoneToMaxHeap(level, false /* end_key */);
+      InsertRangeTombstoneToMaxHeap(maxHeap_->top()->level, false /* end_key */,
+                                    true /* replace_top */);
     }
   }
 
@@ -565,6 +591,10 @@ class MergingIterator : public InternalIterator {
   std::unique_ptr<MergerMaxIterHeap> maxHeap_;
   PinnedIteratorsManager* pinned_iters_mgr_;
 
+  // Used to bound range tombstones. For point keys, DBIter and SSTable iterator
+  // take care of boundary checking.
+  const Slice* iterate_upper_bound_;
+
   // In forward direction, process a child that is not in the min heap.
   // If valid, add to the min heap. Otherwise, check status.
   void AddToMinHeapOrCheckStatus(HeapItem*);
@@ -626,9 +656,19 @@ void MergingIterator::SeekImpl(const Slice& target, size_t starting_level,
     for (size_t level = 0; level < starting_level; ++level) {
       if (range_tombstone_iters_[level] &&
           range_tombstone_iters_[level]->Valid()) {
-        assert(static_cast<bool>(active_.count(level)) ==
-               (pinned_heap_item_[level].type == HeapItem::DELETE_RANGE_END));
-        minHeap_.push(&pinned_heap_item_[level]);
+        // use an iterator on active_ if performance becomes an issue here
+        if (active_.count(level) > 0) {
+          assert(pinned_heap_item_[level].type == HeapItem::DELETE_RANGE_END);
+          // if it was active, then start key must be within upper_bound,
+          // so we can add to minHeap_ directly.
+          minHeap_.push(&pinned_heap_item_[level]);
+        } else {
+          // this takes care of checking iterate_upper_bound, but with an extra
+          // key comparison if range_tombstone_iters_[level] was already out of
+          // bound. Consider using a new HeapItem type or some flag to remember
+          // boundary checking result.
+          InsertRangeTombstoneToMinHeap(level);
+        }
       } else {
         assert(!active_.count(level));
       }
@@ -761,13 +801,15 @@ bool MergingIterator::SkipNextDeleted() {
   // - range deletion end key
   auto current = minHeap_.top();
   if (current->type == HeapItem::DELETE_RANGE_END) {
-    minHeap_.pop();
     active_.erase(current->level);
     assert(range_tombstone_iters_[current->level] &&
            range_tombstone_iters_[current->level]->Valid());
     range_tombstone_iters_[current->level]->Next();
     if (range_tombstone_iters_[current->level]->Valid()) {
-      InsertRangeTombstoneToMinHeap(current->level);
+      InsertRangeTombstoneToMinHeap(current->level, true /* start_key */,
+                                    true /* replace_top */);
+    } else {
+      minHeap_.pop();
     }
     return true /* current key deleted */;
   }
@@ -977,13 +1019,15 @@ bool MergingIterator::SkipPrevDeleted() {
   // - range deletion start key
   auto current = maxHeap_->top();
   if (current->type == HeapItem::DELETE_RANGE_START) {
-    maxHeap_->pop();
     active_.erase(current->level);
     assert(range_tombstone_iters_[current->level] &&
            range_tombstone_iters_[current->level]->Valid());
     range_tombstone_iters_[current->level]->Prev();
     if (range_tombstone_iters_[current->level]->Valid()) {
-      InsertRangeTombstoneToMaxHeap(current->level);
+      InsertRangeTombstoneToMaxHeap(current->level, true /* end_key */,
+                                    true /* replace_top */);
+    } else {
+      maxHeap_->pop();
     }
     return true /* current key deleted */;
   }
@@ -1266,11 +1310,12 @@ InternalIterator* NewMergingIterator(const InternalKeyComparator* cmp,
 }
 
 MergeIteratorBuilder::MergeIteratorBuilder(
-    const InternalKeyComparator* comparator, Arena* a, bool prefix_seek_mode)
+    const InternalKeyComparator* comparator, Arena* a, bool prefix_seek_mode,
+    const Slice* iterate_upper_bound)
     : first_iter(nullptr), use_merging_iter(false), arena(a) {
   auto mem = arena->AllocateAligned(sizeof(MergingIterator));
-  merge_iter =
-      new (mem) MergingIterator(comparator, nullptr, 0, true, prefix_seek_mode);
+  merge_iter = new (mem) MergingIterator(comparator, nullptr, 0, true,
+                                         prefix_seek_mode, iterate_upper_bound);
 }
 
 MergeIteratorBuilder::~MergeIteratorBuilder() {

table/merging_iterator.h

@@ -45,7 +45,8 @@ class MergeIteratorBuilder {
   // comparator: the comparator used in merging comparator
   // arena: where the merging iterator needs to be allocated from.
   explicit MergeIteratorBuilder(const InternalKeyComparator* comparator,
-                                Arena* arena, bool prefix_seek_mode = false);
+                                Arena* arena, bool prefix_seek_mode = false,
+                                const Slice* iterate_upper_bound = nullptr);
   ~MergeIteratorBuilder();
 
   // Add iter to the merging iterator.

tools/db_crashtest.py

@@ -131,6 +131,8 @@ default_params = {
     "use_multiget" : lambda: random.randint(0, 1),
     "periodic_compaction_seconds" :
         lambda: random.choice([0, 0, 1, 2, 10, 100, 1000]),
+    # 0 = never (used by some), 10 = often (for threading bugs), 600 = default
+    "stats_dump_period_sec" : lambda: random.choice([0, 10, 600]),
     "compaction_ttl" : lambda: random.choice([0, 0, 1, 2, 10, 100, 1000]),
     # Test small max_manifest_file_size in a smaller chance, as most of the
     # time we wnat manifest history to be preserved to help debug