From b44cbbf7094c6341eefc0f04782c877e93fd1902 Mon Sep 17 00:00:00 2001
From: Andrew Kryczka <andrewkr@fb.com>
Date: Thu, 26 Jan 2023 15:11:19 -0800
Subject: [PATCH] Fix GetMergeOperands() returning MergeInProgress (#11136)

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

Test Plan: the provided unit test used to fail due to `GetMergeOperands()` returning `Status::MergeInProgress()`; it passes now because the `GetMergeOperands()` call returns `Status::OK()`

Reviewed By: pdillinger

Differential Revision: D42759198

Pulled By: ajkr

fbshipit-source-id: 878f9f40ccc1d7e2fe7b1352814bae3a49c19939
---
 db/db_merge_operand_test.cc | 42 +++++++++++++++++++++++++++++++++++++
 db/memtable.cc              |  5 +++++
 2 files changed, 47 insertions(+)

diff --git a/db/db_merge_operand_test.cc b/db/db_merge_operand_test.cc
index cbec37138..629d3923f 100644
--- a/db/db_merge_operand_test.cc
+++ b/db/db_merge_operand_test.cc
@@ -439,6 +439,48 @@ TEST_F(DBMergeOperandTest, GetMergeOperandsLargeResultOptimization) {
   }
 }
 
+TEST_F(DBMergeOperandTest, GetMergeOperandsBaseDeletionInImmMem) {
+  // In this test, "k1" has a MERGE in a mutable memtable on top of a base
+  // DELETE in an immutable memtable.
+  Options opts = CurrentOptions();
+  opts.max_write_buffer_number = 10;
+  opts.min_write_buffer_number_to_merge = 10;
+  opts.merge_operator = MergeOperators::CreateDeprecatedPutOperator();
+  Reopen(opts);
+
+  ASSERT_OK(Put("k1", "val"));
+  ASSERT_OK(Flush());
+
+  ASSERT_OK(Put("k0", "val"));
+  ASSERT_OK(Delete("k1"));
+  ASSERT_OK(Put("k2", "val"));
+  ASSERT_OK(dbfull()->TEST_SwitchMemtable());
+  ASSERT_OK(Merge("k1", "val"));
+
+  {
+    std::vector<PinnableSlice> values(2);
+
+    GetMergeOperandsOptions merge_operands_info;
+    merge_operands_info.expected_max_number_of_operands =
+        static_cast<int>(values.size());
+
+    std::string key = "k1", from_db;
+    int number_of_operands = 0;
+    ASSERT_OK(db_->GetMergeOperands(ReadOptions(), db_->DefaultColumnFamily(),
+                                    key, values.data(), &merge_operands_info,
+                                    &number_of_operands));
+    ASSERT_EQ(1, number_of_operands);
+    from_db = values[0].ToString();
+    ASSERT_EQ("val", from_db);
+  }
+
+  {
+    std::string val;
+    ASSERT_OK(db_->Get(ReadOptions(), "k1", &val));
+    ASSERT_EQ("val", val);
+  }
+}
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {
diff --git a/db/memtable.cc b/db/memtable.cc
index 1ea12d8c6..943bd73c8 100644
--- a/db/memtable.cc
+++ b/db/memtable.cc
@@ -1207,6 +1207,11 @@ static bool SaveValue(void* arg, const char* entry) {
                 s->columns->SetPlainValue(result);
               }
             }
+          } else {
+            // We have found a final value (a base deletion) and have newer
+            // merge operands that we do not intend to merge. Nothing remains
+            // to be done so assign status to OK.
+            *(s->status) = Status::OK();
           }
         } else {
           *(s->status) = Status::NotFound();