SERVER-44477 Use correct collection count in cluster MR when determining whether to drop and reshard target

4 files changed, 70 insertions, 17 deletions

diff --git a/buildscripts/resmokeconfig/suites/sharding_last_stable_mongos_and_mixed_shards.yml b/buildscripts/resmokeconfig/suites/sharding_last_stable_mongos_and_mixed_shards.yml
index a6de0680202..4d94f01803b 100644
--- a/buildscripts/resmokeconfig/suites/sharding_last_stable_mongos_and_mixed_shards.yml
+++ b/buildscripts/resmokeconfig/suites/sharding_last_stable_mongos_and_mixed_shards.yml
@@ -83,6 +83,7 @@ selector:
   - jstests/sharding/json_schema.js
   - jstests/sharding/current_op_no_shards.js
   - jstests/sharding/mapReduce_outSharded_checkUUID.js
+  - jstests/sharding/mr_merge_to_existing.js
   # New feature in v3.6 mongo shell.
   - jstests/sharding/causal_consistency_shell_support.js
   - jstests/sharding/keys_rotation_interval_sec.js
@@ -121,7 +122,6 @@ selector:
   - jstests/sharding/explain_cmd.js
   # Enable when SERVER-44733 is backported
   - jstests/sharding/change_streams_update_lookup_shard_metadata_missing.js
-
 executor:
   config:
     shell_options:
diff --git a/jstests/sharding/mapReduce_outSharded_checkUUID.js b/jstests/sharding/mapReduce_outSharded_checkUUID.js
index 1845dbf45dc..c77275fcff6 100644
--- a/jstests/sharding/mapReduce_outSharded_checkUUID.js
+++ b/jstests/sharding/mapReduce_outSharded_checkUUID.js
@@ -35,6 +35,7 @@
             bulk.insert({j: j, i: i});
         }
     }
+
     assert.writeOK(bulk.execute());
 
     function map() {
@@ -47,14 +48,14 @@
     // sharded src sharded dst
     var suffix = "InShardedOutSharded";
 
-    // Check that merge to an existing empty sharded collection works and creates a new UUID after
+    // Check that merge to an existing empty sharded collection works and preserves the UUID after
     // M/R
     st.adminCommand({shardcollection: "mrShard.outSharded", key: {"_id": 1}});
     var origUUID = getUUIDFromConfigCollections(st.s, "mrShard.outSharded");
     var out = db.srcSharded.mapReduce(map, reduce, {out: {merge: "outSharded", sharded: true}});
     verifyOutput(out, 512);
     var newUUID = getUUIDFromConfigCollections(st.s, "mrShard.outSharded");
-    assert.neq(origUUID, newUUID);
+    assert.eq(origUUID, newUUID);
 
     // Shard1 is the primary shard and only one chunk should have been written, so the chunk with
     // the new UUID should have been written to it.
@@ -166,5 +167,4 @@
     assert.eq(newUUID, getUUIDFromListCollections(st.shard1.getDB("mrShard"), "replaceUnsharded"));
 
     st.stop();
-
 })();
diff --git a/jstests/sharding/mr_merge_to_existing.js b/jstests/sharding/mr_merge_to_existing.js
new file mode 100644
index 00000000000..fcf85091304
--- /dev/null
+++ b/jstests/sharding/mr_merge_to_existing.js
@@ -0,0 +1,61 @@
+// Test that running mapReduce to a target collection which is empty on the primary shard does *not*
+// drop and reshard it.
+(function() {
+
+    load("jstests/libs/fixture_helpers.js");
+
+    var st = new ShardingTest({
+        shards: 2,
+        rs: {nodes: 2},
+    });
+
+    const dbName = "test";
+    const collName = jsTestName();
+
+    let mongosConn = st.s;
+    assert.commandWorked(mongosConn.getDB(dbName).runCommand({create: collName}));
+    st.ensurePrimaryShard(dbName, st.shard0.shardName);
+
+    // Shard the test collection and split it into two chunks.
+    st.shardColl(collName,
+                 {_id: 1} /* Shard key */,
+                 {_id: 2} /* Split at */,
+                 {_id: 2} /* Move the chunk to its own shard */,
+                 dbName,
+                 true /* Wait until documents orphaned by the move get deleted */);
+
+    // Seed the source collection.
+    let sourceColl = mongosConn.getDB(dbName)[collName];
+    assert.writeOK(sourceColl.insert({key: 1}));
+    assert.writeOK(sourceColl.insert({key: 2}));
+    assert.writeOK(sourceColl.insert({key: 3}));
+    assert.writeOK(sourceColl.insert({key: 4}));
+
+    // Shard the target collection.
+    let mergeColl = mongosConn.getDB(dbName).mr_merge_out;
+    st.shardColl("mr_merge_out",
+                 {_id: 1} /* Shard key */,
+                 {_id: 2} /* Split at */,
+                 {_id: 2} /* Move the chunk containing {_id: 2} to its own shard */,
+                 dbName);
+
+    // Insert a single document to the target collection and ensure that it lives on the non-primary
+    // shard.
+    assert.writeOK(mergeColl.insert({_id: 5, value: 1}));
+
+    function map() {
+        emit(this.key, 1);
+    }
+    function reduce(key, values) {
+        return Array.sum(values);
+    }
+
+    // Run the mapReduce to merge to the existing sharded collection.
+    assert.commandWorked(
+        sourceColl.mapReduce(map, reduce, {out: {merge: mergeColl.getName(), sharded: true}}));
+
+    // Verify that the previous document still exists in the target collection.
+    assert.eq(mergeColl.find({_id: 5}).itcount(), 1);
+
+    st.stop();
+})();
diff --git a/src/mongo/s/commands/cluster_map_reduce_cmd.cpp b/src/mongo/s/commands/cluster_map_reduce_cmd.cpp
index ef892481c02..01d34b9f318 100644
--- a/src/mongo/s/commands/cluster_map_reduce_cmd.cpp
+++ b/src/mongo/s/commands/cluster_map_reduce_cmd.cpp
@@ -472,18 +472,16 @@ public:
             // We need to determine whether we need to drop and shard the output collection and
             // send the UUID to the shards. We will always do this if we are using replace so we
             // can skip this check in that case. If using merge or reduce, we only want to do this
-            // if the output collection does not exist or if it exists and is an empty sharded
-            // collection.
+            // if the output collection does not exist.
             bool shouldDropAndShard = replaceOutput;
             if (!replaceOutput && outputCollNss.isValid()) {
-                const auto primaryShard =
-                    uassertStatusOK(shardRegistry->getShard(opCtx, outputDbInfo.primaryId()));
-                ScopedDbConnection conn(primaryShard->getConnString());
-
                 if (!outputRoutingInfo.cm()) {
                     // The output collection either exists and is unsharded, or does not exist. If
                     // the output collection exists and is unsharded, fail because we should not go
                     // from unsharded to sharded.
+                    const auto primaryShard =
+                        uassertStatusOK(shardRegistry->getShard(opCtx, outputDbInfo.primaryId()));
+                    ScopedDbConnection conn(primaryShard->getConnString());
                     BSONObj listCollsCmdResponse;
                     ok = conn->runCommand(
                         outDB,
@@ -500,14 +498,8 @@ public:
 
                     // If we reach here, the collection does not exist at all.
                     shouldDropAndShard = true;
-                } else {
-                    // The output collection exists and is sharded. We need to determine whether the
-                    // collection is empty in order to decide whether we should drop and
-                    // re-shard it. We don't want to do this if the collection is not empty.
-                    shouldDropAndShard = (conn->count(outputCollNss.ns()) == 0);
+                    conn.done();
                 }
-
-                conn.done();
             }
 
             // If we are using replace, the output collection exists and is sharded, or the output