revert Prev() in MergingIterator to use previous code in non-prefix-seek mode

Summary: Siying suggested to keep old code for normal mode prev() for safety

Test Plan: make check -j64

Reviewers: yiwu, andrewkr, sdong

Reviewed By: sdong

Subscribers: andrewkr, dhruba, leveldb

Differential Revision: https://reviews.facebook.net/D65439

db/db_impl.cc

@@ -3850,7 +3850,10 @@ InternalIterator* DBImpl::NewInternalIterator(const ReadOptions& read_options,
   InternalIterator* internal_iter;
   assert(arena != nullptr);
   // Need to create internal iterator from the arena.
-  MergeIteratorBuilder merge_iter_builder(&cfd->internal_comparator(), arena);
+  MergeIteratorBuilder merge_iter_builder(
+      &cfd->internal_comparator(), arena,
+      !read_options.total_order_seek &&
+          cfd->ioptions()->prefix_extractor != nullptr);
   // Collect iterator for mutable mem
   merge_iter_builder.AddIterator(
       super_version->mem->NewIterator(read_options, arena));

table/merger.cc

@@ -36,12 +36,13 @@ const size_t kNumIterReserve = 4;
 class MergingIterator : public InternalIterator {
  public:
   MergingIterator(const Comparator* comparator, InternalIterator** children,
-                  int n, bool is_arena_mode)
+                  int n, bool is_arena_mode, bool prefix_seek_mode)
       : is_arena_mode_(is_arena_mode),
         comparator_(comparator),
         current_(nullptr),
         direction_(kForward),
         minHeap_(comparator_),
+        prefix_seek_mode_(prefix_seek_mode),
         pinned_iters_mgr_(nullptr) {
     children_.resize(n);
     for (int i = 0; i < n; i++) {
@@ -204,16 +205,37 @@ class MergingIterator : public InternalIterator {
       InitMaxHeap();
       for (auto& child : children_) {
         if (&child != current_) {
+          if (!prefix_seek_mode_) {
+            child.Seek(key());
+            if (child.Valid()) {
+              // Child is at first entry >= key().  Step back one to be < key()
+              TEST_SYNC_POINT_CALLBACK("MergeIterator::Prev:BeforePrev",
+                                       &child);
+              child.Prev();
+            } else {
+              // Child has no entries >= key().  Position at last entry.
+              TEST_SYNC_POINT("MergeIterator::Prev:BeforeSeekToLast");
+              child.SeekToLast();
+            }
+          } else {
             child.SeekForPrev(key());
             if (child.Valid() && comparator_->Equal(key(), child.key())) {
               child.Prev();
             }
           }
+        }
         if (child.Valid()) {
           maxHeap_->push(&child);
         }
       }
       direction_ = kReverse;
+      if (!prefix_seek_mode_) {
+        // Note that we don't do assert(current_ == CurrentReverse()) here
+        // because it is possible to have some keys larger than the seek-key
+        // inserted between Seek() and SeekToLast(), which makes current_ not
+        // equal to CurrentReverse().
+        current_ = CurrentReverse();
+      }
       // The loop advanced all non-current children to be < key() so current_
       // should still be strictly the smallest key.
       assert(current_ == CurrentReverse());
@@ -299,6 +321,8 @@ class MergingIterator : public InternalIterator {
   };
   Direction direction_;
   MergerMinIterHeap minHeap_;
+  bool prefix_seek_mode_;
+
   // Max heap is used for reverse iteration, which is way less common than
   // forward.  Lazily initialize it to save memory.
   std::unique_ptr<MergerMaxIterHeap> maxHeap_;
@@ -331,7 +355,7 @@ void MergingIterator::InitMaxHeap() {
 
 InternalIterator* NewMergingIterator(const Comparator* cmp,
                                      InternalIterator** list, int n,
-                                     Arena* arena) {
+                                     Arena* arena, bool prefix_seek_mode) {
   assert(n >= 0);
   if (n == 0) {
     return NewEmptyInternalIterator(arena);
@@ -339,19 +363,20 @@ InternalIterator* NewMergingIterator(const Comparator* cmp,
     return list[0];
   } else {
     if (arena == nullptr) {
-      return new MergingIterator(cmp, list, n, false);
+      return new MergingIterator(cmp, list, n, false, prefix_seek_mode);
     } else {
       auto mem = arena->AllocateAligned(sizeof(MergingIterator));
-      return new (mem) MergingIterator(cmp, list, n, true);
+      return new (mem) MergingIterator(cmp, list, n, true, prefix_seek_mode);
     }
   }
 }
 
 MergeIteratorBuilder::MergeIteratorBuilder(const Comparator* comparator,
-                                           Arena* a)
+                                           Arena* a, bool prefix_seek_mode)
     : first_iter(nullptr), use_merging_iter(false), arena(a) {
   auto mem = arena->AllocateAligned(sizeof(MergingIterator));
-  merge_iter = new (mem) MergingIterator(comparator, nullptr, 0, true);
+  merge_iter =
+      new (mem) MergingIterator(comparator, nullptr, 0, true, prefix_seek_mode);
 }
 
 void MergeIteratorBuilder::AddIterator(InternalIterator* iter) {

table/merger.h

@@ -28,7 +28,8 @@ class Arena;
 // REQUIRES: n >= 0
 extern InternalIterator* NewMergingIterator(const Comparator* comparator,
                                             InternalIterator** children, int n,
-                                            Arena* arena = nullptr);
+                                            Arena* arena = nullptr,
+                                            bool prefix_seek_mode = false);
 
 class MergingIterator;
 
@@ -37,7 +38,8 @@ class MergeIteratorBuilder {
  public:
   // comparator: the comparator used in merging comparator
   // arena: where the merging iterator needs to be allocated from.
-  explicit MergeIteratorBuilder(const Comparator* comparator, Arena* arena);
+  explicit MergeIteratorBuilder(const Comparator* comparator, Arena* arena,
+                                bool prefix_seek_mode = false);
   ~MergeIteratorBuilder() {}
 
   // Add iter to the merging iterator.