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

2 files changed, 31 insertions, 12 deletions

diff --git a/jstests/sharding/mapReduce_outSharded_checkUUID.js b/jstests/sharding/mapReduce_outSharded_checkUUID.js
index 25a499c4bed..6745e524a74 100644
--- a/jstests/sharding/mapReduce_outSharded_checkUUID.js
+++ b/jstests/sharding/mapReduce_outSharded_checkUUID.js
@@ -81,6 +81,25 @@ assert.eq(origUUID, newUUID);
 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 f11ed660b9f..500e8b733fe 100644
--- a/src/mongo/db/commands/mr.cpp
+++ b/src/mongo/db/commands/mr.cpp
@@ -758,11 +758,16 @@ long long State::postProcessCollectionNonAtomic(OperationContext* opCtx,
     // Make sure we enforce prepare conflicts before writing.
     EnforcePrepareConflictsBlock enforcePrepare(opCtx);
 
-    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
@@ -780,12 +785,10 @@ 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);
-
         ProgressMeterHolder pm;
         {
             stdx::unique_lock<Client> lk(*opCtx->getClient());
-            pm.set(curOp->setProgress_inlock("M/R Merge Post Processing", count));
+            pm.set(curOp->setProgress_inlock("M/R Merge Post Processing", outputCount));
         }
 
         unique_ptr<DBClientCursor> cursor = _db.query(_config.tempNamespace, BSONObj());
@@ -801,12 +804,10 @@ long long State::postProcessCollectionNonAtomic(OperationContext* opCtx,
         // reduce: apply reduce op on new result and existing one
         BSONList values;
 
-        const auto count = collectionCount(opCtx, _config.tempNamespace, callerHoldsGlobalLock);
-
         ProgressMeterHolder pm;
         {
             stdx::unique_lock<Client> lk(*opCtx->getClient());
-            pm.set(curOp->setProgress_inlock("M/R Reduce Post Processing", count));
+            pm.set(curOp->setProgress_inlock("M/R Reduce Post Processing", outputCount));
         }
 
         unique_ptr<DBClientCursor> cursor = _db.query(_config.tempNamespace, BSONObj());
@@ -818,7 +819,6 @@ long long State::postProcessCollectionNonAtomic(OperationContext* opCtx,
 
             const bool found = [&] {
                 AutoGetCollection autoColl(opCtx, _config.outputOptions.finalNamespace, MODE_IS);
-                assertCollectionNotNull(_config.outputOptions.finalNamespace, autoColl);
                 return Helpers::findOne(
                     opCtx, autoColl.getCollection(), temp["_id"].wrap(), old, true);
             }();