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

@@ -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) {

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();