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

4 files changed, 72 insertions, 16 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 c1ed49a68df..09ce1fc9dc7 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
@@ -150,7 +150,8 @@ selector:
   - jstests/sharding/change_streams_update_lookup_shard_metadata_missing.js
   # Enable if SERVER-43310 is backported to 4.0
   - jstests/sharding/cluster_create_indexes_always_routes_through_primary.js
-
+   # Enable after SERVER-44477 is backported and available on 4.0 binaries
+  - jstests/sharding/mr_merge_to_existing.js
 executor:
   config:
     shell_options:
diff --git a/jstests/sharding/mapReduce_outSharded_checkUUID.js b/jstests/sharding/mapReduce_outSharded_checkUUID.js
index 6745e524a74..e7d0c21f521 100644
--- a/jstests/sharding/mapReduce_outSharded_checkUUID.js
+++ b/jstests/sharding/mapReduce_outSharded_checkUUID.js
@@ -46,14 +46,14 @@ function reduce(key, values) {
 // 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.
diff --git a/jstests/sharding/mr_merge_to_existing.js b/jstests/sharding/mr_merge_to_existing.js
new file mode 100644
index 00000000000..843e0da6b0e
--- /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.commandWorked(sourceColl.insert({key: 1}));
+assert.commandWorked(sourceColl.insert({key: 2}));
+assert.commandWorked(sourceColl.insert({key: 3}));
+assert.commandWorked(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.commandWorked(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 53432a49499..abd98902feb 100644
--- a/src/mongo/s/commands/cluster_map_reduce_cmd.cpp
+++ b/src/mongo/s/commands/cluster_map_reduce_cmd.cpp
@@ -483,19 +483,18 @@ 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,
                                           BSON("listCollections"
                                                << 1 << "filter"
@@ -511,15 +510,10 @@ public:
 
                     // If we reach here, the collection does not exist at all.
                     shouldDropAndShard = true;
+                    conn.done();
                 } 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);
+                    // The output collection exists and is sharded. Do not drop and reshard it.
                 }
-
-                conn.done();
             }
 
             // If we are using replace, the output collection exists and is sharded, or the output