commit 767c953b4465f84f25a63806c4c95d44ab9c4e81
Author: Charlie Swanson <charlie.swanson@mongodb.com>
Date:   Wed Aug 21 13:00:17 2019 -0400

    I think this is the patch I had?

diff --git a/repro.js b/repro.js
new file mode 100644
index 0000000000..1bc5500e48
--- /dev/null
+++ b/repro.js
@@ -0,0 +1,25 @@
+(function() {
+"use strict";
+
+const st = new ShardingTest({shards: 2, config: 1});
+
+const mongosDB = st.s.getDB("test");
+const source = mongosDB.source;
+const target = mongosDB.target;
+
+// Shard collection, move some to each shard.
+st.shardColl(source, {_id: 1}, {_id: 1}, {_id: 2});
+
+assert.commandWorked(source.insert([{_id: 0}, {_id: 1}, {_id: 2}, {_id: 3}]));
+for (let i = 0; i < 500; i++) {
+    assert.eq(0, target.find().itcount());
+    assert.commandWorked(
+        mongosDB.runCommand({create: target.getName(), validator: {_id: {$nin: [0, 2]}}}));
+    st.shardColl(target, {_id: 1}, {_id: 1}, {_id: 2});
+    const explain = source.explain().aggregate([{$merge: {into: target.getName()}}]);
+    assert.throws(() => source.aggregate([{$merge: {into: target.getName()}}]));
+    target.drop();
+}
+
+st.stop();
+})();
diff --git a/src/mongo/db/pipeline/SConscript b/src/mongo/db/pipeline/SConscript
index 25efa885ef..f925c3dd73 100644
--- a/src/mongo/db/pipeline/SConscript
+++ b/src/mongo/db/pipeline/SConscript
@@ -241,6 +241,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/session_catalog',
         '$BUILD_DIR/mongo/db/storage/backup_cursor_hooks',
         '$BUILD_DIR/mongo/db/transaction',
+        '$BUILD_DIR/mongo/db/catalog/catalog_helpers',
         '$BUILD_DIR/mongo/scripting/scripting_common',
     ],
 )
diff --git a/src/mongo/db/pipeline/document_source_merge.cpp b/src/mongo/db/pipeline/document_source_merge.cpp
index 99c80f74ec..e07729e57d 100644
--- a/src/mongo/db/pipeline/document_source_merge.cpp
+++ b/src/mongo/db/pipeline/document_source_merge.cpp
@@ -422,6 +422,20 @@ boost::intrusive_ptr<DocumentSource> DocumentSourceMerge::createFromBson(
                                        targetCollectionVersion);
 }
 
+void DocumentSourceMerge::initialize() {
+    if (_targetCollectionVersion) {
+        return;  // If we know the version we're targeting we must have already detected that the
+                 // collection exists.
+    }
+    DocumentSourceWriteBlock writeBlock(pExpCtx->opCtx);
+
+    try {
+        pExpCtx->mongoProcessInterface->createCollection(pExpCtx, _outputNs, BSONObj());
+    } catch (const ExceptionFor<ErrorCodes::NamespaceExists>&) {
+        LOG(2) << "Attempted to create a collection that already exists. Ignoring.";
+    }
+}
+
 Value DocumentSourceMerge::serialize(boost::optional<ExplainOptions::Verbosity> explain) const {
     DocumentSourceMergeSpec spec;
     spec.setTargetNss(_outputNs);
diff --git a/src/mongo/db/pipeline/document_source_merge.h b/src/mongo/db/pipeline/document_source_merge.h
index f788952893..dc75b34333 100644
--- a/src/mongo/db/pipeline/document_source_merge.h
+++ b/src/mongo/db/pipeline/document_source_merge.h
@@ -143,6 +143,12 @@ public:
     static boost::intrusive_ptr<DocumentSource> createFromBson(
         BSONElement elem, const boost::intrusive_ptr<ExpressionContext>& pExpCtx);
 
+protected:
+    /**
+     * Creates the collection we will be writing to if necessary.
+     */
+    void initialize() final;
+
 private:
     /**
      * Builds a new $merge stage which will merge all documents into 'outputNs'. If
diff --git a/src/mongo/db/pipeline/document_source_out.cpp b/src/mongo/db/pipeline/document_source_out.cpp
index f4b7127a83..7d7112bcc8 100644
--- a/src/mongo/db/pipeline/document_source_out.cpp
+++ b/src/mongo/db/pipeline/document_source_out.cpp
@@ -128,16 +128,10 @@ void DocumentSourceOut::initialize() {
 
     // Create temp collection, copying options from the existing output collection if any.
     {
-        BSONObjBuilder cmd;
-        cmd << "create" << _tempNs.coll();
-        cmd << "temp" << true;
-        cmd.appendElementsUnique(_originalOutOptions);
-
-        BSONObj info;
-        uassert(16994,
-                "failed to create temporary {} collection '{}': {}"_format(
-                    kStageName, _tempNs.ns(), getStatusFromCommandResult(info).reason()),
-                conn->runCommand(outputNs.db().toString(), cmd.done(), info));
+        BSONObjBuilder opts;
+        opts << "temp" << true;
+        opts.appendElementsUnique(_originalOutOptions);
+        pExpCtx->mongoProcessInterface->createCollection(pExpCtx, _tempNs, opts.obj());
     }
 
     if (_originalIndexes.empty()) {
diff --git a/src/mongo/db/pipeline/mongo_process_common.cpp b/src/mongo/db/pipeline/mongo_process_common.cpp
index 5cbd150128..b54e6cadac 100644
--- a/src/mongo/db/pipeline/mongo_process_common.cpp
+++ b/src/mongo/db/pipeline/mongo_process_common.cpp
@@ -40,6 +40,7 @@
 #include "mongo/db/pipeline/expression_context.h"
 #include "mongo/db/service_context.h"
 #include "mongo/s/catalog_cache.h"
+#include "mongo/s/cluster_commands_helpers.h"
 #include "mongo/s/grid.h"
 #include "mongo/util/net/socket_utils.h"
 
@@ -181,4 +182,13 @@ std::set<FieldPath> MongoProcessCommon::_convertToFieldPaths(
     return fieldPaths;
 }
 
+void MongoProcessCommon::createCollection(const boost::intrusive_ptr<ExpressionContext>& expCtx,
+                                          const NamespaceString& nss,
+                                          BSONObj options) {
+    BSONObjBuilder cmdBuilder;
+    cmdBuilder.append("create", nss.coll());
+    cmdBuilder.appendElementsUnique(options);
+    mongo::createCollection(expCtx->opCtx, nss, cmdBuilder.obj());
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/mongo_process_common.h b/src/mongo/db/pipeline/mongo_process_common.h
index 3ba69ba20a..2bfecdae05 100644
--- a/src/mongo/db/pipeline/mongo_process_common.h
+++ b/src/mongo/db/pipeline/mongo_process_common.h
@@ -67,6 +67,10 @@ public:
         const boost::intrusive_ptr<ExpressionContext>& expCtx,
         const NamespaceString& nss) const final;
 
+    void createCollection(const boost::intrusive_ptr<ExpressionContext>&,
+                          const NamespaceString&,
+                          BSONObj options) override;
+
 protected:
     /**
      * Converts the fields from a ShardKeyPattern to a vector of FieldPaths, including the _id if
diff --git a/src/mongo/db/pipeline/mongo_process_interface.h b/src/mongo/db/pipeline/mongo_process_interface.h
index 58958aced6..b9e7a8c661 100644
--- a/src/mongo/db/pipeline/mongo_process_interface.h
+++ b/src/mongo/db/pipeline/mongo_process_interface.h
@@ -416,6 +416,10 @@ public:
      */
     virtual std::pair<JsExecution*, bool> getJsExec(const BSONObj& scope) = 0;
     virtual void releaseJsExec() = 0;
+
+    virtual void createCollection(const boost::intrusive_ptr<ExpressionContext>&,
+                                  const NamespaceString&,
+                                  BSONObj options) = 0;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/process_interface_shardsvr.cpp b/src/mongo/db/pipeline/process_interface_shardsvr.cpp
index 896de887cb..f1e596e3f7 100644
--- a/src/mongo/db/pipeline/process_interface_shardsvr.cpp
+++ b/src/mongo/db/pipeline/process_interface_shardsvr.cpp
@@ -124,6 +124,7 @@ Status MongoInterfaceShardServer::insert(const boost::intrusive_ptr<ExpressionCo
 
     // If applicable, attach a write concern to the batched command request.
     attachWriteConcern(&insertCommand, wc);
+    insertCommand.setAllowImplicitCreate(false);
 
     ClusterWriter::write(expCtx->opCtx, insertCommand, &stats, &response, targetEpoch);
 
@@ -145,6 +146,7 @@ StatusWith<MongoProcessInterface::UpdateResult> MongoInterfaceShardServer::updat
 
     // If applicable, attach a write concern to the batched command request.
     attachWriteConcern(&updateCommand, wc);
+    updateCommand.setAllowImplicitCreate(false);
 
     ClusterWriter::write(expCtx->opCtx, updateCommand, &stats, &response, targetEpoch);
 
diff --git a/src/mongo/db/pipeline/process_interface_standalone.cpp b/src/mongo/db/pipeline/process_interface_standalone.cpp
index ac6198dd79..45c6bc8adc 100644
--- a/src/mongo/db/pipeline/process_interface_standalone.cpp
+++ b/src/mongo/db/pipeline/process_interface_standalone.cpp
@@ -36,6 +36,7 @@
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_catalog.h"
+#include "mongo/db/catalog/create_collection.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/document_validation.h"
 #include "mongo/db/catalog/index_catalog_entry.h"
@@ -688,4 +689,15 @@ MongoInterfaceStandalone::ensureFieldsUniqueOrResolveDocumentKey(
     return {fieldPaths, targetCollectionVersion};
 }
 
+void MongoInterfaceStandalone::createCollection(
+    const boost::intrusive_ptr<ExpressionContext>& expCtx,
+    const NamespaceString& nss,
+    BSONObj options) {
+    BSONObjBuilder cmdBuilder;
+    cmdBuilder.append("create", nss.coll());
+    cmdBuilder.appendElementsUnique(options);
+    return uassertStatusOK(
+        mongo::createCollection(expCtx->opCtx, nss.db().toString(), cmdBuilder.obj()));
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/process_interface_standalone.h b/src/mongo/db/pipeline/process_interface_standalone.h
index 7e45dd0eaa..9dc3b9f41b 100644
--- a/src/mongo/db/pipeline/process_interface_standalone.h
+++ b/src/mongo/db/pipeline/process_interface_standalone.h
@@ -170,6 +170,10 @@ public:
         _jsExec.reset();
     }
 
+    void createCollection(const boost::intrusive_ptr<ExpressionContext>&,
+                          const NamespaceString&,
+                          BSONObj options) override;
+
 protected:
     BSONObj _reportCurrentOpForClient(OperationContext* opCtx,
                                       Client* client,
diff --git a/src/mongo/db/pipeline/stub_mongo_process_interface.h b/src/mongo/db/pipeline/stub_mongo_process_interface.h
index 1644d54536..beb7e84fd6 100644
--- a/src/mongo/db/pipeline/stub_mongo_process_interface.h
+++ b/src/mongo/db/pipeline/stub_mongo_process_interface.h
@@ -246,5 +246,12 @@ public:
         MONGO_UNREACHABLE;
     }
     void releaseJsExec() {}
-};
+
+    void createCollection(const boost::intrusive_ptr<ExpressionContext>& expCtx,
+                          const NamespaceString& nss,
+                          BSONObj options) final {
+        MONGO_UNREACHABLE;
+    }
+
+};  // namespace mongo
 }  // namespace mongo
diff --git a/src/mongo/s/cluster_commands_helpers.cpp b/src/mongo/s/cluster_commands_helpers.cpp
index 7eead10ebf..afb9492587 100644
--- a/src/mongo/s/cluster_commands_helpers.cpp
+++ b/src/mongo/s/cluster_commands_helpers.cpp
@@ -50,6 +50,7 @@
 #include "mongo/s/database_version_helpers.h"
 #include "mongo/s/grid.h"
 #include "mongo/s/multi_statement_transaction_requests_sender.h"
+#include "mongo/s/request_types/create_collection_gen.h"
 #include "mongo/s/request_types/create_database_gen.h"
 #include "mongo/s/shard_id.h"
 #include "mongo/s/stale_exception.h"
@@ -570,4 +571,35 @@ StatusWith<CachedCollectionRoutingInfo> getCollectionRoutingInfoForTxnCmd(
     return catalogCache->getCollectionRoutingInfoAt(opCtx, nss, atClusterTime.asTimestamp());
 }
 
+void createCollection(OperationContext* opCtx, const NamespaceString& nss, BSONObj cmdObj) {
+    createShardDatabase(opCtx, nss.db());
+
+    uassert(ErrorCodes::InvalidOptions,
+            "specify size:<n> when capped is true",
+            !cmdObj["capped"].trueValue() || cmdObj["size"].isNumber() ||
+                cmdObj.hasField("$nExtents"));
+
+    ConfigsvrCreateCollection configCreateCmd(nss);
+    configCreateCmd.setDbName(NamespaceString::kAdminDb);
+
+    {
+        BSONObjIterator cmdIter(cmdObj);
+        invariant(cmdIter.more());  // At least the command namespace should be present
+        cmdIter.next();
+        BSONObjBuilder optionsBuilder;
+        CommandHelpers::filterCommandRequestForPassthrough(&cmdIter, &optionsBuilder);
+        configCreateCmd.setOptions(optionsBuilder.obj());
+    }
+
+    const auto shardRegistry = Grid::get(opCtx)->shardRegistry();
+    auto response = shardRegistry->getConfigShard()->runCommandWithFixedRetryAttempts(
+        opCtx,
+        ReadPreferenceSetting{ReadPreference::PrimaryOnly},
+        "admin",
+        CommandHelpers::appendMajorityWriteConcern(
+            CommandHelpers::appendPassthroughFields(cmdObj, configCreateCmd.toBSON({}))),
+        Shard::RetryPolicy::kIdempotent);
+
+    uassertStatusOK(Shard::CommandResponse::getEffectiveStatus(response));
+}
 }  // namespace mongo
diff --git a/src/mongo/s/cluster_commands_helpers.h b/src/mongo/s/cluster_commands_helpers.h
index 1e5ee9685f..e7da4048c8 100644
--- a/src/mongo/s/cluster_commands_helpers.h
+++ b/src/mongo/s/cluster_commands_helpers.h
@@ -233,4 +233,8 @@ std::set<ShardId> getTargetedShardsForQuery(OperationContext* opCtx,
 StatusWith<CachedCollectionRoutingInfo> getCollectionRoutingInfoForTxnCmd(
     OperationContext* opCtx, const NamespaceString& nss);
 
+/**
+ * TODO
+ */
+void createCollection(OperationContext* opCtx, const NamespaceString&, BSONObj cmdObj);
 }  // namespace mongo
diff --git a/src/mongo/s/commands/cluster_create_cmd.cpp b/src/mongo/s/commands/cluster_create_cmd.cpp
index 9b9ebab29e..dc8c63fa9d 100644
--- a/src/mongo/s/commands/cluster_create_cmd.cpp
+++ b/src/mongo/s/commands/cluster_create_cmd.cpp
@@ -34,8 +34,6 @@
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/commands.h"
 #include "mongo/s/cluster_commands_helpers.h"
-#include "mongo/s/grid.h"
-#include "mongo/s/request_types/create_collection_gen.h"
 
 namespace mongo {
 namespace {
@@ -72,36 +70,7 @@ public:
              const BSONObj& cmdObj,
              BSONObjBuilder& result) override {
         const NamespaceString nss(parseNs(dbName, cmdObj));
-
-        createShardDatabase(opCtx, dbName);
-
-        uassert(ErrorCodes::InvalidOptions,
-                "specify size:<n> when capped is true",
-                !cmdObj["capped"].trueValue() || cmdObj["size"].isNumber() ||
-                    cmdObj.hasField("$nExtents"));
-
-        ConfigsvrCreateCollection configCreateCmd(nss);
-        configCreateCmd.setDbName(NamespaceString::kAdminDb);
-
-        {
-            BSONObjIterator cmdIter(cmdObj);
-            invariant(cmdIter.more());  // At least the command namespace should be present
-            cmdIter.next();
-            BSONObjBuilder optionsBuilder;
-            CommandHelpers::filterCommandRequestForPassthrough(&cmdIter, &optionsBuilder);
-            configCreateCmd.setOptions(optionsBuilder.obj());
-        }
-
-        const auto shardRegistry = Grid::get(opCtx)->shardRegistry();
-        auto response = shardRegistry->getConfigShard()->runCommandWithFixedRetryAttempts(
-            opCtx,
-            ReadPreferenceSetting{ReadPreference::PrimaryOnly},
-            "admin",
-            CommandHelpers::appendMajorityWriteConcern(
-                CommandHelpers::appendPassthroughFields(cmdObj, configCreateCmd.toBSON({}))),
-            Shard::RetryPolicy::kIdempotent);
-
-        uassertStatusOK(Shard::CommandResponse::getEffectiveStatus(response));
+        createCollection(opCtx, nss, cmdObj);
         return true;
     }
 
