SERVER-43860 Always upsert exact source document for pipeline-insert $merge

1 files changed, 129 insertions, 100 deletions

diff --git a/src/mongo/db/pipeline/document_source_merge_test.cpp b/src/mongo/db/pipeline/document_source_merge_test.cpp
index f46df8bd212..b05a30e1507 100644
--- a/src/mongo/db/pipeline/document_source_merge_test.cpp
+++ b/src/mongo/db/pipeline/document_source_merge_test.cpp
@@ -729,20 +729,23 @@ TEST_F(DocumentSourceMergeTest, LetVariablesCanOnlyBeUsedWithPipelineMode) {
     ASSERT_THROWS_CODE(createMergeStage(spec), AssertionException, 51199);
 }
 
+// We always serialize the default let variables as {new: "$$ROOT"} if omitted.
 TEST_F(DocumentSourceMergeTest, SerializeDefaultLetVariable) {
-    auto spec =
-        BSON("$merge" << BSON("into"
-                              << "target_collection"
-                              << "whenMatched" << BSON_ARRAY(BSON("$project" << BSON("x" << 1)))
-                              << "whenNotMatched"
-                              << "insert"));
-    auto mergeStage = createMergeStage(spec);
-    auto serialized = mergeStage->serialize().getDocument();
-    ASSERT_VALUE_EQ(serialized["$merge"]["let"],
-                    Value(BSON("new"
-                               << "$$ROOT")));
+    for (auto&& whenNotMatched : {"insert", "fail", "discard"}) {
+        auto spec =
+            BSON("$merge" << BSON("into"
+                                  << "target_collection"
+                                  << "whenMatched" << BSON_ARRAY(BSON("$project" << BSON("x" << 1)))
+                                  << "whenNotMatched" << whenNotMatched));
+        auto mergeStage = createMergeStage(spec);
+        auto serialized = mergeStage->serialize().getDocument();
+        ASSERT_VALUE_EQ(serialized["$merge"]["let"],
+                        Value(BSON("new"
+                                   << "$$ROOT")));
+    }
 }
 
+// Test the behaviour of 'let' serialization for each whenNotMatched mode.
 TEST_F(DocumentSourceMergeTest, SerializeLetVariables) {
     auto pipeline = BSON_ARRAY(BSON("$project" << BSON("x"
                                                        << "$$v1"
@@ -750,50 +753,70 @@ TEST_F(DocumentSourceMergeTest, SerializeLetVariables) {
                                                        << "$$v2"
                                                        << "z"
                                                        << "$$v3")));
-    auto spec = BSON("$merge" << BSON("into"
-                                      << "target_collection"
-                                      << "let"
-                                      << BSON("v1" << 10 << "v2"
-                                                   << "foo"
-                                                   << "v3"
-                                                   << BSON("x" << 1 << "y"
-                                                               << BSON("z"
-                                                                       << "bar")))
-                                      << "whenMatched" << pipeline << "whenNotMatched"
-                                      << "insert"));
-    auto mergeStage = createMergeStage(spec);
-    ASSERT(mergeStage);
-    auto serialized = mergeStage->serialize().getDocument();
-    ASSERT_VALUE_EQ(serialized["$merge"]["let"]["v1"], Value(BSON("$const" << 10)));
-    ASSERT_VALUE_EQ(serialized["$merge"]["let"]["v2"],
-                    Value(BSON("$const"
-                               << "foo")));
-    ASSERT_VALUE_EQ(serialized["$merge"]["let"]["v3"],
-                    Value(BSON("x" << BSON("$const" << 1) << "y"
-                                   << BSON("z" << BSON("$const"
-                                                       << "bar")))));
-    ASSERT_VALUE_EQ(serialized["$merge"]["whenMatched"], Value(pipeline));
+
+    const auto createAndSerializeMergeStage = [this, &pipeline](StringData whenNotMatched) {
+        auto spec = BSON("$merge" << BSON("into"
+                                          << "target_collection"
+                                          << "let"
+                                          << BSON("v1" << 10 << "v2"
+                                                       << "foo"
+                                                       << "v3"
+                                                       << BSON("x" << 1 << "y"
+                                                                   << BSON("z"
+                                                                           << "bar")))
+                                          << "whenMatched" << pipeline << "whenNotMatched"
+                                          << whenNotMatched));
+        auto mergeStage = createMergeStage(spec);
+        ASSERT(mergeStage);
+
+        return mergeStage->serialize().getDocument();
+    };
+
+    for (auto&& whenNotMatched : {"insert", "fail", "discard"}) {
+        const auto serialized = createAndSerializeMergeStage(whenNotMatched);
+
+        // For {whenNotMatched:insert}, we always attach the 'new' document even if the user has
+        // already specified a set of variables. This is because a {whenNotMatched: insert} merge
+        // generates an upsert, and if no documents in the target collection match the query we must
+        // insert the original document. For other 'whenNotMatched' modes, we do not serialize the
+        // new document, since neither 'fail' nor 'discard' can result in an upsert.
+        ASSERT_VALUE_EQ(serialized["$merge"]["let"]["new"],
+                        (whenNotMatched == "insert"_sd ? Value("$$ROOT"_sd) : Value()));
+
+        // The user's variables should be serialized in all cases.
+        ASSERT_VALUE_EQ(serialized["$merge"]["let"]["v1"], Value(BSON("$const" << 10)));
+        ASSERT_VALUE_EQ(serialized["$merge"]["let"]["v2"],
+                        Value(BSON("$const"
+                                   << "foo")));
+        ASSERT_VALUE_EQ(serialized["$merge"]["let"]["v3"],
+                        Value(BSON("x" << BSON("$const" << 1) << "y"
+                                       << BSON("z" << BSON("$const"
+                                                           << "bar")))));
+        ASSERT_VALUE_EQ(serialized["$merge"]["whenMatched"], Value(pipeline));
+    }
 }
 
 TEST_F(DocumentSourceMergeTest, SerializeLetArrayVariable) {
-    auto pipeline = BSON_ARRAY(BSON("$project" << BSON("x"
-                                                       << "$$v1")));
-    auto spec =
-        BSON("$merge" << BSON("into"
-                              << "target_collection"
-                              << "let"
-                              << BSON("v1" << BSON_ARRAY(1 << "2" << BSON("x" << 1 << "y" << 2)))
-                              << "whenMatched" << pipeline << "whenNotMatched"
-                              << "insert"));
-    auto mergeStage = createMergeStage(spec);
-    ASSERT(mergeStage);
-    auto serialized = mergeStage->serialize().getDocument();
-    ASSERT_VALUE_EQ(serialized["$merge"]["let"]["v1"],
-                    Value(BSON_ARRAY(BSON("$const" << 1) << BSON("$const"
-                                                                 << "2")
-                                                         << BSON("x" << BSON("$const" << 1) << "y"
-                                                                     << BSON("$const" << 2)))));
-    ASSERT_VALUE_EQ(serialized["$merge"]["whenMatched"], Value(pipeline));
+    for (auto&& whenNotMatched : {"insert", "fail", "discard"}) {
+        auto pipeline = BSON_ARRAY(BSON("$project" << BSON("x"
+                                                           << "$$v1")));
+        auto spec = BSON(
+            "$merge" << BSON("into"
+                             << "target_collection"
+                             << "let"
+                             << BSON("v1" << BSON_ARRAY(1 << "2" << BSON("x" << 1 << "y" << 2)))
+                             << "whenMatched" << pipeline << "whenNotMatched" << whenNotMatched));
+        auto mergeStage = createMergeStage(spec);
+        ASSERT(mergeStage);
+        auto serialized = mergeStage->serialize().getDocument();
+        ASSERT_VALUE_EQ(
+            serialized["$merge"]["let"]["v1"],
+            Value(BSON_ARRAY(BSON("$const" << 1)
+                             << BSON("$const"
+                                     << "2")
+                             << BSON("x" << BSON("$const" << 1) << "y" << BSON("$const" << 2)))));
+        ASSERT_VALUE_EQ(serialized["$merge"]["whenMatched"], Value(pipeline));
+    }
 }
 
 // This test verifies that when the 'let' argument is specified as 'null', the default 'new'
@@ -803,60 +826,66 @@ TEST_F(DocumentSourceMergeTest, SerializeLetArrayVariable) {
 // this test ensures that we're aware of this limitation. Once the limitation is addressed in
 // SERVER-41272, this test should be updated to accordingly.
 TEST_F(DocumentSourceMergeTest, SerializeNullLetVariablesAsDefault) {
-    auto pipeline = BSON_ARRAY(BSON("$project" << BSON("x"
-                                                       << "1")));
-    auto spec =
-        BSON("$merge" << BSON("into"
-                              << "target_collection"
-                              << "let" << BSONNULL << "whenMatched" << pipeline << "whenNotMatched"
-                              << "insert"));
-    auto mergeStage = createMergeStage(spec);
-    ASSERT(mergeStage);
-    auto serialized = mergeStage->serialize().getDocument();
-    ASSERT_VALUE_EQ(serialized["$merge"]["let"],
-                    Value(BSON("new"
-                               << "$$ROOT")));
-    ASSERT_VALUE_EQ(serialized["$merge"]["whenMatched"], Value(pipeline));
+    for (auto&& whenNotMatched : {"insert", "fail", "discard"}) {
+        auto pipeline = BSON_ARRAY(BSON("$project" << BSON("x"
+                                                           << "1")));
+        auto spec = BSON("$merge" << BSON("into"
+                                          << "target_collection"
+                                          << "let" << BSONNULL << "whenMatched" << pipeline
+                                          << "whenNotMatched" << whenNotMatched));
+        auto mergeStage = createMergeStage(spec);
+        ASSERT(mergeStage);
+        auto serialized = mergeStage->serialize().getDocument();
+        ASSERT_VALUE_EQ(serialized["$merge"]["let"],
+                        Value(BSON("new"
+                                   << "$$ROOT")));
+        ASSERT_VALUE_EQ(serialized["$merge"]["whenMatched"], Value(pipeline));
+    }
 }
 
 TEST_F(DocumentSourceMergeTest, SerializeEmptyLetVariables) {
-    auto pipeline = BSON_ARRAY(BSON("$project" << BSON("x"
-                                                       << "1")));
-    auto spec =
-        BSON("$merge" << BSON("into"
-                              << "target_collection"
-                              << "let" << BSONObj() << "whenMatched" << pipeline << "whenNotMatched"
-                              << "insert"));
-    auto mergeStage = createMergeStage(spec);
-    ASSERT(mergeStage);
-    auto serialized = mergeStage->serialize().getDocument();
-    ASSERT_VALUE_EQ(serialized["$merge"]["let"], Value(BSONObj()));
-    ASSERT_VALUE_EQ(serialized["$merge"]["whenMatched"], Value(pipeline));
+    for (auto&& whenNotMatched : {"insert", "fail", "discard"}) {
+        auto pipeline = BSON_ARRAY(BSON("$project" << BSON("x"
+                                                           << "1")));
+        auto spec = BSON("$merge" << BSON("into"
+                                          << "target_collection"
+                                          << "let" << BSONObj() << "whenMatched" << pipeline
+                                          << "whenNotMatched" << whenNotMatched));
+        auto mergeStage = createMergeStage(spec);
+        ASSERT(mergeStage);
+        auto serialized = mergeStage->serialize().getDocument();
+        ASSERT_VALUE_EQ(serialized["$merge"]["let"],
+                        (whenNotMatched == "insert"_sd ? Value(BSON("new"
+                                                                    << "$$ROOT"))
+                                                       : Value(BSONObj())));
+        ASSERT_VALUE_EQ(serialized["$merge"]["whenMatched"], Value(pipeline));
+    }
 }
 
 TEST_F(DocumentSourceMergeTest, OnlyObjectCanBeUsedAsLetVariables) {
-    auto pipeline = BSON_ARRAY(BSON("$project" << BSON("x"
-                                                       << "1")));
-    auto spec = BSON("$merge" << BSON("into"
-                                      << "target_collection"
-                                      << "let" << 1 << "whenMatched" << pipeline << "whenNotMatched"
-                                      << "insert"));
-    ASSERT_THROWS_CODE(createMergeStage(spec), AssertionException, ErrorCodes::TypeMismatch);
-
-    spec = BSON("$merge" << BSON("into"
-                                 << "target_collection"
-                                 << "let"
-                                 << "foo"
-                                 << "whenMatched" << pipeline << "whenNotMatched"
-                                 << "insert"));
-    ASSERT_THROWS_CODE(createMergeStage(spec), AssertionException, ErrorCodes::TypeMismatch);
-
-    spec = BSON("$merge" << BSON("into"
-                                 << "target_collection"
-                                 << "let" << BSON_ARRAY(1 << "2") << "whenMatched" << pipeline
-                                 << "whenNotMatched"
-                                 << "insert"));
-    ASSERT_THROWS_CODE(createMergeStage(spec), AssertionException, ErrorCodes::TypeMismatch);
+    for (auto&& whenNotMatched : {"insert", "fail", "discard"}) {
+        auto pipeline = BSON_ARRAY(BSON("$project" << BSON("x"
+                                                           << "1")));
+        auto spec = BSON("$merge" << BSON("into"
+                                          << "target_collection"
+                                          << "let" << 1 << "whenMatched" << pipeline
+                                          << "whenNotMatched" << whenNotMatched));
+        ASSERT_THROWS_CODE(createMergeStage(spec), AssertionException, ErrorCodes::TypeMismatch);
+
+        spec = BSON("$merge" << BSON("into"
+                                     << "target_collection"
+                                     << "let"
+                                     << "foo"
+                                     << "whenMatched" << pipeline << "whenNotMatched"
+                                     << whenNotMatched));
+        ASSERT_THROWS_CODE(createMergeStage(spec), AssertionException, ErrorCodes::TypeMismatch);
+
+        spec = BSON("$merge" << BSON("into"
+                                     << "target_collection"
+                                     << "let" << BSON_ARRAY(1 << "2") << "whenMatched" << pipeline
+                                     << "whenNotMatched" << whenNotMatched));
+        ASSERT_THROWS_CODE(createMergeStage(spec), AssertionException, ErrorCodes::TypeMismatch);
+    }
 }
 
 }  // namespace