Introduce a transaction option to skip memtable write during commit (#13144)

Summary:
add a new transaction option `TransactionOptions::commit_bypass_memtable` that will ingest the transaction into a DB as an immutable memtables, skipping memtable writes during transaction commit. This helps to reduce the blocking time of committing a large transaction, which is mostly spent on memtable writes. The ingestion is done by creating WBWIMemTable using transaction's underlying WBWI, and ingest it as the latest immutable memtable. The feature will be experimental.

Major changes are:
1. write path change to ingest the transaction, mostly in WriteImpl() and IngestWBWI() in db_impl_write.cc.
2. WBWI changes to track some per CF stats like entry count and overwritten single deletion count, and track which keys have overwritten single deletions (see 3.). Per CF stat is used to precompute the number of entries in each WBWIMemTable.
3. WBWIMemTable Iterator changes to emit overwritten single deletions. The motivation is explained in the comment above class WBWIMemTable definition. The rest of the changes in WBWIMemTable are moving the iterator definition around.

Some intended follow ups:
1. support for merge operations
2. stats/logging around this option
3. tests improvement, including stress test support for the more comprehensive no_batched_op_stress.

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

Test Plan:
* added new unit tests
* enabled in multi_ops_txns_stress test
* Benchmark: applying the change in 8222c0cafc4c6eb3a0d05807f7014b44998acb7a, I tested txn size of 10k and check perf context for write_memtable_time, write_wal_time and key_lock_wait_time(repurposed for transaction unlock time). Though the benchmark result number can be flaky, this shows memtable write time improved a lot (more than 100 times). The benchmark also shows that the remaining commit latency is from transaction unlock.
```
./db_bench --benchmarks=fillrandom --seed=1727376962 --threads=1 --disable_auto_compactions=1 --max_write_buffer_number=100 --min_write_buffer_number_to_merge=100 --writes=100000 --batch_size=10000 --transaction_db=1 --perf_level=4 --enable_pipelined_write=false --commit_bypass_memtable=1

commit_bypass_memtable = false
fillrandom   :       3.982 micros/op 251119 ops/sec 0.398 seconds 100000 operations;   27.8 MB/s PERF_CONTEXT:
write_memtable_time = 116950422
write_wal_time =      8535565
txn unlock time =     32979883

commit_bypass_memtable = true
fillrandom   :       2.627 micros/op 380559 ops/sec 0.263 seconds 100000 operations;   42.1 MB/s PERF_CONTEXT:
write_memtable_time = 740784
write_wal_time =      11993119
txn unlock time =     21735685
```

Reviewed By: jowlyzhang

Differential Revision: D66307632

Pulled By: cbi42

fbshipit-source-id: 6619af58c4c537aed1f76c4a7e869fb3f5098999

db/column_family.cc

@@ -1208,7 +1208,7 @@ Status ColumnFamilyData::RangesOverlapWithMemtables(
   MergeIteratorBuilder merge_iter_builder(&internal_comparator_, &arena);
   merge_iter_builder.AddIterator(super_version->mem->NewIterator(
       read_opts, /*seqno_to_time_mapping=*/nullptr, &arena,
-      /*prefix_extractor=*/nullptr));
+      /*prefix_extractor=*/nullptr, /*for_flush=*/false));
   super_version->imm->AddIterators(read_opts, /*seqno_to_time_mapping=*/nullptr,
                                    /*prefix_extractor=*/nullptr,
                                    &merge_iter_builder,

db/column_family.h

@@ -388,10 +388,14 @@ class ColumnFamilyData {
   uint64_t GetTotalBlobFileSize() const;  // REQUIRE: DB mutex held
   // REQUIRE: DB mutex held
   void SetMemtable(MemTable* new_mem) {
-    new_mem->SetID(++last_memtable_id_);
+    AssignMemtableID(new_mem);
     mem_ = new_mem;
   }
 
+  void AssignMemtableID(ReadOnlyMemTable* new_imm) {
+    new_imm->SetID(++last_memtable_id_);
+  }
+
   // calculate the oldest log needed for the durability of this column family
   uint64_t OldestLogToKeep();
 

db/db_impl/db_impl.cc

@@ -2098,7 +2098,8 @@ InternalIterator* DBImpl::NewInternalIterator(
   // Collect iterator for mutable memtable
   auto mem_iter = super_version->mem->NewIterator(
       read_options, super_version->GetSeqnoToTimeMapping(), arena,
-      super_version->mutable_cf_options.prefix_extractor.get());
+      super_version->mutable_cf_options.prefix_extractor.get(),
+      /*for_flush=*/false);
   Status s;
   if (!read_options.ignore_range_deletions) {
     std::unique_ptr<TruncatedRangeDelIterator> mem_tombstone_iter;

db/db_impl/db_impl.h

@@ -48,6 +48,7 @@
 #include "db/write_controller.h"
 #include "db/write_thread.h"
 #include "logging/event_logger.h"
+#include "memtable/wbwi_memtable.h"
 #include "monitoring/instrumented_mutex.h"
 #include "options/db_options.h"
 #include "port/port.h"
@@ -60,6 +61,7 @@
 #include "rocksdb/transaction_log.h"
 #include "rocksdb/user_write_callback.h"
 #include "rocksdb/utilities/replayer.h"
+#include "rocksdb/utilities/write_batch_with_index.h"
 #include "rocksdb/write_buffer_manager.h"
 #include "table/merging_iterator.h"
 #include "util/autovector.h"
@@ -1507,6 +1509,23 @@ class DBImpl : public DB {
 
   void EraseThreadStatusDbInfo() const;
 
+  // For CFs that has updates in `wbwi`, their memtable will be switched,
+  // and `wbwi` will be added as the latest immutable memtable.
+  //
+  // REQUIRES: this thread is currently at the front of the main writer queue.
+  // @param prep_log refers to the WAL that contains prepare record
+  // for the transaction based on wbwi.
+  // @param assigned_seqno Sequence numbers for the ingested memtable.
+  // @param last_seqno the value of versions_->LastSequence() after the write
+  // ingests `wbwi` is done.
+  // @param memtable_updated Whether the same write that ingests wbwi has
+  // updated memtable. This is useful for determining whether to set bg
+  // error when IngestWBWI fails.
+  Status IngestWBWI(std::shared_ptr<WriteBatchWithIndex> wbwi,
+                    const WBWIMemTable::SeqnoRange& assigned_seqno,
+                    uint64_t min_prep_log, SequenceNumber last_seqno,
+                    bool memtable_updated, bool ignore_missing_cf);
+
   // If disable_memtable is set the application logic must guarantee that the
   // batch will still be skipped from memtable during the recovery. An excption
   // to this is seq_per_batch_ mode, in which since each batch already takes one
@@ -1522,6 +1541,16 @@ class DBImpl : public DB {
   // batch_cnt is expected to be non-zero in seq_per_batch mode and
   // indicates the number of sub-patches. A sub-patch is a subset of the write
   // batch that does not have duplicate keys.
+  // `callback` is called before WAL write.
+  // See more in comment above WriteCallback::Callback().
+  // pre_release_callback is called after WAL write and before memtable write.
+  // See more in comment above PreReleaseCallback::Callback().
+  // post_memtable_callback is called after memtable write but before publishing
+  // the sequence number to readers.
+  //
+  // The main write queue. This is the only write queue that updates
+  // LastSequence. When using one write queue, the same sequence also indicates
+  // the last published sequence.
   Status WriteImpl(const WriteOptions& options, WriteBatch* updates,
                    WriteCallback* callback = nullptr,
                    UserWriteCallback* user_write_cb = nullptr,
@@ -1529,7 +1558,9 @@ class DBImpl : public DB {
                    bool disable_memtable = false, uint64_t* seq_used = nullptr,
                    size_t batch_cnt = 0,
                    PreReleaseCallback* pre_release_callback = nullptr,
-                   PostMemTableCallback* post_memtable_callback = nullptr);
+                   PostMemTableCallback* post_memtable_callback = nullptr,
+                   std::shared_ptr<WriteBatchWithIndex> wbwi = nullptr,
+                   uint64_t min_prep_log = 0);
 
   Status PipelinedWriteImpl(const WriteOptions& options, WriteBatch* updates,
                             WriteCallback* callback = nullptr,
@@ -2053,7 +2084,19 @@ class DBImpl : public DB {
 
   // Switches the current live memtable to immutable/read-only memtable.
   // A new WAL is created if the current WAL is not empty.
-  Status SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context);
+  // If `new_imm` is not nullptr, it will be added as the newest immutable
+  // memtable, if and only if OK status is returned.
+  // `last_seqno` needs to be provided if `new_imm` is not nullptr. It is
+  // the value of versions_->LastSequence() after the write that ingests new_imm
+  // is done.
+  //
+  // REQUIRES: mutex_ is held
+  // REQUIRES: this thread is currently at the front of the writer queue
+  // REQUIRES: this thread is currently at the front of the 2nd writer queue if
+  // two_write_queues_ is true (This is to simplify the reasoning.)
+  Status SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context,
+                        ReadOnlyMemTable* new_imm = nullptr,
+                        SequenceNumber last_seqno = 0);
 
   // Select and output column families qualified for atomic flush in
   // `selected_cfds`. If `provided_candidate_cfds` is non-empty, it will be used

db/db_impl/db_impl_open.cc

@@ -1373,6 +1373,9 @@ Status DBImpl::RecoverLogFiles(const std::vector<uint64_t>& wal_numbers,
         }
       }
     }
+    ROCKS_LOG_INFO(immutable_db_options_.info_log,
+                   "Recovered to log #%" PRIu64 " seq #%" PRIu64, wal_number,
+                   *next_sequence);
 
     if (!status.ok() || old_log_record) {
       if (status.IsNotSupported()) {
@@ -1405,10 +1408,6 @@ Status DBImpl::RecoverLogFiles(const std::vector<uint64_t>& wal_numbers,
         if (corrupted_wal_found != nullptr) {
           *corrupted_wal_found = true;
         }
-        ROCKS_LOG_INFO(immutable_db_options_.info_log,
-                       "Point in time recovered to log #%" PRIu64
-                       " seq #%" PRIu64,
-                       wal_number, *next_sequence);
       } else {
         assert(immutable_db_options_.wal_recovery_mode ==
                    WALRecoveryMode::kTolerateCorruptedTailRecords ||
@@ -1681,7 +1680,8 @@ Status DBImpl::WriteLevel0TableForRecovery(int job_id, ColumnFamilyData* cfd,
                   ro, /*seqno_to_time_mapping=*/nullptr, &arena,
                   /*prefix_extractor=*/nullptr, ts_sz)
             : mem->NewIterator(ro, /*seqno_to_time_mapping=*/nullptr, &arena,
-                               /*prefix_extractor=*/nullptr));
+                               /*prefix_extractor=*/nullptr,
+                               /*for_flush=*/true));
     ROCKS_LOG_DEBUG(immutable_db_options_.info_log,
                     "[%s] [WriteLevel0TableForRecovery]"
                     " Level-0 table #%" PRIu64 ": started",

db/db_impl/db_impl_write.cc

@@ -12,6 +12,7 @@
 #include "db/error_handler.h"
 #include "db/event_helpers.h"
 #include "logging/logging.h"
+#include "memtable/wbwi_memtable.h"
 #include "monitoring/perf_context_imp.h"
 #include "options/options_helper.h"
 #include "test_util/sync_point.h"
@@ -189,16 +190,137 @@ Status DBImpl::WriteWithCallback(const WriteOptions& write_options,
   return s;
 }
 
-// The main write queue. This is the only write queue that updates LastSequence.
-// When using one write queue, the same sequence also indicates the last
-// published sequence.
+Status DBImpl::IngestWBWI(std::shared_ptr<WriteBatchWithIndex> wbwi,
+                          const WBWIMemTable::SeqnoRange& assigned_seqno,
+                          uint64_t prep_log,
+                          SequenceNumber last_seqno_after_ingest,
+                          bool memtable_updated, bool ignore_missing_cf) {
+  // Keys in new memtable have seqno > last_seqno_after_ingest >= keys in wbwi.
+  assert(assigned_seqno.upper_bound <= last_seqno_after_ingest);
+  // Keys in the current memtable have seqno <= LastSequence() < keys in wbwi.
+  assert(assigned_seqno.lower_bound > versions_->LastSequence());
+  autovector<ReadOnlyMemTable*> memtables;
+  autovector<ColumnFamilyData*> cfds;
+  InstrumentedMutexLock lock(&mutex_);
+  ColumnFamilySet* cf_set = versions_->GetColumnFamilySet();
+
+  // Create WBWIMemTables
+  for (const auto [cf_id, stat] : wbwi->GetCFStats()) {
+    ColumnFamilyData* cfd = cf_set->GetColumnFamily(cf_id);
+    if (!cfd) {
+      if (ignore_missing_cf) {
+        continue;
+      }
+      for (auto mem : memtables) {
+        mem->Unref();
+        delete mem;
+      }
+      for (auto cfd_ptr : cfds) {
+        cfd_ptr->UnrefAndTryDelete();
+      }
+      Status s = Status::InvalidArgument(
+          "Invalid column family id from WriteBatchWithIndex: " +
+          std::to_string(cf_id));
+      if (memtable_updated) {
+        s = Status::Corruption(
+            "Part of the write batch is applied. Memtable is in a inconsistent "
+            "state. " +
+            s.ToString());
+        error_handler_.SetBGError(s, BackgroundErrorReason::kMemTable);
+      }
+
+      return s;
+    }
+    WBWIMemTable* wbwi_memtable =
+        new WBWIMemTable(wbwi, cfd->user_comparator(), cf_id, cfd->ioptions(),
+                         cfd->GetLatestMutableCFOptions(), stat);
+    wbwi_memtable->Ref();
+    wbwi_memtable->AssignSequenceNumbers(assigned_seqno);
+    // This is needed to keep the WAL that contains Prepare alive until
+    // committed data in this memtable is persisted.
+    wbwi_memtable->SetMinPrepLog(prep_log);
+    memtables.push_back(wbwi_memtable);
+    cfd->Ref();
+    cfds.push_back(cfd);
+  }
+
+  // Stop writes to the DB by entering both write threads
+  WriteThread::Writer nonmem_w;
+  if (two_write_queues_) {
+    nonmem_write_thread_.EnterUnbatched(&nonmem_w, &mutex_);
+  }
+  WaitForPendingWrites();
+
+  // Switch memtable and add WBWIMemTables
+  Status s;
+  for (size_t i = 0; i < memtables.size(); ++i) {
+    assert(!immutable_db_options_.atomic_flush);
+    // NOTE: to support atomic flush, need to call
+    // SelectColumnFamiliesForAtomicFlush()
+    WriteContext write_context;
+    // TODO: not switch on empty memtable, may need to update metadata
+    //   like NextLogNumber(), earliest_seqno and memtable id.
+    s = SwitchMemtable(cfds[i], &write_context, memtables[i],
+                       last_seqno_after_ingest);
+    if (!s.ok()) {
+      // SwitchMemtable() can only fail if a new WAL is to be created, this
+      // should only happen for the first call to SwitchMemtable(). log will
+      // be empty and no new WAL is created for the rest of the calls.
+      assert(i == 0);
+      if (i != 0 || memtable_updated) {
+        // escalate error to non-recoverable
+        s = Status::Corruption(
+            "Part of the write batch is applied. Memtable is in a inconsistent "
+            "state. " +
+            s.ToString());
+        error_handler_.SetBGError(s, BackgroundErrorReason::kMemTable);
+      } else {
+        // SwitchMemtable() already sets appropriate bg error
+      }
+      for (size_t j = i; j < memtables.size(); j++) {
+        memtables[j]->Unref();
+        delete memtables[j];
+      }
+      break;
+    }
+  }
+  for (size_t i = 0; i < cfds.size(); ++i) {
+    if (cfds[i]->UnrefAndTryDelete()) {
+      cfds[i] = nullptr;
+    }
+  }
+
+  // exit the second queue before returning
+  if (two_write_queues_) {
+    nonmem_write_thread_.ExitUnbatched(&nonmem_w);
+  }
+  if (s.ok()) {
+    // Trigger flushes for the new immutable memtables.
+    for (const auto cfd : cfds) {
+      if (cfd == nullptr) {
+        continue;
+      }
+      cfd->imm()->FlushRequested();
+      FlushRequest flush_req;
+      // TODO: a new flush reason for ingesting memtable
+      GenerateFlushRequest({cfd}, FlushReason::kExternalFileIngestion,
+                           &flush_req);
+      EnqueuePendingFlush(flush_req);
+    }
+    MaybeScheduleFlushOrCompaction();
+  }
+  return s;
+}
+
 Status DBImpl::WriteImpl(const WriteOptions& write_options,
                          WriteBatch* my_batch, WriteCallback* callback,
                          UserWriteCallback* user_write_cb, uint64_t* log_used,
                          uint64_t log_ref, bool disable_memtable,
                          uint64_t* seq_used, size_t batch_cnt,
                          PreReleaseCallback* pre_release_callback,
-                         PostMemTableCallback* post_memtable_callback) {
+                         PostMemTableCallback* post_memtable_callback,
+                         std::shared_ptr<WriteBatchWithIndex> wbwi,
+                         uint64_t prep_log) {
   assert(!seq_per_batch_ || batch_cnt != 0);
   assert(my_batch == nullptr || my_batch->Count() == 0 ||
          write_options.protection_bytes_per_key == 0 ||
@@ -287,6 +409,23 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
     return Status::NotSupported(
         "DeleteRange is not compatible with row cache.");
   }
+  if (wbwi) {
+    assert(prep_log > 0);
+    // Used only in WriteCommittedTxn::CommitInternal() with no `callback`.
+    assert(!callback);
+    if (immutable_db_options_.unordered_write) {
+      return Status::NotSupported(
+          "Ingesting WriteBatch does not support unordered_write");
+    }
+    if (immutable_db_options_.enable_pipelined_write) {
+      return Status::NotSupported(
+          "Ingesting WriteBatch does not support pipelined_write");
+    }
+    if (immutable_db_options_.atomic_flush) {
+      return Status::NotSupported(
+          "Ingesting WriteBatch does not support atomic_flush");
+    }
+  }
   // Otherwise IsLatestPersistentState optimization does not make sense
   assert(!WriteBatchInternal::IsLatestPersistentState(my_batch) ||
          disable_memtable);
@@ -344,7 +483,8 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
   PERF_TIMER_GUARD(write_pre_and_post_process_time);
   WriteThread::Writer w(write_options, my_batch, callback, user_write_cb,
                         log_ref, disable_memtable, batch_cnt,
-                        pre_release_callback, post_memtable_callback);
+                        pre_release_callback, post_memtable_callback,
+                        /*_ingest_wbwi=*/wbwi != nullptr);
   StopWatch write_sw(immutable_db_options_.clock, stats_, DB_WRITE);
 
   write_thread_.JoinBatchGroup(&w);
@@ -441,6 +581,9 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
   TEST_SYNC_POINT("DBImpl::WriteImpl:BeforeLeaderEnters");
   last_batch_group_size_ =
       write_thread_.EnterAsBatchGroupLeader(&w, &write_group);
+  if (wbwi) {
+    assert(write_group.size == 1);
+  }
 
   IOStatus io_s;
   Status pre_release_cb_status;
@@ -494,10 +637,25 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
     // Note about seq_per_batch_: either disableWAL is set for the entire write
     // group or not. In either case we inc seq for each write batch with no
     // failed callback. This means that there could be a batch with
-    // disalbe_memtable in between; although we do not write this batch to
+    // 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;
+    if (wbwi) {
+      // Reserve sequence numbers for the ingested memtable. We need to reserve
+      // at lease this amount for recovery. During recovery,
+      // transactions do not commit by ingesting WBWI. The sequence number
+      // associated with the commit entry in WAL is used as the starting
+      // sequence number for inserting into memtable. We need to reserve
+      // enough sequence numbers here (at least the number of operations
+      // in write batch) to assign to memtable entries for this transaction.
+      // This prevents updates in different transactions from using out-of-order
+      // sequence numbers or the same key+seqno.
+      //
+      // WBWI ingestion requires not grouping writes, so we don't need to
+      // consider incrementing sequence number for WBWI from other writers.
+      seq_inc += wbwi->GetWriteBatch()->Count();
+    }
 
     const bool concurrent_update = two_write_queues_;
     // Update stats while we are an exclusive group leader, so we know
@@ -674,6 +832,27 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
     // handle exit, false means somebody else did
     should_exit_batch_group = write_thread_.CompleteParallelMemTableWriter(&w);
   }
+  if (wbwi) {
+    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);
+      if (two_write_queues_) {
+        assert(ub <= versions_->LastAllocatedSequence());
+      }
+      status = IngestWBWI(wbwi, {/*lower_bound=*/lb, /*upper_bound=*/ub},
+                          prep_log, last_sequence,
+                          /*memtable_updated=*/memtable_update_count > 0,
+                          write_options.ignore_missing_column_families);
+    }
+  }
+
   if (should_exit_batch_group) {
     if (status.ok()) {
       for (auto* tmp_w : write_group) {
@@ -2204,16 +2383,13 @@ void DBImpl::NotifyOnMemTableSealed(ColumnFamilyData* /*cfd*/,
   mutex_.Lock();
 }
 
-// REQUIRES: mutex_ is held
-// REQUIRES: this thread is currently at the front of the writer queue
-// REQUIRES: this thread is currently at the front of the 2nd writer queue if
-// two_write_queues_ is true (This is to simplify the reasoning.)
-Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) {
+Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context,
+                              ReadOnlyMemTable* new_imm,
+                              SequenceNumber last_seqno) {
   mutex_.AssertHeld();
   assert(lock_wal_count_ == 0);
 
   // TODO: plumb Env::IOActivity, Env::IOPriority
-  const ReadOptions read_options;
   const WriteOptions write_options;
 
   log::Writer* new_log = nullptr;
@@ -2249,6 +2425,7 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) {
   const MutableCFOptions mutable_cf_options = *cfd->GetLatestMutableCFOptions();
 
   // Set memtable_info for memtable sealed callback
+  // TODO: memtable_info for `new_imm`
   MemTableInfo memtable_info;
   memtable_info.cf_name = cfd->GetName();
   memtable_info.first_seqno = cfd->mem()->GetFirstSequenceNumber();
@@ -2276,8 +2453,20 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) {
     }
   }
   if (s.ok()) {
-    SequenceNumber seq = versions_->LastSequence();
-    new_mem = cfd->ConstructNewMemtable(mutable_cf_options, seq);
+    // FIXME: from the comment for GetEarliestSequenceNumber(), any key with
+    //  seqno >= earliest_seqno should be in this or later memtable. This means
+    //  we should use LastSequence() + 1 or last_seqno + 1 here. And it needs to
+    //  be incremented with file ingestion and other operations that consumes
+    //  sequence number.
+    SequenceNumber seq;
+    if (new_imm) {
+      assert(last_seqno > versions_->LastSequence());
+      seq = last_seqno;
+    } else {
+      seq = versions_->LastSequence();
+    }
+    new_mem =
+        cfd->ConstructNewMemtable(mutable_cf_options, /*earliest_seq=*/seq);
     context->superversion_context.NewSuperVersion();
 
     ROCKS_LOG_INFO(immutable_db_options_.info_log,
@@ -2359,6 +2548,8 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) {
         versions_->PreComputeMinLogNumberWithUnflushedData(logfile_number_);
     if (min_wal_number_to_keep >
         versions_->GetWalSet().GetMinWalNumberToKeep()) {
+      // TODO: plumb Env::IOActivity, Env::IOPriority
+      const ReadOptions read_options;
       // Get a snapshot of the empty column families.
       // LogAndApply may release and reacquire db
       // mutex, during that period, column family may become empty (e.g. its
@@ -2416,6 +2607,18 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) {
   cfd->mem()->SetNextLogNumber(logfile_number_);
   assert(new_mem != nullptr);
   cfd->imm()->Add(cfd->mem(), &context->memtables_to_free_);
+  if (new_imm) {
+    // Need to assign memtable id here before SetMemtable() below assigns id to
+    // the new live memtable
+    cfd->AssignMemtableID(new_imm);
+    // NOTE: new_imm and cfd->mem() references the same WAL and has the same
+    // NextLogNumber(). They should be flushed together. For non-atomic-flush,
+    // we always try to flush all immutable memtable. For atomic flush, these
+    // two memtables will be marked eligible for flush in the same call to
+    // AssignAtomicFlushSeq().
+    new_imm->SetNextLogNumber(logfile_number_);
+    cfd->imm()->Add(new_imm, &context->memtables_to_free_);
+  }
   new_mem->Ref();
   cfd->SetMemtable(new_mem);
   InstallSuperVersionAndScheduleWork(cfd, &context->superversion_context,
@@ -2428,6 +2631,9 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) {
   // that is okay.  If we did, it most likely means that s was already an error.
   // In any case, ignore any unchecked error for i_os here.
   io_s.PermitUncheckedError();
+  // We guarantee that if a non-ok status is returned, `new_imm` was not added
+  // to the db.
+  assert(s.ok());
   return s;
 }
 

db/db_test_util.cc

@@ -1600,42 +1600,74 @@ std::vector<std::uint64_t> DBTestBase::ListTableFiles(Env* env,
   return file_numbers;
 }
 
-void DBTestBase::VerifyDBFromMap(std::map<std::string, std::string> true_data,
-                                 size_t* total_reads_res, bool tailing_iter,
-                                 std::map<std::string, Status> status) {
-  size_t total_reads = 0;
+void DBTestBase::VerifyDBFromMap(
+    std::map<std::string, std::string> true_data, size_t* total_reads_res,
+    bool tailing_iter, ReadOptions* ro, ColumnFamilyHandle* cf,
+    std::unordered_set<std::string>* not_found) const {
+  ReadOptions temp_ro;
+  if (!ro) {
+    ro = &temp_ro;
+    ro->verify_checksums = true;
+  }
+  if (!cf) {
+    cf = db_->DefaultColumnFamily();
+  }
 
-  for (auto& kv : true_data) {
-    Status s = status[kv.first];
-    if (s.ok()) {
-      ASSERT_EQ(Get(kv.first), kv.second);
-    } else {
-      std::string value;
-      ASSERT_EQ(s, db_->Get(ReadOptions(), kv.first, &value));
-    }
+  // Get
+  size_t total_reads = 0;
+  std::string result;
+  for (auto& [k, v] : true_data) {
+    ASSERT_OK(db_->Get(*ro, cf, k, &result)) << "key is " << k;
+    ASSERT_EQ(v, result);
     total_reads++;
   }
+  if (not_found) {
+    for (const auto& k : *not_found) {
+      ASSERT_TRUE(db_->Get(*ro, cf, k, &result).IsNotFound())
+          << "key is " << k << " val is " << result;
+    }
+  }
+
+  // MultiGet
+  std::vector<Slice> key_slice;
+  for (const auto& [k, _] : true_data) {
+    key_slice.emplace_back(k);
+  }
+  std::vector<std::string> values;
+  std::vector<ColumnFamilyHandle*> cfs(key_slice.size(), cf);
+  std::vector<Status> status = db_->MultiGet(*ro, cfs, key_slice, &values);
+  total_reads += key_slice.size();
+  auto data_iter = true_data.begin();
+  for (size_t i = 0; i < key_slice.size(); ++i, ++data_iter) {
+    ASSERT_OK(status[i]);
+    ASSERT_EQ(values[i], data_iter->second);
+  }
+  // MultiGet - not found
+  if (not_found) {
+    key_slice.clear();
+    for (const auto& k : *not_found) {
+      key_slice.emplace_back(k);
+    }
+    cfs = std::vector<ColumnFamilyHandle*>(key_slice.size(), cf);
+    values.clear();
+    status = db_->MultiGet(*ro, cfs, key_slice, &values);
+    for (const auto& s : status) {
+      ASSERT_TRUE(s.IsNotFound());
+    }
+  }
 
   // Normal Iterator
   {
     int iter_cnt = 0;
-    ReadOptions ro;
-    ro.total_order_seek = true;
-    Iterator* iter = db_->NewIterator(ro);
+    ReadOptions ro_ = *ro;
+    ro_.total_order_seek = true;
+    Iterator* iter = db_->NewIterator(ro_, cf);
     // Verify Iterator::Next()
     iter_cnt = 0;
-    auto data_iter = true_data.begin();
-    Status s;
-    for (iter->SeekToFirst(); iter->Valid(); iter->Next(), data_iter++) {
+    data_iter = true_data.begin();
+    for (iter->SeekToFirst(); iter->Valid(); iter->Next(), ++data_iter) {
       ASSERT_EQ(iter->key().ToString(), data_iter->first);
-      Status current_status = status[data_iter->first];
-      if (!current_status.ok()) {
-        s = current_status;
-      }
-      ASSERT_EQ(iter->status(), s);
-      if (current_status.ok()) {
         ASSERT_EQ(iter->value().ToString(), data_iter->second);
-      }
       iter_cnt++;
       total_reads++;
     }
@@ -1646,20 +1678,12 @@ void DBTestBase::VerifyDBFromMap(std::map<std::string, std::string> true_data,
 
     // Verify Iterator::Prev()
     // Use a new iterator to make sure its status is clean.
-    iter = db_->NewIterator(ro);
+    iter = db_->NewIterator(ro_, cf);
     iter_cnt = 0;
-    s = Status::OK();
     auto data_rev = true_data.rbegin();
     for (iter->SeekToLast(); iter->Valid(); iter->Prev(), data_rev++) {
       ASSERT_EQ(iter->key().ToString(), data_rev->first);
-      Status current_status = status[data_rev->first];
-      if (!current_status.ok()) {
-        s = current_status;
-      }
-      ASSERT_EQ(iter->status(), s);
-      if (current_status.ok()) {
         ASSERT_EQ(iter->value().ToString(), data_rev->second);
-      }
       iter_cnt++;
       total_reads++;
     }
@@ -1667,12 +1691,20 @@ void DBTestBase::VerifyDBFromMap(std::map<std::string, std::string> true_data,
     ASSERT_EQ(data_rev, true_data.rend())
         << iter_cnt << " / " << true_data.size();
 
-    // Verify Iterator::Seek()
-    for (const auto& kv : true_data) {
-      iter->Seek(kv.first);
-      ASSERT_EQ(kv.first, iter->key().ToString());
-      ASSERT_EQ(kv.second, iter->value().ToString());
-      total_reads++;
+    // Verify Iterator::Seek() and SeekForPrev()
+    for (const auto& [k, v] : true_data) {
+      for (bool prev : {false, true}) {
+        if (prev) {
+          iter->SeekForPrev(k);
+        } else {
+          iter->Seek(k);
+        }
+        ASSERT_TRUE(iter->Valid());
+        ASSERT_OK(iter->status());
+        ASSERT_EQ(iter->key(), k);
+        ASSERT_EQ(iter->value(), v);
+        ++total_reads;
+      }
     }
     delete iter;
   }
@@ -1680,14 +1712,14 @@ void DBTestBase::VerifyDBFromMap(std::map<std::string, std::string> true_data,
   if (tailing_iter) {
     // Tailing iterator
     int iter_cnt = 0;
-    ReadOptions ro;
-    ro.tailing = true;
-    ro.total_order_seek = true;
-    Iterator* iter = db_->NewIterator(ro);
+    ReadOptions ro_ = *ro;
+    ro_.tailing = true;
+    ro_.total_order_seek = true;
+    Iterator* iter = db_->NewIterator(ro_, cf);
 
     // Verify ForwardIterator::Next()
     iter_cnt = 0;
-    auto data_iter = true_data.begin();
+    data_iter = true_data.begin();
     for (iter->SeekToFirst(); iter->Valid(); iter->Next(), data_iter++) {
       ASSERT_EQ(iter->key().ToString(), data_iter->first);
       ASSERT_EQ(iter->value().ToString(), data_iter->second);

db/db_test_util.h

@@ -1367,7 +1367,8 @@ class DBTestBase : public testing::Test {
   void VerifyDBFromMap(
       std::map<std::string, std::string> true_data,
       size_t* total_reads_res = nullptr, bool tailing_iter = false,
-      std::map<std::string, Status> status = std::map<std::string, Status>());
+      ReadOptions* ro = nullptr, ColumnFamilyHandle* cf = nullptr,
+      std::unordered_set<std::string>* not_found = nullptr) const;
 
   void VerifyDBInternal(
       std::vector<std::pair<std::string, std::string>> true_data);

db/flush_job.cc

@@ -422,7 +422,8 @@ Status FlushJob::MemPurge() {
       range_del_iters;
   for (ReadOnlyMemTable* m : mems_) {
     memtables.push_back(m->NewIterator(ro, /*seqno_to_time_mapping=*/nullptr,
-                                       &arena, /*prefix_extractor=*/nullptr));
+                                       &arena, /*prefix_extractor=*/nullptr,
+                                       /*for_flush=*/true));
     auto* range_del_iter = m->NewRangeTombstoneIterator(
         ro, kMaxSequenceNumber, true /* immutable_memtable */);
     if (range_del_iter != nullptr) {
@@ -624,10 +625,13 @@ Status FlushJob::MemPurge() {
         // Construct fragmented memtable range tombstones without mutex
         new_mem->ConstructFragmentedRangeTombstones();
         db_mutex_->Lock();
-        uint64_t new_mem_id = mems_[0]->GetID();
+        // Take the newest id, so that memtables in MemtableList don't have
+        // out-of-order memtable ids.
+        uint64_t new_mem_id = mems_.back()->GetID();
 
         new_mem->SetID(new_mem_id);
-        new_mem->SetNextLogNumber(mems_[0]->GetNextLogNumber());
+        // Take the latest memtable's next log number.
+        new_mem->SetNextLogNumber(mems_.back()->GetNextLogNumber());
 
         // This addition will not trigger another flush, because
         // we do not call EnqueuePendingFlush().
@@ -907,7 +911,7 @@ Status FlushJob::WriteLevel0Table() {
       } else {
         memtables.push_back(
             m->NewIterator(ro, /*seqno_to_time_mapping=*/nullptr, &arena,
-                           /*prefix_extractor=*/nullptr));
+                           /*prefix_extractor=*/nullptr, /*for_flush=*/true));
       }
       auto* range_del_iter =
           logical_strip_timestamp

db/forward_iterator.cc

@@ -714,7 +714,8 @@ void ForwardIterator::RebuildIterators(bool refresh_sv) {
       sv_->GetSeqnoToTimeMapping();
   mutable_iter_ =
       sv_->mem->NewIterator(read_options_, seqno_to_time_mapping, &arena_,
-                            sv_->mutable_cf_options.prefix_extractor.get());
+                            sv_->mutable_cf_options.prefix_extractor.get(),
+                            /*for_flush=*/false);
   sv_->imm->AddIterators(read_options_, seqno_to_time_mapping,
                          sv_->mutable_cf_options.prefix_extractor.get(),
                          &imm_iters_, &arena_);
@@ -784,7 +785,8 @@ void ForwardIterator::RenewIterators() {
       svnew->GetSeqnoToTimeMapping();
   mutable_iter_ =
       svnew->mem->NewIterator(read_options_, seqno_to_time_mapping, &arena_,
-                              svnew->mutable_cf_options.prefix_extractor.get());
+                              svnew->mutable_cf_options.prefix_extractor.get(),
+                              /*for_flush=*/false);
   svnew->imm->AddIterators(read_options_, seqno_to_time_mapping,
                            svnew->mutable_cf_options.prefix_extractor.get(),
                            &imm_iters_, &arena_);

db/memtable.cc

@@ -599,7 +599,7 @@ class MemTableIterator : public InternalIterator {
 InternalIterator* MemTable::NewIterator(
     const ReadOptions& read_options,
     UnownedPtr<const SeqnoToTimeMapping> seqno_to_time_mapping, Arena* arena,
-    const SliceTransform* prefix_extractor) {
+    const SliceTransform* prefix_extractor, bool /*for_flush*/) {
   assert(arena != nullptr);
   auto mem = arena->AllocateAligned(sizeof(MemTableIterator));
   return new (mem)

db/memtable.h

@@ -153,7 +153,7 @@ class ReadOnlyMemTable {
   virtual InternalIterator* NewIterator(
       const ReadOptions& read_options,
       UnownedPtr<const SeqnoToTimeMapping> seqno_to_time_mapping, Arena* arena,
-      const SliceTransform* prefix_extractor) = 0;
+      const SliceTransform* prefix_extractor, bool for_flush) = 0;
 
   // Returns an iterator that wraps a MemTableIterator and logically strips the
   // user-defined timestamp of each key. This API is only used by flush when
@@ -541,7 +541,7 @@ class MemTable final : public ReadOnlyMemTable {
   InternalIterator* NewIterator(
       const ReadOptions& read_options,
       UnownedPtr<const SeqnoToTimeMapping> seqno_to_time_mapping, Arena* arena,
-      const SliceTransform* prefix_extractor) override;
+      const SliceTransform* prefix_extractor, bool for_flush) override;
 
   InternalIterator* NewTimestampStrippingIterator(
       const ReadOptions& read_options,

db/memtable_list.cc

@@ -32,6 +32,10 @@ class Mutex;
 class VersionSet;
 
 void MemTableListVersion::AddMemTable(ReadOnlyMemTable* m) {
+  if (!memlist_.empty()) {
+    // ID can be equal for MemPurge
+    assert(m->GetID() >= memlist_.front()->GetID());
+  }
   memlist_.push_front(m);
   *parent_memtable_list_memory_usage_ += m->ApproximateMemoryUsage();
 }
@@ -216,7 +220,8 @@ void MemTableListVersion::AddIterators(
     std::vector<InternalIterator*>* iterator_list, Arena* arena) {
   for (auto& m : memlist_) {
     iterator_list->push_back(m->NewIterator(options, seqno_to_time_mapping,
-                                            arena, prefix_extractor));
+                                            arena, prefix_extractor,
+                                            /*for_flush=*/false));
   }
 }
 
@@ -228,7 +233,8 @@ void MemTableListVersion::AddIterators(
   for (auto& m : memlist_) {
     auto mem_iter =
         m->NewIterator(options, seqno_to_time_mapping,
-                       merge_iter_builder->GetArena(), prefix_extractor);
+                       merge_iter_builder->GetArena(), prefix_extractor,
+                       /*for_flush=*/false);
     if (!add_range_tombstone_iter || options.ignore_range_deletions) {
       merge_iter_builder->AddIterator(mem_iter);
     } else {
@@ -410,7 +416,8 @@ void MemTableList::PickMemtablesToFlush(uint64_t max_memtable_id,
   // ret is filled with memtables already sorted in increasing MemTable ID.
   // However, when the mempurge feature is activated, new memtables with older
   // IDs will be added to the memlist.
-  for (auto it = memlist.rbegin(); it != memlist.rend(); ++it) {
+  auto it = memlist.rbegin();
+  for (; it != memlist.rend(); ++it) {
     ReadOnlyMemTable* m = *it;
     if (!atomic_flush && m->atomic_flush_seqno_ != kMaxSequenceNumber) {
       atomic_flush = true;
@@ -439,6 +446,12 @@ void MemTableList::PickMemtablesToFlush(uint64_t max_memtable_id,
       break;
     }
   }
+  if (!ret->empty() && it != memlist.rend()) {
+    // checks that the first memtable not picked to flush is not ingested wbwi.
+    // Ingested memtable should be flushed together with the memtable before it
+    // since they map to the same WAL and have the same NextLogNumber().
+    assert(strcmp((*it)->Name(), "WBWIMemTable") != 0);
+  }
   if (!atomic_flush || num_flush_not_started_ == 0) {
     flush_requested_ = false;  // start-flush request is complete
   }

db/memtable_list_test.cc

@@ -330,6 +330,7 @@ TEST_F(MemTableListTest, GetTest) {
   // This is to make assert(memtable->IsFragmentedRangeTombstonesConstructed())
   // in MemTableListVersion::GetFromList work.
   mem->ConstructFragmentedRangeTombstones();
+  mem->SetID(1);
   list.Add(mem, &to_delete);
 
   SequenceNumber saved_seq = seq;
@@ -338,6 +339,7 @@ TEST_F(MemTableListTest, GetTest) {
   WriteBufferManager wb2(options.db_write_buffer_size);
   MemTable* mem2 = new MemTable(cmp, ioptions, MutableCFOptions(options), &wb2,
                                 kMaxSequenceNumber, 0 /* column_family_id */);
+  mem2->SetID(2);
   mem2->Ref();
 
   ASSERT_OK(

db/repair.cc

@@ -449,7 +449,7 @@ class Repairer {
       Arena arena;
       ScopedArenaPtr<InternalIterator> iter(
           mem->NewIterator(ro, /*seqno_to_time_mapping=*/nullptr, &arena,
-                           /*prefix_extractor=*/nullptr));
+                           /*prefix_extractor=*/nullptr, /*for_flush=*/true));
       int64_t _current_time = 0;
       immutable_db_options_.clock->GetCurrentTime(&_current_time)
           .PermitUncheckedError();  // ignore error

db/write_batch_test.cc

@@ -59,7 +59,8 @@ static std::string PrintContents(WriteBatch* b,
     InternalIterator* iter;
     if (i == 0) {
       iter = mem->NewIterator(ReadOptions(), /*seqno_to_time_mapping=*/nullptr,
-                              &arena, /*prefix_extractor=*/nullptr);
+                              &arena, /*prefix_extractor=*/nullptr,
+                              /*for_flush=*/false);
       arena_iter_guard.reset(iter);
     } else {
       iter = mem->NewRangeTombstoneIterator(ReadOptions(),

db/write_thread.cc

@@ -523,8 +523,10 @@ size_t WriteThread::EnterAsBatchGroupLeader(Writer* leader,
         // those are something else. They want to be alone
         (w->callback != nullptr && !w->callback->AllowWriteBatching()) ||
         // dont batch writes that don't want to be batched
-        (size + WriteBatchInternal::ByteSize(w->batch) > max_size)
+        (size + WriteBatchInternal::ByteSize(w->batch) > max_size) ||
         // Do not make batch too big
+        (leader->ingest_wbwi || w->ingest_wbwi)
+        // ingesting WBWI needs to be its own group
     ) {
       // remove from list
       w->link_older->link_newer = w->link_newer;

db/write_thread.h

@@ -148,6 +148,8 @@ class WriteThread {
     Writer* link_older;  // read/write only before linking, or as leader
     Writer* link_newer;  // lazy, read/write only before linking, or as leader
 
+    bool ingest_wbwi;
+
     Writer()
         : batch(nullptr),
           sync(false),
@@ -174,7 +176,8 @@ class WriteThread {
            WriteCallback* _callback, UserWriteCallback* _user_write_cb,
            uint64_t _log_ref, bool _disable_memtable, size_t _batch_cnt = 0,
            PreReleaseCallback* _pre_release_callback = nullptr,
-           PostMemTableCallback* _post_memtable_callback = nullptr)
+           PostMemTableCallback* _post_memtable_callback = nullptr,
+           bool _ingest_wbwi = false)
         : batch(_batch),
           // TODO: store a copy of WriteOptions instead of its seperated data
           // members
@@ -196,7 +199,8 @@ class WriteThread {
           write_group(nullptr),
           sequence(kMaxSequenceNumber),
           link_older(nullptr),
-          link_newer(nullptr) {}
+          link_newer(nullptr),
+          ingest_wbwi(_ingest_wbwi) {}
 
     ~Writer() {
       if (made_waitable) {

db_stress_tool/db_stress_common.h

@@ -420,6 +420,7 @@ DECLARE_uint32(uncache_aggressiveness);
 DECLARE_int32(test_ingest_standalone_range_deletion_one_in);
 DECLARE_bool(allow_unprepared_value);
 DECLARE_string(file_temperature_age_thresholds);
+DECLARE_uint32(commit_bypass_memtable_one_in);
 
 constexpr long KB = 1024;
 constexpr int kRandomValueMaxFactor = 3;

db_stress_tool/db_stress_gflags.cc

@@ -1469,4 +1469,7 @@ DEFINE_bool(paranoid_memory_checks,
             ROCKSDB_NAMESPACE::Options().paranoid_memory_checks,
             "Sets CF option paranoid_memory_checks.");
 
+DEFINE_uint32(commit_bypass_memtable_one_in, 0,
+              "If greater than zero, transaction option will set "
+              "commit_bypass_memtable to per every N transactions on average.");
 #endif  // GFLAGS

db_stress_tool/db_stress_test_base.cc

@@ -805,7 +805,7 @@ void StressTest::ProcessRecoveredPreparedTxnsHelper(Transaction* txn,
   }
 }
 
-Status StressTest::NewTxn(WriteOptions& write_opts,
+Status StressTest::NewTxn(WriteOptions& write_opts, ThreadState* thread,
                           std::unique_ptr<Transaction>* out_txn) {
   if (!FLAGS_use_txn) {
     return Status::InvalidArgument("NewTxn when FLAGS_use_txn is not set");
@@ -821,6 +821,12 @@ Status StressTest::NewTxn(WriteOptions& write_opts,
         FLAGS_use_only_the_last_commit_time_batch_for_recovery;
     txn_options.lock_timeout = 600000;  // 10 min
     txn_options.deadlock_detect = true;
+    if (FLAGS_commit_bypass_memtable_one_in > 0) {
+      assert(FLAGS_txn_write_policy == 0);
+      assert(FLAGS_user_timestamp_size == 0);
+      txn_options.commit_bypass_memtable =
+          thread->rand.OneIn(FLAGS_commit_bypass_memtable_one_in);
+    }
     out_txn->reset(txn_db_->BeginTransaction(write_opts, txn_options));
     auto istr = std::to_string(txn_id.fetch_add(1));
     Status s = (*out_txn)->SetName("xid" + istr);
@@ -880,7 +886,7 @@ Status StressTest::ExecuteTransaction(
     WriteOptions& write_opts, ThreadState* thread,
     std::function<Status(Transaction&)>&& ops) {
   std::unique_ptr<Transaction> txn;
-  Status s = NewTxn(write_opts, &txn);
+  Status s = NewTxn(write_opts, thread, &txn);
   std::string try_again_messages;
   if (s.ok()) {
     for (int tries = 1;; ++tries) {

db_stress_tool/db_stress_test_base.h

@@ -138,7 +138,7 @@ class StressTest {
                                                   SharedState* shared);
 
   // ExecuteTransaction is recommended instead
-  Status NewTxn(WriteOptions& write_opts,
+  Status NewTxn(WriteOptions& write_opts, ThreadState* thread,
                 std::unique_ptr<Transaction>* out_txn);
   Status CommitTxn(Transaction& txn, ThreadState* thread = nullptr);
 

db_stress_tool/multi_ops_txns_stress.cc

@@ -544,6 +544,11 @@ Status MultiOpsTxnsStressTest::TestCustomOperations(
     // Should never reach here.
     assert(false);
   }
+  if (!s.ok()) {
+    fprintf(stderr, "Transaction failed %s\n", s.ToString().c_str());
+    fflush(stderr);
+    thread->shared->SafeTerminate();
+  }
 
   return s;
 }
@@ -579,7 +584,7 @@ Status MultiOpsTxnsStressTest::PrimaryKeyUpdateTxn(ThreadState* thread,
   std::string new_pk = Record::EncodePrimaryKey(new_a);
   std::unique_ptr<Transaction> txn;
   WriteOptions wopts;
-  Status s = NewTxn(wopts, &txn);
+  Status s = NewTxn(wopts, thread, &txn);
   if (!s.ok()) {
     assert(!txn);
     thread->stats.AddErrors(1);
@@ -611,7 +616,12 @@ Status MultiOpsTxnsStressTest::PrimaryKeyUpdateTxn(ThreadState* thread,
     }
     auto& key_gen = key_gen_for_a_[thread->tid];
     key_gen->UndoAllocation(new_a);
-    txn->Rollback().PermitUncheckedError();
+    s = txn->Rollback();
+    if (!s.ok()) {
+      fprintf(stderr, "Transaction rollback failed %s\n", s.ToString().c_str());
+      fflush(stderr);
+      assert(false);
+    }
   });
 
   ReadOptions ropts;
@@ -699,7 +709,7 @@ Status MultiOpsTxnsStressTest::SecondaryKeyUpdateTxn(ThreadState* thread,
                                                      uint32_t new_c) {
   std::unique_ptr<Transaction> txn;
   WriteOptions wopts;
-  Status s = NewTxn(wopts, &txn);
+  Status s = NewTxn(wopts, thread, &txn);
   if (!s.ok()) {
     assert(!txn);
     thread->stats.AddErrors(1);
@@ -735,7 +745,12 @@ Status MultiOpsTxnsStressTest::SecondaryKeyUpdateTxn(ThreadState* thread,
     }
     auto& key_gen = key_gen_for_c_[thread->tid];
     key_gen->UndoAllocation(new_c);
-    txn->Rollback().PermitUncheckedError();
+    s = txn->Rollback();
+    if (!s.ok()) {
+      fprintf(stderr, "Transaction rollback failed %s\n", s.ToString().c_str());
+      fflush(stderr);
+      assert(false);
+    }
   });
 
   // TODO (yanqin) try SetSnapshotOnNextOperation(). We currently need to take
@@ -901,7 +916,7 @@ Status MultiOpsTxnsStressTest::UpdatePrimaryIndexValueTxn(ThreadState* thread,
   std::string pk_str = Record::EncodePrimaryKey(a);
   std::unique_ptr<Transaction> txn;
   WriteOptions wopts;
-  Status s = NewTxn(wopts, &txn);
+  Status s = NewTxn(wopts, thread, &txn);
   if (!s.ok()) {
     assert(!txn);
     thread->stats.AddErrors(1);
@@ -927,7 +942,12 @@ Status MultiOpsTxnsStressTest::UpdatePrimaryIndexValueTxn(ThreadState* thread,
     } else {
       thread->stats.AddErrors(1);
     }
-    txn->Rollback().PermitUncheckedError();
+    s = txn->Rollback();
+    if (!s.ok()) {
+      fprintf(stderr, "Transaction rollback failed %s\n", s.ToString().c_str());
+      fflush(stderr);
+      assert(false);
+    }
   });
   ReadOptions ropts;
   ropts.rate_limiter_priority =
@@ -982,7 +1002,7 @@ Status MultiOpsTxnsStressTest::PointLookupTxn(ThreadState* thread,
 
   std::unique_ptr<Transaction> txn;
   WriteOptions wopts;
-  Status s = NewTxn(wopts, &txn);
+  Status s = NewTxn(wopts, thread, &txn);
   if (!s.ok()) {
     assert(!txn);
     thread->stats.AddErrors(1);
@@ -1026,7 +1046,7 @@ Status MultiOpsTxnsStressTest::RangeScanTxn(ThreadState* thread,
 
   std::unique_ptr<Transaction> txn;
   WriteOptions wopts;
-  Status s = NewTxn(wopts, &txn);
+  Status s = NewTxn(wopts, thread, &txn);
   if (!s.ok()) {
     assert(!txn);
     thread->stats.AddErrors(1);
@@ -1384,6 +1404,12 @@ Status MultiOpsTxnsStressTest::CommitAndCreateTimestampedSnapshotIfNeeded(
   } else {
     s = txn.Commit();
   }
+  if (!s.ok()) {
+    fprintf(stderr, "Txn %s commit failed with %s\n", txn.GetName().c_str(),
+            s.ToString().c_str());
+    fflush(stderr);
+  }
+
   assert(txn_db_);
   if (FLAGS_create_timestamped_snapshot_one_in > 0 &&
       thread->rand.OneInOpt(50000)) {

db_stress_tool/no_batched_ops_stress.cc

@@ -666,7 +666,7 @@ class NonBatchedOpsStressTest : public StressTest {
       if (FLAGS_rate_limit_auto_wal_flush) {
         wo.rate_limiter_priority = Env::IO_USER;
       }
-      Status s = NewTxn(wo, &txn);
+      Status s = NewTxn(wo, thread, &txn);
       if (!s.ok()) {
         fprintf(stderr, "NewTxn error: %s\n", s.ToString().c_str());
         shared->SafeTerminate();
@@ -1127,7 +1127,7 @@ class NonBatchedOpsStressTest : public StressTest {
         write_options.rate_limiter_priority = Env::IO_USER;
       }
 
-      const Status s = NewTxn(write_options, &txn);
+      const Status s = NewTxn(write_options, thread, &txn);
       if (!s.ok()) {
         fprintf(stderr, "NewTxn error: %s\n", s.ToString().c_str());
         thread->shared->SafeTerminate();

include/rocksdb/utilities/transaction.h

@@ -750,6 +750,8 @@ class Transaction {
 
   virtual TxnTimestamp GetCommitTimestamp() const { return kMaxTxnTimestamp; }
 
+  virtual bool GetCommitBypassMemTable() const { return false; }
+
  protected:
   explicit Transaction(const TransactionDB* /*db*/) {}
   Transaction() : log_number_(0), txn_state_(STARTED) {}

include/rocksdb/utilities/transaction_db.h

@@ -339,6 +339,23 @@ struct TransactionOptions {
   // size in APIs that MyRocks currently are using, including Put, Merge, Delete
   // DeleteRange, SingleDelete.
   bool write_batch_track_timestamp_size = false;
+
+  // EXPERIMENTAL
+  // Only supports write-committed policy. If set to true, the transaction will
+  // skip memtable write and ingest into the DB directly during Commit(). This
+  // makes Commit() much faster for transactions with many operations.
+  // Transactions with Merge() or PutEntity() is not supported yet.
+  //
+  // Note that the transaction will be ingested as an immutable memtable for
+  // CFs it updates, and the current memtable will be switched to a new one.
+  // So ingesting many transactions in a short period of time may cause stall
+  // due to too many memtables.
+  // Note that the ingestion relies on the transaction's underlying index,
+  // (WriteBatchWithIndex), so updates that are added to the transaction
+  // 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.
+  bool commit_bypass_memtable = false;
 };
 
 // The per-write optimizations that do not involve transactions. TransactionDB

include/rocksdb/utilities/write_batch_with_index.h

@@ -74,6 +74,10 @@ class WBWIIterator {
   // WriteBatchWithIndex
   virtual WriteEntry Entry() const = 0;
 
+  // For this user key, there is a single delete in this write batch,
+  // and it was overwritten by another update.
+  virtual bool HasOverWrittenSingleDel() const { return false; }
+
   virtual Status status() const = 0;
 };
 
@@ -350,6 +354,17 @@ class WriteBatchWithIndex : public WriteBatchBase {
   void SetMaxBytes(size_t max_bytes) override;
   size_t GetDataSize() const;
 
+  struct CFStat {
+    uint32_t entry_count = 0;
+    uint32_t overwritten_sd_count = 0;
+  };
+  // Will track CF ID, per CF entry count and overwritten sd count.
+  // Should be enabled when WBWI is empty for correct tracking.
+  void SetTrackPerCFStat(bool track);
+  const std::unordered_map<uint32_t, CFStat>& GetCFStats() const;
+
+  bool GetOverwriteKey() const;
+
  private:
   friend class PessimisticTransactionDB;
   friend class WritePreparedTxn;

java/rocksjni/write_batch_test.cc

@@ -61,7 +61,7 @@ jbyteArray Java_org_rocksdb_WriteBatchTest_getContents(JNIEnv* env,
   ROCKSDB_NAMESPACE::ScopedArenaPtr<ROCKSDB_NAMESPACE::InternalIterator> iter(
       mem->NewIterator(ROCKSDB_NAMESPACE::ReadOptions(),
                        /*seqno_to_time_mapping=*/nullptr, &arena,
-                       /*prefix_extractor=*/nullptr));
+                       /*prefix_extractor=*/nullptr, /*for_flush=*/false));
   for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
     ROCKSDB_NAMESPACE::ParsedInternalKey ikey;
     ikey.clear();

memtable/wbwi_memtable.cc

@@ -21,6 +21,27 @@ const std::unordered_map<WriteType, ValueType>
         // kLogDataRecord, kXIDRecord, kUnknownRecord
 };
 
+InternalIterator* WBWIMemTable::NewIterator(
+    const ReadOptions&, UnownedPtr<const SeqnoToTimeMapping>, Arena* arena,
+    const SliceTransform* /* prefix_extractor */, bool for_flush) {
+  // Ingested WBWIMemTable should have an assigned seqno
+  assert(assigned_seqno_.upper_bound != kMaxSequenceNumber);
+  assert(assigned_seqno_.lower_bound != kMaxSequenceNumber);
+  assert(arena);
+  auto mem = arena->AllocateAligned(sizeof(WBWIMemTableIterator));
+  return new (mem) WBWIMemTableIterator(
+      std::unique_ptr<WBWIIterator>(wbwi_->NewIterator(cf_id_)),
+      assigned_seqno_, comparator_, for_flush);
+}
+
+inline InternalIterator* WBWIMemTable::NewIterator() const {
+  assert(assigned_seqno_.upper_bound != kMaxSequenceNumber);
+  assert(assigned_seqno_.lower_bound != kMaxSequenceNumber);
+  return new WBWIMemTableIterator(
+      std::unique_ptr<WBWIIterator>(wbwi_->NewIterator(cf_id_)),
+      assigned_seqno_, comparator_, /*for_flush=*/false);
+}
+
 bool WBWIMemTable::Get(const LookupKey& key, std::string* value,
                        PinnableWideColumns* columns, std::string* timestamp,
                        Status* s, MergeContext* merge_context,
@@ -34,7 +55,8 @@ bool WBWIMemTable::Get(const LookupKey& key, std::string* value,
   assert(immutable_memtable);
   assert(!timestamp);  // TODO: support UDT
   assert(!columns);    // TODO: support WideColumn
-  assert(global_seqno_ != kMaxSequenceNumber);
+  assert(assigned_seqno_.upper_bound != kMaxSequenceNumber);
+  assert(assigned_seqno_.lower_bound != kMaxSequenceNumber);
   // WBWI does not support DeleteRange yet.
   assert(!wbwi_->GetWriteBatch()->HasDeleteRange());
 

memtable/wbwi_memtable.h

@@ -8,157 +8,56 @@
 #include "rocksdb/utilities/write_batch_with_index.h"
 
 namespace ROCKSDB_NAMESPACE {
-
-class WBWIMemTableIterator final : public InternalIterator {
- public:
-  WBWIMemTableIterator(std::unique_ptr<WBWIIterator>&& it, SequenceNumber seqno,
-                       const Comparator* comparator)
-      : it_(std::move(it)), global_seqno_(seqno), comparator_(comparator) {
-    assert(seqno != kMaxSequenceNumber);
-    s_.PermitUncheckedError();
-  }
-
-  // No copying allowed
-  WBWIMemTableIterator(const WBWIMemTableIterator&) = delete;
-  WBWIMemTableIterator& operator=(const WBWIMemTableIterator&) = delete;
-
-  bool Valid() const override { return valid_; }
-
-  void SeekToFirst() override {
-    it_->SeekToFirst();
-    UpdateKey();
-  }
-
-  void SeekToLast() override {
-    it_->SeekToLast();
-    UpdateKey();
-  }
-
-  void Seek(const Slice& target) override {
-    Slice target_user_key = ExtractUserKey(target);
-    it_->Seek(target_user_key);
-    if (it_->Valid()) {
-      // compare seqno
-      SequenceNumber seqno = GetInternalKeySeqno(target);
-      if (seqno < global_seqno_ &&
-          comparator_->Compare(it_->Entry().key, target_user_key) == 0) {
-        it_->Next();
-        // TODO: cannot assume distinct keys once Merge is supported
-        if (it_->Valid()) {
-          assert(comparator_->Compare(it_->Entry().key, target_user_key) > 0);
-        }
-      }
-    }
-    UpdateKey();
-  }
-
-  void SeekForPrev(const Slice& target) override {
-    Slice target_user_key = ExtractUserKey(target);
-    it_->SeekForPrev(target_user_key);
-    if (it_->Valid()) {
-      SequenceNumber seqno = GetInternalKeySeqno(target);
-      if (seqno > global_seqno_ &&
-          comparator_->Compare(it_->Entry().key, target_user_key) == 0) {
-        it_->Prev();
-        if (it_->Valid()) {
-          // TODO: cannot assume distinct keys once Merge is supported
-          assert(comparator_->Compare(it_->Entry().key, target_user_key) < 0);
-        }
-      }
-    }
-    UpdateKey();
-  }
-
-  void Next() override {
-    assert(Valid());
-    it_->Next();
-    UpdateKey();
-  }
-
-  bool NextAndGetResult(IterateResult* result) override {
-    assert(Valid());
-    Next();
-    bool is_valid = Valid();
-    if (is_valid) {
-      result->key = key();
-      result->bound_check_result = IterBoundCheck::kUnknown;
-      result->value_prepared = true;
-    }
-    return is_valid;
-  }
-
-  void Prev() override {
-    assert(Valid());
-    it_->Prev();
-    UpdateKey();
-  }
-
-  Slice key() const override {
-    assert(Valid());
-    return key_;
-  }
-
-  Slice value() const override {
-    assert(Valid());
-    return it_->Entry().value;
-  }
-
-  Status status() const override {
-    assert(it_->status().ok());
-    return s_;
-  }
-
- private:
-  static const std::unordered_map<WriteType, ValueType> WriteTypeToValueTypeMap;
-
-  void UpdateKey() {
-    valid_ = it_->Valid();
-    if (!Valid()) {
-      key_.clear();
-      return;
-    }
-    auto t = WriteTypeToValueTypeMap.find(it_->Entry().type);
-    assert(t != WriteTypeToValueTypeMap.end());
-    if (t == WriteTypeToValueTypeMap.end()) {
-      key_.clear();
-      valid_ = false;
-      s_ = Status::Corruption("Unexpected write_batch_with_index entry type " +
-                              std::to_string(it_->Entry().type));
-      return;
-    }
-    key_buf_.SetInternalKey(it_->Entry().key, global_seqno_, t->second);
-    key_ = key_buf_.GetInternalKey();
-  }
-
-  std::unique_ptr<WBWIIterator> it_;
-  // The sequence number of entries in this write batch.
-  SequenceNumber global_seqno_;
-  const Comparator* comparator_;
-  IterKey key_buf_;
-  // The current internal key.
-  Slice key_;
-  Status s_;
-  bool valid_ = false;
-};
-
+// An implementation of the ReadOnlyMemTable interface based on the content
+// 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
+// Since the keys in WBWI do not have sequence number, the memtable needs to be
+// assigned a range of sequence numbers through AssignSequenceNumbers(seqno)
+// before being available for reads.
+// With overwrite_key = true, WBWI keeps track of the most recent update for
+// each key, and each such key will be assigned seqno.upper_bound during reads.
+// One exception is during flush, consider the following scenario:
+// - WBWI has SD(k) then PUT(k, v1)
+// - DB has PUT(k, v2) in L1
+// - flush WBWI adds PUT(k, v1) into L0
+// - live memtable contains SD(k)
+// - flush live memtable and compact it with L0 will drop SD(k) and PUT(k, v1)
+// - the PUT(k, v2) in L1 incorrectly becomes visible
+// So during flush, iterator from this memtable will need emit overwritten
+// single deletion. These single deletion entries will be
+// assigned seqno.upper_bound - 1.
 class WBWIMemTable final : public ReadOnlyMemTable {
  public:
+  struct SeqnoRange {
+    SequenceNumber lower_bound = kMaxSequenceNumber;
+    SequenceNumber upper_bound = kMaxSequenceNumber;
+  };
   WBWIMemTable(const std::shared_ptr<WriteBatchWithIndex>& wbwi,
                const Comparator* cmp, uint32_t cf_id,
                const ImmutableOptions* immutable_options,
-               const MutableCFOptions* cf_options)
+               const MutableCFOptions* cf_options,
+               const WriteBatchWithIndex::CFStat& stat)
       : wbwi_(wbwi),
         comparator_(cmp),
         ikey_comparator_(comparator_),
         moptions_(*immutable_options, *cf_options),
         clock_(immutable_options->clock),
-        cf_id_(cf_id) {}
+        // We need to include overwritten_sd_count in num_entries_ since flush
+        // verifies number of entries processed and that iterator for this
+        // memtable will emit overwritten SingleDelete entries during flush, See
+        // comment above WBWIMemTableIterator for more detail.
+        num_entries_(stat.entry_count + stat.overwritten_sd_count),
+        cf_id_(cf_id) {
+    assert(wbwi->GetOverwriteKey());
+  }
 
   // No copying allowed
   WBWIMemTable(const WBWIMemTable&) = delete;
   WBWIMemTable& operator=(const WBWIMemTable&) = delete;
 
-  ~WBWIMemTable() override = default;
+  ~WBWIMemTable() override { assert(refs_ == 0); }
 
   const char* Name() const override { return "WBWIMemTable"; }
 
@@ -182,17 +81,11 @@ class WBWIMemTable final : public ReadOnlyMemTable {
     assert(false);
   }
 
-  InternalIterator* NewIterator(
-      const ReadOptions&, UnownedPtr<const SeqnoToTimeMapping>, Arena* arena,
-      const SliceTransform* /* prefix_extractor */) override {
-    // Ingested WBWIMemTable should have an assigned seqno
-    assert(global_seqno_ != kMaxSequenceNumber);
-    assert(arena);
-    auto mem = arena->AllocateAligned(sizeof(WBWIMemTableIterator));
-    return new (mem) WBWIMemTableIterator(
-        std::unique_ptr<WBWIIterator>(wbwi_->NewIterator(cf_id_)),
-        global_seqno_, comparator_);
-  }
+  InternalIterator* NewIterator(const ReadOptions&,
+                                UnownedPtr<const SeqnoToTimeMapping>,
+                                Arena* arena,
+                                const SliceTransform* /* prefix_extractor */,
+                                bool for_flush) override;
 
   // Returns an iterator that wraps a MemTableIterator and logically strips the
   // user-defined timestamp of each key. This API is only used by flush when
@@ -235,13 +128,7 @@ class WBWIMemTable final : public ReadOnlyMemTable {
   void MultiGet(const ReadOptions& read_options, MultiGetRange* range,
                 ReadCallback* callback, bool immutable_memtable) override;
 
-  uint64_t NumEntries() const override {
-    // FIXME: used in
-    // - verify number of entries processed during flush
-    // - stats for estimate num entries and num entries in immutable memtables
-    // - MemPurgeDecider
-    return 0;
-  }
+  uint64_t NumEntries() const override { return num_entries_; }
 
   uint64_t NumDeletion() const override {
     // FIXME: this is used for stats and event logging
@@ -259,7 +146,9 @@ class WBWIMemTable final : public ReadOnlyMemTable {
     return 0;
   }
 
-  SequenceNumber GetEarliestSequenceNumber() override { return global_seqno_; }
+  SequenceNumber GetEarliestSequenceNumber() override {
+    return assigned_seqno_.lower_bound;
+  }
 
   bool IsEmpty() const override {
     // Ideally also check that wbwi contains updates from this CF. For now, we
@@ -267,7 +156,9 @@ class WBWIMemTable final : public ReadOnlyMemTable {
     return wbwi_->GetWriteBatch()->Count() == 0;
   }
 
-  SequenceNumber GetFirstSequenceNumber() override { return global_seqno_; }
+  SequenceNumber GetFirstSequenceNumber() override {
+    return assigned_seqno_.lower_bound;
+  }
 
   uint64_t GetMinLogContainingPrepSection() override {
     // FIXME: used to retain WAL with pending Prepare
@@ -304,31 +195,206 @@ class WBWIMemTable final : public ReadOnlyMemTable {
   }
 
   // Assign a sequence number to the entries in this memtable.
-  void SetGlobalSequenceNumber(SequenceNumber global_seqno) {
+  void AssignSequenceNumbers(const SeqnoRange& seqno_range) {
     // Not expecting to assign seqno multiple times.
-    assert(global_seqno_ == kMaxSequenceNumber);
-    global_seqno_ = global_seqno;
+    assert(assigned_seqno_.lower_bound == kMaxSequenceNumber);
+    assert(assigned_seqno_.upper_bound == kMaxSequenceNumber);
+
+    assigned_seqno_ = seqno_range;
+
+    assert(assigned_seqno_.lower_bound <= assigned_seqno_.upper_bound);
+    assert(assigned_seqno_.upper_bound != kMaxSequenceNumber);
+  }
+
+  void SetMinPrepLog(uint64_t min_prep_log) {
+    min_prep_log_referenced_ = min_prep_log;
   }
 
  private:
-  InternalIterator* NewIterator() const {
-    assert(global_seqno_ != kMaxSequenceNumber);
-    return new WBWIMemTableIterator(
-        std::unique_ptr<WBWIIterator>(wbwi_->NewIterator(cf_id_)),
-        global_seqno_, comparator_);
-  }
+  inline InternalIterator* NewIterator() const;
 
   Slice newest_udt_;
   std::shared_ptr<WriteBatchWithIndex> wbwi_;
   const Comparator* comparator_;
   InternalKeyComparator ikey_comparator_;
-  SequenceNumber global_seqno_ = kMaxSequenceNumber;
+  SeqnoRange assigned_seqno_;
   const ImmutableMemTableOptions moptions_;
   SystemClock* clock_;
   uint64_t min_prep_log_referenced_{0};
+  uint64_t num_entries_;
   // WBWI can contains updates to multiple CFs. `cf_id_` determines which CF
   // this memtable is for.
   uint32_t cf_id_;
 };
 
+class WBWIMemTableIterator final : public InternalIterator {
+ public:
+  WBWIMemTableIterator(std::unique_ptr<WBWIIterator>&& it,
+                       const WBWIMemTable::SeqnoRange& assigned_seqno,
+                       const Comparator* comparator, bool for_flush)
+      : it_(std::move(it)),
+        assigned_seqno_(assigned_seqno),
+        comparator_(comparator),
+        emit_overwritten_single_del_(for_flush) {
+    assert(assigned_seqno_.lower_bound <= assigned_seqno_.upper_bound);
+    assert(assigned_seqno_.upper_bound < kMaxSequenceNumber);
+    s_.PermitUncheckedError();
+  }
+
+  // No copying allowed
+  WBWIMemTableIterator(const WBWIMemTableIterator&) = delete;
+  WBWIMemTableIterator& operator=(const WBWIMemTableIterator&) = delete;
+
+  bool Valid() const override { return valid_; }
+
+  void SeekToFirst() override {
+    it_->SeekToFirst();
+    UpdateKey();
+  }
+
+  void SeekToLast() override {
+    it_->SeekToLast();
+    UpdateKey();
+  }
+
+  void Seek(const Slice& target) override {
+    Slice target_user_key = ExtractUserKey(target);
+    it_->Seek(target_user_key);
+    if (it_->Valid()) {
+      // compare seqno
+      SequenceNumber seqno = GetInternalKeySeqno(target);
+      assert(!emit_overwritten_single_del_);
+      // For now all keys are assigned seqno_ub_, this may change after merge
+      // is supported.
+      assert(seqno <= assigned_seqno_.lower_bound ||
+             seqno >= assigned_seqno_.upper_bound);
+      if (seqno < assigned_seqno_.upper_bound &&
+          comparator_->Compare(it_->Entry().key, target_user_key) == 0) {
+        it_->Next();
+        // TODO: cannot assume distinct keys once Merge is supported
+        if (it_->Valid()) {
+          assert(comparator_->Compare(it_->Entry().key, target_user_key) > 0);
+        }
+      }
+    }
+    UpdateKey();
+  }
+
+  void SeekForPrev(const Slice& target) override {
+    Slice target_user_key = ExtractUserKey(target);
+    it_->SeekForPrev(target_user_key);
+    if (it_->Valid()) {
+      SequenceNumber seqno = GetInternalKeySeqno(target);
+      assert(seqno <= assigned_seqno_.lower_bound ||
+             seqno >= assigned_seqno_.upper_bound);
+      if (seqno > assigned_seqno_.upper_bound &&
+          comparator_->Compare(it_->Entry().key, target_user_key) == 0) {
+        it_->Prev();
+        if (it_->Valid()) {
+          // TODO: cannot assume distinct keys once Merge is supported
+          assert(comparator_->Compare(it_->Entry().key, target_user_key) < 0);
+        }
+      }
+    }
+    UpdateKey();
+  }
+
+  void Next() override {
+    // Only need to emit single deletion during flush. Since Flush does
+    // sequential forward scan, we only need to emit single deletion in Next(),
+    // and do not need to consider iterator direction change.
+    assert(Valid());
+    if (emit_overwritten_single_del_) {
+      if (it_->HasOverWrittenSingleDel() && !at_overwritten_single_del_) {
+        UpdateSingleDeleteKey();
+        return;
+      }
+      at_overwritten_single_del_ = false;
+    }
+
+    it_->Next();
+    UpdateKey();
+  }
+
+  bool NextAndGetResult(IterateResult* result) override {
+    assert(Valid());
+    Next();
+    bool is_valid = Valid();
+    if (is_valid) {
+      result->key = key();
+      result->bound_check_result = IterBoundCheck::kUnknown;
+      result->value_prepared = true;
+    }
+    return is_valid;
+  }
+
+  void Prev() override {
+    assert(Valid());
+    it_->Prev();
+    UpdateKey();
+  }
+
+  Slice key() const override {
+    assert(Valid());
+    return key_;
+  }
+
+  Slice value() const override {
+    assert(Valid());
+    // TODO: it_->Entry() is not trivial, cache it
+    return it_->Entry().value;
+  }
+
+  Status status() const override {
+    assert(it_->status().ok());
+    return s_;
+  }
+
+  bool IsValuePinned() const override { return true; }
+
+ private:
+  static const std::unordered_map<WriteType, ValueType> WriteTypeToValueTypeMap;
+
+  void UpdateKey() {
+    valid_ = it_->Valid();
+    if (!Valid()) {
+      key_.clear();
+      return;
+    }
+    auto t = WriteTypeToValueTypeMap.find(it_->Entry().type);
+    assert(t != WriteTypeToValueTypeMap.end());
+    if (t == WriteTypeToValueTypeMap.end()) {
+      key_.clear();
+      valid_ = false;
+      s_ = Status::Corruption("Unexpected write_batch_with_index entry type " +
+                              std::to_string(t->second));
+      return;
+    }
+    key_buf_.SetInternalKey(it_->Entry().key, assigned_seqno_.upper_bound,
+                            t->second);
+    key_ = key_buf_.GetInternalKey();
+  }
+
+  void UpdateSingleDeleteKey() {
+    assert(it_->Valid());
+    assert(Valid());
+    assert(assigned_seqno_.lower_bound < assigned_seqno_.upper_bound);
+    key_buf_.SetInternalKey(it_->Entry().key, assigned_seqno_.upper_bound - 1,
+                            kTypeSingleDeletion);
+    key_ = key_buf_.GetInternalKey();
+    at_overwritten_single_del_ = true;
+  }
+
+  std::unique_ptr<WBWIIterator> it_;
+  const WBWIMemTable::SeqnoRange assigned_seqno_;
+  const Comparator* comparator_;
+  IterKey key_buf_;
+  // The current internal key.
+  Slice key_;
+  Status s_;
+  bool valid_ = false;
+  bool at_overwritten_single_del_ = false;
+  bool emit_overwritten_single_del_ = false;
+};
+
 }  // namespace ROCKSDB_NAMESPACE

table/table_test.cc

@@ -534,7 +534,8 @@ class MemTableConstructor : public Constructor {
       const SliceTransform* /*prefix_extractor*/) const override {
     return new KeyConvertingIterator(
         memtable_->NewIterator(ReadOptions(), /*seqno_to_time_mapping=*/nullptr,
-                               &arena_, /*prefix_extractor=*/nullptr),
+                               &arena_, /*prefix_extractor=*/nullptr,
+                               /*for_flush=*/false),
         true);
   }
 
@@ -4911,7 +4912,8 @@ TEST_F(MemTableTest, Simple) {
     if (i == 0) {
       iter = GetMemTable()->NewIterator(ReadOptions(),
                                         /*seqno_to_time_mapping=*/nullptr,
-                                        &arena, /*prefix_extractor=*/nullptr);
+                                        &arena, /*prefix_extractor=*/nullptr,
+                                        /*for_flush=*/false);
       arena_iter_guard.reset(iter);
     } else {
       iter = GetMemTable()->NewRangeTombstoneIterator(

tools/db_crashtest.py

@@ -674,6 +674,7 @@ multiops_wc_txn_params = {
     "txn_write_policy": 0,
     # TODO re-enable pipelined write. Not well tested atm
     "enable_pipelined_write": 0,
+    "commit_bypass_memtable_one_in": random.choice([0] * 4 + [100]),
 }
 
 multiops_wp_txn_params = {
@@ -988,6 +989,18 @@ def finalize_and_sanitize(src_params):
         or dest_params.get("delrangepercent") == 0
     ):
         dest_params["test_ingest_standalone_range_deletion_one_in"] = 0
+    if dest_params.get("commit_bypass_memtable_one_in", 0) > 0:
+        dest_params["enable_blob_files"] = 0
+        dest_params["allow_setting_blob_options_dynamically"] = 0
+        dest_params["atomic_flush"] = 0
+        dest_params["allow_concurrent_memtable_write"] = 0
+        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
+
     return dest_params
 
 

unreleased_history/new_features/txn-bypass-memtable.md

@@ -0,0 +1 @@
+* Introduce `TransactionOptions::commit_bypass_memtable` to enable transaction commit to bypass memtable insertions. This can be beneficial for transactions with many operations, as it reduces commit time that is mostly spent on memtable insertion.  

utilities/transactions/pessimistic_transaction.cc

@@ -103,6 +103,9 @@ void PessimisticTransaction::Initialize(const TransactionOptions& txn_options) {
 
   read_timestamp_ = kMaxTxnTimestamp;
   commit_timestamp_ = kMaxTxnTimestamp;
+
+  commit_bypass_memtable_ = txn_options.commit_bypass_memtable;
+  write_batch_.SetTrackPerCFStat(txn_options.commit_bypass_memtable);
 }
 
 PessimisticTransaction::~PessimisticTransaction() {
@@ -843,6 +846,7 @@ Status WriteCommittedTxn::CommitInternal() {
   if (!needs_ts) {
     s = WriteBatchInternal::MarkCommit(working_batch, name_);
   } else {
+    assert(!commit_bypass_memtable_);
     assert(commit_timestamp_ != kMaxTxnTimestamp);
     char commit_ts_buf[sizeof(kMaxTxnTimestamp)];
     EncodeFixed64(commit_ts_buf, commit_timestamp_);
@@ -878,11 +882,14 @@ Status WriteCommittedTxn::CommitInternal() {
   // any operations appended to this working_batch will be ignored from WAL
   working_batch->MarkWalTerminationPoint();
 
-  // insert prepared batch into Memtable only skipping WAL.
-  // Memtable will ignore BeginPrepare/EndPrepare markers
-  // in non recovery mode and simply insert the values
-  s = WriteBatchInternal::Append(working_batch, wb);
-  assert(s.ok());
+  const bool bypass_memtable = commit_bypass_memtable_ && wb->Count() > 0;
+  if (!bypass_memtable) {
+    // insert prepared batch into Memtable only skipping WAL.
+    // Memtable will ignore BeginPrepare/EndPrepare markers
+    // in non recovery mode and simply insert the values
+    s = WriteBatchInternal::Append(working_batch, wb);
+    assert(s.ok());
+  }
 
   uint64_t seq_used = kMaxSequenceNumber;
   SnapshotCreationCallback snapshot_creation_cb(db_impl_, commit_timestamp_,
@@ -896,12 +903,28 @@ Status WriteCommittedTxn::CommitInternal() {
       post_mem_cb = &snapshot_creation_cb;
     }
   }
-  s = db_impl_->WriteImpl(write_options_, working_batch, /*callback*/ nullptr,
-                          /*user_write_cb=*/nullptr,
-                          /*log_used*/ nullptr, /*log_ref*/ log_number_,
-                          /*disable_memtable*/ false, &seq_used,
-                          /*batch_cnt=*/0, /*pre_release_callback=*/nullptr,
-                          post_mem_cb);
+  assert(log_number_ > 0);
+  if (bypass_memtable) {
+    s = db_impl_->WriteImpl(
+        write_options_, working_batch, /*callback*/ nullptr,
+        /*user_write_cb=*/nullptr,
+        /*log_used*/ nullptr, /*log_ref*/ log_number_,
+        /*disable_memtable*/ false, &seq_used,
+        /*batch_cnt=*/0, /*pre_release_callback=*/nullptr, post_mem_cb,
+        /*wbwi=*/std::make_shared<WriteBatchWithIndex>(std::move(write_batch_)),
+        /*min_prep_log=*/log_number_);
+    // Reset write_batch_ since it's accessed in transaction clean up and
+    // might be used for transaction reuse.
+    write_batch_ = WriteBatchWithIndex(cmp_, 0, true, 0,
+                                       write_options_.protection_bytes_per_key);
+  } else {
+    s = db_impl_->WriteImpl(write_options_, working_batch, /*callback*/ nullptr,
+                            /*user_write_cb=*/nullptr,
+                            /*log_used*/ nullptr, /*log_ref*/ log_number_,
+                            /*disable_memtable*/ false, &seq_used,
+                            /*batch_cnt=*/0, /*pre_release_callback=*/nullptr,
+                            post_mem_cb);
+  }
   assert(!s.ok() || seq_used != kMaxSequenceNumber);
   if (s.ok()) {
     SetId(seq_used);

utilities/transactions/pessimistic_transaction.h

@@ -122,13 +122,6 @@ class PessimisticTransaction : public TransactionBaseImpl {
                      ColumnFamilyHandle* column_family = nullptr) override;
 
  protected:
-  // Refer to
-  // TransactionOptions::use_only_the_last_commit_time_batch_for_recovery
-  bool use_only_the_last_commit_time_batch_for_recovery_ = false;
-  // Refer to
-  // TransactionOptions::skip_prepare
-  bool skip_prepare_ = false;
-
   virtual Status PrepareInternal() = 0;
 
   virtual Status CommitWithoutPrepareInternal() = 0;
@@ -167,6 +160,16 @@ class PessimisticTransaction : public TransactionBaseImpl {
   TxnTimestamp read_timestamp_{kMaxTxnTimestamp};
   TxnTimestamp commit_timestamp_{kMaxTxnTimestamp};
 
+  // Refer to
+  // TransactionOptions::use_only_the_last_commit_time_batch_for_recovery
+  bool use_only_the_last_commit_time_batch_for_recovery_ = false;
+  // Refer to
+  // TransactionOptions::skip_prepare
+  bool skip_prepare_ = false;
+  // Refer to
+  // TransactionOptions::commit_bypass_memtable
+  bool commit_bypass_memtable_ = false;
+
  private:
   friend class TransactionTest_ValidateSnapshotTest_Test;
   // Used to create unique ids for transactions.
@@ -304,6 +307,10 @@ class WriteCommittedTxn : public PessimisticTransaction {
   Status SetCommitTimestamp(TxnTimestamp ts) override;
   TxnTimestamp GetCommitTimestamp() const override { return commit_timestamp_; }
 
+  bool GetCommitBypassMemTable() const override {
+    return commit_bypass_memtable_;
+  }
+
  private:
   template <typename TValue>
   Status GetForUpdateImpl(const ReadOptions& read_options,

utilities/transactions/transaction_test.cc

@@ -8,6 +8,7 @@
 #include <algorithm>
 #include <array>
 #include <functional>
+#include <numeric>
 #include <string>
 #include <thread>
 
@@ -8100,6 +8101,485 @@ TEST_F(TransactionDBTest, FlushedLogWithPendingPrepareIsSynced) {
   }
 }
 
+class CommitBypassMemtableTest : public DBTestBase,
+                                 public ::testing::WithParamInterface<bool> {
+ public:
+  CommitBypassMemtableTest() : DBTestBase("commit_bypass_memtable_test", true) {
+    SetUpTransactionDB();
+  }
+
+ protected:
+  TransactionDB* txn_db = nullptr;
+  Options options;
+  TransactionDBOptions txn_db_opts;
+
+  void SetUpTransactionDB() {
+    options = CurrentOptions();
+    options.create_if_missing = true;
+    options.allow_2pc = true;
+    options.two_write_queues = GetParam();
+    // Avoid write stall
+    options.max_write_buffer_number = 8;
+    // Destroy the DB to recreate as a TransactionDB.
+    Close();
+    Destroy(options, true);
+
+    txn_db_opts.write_policy = TxnDBWritePolicy::WRITE_COMMITTED;
+    ASSERT_OK(TransactionDB::Open(options, txn_db_opts, dbname_, &txn_db));
+    ASSERT_NE(txn_db, nullptr);
+    db_ = txn_db;
+  }
+};
+
+INSTANTIATE_TEST_CASE_P(, CommitBypassMemtableTest, testing::Bool());
+
+// TODO: parameterize other tests in the file with commit_bypass_memtable
+TEST_P(CommitBypassMemtableTest, SingleCFUpdate) {
+  // 10000 updates for one CF in a single transaction.
+  // Tests basic read before and after flush, with and w/o snapshot.
+  for (bool disable_flush : {false, true}) {
+    SetUpTransactionDB();
+    if (disable_flush) {
+      ASSERT_OK(db_->PauseBackgroundWork());
+    }
+
+    WriteOptions wopts;
+    std::unordered_set<std::string> not_found_at_snapshot;
+    std::map<std::string, std::string> snapshot_map;
+    std::map<std::string, std::string> expected_map;
+
+    for (int i = 0; i < 10000; i += 3) {
+      ASSERT_OK(db_->Put(wopts, Key(i), "foo"));
+      not_found_at_snapshot.insert(Key(i + 1));
+      snapshot_map[Key(i)] = "foo";
+    }
+    const Snapshot* snapshot = db_->GetSnapshot();
+
+    TransactionOptions txn_opts;
+    txn_opts.commit_bypass_memtable = true;
+    Transaction* txn1 = txn_db->BeginTransaction(wopts, txn_opts, nullptr);
+    std::unordered_set<std::string> expected_not_found;
+    for (int i = 0; i < 10000; i += 2) {
+      std::string v = "val" + std::to_string(i);
+      ASSERT_OK(txn1->Put(Key(i), v));
+      expected_map[Key(i)] = v;
+      ASSERT_OK(txn1->Delete(Key(i + 1)));
+      expected_not_found.insert(Key(i + 1));
+    }
+    ASSERT_OK(txn1->SetName("xid1"));
+    ASSERT_OK(txn1->Prepare());
+    ASSERT_OK(txn1->Commit());
+    delete txn1;
+
+    ReadOptions ro_snapshot;
+    ro_snapshot.snapshot = snapshot;
+    // Verify at snapshot
+    VerifyDBFromMap(snapshot_map, nullptr, false, &ro_snapshot, nullptr,
+                    &not_found_at_snapshot);
+    // Verify latest state
+    VerifyDBFromMap(expected_map, nullptr, false, nullptr, nullptr,
+                    &expected_not_found);
+
+    db_->ReleaseSnapshot(snapshot);
+    if (disable_flush) {
+      ASSERT_OK(db_->ContinueBackgroundWork());
+    }
+    ASSERT_OK(db_->Flush({}));
+    VerifyDBFromMap(expected_map, nullptr, false, nullptr, nullptr,
+                    &expected_not_found);
+  }
+}
+
+TEST_P(CommitBypassMemtableTest, SingleCFUpdateWithOverWrite) {
+  // Test the case where DB has base data and there are overwrites
+  // over the data in WBWI for one CF.
+  //
+  // live mem has k2
+  // ingested wbwi has k2 del k4, del k5, k6
+  // older imm mem has k3, k4, k5, k6
+  // SST has k1, k2, k4 (if not using single del)
+  for (bool single_del : {false, true}) {
+    SCOPED_TRACE("use single_del " + std::to_string(single_del));
+    for (bool disable_flush : {false, true}) {
+      SCOPED_TRACE("disable_flush: " + std::to_string(disable_flush));
+      SetUpTransactionDB();
+
+      std::map<std::string, std::string> expected = {{"k1", "sst"},
+                                                     {"k2", "sst"}};
+      WriteOptions wopts;
+      ASSERT_OK(txn_db->Put(wopts, "k1", "sst"));
+      ASSERT_OK(txn_db->Put(wopts, "k2", "sst"));
+      if (!single_del) {
+        // single del does not allow overwrite. We write to this key below.
+        ASSERT_OK(txn_db->Put(wopts, "k4", "sst"));
+        expected["k4"] = "sst";
+      }
+      ASSERT_OK(txn_db->Flush({}));
+      std::unordered_set<std::string> not_found;
+      VerifyDBFromMap(expected, nullptr, false, nullptr, nullptr, &not_found);
+
+      if (disable_flush) {
+        ASSERT_OK(db_->PauseBackgroundWork());
+      }
+
+      // immutable mem
+      TransactionOptions topts;
+      Transaction* txn = txn_db->BeginTransaction(wopts, topts);
+      ASSERT_OK(txn->Put("k3", "imm"));
+      ASSERT_OK(txn->Put("k4", "imm"));
+      ASSERT_OK(txn->Put("k5", "imm"));
+      ASSERT_OK(txn->Put("k6", "imm"));
+      ASSERT_OK(txn->SetName("xid1"));
+      ASSERT_OK(txn->Prepare());
+      ASSERT_OK(txn->Commit());
+      auto dbimpl = static_cast_with_check<DBImpl>(txn_db->GetRootDB());
+      ASSERT_OK(dbimpl->TEST_SwitchMemtable());
+      for (const auto k : {"k3", "k4", "k5", "k6"}) {
+        expected[k] = "imm";
+      }
+      VerifyDBFromMap(expected, nullptr, false, nullptr, nullptr, &not_found);
+
+      uint64_t num_imm_mems;
+      if (disable_flush) {
+        ASSERT_TRUE(txn_db->GetIntProperty(
+            DB::Properties::kNumImmutableMemTable, &num_imm_mems));
+        ASSERT_EQ(num_imm_mems, 1);
+      }
+
+      // ingest wbwi
+      const Snapshot* snapshot = txn_db->GetSnapshot();
+      TransactionOptions topts_ingest;
+      topts_ingest.commit_bypass_memtable = true;
+      // reuse txn1
+      txn = txn_db->BeginTransaction(wopts, topts_ingest, txn);
+      ASSERT_OK(txn->Put("k2", "wbwi"));
+      if (single_del) {
+        ASSERT_OK(txn->SingleDelete("k4"));
+        ASSERT_OK(txn->SingleDelete("k5"));
+      } else {
+        ASSERT_OK(txn->Delete("k4"));
+        ASSERT_OK(txn->Delete("k5"));
+      }
+      ASSERT_OK(txn->Put("k6", "wbwi"));
+      ASSERT_OK(txn->SetName("xid2"));
+      ASSERT_OK(txn->Prepare());
+      VerifyDBFromMap(expected, nullptr, false, nullptr, nullptr, &not_found);
+
+      ASSERT_OK(txn->Commit());
+      if (disable_flush) {
+        ASSERT_TRUE(txn_db->GetIntProperty(
+            DB::Properties::kNumImmutableMemTable, &num_imm_mems));
+        // During Commit(), current empty memtable becomes a new immutable
+        // memtable. 3 here includes wbwi for this transaction, the empty
+        // immutable memtable, and the immutable memtable created
+        // by TEST_SwitchMemtable() above.
+        ASSERT_EQ(num_imm_mems, 3);
+      }
+
+      ReadOptions ro_snapshot;
+      ro_snapshot.snapshot = snapshot;
+      auto expected_at_snapshot = expected;
+      auto not_found_at_snapshot = not_found;
+      VerifyDBFromMap(expected, nullptr, false, &ro_snapshot, nullptr,
+                      &not_found);
+      not_found = {"k4", "k5"};
+      expected.erase("k4");
+      expected.erase("k5");
+      expected["k2"] = "wbwi";
+      expected["k6"] = "wbwi";
+      VerifyDBFromMap(expected, nullptr, false, nullptr, nullptr, &not_found);
+
+      // live mem
+      txn = txn_db->BeginTransaction(wopts, topts, txn);
+      ASSERT_OK(txn->Put("k2", "live_mem"));
+      ASSERT_OK(txn->SetName("xid3"));
+      ASSERT_OK(txn->Prepare());
+      ASSERT_OK(txn->Commit());
+      delete txn;
+      expected["k2"] = "live_mem";
+      VerifyDBFromMap(expected, nullptr, false, nullptr, nullptr, &not_found);
+
+      if (disable_flush) {
+        ASSERT_OK(db_->ContinueBackgroundWork());
+        ASSERT_OK(db_->Flush({}));
+      } else {
+        ASSERT_OK(db_->WaitForCompact({}));
+      }
+      ASSERT_TRUE(txn_db->GetIntProperty(DB::Properties::kNumImmutableMemTable,
+                                         &num_imm_mems));
+      ASSERT_EQ(num_imm_mems, 0);
+
+      VerifyDBFromMap(expected, nullptr, false, nullptr, nullptr, &not_found);
+      VerifyDBFromMap(expected_at_snapshot, nullptr, false, &ro_snapshot,
+                      nullptr, &not_found_at_snapshot);
+      txn_db->ReleaseSnapshot(snapshot);
+    }
+  }
+}
+
+TEST_P(CommitBypassMemtableTest, MultiCFOverwrite) {
+  // mini-stress test multi CF update
+  const int kNumKeys = 10000;
+  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 = {"puppy", "kitty", "meta"};
+    CreateColumnFamilies(cfs, options);
+    ASSERT_EQ(handles_.size(), 3);
+    ASSERT_EQ(handles_[0]->GetName(), cfs[0]);
+    ASSERT_EQ(handles_[1]->GetName(), cfs[1]);
+    ASSERT_EQ(handles_[2]->GetName(), cfs[2]);
+
+    std::map<std::string, std::string> expected_cf0;
+    std::unordered_set<std::string> not_found_cf0;
+    std::map<std::string, std::string> expected_cf1;
+    std::unordered_set<std::string> not_found_cf1;
+    WriteOptions wopts;
+    Random* rnd = Random::GetTLSInstance();
+    // Some base data
+    for (int i = 0; i < kNumKeys; ++i) {
+      std::string val_cf0 = "cf0_sst" + std::to_string(i);
+      if (rnd->OneIn(2)) {
+        ASSERT_OK(txn_db->Put(wopts, handles_[0], Key(i), val_cf0));
+        expected_cf0[Key(i)] = val_cf0;
+      } else {
+        not_found_cf0.insert(Key(i));
+      }
+      std::string val_cf1 = "cf1_sst" + std::to_string(i);
+      ASSERT_OK(txn_db->Put(wopts, handles_[1], Key(i), val_cf1));
+      expected_cf1[Key(i)] = val_cf1;
+    }
+    SCOPED_TRACE("Verify after SST");
+    ASSERT_OK(txn_db->Put(wopts, handles_[2], "id", "0"));
+    std::map<std::string, std::string> expected_cf2 = {{"id", "0"}};
+    VerifyDBFromMap(expected_cf0, nullptr, false, nullptr, handles_[0],
+                    &not_found_cf0);
+    VerifyDBFromMap(expected_cf1, nullptr, false, nullptr, handles_[1],
+                    &not_found_cf1);
+
+    if (!disable_flush) {
+      ASSERT_OK(txn_db->Flush({}, handles_[0]));
+      ASSERT_OK(txn_db->Flush({}, handles_[1]));
+    }
+
+    const Snapshot* snapshot = nullptr;
+    std::map<std::string, std::string> expected_cf0_snapshot;
+    std::unordered_set<std::string> not_found_cf0_snapshot;
+    std::map<std::string, std::string> expected_cf1_snapshot;
+    std::unordered_set<std::string> not_found_cf1_snapshot;
+    std::map<std::string, std::string> expected_cf2_snapshot;
+    auto init_snapshot_states = [&]() {
+      snapshot = txn_db->GetSnapshot();
+      expected_cf0_snapshot = expected_cf0;
+      not_found_cf0_snapshot = not_found_cf0;
+      expected_cf1_snapshot = expected_cf1;
+      not_found_cf1_snapshot = not_found_cf1;
+      expected_cf2_snapshot = expected_cf2;
+    };
+    if (rnd->OneIn(4)) {
+      init_snapshot_states();
+    }
+
+    // Transaction 1
+    TransactionOptions topts;
+    topts.commit_bypass_memtable = true;
+    Transaction* txn1 = txn_db->BeginTransaction(wopts, topts);
+    auto fill_txn = [&](Transaction* txn, const std::string& v) {
+      std::vector<int> indices(kNumKeys);
+      std::iota(indices.begin(), indices.end(), 0);
+      RandomShuffle(indices.begin(), indices.end());
+      for (int i : indices) {
+        // CF0 update
+        if (rnd->OneIn(4)) {
+          std::string val_cf0 = "cf0_" + v + std::to_string(i);
+          ASSERT_OK(txn->Put(handles_[0], Key(i), val_cf0));
+          expected_cf0[Key(i)] = val_cf0;
+          not_found_cf0.erase(Key(i));
+        } else if (rnd->OneIn(4)) {
+          ASSERT_OK(txn->Delete(handles_[0], Key(i)));
+          expected_cf0.erase(Key(i));
+          not_found_cf0.insert(Key(i));
+        }
+
+        // CF1 update
+        if (rnd->OneIn(4)) {
+          ASSERT_OK(txn->SingleDelete(handles_[1], Key(i)));
+          expected_cf1.erase(Key(i));
+          not_found_cf1.insert(Key(i));
+        }
+        if (rnd->OneIn(4) &&
+            not_found_cf1.find(Key(i)) != not_found_cf1.end()) {
+          // Can not overwrite when using SD.
+          std::string val_cf1 = "cf1_" + v + std::to_string(i);
+          ASSERT_OK(txn->Put(handles_[1], Key(i), val_cf1));
+          expected_cf1[Key(i)] = val_cf1;
+          not_found_cf1.erase(Key(i));
+        }
+      }
+    };
+    fill_txn(txn1, "txn1");
+    ASSERT_OK(txn1->SetName("xid1"));
+    ASSERT_OK(txn1->Prepare());
+    if (rnd->OneIn(2)) {
+      ASSERT_OK(txn1->GetCommitTimeWriteBatch()->Put(handles_[2], "id", "1"));
+      expected_cf2["id"] = "1";
+    }
+    ASSERT_OK(txn1->Commit());
+    delete txn1;
+
+    uint64_t num_imm_mems = 0;
+    if (disable_flush) {
+      ASSERT_TRUE(txn_db->GetIntProperty(
+          handles_[0], DB::Properties::kNumImmutableMemTable, &num_imm_mems));
+      ASSERT_EQ(num_imm_mems, 2);
+
+      ASSERT_TRUE(txn_db->GetIntProperty(
+          handles_[1], DB::Properties::kNumImmutableMemTable, &num_imm_mems));
+      ASSERT_EQ(num_imm_mems, 2);
+      ASSERT_TRUE(txn_db->GetIntProperty(
+          handles_[2], DB::Properties::kNumImmutableMemTable, &num_imm_mems));
+      ASSERT_EQ(num_imm_mems, 0);
+    }
+    SCOPED_TRACE("Verify cf0 after txn1");
+    VerifyDBFromMap(expected_cf0, nullptr, false, nullptr, handles_[0],
+                    &not_found_cf0);
+    SCOPED_TRACE("Verify cf1 after txn1");
+    SCOPED_TRACE("db is " + txn_db->GetName());
+    VerifyDBFromMap(expected_cf1, nullptr, false, nullptr, handles_[1],
+                    &not_found_cf1);
+    VerifyDBFromMap(expected_cf2, nullptr, false, nullptr, handles_[2]);
+    auto verify_at_snapshot = [&](const std::string& m) {
+      if (snapshot) {
+        SCOPED_TRACE("Verify at snapshot " + m);
+        ReadOptions ro;
+        ro.snapshot = snapshot;
+        SCOPED_TRACE("Verify cf0 after txn1");
+        VerifyDBFromMap(expected_cf0_snapshot, nullptr, false, &ro, handles_[0],
+                        &not_found_cf0_snapshot);
+        SCOPED_TRACE("Verify cf1 after txn1");
+        VerifyDBFromMap(expected_cf1_snapshot, nullptr, false, &ro, handles_[1],
+                        &not_found_cf1_snapshot);
+        VerifyDBFromMap(expected_cf2_snapshot, nullptr, false, &ro,
+                        handles_[2]);
+        if (rnd->OneIn(2)) {
+          txn_db->ReleaseSnapshot(snapshot);
+          snapshot = nullptr;
+        }
+      }
+    };
+    verify_at_snapshot("txn1");
+    if (!snapshot && rnd->OneIn(4)) {
+      init_snapshot_states();
+    }
+
+    // Live memtable or another transaction with commit_bypass_memtable
+    TransactionOptions topts2;
+    topts2.commit_bypass_memtable = rnd->OneIn(2);
+    SCOPED_TRACE("txn2 commit_bypass_memtable = " +
+                 std::to_string(topts2.commit_bypass_memtable));
+    Transaction* txn2 = txn_db->BeginTransaction(wopts, topts2);
+    fill_txn(txn2, "txn2");
+    ASSERT_OK(txn2->SetName("xid2"));
+    ASSERT_OK(txn2->Prepare());
+    if (rnd->OneIn(2)) {
+      ASSERT_OK(txn2->GetCommitTimeWriteBatch()->Put(handles_[2], "id", "2"));
+      expected_cf2["id"] = "2";
+    }
+    ASSERT_OK(txn2->Commit());
+    delete txn2;
+    if (disable_flush) {
+      ASSERT_TRUE(txn_db->GetIntProperty(
+          handles_[0], DB::Properties::kNumImmutableMemTable, &num_imm_mems));
+      ASSERT_EQ(num_imm_mems, topts2.commit_bypass_memtable ? 4 : 2);
+
+      ASSERT_TRUE(txn_db->GetIntProperty(
+          handles_[1], DB::Properties::kNumImmutableMemTable, &num_imm_mems));
+      ASSERT_EQ(num_imm_mems, topts2.commit_bypass_memtable ? 4 : 2);
+      ASSERT_TRUE(txn_db->GetIntProperty(
+          handles_[2], DB::Properties::kNumImmutableMemTable, &num_imm_mems));
+      ASSERT_EQ(num_imm_mems, 0);
+    }
+
+    SCOPED_TRACE("Verify cf0 after txn2");
+    VerifyDBFromMap(expected_cf0, nullptr, false, nullptr, handles_[0],
+                    &not_found_cf0);
+    SCOPED_TRACE("Verify cf1 after txn2");
+    VerifyDBFromMap(expected_cf1, nullptr, false, nullptr, handles_[1],
+                    &not_found_cf1);
+    // cf2 should not have keys in not_found_cf1 either
+    VerifyDBFromMap(expected_cf2, nullptr, false, nullptr, handles_[2],
+                    &not_found_cf1);
+    verify_at_snapshot("txn2");
+    if (!snapshot && rnd->OneIn(4)) {
+      init_snapshot_states();
+    }
+
+    // Verify all data is flushed
+    if (disable_flush) {
+      ASSERT_OK(db_->ContinueBackgroundWork());
+      ASSERT_OK(db_->Flush({}, handles_[0]));
+      ASSERT_OK(db_->Flush({}, handles_[1]));
+    } else {
+      ASSERT_OK(db_->WaitForCompact({}));
+    }
+
+    VerifyDBFromMap(expected_cf0, nullptr, false, nullptr, handles_[0],
+                    &not_found_cf0);
+    VerifyDBFromMap(expected_cf1, nullptr, false, nullptr, handles_[1],
+                    &not_found_cf1);
+    VerifyDBFromMap(expected_cf2, nullptr, false, nullptr, handles_[2],
+                    &not_found_cf1);
+    verify_at_snapshot("flush");
+    if (snapshot) {
+      db_->ReleaseSnapshot(snapshot);
+    }
+  }
+}
+
+TEST_P(CommitBypassMemtableTest, Recovery) {
+  // Test that ingested txn can be recovered.
+  // Implementation detail: this tests that write path reserves enough sequence
+  // number for the ingested memtables (see seq_inc in WriteImpl()). For
+  // example, suppose that we only assign one sequence number per ingested
+  // memtable. Then txn1 will be assigned seqno 1 and txn2 will be assigned 2.
+  // During recovery, we will not use memtable ingestion and will fall back
+  // to memtable insertion. The sequence number for the first key in a
+  // transaction is the commit sequence number. So for txn1, we will insert into
+  // memtable k2@1 and k1@2. And for txn2, we will insert k1@2 and k2@3. This
+  // results in duplicated key@seqno.
+  WriteOptions wopts;
+  TransactionOptions txn_opts;
+  txn_opts.commit_bypass_memtable = true;
+  Transaction* txn1 = txn_db->BeginTransaction(wopts, txn_opts, nullptr);
+  ASSERT_OK(txn1->SetName("xid1"));
+  ASSERT_OK(txn1->Put("k2", "v2"));
+  ASSERT_OK(txn1->Put("k1", "v1"));
+  ASSERT_OK(txn1->Prepare());
+  ASSERT_OK(txn1->Commit());
+
+  // Test txn reuse code path
+  txn1 = txn_db->BeginTransaction(wopts, txn_opts, txn1);
+  ASSERT_OK(txn1->SetName("xid2"));
+  ASSERT_OK(txn1->Put("k1", "v3"));
+  ASSERT_OK(txn1->Put("k2", "v4"));
+  ASSERT_OK(txn1->Prepare());
+  ASSERT_OK(txn1->Commit());
+  delete txn1;
+
+  std::map<std::string, std::string> expected = {{"k1", "v3"}, {"k2", "v4"}};
+  VerifyDBFromMap(expected);
+
+  ASSERT_OK(txn_db->Close());
+  delete txn_db;
+  ASSERT_OK(TransactionDB::Open(options, txn_db_opts, dbname_, &txn_db));
+  db_ = txn_db;
+
+  VerifyDBFromMap(expected);
+}
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

utilities/write_batch_with_index/write_batch_with_index.cc

@@ -35,7 +35,8 @@ struct WriteBatchWithIndex::Rep {
         overwrite_key(_overwrite_key),
         last_entry_offset(0),
         last_sub_batch_offset(0),
-        sub_batch_cnt(1) {}
+        sub_batch_cnt(1),
+        track_cf_stat(false) {}
   ReadableWriteBatch write_batch;
   WriteBatchEntryComparator comparator;
   Arena arena;
@@ -49,6 +50,11 @@ struct WriteBatchWithIndex::Rep {
   // Total number of sub-batches in the write batch. Default is 1.
   size_t sub_batch_cnt;
 
+  bool track_cf_stat;
+  // Tracks ids of CFs that have updates in this WBWI, number of updates and
+  // number of overwritten single deletions per cf.
+  std::unordered_map<uint32_t, CFStat> cf_id_to_stat;
+
   // Remember current offset of internal write batch, which is used as
   // the starting offset of the next record.
   void SetLastEntryOffset() { last_entry_offset = write_batch.GetDataSize(); }
@@ -69,7 +75,7 @@ struct WriteBatchWithIndex::Rep {
 
   // Allocate an index entry pointing to the last entry in the write batch and
   // put it to skip list.
-  void AddNewEntry(uint32_t column_family_id);
+  void AddNewEntry(uint32_t column_family_id, WriteType type);
 
   // Clear all updates buffered in this batch.
   void Clear();
@@ -114,6 +120,16 @@ bool WriteBatchWithIndex::Rep::UpdateExistingEntryWithCfId(
     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) {
+      cf_id_to_stat[column_family_id].overwritten_sd_count++;
+      non_const_entry->has_overwritten_single_del = true;
+    }
+    if (type == kSingleDeleteRecord) {
+      non_const_entry->has_single_del = true;
+    }
+  }
   if (type == kMergeRecord) {
     return false;
   } else {
@@ -130,18 +146,19 @@ void WriteBatchWithIndex::Rep::AddOrUpdateIndex(
     if (cf_cmp != nullptr) {
       comparator.SetComparatorForCF(cf_id, cf_cmp);
     }
-    AddNewEntry(cf_id);
+    AddNewEntry(cf_id, type);
   }
 }
 
 void WriteBatchWithIndex::Rep::AddOrUpdateIndex(const Slice& key,
                                                 WriteType type) {
   if (!UpdateExistingEntryWithCfId(0, key, type)) {
-    AddNewEntry(0);
+    AddNewEntry(0, type);
   }
 }
 
-void WriteBatchWithIndex::Rep::AddNewEntry(uint32_t column_family_id) {
+void WriteBatchWithIndex::Rep::AddNewEntry(uint32_t column_family_id,
+                                           WriteType type) {
   const std::string& wb_data = write_batch.Data();
   Slice entry_ptr = Slice(wb_data.data() + last_entry_offset,
                           wb_data.size() - last_entry_offset);
@@ -166,10 +183,18 @@ void WriteBatchWithIndex::Rep::AddNewEntry(uint32_t column_family_id) {
       new (mem) WriteBatchIndexEntry(last_entry_offset, column_family_id,
                                      key.data() - wb_data.data(), key.size());
   skip_list.Insert(index_entry);
+
+  if (track_cf_stat) {
+    if (type == kSingleDeleteRecord) {
+      index_entry->has_single_del = true;
+    }
+    cf_id_to_stat[column_family_id].entry_count++;
+  }
 }
 
 void WriteBatchWithIndex::Rep::Clear() {
   write_batch.Clear();
+  cf_id_to_stat.clear();
   ClearIndex();
 }
 
@@ -220,7 +245,7 @@ Status WriteBatchWithIndex::Rep::ReBuildIndex() {
       case kTypeValue:
         found++;
         if (!UpdateExistingEntryWithCfId(column_family_id, key, kPutRecord)) {
-          AddNewEntry(column_family_id);
+          AddNewEntry(column_family_id, kPutRecord);
         }
         break;
       case kTypeColumnFamilyDeletion:
@@ -228,7 +253,7 @@ Status WriteBatchWithIndex::Rep::ReBuildIndex() {
         found++;
         if (!UpdateExistingEntryWithCfId(column_family_id, key,
                                          kDeleteRecord)) {
-          AddNewEntry(column_family_id);
+          AddNewEntry(column_family_id, kDeleteRecord);
         }
         break;
       case kTypeColumnFamilySingleDeletion:
@@ -236,14 +261,14 @@ Status WriteBatchWithIndex::Rep::ReBuildIndex() {
         found++;
         if (!UpdateExistingEntryWithCfId(column_family_id, key,
                                          kSingleDeleteRecord)) {
-          AddNewEntry(column_family_id);
+          AddNewEntry(column_family_id, kSingleDeleteRecord);
         }
         break;
       case kTypeColumnFamilyMerge:
       case kTypeMerge:
         found++;
         if (!UpdateExistingEntryWithCfId(column_family_id, key, kMergeRecord)) {
-          AddNewEntry(column_family_id);
+          AddNewEntry(column_family_id, kMergeRecord);
         }
         break;
       case kTypeLogData:
@@ -261,7 +286,7 @@ Status WriteBatchWithIndex::Rep::ReBuildIndex() {
         found++;
         if (!UpdateExistingEntryWithCfId(column_family_id, key,
                                          kPutEntityRecord)) {
-          AddNewEntry(column_family_id);
+          AddNewEntry(column_family_id, kPutEntityRecord);
         }
         break;
       case kTypeColumnFamilyValuePreferredSeqno:
@@ -1138,4 +1163,17 @@ const Comparator* WriteBatchWithIndexInternal::GetUserComparator(
   return ucmps.GetComparator(cf_id);
 }
 
+void WriteBatchWithIndex::SetTrackPerCFStat(bool track) {
+  // Should be set when the wbwi contains no update.
+  assert(GetWriteBatch()->Count() == 0);
+  rep->track_cf_stat = track;
+}
+
+const std::unordered_map<uint32_t, WriteBatchWithIndex::CFStat>&
+WriteBatchWithIndex::GetCFStats() const {
+  assert(rep->track_cf_stat);
+  return rep->cf_id_to_stat;
+}
+
+bool WriteBatchWithIndex::GetOverwriteKey() const { return rep->overwrite_key; }
 }  // namespace ROCKSDB_NAMESPACE

utilities/write_batch_with_index/write_batch_with_index_internal.h

@@ -104,6 +104,8 @@ struct WriteBatchIndexEntry {
   WriteBatchIndexEntry(size_t o, uint32_t c, size_t ko, size_t ksz)
       : offset(o),
         column_family(c),
+        has_single_del(false),
+        has_overwritten_single_del(false),
         key_offset(ko),
         key_size(ksz),
         search_key(nullptr) {}
@@ -121,6 +123,8 @@ struct WriteBatchIndexEntry {
       // entry who has the same search key. Otherwise, we'll miss those entries.
       : offset(is_forward_direction ? 0 : std::numeric_limits<size_t>::max()),
         column_family(_column_family),
+        has_single_del(false),
+        has_overwritten_single_del(false),
         key_offset(0),
         key_size(is_seek_to_first ? kFlagMinInCf : 0),
         search_key(_search_key) {
@@ -147,6 +151,9 @@ struct WriteBatchIndexEntry {
   // 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
   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
@@ -209,7 +216,7 @@ class WriteBatchEntryComparator {
 using WriteBatchEntrySkipList =
     SkipList<WriteBatchIndexEntry*, const WriteBatchEntryComparator&>;
 
-class WBWIIteratorImpl : public WBWIIterator {
+class WBWIIteratorImpl final : public WBWIIterator {
  public:
   enum Result : uint8_t {
     kFound,
@@ -325,6 +332,11 @@ class WBWIIteratorImpl : public WBWIIterator {
 
   WriteEntry Entry() const override;
 
+  bool HasOverWrittenSingleDel() const override {
+    assert(Valid());
+    return skip_list_iter_.key()->has_overwritten_single_del;
+  }
+
   Status status() const override {
     // this is in-memory data structure, so the only way status can be non-ok is
     // through memory corruption

utilities/write_batch_with_index/write_batch_with_index_test.cc

@@ -3417,6 +3417,52 @@ TEST_P(WriteBatchWithIndexTest, EntityReadSanityChecks) {
   }
 }
 
+TEST_P(WriteBatchWithIndexTest, TrackAndClearCFStats) {
+  std::string value;
+  batch_->SetTrackPerCFStat(true);
+  ASSERT_OK(batch_->Put("A", "val"));
+  ASSERT_OK(batch_->SingleDelete("B"));
+
+  ColumnFamilyHandleImplDummy cf1(/*id=*/1, BytewiseComparator());
+  ASSERT_OK(batch_->Put(&cf1, "bar", "foo"));
+
+  {
+    auto& cf_id_to_count = batch_->GetCFStats();
+    ASSERT_EQ(2, cf_id_to_count.size());
+    for (const auto [cf_id, stat] : cf_id_to_count) {
+      if (cf_id == 0) {
+        ASSERT_EQ(2, stat.entry_count);
+        ASSERT_EQ(0, stat.overwritten_sd_count);
+      } else {
+        ASSERT_EQ(cf_id, 1);
+        ASSERT_EQ(1, stat.entry_count);
+        ASSERT_EQ(0, stat.overwritten_sd_count);
+      }
+    }
+  }
+
+  batch_->Clear();
+  ASSERT_TRUE(batch_->GetCFStats().empty());
+
+  // Now do a version with overwritten SD
+  ASSERT_OK(batch_->Put("A", "val"));
+  ASSERT_OK(batch_->SingleDelete("A"));
+  bool overwrite = GetParam();
+  {
+    auto& cf_id_to_count = batch_->GetCFStats();
+    ASSERT_EQ(1, cf_id_to_count.size());
+    ASSERT_EQ(overwrite ? 1 : 2, cf_id_to_count.at(0).entry_count);
+    ASSERT_EQ(0, cf_id_to_count.at(0).overwritten_sd_count);
+  }
+  ASSERT_OK(batch_->Put("A", "new_val"));
+  {
+    auto& cf_id_to_count = batch_->GetCFStats();
+    ASSERT_EQ(1, cf_id_to_count.size());
+    ASSERT_EQ(overwrite ? 1 : 3, cf_id_to_count.at(0).entry_count);
+    ASSERT_EQ(overwrite ? 1 : 0, cf_id_to_count.at(0).overwritten_sd_count);
+  }
+}
+
 INSTANTIATE_TEST_CASE_P(WBWI, WriteBatchWithIndexTest, testing::Bool());
 
 std::string Get(const std::string& k, std::unique_ptr<WBWIMemTable>& wbwi_mem,
@@ -3453,8 +3499,10 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) {
   ImmutableOptions immutable_opts(opts);
   MutableCFOptions mutable_cf_options(opts);
 
-  Random rnd(301);
-  auto wbwi = std::make_shared<WriteBatchWithIndex>(cmp, 0, true, 0, 0);
+  Random& rnd = *Random::GetTLSInstance();
+  auto wbwi = std::make_shared<WriteBatchWithIndex>(
+      cmp, 0, /*overwrite_key=*/true, 0, 0);
+  wbwi->SetTrackPerCFStat(true);
   std::vector<std::pair<std::string, std::string>> expected;
   expected.resize(10000);
   for (int i = 0; i < 10000; ++i) {
@@ -3468,12 +3516,14 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) {
   RandomShuffle(expected.begin(), expected.end());
   std::unique_ptr<WBWIMemTable> wbwi_mem{
       new WBWIMemTable(wbwi, cmp,
-                       /*cf_id=*/0, &immutable_opts, &mutable_cf_options)};
+                       /*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 visible_seq = 3;
+  constexpr SequenceNumber visible_seq = 10002;
   constexpr SequenceNumber non_visible_seq = 1;
-  constexpr SequenceNumber assigned_seq = 2;
-  wbwi_mem->SetGlobalSequenceNumber(assigned_seq);
+  constexpr WBWIMemTable::SeqnoRange assigned_seq = {2, 10001};
+  wbwi_mem->AssignSequenceNumbers(assigned_seq);
 
   bool found_final_value = false;
   for (const auto& [key, val] : expected) {
@@ -3501,7 +3551,7 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) {
   RandomShuffle(expected.begin(), expected.end());
   // overwrites
   for (size_t i = 0; i < 2000; ++i) {
-    // We don't expect mixing SD and DEL, or issue multiple SD consecutively in
+    // We don't expect to mix SD and DEL, or issue multiple SD consecutively in
     // a DB. Read from WBWI should still work so we do it here to keep the test
     // simple.
     if (rnd.OneIn(2)) {
@@ -3540,7 +3590,6 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) {
     ASSERT_TRUE(val == Get(key, wbwi_mem, visible_seq, &found_final_value));
     ASSERT_TRUE(found_final_value);
   }
-
   // MultiGet
   int batch_size = 30;
   for (int i = 0; i < 10000; i += batch_size) {
@@ -3601,9 +3650,9 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) {
               return a.first < b.first;
             });
   Arena arena;
-  InternalIterator* iter =
-      wbwi_mem->NewIterator(ReadOptions(), /*seqno_to_time_mapping=*/nullptr,
-                            &arena, /*prefix_extractor=*/nullptr);
+  InternalIterator* iter = wbwi_mem->NewIterator(
+      ReadOptions(), /*seqno_to_time_mapping=*/nullptr, &arena,
+      /*prefix_extractor=*/nullptr, /*for_flush=*/false);
   ASSERT_OK(iter->status());
 
   auto verify_iter_at = [&](size_t idx) {
@@ -3615,7 +3664,7 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) {
     ValueType val_type;
     UnPackSequenceAndType(ExtractInternalKeyFooter(iter->key()), &seq,
                           &val_type);
-    ASSERT_EQ(seq, assigned_seq);
+    ASSERT_EQ(seq, assigned_seq.upper_bound);
     if (expected[idx].second == "NOT_FOUND") {
       ASSERT_TRUE(val_type == kTypeDeletion || val_type == kTypeSingleDeletion);
     } else {
@@ -3639,14 +3688,20 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) {
         iter->Seek(seek_key.GetInternalKey());
       }
       verify_iter_at(key_idx);
-      uint32_t j = key_idx + 1;
-      for (; j < std::min(10000u, key_idx + 5); ++j) {
+
+      // verify next/prev 5 times
+      for (int j = 0; j < 5; ++j) {
+        if (++key_idx >= 10000) {
+          --key_idx;
+          break;
+        }
         iter->Next();
-        verify_iter_at(j);
+        verify_iter_at(key_idx);
       }
-      for (j -= 2; j >= key_idx; --j) {
+      for (int j = 0; j < 5; ++j) {
         iter->Prev();
-        verify_iter_at(j);
+        assert(key_idx >= 1);
+        verify_iter_at(--key_idx);
       }
     }
   }
@@ -3668,8 +3723,9 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) {
 
   // Read from another CF
   std::unique_ptr<WBWIMemTable> meta_wbwi_mem{new WBWIMemTable(
-      wbwi, cmp, /*cf_id=*/1, &immutable_opts, &mutable_cf_options)};
-  meta_wbwi_mem->SetGlobalSequenceNumber(assigned_seq);
+      wbwi, cmp, /*cf_id=*/1, &immutable_opts, &mutable_cf_options,
+      /*stat=*/{1, 0})};
+  meta_wbwi_mem->AssignSequenceNumbers(assigned_seq);
   found_final_value = false;
   ASSERT_TRUE("foo" == Get(DBTestBase::Key(0), meta_wbwi_mem, visible_seq,
                            &found_final_value));
@@ -3682,7 +3738,8 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) {
   wbwi_mem.reset();
   // allocated by arena
   iter->~InternalIterator();
-  iter = meta_wbwi_mem->NewIterator(ReadOptions(), nullptr, &arena, nullptr);
+  iter = meta_wbwi_mem->NewIterator(ReadOptions(), nullptr, &arena, nullptr,
+                                    /*for_flush=*/false);
   iter->SeekToFirst();
   ASSERT_OK(iter->status());
   ASSERT_TRUE(iter->Valid());
@@ -3693,6 +3750,64 @@ TEST_F(WBWIMemTableTest, ReadFromWBWIMemtable) {
   ASSERT_FALSE(iter->Valid());
   iter->~InternalIterator();
 }
+
+TEST_F(WBWIMemTableTest, IterEmitSingleDelete) {
+  const Comparator* cmp = BytewiseComparator();
+  Options opts;
+  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);
+
+  ASSERT_OK(wbwi->Put(DBTestBase::Key(0), "val0"));
+  ASSERT_OK(wbwi->SingleDelete(DBTestBase::Key(0)));
+  ASSERT_OK(wbwi->SingleDelete(DBTestBase::Key(1)));
+  ASSERT_OK(wbwi->SingleDelete(DBTestBase::Key(2)));
+  ASSERT_OK(wbwi->Put(DBTestBase::Key(3), "val3"));
+  // SD at key1 overwritten
+  ASSERT_OK(wbwi->Put(DBTestBase::Key(1), "val1"));
+
+  std::unique_ptr<WBWIMemTable> wbwi_mem{
+      new WBWIMemTable(wbwi, cmp,
+                       /*cf_id=*/0, &immutable_opts, &mutable_cf_options,
+                       /*stat=*/wbwi->GetCFStats().at(0))};
+  WBWIMemTable::SeqnoRange assigned_seqno = {
+      1, 1 + wbwi->GetWriteBatch()->Count()};
+  wbwi_mem->AssignSequenceNumbers(assigned_seqno);
+  Arena arena;
+  InternalIterator* iter_for_flush = wbwi_mem->NewIterator(
+      ReadOptions(), /*seqno_to_time_mapping=*/nullptr, &arena,
+      /*prefix_extractor=*/nullptr, /*for_flush=*/true);
+  InternalIterator* iter = wbwi_mem->NewIterator(
+      ReadOptions(), /*seqno_to_time_mapping=*/nullptr, &arena,
+      /*prefix_extractor=*/nullptr, /*for_flush=*/false);
+  iter->SeekToFirst();
+  iter_for_flush->SeekToFirst();
+  for (int i = 0; i < 4; ++i) {
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_TRUE(iter_for_flush->Valid());
+    ASSERT_EQ(iter->key(), iter_for_flush->key());
+
+    iter->Next();
+    iter_for_flush->Next();
+    if (i == 1) {
+      // overwritten SD at key1
+      // See WBWIMemTableIterator::UpdateSingleDeleteKey() for seqno assignment
+      InternalKey ikey(DBTestBase::Key(1), assigned_seqno.upper_bound - 1,
+                       kTypeSingleDeletion);
+      ASSERT_EQ(ikey.Encode(), iter_for_flush->key());
+      iter_for_flush->Next();
+    }
+  }
+  ASSERT_FALSE(iter->Valid());
+  ASSERT_FALSE(iter_for_flush->Valid());
+  ASSERT_OK(iter->status());
+  ASSERT_OK(iter_for_flush->status());
+  iter->~InternalIterator();
+  iter_for_flush->~InternalIteratorBase();
+}
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {