MergingIterator.Seek() to lazily initialize MinHeap

Summary:
For the use cases that prefix filtering is enabled, initializing heaps when doing MergingIterator.Seek() might introduce non-negligible costs. This patch makes it lazily done.

Test Plan: make all check

Reviewers: haobo,dhruba,kailiu

CC:

Task ID: #

Blame Rev:

table/merger.cc

@@ -26,6 +26,7 @@ class MergingIterator : public Iterator {
       : comparator_(comparator),
         children_(n),
         current_(nullptr),
+        use_heap_(true),
         direction_(kForward),
         maxHeap_(NewMaxIterHeap(comparator_)),
         minHeap_ (NewMinIterHeap(comparator_)) {
@@ -70,14 +71,38 @@ class MergingIterator : public Iterator {
   }
 
   virtual void Seek(const Slice& target) {
-    ClearHeaps();
+    // Invalidate the heap.
+    use_heap_ = false;
+    IteratorWrapper* first_child = nullptr;
     for (auto& child : children_) {
       child.Seek(target);
       if (child.Valid()) {
-        minHeap_.push(&child);
+        // This child has valid key
+        if (!use_heap_) {
+          if (first_child == nullptr) {
+            // It's the first child has valid key. Only put it int
+            // current_. Now the values in the heap should be invalid.
+            first_child = &child;
+          } else {
+            // We have more than one children with valid keys. Initialize
+            // the heap and put the first child into the heap.
+            ClearHeaps();
+            minHeap_.push(first_child);
+          }
+        }
+        if (use_heap_) {
+          minHeap_.push(&child);
+        }
       }
     }
-    FindSmallest();
+    if (use_heap_) {
+      // If heap is valid, need to put the smallest key to curent_.
+      FindSmallest();
+    } else {
+      // The heap is not valid, then the current_ iterator is the first
+      // one, or null if there is no first child.
+      current_ = first_child;
+    }
     direction_ = kForward;
   }
 
@@ -109,10 +134,14 @@ class MergingIterator : public Iterator {
     // as the current points to the current record. move the iterator forward.
     // and if it is valid add it to the heap.
     current_->Next();
-    if (current_->Valid()){
-      minHeap_.push(current_);
+    if (use_heap_) {
+      if (current_->Valid()) {
+        minHeap_.push(current_);
+      }
+      FindSmallest();
+    } else if (!current_->Valid()) {
+      current_ = nullptr;
     }
-    FindSmallest();
   }
 
   virtual void Prev() {
@@ -178,6 +207,10 @@ class MergingIterator : public Iterator {
   const Comparator* comparator_;
   std::vector<IteratorWrapper> children_;
   IteratorWrapper* current_;
+  // If the value is true, both of iterators in the heap and current_
+  // contain valid rows. If it is false, only current_ can possibly contain
+  // valid rows.
+  bool use_heap_;
   // Which direction is the iterator moving?
   enum Direction {
     kForward,
@@ -189,6 +222,7 @@ class MergingIterator : public Iterator {
 };
 
 void MergingIterator::FindSmallest() {
+  assert(use_heap_);
   if (minHeap_.empty()) {
     current_ = nullptr;
   } else {
@@ -199,6 +233,7 @@ void MergingIterator::FindSmallest() {
 }
 
 void MergingIterator::FindLargest() {
+  assert(use_heap_);
   if (maxHeap_.empty()) {
     current_ = nullptr;
   } else {
@@ -209,6 +244,7 @@ void MergingIterator::FindLargest() {
 }
 
 void MergingIterator::ClearHeaps() {
+  use_heap_ = true;
   maxHeap_ = NewMaxIterHeap(comparator_);
   minHeap_ = NewMinIterHeap(comparator_);
 }