Merge support in WBWIMemTable (#13410)

Summary:
added merge support for WBWIMemTable. Most of the preparation work is done in https://github.com/facebook/rocksdb/issues/13387 and https://github.com/facebook/rocksdb/issues/13400. The main code change to support merge is in wbwi_memtable.cc to support reading the Merge value type. The rest of the changes are mostly comment change and tests.

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

Test Plan:
- new unit test
- ran `python3 ./tools/db_crashtest.py --txn blackbox  --txn_write_policy=0 --commit_bypass_memtable_one_in=100 --test_batches_snapshots=0 --use_merge=1` for several runs.

Reviewed By: jowlyzhang

Differential Revision: D69885868

Pulled By: cbi42

fbshipit-source-id: b127d95a3027dc35910f6e5d65f3409ba27e2b6b

db/db_impl/db_impl_write.cc

@@ -587,6 +587,7 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
 
   IOStatus io_s;
   Status pre_release_cb_status;
+  size_t seq_inc = 0;
   if (status.ok()) {
     // Rules for when we can update the memtable concurrently
     // 1. supported by memtable
@@ -640,7 +641,7 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
     // disable_memtable in between; although we do not write this batch to
     // memtable it still consumes a seq. Otherwise, if !seq_per_batch_, we inc
     // the seq per valid written key to mem.
-    size_t seq_inc = seq_per_batch_ ? valid_batches : total_count;
+    seq_inc = seq_per_batch_ ? valid_batches : total_count;
     if (wbwi) {
       // Reserve sequence numbers for the ingested memtable. We need to reserve
       // at lease this amount for recovery. During recovery,
@@ -713,6 +714,7 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
     assert(last_sequence != kMaxSequenceNumber);
     const SequenceNumber current_sequence = last_sequence + 1;
     last_sequence += seq_inc;
+    // Seqno assigned to this write are [current_sequence, last_sequence]
 
     if (log_context.need_log_sync) {
       VersionEdit synced_wals;
@@ -836,13 +838,17 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
     if (status.ok() && w.status.ok()) {
       // w.batch contains (potentially empty) commit time batch updates,
       // only ingest wbwi if w.batch is applied to memtable successfully
-      assert(wbwi->GetWriteBatch()->Count() > 0);
-
       uint32_t memtable_update_count = w.batch->Count();
-      SequenceNumber lb = versions_->LastSequence() + memtable_update_count + 1;
-      SequenceNumber ub = versions_->LastSequence() + memtable_update_count +
-                          wbwi->GetWriteBatch()->Count();
-      assert(ub == last_sequence);
+      uint32_t wbwi_count = wbwi->GetWriteBatch()->Count();
+      // Seqno assigned to this write are [last_seq + 1 - seq_inc, last_seq].
+      // seq_inc includes w.batch (memtable updates) and wbwi
+      // w.batch gets first `memtable_update_count` sequence numbers.
+      // wbwi gets the rest `wbwi_count` sequence numbers.
+      assert(seq_inc == memtable_update_count + wbwi_count);
+      assert(wbwi_count > 0);
+      assert(last_sequence != kMaxSequenceNumber);
+      SequenceNumber lb = last_sequence + 1 - wbwi_count;
+      SequenceNumber ub = last_sequence;
       if (two_write_queues_) {
         assert(ub <= versions_->LastAllocatedSequence());
       }

db/db_merge_operator_test.cc

@@ -120,6 +120,10 @@ TEST_F(DBMergeOperatorTest, LimitMergeOperands) {
   ASSERT_OK(Merge("k3", "de"));
   ASSERT_OK(db_->Get(ReadOptions(), "k3", &value));
   ASSERT_EQ(value, "cd,de");
+  // Tests that merge operands reach exact limit at memtable.
+  ASSERT_OK(Merge("k3", "fg"));
+  ASSERT_OK(db_->Get(ReadOptions(), "k3", &value));
+  ASSERT_EQ(value, "de,fg");
 
   // All K4 values are in different levels
   ASSERT_OK(Merge("k4", "ab"));

db/memtable.cc

@@ -1325,50 +1325,13 @@ static bool SaveValue(void* arg, const char* entry) {
         return false;
       }
       case kTypeMerge: {
-        if (!merge_operator) {
-          *(s->status) = Status::InvalidArgument(
-              "merge_operator is not properly initialized.");
-          // Normally we continue the loop (return true) when we see a merge
-          // operand.  But in case of an error, we should stop the loop
-          // immediately and pretend we have found the value to stop further
-          // seek.  Otherwise, the later call will override this error status.
-          *(s->found_final_value) = true;
-          return false;
-        }
         Slice v = GetLengthPrefixedSlice(key_ptr + key_length);
         *(s->merge_in_progress) = true;
-        merge_context->PushOperand(
-            v, s->inplace_update_support == false /* operand_pinned */);
-        PERF_COUNTER_ADD(internal_merge_point_lookup_count, 1);
-
-        if (s->do_merge && merge_operator->ShouldMerge(
-                               merge_context->GetOperandsDirectionBackward())) {
-          if (s->value || s->columns) {
-            // `op_failure_scope` (an output parameter) is not provided (set to
-            // nullptr) since a failure must be propagated regardless of its
-            // value.
-            *(s->status) = MergeHelper::TimedFullMerge(
-                merge_operator, s->key->user_key(), MergeHelper::kNoBaseValue,
-                merge_context->GetOperands(), s->logger, s->statistics,
-                s->clock, /* update_num_ops_stats */ true,
-                /* op_failure_scope */ nullptr, s->value, s->columns);
-          }
-
-          *(s->found_final_value) = true;
-          return false;
-        }
-        if (merge_context->get_merge_operands_options != nullptr &&
-            merge_context->get_merge_operands_options->continue_cb != nullptr &&
-            !merge_context->get_merge_operands_options->continue_cb(v)) {
-          // We were told not to continue. `status` may be MergeInProress(),
-          // overwrite to signal the end of successful get. This status
-          // will be checked at the end of GetImpl().
-          *(s->status) = Status::OK();
-          *(s->found_final_value) = true;
-          return false;
-        }
-
-        return true;
+        *(s->found_final_value) = ReadOnlyMemTable::HandleTypeMerge(
+            s->key->user_key(), v, s->inplace_update_support == false,
+            s->do_merge, merge_context, s->merge_operator, s->clock,
+            s->statistics, s->logger, s->status, s->value, s->columns);
+        return !*(s->found_final_value);
       }
       default: {
         std::string msg("Corrupted value not expected.");

db/memtable.h

@@ -186,10 +186,13 @@ class ReadOnlyMemTable {
                                               SequenceNumber read_seq,
                                               size_t ts_sz) = 0;
 
-  // Used to Get value associated with key or Get Merge Operands associated
-  // with key.
-  // Keys are considered if they are no larger than the parameter `key` in
+  // Used to get value associated with `key`, or Merge operands associated
+  // with key, or get the latest sequence number of `key` (e.g. transaction
+  // conflict checking).
+  //
+  // Keys are considered if they are no smaller than the parameter `key` in
   // the order defined by comparator and share the save user key with `key`.
+  //
   // If do_merge = true the default behavior which is Get value for key is
   // executed. Expected behavior is described right below.
   // If memtable contains a value for key, store it in *value and return true.
@@ -207,6 +210,7 @@ class ReadOnlyMemTable {
   // returned).  Otherwise, *seq will be set to kMaxSequenceNumber.
   // On success, *s may be set to OK, NotFound, or MergeInProgress.  Any other
   // status returned indicates a corruption or other unexpected error.
+  //
   // If do_merge = false then any Merge Operands encountered for key are simply
   // stored in merge_context.operands_list and never actually merged to get a
   // final value. The raw Merge Operands are eventually returned to the user.
@@ -215,6 +219,9 @@ class ReadOnlyMemTable {
   // @param column If not null and memtable contains a value/WideColumn for key,
   // `column` will be set to the result value/WideColumn.
   // Note: only one of `value` and `column` can be non-nullptr.
+  // To only query for key existence or the latest sequence number of a key,
+  // `value` and `column` can be both nullptr. In this case, returned status can
+  // be OK, NotFound or MergeInProgress if a key is found.
   // @param immutable_memtable Whether this memtable is immutable. Used
   // internally by NewRangeTombstoneIterator(). See comment above
   // NewRangeTombstoneIterator() for more detail.
@@ -394,7 +401,6 @@ class ReadOnlyMemTable {
       // can also be retained.
       merge_context->PushOperand(value, value_pinned);
     } else if (merge_in_progress) {
-      assert(do_merge);
       // `op_failure_scope` (an output parameter) is not provided (set to
       // nullptr) since a failure must be propagated regardless of its
       // value.
@@ -442,6 +448,54 @@ class ReadOnlyMemTable {
     }
   }
 
+  // Returns if a final value is found.
+  static bool HandleTypeMerge(const Slice& lookup_user_key, const Slice& value,
+                              bool value_pinned, bool do_merge,
+                              MergeContext* merge_context,
+                              const MergeOperator* merge_operator,
+                              SystemClock* clock, Statistics* statistics,
+                              Logger* logger, Status* s, std::string* out_value,
+                              PinnableWideColumns* out_columns) {
+    if (!merge_operator) {
+      *s = Status::InvalidArgument(
+          "merge_operator is not properly initialized.");
+      // Normally we continue the loop (return true) when we see a merge
+      // operand.  But in case of an error, we should stop the loop
+      // immediately and pretend we have found the value to stop further
+      // seek.  Otherwise, the later call will override this error status.
+      return true;
+    }
+    merge_context->PushOperand(value, value_pinned /* operand_pinned */);
+    PERF_COUNTER_ADD(internal_merge_point_lookup_count, 1);
+
+    if (do_merge && merge_operator->ShouldMerge(
+                        merge_context->GetOperandsDirectionBackward())) {
+      if (out_value || out_columns) {
+        // `op_failure_scope` (an output parameter) is not provided (set to
+        // nullptr) since a failure must be propagated regardless of its
+        // value.
+        *s = MergeHelper::TimedFullMerge(
+            merge_operator, lookup_user_key, MergeHelper::kNoBaseValue,
+            merge_context->GetOperands(), logger, statistics, clock,
+            /* update_num_ops_stats */ true,
+            /* op_failure_scope */ nullptr, out_value, out_columns);
+      }
+      return true;
+    }
+    if (merge_context->get_merge_operands_options != nullptr &&
+        merge_context->get_merge_operands_options->continue_cb != nullptr &&
+        !merge_context->get_merge_operands_options->continue_cb(value)) {
+      // We were told not to continue. `status` may be MergeInProress(),
+      // overwrite to signal the end of successful get. This status
+      // will be checked at the end of GetImpl().
+      *s = Status::OK();
+      return true;
+    }
+
+    // no final value found yet
+    return false;
+  }
+
  protected:
   friend class MemTableList;
 

include/rocksdb/utilities/transaction_db.h

@@ -372,6 +372,11 @@ struct TransactionOptions {
   // without indexing (e.g. added directly to the transaction underlying
   // write batch through Transaction::GetWriteBatch()->GetWriteBatch())
   // are not supported. They will not be applied to the DB.
+  //
+  // NOTE: since WBWI keep track of the most recent update per key, a Put
+  // followed by a SingleDelete will be written to DB as a SingleDelete. This
+  // can cause flush/compaction to report `num_single_del_mismatch` due to
+  // consecutive SingleDeletes.
   bool commit_bypass_memtable = false;
 };
 

memtable/wbwi_memtable.cc

@@ -49,16 +49,17 @@ bool WBWIMemTable::Get(const LookupKey& key, std::string* value,
                        SequenceNumber* out_seq, const ReadOptions&,
                        bool immutable_memtable, ReadCallback* callback,
                        bool* is_blob_index, bool do_merge) {
+  assert(s->ok() || s->IsMergeInProgress());
   (void)immutable_memtable;
   (void)timestamp;
   (void)columns;
   assert(immutable_memtable);
   assert(!timestamp);  // TODO: support UDT
-  assert(!columns);    // TODO: support WideColumn
   assert(assigned_seqno_.upper_bound != kMaxSequenceNumber);
   assert(assigned_seqno_.lower_bound != kMaxSequenceNumber);
   // WBWI does not support DeleteRange yet.
   assert(!wbwi_->GetWriteBatch()->HasDeleteRange());
+  assert(merge_context);
 
   [[maybe_unused]] SequenceNumber read_seq =
       GetInternalKeySeqno(key.internal_key());
@@ -69,6 +70,7 @@ bool WBWIMemTable::Get(const LookupKey& key, std::string* value,
   std::unique_ptr<InternalIterator> iter{NewIterator()};
   iter->Seek(key.internal_key());
   const Slice lookup_user_key = key.user_key();
+  bool merge_in_progress = s->IsMergeInProgress();
 
   while (iter->Valid() && comparator_->EqualWithoutTimestamp(
                               ExtractUserKey(iter->key()), lookup_user_key)) {
@@ -81,7 +83,6 @@ bool WBWIMemTable::Get(const LookupKey& key, std::string* value,
     assert(type != kTypeWideColumnEntity);
     assert(type != kTypeValuePreferredSeqno);
     assert(type != kTypeDeletionWithTimestamp);
-    assert(type != kTypeMerge);
     if (!callback || callback->IsVisible(seq)) {
       if (*out_seq == kMaxSequenceNumber) {
         *out_seq = std::max(seq, *max_covering_tombstone_seq);
@@ -92,7 +93,7 @@ bool WBWIMemTable::Get(const LookupKey& key, std::string* value,
       switch (type) {
         case kTypeValue: {
           HandleTypeValue(lookup_user_key, iter->value(), iter->IsValuePinned(),
-                          do_merge, s->IsMergeInProgress(), merge_context,
+                          do_merge, merge_in_progress, merge_context,
                           moptions_.merge_operator, clock_,
                           moptions_.statistics, moptions_.info_log, s, value,
                           columns, is_blob_index);
@@ -102,16 +103,29 @@ bool WBWIMemTable::Get(const LookupKey& key, std::string* value,
         case kTypeDeletion:
         case kTypeSingleDeletion:
         case kTypeRangeDeletion: {
-          HandleTypeDeletion(lookup_user_key, s->IsMergeInProgress(),
-                             merge_context, moptions_.merge_operator, clock_,
+          HandleTypeDeletion(lookup_user_key, merge_in_progress, merge_context,
+                             moptions_.merge_operator, clock_,
                              moptions_.statistics, moptions_.info_log, s, value,
                              columns);
           assert(seq <= read_seq);
           return /*found_final_value=*/true;
         }
+        case kTypeMerge: {
+          merge_in_progress = true;
+          if (ReadOnlyMemTable::HandleTypeMerge(
+                  lookup_user_key, iter->value(), iter->IsValuePinned(),
+                  do_merge, merge_context, moptions_.merge_operator, clock_,
+                  moptions_.statistics, moptions_.info_log, s, value,
+                  columns)) {
+            return true;
+          }
+          break;
+        }
         default: {
-          std::string msg("Unrecognized or unsupported value type: " +
-                          std::to_string(static_cast<int>(type)) + ". ");
+          std::string msg(
+              "Unrecognized or unsupported value type for "
+              "WBWI-based memtable: " +
+              std::to_string(static_cast<int>(type)) + ". ");
           msg.append("User key: " +
                      ExtractUserKey(iter->key()).ToString(/*hex=*/true) + ". ");
           msg.append("seq: " + std::to_string(seq) + ".");
@@ -120,8 +134,8 @@ bool WBWIMemTable::Get(const LookupKey& key, std::string* value,
         }
       }
     }
-    // Current key not visible or we read a merge key
-    assert(s->IsMergeInProgress() || (callback && !callback->IsVisible(seq)));
+    // Current key is a merge key or not visible
+    assert(merge_in_progress || (callback && !callback->IsVisible(seq)));
     iter->Next();
   }
   if (!iter->status().ok() &&
@@ -130,6 +144,10 @@ bool WBWIMemTable::Get(const LookupKey& key, std::string* value,
     // stop further look up
     return true;
   }
+  if (merge_in_progress) {
+    assert(s->ok() || s->IsMergeInProgress());
+    *s = Status::MergeInProgress();
+  }
   return /*found_final_value=*/false;
 }
 

memtable/wbwi_memtable.h

@@ -12,7 +12,7 @@ namespace ROCKSDB_NAMESPACE {
 // of the given write batch with index (WBWI) object. This can be used to ingest
 // a transaction (which is based on WBWI) into the DB as an immutable memtable.
 //
-// REQUIRE overwrite_key to be true for the WBWI
+// REQUIRES: overwrite_key to be true for the WBWI
 // Since the keys in WBWI do not have sequence number, this class is responsible
 // for assigning sequence numbers to the keys. This memtable needs to be
 // assigned a range of sequence numbers through AssignSequenceNumbers(seqno)
@@ -23,10 +23,13 @@ namespace ROCKSDB_NAMESPACE {
 // sequence number assigned is seqno.lower_bound + update_count - 1. So more
 // recent updates will have higher sequence number.
 //
-// WBWI with overwrite mode keeps track of the most recent update for each key,
-// so this memtable contains one update per key usually. However, there is a
-// special case where this memtable needs to emit an extra SingleDelete even
-// when the SD is overwritten by another update.
+// Since WBWI with overwrite mode keeps track of the most recent update for
+// each key, this memtable contains one update per key usually. However, there
+// are two exceptions:
+// 1. Merge operations: Each Merge operation do not overwrite existing entries,
+// if a user uses Merge, multiple entries may be kept.
+// 2. Overwriten SingleDelete: this memtable needs to emit an extra
+// SingleDelete even when the SD is overwritten by another update.
 // Consider the following scenario:
 // - WBWI has SD(k) then PUT(k, v1)
 // - DB has PUT(k, v2) in L1
@@ -261,6 +264,7 @@ class WBWIMemTableIterator final : public InternalIterator {
   }
 
   void SeekToLast() override {
+    assert(!emit_overwritten_single_del_);
     it_->SeekToLast();
     UpdateKey();
   }
@@ -303,6 +307,8 @@ class WBWIMemTableIterator final : public InternalIterator {
     assert(Valid());
     if (emit_overwritten_single_del_) {
       if (it_->HasOverWrittenSingleDel() && !at_overwritten_single_del_) {
+        // Merge and SingleDelete on the same key is undefined behavior.
+        assert(it_->Entry().type != kMergeRecord);
         UpdateSingleDeleteKey();
         return;
       }

tools/db_crashtest.py

@@ -1025,8 +1025,6 @@ def finalize_and_sanitize(src_params):
         dest_params["use_put_entity_one_in"] = 0
         dest_params["use_get_entity"] = 0
         dest_params["use_multi_get_entity"] = 0
-        dest_params["use_merge"] = 0
-        dest_params["use_full_merge_v1"] = 0
         dest_params["enable_pipelined_write"] = 0
         dest_params["use_attribute_group"] = 0
     # Continuous verification fails with secondaries inside NonBatchedOpsStressTest

utilities/transactions/transaction_test.cc

@@ -9450,6 +9450,212 @@ TEST_P(CommitBypassMemtableTest, ThresholdTxnDBOption) {
   }
 }
 
+TEST_P(CommitBypassMemtableTest, MergeAndMultiCF) {
+  // disable_flush allows testing Get path with memtables.
+  for (bool disable_flush : {false, true}) {
+    SCOPED_TRACE("disable_flush: " + std::to_string(disable_flush));
+    SetUpTransactionDB();
+    if (disable_flush) {
+      ASSERT_OK(txn_db->PauseBackgroundWork());
+    }
+
+    std::vector<std::string> cfs = {"appendmerge"};
+    Options opts;
+    opts.max_write_buffer_number = 8;
+    opts.merge_operator = MergeOperators::CreateFromStringId("stringappend");
+    CreateColumnFamilies(cfs, opts);
+
+    cfs = {"uint64addmerge"};
+    opts.merge_operator = MergeOperators::CreateFromStringId("uint64add");
+    CreateColumnFamilies(cfs, opts);
+
+    cfs = {"data"};
+    opts.merge_operator = nullptr;
+    CreateColumnFamilies(cfs, opts);
+    ASSERT_TRUE(handles_.size() == 3);
+
+    std::string buf_count;
+    PutFixed64(&buf_count, 1);
+    // Some base data in SST or memtable
+    ASSERT_OK(db_->Merge({}, handles_[1], "count", buf_count));
+    ASSERT_OK(db_->Put({}, handles_[0], "k5", "5v1"));
+    ASSERT_OK(db_->Merge({}, handles_[0], "k7", "7v1"));
+    if (!disable_flush) {
+      ASSERT_OK(db_->Flush({}, handles_[1]));
+    }
+
+    WriteOptions wopts;
+    TransactionOptions txn_opts;
+    txn_opts.commit_bypass_memtable = true;
+    Transaction* txn = txn_db->BeginTransaction(wopts, txn_opts);
+    ASSERT_OK(txn->SetName("xid1"));
+    ASSERT_OK(txn->Put(handles_[0], "k1", "v1"));
+    ASSERT_OK(txn->Merge(handles_[0], "k1", "v2"));
+
+    ASSERT_OK(txn->Delete(handles_[0], "k2"));
+    ASSERT_OK(txn->Merge(handles_[0], "k2", "v1"));
+
+    ASSERT_OK(txn->Merge(handles_[0], "k3", "v1"));
+    ASSERT_OK(txn->Delete(handles_[0], "k3"));
+
+    ASSERT_OK(txn->Merge(handles_[0], "k4", "v1"));
+    ASSERT_OK(txn->Put(handles_[0], "k4", "v4"));
+
+    ASSERT_OK(txn->Merge(handles_[0], "k5", "5v2"));
+
+    ASSERT_OK(txn->Merge(handles_[0], "k6", "6v1"));
+    ASSERT_OK(txn->Merge(handles_[0], "k6", "6v2"));
+
+    ASSERT_OK(txn->Merge(handles_[0], "k7", "7v2"));
+    ASSERT_OK(txn->Merge(handles_[0], "k7", "7v3"));
+
+    ASSERT_OK(txn->Merge(handles_[1], "count", buf_count));
+    ASSERT_OK(txn->Merge(handles_[1], "count", buf_count));
+
+    ASSERT_OK(txn->Put(handles_[2], "a", "a1"));
+    ASSERT_OK(txn->Put(handles_[2], "c", "c1"));
+
+    ASSERT_OK(txn->Prepare());
+    ASSERT_OK(txn->Commit());
+
+    // Data in mutable memtable
+    txn_opts.commit_bypass_memtable = false;
+    txn = txn_db->BeginTransaction(wopts, txn_opts, txn);
+    ASSERT_OK(txn->SetName("xid2"));
+    ASSERT_OK(txn->Merge(handles_[0], "k1", "v3"));
+    ASSERT_OK(txn->Merge(handles_[1], "count", buf_count));
+    ASSERT_OK(txn->Prepare());
+    ASSERT_OK(txn->Commit());
+    delete txn;
+
+    std::map<std::string, std::string> expected_cf0 = {
+        {"k1", "v1,v2,v3"}, {"k2", "v1"},      {"k4", "v4"},
+        {"k5", "5v1,5v2"},  {"k6", "6v1,6v2"}, {"k7", "7v1,7v2,7v3"},
+    };
+    std::unordered_set<std::string> not_found_cf0 = {"k3"};
+    VerifyDBFromMap(expected_cf0, nullptr, false, nullptr, handles_[0],
+                    &not_found_cf0);
+
+    std::string count;
+    PutFixed64(&count, 4);
+    std::map<std::string, std::string> expected_cf1 = {
+        {"count", count},
+    };
+    VerifyDBFromMap(expected_cf1, nullptr, false, nullptr, handles_[1]);
+
+    std::map<std::string, std::string> expected_cf2 = {
+        {"a", "a1"},
+        {"c", "c1"},
+    };
+    VerifyDBFromMap(expected_cf2, nullptr, false, nullptr, handles_[2]);
+
+    // Verify all data is flushed
+    if (disable_flush) {
+      uint64_t num_imm_mems = 0;
+      ASSERT_TRUE(txn_db->GetIntProperty(
+          handles_[0], DB::Properties::kNumImmutableMemTable, &num_imm_mems));
+      ASSERT_EQ(2,
+                num_imm_mems);  // 1 imm mem before WBWI, 1 imm is WBWI itself
+
+      // Test GetMergeOperands() for CF0
+      std::vector<PinnableSlice> merge_operands(4);
+      GetMergeOperandsOptions merge_operand_opts;
+      merge_operand_opts.expected_max_number_of_operands = 4;
+      int num_operands = 0;
+      ASSERT_OK(db_->GetMergeOperands({}, handles_[0], "k1",
+                                      merge_operands.data(),
+                                      &merge_operand_opts, &num_operands));
+      ASSERT_EQ(num_operands, 3);
+      ASSERT_EQ(merge_operands[0], "v1");
+      ASSERT_EQ(merge_operands[1], "v2");
+      ASSERT_EQ(merge_operands[2], "v3");
+
+      ASSERT_OK(db_->ContinueBackgroundWork());
+      ASSERT_OK(db_->Flush({}, {handles_[0], handles_[1], handles_[2]}));
+    } else {
+      ASSERT_OK(db_->WaitForCompact({}));
+    }
+
+    VerifyDBFromMap(expected_cf0, nullptr, false, nullptr, handles_[0],
+                    &not_found_cf0);
+    VerifyDBFromMap(expected_cf1, nullptr, false, nullptr, handles_[1]);
+    VerifyDBFromMap(expected_cf2, nullptr, false, nullptr, handles_[2]);
+  }
+}
+
+TEST_P(CommitBypassMemtableTest, MergeMiniStress) {
+  // To test the merge path with various LSM shapes
+  std::string key_prefix = "key";
+  std::string value_prefix = "val";
+  Random* rnd = Random::GetTLSInstance();
+  const int kBatchSize = 50;
+  for (int num_memtable_to_merge : {1, 4}) {
+    SetUpTransactionDB();
+    std::vector<std::string> cfs = {"appendmerge"};
+    Options opts;
+    opts.max_write_buffer_number = 10;
+    // Exercise read path of memtables.
+    opts.min_write_buffer_number_to_merge = num_memtable_to_merge;
+    opts.merge_operator = MergeOperators::CreateFromStringId("stringappend");
+    CreateColumnFamilies(cfs, opts);
+    ASSERT_TRUE(handles_.size() == 1);
+
+    std::map<std::string, std::string> expected_cf;
+    std::unordered_set<std::string> not_found_cf;
+    for (int i = 0; i < 10000; i += kBatchSize) {
+      WriteOptions wopts;
+      TransactionOptions txn_opts;
+      txn_opts.commit_bypass_memtable = rnd->OneIn(2);
+      std::unique_ptr<Transaction> txn{
+          txn_db->BeginTransaction(wopts, txn_opts)};
+      const int txn_count = i / kBatchSize;
+      ASSERT_OK(txn->SetName("xid" + std::to_string(txn_count)));
+
+      const Snapshot* snapshot = txn_db->GetSnapshot();
+      // Remember the state for the snapshot
+      auto expected_cf_snapshot = expected_cf;
+      auto not_found_cf_snapshot = not_found_cf;
+
+      for (int j = 0; j < kBatchSize; ++j) {
+        std::string key = key_prefix + std::to_string(rnd->Uniform(1000));
+        std::string value = value_prefix + std::to_string(i + j);
+        int operation = rnd->Uniform(10);
+        if (operation < 8) {  // 80% probability for Merge
+          ASSERT_OK(txn->Merge(handles_[0], key, value));
+          if (expected_cf.find(key) != expected_cf.end()) {
+            expected_cf[key] += "," + value;
+          } else {
+            expected_cf[key] = value;
+          }
+          not_found_cf.erase(key);
+        } else if (operation == 8) {  // 10% probability for PUT
+          ASSERT_OK(txn->Put(handles_[0], key, value));
+          expected_cf[key] = value;
+          not_found_cf.erase(key);
+        } else {  // 10% probability for DEL
+          ASSERT_OK(txn->Delete(handles_[0], key));
+          expected_cf.erase(key);
+          not_found_cf.insert(key);
+        }
+      }
+      ASSERT_OK(txn->Prepare());
+      ASSERT_OK(txn->Commit());
+
+      if (txn_count % 10 == 0) {
+        VerifyDBFromMap(expected_cf, nullptr, false, nullptr, handles_[0],
+                        &not_found_cf);
+        // Verify read at snapshot
+        ReadOptions ro;
+        ro.snapshot = snapshot;
+        VerifyDBFromMap(expected_cf_snapshot, nullptr, false, &ro, handles_[0],
+                        &not_found_cf_snapshot);
+      }
+      txn_db->ReleaseSnapshot(snapshot);
+    }
+
+    VerifyDBFromMap(expected_cf, nullptr, false, nullptr, handles_[0]);
+  }
+}
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

utilities/write_batch_with_index/write_batch_with_index.cc

@@ -136,7 +136,12 @@ bool WriteBatchWithIndex::Rep::UpdateExistingEntryWithCfId(
       most_recent_entry->has_single_del = true;
     }
   }
+  // Some sanity check for using Merge and SD on the same key.
+  if (iter.Entry().type == kSingleDeleteRecord) {
+    assert(type != kMergeRecord);
+  }
   if (type == kMergeRecord) {
+    assert(iter.Entry().type != kSingleDeleteRecord);
     return false;
   } else {
     // We still increment the update count when updating in-place. This is

utilities/write_batch_with_index/write_batch_with_index_test.cc

@@ -15,7 +15,6 @@
 #include <memory>
 
 #include "db/column_family.h"
-#include "db/wide/wide_columns_helper.h"
 #include "memtable/wbwi_memtable.h"
 #include "port/stack_trace.h"
 #include "test_util/testharness.h"
@@ -23,7 +22,6 @@
 #include "util/random.h"
 #include "util/string_util.h"
 #include "utilities/merge_operators.h"
-#include "utilities/merge_operators/string_append/stringappend.h"
 #include "utilities/write_batch_with_index/write_batch_with_index_internal.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -3695,15 +3693,20 @@ TEST_P(WriteBatchWithIndexTest, TrackAndClearCFStats) {
 INSTANTIATE_TEST_CASE_P(WBWI, WriteBatchWithIndexTest, testing::Bool());
 
 std::string Get(const std::string& k, std::unique_ptr<WBWIMemTable>& wbwi_mem,
-                SequenceNumber snapshot_seq, bool* found_final_value) {
+                SequenceNumber snapshot_seq, bool* found_final_value,
+                MergeContext* merge_context = nullptr) {
   LookupKey lkey(k, snapshot_seq);
   std::string val;
   SequenceNumber max_range_del_seqno = 0;
   SequenceNumber out_seqno = 0;
   bool is_blob_index = false;
   Status s;
+  std::unique_ptr<MergeContext> merge_context_guard{new MergeContext};
+  if (merge_context == nullptr) {
+    merge_context = merge_context_guard.get();
+  }
   *found_final_value = wbwi_mem->Get(
-      lkey, &val, nullptr, nullptr, &s, nullptr, &max_range_del_seqno,
+      lkey, &val, nullptr, nullptr, &s, merge_context, &max_range_del_seqno,
       &out_seqno, ReadOptions(), true, nullptr, &is_blob_index, true);
   if (s.ok()) {
     if (*found_final_value) {
@@ -3711,10 +3714,11 @@ std::string Get(const std::string& k, std::unique_ptr<WBWIMemTable>& wbwi_mem,
       return val;
     }
     return "NOT_FOUND";
+  } else if (s.IsNotFound()) {
+    EXPECT_TRUE(*found_final_value);
+    return "NOT_FOUND";
   }
-  EXPECT_TRUE(s.IsNotFound());
-  EXPECT_TRUE(*found_final_value);
-  return "NOT_FOUND";
+  return s.ToString();
 }
 
 class WBWIMemTableTest : public testing::Test {};
@@ -4056,6 +4060,177 @@ TEST_F(WBWIMemTableTest, IterEmitSingleDelete) {
   iter->~InternalIterator();
   iter_for_flush->~InternalIteratorBase();
 }
+
+void VerifyIterator(
+    InternalIterator* iter,
+    const std::vector<std::pair<std::string, std::string>>& expected) {
+  // Verify SeekToFirst and Next
+  iter->SeekToFirst();
+  auto k = expected.begin();
+  while (iter->Valid()) {
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+
+    ASSERT_EQ(iter->key(), k->first);
+    ASSERT_EQ(iter->value(), k->second);
+
+    iter->Next();
+    ++k;
+  }
+  ASSERT_OK(iter->status());
+  ASSERT_TRUE(k == expected.end());
+
+  // Verify SeekToLast and Prev
+  iter->SeekToLast();
+  k = expected.end();
+  while (iter->Valid()) {
+    --k;
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+
+    ASSERT_EQ(iter->key(), k->first);
+    ASSERT_EQ(iter->value(), k->second);
+
+    iter->Prev();
+  }
+  ASSERT_OK(iter->status());
+  ASSERT_TRUE(k == expected.begin());
+
+  // Verify Seek and SeekForPrev
+  for (auto exp = expected.begin(); exp != expected.end(); ++exp) {
+    iter->Seek(exp->first);
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), exp->first);
+    ASSERT_EQ(iter->value(), exp->second);
+
+    iter->Next();
+    if (iter->Valid()) {
+      ASSERT_OK(iter->status());
+      ++exp;
+      ASSERT_EQ(iter->key(), exp->first);
+      ASSERT_EQ(iter->value(), exp->second);
+      iter->Prev();
+      --exp;
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), exp->first);
+      ASSERT_EQ(iter->value(), exp->second);
+    } else {
+      iter->SeekToLast();
+    }
+
+    iter->SeekForPrev(exp->first);
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(iter->key(), exp->first);
+    ASSERT_EQ(iter->value(), exp->second);
+
+    iter->Prev();
+    if (iter->Valid()) {
+      ASSERT_OK(iter->status());
+      --exp;
+      ASSERT_EQ(iter->key(), exp->first);
+      ASSERT_EQ(iter->value(), exp->second);
+      iter->Next();
+      ++exp;
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), exp->first);
+      ASSERT_EQ(iter->value(), exp->second);
+    } else {
+      iter->SeekToFirst();
+    }
+  }
+}
+
+TEST_F(WBWIMemTableTest, WBWIMemTableWithMerge) {
+  const Comparator* cmp = BytewiseComparator();
+  Options opts;
+  opts.merge_operator = MergeOperators::CreateFromStringId("stringappend");
+  ImmutableOptions immutable_opts(opts);
+  MutableCFOptions mutable_cf_options(opts);
+
+  auto wbwi = std::make_shared<WriteBatchWithIndex>(
+      cmp, 0, /*overwrite_key=*/true, 0, 0);
+  wbwi->SetTrackPerCFStat(true);
+  std::unique_ptr<WBWIMemTable> wbwi_mem{
+      new WBWIMemTable(wbwi, cmp,
+                       /*cf_id=*/0, &immutable_opts, &mutable_cf_options,
+                       // stats is inaccurate but read path should still work
+                       /*stat=*/{})};
+  ASSERT_TRUE(wbwi_mem->IsEmpty());
+  constexpr SequenceNumber seqno_lb = 10;
+  constexpr SequenceNumber seqno_ub = 100;
+  constexpr WBWIMemTable::SeqnoRange assigned_seq = {seqno_lb, seqno_ub};
+  wbwi_mem->AssignSequenceNumbers(assigned_seq);
+
+  // Update then Merge
+  ASSERT_OK(wbwi->Put("a", "a1"));
+  ASSERT_OK(wbwi->Merge("a", "a2"));
+  ASSERT_OK(wbwi->Merge("a", "a3"));
+  ASSERT_OK(wbwi->Delete("b"));
+  ASSERT_OK(wbwi->Merge("b", "b1"));
+
+  // Merge then Update
+  ASSERT_OK(wbwi->Merge("c", "c1"));
+  ASSERT_OK(wbwi->Put("c", "c2"));
+  ASSERT_OK(wbwi->Merge("d", "d1"));
+  ASSERT_OK(wbwi->Merge("d", "d2"));
+  ASSERT_OK(wbwi->Delete("d"));
+
+  // Just Merge
+  ASSERT_OK(wbwi->Merge("e", "e1"));
+  ASSERT_OK(wbwi->Merge("f", "f1"));
+  ASSERT_OK(wbwi->Merge("f", "f2"));
+
+  // Just Update
+  ASSERT_OK(wbwi->SingleDelete("g"));
+
+  // key <-> val
+  // Refer to the sequence number assignment method described in the comments
+  // above the WBWIMemTable class.
+  std::vector<std::pair<std::string, std::string>> expected = {
+      {InternalKey("a", seqno_lb + 2, kTypeMerge).Encode().ToString(), "a3"},
+      {InternalKey("a", seqno_lb + 1, kTypeMerge).Encode().ToString(), "a2"},
+      {InternalKey("a", seqno_lb, kTypeValue).Encode().ToString(), "a1"},
+      {InternalKey("b", seqno_lb + 1, kTypeMerge).Encode().ToString(), "b1"},
+      {InternalKey("b", seqno_lb, kTypeDeletion).Encode().ToString(), ""},
+      {InternalKey("c", seqno_lb + 1, kTypeValue).Encode().ToString(), "c2"},
+      {InternalKey("d", seqno_lb + 2, kTypeDeletion).Encode().ToString(), ""},
+      {InternalKey("d", seqno_lb, kTypeMerge).Encode().ToString(), "d1"},
+      {InternalKey("e", seqno_lb, kTypeMerge).Encode().ToString(), "e1"},
+      {InternalKey("f", seqno_lb + 1, kTypeMerge).Encode().ToString(), "f2"},
+      {InternalKey("f", seqno_lb, kTypeMerge).Encode().ToString(), "f1"},
+      {InternalKey("g", seqno_lb, kTypeSingleDeletion).Encode().ToString(), ""},
+  };
+
+  Arena arena;
+  InternalIterator* iter = wbwi_mem->NewIterator(
+      ReadOptions(), /*seqno_to_time_mapping=*/nullptr, &arena,
+      /*prefix_extractor=*/nullptr, /*for_flush=*/false);
+  VerifyIterator(iter, expected);
+  iter->~InternalIterator();
+
+  // Test Get
+  bool found_final_value = false;
+  ASSERT_EQ("a1,a2,a3", Get("a", wbwi_mem, seqno_ub, &found_final_value));
+  ASSERT_EQ("b1", Get("b", wbwi_mem, seqno_ub, &found_final_value));
+  ASSERT_EQ("c2", Get("c", wbwi_mem, seqno_ub, &found_final_value));
+  ASSERT_EQ("NOT_FOUND", Get("d", wbwi_mem, seqno_ub, &found_final_value));
+  MergeContext merge_context;
+  ASSERT_EQ(Status::MergeInProgress().ToString(),
+            Get("e", wbwi_mem, seqno_ub, &found_final_value, &merge_context));
+  ASSERT_EQ(merge_context.GetNumOperands(), 1);
+  ASSERT_EQ(merge_context.GetOperand(0), "e1");
+  merge_context.Clear();
+  ASSERT_EQ(Status::MergeInProgress().ToString(),
+            Get("f", wbwi_mem, seqno_ub, &found_final_value, &merge_context));
+  ASSERT_EQ(merge_context.GetNumOperands(), 2);
+  ASSERT_EQ(merge_context.GetOperand(0), "f1");
+  ASSERT_EQ(merge_context.GetOperand(1), "f2");
+  ASSERT_EQ("NOT_FOUND", Get("g", wbwi_mem, seqno_ub, &found_final_value));
+}
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {