SERVER-45540 Gate $unionWith behind FCV check and test upgrade/downgrade

5 files changed, 120 insertions, 14 deletions

diff --git a/jstests/multiVersion/genericSetFCVUsage/view_definition_feature_compatibility_version.js b/jstests/multiVersion/genericSetFCVUsage/view_definition_feature_compatibility_version.js
index d7e6ce36bfc..2a575367b64 100644
--- a/jstests/multiVersion/genericSetFCVUsage/view_definition_feature_compatibility_version.js
+++ b/jstests/multiVersion/genericSetFCVUsage/view_definition_feature_compatibility_version.js
@@ -36,6 +36,10 @@ const pipelinesWithNewFeatures = [
     [{$project: {x: {$replaceAll: {input: '', find: '', replacement: ''}}}}],
     [{$project: {x: {$first: {$literal: ['a']}}}}],
     [{$project: {x: {$last: {$literal: ['a']}}}}],
+    [{$unionWith: "A"}],
+    [{$unionWith: {coll: "A", pipeline: [{$match: {b: 1}}]}}],
+    [{$lookup: {from: "A", pipeline: [{$unionWith: "B"}], as: "result"}}],
+    [{$facet: {sub_pipe_invalid: [{$unionWith: "B"}], sub_pipe_valid: [{$match: {b: 1}}]}}],
 ];
 
 let conn = MongoRunner.runMongod({dbpath: dbpath, binVersion: "latest"});
@@ -55,7 +59,7 @@ pipelinesWithNewFeatures.forEach(
         `Expected to be able to create view with pipeline ${tojson(pipe)} while in FCV` +
             ` ${latestFCV}`));
 
-// Test that we are able to create a new view with any of the new features.
+// Test that we are able to update an existing view with any of the new features.
 pipelinesWithNewFeatures.forEach(function(pipe, i) {
     assert(testDB["firstView" + i].drop(), `Drop of view with pipeline ${tojson(pipe)} failed`);
     assert.commandWorked(testDB.createView("firstView" + i, "coll", []));
diff --git a/jstests/multiVersion/unionWith_fcv.js b/jstests/multiVersion/unionWith_fcv.js
new file mode 100644
index 00000000000..35b9a3c15f2
--- /dev/null
+++ b/jstests/multiVersion/unionWith_fcv.js
@@ -0,0 +1,78 @@
+/**
+ * Test the behavior of the $unionWith aggregation stage against a standalone/sharded cluster during
+ * upgrade.
+ *
+ *  Checking UUID consistency uses cached connections, which are not valid across restarts or
+ *  stepdowns.
+ */
+TestData.skipCheckingUUIDsConsistentAcrossCluster = true;
+
+(function() {
+"use strict";
+
+load("jstests/multiVersion/libs/multi_cluster.js");  // For upgradeCluster.
+
+let conn = MongoRunner.runMongod({binVersion: "latest"});
+assert.neq(null, conn, "mongod was unable to start up");
+let testDB = conn.getDB(jsTestName());
+
+// Set the feature compatibility version to the last-stable version.
+assert.commandWorked(testDB.adminCommand({setFeatureCompatibilityVersion: lastStableFCV}));
+
+// Seed the two involved collections.
+assert.commandWorked(testDB.collA.insert({fromA: 1}));
+assert.commandWorked(testDB.collB.insert({fromB: 1}));
+
+// Verify that we can still use $unionWith since the binary version is 4.4.
+const pipeline = [{$unionWith: "collB"}, {$project: {_id: 0}}];
+assert.sameMembers([{fromA: 1}, {fromB: 1}], testDB.collA.aggregate(pipeline).toArray());
+
+// Set the feature compatibility version to the latest version.
+assert.commandWorked(testDB.adminCommand({setFeatureCompatibilityVersion: latestFCV}));
+
+// Verify that we can still use $unionWith.
+assert.sameMembers([{fromA: 1}, {fromB: 1}], testDB.collA.aggregate(pipeline).toArray());
+
+MongoRunner.stopMongod(conn);
+
+// Start a sharded cluster in which all mongod and mongos processes are "last-stable" binVersion.
+let st = new ShardingTest({
+    shards: 2,
+    rs: {nodes: 2, binVersion: "last-stable"},
+    other: {mongosOptions: {binVersion: "last-stable"}}
+});
+
+testDB = st.s.getDB(jsTestName());
+assert.commandWorked(testDB.runCommand({create: "collA"}));
+st.ensurePrimaryShard(testDB.getName(), st.shard0.shardName);
+
+// Seed the two involved collections.
+assert.commandWorked(testDB.collA.insert({fromA: 1}));
+assert.commandWorked(testDB.collB.insert({fromB: 1}));
+
+// Aggregations with $unionWith should fail against older binary versions.
+assert.commandFailedWithCode(
+    testDB.runCommand({aggregate: "collA", pipeline: pipeline, cursor: {}}), 40324);
+
+// Upgrade the config servers and the shards to the "latest" binVersion.
+st.upgradeCluster("latest", {upgradeShards: true, upgradeConfigs: true, upgradeMongos: false});
+
+// Since mongos is still on 4.2, $unionWith should fail to parse.
+assert.commandFailedWithCode(
+    testDB.runCommand({aggregate: "collA", pipeline: pipeline, cursor: {}}), 40324);
+
+// Upgrade mongos to the "latest" binVersion but keep the old FCV.
+st.upgradeCluster("latest", {upgradeShards: false, upgradeConfigs: false, upgradeMongos: true});
+testDB = st.s.getDB(jsTestName());
+
+// Now an aggregation containing $unionWith should pass because all nodes are on binary version 4.4.
+assert.sameMembers([{fromA: 1}, {fromB: 1}], testDB.collA.aggregate(pipeline).toArray());
+
+// For completeness, set the FCV to the latest.
+assert.commandWorked(testDB.adminCommand({setFeatureCompatibilityVersion: latestFCV}));
+
+// Verify that $unionWith is allowed in a fully upgraded cluster.
+assert.sameMembers([{fromA: 1}, {fromB: 1}], testDB.collA.aggregate(pipeline).toArray());
+
+st.stop();
+}());
diff --git a/src/mongo/db/pipeline/document_source.cpp b/src/mongo/db/pipeline/document_source.cpp
index 91e5c77f5a5..f6b5b38c486 100644
--- a/src/mongo/db/pipeline/document_source.cpp
+++ b/src/mongo/db/pipeline/document_source.cpp
@@ -31,6 +31,7 @@
 
 #include "mongo/db/pipeline/document_source.h"
 
+#include "mongo/db/commands/feature_compatibility_version_documentation.h"
 #include "mongo/db/exec/document_value/value.h"
 #include "mongo/db/matcher/expression_algo.h"
 #include "mongo/db/pipeline/document_source_group.h"
@@ -56,16 +57,23 @@ DocumentSource::DocumentSource(const StringData stageName,
     : pSource(nullptr), pExpCtx(pCtx), _commonStats(stageName.rawData()) {}
 
 namespace {
+struct ParserRegistration {
+    Parser parser;
+    boost::optional<ServerGlobalParams::FeatureCompatibility::Version> requiredMinVersion;
+};
 // Used to keep track of which DocumentSources are registered under which name.
-static StringMap<Parser> parserMap;
+static StringMap<ParserRegistration> parserMap;
 }  // namespace
 
-void DocumentSource::registerParser(string name, Parser parser) {
+void DocumentSource::registerParser(
+    string name,
+    Parser parser,
+    boost::optional<ServerGlobalParams::FeatureCompatibility::Version> requiredMinVersion) {
     auto it = parserMap.find(name);
     massert(28707,
             str::stream() << "Duplicate document source (" << name << ") registered.",
             it == parserMap.end());
-    parserMap[name] = parser;
+    parserMap[name] = {parser, requiredMinVersion};
 }
 
 list<intrusive_ptr<DocumentSource>> DocumentSource::parse(
@@ -83,7 +91,15 @@ list<intrusive_ptr<DocumentSource>> DocumentSource::parse(
             str::stream() << "Unrecognized pipeline stage name: '" << stageName << "'",
             it != parserMap.end());
 
-    return it->second(stageSpec, expCtx);
+    uassert(ErrorCodes::QueryFeatureNotAllowed,
+            str::stream() << stageName
+                          << " is not allowed in the current feature compatibility version. See "
+                          << feature_compatibility_version_documentation::kCompatibilityLink
+                          << " for more information.",
+            !expCtx->maxFeatureCompatibilityVersion || !it->second.requiredMinVersion ||
+                (*it->second.requiredMinVersion <= *expCtx->maxFeatureCompatibilityVersion));
+
+    return it->second.parser(stageSpec, expCtx);
 }
 
 const char* DocumentSource::getSourceName() const {
diff --git a/src/mongo/db/pipeline/document_source.h b/src/mongo/db/pipeline/document_source.h
index 5f956431c36..6710a1c78ce 100644
--- a/src/mongo/db/pipeline/document_source.h
+++ b/src/mongo/db/pipeline/document_source.h
@@ -87,7 +87,7 @@ class Document;
  * If your stage is actually an alias which needs to return more than one stage (such as
  * $sortByCount), you should use the REGISTER_MULTI_STAGE_ALIAS macro instead.
  */
-#define REGISTER_DOCUMENT_SOURCE_CONDITIONALLY(key, liteParser, fullParser, ...)             \
+#define REGISTER_DOCUMENT_SOURCE_CONDITIONALLY(key, liteParser, fullParser, minVersion, ...) \
     MONGO_INITIALIZER(addToDocSourceParserMap_##key)(InitializerContext*) {                  \
         if (!__VA_ARGS__) {                                                                  \
             return Status::OK();                                                             \
@@ -98,16 +98,19 @@ class Document;
                 (fullParser)(stageSpec, expCtx)};                                            \
         };                                                                                   \
         LiteParsedDocumentSource::registerParser("$" #key, liteParser);                      \
-        DocumentSource::registerParser("$" #key, fullParserWrapper);                         \
+        DocumentSource::registerParser("$" #key, fullParserWrapper, minVersion);             \
         return Status::OK();                                                                 \
     }
 
 #define REGISTER_DOCUMENT_SOURCE(key, liteParser, fullParser) \
-    REGISTER_DOCUMENT_SOURCE_CONDITIONALLY(key, liteParser, fullParser, true)
+    REGISTER_DOCUMENT_SOURCE_CONDITIONALLY(key, liteParser, fullParser, boost::none, true)
 
 #define REGISTER_TEST_DOCUMENT_SOURCE(key, liteParser, fullParser) \
     REGISTER_DOCUMENT_SOURCE_CONDITIONALLY(                        \
-        key, liteParser, fullParser, ::mongo::getTestCommandsEnabled())
+        key, liteParser, fullParser, boost::none, ::mongo::getTestCommandsEnabled())
+
+#define REGISTER_DOCUMENT_SOURCE_WITH_MIN_VERSION(key, liteParser, fullParser, minVersion) \
+    REGISTER_DOCUMENT_SOURCE_CONDITIONALLY(key, liteParser, fullParser, minVersion, true)
 
 /**
  * Registers a multi-stage alias (such as $sortByCount) to have the single name 'key'. When a stage
@@ -125,7 +128,7 @@ class Document;
 #define REGISTER_MULTI_STAGE_ALIAS(key, liteParser, fullParser)                  \
     MONGO_INITIALIZER(addAliasToDocSourceParserMap_##key)(InitializerContext*) { \
         LiteParsedDocumentSource::registerParser("$" #key, (liteParser));        \
-        DocumentSource::registerParser("$" #key, (fullParser));                  \
+        DocumentSource::registerParser("$" #key, (fullParser), boost::none);     \
         return Status::OK();                                                     \
     }
 
@@ -354,7 +357,10 @@ public:
      * DO NOT call this method directly. Instead, use the REGISTER_DOCUMENT_SOURCE macro defined in
      * this file.
      */
-    static void registerParser(std::string name, Parser parser);
+    static void registerParser(
+        std::string name,
+        Parser parser,
+        boost::optional<ServerGlobalParams::FeatureCompatibility::Version> requiredMinVersion);
 
 private:
     /**
diff --git a/src/mongo/db/pipeline/document_source_union_with.cpp b/src/mongo/db/pipeline/document_source_union_with.cpp
index 6d77dc4ca1e..52d730cff39 100644
--- a/src/mongo/db/pipeline/document_source_union_with.cpp
+++ b/src/mongo/db/pipeline/document_source_union_with.cpp
@@ -41,9 +41,11 @@
 
 namespace mongo {
 
-REGISTER_DOCUMENT_SOURCE(unionWith,
-                         DocumentSourceUnionWith::LiteParsed::parse,
-                         DocumentSourceUnionWith::createFromBson);
+REGISTER_DOCUMENT_SOURCE_WITH_MIN_VERSION(
+    unionWith,
+    DocumentSourceUnionWith::LiteParsed::parse,
+    DocumentSourceUnionWith::createFromBson,
+    ServerGlobalParams::FeatureCompatibility::Version::kFullyUpgradedTo44);
 
 namespace {
 std::unique_ptr<Pipeline, PipelineDeleter> buildPipelineFromViewDefinition(