SERVER-40429 Add merge stage to write output to existing collection

6 files changed, 87 insertions, 16 deletions

diff --git a/jstests/aggregation/sources/explain_out.js b/jstests/aggregation/sources/explain_out.js
index bec7a83765e..9304afcec67 100644
--- a/jstests/aggregation/sources/explain_out.js
+++ b/jstests/aggregation/sources/explain_out.js
@@ -44,7 +44,7 @@
             explain: {aggregate: sourceColl.getName(), pipeline: [outStage], cursor: {}},
             verbosity: verbosity
         }),
-                                     51029);
+                                     [51029, 51184]);
         assert.eq(targetColl.find().itcount(), 0);
     }
 
diff --git a/jstests/aggregation/sources/merge/all_modes.js b/jstests/aggregation/sources/merge/all_modes.js
new file mode 100644
index 00000000000..bb21b24029f
--- /dev/null
+++ b/jstests/aggregation/sources/merge/all_modes.js
@@ -0,0 +1,71 @@
+// Tests basic use cases for all $merge modes.
+(function() {
+    "use strict";
+
+    load("jstests/aggregation/extras/utils.js");  // For arrayEq.
+
+    // Asserts that two arrays are equal - that is, if their sizes are equal and each element in
+    // the 'actual' array has a matching element in the 'expected' array, without honoring
+    // elements order.
+    function assertArrayEq({actual = [], expected = []} = {}) {
+        assert(arrayEq(actual, expected), `actual=${tojson(actual)}, expected=${tojson(expected)}`);
+    }
+
+    const source = db.all_modes_source;
+    const target = db.all_modes_target;
+
+    (function setup() {
+        source.drop();
+        target.drop();
+
+        // All tests use the same data in the source collection.
+        assert.commandWorked(source.insert(
+            [{_id: 1, a: 1, b: "a"}, {_id: 2, a: 2, b: "b"}, {_id: 3, a: 3, b: "c"}]));
+
+    })();
+
+    // Test 'whenMatched=fail whenNotMatched=insert' mode. This is an equivalent of a
+    // replacemnt-style update with upsert=true.
+    (function testWhenMatchedReplaceWithNewWhenNotMatchedInsert() {
+        assert.commandWorked(target.insert([{_id: 1, a: 10}, {_id: 3, a: 30}, {_id: 4, a: 40}]));
+        assert.doesNotThrow(() => source.aggregate([{
+            $merge: {
+                into: target.getName(),
+                whenMatched: "replaceWithNew",
+                whenNotMatched: "insert"
+            }
+        }]));
+        assertArrayEq({
+            actual: target.find().toArray(),
+            expected: [
+                {_id: 1, a: 1, b: "a"},
+                {_id: 2, a: 2, b: "b"},
+                {_id: 3, a: 3, b: "c"},
+                {_id: 4, a: 40}
+            ]
+        });
+    })();
+
+    // Test 'whenMatched=fail whenNotMatched=insert' mode. For matched documents the update should
+    // be unordered and report an error at the end when all documents in a batch have been
+    // processed, it will not fail as soon as we hit the first document without a match.
+    (function testWhenMatchedFailWhenNotMatchedInsert() {
+        assert(target.drop());
+        assert.commandWorked(target.insert(
+            [{_id: 10, a: 10, c: "x"}, {_id: 3, a: 30, c: "y"}, {_id: 4, a: 40, c: "z"}]));
+        const error = assert.throws(() => source.aggregate([
+            {$merge: {into: target.getName(), whenMatched: "fail", whenNotMatched: "insert"}}
+        ]));
+        assert.commandFailedWithCode(error, ErrorCodes.DuplicateKey);
+        assertArrayEq({
+            actual: target.find().toArray(),
+            expected: [
+                {_id: 1, a: 1, b: "a"},
+                {_id: 2, a: 2, b: "b"},
+                {_id: 3, a: 30, c: "y"},
+                {_id: 4, a: 40, c: "z"},
+                {_id: 10, a: 10, c: "x"}
+            ]
+        });
+    })();
+}());
diff --git a/jstests/aggregation/sources/out/exchange_explain.js b/jstests/aggregation/sources/out/exchange_explain.js
index 84a5b02588f..ebf85336677 100644
--- a/jstests/aggregation/sources/out/exchange_explain.js
+++ b/jstests/aggregation/sources/out/exchange_explain.js
@@ -105,7 +105,7 @@ load('jstests/aggregation/extras/utils.js');
                            mode: "replaceDocuments"
                        }
                     }],
-                    50905);
+                    51132);
 
     // Turn off the exchange and rerun the query.
     assert.commandWorked(mongosDB.adminCommand({setParameter: 1, internalQueryDisableExchange: 1}));
@@ -126,7 +126,7 @@ load('jstests/aggregation/extras/utils.js');
                            mode: "replaceDocuments"
                        }
                     }],
-                    50905);
+                    51132);
 
     // SERVER-38349 Make sure mongos rejects specifying exchange directly.
     assert.commandFailedWithCode(mongosDB.runCommand({
diff --git a/jstests/aggregation/sources/out/mode_replace_documents.js b/jstests/aggregation/sources/out/mode_replace_documents.js
index 4281b2b7c08..269d3ca2109 100644
--- a/jstests/aggregation/sources/out/mode_replace_documents.js
+++ b/jstests/aggregation/sources/out/mode_replace_documents.js
@@ -97,7 +97,7 @@
     assertErrorCode(
         coll,
         [{$out: {to: outColl.getName(), mode: "replaceDocuments", uniqueKey: {missing: 1}}}],
-        50905  // This attempt should fail because there's no field 'missing' in the document.
+        51132  // This attempt should fail because there's no field 'missing' in the document.
         );
 
     // Test that a replace fails to insert a document if it violates a unique index constraint. In
@@ -123,7 +123,7 @@
           {$addFields: {_id: 0}},
           {$out: {to: outColl.getName(), mode: "replaceDocuments", uniqueKey: {_id: 1, "a.b": 1}}}
         ],
-        50905);
+        51132);
 
     coll.drop();
     assert.commandWorked(coll.insert({_id: 0, a: [{b: 1}]}));
@@ -133,7 +133,7 @@
           {$addFields: {_id: 0}},
           {$out: {to: outColl.getName(), mode: "replaceDocuments", uniqueKey: {_id: 1, "a.b": 1}}}
         ],
-        50905);
+        51132);
 
     // Tests for $out to a database that differs from the aggregation database.
     const foreignDb = db.getSiblingDB("mode_replace_documents_foreign");
diff --git a/jstests/aggregation/sources/out/unique_key_requires_index.js b/jstests/aggregation/sources/out/unique_key_requires_index.js
index 5d42626e7a1..6448fcd5074 100644
--- a/jstests/aggregation/sources/out/unique_key_requires_index.js
+++ b/jstests/aggregation/sources/out/unique_key_requires_index.js
@@ -282,7 +282,7 @@
                 ],
                 cursor: {}
             }),
-                                         50905);
+                                         [50905, 51132]);
         });
     }());
 
diff --git a/jstests/aggregation/sources/out/unique_key_validation.js b/jstests/aggregation/sources/out/unique_key_validation.js
index 9af0486f7dd..73d3a49c29b 100644
--- a/jstests/aggregation/sources/out/unique_key_validation.js
+++ b/jstests/aggregation/sources/out/unique_key_validation.js
@@ -66,15 +66,15 @@
         [{$out: {to: target.getName(), mode: "replaceDocuments", uniqueKey: {name: 1, team: 1}}}];
 
     // Missing both "name" and "team".
-    assertErrorCode(source, pipelineNameTeam, 50905);
+    assertErrorCode(source, pipelineNameTeam, 51132);
 
     // Missing "name".
     assert.commandWorked(source.update({_id: 0}, {_id: 0, team: "query"}));
-    assertErrorCode(source, pipelineNameTeam, 50905);
+    assertErrorCode(source, pipelineNameTeam, 51132);
 
     // Missing "team".
     assert.commandWorked(source.update({_id: 0}, {_id: 0, name: "nicholas"}));
-    assertErrorCode(source, pipelineNameTeam, 50905);
+    assertErrorCode(source, pipelineNameTeam, 51132);
 
     // A document with both "name" and "team" will be accepted.
     assert.commandWorked(source.update({_id: 0}, {_id: 0, name: "nicholas", team: "query"}));
@@ -92,19 +92,19 @@
 
     // Explicit null "song" (a prefix of a "uniqueKey" field).
     assert.commandWorked(source.update({_id: 0}, {_id: 0, song: null}));
-    assertErrorCode(source, pipelineSongDotArtist, 50905);
+    assertErrorCode(source, pipelineSongDotArtist, 51132);
 
     // Explicit undefined "song" (a prefix of a "uniqueKey" field).
     assert.commandWorked(source.update({_id: 0}, {_id: 0, song: undefined}));
-    assertErrorCode(source, pipelineSongDotArtist, 50905);
+    assertErrorCode(source, pipelineSongDotArtist, 51132);
 
     // Explicit null "song.artist".
     assert.commandWorked(source.update({_id: 0}, {_id: 0, song: {artist: null}}));
-    assertErrorCode(source, pipelineSongDotArtist, 50905);
+    assertErrorCode(source, pipelineSongDotArtist, 51132);
 
     // Explicit undefined "song.artist".
     assert.commandWorked(source.update({_id: 0}, {_id: 0, song: {artist: undefined}}));
-    assertErrorCode(source, pipelineSongDotArtist, 50905);
+    assertErrorCode(source, pipelineSongDotArtist, 51132);
 
     // A valid "artist" will be accepted.
     assert.commandWorked(source.update({_id: 0}, {_id: 0, song: {artist: "Illenium"}}));
@@ -124,11 +124,11 @@
     // "address.street" is an array.
     assert.commandWorked(
         source.update({_id: 0}, {_id: 0, address: {street: ["West 43rd St", "1633 Broadway"]}}));
-    assertErrorCode(source, pipelineAddressDotStreet, 50943);
+    assertErrorCode(source, pipelineAddressDotStreet, 51185);
 
     // "address" is an array (a prefix of a "uniqueKey" field).
     assert.commandWorked(source.update({_id: 0}, {_id: 0, address: [{street: "1633 Broadway"}]}));
-    assertErrorCode(source, pipelineAddressDotStreet, 50905);
+    assertErrorCode(source, pipelineAddressDotStreet, 51132);
 
     // A scalar "address.street" is accepted.
     assert.commandWorked(source.update({_id: 0}, {_id: 0, address: {street: "1633 Broadway"}}));