Reverse the order of updates to the same key in WriteBatchWithIndex (#13387)

Summary:
as a preparation to support merge in [WBWIMemtable](https://github.com/facebook/rocksdb/blob/d48af213860054a7696e7ea2764f266c88a3263e/memtable/wbwi_memtable.h#L31), this PR updates how we [order updates to the same key](https://github.com/facebook/rocksdb/blob/d48af213860054a7696e7ea2764f266c88a3263e/utilities/write_batch_with_index/write_batch_with_index_internal.cc#L694-L697) in WriteBatchWithIndex. Specifically, the order is now reversed such that more recent update is ordered first. This will make iterating from WriteBatchWithIndex much easier since the key ordering in WBWI now matches internal key order where keys with larger sequence number are ordered first. The ordering is now explicitly documented above the declaration for `WriteBatchWithIndex` class.

Places that use `WBWIIteratorImpl` and assume key ordering are updated. The rest is test and comments update.
This will affect users who use WBWIIterator directly, the output of GetFromBatch, GetFromBatchAndDB or NewIteratorWithBase are not affected. Users are only affected if they may issue multiple updates to the same key. If WriteBatchWithIndex is created with `overwrite_key=true`, one the the updates needs to be Merge.

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

Test Plan: we have some good coverage of WBWI, I updated some existing tests and added a test for `WBWIIteratorImpl`.

Reviewed By: pdillinger

Differential Revision: D69421268

Pulled By: cbi42

fbshipit-source-id: d97eec4ee74aeac3937c9758041c7713f07f9676

db/merge_context.h

@@ -140,10 +140,11 @@ class MergeContext {
     }
   }
 
-  // List of operands
+  // List of operands, the order of operands depends on operands_reversed_.
   mutable std::unique_ptr<std::vector<Slice>> operand_list_;
   // Copy of operands that are not pinned.
   std::unique_ptr<std::vector<std::unique_ptr<std::string>>> copied_operands_;
+  // Reversed means the newest update is ordered first.
   mutable bool operands_reversed_ = true;
 };
 

db/write_batch_test.cc

@@ -856,15 +856,16 @@ TEST_F(WriteBatchTest, ColumnFamiliesBatchWithIndexTest) {
   iter->Seek("eightfoo");
   ASSERT_OK(iter->status());
   ASSERT_TRUE(iter->Valid());
-  ASSERT_EQ(WriteType::kPutRecord, iter->Entry().type);
+  // For the same key, most recent update is ordered first.
+  ASSERT_EQ(WriteType::kDeleteRecord, iter->Entry().type);
   ASSERT_EQ("eightfoo", iter->Entry().key.ToString());
-  ASSERT_EQ("bar8", iter->Entry().value.ToString());
 
   iter->Next();
   ASSERT_OK(iter->status());
   ASSERT_TRUE(iter->Valid());
-  ASSERT_EQ(WriteType::kDeleteRecord, iter->Entry().type);
+  ASSERT_EQ(WriteType::kPutRecord, iter->Entry().type);
   ASSERT_EQ("eightfoo", iter->Entry().key.ToString());
+  ASSERT_EQ("bar8", iter->Entry().value.ToString());
 
   iter->Next();
   ASSERT_OK(iter->status());
@@ -874,15 +875,15 @@ TEST_F(WriteBatchTest, ColumnFamiliesBatchWithIndexTest) {
   iter->Seek("twofoo");
   ASSERT_OK(iter->status());
   ASSERT_TRUE(iter->Valid());
-  ASSERT_EQ(WriteType::kPutRecord, iter->Entry().type);
+  ASSERT_EQ(WriteType::kSingleDeleteRecord, iter->Entry().type);
   ASSERT_EQ("twofoo", iter->Entry().key.ToString());
-  ASSERT_EQ("bar2", iter->Entry().value.ToString());
 
   iter->Next();
   ASSERT_OK(iter->status());
   ASSERT_TRUE(iter->Valid());
-  ASSERT_EQ(WriteType::kSingleDeleteRecord, iter->Entry().type);
+  ASSERT_EQ(WriteType::kPutRecord, iter->Entry().type);
   ASSERT_EQ("twofoo", iter->Entry().key.ToString());
+  ASSERT_EQ("bar2", iter->Entry().value.ToString());
 
   iter->Next();
   ASSERT_OK(iter->status());

include/rocksdb/utilities/write_batch_with_index.h

@@ -62,16 +62,23 @@ class WBWIIterator {
 
   virtual void SeekToLast() = 0;
 
-  virtual void Seek(const Slice& key) = 0;
+  // Move to the first entry with key >= target.
+  // If there are multiple updates to the same key, the most recent update is
+  // ordered first. If `overwrite_key` is true for this WBWI, this should only
+  // affect iterator output if the write batch contains Merge.
+  virtual void Seek(const Slice& target) = 0;
 
-  virtual void SeekForPrev(const Slice& key) = 0;
+  // Move to the last entry with key <= target.
+  // If there are multiple updates to the same key, this will move iterator
+  // to the last entry, which is the oldest update.
+  virtual void SeekForPrev(const Slice& target) = 0;
 
   virtual void Next() = 0;
 
   virtual void Prev() = 0;
 
-  // the return WriteEntry is only valid until the next mutation of
-  // WriteBatchWithIndex
+  // The returned WriteEntry is only valid until the next mutation of
+  // WriteBatchWithIndex.
   virtual WriteEntry Entry() const = 0;
 
   // For this user key, there is a single delete in this write batch,
@@ -87,6 +94,8 @@ class WBWIIterator {
 // time, indexes will be built. By calling GetWriteBatch(), a user will get the
 // WriteBatch for the data they inserted, which can be used for DB::Write(). A
 // user can call NewIterator() to create an iterator.
+// If there are multiple updates to the same key, the most recent update is
+// ordered first (i.e. the iterator will return the most recent update first).
 class WriteBatchWithIndex : public WriteBatchBase {
  public:
   // backup_index_comparator: the backup comparator used to compare keys
@@ -98,6 +107,8 @@ class WriteBatchWithIndex : public WriteBatchBase {
   // overwrite_key: if true, overwrite the key in the index when inserting
   //                the same key as previously, so iterator will never
   //                show two entries with the same key.
+  //                Note that for Merge, it's added as a new update instead
+  //                of overwriting the existing one.
   explicit WriteBatchWithIndex(
       const Comparator* backup_index_comparator = BytewiseComparator(),
       size_t reserved_bytes = 0, bool overwrite_key = false,
@@ -195,8 +206,8 @@ class WriteBatchWithIndex : public WriteBatchBase {
   // if overwrite_key=false, then each update will be returned as a separate
   // entry, in the order of update time.
   // if overwrite_key=true, then one entry per key will be returned. Merge
-  // updates on the same key will be returned as separate entries.
-  //
+  // updates on the same key will be returned as separate entries, with most
+  // recent update ordered first.
   // The returned iterator should be deleted by the caller.
   WBWIIterator* NewIterator(ColumnFamilyHandle* column_family);
   // Create an iterator of the default column family.
@@ -229,7 +240,7 @@ class WriteBatchWithIndex : public WriteBatchBase {
 
   // Similar to previous function but does not require a column_family.
   // Note:  An InvalidArgument status will be returned if there are any Merge
-  // operators for this key.  Use previous method instead.
+  // operators for this key. Use previous method instead.
   Status GetFromBatch(const DBOptions& options, const Slice& key,
                       std::string* value) {
     return GetFromBatch(nullptr, options, key, value);

unreleased_history/behavior_changes/wbwi-ordering.md

@@ -0,0 +1 @@
+* Reversed the order of updates to the same key in WriteBatchWithIndex. This means if there are multiple updates to the same key, the most recent update is ordered first. This affects the output of WBWIIterator. When WriteBatchWithIndex is created with `overwrite_key=true`, this affects the output only if Merge is used (#13387).

utilities/write_batch_with_index/write_batch_with_index.cc

@@ -105,35 +105,29 @@ bool WriteBatchWithIndex::Rep::UpdateExistingEntryWithCfId(
     return false;
   } else if (!iter.MatchesKey(column_family_id, key)) {
     return false;
-  } else {
-    // Move to the end of this key (NextKey-Prev)
-    iter.NextKey();  // Move to the next key
-    if (iter.Valid()) {
-      iter.Prev();  // Move back one entry
-    } else {
-      iter.SeekToLast();
-    }
   }
-  WriteBatchIndexEntry* non_const_entry =
+  // Seek() and MatchesKey() guarantees that we are at the first entry with
+  // key == `key`, which is the most recently update to `key`.
+  WriteBatchIndexEntry* most_recent_entry =
       const_cast<WriteBatchIndexEntry*>(iter.GetRawEntry());
-  if (LIKELY(last_sub_batch_offset <= non_const_entry->offset)) {
+  if (LIKELY(last_sub_batch_offset <= most_recent_entry->offset)) {
     last_sub_batch_offset = last_entry_offset;
     sub_batch_cnt++;
   }
   if (track_cf_stat) {
-    if (non_const_entry->has_single_del &&
-        !non_const_entry->has_overwritten_single_del) {
+    if (most_recent_entry->has_single_del &&
+        !most_recent_entry->has_overwritten_single_del) {
       cf_id_to_stat[column_family_id].overwritten_sd_count++;
-      non_const_entry->has_overwritten_single_del = true;
+      most_recent_entry->has_overwritten_single_del = true;
     }
     if (type == kSingleDeleteRecord) {
-      non_const_entry->has_single_del = true;
+      most_recent_entry->has_single_del = true;
     }
   }
   if (type == kMergeRecord) {
     return false;
   } else {
-    non_const_entry->offset = last_entry_offset;
+    most_recent_entry->offset = last_entry_offset;
     return true;
   }
 }

utilities/write_batch_with_index/write_batch_with_index_internal.cc

@@ -42,7 +42,8 @@ BaseDeltaIterator::BaseDeltaIterator(ColumnFamilyHandle* column_family,
 BaseDeltaIterator::~BaseDeltaIterator() = default;
 
 bool BaseDeltaIterator::Valid() const {
-  return status_.ok() ? (current_at_base_ ? BaseValid() : DeltaValid()) : false;
+  return status().ok() ? (current_at_base_ ? BaseValid() : DeltaValid())
+                       : false;
 }
 
 void BaseDeltaIterator::SeekToFirst() {
@@ -89,14 +90,22 @@ void BaseDeltaIterator::Next() {
     equal_keys_ = false;
     if (!BaseValid()) {
       assert(DeltaValid());
+      // During reverse scan, base iter was exhausted. This means the
+      // base iter has no smaller key than the current key at delta iter.
+      // So seeking to the first key of base iter will point to base iter's
+      // first key that is at least the current key.
       base_iterator_->SeekToFirst();
     } else if (!DeltaValid()) {
+      // Similar to the base iter case above, position delta iter to the first
+      // entry that is at least the current key.
       delta_iterator_->SeekToFirst();
     } else if (current_at_base_) {
-      // Change delta from larger than base to smaller
+      // current_at_base_ means delta iter is at its first key that is less
+      // than current key. So advance delta one step moves it to the first
+      // key that is at least the current key.
       AdvanceDelta();
     } else {
-      // Change base from larger than delta to smaller
+      // Similar to the case above.
       AdvanceBase();
     }
     if (DeltaValid() && BaseValid()) {
@@ -289,6 +298,9 @@ void BaseDeltaIterator::SetValueAndColumnsFromDelta() {
   WriteEntry delta_entry = delta_iterator_->Entry();
 
   if (merge_context_.GetNumOperands() == 0) {
+    // delete case is handled at callsites, see UpdateCurrent().
+    assert(delta_entry.type == kPutRecord ||
+           delta_entry.type == kPutEntityRecord);
     if (delta_entry.type == kPutRecord) {
       value_ = delta_entry.value;
       columns_.emplace_back(kDefaultWideColumnName, value_);
@@ -411,13 +423,14 @@ void BaseDeltaIterator::UpdateCurrent() {
         return;
       }
 
-      // Base has finished.
       if (!DeltaValid()) {
-        // Finished
+        // Finished, neither base nor delta is valid.
         return;
       }
+      // Base is done, delta has value.
       if (delta_result == WBWIIteratorImpl::kDeleted &&
           merge_context_.GetNumOperands() == 0) {
+        // This key is deleted and no Merge is done after Delete.
         AdvanceDelta();
       } else {
         current_at_base_ = false;
@@ -425,13 +438,14 @@ void BaseDeltaIterator::UpdateCurrent() {
         return;
       }
     } else if (!DeltaValid()) {
-      // Delta has finished.
+      // Delta is done, base has value.
       current_at_base_ = true;
       if (!allow_unprepared_value_) {
         SetValueAndColumnsFromBase();
       }
       return;
     } else {
+      // Both are valid.
       int compare =
           (forward_ ? 1 : -1) * comparator_->CompareWithoutTimestamp(
                                     delta_entry.key, /*a_has_ts=*/false,
@@ -442,6 +456,7 @@ void BaseDeltaIterator::UpdateCurrent() {
         }
         if (delta_result != WBWIIteratorImpl::kDeleted ||
             merge_context_.GetNumOperands() > 0) {
+          // delta is visible
           current_at_base_ = false;
           SetValueAndColumnsFromDelta();
           return;
@@ -452,6 +467,7 @@ void BaseDeltaIterator::UpdateCurrent() {
           AdvanceBase();
         }
       } else {
+        // base smaller than delta
         current_at_base_ = true;
         if (!allow_unprepared_value_) {
           SetValueAndColumnsFromBase();
@@ -459,12 +475,15 @@ void BaseDeltaIterator::UpdateCurrent() {
         return;
       }
     }
+    // delta <= base and is not visible (most recent update is delete)
+    // move on to the next key
   }
 
   AssertInvariants();
 #endif  // __clang_analyzer__
 }
 
+// Move forward/backward until the iterator is at a different key.
 void WBWIIteratorImpl::AdvanceKey(bool forward) {
   if (Valid()) {
     Slice key = Entry().key;
@@ -515,18 +534,17 @@ WBWIIteratorImpl::Result WBWIIteratorImpl::FindLatestUpdate(
              0) {
     return result;
   } else {
-    // We want to iterate in the reverse order that the writes were added to the
-    // batch.  Since we don't have a reverse iterator, we must seek past the
-    // end. We do this by seeking to the next key, and then back one step
-    NextKey();
+    // Need to move to the first entry of the current key
+    // We may not always be at the first entry, e.g., after SeekToLast() or a
+    // SeekForPrev().
+    AdvanceKey(false);
     if (Valid()) {
-      Prev();
+      Next();
     } else {
-      SeekToLast();
+      SeekToFirst();
     }
-
-    // We are at the end of the iterator for this key.  Search backwards for the
-    // last Put or Delete, accumulating merges along the way.
+    // We are at the first and most recent entry for this key, search forward
+    // until a non-Merge entry, accumulating merges along the way.
     while (Valid()) {
       const WriteEntry entry = Entry();
       if (comparator_->CompareKey(column_family_id_, entry.key, key) != 0) {
@@ -551,18 +569,21 @@ WBWIIteratorImpl::Result WBWIIteratorImpl::FindLatestUpdate(
         case kPutEntityRecord:
           return WBWIIteratorImpl::kFound;
         default:
+          assert(false);
           return WBWIIteratorImpl::kError;
       }  // end switch statement
-      Prev();
+      Next();
     }  // End while Valid()
     // At this point, we have been through the whole list and found no Puts or
-    // Deletes. The iterator points to the previous key.  Move the iterator back
-    // onto this one.
+    // Deletes. The iterator points to the next key.  Move the iterator back
+    // to the last entry of this key (see comment for FindLatestUpdate()).
     if (Valid()) {
-      Next();
+      Prev();
     } else {
-      SeekToFirst();
+      SeekToLast();
     }
+    assert(Valid());
+    assert(comparator_->CompareKey(column_family_id_, Entry().key, key) == 0);
   }
   return result;
 }
@@ -650,7 +671,8 @@ Status ReadableWriteBatch::GetEntryFromDataOffset(size_t data_offset,
 // 2. Inside the same CF, we first decode the entry to find the key of the entry
 //    and the entry with larger key will be larger;
 // 3. If two entries are of the same CF and key, the one with larger offset
-//    will be larger.
+//    will be smaller. This follows the internal key order where keys
+//    with larger sequence number (larger offset here) will be ordered first.
 // Some times either `entry1` or `entry2` is dummy entry, which is actually
 // a search key. In this case, in step 2, we don't go ahead and decode the
 // entry but use the value in WriteBatchIndexEntry::search_key.
@@ -691,9 +713,9 @@ int WriteBatchEntryComparator::operator()(
   int cmp = CompareKey(entry1->column_family, key1, key2);
   if (cmp != 0) {
     return cmp;
-  } else if (entry1->offset > entry2->offset) {
-    return 1;
   } else if (entry1->offset < entry2->offset) {
+    return 1;
+  } else if (entry1->offset > entry2->offset) {
     return -1;
   }
   return 0;

utilities/write_batch_with_index/write_batch_with_index_internal.h

@@ -74,6 +74,8 @@ class BaseDeltaIterator : public Iterator {
 
  private:
   void AssertInvariants();
+  // Advance the current iterator to the next/prev key depending on forward_.
+  // If equal_keys_ is true, advance both base and delta iterators.
   void Advance();
   void AdvanceDelta();
   void AdvanceBase();
@@ -81,10 +83,17 @@ class BaseDeltaIterator : public Iterator {
   bool DeltaValid() const;
   void ResetValueAndColumns();
   void SetValueAndColumnsFromBase();
+  // Requires calling delta_iterator_->FindLatestUpdate() before this call.
+  // Will update value_ and column_ accordingly and assumes that delta iterator
+  // is visible.
   void SetValueAndColumnsFromDelta();
+  // Determine the current key and value for the iterator from base and delta
+  // iterators. Advance base or delta iters if needed.
   void UpdateCurrent();
 
   bool forward_;
+  // The non-current iterator, if valid, is at its first key that is ahead of
+  // the current iterator.
   bool current_at_base_;
   bool equal_keys_;
   bool allow_unprepared_value_;
@@ -119,9 +128,10 @@ struct WriteBatchIndexEntry {
   //                    _search_key should be null in this case.
   WriteBatchIndexEntry(const Slice* _search_key, uint32_t _column_family,
                        bool is_forward_direction, bool is_seek_to_first)
-      // For SeekForPrev(), we need to make the dummy entry larger than any
+      // For SeekForPrev(), we need to make the dummy entry no smaller than any
       // entry who has the same search key. Otherwise, we'll miss those entries.
-      : offset(is_forward_direction ? 0 : std::numeric_limits<size_t>::max()),
+      // Keys are ordered by descending offset.
+      : offset(is_forward_direction ? std::numeric_limits<size_t>::max() : 0),
         column_family(_column_family),
         has_single_del(false),
         has_overwritten_single_del(false),
@@ -141,19 +151,20 @@ struct WriteBatchIndexEntry {
     return key_size == kFlagMinInCf;
   }
 
-  // offset of an entry in write batch's string buffer. If this is a dummy
+  // The offset of an entry in write batch's string buffer. If this is a dummy
   // lookup key, in which case search_key != nullptr, offset is set to either
   // 0 or max, only for comparison purpose. Because when entries have the same
-  // key, the entry with larger offset is larger, offset = 0 will make a seek
-  // key small or equal than all the entries with the seek key, so that Seek()
-  // will find all the entries of the same key. Similarly, offset = MAX will
-  // make the entry just larger than all entries with the search key so
+  // key, the entry with larger offset is smaller, offset = MAX will make a seek
+  // key smaller than all the entries with the seek key, so that Seek()
+  // will find all the entries of the same key. Similarly, offset = 0 will
+  // make the entry larger than or equal to all entries with the seek key so
   // SeekForPrev() will see all the keys with the same key.
   size_t offset;
   uint32_t column_family;  // column family of the entry.
   bool has_single_del;     // whether single del was issued for this key
   bool has_overwritten_single_del;  // whether a single del for this key was
                                     // overwritten by another key
+  // The following two fields are used when search_key is null.
   size_t key_offset;  // offset of the key in write batch's string buffer.
   size_t key_size;    // size of the key. kFlagMinInCf indicates
                       // that this is a dummy look up entry for
@@ -354,16 +365,21 @@ class WBWIIteratorImpl final : public WBWIIterator {
   // Moves the iterator to first entry of the next key.
   void NextKey();
 
-  // Moves the iterator to the Update (Put, PutEntity or Delete) for the current
-  // key. If there is no Put/PutEntity/Delete, the Iterator will point to the
-  // first entry for this key.
-  // @return kFound if a Put/PutEntity was found for the key
-  // @return kDeleted if a delete was found for the key
-  // @return kMergeInProgress if only merges were found for the key
-  // @return kError if an unsupported operation was found for the key
-  // @return kNotFound if no operations were found for this key
+  // If the iterator's current entry equals to `key`, then
+  // - moves the iterator to the most recent update (Put, PutEntity or Delete)
+  // for `key`.
+  // - if there is no update (only Merge), the iterator will point to the last
+  // (oldest) Merge for `key`.
+  // - merge operands will be accumulated in merge_context.
+  // Else, the iterator will not move.
   //
+  // @return kFound if a Put/PutEntity was found for `key`.
+  // @return kDeleted if a Delete was found for `key`
+  // @return kMergeInProgress if only Merges were found for `key`
+  // @return kError if an unsupported operation was found for `key`
+  // @return kNotFound if no operations were found for `key`
   Result FindLatestUpdate(const Slice& key, MergeContext* merge_context);
+  // Find the latest update for iterator's current key.
   Result FindLatestUpdate(MergeContext* merge_context);
 
  protected:

utilities/write_batch_with_index/write_batch_with_index_test.cc

@@ -415,7 +415,8 @@ class WriteBatchWithIndexTest : public WBWIBaseTest,
 };
 
 void TestValueAsSecondaryIndexHelper(std::vector<Entry> entries,
-                                     WriteBatchWithIndex* batch) {
+                                     WriteBatchWithIndex* batch,
+                                     bool overwrite) {
   // In this test, we insert <key, value> to column family `data`, and
   // <value, key> to column family `index`. Then iterator them in order
   // and seek them by key.
@@ -425,8 +426,25 @@ void TestValueAsSecondaryIndexHelper(std::vector<Entry> entries,
   // Sort entries by value
   std::map<std::string, std::vector<Entry*>> index_map;
   for (auto& e : entries) {
-    data_map[e.key].push_back(&e);
-    index_map[e.value].push_back(&e);
+    if (overwrite && e.type != kMergeRecord && data_map[e.key].size() > 0) {
+      data_map[e.key].back() = &e;
+    } else {
+      data_map[e.key].push_back(&e);
+    }
+
+    // index does not use Merge, so we overwrite the expected value
+    if (overwrite && index_map[e.value].size() > 0) {
+      index_map[e.value].back() = &e;
+    } else {
+      index_map[e.value].push_back(&e);
+    }
+  }
+  // Most recent update for the same key is ordered first.
+  for (auto& [_, v] : data_map) {
+    std::reverse(v.begin(), v.end());
+  }
+  for (auto& [_, v] : index_map) {
+    std::reverse(v.begin(), v.end());
   }
 
   ColumnFamilyHandleImplDummy data(6, BytewiseComparator());
@@ -554,6 +572,25 @@ void TestValueAsSecondaryIndexHelper(std::vector<Entry> entries,
         ASSERT_OK(iter->status());
       }
     }
+
+    // SeekForPrev
+    for (auto pair = data_map.begin(); pair != data_map.end(); ++pair) {
+      iter->SeekForPrev(pair->first);
+      ASSERT_OK(iter->status());
+      // SeekForPrev positions iter at the last entry <= target.
+      // So we iterate updates from oldest to most recent.
+      for (auto v = pair->second.rbegin(); v != pair->second.rend(); ++v) {
+        ASSERT_TRUE(iter->Valid());
+        auto write_entry = iter->Entry();
+        ASSERT_EQ(pair->first, write_entry.key.ToString());
+        ASSERT_EQ((*v)->type, write_entry.type);
+        if (write_entry.type != kDeleteRecord) {
+          ASSERT_EQ((*v)->value, write_entry.value.ToString());
+        }
+        iter->Prev();
+        ASSERT_OK(iter->status());
+      }
+    }
   }
 
   // Seek to every index
@@ -576,6 +613,24 @@ void TestValueAsSecondaryIndexHelper(std::vector<Entry> entries,
         ASSERT_OK(iter->status());
       }
     }
+
+    // SeekForPrev
+    for (auto pair = index_map.begin(); pair != index_map.end(); ++pair) {
+      iter->SeekForPrev(pair->first);
+      ASSERT_OK(iter->status());
+      // SeekForPrev positions iter at the last entry <= target.
+      // So we iterate updates from oldest to most recent.
+      for (auto v = pair->second.rbegin(); v != pair->second.rend(); ++v) {
+        ASSERT_TRUE(iter->Valid());
+        auto write_entry = iter->Entry();
+        ASSERT_EQ(pair->first, write_entry.key.ToString());
+        if ((*v)->type != kDeleteRecord) {
+          ASSERT_EQ((*v)->key, write_entry.value.ToString());
+        }
+        iter->Prev();
+        ASSERT_OK(iter->status());
+      }
+    }
   }
 
   // Verify WriteBatch can be iterated
@@ -610,7 +665,7 @@ void TestValueAsSecondaryIndexHelper(std::vector<Entry> entries,
   }
 }
 
-TEST_F(WBWIKeepTest, TestValueAsSecondaryIndex) {
+TEST_P(WriteBatchWithIndexTest, TestValueAsSecondaryIndex) {
   Entry entries[] = {
       {"aaa", "0005", kPutRecord},   {"b", "0002", kPutRecord},
       {"cdd", "0002", kMergeRecord}, {"aab", "00001", kPutRecord},
@@ -619,23 +674,138 @@ TEST_F(WBWIKeepTest, TestValueAsSecondaryIndex) {
   };
   std::vector<Entry> entries_list(entries, entries + 8);
 
-  batch_.reset(new WriteBatchWithIndex(nullptr, 20, false));
+  batch_.reset(new WriteBatchWithIndex(nullptr, 20, GetParam()));
 
-  TestValueAsSecondaryIndexHelper(entries_list, batch_.get());
+  TestValueAsSecondaryIndexHelper(entries_list, batch_.get(), GetParam());
 
   // Clear batch and re-run test with new values
   batch_->Clear();
 
   Entry new_entries[] = {
-      {"aaa", "0005", kPutRecord},   {"e", "0002", kPutRecord},
-      {"add", "0002", kMergeRecord}, {"aab", "00001", kPutRecord},
-      {"zz", "00005", kPutRecord},   {"add", "0002", kPutRecord},
-      {"aab", "0003", kPutRecord},   {"zz", "00005", kDeleteRecord},
+      {"aaa", "0005", kPutRecord}, {"e", "0002", kPutRecord},
+      {"add", "0002", kPutRecord}, {"aab", "00001", kPutRecord},
+      {"zz", "00005", kPutRecord}, {"add", "0002", kMergeRecord},
+      {"aab", "0003", kPutRecord}, {"zz", "00005", kDeleteRecord},
   };
 
   entries_list = std::vector<Entry>(new_entries, new_entries + 8);
 
-  TestValueAsSecondaryIndexHelper(entries_list, batch_.get());
+  TestValueAsSecondaryIndexHelper(entries_list, batch_.get(), GetParam());
+}
+
+TEST_P(WriteBatchWithIndexTest, WBWIIteratorImpl) {
+  // Tests methods of WBWIIteratorImpl, with some overwrites and merges.
+  ASSERT_OK(batch_->Merge("k0", "k0m0"));
+  ASSERT_OK(batch_->Put("k0", "k0p1"));
+
+  // a merge and a non-merge
+  ASSERT_OK(batch_->Merge("k1", "k1m0"));
+  ASSERT_OK(batch_->Merge("k1", "k1m1"));
+
+  ASSERT_OK(batch_->SingleDelete("k2"));
+
+  // put then merge
+  ASSERT_OK(batch_->Put("k3", "k3p0"));
+  ASSERT_OK(batch_->Merge("k3", "k3m1"));
+
+  std::unique_ptr<WBWIIteratorImpl> iter(
+      static_cast<WBWIIteratorImpl*>(batch_->NewIterator()));
+
+  auto verify_iter = [&iter](const std::string& k, const std::string& v,
+                             WriteType t, int line) {
+    SCOPED_TRACE("Called from line " + std::to_string(line));
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    const auto entry = iter->Entry();
+    ASSERT_EQ(entry.key, k);
+    if (t != kDeleteRecord && t != kSingleDeleteRecord) {
+      ASSERT_EQ(entry.value, v);
+    }
+    ASSERT_EQ(entry.type, t);
+  };
+
+  // Should land on first key >= k0, recent update is ordered first
+  iter->Seek("k0");
+  verify_iter("k0", "k0p1", kPutRecord, __LINE__);
+  // Should land on the first update of the next key
+  iter->NextKey();
+  verify_iter("k1", "k1m1", kMergeRecord, __LINE__);
+  iter->NextKey();
+  verify_iter("k2", "", kSingleDeleteRecord, __LINE__);
+  iter->NextKey();
+  verify_iter("k3", "k3m1", kMergeRecord, __LINE__);
+  iter->PrevKey();
+  verify_iter("k2", "", kSingleDeleteRecord, __LINE__);
+
+  // Should land on last key <= k0, recent update is ordered first
+  iter->SeekForPrev("k3");
+  verify_iter("k3", "k3p0", kPutRecord, __LINE__);
+  iter->PrevKey();
+  verify_iter("k2", "", kSingleDeleteRecord, __LINE__);
+  iter->PrevKey();
+  verify_iter("k1", "k1m1", kMergeRecord, __LINE__);
+  iter->PrevKey();
+  verify_iter("k0", "k0p1", kPutRecord, __LINE__);
+  iter->NextKey();
+  verify_iter("k1", "k1m1", kMergeRecord, __LINE__);
+
+  // test FindLatestUpdate
+  iter->SeekToFirst();
+  verify_iter("k0", "k0p1", kPutRecord, __LINE__);
+  MergeContext merge_context;
+  // iterator is not at k1
+  ASSERT_EQ(iter->FindLatestUpdate("k1", &merge_context),
+            WBWIIteratorImpl::Result::kNotFound);
+  ASSERT_EQ(merge_context.GetNumOperands(), 0);
+  // iterator was not moved
+  verify_iter("k0", "k0p1", kPutRecord, __LINE__);
+
+  // k0's most recent update is a Put
+  ASSERT_EQ(iter->FindLatestUpdate("k0", &merge_context),
+            WBWIIteratorImpl::Result::kFound);
+  ASSERT_EQ(merge_context.GetNumOperands(), 0);
+  verify_iter("k0", "k0p1", kPutRecord, __LINE__);
+
+  iter->NextKey();
+  verify_iter("k1", "k1m1", kMergeRecord, __LINE__);
+  ASSERT_EQ(iter->FindLatestUpdate("k1", &merge_context),
+            WBWIIteratorImpl::Result::kMergeInProgress);
+  ASSERT_EQ(merge_context.GetNumOperands(), 2);
+  auto operands = merge_context.GetOperands();
+  // oldest merge is ordered first
+  ASSERT_EQ(operands[0], "k1m0");
+  ASSERT_EQ(operands[1], "k1m1");
+  // iterator moved to last merge
+  verify_iter("k1", "k1m0", kMergeRecord, __LINE__);
+  merge_context.Clear();
+
+  iter->Next();
+  verify_iter("k2", "", kSingleDeleteRecord, __LINE__);
+  ASSERT_EQ(iter->FindLatestUpdate("k2", &merge_context),
+            WBWIIteratorImpl::Result::kDeleted);
+  ASSERT_EQ(merge_context.GetNumOperands(), 0);
+  verify_iter("k2", "", kSingleDeleteRecord, __LINE__);
+
+  iter->Next();
+  verify_iter("k3", "k3m1", kMergeRecord, __LINE__);
+  // This will find latest updated of the current key
+  ASSERT_EQ(iter->FindLatestUpdate(&merge_context),
+            WBWIIteratorImpl::Result::kFound);
+  ASSERT_EQ(merge_context.GetNumOperands(), 1);
+  operands = merge_context.GetOperands();
+  ASSERT_EQ(operands[0], "k3m1");
+  // iterator moved to last merge
+  verify_iter("k3", "k3p0", kPutRecord, __LINE__);
+
+  // SeekToLast
+  iter->SeekToLast();
+  verify_iter("k3", "k3p0", kPutRecord, __LINE__);
+  // FindLatestUpdate() while we are at the older update of k3
+  ASSERT_EQ(iter->FindLatestUpdate(&merge_context),
+            WBWIIteratorImpl::Result::kFound);
+  ASSERT_EQ(merge_context.GetNumOperands(), 1);
+  operands = merge_context.GetOperands();
+  ASSERT_EQ(operands[0], "k3m1");
 }
 
 TEST_P(WriteBatchWithIndexTest, TestComparatorForCF) {
@@ -2599,7 +2769,7 @@ TEST_P(WriteBatchWithIndexTest, GetFromBatchAndDBAfterMerge) {
   ASSERT_EQ(value, "cc");
 }
 
-TEST_F(WBWIKeepTest, GetAfterPut) {
+TEST_P(WriteBatchWithIndexTest, GetAfterPut) {
   std::string value;
   ASSERT_OK(OpenDB());
   ColumnFamilyHandle* cf0 = db_->DefaultColumnFamily();
@@ -2691,7 +2861,7 @@ TEST_P(WriteBatchWithIndexTest, GetAfterMergeDelete) {
   ASSERT_EQ(value, "cc,dd");
 }
 
-TEST_F(WBWIOverwriteTest, TestBadMergeOperator) {
+TEST_P(WriteBatchWithIndexTest, TestBadMergeOperator) {
   class FailingMergeOperator : public MergeOperator {
    public:
     FailingMergeOperator() = default;
@@ -2837,16 +3007,17 @@ TEST_P(WriteBatchWithIndexTest, ColumnFamilyWithTimestamp) {
       std::string key;
       PutFixed32(&key, start);
       ASSERT_EQ(key, it->Entry().key);
-      if (!overwrite) {
-        ASSERT_EQ(WriteType::kPutRecord, it->Entry().type);
-        it->Next();
-        ASSERT_TRUE(it->Valid());
-      }
+      // For each key, most recent update (Del) is ordered first.
       if (0 == (start % 2)) {
         ASSERT_EQ(WriteType::kDeleteRecord, it->Entry().type);
       } else {
         ASSERT_EQ(WriteType::kSingleDeleteRecord, it->Entry().type);
       }
+      if (!overwrite) {
+        it->Next();
+        ASSERT_TRUE(it->Valid());
+        ASSERT_EQ(WriteType::kPutRecord, it->Entry().type);
+      }
     }
   }
 }