SERVER-44527 Avoid renaming to target collection in mapReduce if the shard does not own any documents

(cherry picked from commit b4db881a18cbe15127a5a60c971cd393e0621466)

2 files changed, 40 insertions, 21 deletions

diff --git a/jstests/sharding/mapReduce_outSharded_checkUUID.js b/jstests/sharding/mapReduce_outSharded_checkUUID.js
index 9faa35cb836..0de45c127c7 100644
--- a/jstests/sharding/mapReduce_outSharded_checkUUID.js
+++ b/jstests/sharding/mapReduce_outSharded_checkUUID.js
@@ -82,6 +82,25 @@
     assert.eq(newUUID, getUUIDFromListCollections(st.shard0.getDB("mrShard"), "outSharded"));
     assert.eq(newUUID, getUUIDFromListCollections(st.shard1.getDB("mrShard"), "outSharded"));
 
+    // Check that merge to an existing sharding collection that has data only on the primary shard
+    // works and that the collection uses the same UUID after M/R.
+    assert.commandWorked(st.s.getCollection("mrShard.outSharded").remove({_id: 2001}));
+    out = db.srcSharded.mapReduce(map, reduce, {out: {merge: "outSharded", sharded: true}});
+    verifyOutput(out, 513);
+    newUUID = getUUIDFromConfigCollections(st.s, "mrShard.outSharded");
+    assert.eq(origUUID, newUUID);
+    assert.eq(newUUID, getUUIDFromListCollections(st.shard0.getDB(db.getName()), "outSharded"));
+    assert.eq(newUUID, getUUIDFromListCollections(st.shard1.getDB(db.getName()), "outSharded"));
+
+    // Similarly, check that reduce to an existing sharding collection that has data only on the
+    // primary shard works and that the collection uses the same UUID after M/R.
+    out = db.srcSharded.mapReduce(map, reduce, {out: {reduce: "outSharded", sharded: true}});
+    verifyOutput(out, 513);
+    newUUID = getUUIDFromConfigCollections(st.s, "mrShard.outSharded");
+    assert.eq(origUUID, newUUID);
+    assert.eq(newUUID, getUUIDFromListCollections(st.shard0.getDB(db.getName()), "outSharded"));
+    assert.eq(newUUID, getUUIDFromListCollections(st.shard1.getDB(db.getName()), "outSharded"));
+
     // Check that replace to an existing sharded collection has data on all shards works and that
     // the collection creates a new UUID after M/R.
     origUUID = getUUIDFromConfigCollections(st.s, "mrShard.outSharded");
diff --git a/src/mongo/db/commands/mr.cpp b/src/mongo/db/commands/mr.cpp
index 5dcee0dd8bd..3637948a1e7 100644
--- a/src/mongo/db/commands/mr.cpp
+++ b/src/mongo/db/commands/mr.cpp
@@ -664,11 +664,16 @@ long long State::postProcessCollectionNonAtomic(OperationContext* opCtx,
                                                 CurOp* curOp,
                                                 ProgressMeterHolder& pm,
                                                 bool callerHoldsGlobalLock) {
-    if (_config.outputOptions.finalNamespace == _config.tempNamespace)
-        return _collectionCount(opCtx, _config.outputOptions.finalNamespace, callerHoldsGlobalLock);
-
-    if (_config.outputOptions.outType == Config::REPLACE ||
-        _collectionCount(opCtx, _config.outputOptions.finalNamespace, callerHoldsGlobalLock) == 0) {
+    auto outputCount =
+        _collectionCount(opCtx, _config.outputOptions.finalNamespace, callerHoldsGlobalLock);
+
+    // Determine whether the temp collection should be renamed to the final output collection and
+    // thus preserve the UUID. This is possible in the following cases:
+    //  * Output mode "replace"
+    //  * If this mapReduce is creating a new sharded output collection, which can be determined by
+    //  whether mongos sent the UUID that the final output collection should have (that is, whether
+    //  _config.finalOutputCollUUID is set).
+    if (_config.outputOptions.outType == Config::REPLACE || _config.finalOutputCollUUID) {
         // This must be global because we may write across different databases.
         Lock::GlobalWrite lock(opCtx);
         // replace: just rename from temp to final collection name, dropping previous collection
@@ -687,12 +692,11 @@ long long State::postProcessCollectionNonAtomic(OperationContext* opCtx,
         _db.dropCollection(_config.tempNamespace.ns());
     } else if (_config.outputOptions.outType == Config::MERGE) {
         // merge: upsert new docs into old collection
+
         {
-            const auto count =
-                _collectionCount(opCtx, _config.tempNamespace, callerHoldsGlobalLock);
-            stdx::lock_guard<Client> lk(*opCtx->getClient());
+            stdx::unique_lock<Client> lk(*opCtx->getClient());
             curOp->setMessage_inlock(
-                "m/r: merge post processing", "M/R Merge Post Processing Progress", count);
+                "m/r: merge post processing", "M/R Merge Post Processing Progress", outputCount);
         }
         unique_ptr<DBClientCursor> cursor = _db.query(_config.tempNamespace.ns(), BSONObj());
         while (cursor->more()) {
@@ -708,12 +712,11 @@ long long State::postProcessCollectionNonAtomic(OperationContext* opCtx,
         BSONList values;
 
         {
-            const auto count =
-                _collectionCount(opCtx, _config.tempNamespace, callerHoldsGlobalLock);
-            stdx::lock_guard<Client> lk(*opCtx->getClient());
+            stdx::unique_lock<Client> lk(*opCtx->getClient());
             curOp->setMessage_inlock(
-                "m/r: reduce post processing", "M/R Reduce Post Processing Progress", count);
+                "m/r: reduce post processing", "M/R Reduce Post Processing Progress", outputCount);
         }
+
         unique_ptr<DBClientCursor> cursor = _db.query(_config.tempNamespace.ns(), BSONObj());
         while (cursor->more()) {
             // This must be global because we may write across different databases.
@@ -721,14 +724,11 @@ long long State::postProcessCollectionNonAtomic(OperationContext* opCtx,
             BSONObj temp = cursor->nextSafe();
             BSONObj old;
 
-            bool found;
-            {
-                OldClientContext tx(opCtx, _config.outputOptions.finalNamespace.ns());
-                Collection* coll =
-                    getCollectionOrUassert(opCtx, tx.db(), _config.outputOptions.finalNamespace);
-                found = Helpers::findOne(opCtx, coll, temp["_id"].wrap(), old, true);
-            }
-
+            const bool found = [&] {
+                AutoGetCollection autoColl(opCtx, _config.outputOptions.finalNamespace, MODE_IS);
+                return Helpers::findOne(
+                    opCtx, autoColl.getCollection(), temp["_id"].wrap(), old, true);
+            }();
             if (found) {
                 // need to reduce
                 values.clear();