From 8e01bd1144abc43ebe68d8ea7393a3f1a31abd06 Mon Sep 17 00:00:00 2001
From: sdong <siying.d@fb.com>
Date: Tue, 4 Aug 2015 16:50:40 -0700
Subject: [PATCH] Fix misplaced position for reversing iterator direction while
 current key is a merge

Summary:
While doing forward iterating, if current key is merge, internal iterator position is placed to the next key. If Prev() is called now, needs to do extra Prev() to recover the location.
This is second attempt of fixing after reverting ec70fea4c4025351190eba7a02bd09bb5f083790. This time shrink the fix to only merge key is the current key and avoid the reseeking logic for max_iterating skipping

Test Plan: enable the two disabled tests and make sure they pass

Reviewers: rven, IslamAbdelRahman, kradhakrishnan, tnovak, yhchiang

Reviewed By: yhchiang

Subscribers: leveldb, dhruba

Differential Revision: https://reviews.facebook.net/D43557
---
 db/db_iter.cc      | 31 +++++++++++++++++++++++++++++--
 db/db_iter_test.cc |  4 ++--
 db/db_test.cc      |  2 +-
 3 files changed, 32 insertions(+), 5 deletions(-)

diff --git a/db/db_iter.cc b/db/db_iter.cc
index 6bee64635..6ef2353d5 100644
--- a/db/db_iter.cc
+++ b/db/db_iter.cc
@@ -115,6 +115,7 @@ class DBIter: public Iterator {
   virtual void SeekToLast() override;
 
  private:
+  void ReverseToBackward();
   void PrevInternal();
   void FindParseableKey(ParsedInternalKey* ikey, Direction direction);
   bool FindValueForCurrentKey();
@@ -350,12 +351,38 @@ void DBIter::MergeValuesNewToOld() {
 void DBIter::Prev() {
   assert(valid_);
   if (direction_ == kForward) {
-    FindPrevUserKey();
-    direction_ = kReverse;
+    ReverseToBackward();
   }
   PrevInternal();
 }
 
+void DBIter::ReverseToBackward() {
+  if (current_entry_is_merged_) {
+    // Not placed in the same key. Need to call Prev() until finding the
+    // previous key.
+    if (!iter_->Valid()) {
+      iter_->SeekToLast();
+    }
+    ParsedInternalKey ikey;
+    FindParseableKey(&ikey, kReverse);
+    while (iter_->Valid() &&
+           user_comparator_->Compare(ikey.user_key, saved_key_.GetKey()) > 0) {
+      iter_->Prev();
+      FindParseableKey(&ikey, kReverse);
+    }
+  }
+#ifndef NDEBUG
+  if (iter_->Valid()) {
+    ParsedInternalKey ikey;
+    assert(ParseKey(&ikey));
+    assert(user_comparator_->Compare(ikey.user_key, saved_key_.GetKey()) <= 0);
+  }
+#endif
+
+  FindPrevUserKey();
+  direction_ = kReverse;
+}
+
 void DBIter::PrevInternal() {
   if (!iter_->Valid()) {
     valid_ = false;
diff --git a/db/db_iter_test.cc b/db/db_iter_test.cc
index 2cb81aab1..dce8c8282 100644
--- a/db/db_iter_test.cc
+++ b/db/db_iter_test.cc
@@ -1670,7 +1670,7 @@ TEST_F(DBIteratorTest, DBIterator8) {
 
 // TODO(3.13): fix the issue of Seek() then Prev() which might not necessary
 //             return the biggest element smaller than the seek key.
-TEST_F(DBIteratorTest, DISABLED_DBIterator9) {
+TEST_F(DBIteratorTest, DBIterator9) {
   Options options;
   options.merge_operator = MergeOperators::CreateFromStringId("stringappend");
   {
@@ -1720,7 +1720,7 @@ TEST_F(DBIteratorTest, DISABLED_DBIterator9) {
 
 // TODO(3.13): fix the issue of Seek() then Prev() which might not necessary
 //             return the biggest element smaller than the seek key.
-TEST_F(DBIteratorTest, DISABLED_DBIterator10) {
+TEST_F(DBIteratorTest, DBIterator10) {
   Options options;
 
   TestIterator* internal_iter = new TestIterator(BytewiseComparator());
diff --git a/db/db_test.cc b/db/db_test.cc
index b691d892a..cb1b8b082 100644
--- a/db/db_test.cc
+++ b/db/db_test.cc
@@ -8226,7 +8226,7 @@ TEST_F(DBTest, RowCache) {
 
 // TODO(3.13): fix the issue of Seek() + Prev() which might not necessary
 //             return the biggest key which is smaller than the seek key.
-TEST_F(DBTest, DISABLED_PrevAfterMerge) {
+TEST_F(DBTest, PrevAfterMerge) {
   Options options;
   options.create_if_missing = true;
   options.merge_operator = MergeOperators::CreatePutOperator();