Change SingleDelete to support conflict checking

Summary: For Transactions, we want to start using the SST files to do write conflict checking.  To do this, we need to make sure that compaction never removes all writes if an earlier snapshot exists.  So I had to change the way we process SingleDeletes to sometimes leave a SingleDelete behind when we encounter a Put followed by a SingleDelete.  See the comments in this diff for a more detailed explanation.

Test Plan: added more unit tests

Reviewers: rven, igor, kradhakrishnan, IslamAbdelRahman, yhchiang, sdong

Reviewed By: sdong

Subscribers: dhruba, leveldb

Differential Revision: https://reviews.facebook.net/D50295

db/compaction_iterator.cc

@@ -97,6 +97,11 @@ void CompactionIterator::Next() {
     NextFromInput();
   }
 
+  if (valid_) {
+    // Record that we've ouputted a record for the current key.
+    has_outputted_key_ = true;
+  }
+
   PrepareOutput();
 }
 
@@ -143,8 +148,10 @@ void CompactionIterator::NextFromInput() {
       key_ = current_key_.SetKey(key_, &ikey_);
       current_user_key_ = ikey_.user_key;
       has_current_user_key_ = true;
+      has_outputted_key_ = false;
       current_user_key_sequence_ = kMaxSequenceNumber;
       current_user_key_snapshot_ = 0;
+
       // apply the compaction filter to the first occurrence of the user key
       if (compaction_filter_ != nullptr && ikey_.type == kTypeValue &&
           (visible_at_tip_ || ikey_.sequence > latest_snapshot_ ||
@@ -198,36 +205,99 @@ void CompactionIterator::NextFromInput() {
         visible_at_tip_ ? visible_at_tip_ : findEarliestVisibleSnapshot(
                                                 ikey_.sequence, &prev_snapshot);
 
-    if (ikey_.type == kTypeSingleDeletion) {
+    if (clear_and_output_next_key_) {
+      // In the previous iteration we encountered a single delete that we could
+      // not compact out.  We will keep this Put, but can drop it's data.
+      // (See Optimization 3, below.)
+      assert(ikey_.type == kTypeValue);
+      assert(current_user_key_snapshot_ == last_snapshot);
+
+      value_.clear();
+      valid_ = true;
+      clear_and_output_next_key_ = false;
+    } else if (ikey_.type == kTypeSingleDeletion) {
+      // We can compact out a SingleDelete if:
+      // 1) We encounter the corresponding PUT -OR- we know that this key
+      //    doesn't appear past this output level
+      // =AND=
+      // 2) We've already returned a record in this snapshot -OR-
+      //    there are no earlier earliest_write_conflict_snapshot.
+      //
+      // Rule 1 is needed for SingleDelete correctness.  Rule 2 is needed to
+      // allow Transactions to do write-conflict checking (if we compacted away
+      // all keys, then we wouldn't know that a write happened in this
+      // snapshot).  If there is no earlier snapshot, then we know that there
+      // are no active transactions that need to know about any writes.
+      //
+      // Optimization 3:
+      // If we encounter a SingleDelete followed by a PUT and Rule 2 is NOT
+      // true, then we must output a SingleDelete.  In this case, we will decide
+      // to also output the PUT.  While we are compacting less by outputting the
+      // PUT now, hopefully this will lead to better compaction in the future
+      // when Rule 2 is later true (Ie, We are hoping we can later compact out
+      // both the SingleDelete and the Put, while we couldn't if we only
+      // outputted the SingleDelete now).
+      // In this case, we can save space by removing the PUT's value as it will
+      // never be read.
+      //
+      // Deletes and Merges are not supported on the same key that has a
+      // SingleDelete as it is not possible to correctly do any partial
+      // compaction of such a combination of operations.  The result of mixing
+      // those operations for a given key is documented as being undefined.  So
+      // we can choose how to handle such a combinations of operations.  We will
+      // try to compact out as much as we can in these cases.
+
+      // The easiest way to process a SingleDelete during iteration is to peek
+      // ahead at the next key.
       ParsedInternalKey next_ikey;
       input_->Next();
 
-      if (earliest_write_conflict_snapshot_) {
+      // Check whether the next key exists, is not corrupt, and is the same key
-        // TODO(agiardullo): to be used in D50295
-        // adding this if statement to keep CLANG happy in the meantime
-      }
-
-      // Check whether the current key is valid, not corrupt and the same
       // as the single delete.
       if (input_->Valid() && ParseInternalKey(input_->key(), &next_ikey) &&
           cmp_->Equal(ikey_.user_key, next_ikey.user_key)) {
-        // Mixing single deletes and merges is not supported. Consecutive
+        // Check whether the next key belongs to the same snapshot as the
-        // single deletes are not valid.
+        // SingleDelete.
-        if (next_ikey.type != kTypeValue) {
-          assert(false);
-          status_ =
-              Status::InvalidArgument("Put expected after single delete.");
-          break;
-        }
-
-        // Check whether the current key belongs to the same snapshot as the
-        // single delete.
         if (prev_snapshot == 0 || next_ikey.sequence > prev_snapshot) {
-          // Found the matching value, we can drop the single delete and the
+          if (next_ikey.type == kTypeSingleDeletion) {
-          // value.
+            // We encountered two SingleDeletes in a row.  This could be due to
-          ++iter_stats_.num_record_drop_hidden;
+            // unexpected user input.
-          ++iter_stats_.num_record_drop_obsolete;
+            // Skip the first SingleDelete and let the next iteration decide how
-          input_->Next();
+            // to handle the second SingleDelete
+
+            // First SingleDelete has been skipped since we already called
+            // input_->Next().
+            ++iter_stats_.num_record_drop_obsolete;
+          } else if ((ikey_.sequence <= earliest_write_conflict_snapshot_) ||
+                     has_outputted_key_) {
+            // Found a matching value, we can drop the single delete and the
+            // value.  It is safe to drop both records since we've already
+            // outputted a key in this snapshot, or there is no earlier
+            // snapshot (Rule 2 above).
+
+            // Note: it doesn't matter whether the second key is a Put or if it
+            // is an unexpected Merge or Delete.  We will compact it out
+            // either way.
+            ++iter_stats_.num_record_drop_hidden;
+            ++iter_stats_.num_record_drop_obsolete;
+            // Already called input_->Next() once.  Call it a second time to
+            // skip past the second key.
+            input_->Next();
+          } else {
+            // Found a matching value, but we cannot drop both keys since
+            // there is an earlier snapshot and we need to leave behind a record
+            // to know that a write happened in this snapshot (Rule 2 above).
+            // Clear the value and output the SingleDelete. (The value will be
+            // outputted on the next iteration.)
+            ++iter_stats_.num_record_drop_hidden;
+
+            // Setting valid_ to true will output the current SingleDelete
+            valid_ = true;
+
+            // Set up the Put to be outputted in the next iteration.
+            // (Optimization 3).
+            clear_and_output_next_key_ = true;
+          }
         } else {
           // We hit the next snapshot without hitting a put, so the iterator
           // returns the single delete.
@@ -242,11 +312,14 @@ void CompactionIterator::NextFromInput() {
         // iteration. If the next key is corrupt, we return before the
         // comparison, so the value of has_current_user_key does not matter.
         has_current_user_key_ = false;
-        if (compaction_ != nullptr &&
+        if (compaction_ != nullptr && ikey_.sequence <= earliest_snapshot_ &&
             compaction_->KeyNotExistsBeyondOutputLevel(ikey_.user_key,
                                                        &level_ptrs_)) {
+          // Key doesn't exist outside of this range.
+          // Can compact out this SingleDelete.
           ++iter_stats_.num_record_drop_obsolete;
         } else {
+          // Output SingleDelete
           valid_ = true;
         }
       }

db/compaction_iterator.h

@@ -127,6 +127,14 @@ class CompactionIterator {
   Slice current_user_key_;
   SequenceNumber current_user_key_sequence_;
   SequenceNumber current_user_key_snapshot_;
+
+  // True if the iterator has already returned a record for the current key.
+  bool has_outputted_key_ = false;
+
+  // truncated the value of the next key and output it without applying any
+  // compaction rules.  This is used for outputting a put after a single delete.
+  bool clear_and_output_next_key_ = false;
+
   MergeOutputIterator merge_out_iter_;
   std::string compaction_filter_value_;
   // "level_ptrs" holds indices that remember which file of an associated

db/compaction_job_test.cc

@@ -217,9 +217,11 @@ class CompactionJobTest : public testing::Test {
     cfd_ = versions_->GetColumnFamilySet()->GetDefault();
   }
 
-  void RunCompaction(const std::vector<std::vector<FileMetaData*>>& input_files,
+  void RunCompaction(
-                     const stl_wrappers::KVMap& expected_results,
+      const std::vector<std::vector<FileMetaData*>>& input_files,
-                     const std::vector<SequenceNumber>& snapshots = {}) {
+      const stl_wrappers::KVMap& expected_results,
+      const std::vector<SequenceNumber>& snapshots = {},
+      SequenceNumber earliest_write_conflict_snapshot = kMaxSequenceNumber) {
     auto cfd = versions_->GetColumnFamilySet()->GetDefault();
 
     size_t num_input_files = 0;
@@ -246,8 +248,8 @@ class CompactionJobTest : public testing::Test {
     CompactionJob compaction_job(
         0, &compaction, db_options_, env_options_, versions_.get(),
         &shutting_down_, &log_buffer, nullptr, nullptr, nullptr, snapshots,
-        kMaxSequenceNumber, table_cache_, &event_logger, false, false, dbname_,
+        earliest_write_conflict_snapshot, table_cache_, &event_logger, false,
-        &compaction_job_stats_);
+        false, dbname_, &compaction_job_stats_);
 
     VerifyInitializationOfCompactionJobStats(compaction_job_stats_);
 
@@ -559,34 +561,144 @@ TEST_F(CompactionJobTest, SimpleSingleDelete) {
 TEST_F(CompactionJobTest, SingleDeleteSnapshots) {
   NewDB();
 
-  auto file1 = mock::MakeMockFile({{KeyStr("A", 12U, kTypeSingleDeletion), ""},
+  auto file1 = mock::MakeMockFile({
-                                   {KeyStr("a", 12U, kTypeSingleDeletion), ""},
+      {KeyStr("A", 12U, kTypeSingleDeletion), ""},
-                                   {KeyStr("b", 21U, kTypeSingleDeletion), ""},
+      {KeyStr("a", 12U, kTypeSingleDeletion), ""},
-                                   {KeyStr("c", 22U, kTypeSingleDeletion), ""},
+      {KeyStr("b", 21U, kTypeSingleDeletion), ""},
-                                   {KeyStr("d", 9U, kTypeSingleDeletion), ""}});
+      {KeyStr("c", 22U, kTypeSingleDeletion), ""},
+      {KeyStr("d", 9U, kTypeSingleDeletion), ""},
+      {KeyStr("f", 21U, kTypeSingleDeletion), ""},
+      {KeyStr("j", 11U, kTypeSingleDeletion), ""},
+      {KeyStr("j", 9U, kTypeSingleDeletion), ""},
+      {KeyStr("k", 12U, kTypeSingleDeletion), ""},
+      {KeyStr("k", 11U, kTypeSingleDeletion), ""},
+      {KeyStr("l", 3U, kTypeSingleDeletion), ""},
+      {KeyStr("l", 2U, kTypeSingleDeletion), ""},
+  });
   AddMockFile(file1);
 
-  auto file2 = mock::MakeMockFile({{KeyStr("0", 2U, kTypeSingleDeletion), ""},
+  auto file2 = mock::MakeMockFile({
-                                   {KeyStr("a", 11U, kTypeValue), "val1"},
+      {KeyStr("0", 2U, kTypeSingleDeletion), ""},
-                                   {KeyStr("b", 11U, kTypeValue), "val2"},
+      {KeyStr("a", 11U, kTypeValue), "val1"},
-                                   {KeyStr("c", 21U, kTypeValue), "val3"},
+      {KeyStr("b", 11U, kTypeValue), "val2"},
-                                   {KeyStr("d", 8U, kTypeValue), "val4"},
+      {KeyStr("c", 21U, kTypeValue), "val3"},
-                                   {KeyStr("e", 2U, kTypeSingleDeletion), ""}});
+      {KeyStr("d", 8U, kTypeValue), "val4"},
+      {KeyStr("e", 2U, kTypeSingleDeletion), ""},
+      {KeyStr("f", 1U, kTypeValue), "val1"},
+      {KeyStr("g", 11U, kTypeSingleDeletion), ""},
+      {KeyStr("h", 2U, kTypeSingleDeletion), ""},
+      {KeyStr("m", 12U, kTypeValue), "val1"},
+      {KeyStr("m", 11U, kTypeSingleDeletion), ""},
+      {KeyStr("m", 8U, kTypeValue), "val2"},
+  });
   AddMockFile(file2);
 
-  auto file3 = mock::MakeMockFile({{KeyStr("A", 1U, kTypeValue), "val"},
+  auto file3 = mock::MakeMockFile({
-                                   {KeyStr("e", 1U, kTypeValue), "val"}});
+      {KeyStr("A", 1U, kTypeValue), "val"},
+      {KeyStr("e", 1U, kTypeValue), "val"},
+  });
   AddMockFile(file3, 2);
 
-  auto expected_results =
+  auto expected_results = mock::MakeMockFile({
-      mock::MakeMockFile({{KeyStr("A", 12U, kTypeSingleDeletion), ""},
+      {KeyStr("A", 12U, kTypeSingleDeletion), ""},
-                          {KeyStr("b", 21U, kTypeSingleDeletion), ""},
+      {KeyStr("a", 12U, kTypeSingleDeletion), ""},
-                          {KeyStr("b", 11U, kTypeValue), "val2"},
+      {KeyStr("a", 11U, kTypeValue), ""},
-                          {KeyStr("e", 2U, kTypeSingleDeletion), ""}});
+      {KeyStr("b", 21U, kTypeSingleDeletion), ""},
+      {KeyStr("b", 11U, kTypeValue), "val2"},
+      {KeyStr("c", 22U, kTypeSingleDeletion), ""},
+      {KeyStr("c", 21U, kTypeValue), ""},
+      {KeyStr("e", 2U, kTypeSingleDeletion), ""},
+      {KeyStr("f", 21U, kTypeSingleDeletion), ""},
+      {KeyStr("f", 1U, kTypeValue), "val1"},
+      {KeyStr("g", 11U, kTypeSingleDeletion), ""},
+      {KeyStr("j", 11U, kTypeSingleDeletion), ""},
+      {KeyStr("k", 11U, kTypeSingleDeletion), ""},
+      {KeyStr("m", 12U, kTypeValue), "val1"},
+      {KeyStr("m", 11U, kTypeSingleDeletion), ""},
+      {KeyStr("m", 8U, kTypeValue), "val2"},
+  });
 
   SetLastSequence(22U);
   auto files = cfd_->current()->storage_info()->LevelFiles(0);
-  RunCompaction({files}, expected_results, {10U, 20U});
+  RunCompaction({files}, expected_results, {10U, 20U}, 10U);
+}
+
+TEST_F(CompactionJobTest, EarliestWriteConflictSnapshot) {
+  NewDB();
+
+  // Test multiple snapshots where the earliest snapshot is not a
+  // write-conflic-snapshot.
+
+  auto file1 = mock::MakeMockFile({
+      {KeyStr("A", 24U, kTypeSingleDeletion), ""},
+      {KeyStr("A", 23U, kTypeValue), "val"},
+      {KeyStr("B", 24U, kTypeSingleDeletion), ""},
+      {KeyStr("B", 23U, kTypeValue), "val"},
+      {KeyStr("D", 24U, kTypeSingleDeletion), ""},
+      {KeyStr("G", 32U, kTypeSingleDeletion), ""},
+      {KeyStr("G", 31U, kTypeValue), "val"},
+      {KeyStr("G", 24U, kTypeSingleDeletion), ""},
+      {KeyStr("G", 23U, kTypeValue), "val2"},
+      {KeyStr("H", 31U, kTypeValue), "val"},
+      {KeyStr("H", 24U, kTypeSingleDeletion), ""},
+      {KeyStr("H", 23U, kTypeValue), "val"},
+      {KeyStr("I", 35U, kTypeSingleDeletion), ""},
+      {KeyStr("I", 34U, kTypeValue), "val2"},
+      {KeyStr("I", 33U, kTypeSingleDeletion), ""},
+      {KeyStr("I", 32U, kTypeValue), "val3"},
+      {KeyStr("I", 31U, kTypeSingleDeletion), ""},
+      {KeyStr("J", 34U, kTypeValue), "val"},
+      {KeyStr("J", 33U, kTypeSingleDeletion), ""},
+      {KeyStr("J", 25U, kTypeValue), "val2"},
+      {KeyStr("J", 24U, kTypeSingleDeletion), ""},
+  });
+  AddMockFile(file1);
+
+  auto file2 = mock::MakeMockFile({
+      {KeyStr("A", 14U, kTypeSingleDeletion), ""},
+      {KeyStr("A", 13U, kTypeValue), "val2"},
+      {KeyStr("C", 14U, kTypeSingleDeletion), ""},
+      {KeyStr("C", 13U, kTypeValue), "val"},
+      {KeyStr("E", 12U, kTypeSingleDeletion), ""},
+      {KeyStr("F", 4U, kTypeSingleDeletion), ""},
+      {KeyStr("F", 3U, kTypeValue), "val"},
+      {KeyStr("G", 14U, kTypeSingleDeletion), ""},
+      {KeyStr("G", 13U, kTypeValue), "val3"},
+      {KeyStr("H", 14U, kTypeSingleDeletion), ""},
+      {KeyStr("H", 13U, kTypeValue), "val2"},
+      {KeyStr("I", 13U, kTypeValue), "val4"},
+      {KeyStr("I", 12U, kTypeSingleDeletion), ""},
+      {KeyStr("I", 11U, kTypeValue), "val5"},
+      {KeyStr("J", 15U, kTypeValue), "val3"},
+      {KeyStr("J", 14U, kTypeSingleDeletion), ""},
+  });
+  AddMockFile(file2);
+
+  auto expected_results = mock::MakeMockFile({
+      {KeyStr("A", 24U, kTypeSingleDeletion), ""},
+      {KeyStr("A", 23U, kTypeValue), ""},
+      {KeyStr("B", 24U, kTypeSingleDeletion), ""},
+      {KeyStr("B", 23U, kTypeValue), ""},
+      {KeyStr("D", 24U, kTypeSingleDeletion), ""},
+      {KeyStr("E", 12U, kTypeSingleDeletion), ""},
+      {KeyStr("G", 32U, kTypeSingleDeletion), ""},
+      {KeyStr("G", 31U, kTypeValue), ""},
+      {KeyStr("H", 31U, kTypeValue), "val"},
+      {KeyStr("I", 35U, kTypeSingleDeletion), ""},
+      {KeyStr("I", 34U, kTypeValue), ""},
+      {KeyStr("I", 31U, kTypeSingleDeletion), ""},
+      {KeyStr("I", 13U, kTypeValue), "val4"},
+      {KeyStr("J", 34U, kTypeValue), "val"},
+      {KeyStr("J", 33U, kTypeSingleDeletion), ""},
+      {KeyStr("J", 25U, kTypeValue), "val2"},
+      {KeyStr("J", 24U, kTypeSingleDeletion), ""},
+      {KeyStr("J", 15U, kTypeValue), "val3"},
+      {KeyStr("J", 14U, kTypeSingleDeletion), ""},
+  });
+
+  SetLastSequence(24U);
+  auto files = cfd_->current()->storage_info()->LevelFiles(0);
+  RunCompaction({files}, expected_results, {10U, 20U, 30U}, 20U);
 }
 
 TEST_F(CompactionJobTest, SingleDeleteZeroSeq) {
@@ -616,9 +728,22 @@ TEST_F(CompactionJobTest, MultiSingleDelete) {
   // Tests three scenarios involving multiple single delete/put pairs:
   //
   // A: Put Snapshot SDel Put SDel -> Put Snapshot SDel
-  // B: Put SDel Put SDel -> (Removed)
+  // B: Snapshot Put SDel Put SDel Snapshot -> Snapshot SDel Snapshot
   // C: SDel Put SDel Snapshot Put -> Snapshot Put
-  // D: (Put) SDel Snapshot Put SDel -> (Put) SDel Snapshot
+  // D: (Put) SDel Snapshot Put SDel -> (Put) SDel Snapshot SDel
+  // E: Put SDel Snapshot Put SDel -> Snapshot SDel
+  // F: Put SDel Put Sdel Snapshot -> removed
+  // G: Snapshot SDel Put SDel Put -> Snapshot Put SDel
+  // H: (Put) Put SDel Put Sdel Snapshot -> Removed
+  // I: (Put) Snapshot Put SDel Put SDel -> SDel
+  // J: Put Put SDel Put SDel SDel Snapshot Put Put SDel SDel Put
+  //      -> Snapshot Put
+  // K: SDel SDel Put SDel Put Put Snapshot SDel Put SDel SDel Put SDel
+  //      -> Snapshot Put Snapshot SDel
+  // L: SDel Put Del Put SDel Snapshot Del Put Del SDel Put SDel
+  //      -> Snapshot SDel
+  // M: (Put) SDel Put Del Put SDel Snapshot Put Del SDel Put SDel Del
+  //      -> SDel Snapshot Del
   NewDB();
 
   auto file1 = mock::MakeMockFile({
@@ -630,6 +755,34 @@ TEST_F(CompactionJobTest, MultiSingleDelete) {
       {KeyStr("C", 14U, kTypeValue), "val3"},
       {KeyStr("D", 12U, kTypeSingleDeletion), ""},
       {KeyStr("D", 11U, kTypeValue), "val4"},
+      {KeyStr("G", 15U, kTypeValue), "val"},
+      {KeyStr("G", 14U, kTypeSingleDeletion), ""},
+      {KeyStr("G", 13U, kTypeValue), "val"},
+      {KeyStr("I", 14U, kTypeSingleDeletion), ""},
+      {KeyStr("I", 13U, kTypeValue), "val"},
+      {KeyStr("J", 15U, kTypeValue), "val"},
+      {KeyStr("J", 14U, kTypeSingleDeletion), ""},
+      {KeyStr("J", 13U, kTypeSingleDeletion), ""},
+      {KeyStr("J", 12U, kTypeValue), "val"},
+      {KeyStr("J", 11U, kTypeValue), "val"},
+      {KeyStr("K", 16U, kTypeSingleDeletion), ""},
+      {KeyStr("K", 15U, kTypeValue), "val1"},
+      {KeyStr("K", 14U, kTypeSingleDeletion), ""},
+      {KeyStr("K", 13U, kTypeSingleDeletion), ""},
+      {KeyStr("K", 12U, kTypeValue), "val2"},
+      {KeyStr("K", 11U, kTypeSingleDeletion), ""},
+      {KeyStr("L", 16U, kTypeSingleDeletion), ""},
+      {KeyStr("L", 15U, kTypeValue), "val"},
+      {KeyStr("L", 14U, kTypeSingleDeletion), ""},
+      {KeyStr("L", 13U, kTypeDeletion), ""},
+      {KeyStr("L", 12U, kTypeValue), "val"},
+      {KeyStr("L", 11U, kTypeDeletion), ""},
+      {KeyStr("M", 16U, kTypeDeletion), ""},
+      {KeyStr("M", 15U, kTypeSingleDeletion), ""},
+      {KeyStr("M", 14U, kTypeValue), "val"},
+      {KeyStr("M", 13U, kTypeSingleDeletion), ""},
+      {KeyStr("M", 12U, kTypeDeletion), ""},
+      {KeyStr("M", 11U, kTypeValue), "val"},
   });
   AddMockFile(file1);
 
@@ -641,24 +794,87 @@ TEST_F(CompactionJobTest, MultiSingleDelete) {
       {KeyStr("C", 9U, kTypeValue), "val6"},
       {KeyStr("C", 8U, kTypeSingleDeletion), ""},
       {KeyStr("D", 10U, kTypeSingleDeletion), ""},
+      {KeyStr("E", 12U, kTypeSingleDeletion), ""},
+      {KeyStr("E", 11U, kTypeValue), "val"},
+      {KeyStr("E", 5U, kTypeSingleDeletion), ""},
+      {KeyStr("E", 4U, kTypeValue), "val"},
+      {KeyStr("F", 6U, kTypeSingleDeletion), ""},
+      {KeyStr("F", 5U, kTypeValue), "val"},
+      {KeyStr("F", 4U, kTypeSingleDeletion), ""},
+      {KeyStr("F", 3U, kTypeValue), "val"},
+      {KeyStr("G", 12U, kTypeSingleDeletion), ""},
+      {KeyStr("H", 6U, kTypeSingleDeletion), ""},
+      {KeyStr("H", 5U, kTypeValue), "val"},
+      {KeyStr("H", 4U, kTypeSingleDeletion), ""},
+      {KeyStr("H", 3U, kTypeValue), "val"},
+      {KeyStr("I", 12U, kTypeSingleDeletion), ""},
+      {KeyStr("I", 11U, kTypeValue), "val"},
+      {KeyStr("J", 6U, kTypeSingleDeletion), ""},
+      {KeyStr("J", 5U, kTypeSingleDeletion), ""},
+      {KeyStr("J", 4U, kTypeValue), "val"},
+      {KeyStr("J", 3U, kTypeSingleDeletion), ""},
+      {KeyStr("J", 2U, kTypeValue), "val"},
+      {KeyStr("K", 8U, kTypeValue), "val3"},
+      {KeyStr("K", 7U, kTypeValue), "val4"},
+      {KeyStr("K", 6U, kTypeSingleDeletion), ""},
+      {KeyStr("K", 5U, kTypeValue), "val5"},
+      {KeyStr("K", 2U, kTypeSingleDeletion), ""},
+      {KeyStr("K", 1U, kTypeSingleDeletion), ""},
+      {KeyStr("L", 5U, kTypeSingleDeletion), ""},
+      {KeyStr("L", 4U, kTypeValue), "val"},
+      {KeyStr("L", 3U, kTypeDeletion), ""},
+      {KeyStr("L", 2U, kTypeValue), "val"},
+      {KeyStr("L", 1U, kTypeSingleDeletion), ""},
+      {KeyStr("M", 10U, kTypeSingleDeletion), ""},
+      {KeyStr("M", 7U, kTypeValue), "val"},
+      {KeyStr("M", 5U, kTypeDeletion), ""},
+      {KeyStr("M", 4U, kTypeValue), "val"},
+      {KeyStr("M", 3U, kTypeSingleDeletion), ""},
   });
   AddMockFile(file2);
 
   auto file3 = mock::MakeMockFile({
-      {KeyStr("D", 11U, kTypeValue), "val"},
+      {KeyStr("D", 1U, kTypeValue), "val"},
+      {KeyStr("H", 1U, kTypeValue), "val"},
+      {KeyStr("I", 2U, kTypeValue), "val"},
   });
   AddMockFile(file3, 2);
 
-  auto expected_results = mock::MakeMockFile({
+  auto file4 = mock::MakeMockFile({
-      {KeyStr("A", 12U, kTypeSingleDeletion), ""},
+      {KeyStr("M", 1U, kTypeValue), "val"},
-      {KeyStr("A", 10U, kTypeValue), "val"},
-      {KeyStr("C", 14U, kTypeValue), "val3"},
-      {KeyStr("D", 10U, kTypeSingleDeletion), ""},
   });
+  AddMockFile(file4, 2);
+
+  auto expected_results =
+      mock::MakeMockFile({{KeyStr("A", 14U, kTypeSingleDeletion), ""},
+                          {KeyStr("A", 13U, kTypeValue), ""},
+                          {KeyStr("A", 12U, kTypeSingleDeletion), ""},
+                          {KeyStr("A", 10U, kTypeValue), "val"},
+                          {KeyStr("B", 14U, kTypeSingleDeletion), ""},
+                          {KeyStr("B", 13U, kTypeValue), ""},
+                          {KeyStr("C", 14U, kTypeValue), "val3"},
+                          {KeyStr("D", 12U, kTypeSingleDeletion), ""},
+                          {KeyStr("D", 11U, kTypeValue), ""},
+                          {KeyStr("D", 10U, kTypeSingleDeletion), ""},
+                          {KeyStr("E", 12U, kTypeSingleDeletion), ""},
+                          {KeyStr("E", 11U, kTypeValue), ""},
+                          {KeyStr("G", 15U, kTypeValue), "val"},
+                          {KeyStr("G", 12U, kTypeSingleDeletion), ""},
+                          {KeyStr("I", 14U, kTypeSingleDeletion), ""},
+                          {KeyStr("I", 13U, kTypeValue), ""},
+                          {KeyStr("J", 15U, kTypeValue), "val"},
+                          {KeyStr("K", 16U, kTypeSingleDeletion), ""},
+                          {KeyStr("K", 15U, kTypeValue), ""},
+                          {KeyStr("K", 11U, kTypeSingleDeletion), ""},
+                          {KeyStr("K", 8U, kTypeValue), "val3"},
+                          {KeyStr("L", 16U, kTypeSingleDeletion), ""},
+                          {KeyStr("L", 15U, kTypeValue), ""},
+                          {KeyStr("M", 16U, kTypeDeletion), ""},
+                          {KeyStr("M", 3U, kTypeSingleDeletion), ""}});
 
   SetLastSequence(22U);
   auto files = cfd_->current()->storage_info()->LevelFiles(0);
-  RunCompaction({files}, expected_results, {10U});
+  RunCompaction({files}, expected_results, {10U}, 10U);
 }
 
 // This test documents the behavior where a corrupt key follows a deletion or a

db/db_impl.cc

@@ -3813,9 +3813,11 @@ Status DBImpl::NewIterators(
 
 const Snapshot* DBImpl::GetSnapshot() { return GetSnapshotImpl(false); }
 
+#ifndef ROCKSDB_LITE
 const Snapshot* DBImpl::GetSnapshotForWriteConflictBoundary() {
   return GetSnapshotImpl(true);
 }
+#endif  // ROCKSDB_LITE
 
 const Snapshot* DBImpl::GetSnapshotImpl(bool is_write_conflict_boundary) {
   int64_t unix_time = 0;