SERVER-41249 Fix special cases for $out to include $merge

12 files changed, 55 insertions, 19 deletions

diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp
index 6aeb2f40659..fcb8d01237d 100644
--- a/src/mongo/db/commands/run_aggregate.cpp
+++ b/src/mongo/db/commands/run_aggregate.cpp
@@ -284,8 +284,8 @@ StatusWith<StringMap<ExpressionContext::ResolvedNamespace>> resolveInvolvedNames
 
         if (involvedNs.db() != request.getNamespaceString().db()) {
             // If the involved namespace is not in the same database as the aggregation, it must be
-            // from a $out to a collection in a different database. Since we cannot write to views,
-            // simply assume that the namespace is a collection.
+            // from a $merge to a collection in a different database. Since we cannot write to
+            // views, simply assume that the namespace is a collection.
             resolvedNamespaces[involvedNs.coll()] = {involvedNs, std::vector<BSONObj>{}};
         } else if (!db || db->getCollection(opCtx, involvedNs)) {
             // If the aggregation database exists and 'involvedNs' refers to a collection namespace,
diff --git a/src/mongo/db/pipeline/document_source_facet_test.cpp b/src/mongo/db/pipeline/document_source_facet_test.cpp
index d9cbe0146a3..7bc70618e1e 100644
--- a/src/mongo/db/pipeline/document_source_facet_test.cpp
+++ b/src/mongo/db/pipeline/document_source_facet_test.cpp
@@ -152,6 +152,28 @@ TEST_F(DocumentSourceFacetTest, ShouldRejectFacetsContainingAnOutStage) {
                   AssertionException);
 }
 
+TEST_F(DocumentSourceFacetTest, ShouldRejectFacetsContainingAMergeStage) {
+    auto ctx = getExpCtx();
+    auto spec =
+        BSON("$facet" << BSON("a" << BSON_ARRAY(BSON("$merge" << BSON("into"
+                                                                      << "merge_collection")))));
+    ASSERT_THROWS(DocumentSourceFacet::createFromBson(spec.firstElement(), ctx),
+                  AssertionException);
+
+    spec =
+        BSON("$facet" << BSON("a" << BSON_ARRAY(BSON("$skip" << 1)
+                                                << BSON("$merge" << BSON("into"
+                                                                         << "merge_collection")))));
+    ASSERT_THROWS(DocumentSourceFacet::createFromBson(spec.firstElement(), ctx),
+                  AssertionException);
+
+    spec = BSON("$facet" << BSON("a" << BSON_ARRAY(BSON("$merge" << BSON("into"
+                                                                         << "merge_collection"))
+                                                   << BSON("$skip" << 1))));
+    ASSERT_THROWS(DocumentSourceFacet::createFromBson(spec.firstElement(), ctx),
+                  AssertionException);
+}
+
 TEST_F(DocumentSourceFacetTest, ShouldRejectFacetsContainingAFacetStage) {
     auto ctx = getExpCtx();
     auto spec = fromjson("{$facet: {a: [{$facet: {a: [{$skip: 2}]}}]}}");
diff --git a/src/mongo/db/pipeline/document_source_merge_test.cpp b/src/mongo/db/pipeline/document_source_merge_test.cpp
index 4da65466223..0fb4686db82 100644
--- a/src/mongo/db/pipeline/document_source_merge_test.cpp
+++ b/src/mongo/db/pipeline/document_source_merge_test.cpp
@@ -53,7 +53,7 @@ const StringData kDefaultWhenNotMatchedMode =
 
 /**
  * For the purpsoses of this test, assume every collection is unsharded. Stages may ask this during
- * setup. For example, to compute its constraints, the $out stage needs to know if the output
+ * setup. For example, to compute its constraints, the $merge stage needs to know if the output
  * collection is sharded.
  */
 class MongoProcessInterfaceForTest : public StubMongoProcessInterface {
diff --git a/src/mongo/db/pipeline/pipeline.h b/src/mongo/db/pipeline/pipeline.h
index 05e2b77977d..71d94b800de 100644
--- a/src/mongo/db/pipeline/pipeline.h
+++ b/src/mongo/db/pipeline/pipeline.h
@@ -129,7 +129,7 @@ public:
         SourceContainer sources, const boost::intrusive_ptr<ExpressionContext>& expCtx);
 
     /**
-     * Returns true if the provided aggregation command has a $out stage.
+     * Returns true if the provided aggregation command has an $out or $merge stage.
      */
     static bool aggSupportsWriteConcern(const BSONObj& cmd);
 
diff --git a/src/mongo/db/pipeline/pipeline_test.cpp b/src/mongo/db/pipeline/pipeline_test.cpp
index 4453a0b559e..1b1f2748b66 100644
--- a/src/mongo/db/pipeline/pipeline_test.cpp
+++ b/src/mongo/db/pipeline/pipeline_test.cpp
@@ -2418,7 +2418,7 @@ DEATH_TEST_F(PipelineMustRunOnMongoSTest,
 
 /**
  * For the purpsoses of this test, assume every collection is unsharded. Stages may ask this during
- * setup. For example, to compute its constraints, the $out stage needs to know if the output
+ * setup. For example, to compute its constraints, the $merge stage needs to know if the output
  * collection is sharded.
  */
 class FakeMongoProcessInterface : public StubMongoProcessInterface {
diff --git a/src/mongo/db/pipeline/process_interface_shardsvr.h b/src/mongo/db/pipeline/process_interface_shardsvr.h
index b3c0a5885f7..c7044848ae5 100644
--- a/src/mongo/db/pipeline/process_interface_shardsvr.h
+++ b/src/mongo/db/pipeline/process_interface_shardsvr.h
@@ -52,7 +52,7 @@ public:
     std::vector<FieldPath> collectDocumentKeyFieldsActingAsRouter(
         OperationContext*, const NamespaceString&) const final {
         // We don't expect anyone to use this method on the shard itself (yet). This is currently
-        // only used for $out. For $out in a sharded cluster, the mongos is responsible for
+        // only used for $merge. For $out in a sharded cluster, the mongos is responsible for
         // collecting the document key fields before serializing them and sending them to the
         // shards. This is logically a MONGO_UNREACHABLE, but a malicious user could construct a
         // request to send directly to the shards which does not include the uniqueKey, so we must
diff --git a/src/mongo/db/pipeline/stage_constraints.h b/src/mongo/db/pipeline/stage_constraints.h
index b5edef5ef5e..da6133f11da 100644
--- a/src/mongo/db/pipeline/stage_constraints.h
+++ b/src/mongo/db/pipeline/stage_constraints.h
@@ -193,8 +193,8 @@ struct StageConstraints {
 
         // Stages which write data to user collections should not be permitted with readConcern
         // level "snapshot" or inside of a multi-document transaction.
-        // TODO (SERVER-36259): relax this requirement when $out (which writes persistent data)
-        // is allowed in a transaction.
+        // TODO (SERVER-36259): relax this requirement when $out and/or $merge (which write
+        // persistent data) is allowed in a transaction.
         if (diskRequirement == DiskUseRequirement::kWritesPersistentData) {
             invariant(!isAllowedInTransaction());
         }
diff --git a/src/mongo/db/views/view_catalog_test.cpp b/src/mongo/db/views/view_catalog_test.cpp
index 543b5edc323..65142d33cd3 100644
--- a/src/mongo/db/views/view_catalog_test.cpp
+++ b/src/mongo/db/views/view_catalog_test.cpp
@@ -252,6 +252,14 @@ TEST_F(ReplViewCatalogFixture, CreateViewWithPipelineFailsOnIneligibleStagePersi
         viewCatalog.createView(opCtx.get(), viewName, viewOn, invalidPipeline, emptyCollation),
         AssertionException,
         ErrorCodes::OptionNotSupportedOnView);
+
+    invalidPipeline = BSON_ARRAY(BSON("$merge"
+                                      << "someOtherCollection"));
+
+    ASSERT_THROWS_CODE(
+        viewCatalog.createView(opCtx.get(), viewName, viewOn, invalidPipeline, emptyCollation),
+        AssertionException,
+        ErrorCodes::OptionNotSupportedOnView);
 }
 
 TEST_F(ViewCatalogFixture, CreateViewOnInvalidCollectionName) {
diff --git a/src/mongo/s/query/cluster_aggregate.cpp b/src/mongo/s/query/cluster_aggregate.cpp
index 2cbb292adc8..64538f676cc 100644
--- a/src/mongo/s/query/cluster_aggregate.cpp
+++ b/src/mongo/s/query/cluster_aggregate.cpp
@@ -625,8 +625,9 @@ Status runPipelineOnMongoS(const boost::intrusive_ptr<ExpressionContext>& expCtx
     auto cursorResponse = establishMergingMongosCursor(
         opCtx, request, requestedNss, litePipe, std::move(pipeline), privileges);
 
-    // We don't need to storePossibleCursor or propagate writeConcern errors; an $out pipeline
-    // can never run on mongoS. Filter the command response and return immediately.
+    // We don't need to storePossibleCursor or propagate writeConcern errors; a pipeline with
+    // writing stages like $out can never run on mongoS. Filter the command response and return
+    // immediately.
     CommandHelpers::filterCommandReplyForPassthrough(cursorResponse, result);
     return getStatusFromCommandResult(result->asTempObj());
 }
diff --git a/src/mongo/s/query/cluster_aggregation_planner_test.cpp b/src/mongo/s/query/cluster_aggregation_planner_test.cpp
index 8f14dceeb6a..a6a81e7a5d4 100644
--- a/src/mongo/s/query/cluster_aggregation_planner_test.cpp
+++ b/src/mongo/s/query/cluster_aggregation_planner_test.cpp
@@ -55,7 +55,7 @@ const NamespaceString kTestOutNss = NamespaceString{"unittests", "out_ns"};
 
 /**
  * For the purposes of this test, assume every collection is sharded. Stages may ask this during
- * setup. For example, to compute its constraints, the $out stage needs to know if the output
+ * setup. For example, to compute its constraints, the $merge stage needs to know if the output
  * collection is sharded.
  */
 class FakeMongoProcessInterface : public StubMongoProcessInterface {
diff --git a/src/mongo/s/transaction_router.cpp b/src/mongo/s/transaction_router.cpp
index 667ba573f17..a5b8a864bbc 100644
--- a/src/mongo/s/transaction_router.cpp
+++ b/src/mongo/s/transaction_router.cpp
@@ -145,9 +145,9 @@ BSONObjBuilder appendFieldsForStartTransaction(BSONObj cmd,
 }
 
 // Commands that are idempotent in a transaction context and can be blindly retried in the middle of
-// a transaction. Aggregate with $out is disallowed in a transaction, so aggregates must be read
-// operations. Note: aggregate and find do have the side-effect of creating cursors, but any
-// established during an unsuccessful attempt are best-effort killed.
+// a transaction. Writing aggregates (e.g. with a $out or $merge) is disallowed in a transaction, so
+// aggregates must be read operations. Note: aggregate and find do have the side-effect of creating
+// cursors, but any established during an unsuccessful attempt are best-effort killed.
 const StringMap<int> alwaysRetryableCmds = {
     {"aggregate", 1}, {"distinct", 1}, {"find", 1}, {"getMore", 1}, {"killCursors", 1}};
 
diff --git a/src/mongo/shell/db.js b/src/mongo/shell/db.js
index 2d9f0789db9..87559699b7d 100644
--- a/src/mongo/shell/db.js
+++ b/src/mongo/shell/db.js
@@ -232,13 +232,18 @@ var DB;
 
         const pipeline = cmdObj.pipeline;
 
-        // Check whether the pipeline has an $out stage. If not, we may run on a Secondary and
-        // should attach a readPreference.
-        const hasOutStage =
-            pipeline.length >= 1 && pipeline[pipeline.length - 1].hasOwnProperty("$out");
+        // Check whether the pipeline has a stage which performs writes like $out. If not, we may
+        // run on a Secondary and should attach a readPreference.
+        const hasWritingStage = (function() {
+            if (pipeline.length == 0) {
+                return false;
+            }
+            const lastStage = pipeline[pipeline.length - 1];
+            return lastStage.hasOwnProperty("$out") || lastStage.hasOwnProperty("$merge");
+        }());
 
         const doAgg = function(cmdObj) {
-            return hasOutStage ? this.runCommand(cmdObj) : this.runReadCommand(cmdObj);
+            return hasWritingStage ? this.runCommand(cmdObj) : this.runReadCommand(cmdObj);
         }.bind(this);
 
         const res = doAgg(cmdObj);