SERVER-35895: Add ability for $out to write to remote hosts

6 files changed, 156 insertions, 12 deletions

diff --git a/jstests/aggregation/sources/out/batch_writes.js b/jstests/aggregation/sources/out/batch_writes.js
index 6bbaac7737a..941b795d27d 100644
--- a/jstests/aggregation/sources/out/batch_writes.js
+++ b/jstests/aggregation/sources/out/batch_writes.js
@@ -41,14 +41,12 @@
     // Test that both batched updates and inserts will successfully write the first document but
     // fail on the second. We don't guarantee any particular behavior in this case, but this test is
     // meant to characterize the current behavior.
-    assertErrorCode(coll, [{$out: {to: outColl.getName(), mode: "insertDocuments"}}], 16996);
-    assert.soon(() => {
-        return outColl.find().itcount() == 2;
-    });
-
-    assertErrorCode(coll, [{$out: {to: outColl.getName(), mode: "replaceDocuments"}}], 50904);
-    assert.soon(() => {
-        return outColl.find().itcount() == 2;
+    ["insertDocuments", "replaceDocuments"].forEach(mode => {
+        assertErrorCode(
+            coll, [{$out: {to: outColl.getName(), mode: mode}}], ErrorCodes.DuplicateKey);
+        assert.soon(() => {
+            return outColl.find().itcount() == 2;
+        });
     });
 
     // Mode "replaceCollection" will drop the contents of the output collection, so there is no
diff --git a/jstests/aggregation/sources/out/mode_insert_documents.js b/jstests/aggregation/sources/out/mode_insert_documents.js
index b6918424a9c..c15818d8011 100644
--- a/jstests/aggregation/sources/out/mode_insert_documents.js
+++ b/jstests/aggregation/sources/out/mode_insert_documents.js
@@ -32,7 +32,7 @@
     //
     // Test that $out fails if there's a duplicate key error.
     //
-    assertErrorCode(coll, pipeline, 16996);
+    assertErrorCode(coll, pipeline, ErrorCodes.DuplicateKey);
 
     //
     // Test that $out will preserve the indexes and options of the output collection.
@@ -61,6 +61,6 @@
     targetColl.drop();
     assert.commandWorked(targetColl.createIndex({a: 1}, {unique: true}));
 
-    assertErrorCode(coll, pipeline, 16996);
+    assertErrorCode(coll, pipeline, ErrorCodes.DuplicateKey);
 
 }());
diff --git a/jstests/aggregation/sources/out/mode_replace_collection.js b/jstests/aggregation/sources/out/mode_replace_collection.js
index e7e770c6d5b..af41753f3cd 100644
--- a/jstests/aggregation/sources/out/mode_replace_collection.js
+++ b/jstests/aggregation/sources/out/mode_replace_collection.js
@@ -52,7 +52,7 @@
     targetColl.drop();
     assert.commandWorked(targetColl.createIndex({a: 1}, {unique: true}));
 
-    assertErrorCode(coll, pipeline, 16996);
+    assertErrorCode(coll, pipeline, ErrorCodes.DuplicateKey);
 
     // Rerun a similar test, except populate the target collection with a document that conflics
     // with one out of the pipeline. In this case, there is no unique key violation since the target
diff --git a/jstests/aggregation/sources/out/mode_replace_documents.js b/jstests/aggregation/sources/out/mode_replace_documents.js
index 8335981a1ff..a147688adda 100644
--- a/jstests/aggregation/sources/out/mode_replace_documents.js
+++ b/jstests/aggregation/sources/out/mode_replace_documents.js
@@ -68,7 +68,7 @@
     assertErrorCode(
         coll,
         [{$addFields: {a: 0}}, {$out: {to: outColl.getName(), mode: "replaceDocuments"}}],
-        50904);
+        ErrorCodes.DuplicateKey);
 
     // Test that $out fails if the unique key contains an array.
     coll.drop();
diff --git a/jstests/sharding/out_to_existing.js b/jstests/sharding/out_to_existing.js
new file mode 100644
index 00000000000..fe3e17fc9e5
--- /dev/null
+++ b/jstests/sharding/out_to_existing.js
@@ -0,0 +1,96 @@
+// Tests for $out with an existing target collection.
+(function() {
+    "use strict";
+
+    load("jstests/aggregation/extras/utils.js");  // For assertErrorCode.
+
+    const st = new ShardingTest({shards: 2, rs: {nodes: 1}});
+
+    const mongosDB = st.s0.getDB(jsTestName());
+    const mongosColl = mongosDB[jsTestName()];
+    const mongosTargetColl = mongosDB[jsTestName() + "_out"];
+
+    function testOut(sourceColl, targetColl, shardedSource, shardedTarget) {
+        jsTestLog("Testing $out from source collection (sharded: " + shardedSource +
+                  ") to target collection (sharded: " + shardedTarget + ")");
+        sourceColl.drop();
+        targetColl.drop();
+
+        if (shardedSource) {
+            st.shardColl(sourceColl, {_id: 1}, {_id: 0}, {_id: 1}, mongosDB.getName());
+        }
+
+        if (shardedTarget) {
+            st.shardColl(targetColl, {_id: 1}, {_id: 0}, {_id: 1}, mongosDB.getName());
+        }
+
+        for (let i = 0; i < 10; i++) {
+            assert.commandWorked(sourceColl.insert({_id: i}));
+        }
+
+        // Test mode "insertDocuments" to an existing collection with no documents.
+        sourceColl.aggregate([{$out: {to: targetColl.getName(), mode: "insertDocuments"}}]);
+        assert.eq(10, targetColl.find().itcount());
+
+        // Test mode "insertDocuments" to an existing collection with unique key conflicts.
+        assertErrorCode(sourceColl,
+                        [{$out: {to: targetColl.getName(), mode: "insertDocuments"}}],
+                        ErrorCodes.DuplicateKey);
+
+        // Test mode "replaceDocuments" to an existing collection with no documents.
+        targetColl.remove({});
+        sourceColl.aggregate([{$out: {to: targetColl.getName(), mode: "replaceDocuments"}}]);
+        assert.eq(10, targetColl.find().itcount());
+
+        // Test mode "replaceDocuments" to an existing collection with documents that match the
+        // unique key. Re-run the previous aggregation, expecting it to succeed and overwrite the
+        // existing documents because the mode is "replaceDocuments".
+        sourceColl.aggregate([{$out: {to: targetColl.getName(), mode: "replaceDocuments"}}]);
+        assert.eq(10, targetColl.find().itcount());
+
+        // Replace all documents in the target collection with documents that don't conflict with
+        // the insert operations.
+        targetColl.remove({});
+        for (let i = 10; i < 20; i++) {
+            assert.commandWorked(targetColl.insert({_id: i}));
+        }
+
+        sourceColl.aggregate([{$out: {to: targetColl.getName(), mode: "insertDocuments"}}]);
+        assert.eq(20, targetColl.find().itcount());
+
+        // Test that mode "replaceCollection" will drop the target collection and replace with the
+        // contents of the $out.
+        // TODO SERVER-36123: Mode "replaceCollection" should fail (gracefully) if the target exists
+        // and is sharded.
+        if (!shardedTarget) {
+            sourceColl.aggregate([{$out: {to: targetColl.getName(), mode: "replaceCollection"}}]);
+            assert.eq(10, targetColl.find().itcount());
+
+            // Legacy syntax should behave identical to mode "replaceCollection".
+            sourceColl.aggregate([{$out: targetColl.getName()}]);
+            assert.eq(10, targetColl.find().itcount());
+        }
+    }
+
+    //
+    // Test with unsharded source and sharded target collection.
+    //
+    testOut(mongosColl, mongosTargetColl, false, true);
+
+    //
+    // Test with sharded source and sharded target collection.
+    //
+    testOut(mongosColl, mongosTargetColl, true, true);
+
+    //
+    // Test with sharded source and unsharded target collection.
+    //
+    testOut(mongosColl, mongosTargetColl, true, false);
+
+    //
+    // Test with unsharded source and unsharded target collection.
+    //
+    testOut(mongosColl, mongosTargetColl, false, false);
+
+    st.stop();
+}());
diff --git a/jstests/sharding/out_to_non_existing.js b/jstests/sharding/out_to_non_existing.js
new file mode 100644
index 00000000000..d3aab05e112
--- /dev/null
+++ b/jstests/sharding/out_to_non_existing.js
@@ -0,0 +1,50 @@
+// Tests for $out with a non-existing target collection.
+(function() {
+    "use strict";
+
+    const st = new ShardingTest({shards: 2, rs: {nodes: 1}});
+
+    const mongosDB = st.s0.getDB(jsTestName());
+    const mongosColl = mongosDB[jsTestName()];
+    const mongosTargetColl = mongosDB[jsTestName() + "_out"];
+
+    function testOut(sourceColl, targetColl, shardedSource) {
+        jsTestLog("Testing $out to non-existent target collection (source collection sharded : " +
+                  shardedSource + ").");
+        sourceColl.drop();
+        targetColl.drop();
+
+        if (shardedSource) {
+            st.shardColl(sourceColl, {_id: 1}, {_id: 0}, {_id: 1}, mongosDB.getName());
+        }
+
+        for (let i = 0; i < 10; i++) {
+            assert.commandWorked(sourceColl.insert({_id: i}));
+        }
+
+        // Test the behavior for each of the $out modes. Since the target collection does not exist,
+        // the behavior should be identical.
+        ["insertDocuments", "replaceDocuments", "replaceCollection"].forEach(mode => {
+            targetColl.drop();
+            sourceColl.aggregate([{$out: {to: targetColl.getName(), mode: mode}}]);
+            assert.eq(10, targetColl.find().itcount());
+        });
+
+        // Test with legacy syntax, which should behave identical to mode "replaceCollection".
+        targetColl.drop();
+        sourceColl.aggregate([{$out: targetColl.getName()}]);
+        assert.eq(10, targetColl.find().itcount());
+    }
+
+    //
+    // Test with unsharded source collection to a non-existent target collection.
+    //
+    testOut(mongosColl, mongosTargetColl, false);
+
+    //
+    // Test with sharded source collection to a non-existent target collection.
+    //
+    testOut(mongosColl, mongosTargetColl, true);
+
+    st.stop();
+}());