SERVER-58379 Extend REGISTER macros to support API version, FCV, and conditional checks for accumulators, expressions, and window functions

9 files changed, 322 insertions, 187 deletions

diff --git a/src/mongo/db/pipeline/accumulation_statement.cpp b/src/mongo/db/pipeline/accumulation_statement.cpp
index 18cca9fd32a..d5afb1f83d0 100644
--- a/src/mongo/db/pipeline/accumulation_statement.cpp
+++ b/src/mongo/db/pipeline/accumulation_statement.cpp
@@ -37,6 +37,7 @@
 #include "mongo/db/commands/feature_compatibility_version_documentation.h"
 #include "mongo/db/exec/document_value/value.h"
 #include "mongo/db/pipeline/accumulator.h"
+#include "mongo/db/query/allowed_contexts.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/str.h"
 #include "mongo/util/string_map.h"
@@ -46,39 +47,29 @@ namespace mongo {
 using std::string;
 
 namespace {
-// Used to keep track of which Accumulators are registered under which name.
-using ParserRegistration = std::pair<AccumulationStatement::Parser,
-                                     boost::optional<multiversion::FeatureCompatibilityVersion>>;
-static StringMap<ParserRegistration> parserMap;
+// Used to keep track of which Accumulators are registered under which name and in which contexts
+// they can be used in.
+static StringMap<AccumulationStatement::ParserRegistration> parserMap;
 }  // namespace
 
 void AccumulationStatement::registerAccumulator(
     std::string name,
     AccumulationStatement::Parser parser,
+    AllowedWithApiStrict allowedWithApiStrict,
+    AllowedWithClientType allowedWithClientType,
     boost::optional<multiversion::FeatureCompatibilityVersion> requiredMinVersion) {
     auto it = parserMap.find(name);
     massert(28722,
             str::stream() << "Duplicate accumulator (" << name << ") registered.",
             it == parserMap.end());
-    parserMap[name] = {parser, requiredMinVersion};
+    parserMap[name] = {parser, allowedWithApiStrict, allowedWithClientType, requiredMinVersion};
 }
 
-AccumulationStatement::Parser& AccumulationStatement::getParser(
-    StringData name, boost::optional<multiversion::FeatureCompatibilityVersion> allowedMaxVersion) {
+AccumulationStatement::ParserRegistration& AccumulationStatement::getParser(StringData name) {
     auto it = parserMap.find(name);
     uassert(
         15952, str::stream() << "unknown group operator '" << name << "'", it != parserMap.end());
-    auto& [parser, requiredMinVersion] = it->second;
-    uassert(ErrorCodes::QueryFeatureNotAllowed,
-            // We would like to include the current version and the required minimum version in this
-            // error message, but using FeatureCompatibilityVersion::toString() would introduce a
-            // dependency cycle (see SERVER-31968).
-            str::stream() << name
-                          << " is not allowed in the current feature compatibility version. See "
-                          << feature_compatibility_version_documentation::kCompatibilityLink
-                          << " for more information.",
-            !requiredMinVersion || !allowedMaxVersion || *requiredMinVersion <= *allowedMaxVersion);
-    return parser;
+    return it->second;
 }
 
 boost::intrusive_ptr<AccumulatorState> AccumulationStatement::makeAccumulator() const {
@@ -111,8 +102,22 @@ AccumulationStatement AccumulationStatement::parseAccumulationStatement(
             str::stream() << "The " << accName << " accumulator is a unary operator",
             specElem.type() != BSONType::Array);
 
-    auto&& parser =
-        AccumulationStatement::getParser(accName, expCtx->maxFeatureCompatibilityVersion);
+    auto&& [parser, allowedWithApiStrict, allowedWithClientType, requiredMinVersion] =
+        AccumulationStatement::getParser(accName);
+    auto allowedMaxVersion = expCtx->maxFeatureCompatibilityVersion;
+    uassert(ErrorCodes::QueryFeatureNotAllowed,
+            // We would like to include the current version and the required minimum version in this
+            // error message, but using FeatureCompatibilityVersion::toString() would introduce a
+            // dependency cycle (see SERVER-31968).
+            str::stream() << accName
+                          << " is not allowed in the current feature compatibility version. See "
+                          << feature_compatibility_version_documentation::kCompatibilityLink
+                          << " for more information.",
+            !requiredMinVersion || !allowedMaxVersion || *requiredMinVersion <= *allowedMaxVersion);
+
+    tassert(5837900, "Accumulators should only appear in a user operation", expCtx->opCtx);
+    assertLanguageFeatureIsAllowed(
+        expCtx->opCtx, accName.toString(), allowedWithApiStrict, allowedWithClientType);
     auto accExpr = parser(expCtx, specElem, vps);
 
     return AccumulationStatement(fieldName.toString(), std::move(accExpr));
diff --git a/src/mongo/db/pipeline/accumulation_statement.h b/src/mongo/db/pipeline/accumulation_statement.h
index e473d5d7ea1..b056b3ca2d9 100644
--- a/src/mongo/db/pipeline/accumulation_statement.h
+++ b/src/mongo/db/pipeline/accumulation_statement.h
@@ -43,15 +43,45 @@ namespace mongo {
  * you would add this line:
  * REGISTER_ACCUMULATOR(foo, AccumulatorFoo::create);
  */
-#define REGISTER_ACCUMULATOR(key, factory) \
-    REGISTER_ACCUMULATOR_WITH_MIN_VERSION(key, factory, boost::none)
-
-#define REGISTER_ACCUMULATOR_WITH_MIN_VERSION(key, factory, minVersion)                \
-    MONGO_INITIALIZER_GENERAL(addToAccumulatorFactoryMap_##key,                        \
-                              ("BeginAccumulatorRegistration"),                        \
-                              ("EndAccumulatorRegistration"))                          \
-    (InitializerContext*) {                                                            \
-        AccumulationStatement::registerAccumulator("$" #key, (factory), (minVersion)); \
+#define REGISTER_ACCUMULATOR(key, factory)                            \
+    REGISTER_ACCUMULATOR_CONDITIONALLY(key,                           \
+                                       factory,                       \
+                                       AllowedWithApiStrict::kAlways, \
+                                       AllowedWithClientType::kAny,   \
+                                       boost::none,                   \
+                                       true)
+
+#define REGISTER_ACCUMULATOR_WITH_MIN_VERSION(key, factory, minVersion) \
+    REGISTER_ACCUMULATOR_CONDITIONALLY(key,                             \
+                                       factory,                         \
+                                       AllowedWithApiStrict::kAlways,   \
+                                       AllowedWithClientType::kAny,     \
+                                       minVersion,                      \
+                                       true)
+
+/**
+ * Like REGISTER_ACCUMULATOR_WITH_MIN_VERSION, except you can also specify a condition,
+ * evaluated during startup, that decides whether to register the parser.
+ *
+ * For example, you could check a feature flag, and register the parser only when it's enabled.
+ *
+ * Note that the condition is evaluated only once, during a MONGO_INITIALIZER. Don't specify
+ * a condition that can change at runtime, such as FCV. (Feature flags are ok, because they
+ * cannot be toggled at runtime.)
+ *
+ * This is the most general REGISTER_ACCUMULATOR* macro, which all others should delegate to.
+ */
+#define REGISTER_ACCUMULATOR_CONDITIONALLY(                                                  \
+    key, factory, allowedWithApiStrict, allowedClientType, minVersion, ...)                  \
+    MONGO_INITIALIZER_GENERAL(addToAccumulatorFactoryMap_##key,                              \
+                              ("BeginAccumulatorRegistration"),                              \
+                              ("EndAccumulatorRegistration"))                                \
+    (InitializerContext*) {                                                                  \
+        if (!(__VA_ARGS__)) {                                                                \
+            return;                                                                          \
+        }                                                                                    \
+        AccumulationStatement::registerAccumulator(                                          \
+            "$" #key, (factory), (allowedWithApiStrict), (allowedClientType), (minVersion)); \
     }
 
 /**
@@ -179,6 +209,16 @@ public:
     using Parser = std::function<AccumulationExpression(
         ExpressionContext* const, BSONElement, VariablesParseState)>;
 
+    /**
+     * Associates a Parser with information regarding which contexts it can be used in, including
+     * API Version and FCV.
+     */
+    using ParserRegistration =
+        std::tuple<Parser,
+                   AllowedWithApiStrict,
+                   AllowedWithClientType,
+                   boost::optional<multiversion::FeatureCompatibilityVersion>>;
+
     AccumulationStatement(std::string fieldName, AccumulationExpression expr)
         : fieldName(std::move(fieldName)), expr(std::move(expr)) {}
 
@@ -186,7 +226,9 @@ public:
      * Parses a BSONElement that is an accumulated field, and returns an AccumulationStatement for
      * that accumulated field.
      *
-     * Throws an AssertionException if parsing fails.
+     * Throws an AssertionException if parsing fails, if the configured API version is not
+     * compatible with this AccumulationStatement, or if the Parser is registered under an FCV
+     * greater than the specified maximum allowed FCV.
      */
     static AccumulationStatement parseAccumulationStatement(ExpressionContext* expCtx,
                                                             const BSONElement& elem,
@@ -203,16 +245,15 @@ public:
     static void registerAccumulator(
         std::string name,
         Parser parser,
+        AllowedWithApiStrict allowedWithApiStrict,
+        AllowedWithClientType allowedWithClientType,
         boost::optional<multiversion::FeatureCompatibilityVersion> requiredMinVersion);
 
     /**
      * Retrieves the Parser for the accumulator specified by the given name, and raises an error if
-     * there is no such Parser registered, or the Parser is registered under an FCV greater than the
-     * specified maximum allowed FCV.
+     * there is no such Parser registered.
      */
-    static Parser& getParser(
-        StringData name,
-        boost::optional<multiversion::FeatureCompatibilityVersion> allowedMaxVersion);
+    static ParserRegistration& getParser(StringData name);
 
     // The field name is used to store the results of the accumulation in a result document.
     std::string fieldName;
diff --git a/src/mongo/db/pipeline/accumulator_multi.cpp b/src/mongo/db/pipeline/accumulator_multi.cpp
index 78bd99e740d..912d7c56569 100644
--- a/src/mongo/db/pipeline/accumulator_multi.cpp
+++ b/src/mongo/db/pipeline/accumulator_multi.cpp
@@ -34,39 +34,64 @@ namespace mongo {
 using FirstLastSense = AccumulatorFirstLastN::Sense;
 using MinMaxSense = AccumulatorMinMax::Sense;
 
-REGISTER_ACCUMULATOR_WITH_MIN_VERSION(maxN,
-                                      AccumulatorMinMaxN::parseMinMaxN<MinMaxSense::kMax>,
-                                      multiversion::FeatureCompatibilityVersion::kVersion_5_1);
-REGISTER_ACCUMULATOR_WITH_MIN_VERSION(minN,
-                                      AccumulatorMinMaxN::parseMinMaxN<MinMaxSense::kMin>,
-                                      multiversion::FeatureCompatibilityVersion::kVersion_5_1);
-REGISTER_EXPRESSION_WITH_MIN_VERSION(maxN,
-                                     AccumulatorMinMaxN::parseExpression<MinMaxSense::kMax>,
-                                     AllowedWithApiStrict::kNeverInVersion1,
-                                     AllowedWithClientType::kAny,
-                                     multiversion::FeatureCompatibilityVersion::kVersion_5_1);
-REGISTER_EXPRESSION_WITH_MIN_VERSION(minN,
-                                     AccumulatorMinMaxN::parseExpression<MinMaxSense::kMin>,
-                                     AllowedWithApiStrict::kNeverInVersion1,
-                                     AllowedWithClientType::kAny,
-                                     multiversion::FeatureCompatibilityVersion::kVersion_5_1);
-REGISTER_ACCUMULATOR_WITH_MIN_VERSION(
+// TODO SERVER-52247 Replace boost::none with 'gFeatureFlagExactTopNAccumulator.getVersion()' below
+// once 'gFeatureFlagExactTopNAccumulator' is set to true by default and is configured with an FCV.
+REGISTER_ACCUMULATOR_CONDITIONALLY(
+    maxN,
+    AccumulatorMinMaxN::parseMinMaxN<MinMaxSense::kMax>,
+    AllowedWithApiStrict::kNeverInVersion1,
+    AllowedWithClientType::kAny,
+    boost::none,
+    feature_flags::gFeatureFlagExactTopNAccumulator.isEnabledAndIgnoreFCV());
+REGISTER_ACCUMULATOR_CONDITIONALLY(
+    minN,
+    AccumulatorMinMaxN::parseMinMaxN<MinMaxSense::kMin>,
+    AllowedWithApiStrict::kNeverInVersion1,
+    AllowedWithClientType::kAny,
+    boost::none,
+    feature_flags::gFeatureFlagExactTopNAccumulator.isEnabledAndIgnoreFCV());
+REGISTER_EXPRESSION_CONDITIONALLY(
+    maxN,
+    AccumulatorMinMaxN::parseExpression<MinMaxSense::kMax>,
+    AllowedWithApiStrict::kNeverInVersion1,
+    AllowedWithClientType::kAny,
+    boost::none,
+    feature_flags::gFeatureFlagExactTopNAccumulator.isEnabledAndIgnoreFCV());
+REGISTER_EXPRESSION_CONDITIONALLY(
+    minN,
+    AccumulatorMinMaxN::parseExpression<MinMaxSense::kMin>,
+    AllowedWithApiStrict::kNeverInVersion1,
+    AllowedWithClientType::kAny,
+    boost::none,
+    feature_flags::gFeatureFlagExactTopNAccumulator.isEnabledAndIgnoreFCV());
+REGISTER_ACCUMULATOR_CONDITIONALLY(
     firstN,
     AccumulatorFirstLastN::parseFirstLastN<FirstLastSense::kFirst>,
-    multiversion::FeatureCompatibilityVersion::kVersion_5_1);
-REGISTER_ACCUMULATOR_WITH_MIN_VERSION(lastN,
-                                      AccumulatorFirstLastN::parseFirstLastN<FirstLastSense::kLast>,
-                                      multiversion::FeatureCompatibilityVersion::kVersion_5_1);
-REGISTER_EXPRESSION_WITH_MIN_VERSION(firstN,
-                                     AccumulatorFirstLastN::parseExpression<FirstLastSense::kFirst>,
-                                     AllowedWithApiStrict::kNeverInVersion1,
-                                     AllowedWithClientType::kAny,
-                                     multiversion::FeatureCompatibilityVersion::kVersion_5_1);
-REGISTER_EXPRESSION_WITH_MIN_VERSION(lastN,
-                                     AccumulatorFirstLastN::parseExpression<FirstLastSense::kLast>,
-                                     AllowedWithApiStrict::kNeverInVersion1,
-                                     AllowedWithClientType::kAny,
-                                     multiversion::FeatureCompatibilityVersion::kVersion_5_1);
+    AllowedWithApiStrict::kNeverInVersion1,
+    AllowedWithClientType::kAny,
+    boost::none,
+    feature_flags::gFeatureFlagExactTopNAccumulator.isEnabledAndIgnoreFCV());
+REGISTER_ACCUMULATOR_CONDITIONALLY(
+    lastN,
+    AccumulatorFirstLastN::parseFirstLastN<FirstLastSense::kLast>,
+    AllowedWithApiStrict::kNeverInVersion1,
+    AllowedWithClientType::kAny,
+    boost::none,
+    feature_flags::gFeatureFlagExactTopNAccumulator.isEnabledAndIgnoreFCV());
+REGISTER_EXPRESSION_CONDITIONALLY(
+    firstN,
+    AccumulatorFirstLastN::parseExpression<FirstLastSense::kFirst>,
+    AllowedWithApiStrict::kNeverInVersion1,
+    AllowedWithClientType::kAny,
+    boost::none,
+    feature_flags::gFeatureFlagExactTopNAccumulator.isEnabledAndIgnoreFCV());
+REGISTER_EXPRESSION_CONDITIONALLY(
+    lastN,
+    AccumulatorFirstLastN::parseExpression<FirstLastSense::kLast>,
+    AllowedWithApiStrict::kNeverInVersion1,
+    AllowedWithClientType::kAny,
+    boost::none,
+    feature_flags::gFeatureFlagExactTopNAccumulator.isEnabledAndIgnoreFCV());
 // TODO SERVER-57884 Add $firstN/$lastN as window functions.
 
 AccumulatorN::AccumulatorN(ExpressionContext* const expCtx)
@@ -182,12 +207,6 @@ AccumulationExpression AccumulatorMinMaxN::parseMinMaxN(ExpressionContext* const
         }
     }();
 
-    // TODO SERVER-58379 Remove this uassert once the FCV constants are upgraded and the REGISTER
-    // macros above are updated accordingly.
-    uassert(5787909,
-            str::stream() << "Cannot create " << name << " accumulator if feature flag is disabled",
-            feature_flags::gFeatureFlagExactTopNAccumulator.isEnabled(
-                serverGlobalParams.featureCompatibility));
     uassert(5787900,
             str::stream() << "specification must be an object; found " << elem,
             elem.type() == BSONType::Object);
@@ -279,12 +298,6 @@ AccumulationExpression AccumulatorFirstLastN::parseFirstLastN(ExpressionContext*
         }
     }();
 
-    // TODO SERVER-58379 Remove this uassert once the FCV constants are upgraded and the REGISTER
-    // macros above are updated accordingly.
-    uassert(5787800,
-            str::stream() << "Cannot create " << name << " accumulator if feature flag is disabled",
-            feature_flags::gFeatureFlagExactTopNAccumulator.isEnabled(
-                serverGlobalParams.featureCompatibility));
     uassert(5787801,
             str::stream() << "specification must be an object; found " << elem,
             elem.type() == BSONType::Object);
diff --git a/src/mongo/db/pipeline/document_source_group_test.cpp b/src/mongo/db/pipeline/document_source_group_test.cpp
index de6252449eb..9d62e0206a2 100644
--- a/src/mongo/db/pipeline/document_source_group_test.cpp
+++ b/src/mongo/db/pipeline/document_source_group_test.cpp
@@ -74,7 +74,7 @@ TEST_F(DocumentSourceGroupTest, ShouldBeAbleToPauseLoading) {
     auto expCtx = getExpCtx();
     expCtx->inMongos = true;  // Disallow external sort.
                               // This is the only way to do this in a debug build.
-    auto&& parser = AccumulationStatement::getParser("$sum", boost::none);
+    auto&& [parser, _1, _2, _3] = AccumulationStatement::getParser("$sum");
     auto accumulatorArg = BSON("" << 1);
     auto accExpr = parser(expCtx.get(), accumulatorArg.firstElement(), expCtx->variablesParseState);
     AccumulationStatement countStatement{"count", accExpr};
@@ -111,7 +111,7 @@ TEST_F(DocumentSourceGroupTest, ShouldBeAbleToPauseLoadingWhileSpilled) {
     expCtx->allowDiskUse = true;
     const size_t maxMemoryUsageBytes = 1000;
 
-    auto&& parser = AccumulationStatement::getParser("$push", boost::none);
+    auto&& [parser, _1, _2, _3] = AccumulationStatement::getParser("$push");
     auto accumulatorArg = BSON(""
                                << "$largeStr");
     auto accExpr = parser(expCtx.get(), accumulatorArg.firstElement(), expCtx->variablesParseState);
@@ -154,7 +154,7 @@ TEST_F(DocumentSourceGroupTest, ShouldErrorIfNotAllowedToSpillToDiskAndResultSet
     expCtx->inMongos = true;  // Disallow external sort.
                               // This is the only way to do this in a debug build.
 
-    auto&& parser = AccumulationStatement::getParser("$push", boost::none);
+    auto&& [parser, _1, _2, _3] = AccumulationStatement::getParser("$push");
     auto accumulatorArg = BSON(""
                                << "$largeStr");
     auto accExpr = parser(expCtx.get(), accumulatorArg.firstElement(), expCtx->variablesParseState);
@@ -180,7 +180,7 @@ TEST_F(DocumentSourceGroupTest, ShouldCorrectlyTrackMemoryUsageBetweenPauses) {
     expCtx->inMongos = true;  // Disallow external sort.
                               // This is the only way to do this in a debug build.
 
-    auto&& parser = AccumulationStatement::getParser("$push", boost::none);
+    auto&& [parser, _1, _2, _3] = AccumulationStatement::getParser("$push");
     auto accumulatorArg = BSON(""
                                << "$largeStr");
     auto accExpr = parser(expCtx.get(), accumulatorArg.firstElement(), expCtx->variablesParseState);
diff --git a/src/mongo/db/pipeline/expression.h b/src/mongo/db/pipeline/expression.h
index a0a2d08bdee..cb68e7cef5b 100644
--- a/src/mongo/db/pipeline/expression.h
+++ b/src/mongo/db/pipeline/expression.h
@@ -73,16 +73,37 @@ class DocumentSource;
  * An expression registered this way can be used in any featureCompatibilityVersion and will be
  * considered part of the stable API.
  */
-#define REGISTER_STABLE_EXPRESSION(key, parser)                       \
-    MONGO_INITIALIZER_GENERAL(addToExpressionParserMap_##key,         \
-                              ("BeginExpressionRegistration"),        \
-                              ("EndExpressionRegistration"))          \
-    (InitializerContext*) {                                           \
-        Expression::registerExpression("$" #key,                      \
-                                       (parser),                      \
-                                       AllowedWithApiStrict::kAlways, \
-                                       AllowedWithClientType::kAny,   \
-                                       boost::none);                  \
+#define REGISTER_STABLE_EXPRESSION(key, parser)                      \
+    REGISTER_EXPRESSION_CONDITIONALLY(key,                           \
+                                      parser,                        \
+                                      AllowedWithApiStrict::kAlways, \
+                                      AllowedWithClientType::kAny,   \
+                                      boost::none,                   \
+                                      true);
+
+/**
+ * Like REGISTER_EXPRESSION_WITH_MIN_VERSION, except you can also specify a condition,
+ * evaluated during startup, that decides whether to register the parser.
+ *
+ * For example, you could check a feature flag, and register the parser only when it's enabled.
+ *
+ * Note that the condition is evaluated only once, during a MONGO_INITIALIZER. Don't specify
+ * a condition that can change at runtime, such as FCV. (Feature flags are ok, because they
+ * cannot be toggled at runtime.)
+ *
+ * This is the most general REGISTER_EXPRESSION* macro, which all others should delegate to.
+ */
+#define REGISTER_EXPRESSION_CONDITIONALLY(                                                  \
+    key, parser, allowedWithApiStrict, allowedClientType, minVersion, ...)                  \
+    MONGO_INITIALIZER_GENERAL(addToExpressionParserMap_##key,                               \
+                              ("BeginExpressionRegistration"),                              \
+                              ("EndExpressionRegistration"))                                \
+    (InitializerContext*) {                                                                 \
+        if (!(__VA_ARGS__)) {                                                               \
+            return;                                                                         \
+        }                                                                                   \
+        Expression::registerExpression(                                                     \
+            "$" #key, (parser), (allowedWithApiStrict), (allowedClientType), (minVersion)); \
     }
 
 /**
@@ -106,30 +127,22 @@ class DocumentSource;
  * parser and enforce the 'sometimes' behavior during that invocation. No extra validation will be
  * done here.
  */
-#define REGISTER_EXPRESSION_WITH_MIN_VERSION(                                               \
-    key, parser, allowedWithApiStrict, allowedClientType, minVersion)                       \
-    MONGO_INITIALIZER_GENERAL(addToExpressionParserMap_##key,                               \
-                              ("BeginExpressionRegistration"),                              \
-                              ("EndExpressionRegistration"))                                \
-    (InitializerContext*) {                                                                 \
-        Expression::registerExpression(                                                     \
-            "$" #key, (parser), (allowedWithApiStrict), (allowedClientType), (minVersion)); \
-    }
+#define REGISTER_EXPRESSION_WITH_MIN_VERSION(                         \
+    key, parser, allowedWithApiStrict, allowedClientType, minVersion) \
+    REGISTER_EXPRESSION_CONDITIONALLY(                                \
+        key, parser, allowedWithApiStrict, allowedClientType, minVersion, true)
 
 /**
  * Registers a Parser only if test commands are enabled. Use this if your expression is only used
  * for testing purposes.
  */
-#define REGISTER_TEST_EXPRESSION(key, allowedWithApiStrict, allowedClientType, parser)         \
-    MONGO_INITIALIZER_GENERAL(addToExpressionParserMap_##key,                                  \
-                              ("BeginExpressionRegistration"),                                 \
-                              ("EndExpressionRegistration"))                                   \
-    (InitializerContext*) {                                                                    \
-        if (getTestCommandsEnabled()) {                                                        \
-            Expression::registerExpression(                                                    \
-                "$" #key, (parser), (allowedWithApiStrict), (allowedClientType), boost::none); \
-        }                                                                                      \
-    }
+#define REGISTER_TEST_EXPRESSION(key, allowedWithApiStrict, allowedClientType, parser) \
+    REGISTER_EXPRESSION_CONDITIONALLY(key,                                             \
+                                      parser,                                          \
+                                      allowedWithApiStrict,                            \
+                                      allowedClientType,                               \
+                                      boost::none,                                     \
+                                      getTestCommandsEnabled());
 
 class Expression : public RefCountable {
 public:
diff --git a/src/mongo/db/pipeline/expression_test.cpp b/src/mongo/db/pipeline/expression_test.cpp
index 6dd4c4b9974..e579eece98c 100644
--- a/src/mongo/db/pipeline/expression_test.cpp
+++ b/src/mongo/db/pipeline/expression_test.cpp
@@ -40,6 +40,7 @@
 #include "mongo/db/jsobj.h"
 #include "mongo/db/json.h"
 #include "mongo/db/pipeline/accumulator.h"
+#include "mongo/db/pipeline/accumulator_multi.h"
 #include "mongo/db/pipeline/expression.h"
 #include "mongo/db/pipeline/expression_context_for_test.h"
 #include "mongo/db/query/collation/collator_interface_mock.h"
@@ -154,6 +155,22 @@ void assertExpectedArray(const BSONObj& spec, const BSONArray& expected) {
     ASSERT_VALUE_EQ(result, Value(expected));
 };
 
+/**
+ * Given 'parseFn', parses and evaluates 'spec' and verifies that the result is equal to
+ * 'expected'. Useful when the parser for an expression is unavailable in certain contexts (for
+ * instance, when evaluating an expression that's guarded by a feature flag that's off by default).
+ */
+void parseAndVerifyResults(
+    const std::function<boost::intrusive_ptr<Expression>(
+        ExpressionContext* const, BSONElement, const VariablesParseState&)>& parseFn,
+    const BSONElement& elem,
+    Value expected) {
+    auto expCtx = ExpressionContextForTest{};
+    VariablesParseState vps = expCtx.variablesParseState;
+    auto expr = parseFn(&expCtx, elem, vps);
+    ASSERT_VALUE_EQ(expr->evaluate({}, &expCtx.variables), expected);
+}
+
 /* ------------------------- ExpressionArrayToObject -------------------------- */
 
 TEST(ExpressionArrayToObjectTest, KVFormatSimple) {
@@ -731,31 +748,38 @@ TEST(ExpressionFromAccumulators, Avg) {
 }
 
 TEST(ExpressionFromAccumulators, FirstNLastN) {
-    RAIIServerParameterControllerForTest controller("featureFlagExactTopNAccumulator", true);
+    using Sense = AccumulatorFirstLastN::Sense;
 
     // $firstN
-    assertExpectedArray(fromjson("{$firstN: {n: 3, output: [19, 7, 28, 3, 5]}}"),
-                        BSONArray(fromjson("[19, 7, 28]")));
-    assertExpectedArray(fromjson("{$firstN: {n: 6, output: [19, 7, 28, 3, 5]}}"),
-                        BSONArray(fromjson("[19, 7, 28, 3, 5]")));
-    assertExpectedArray(fromjson("{$firstN: {n: 3, output: [1,2,3,4,5,6]}}"),
-                        BSONArray(fromjson("[1,2,3]")));
-    assertExpectedArray(fromjson("{$firstN: {n: 3, output: [1,2,null,null]}}"),
-                        BSONArray(fromjson("[1,2,null]")));
-    assertExpectedArray(fromjson("{$firstN: {n: 3, output: [1.1, 2.713, 3, 3.4]}}"),
-                        BSONArray(fromjson("[1.1, 2.713, 3]")));
+    auto firstNFn = [&](const BSONObj& spec, const BSONArray& expected) {
+        parseAndVerifyResults(AccumulatorFirstLastN::parseExpression<Sense::kFirst>,
+                              spec.firstElement(),
+                              Value(expected));
+    };
+    firstNFn(fromjson("{$firstN: {n: 3, output: [19, 7, 28, 3, 5]}}"),
+             BSONArray(fromjson("[19, 7, 28]")));
+    firstNFn(fromjson("{$firstN: {n: 6, output: [19, 7, 28, 3, 5]}}"),
+             BSONArray(fromjson("[19, 7, 28, 3, 5]")));
+    firstNFn(fromjson("{$firstN: {n: 3, output: [1,2,3,4,5,6]}}"), BSONArray(fromjson("[1,2,3]")));
+    firstNFn(fromjson("{$firstN: {n: 3, output: [1,2,null,null]}}"),
+             BSONArray(fromjson("[1,2,null]")));
+    firstNFn(fromjson("{$firstN: {n: 3, output: [1.1, 2.713, 3, 3.4]}}"),
+             BSONArray(fromjson("[1.1, 2.713, 3]")));
 
     // $lastN
-    assertExpectedArray(fromjson("{$lastN: {n: 3, output: [19, 7, 28, 3, 5]}}"),
-                        BSONArray(fromjson("[28,3,5]")));
-    assertExpectedArray(fromjson("{$lastN: {n: 6, output: [19, 7, 28, 3, 5]}}"),
-                        BSONArray(fromjson("[19, 7, 28, 3, 5]")));
-    assertExpectedArray(fromjson("{$lastN: {n: 3, output: [3,2,1,4,5,6]}}"),
-                        BSONArray(fromjson("[4,5,6]")));
-    assertExpectedArray(fromjson("{$lastN: {n: 3, output: [1,2,null,3]}}"),
-                        BSONArray(fromjson("[2,null,3]")));
-    assertExpectedArray(fromjson("{$lastN: {n: 3, output: [3, 2.713, 1.1, 2.7]}}"),
-                        BSONArray(fromjson("[2.713, 1.1, 2.7]")));
+    auto lastNFn = [&](const BSONObj& spec, const BSONArray& expected) {
+        parseAndVerifyResults(AccumulatorFirstLastN::parseExpression<Sense::kLast>,
+                              spec.firstElement(),
+                              Value(expected));
+    };
+    lastNFn(fromjson("{$lastN: {n: 3, output: [19, 7, 28, 3, 5]}}"),
+            BSONArray(fromjson("[28,3,5]")));
+    lastNFn(fromjson("{$lastN: {n: 6, output: [19, 7, 28, 3, 5]}}"),
+            BSONArray(fromjson("[19, 7, 28, 3, 5]")));
+    lastNFn(fromjson("{$lastN: {n: 3, output: [3,2,1,4,5,6]}}"), BSONArray(fromjson("[4,5,6]")));
+    lastNFn(fromjson("{$lastN: {n: 3, output: [1,2,null,3]}}"), BSONArray(fromjson("[2,null,3]")));
+    lastNFn(fromjson("{$lastN: {n: 3, output: [3, 2.713, 1.1, 2.7]}}"),
+            BSONArray(fromjson("[2.713, 1.1, 2.7]")));
 }
 
 TEST(ExpressionFromAccumulators, Max) {
@@ -782,31 +806,35 @@ TEST(ExpressionFromAccumulators, Min) {
 }
 
 TEST(ExpressionFromAccumulators, MinNMaxN) {
-    RAIIServerParameterControllerForTest controller("featureFlagExactTopNAccumulator", true);
+    using Sense = AccumulatorMinMax::Sense;
+    auto maxNFn = [&](const BSONObj& spec, const BSONArray& expected) {
+        parseAndVerifyResults(
+            AccumulatorMinMaxN::parseExpression<Sense::kMax>, spec.firstElement(), Value(expected));
+    };
 
     // $maxN
-    assertExpectedArray(fromjson("{$maxN: {n: 3, output: [19, 7, 28, 3, 5]}}"),
-                        BSONArray(fromjson("[28, 19, 7]")));
-    assertExpectedArray(fromjson("{$maxN: {n: 6, output: [19, 7, 28, 3, 5]}}"),
-                        BSONArray(fromjson("[28, 19, 7, 5, 3]")));
-    assertExpectedArray(fromjson("{$maxN: {n: 3, output: [1,2,3]}}"),
-                        BSONArray(fromjson("[3,2,1]")));
-    assertExpectedArray(fromjson("{$maxN: {n: 3, output: [1,2,null]}}"),
-                        BSONArray(fromjson("[2,1]")));
-    assertExpectedArray(fromjson("{$maxN: {n: 3, output: [1.1, 2.713, 3]}}"),
-                        BSONArray(fromjson("[3, 2.713, 1.1]")));
+    maxNFn(fromjson("{$maxN: {n: 3, output: [19, 7, 28, 3, 5]}}"),
+           BSONArray(fromjson("[28, 19, 7]")));
+    maxNFn(fromjson("{$maxN: {n: 6, output: [19, 7, 28, 3, 5]}}"),
+           BSONArray(fromjson("[28, 19, 7, 5, 3]")));
+    maxNFn(fromjson("{$maxN: {n: 3, output: [1,2,3]}}"), BSONArray(fromjson("[3,2,1]")));
+    maxNFn(fromjson("{$maxN: {n: 3, output: [1,2,null]}}"), BSONArray(fromjson("[2,1]")));
+    maxNFn(fromjson("{$maxN: {n: 3, output: [1.1, 2.713, 3]}}"),
+           BSONArray(fromjson("[3, 2.713, 1.1]")));
+
+    auto minNFn = [&](const BSONObj& spec, const BSONArray& expected) {
+        parseAndVerifyResults(
+            AccumulatorMinMaxN::parseExpression<Sense::kMin>, spec.firstElement(), Value(expected));
+    };
 
     // $minN
-    assertExpectedArray(fromjson("{$minN: {n: 3, output: [19, 7, 28, 3, 5]}}"),
-                        BSONArray(fromjson("[3,5,7]")));
-    assertExpectedArray(fromjson("{$minN: {n: 6, output: [19, 7, 28, 3, 5]}}"),
-                        BSONArray(fromjson("[3,5,7,19,28]")));
-    assertExpectedArray(fromjson("{$minN: {n: 3, output: [3,2,1]}}"),
-                        BSONArray(fromjson("[1,2,3]")));
-    assertExpectedArray(fromjson("{$minN: {n: 3, output: [1,2,null]}}"),
-                        BSONArray(fromjson("[1,2]")));
-    assertExpectedArray(fromjson("{$minN: {n: 3, output: [3, 2.713, 1.1]}}"),
-                        BSONArray(fromjson("[1.1, 2.713, 3]")));
+    minNFn(fromjson("{$minN: {n: 3, output: [19, 7, 28, 3, 5]}}"), BSONArray(fromjson("[3,5,7]")));
+    minNFn(fromjson("{$minN: {n: 6, output: [19, 7, 28, 3, 5]}}"),
+           BSONArray(fromjson("[3,5,7,19, 28]")));
+    minNFn(fromjson("{$minN: {n: 3, output: [3,2,1]}}"), BSONArray(fromjson("[1,2,3]")));
+    minNFn(fromjson("{$minN: {n: 3, output: [1,2,null]}}"), BSONArray(fromjson("[1,2]")));
+    minNFn(fromjson("{$minN: {n: 3, output: [3, 2.713, 1.1]}}"),
+           BSONArray(fromjson("[1.1, 2.713, 3]")));
 }
 
 TEST(ExpressionFromAccumulators, Sum) {
diff --git a/src/mongo/db/pipeline/sharded_union_test.cpp b/src/mongo/db/pipeline/sharded_union_test.cpp
index e9748aa0a57..395990e5309 100644
--- a/src/mongo/db/pipeline/sharded_union_test.cpp
+++ b/src/mongo/db/pipeline/sharded_union_test.cpp
@@ -205,7 +205,7 @@ TEST_F(ShardedUnionTest, CorrectlySplitsSubPipelineIfRefreshedDistributionRequir
     auto shards = setupNShards(2);
     const auto cm = loadRoutingTableWithTwoChunksAndTwoShards(kTestAggregateNss);
 
-    auto&& parser = AccumulationStatement::getParser("$sum", boost::none);
+    auto&& [parser, _1, _2, _3] = AccumulationStatement::getParser("$sum");
     auto accumulatorArg = BSON("" << 1);
     auto sumStatement =
         parser(expCtx().get(), accumulatorArg.firstElement(), expCtx()->variablesParseState);
@@ -293,7 +293,7 @@ TEST_F(ShardedUnionTest, AvoidsSplittingSubPipelineIfRefreshedDistributionDoesNo
     auto shards = setupNShards(2);
     const auto cm = loadRoutingTableWithTwoChunksAndTwoShards(kTestAggregateNss);
 
-    auto&& parser = AccumulationStatement::getParser("$sum", boost::none);
+    auto&& [parser, _1, _2, _3] = AccumulationStatement::getParser("$sum");
     auto accumulatorArg = BSON("" << 1);
     auto sumStatement =
         parser(expCtx().get(), accumulatorArg.firstElement(), expCtx()->variablesParseState);
@@ -428,7 +428,7 @@ TEST_F(ShardedUnionTest, ForwardsReadConcernToRemotes) {
     setupNShards(2);
     loadRoutingTableWithTwoChunksAndTwoShards(kTestAggregateNss);
 
-    auto&& parser = AccumulationStatement::getParser("$sum", boost::none);
+    auto&& [parser, _1, _2, _3] = AccumulationStatement::getParser("$sum");
     auto accumulatorArg = BSON("" << 1);
     auto sumExpression =
         parser(expCtx().get(), accumulatorArg.firstElement(), expCtx()->variablesParseState);
diff --git a/src/mongo/db/pipeline/window_function/window_function_expression.cpp b/src/mongo/db/pipeline/window_function/window_function_expression.cpp
index 118db84f280..729d47aad0d 100644
--- a/src/mongo/db/pipeline/window_function/window_function_expression.cpp
+++ b/src/mongo/db/pipeline/window_function/window_function_expression.cpp
@@ -29,6 +29,7 @@
 
 #include "mongo/platform/basic.h"
 
+#include "mongo/db/commands/feature_compatibility_version_documentation.h"
 #include "mongo/db/pipeline/accumulation_statement.h"
 #include "mongo/db/pipeline/document_source_add_fields.h"
 #include "mongo/db/pipeline/document_source_project.h"
@@ -53,26 +54,45 @@ using MinMaxSense = AccumulatorMinMax::Sense;
 REGISTER_WINDOW_FUNCTION(derivative, ExpressionDerivative::parse);
 REGISTER_WINDOW_FUNCTION(first, ExpressionFirst::parse);
 REGISTER_WINDOW_FUNCTION(last, ExpressionLast::parse);
-REGISTER_WINDOW_FUNCTION(minN, ExpressionMinMaxN<MinMaxSense::kMin>::parse);
-REGISTER_WINDOW_FUNCTION(maxN, ExpressionMinMaxN<MinMaxSense::kMax>::parse);
 
-StringMap<Expression::Parser> Expression::parserMap;
+// TODO SERVER-52247 Replace boost::none with 'gFeatureFlagExactTopNAccumulator.getVersion()' below
+// once 'gFeatureFlagExactTopNAccumulator' is set to true by default and is configured with an FCV.
+REGISTER_WINDOW_FUNCTION_CONDITIONALLY(
+    minN,
+    ExpressionMinMaxN<MinMaxSense::kMin>::parse,
+    boost::none,
+    feature_flags::gFeatureFlagExactTopNAccumulator.isEnabledAndIgnoreFCV());
+REGISTER_WINDOW_FUNCTION_CONDITIONALLY(
+    maxN,
+    ExpressionMinMaxN<MinMaxSense::kMax>::parse,
+    boost::none,
+    feature_flags::gFeatureFlagExactTopNAccumulator.isEnabledAndIgnoreFCV());
+
+StringMap<Expression::ExpressionParserRegistration> Expression::parserMap;
 
 intrusive_ptr<Expression> Expression::parse(BSONObj obj,
                                             const optional<SortPattern>& sortBy,
                                             ExpressionContext* expCtx) {
-
     for (const auto& field : obj) {
         // Check if window function is $-prefixed.
         auto fieldName = field.fieldNameStringData();
 
         if (fieldName.startsWith("$"_sd)) {
-
-            if (auto parser = parserMap.find(field.fieldNameStringData());
-                parser != parserMap.end()) {
+            auto exprName = field.fieldNameStringData();
+            if (auto parserFCV = parserMap.find(exprName); parserFCV != parserMap.end()) {
                 // Found one valid window function. If there are multiple window functions they will
                 // be caught as invalid arguments to the Expression parser later.
-                return parser->second(obj, sortBy, expCtx);
+                const auto& parser = parserFCV->second.first;
+                auto fcv = parserFCV->second.second;
+                uassert(ErrorCodes::QueryFeatureNotAllowed,
+                        str::stream()
+                            << exprName
+                            << " is not allowed in the current feature compatibility version. See "
+                            << feature_compatibility_version_documentation::kCompatibilityLink
+                            << " for more information.",
+                        !expCtx->maxFeatureCompatibilityVersion || !fcv ||
+                            (*fcv <= *expCtx->maxFeatureCompatibilityVersion));
+                return parser(obj, sortBy, expCtx);
             }
             // The window function provided in the window function expression is invalid.
 
@@ -99,9 +119,13 @@ intrusive_ptr<Expression> Expression::parse(BSONObj obj,
                        : ", "s + obj.firstElementFieldNameStringData()));
 }
 
-void Expression::registerParser(std::string functionName, Parser parser) {
+void Expression::registerParser(
+    std::string functionName,
+    Parser parser,
+    boost::optional<multiversion::FeatureCompatibilityVersion> requiredMinVersion) {
     invariant(parserMap.find(functionName) == parserMap.end());
-    parserMap.emplace(std::move(functionName), std::move(parser));
+    ExpressionParserRegistration r(parser, requiredMinVersion);
+    parserMap.emplace(std::move(functionName), std::move(r));
 }
 
 
@@ -229,12 +253,6 @@ boost::intrusive_ptr<Expression> ExpressionMinMaxN<S>::parse(
             return AccumulatorMaxN::getName();
         }
     }();
-    uassert(5788502,
-            str::stream() << "Cannot create " << name
-                          << " accumulator in $setWindowFields"
-                             " if feature flag is disabled",
-            feature_flags::gFeatureFlagExactTopNAccumulator.isEnabled(
-                serverGlobalParams.featureCompatibility));
 
     boost::intrusive_ptr<::mongo::Expression> nExpr;
     boost::intrusive_ptr<::mongo::Expression> outputExpr;
diff --git a/src/mongo/db/pipeline/window_function/window_function_expression.h b/src/mongo/db/pipeline/window_function/window_function_expression.h
index 4cbbd88914e..ab2675f1dbd 100644
--- a/src/mongo/db/pipeline/window_function/window_function_expression.h
+++ b/src/mongo/db/pipeline/window_function/window_function_expression.h
@@ -46,21 +46,33 @@ class WindowFunctionExec;
 class PartitionIterator;
 }  // namespace mongo
 
-#define REGISTER_WINDOW_FUNCTION(name, parser)                                   \
-    MONGO_INITIALIZER_GENERAL(addToWindowFunctionMap_##name,                     \
-                              ("BeginWindowFunctionRegistration"),               \
-                              ("EndWindowFunctionRegistration"))                 \
-    (InitializerContext*) {                                                      \
-        ::mongo::window_function::Expression::registerParser("$" #name, parser); \
+#define REGISTER_WINDOW_FUNCTION(name, parser) \
+    REGISTER_WINDOW_FUNCTION_CONDITIONALLY(name, parser, boost::none, true)
+
+#define REGISTER_WINDOW_FUNCTION_WITH_MIN_VERSION(name, parser, minVersion) \
+    REGISTER_WINDOW_FUNCTION_CONDITIONALLY(name, parser, minVersion, true)
+
+#define REGISTER_WINDOW_FUNCTION_CONDITIONALLY(name, parser, minVersion, ...)                \
+    MONGO_INITIALIZER_GENERAL(addToWindowFunctionMap_##name,                                 \
+                              ("BeginWindowFunctionRegistration"),                           \
+                              ("EndWindowFunctionRegistration"))                             \
+    (InitializerContext*) {                                                                  \
+        if (!(__VA_ARGS__)) {                                                                \
+            return;                                                                          \
+        }                                                                                    \
+        ::mongo::window_function::Expression::registerParser("$" #name, parser, minVersion); \
     }
 
-#define REGISTER_REMOVABLE_WINDOW_FUNCTION(name, accumClass, wfClass)                              \
-    MONGO_INITIALIZER_GENERAL(addToWindowFunctionMap_##name,                                       \
-                              ("BeginWindowFunctionRegistration"),                                 \
-                              ("EndWindowFunctionRegistration"))                                   \
-    (InitializerContext*) {                                                                        \
-        ::mongo::window_function::Expression::registerParser(                                      \
-            "$" #name, ::mongo::window_function::ExpressionRemovable<accumClass, wfClass>::parse); \
+
+#define REGISTER_REMOVABLE_WINDOW_FUNCTION(name, accumClass, wfClass)                  \
+    MONGO_INITIALIZER_GENERAL(addToWindowFunctionMap_##name,                           \
+                              ("BeginWindowFunctionRegistration"),                     \
+                              ("EndWindowFunctionRegistration"))                       \
+    (InitializerContext*) {                                                            \
+        ::mongo::window_function::Expression::registerParser(                          \
+            "$" #name,                                                                 \
+            ::mongo::window_function::ExpressionRemovable<accumClass, wfClass>::parse, \
+            boost::none);                                                              \
     }
 
 
@@ -107,7 +119,12 @@ public:
      * described above, because some parsers need to switch on the function name.
      */
     using Parser = std::function<decltype(parse)>;
-    static void registerParser(std::string functionName, Parser parser);
+    using ExpressionParserRegistration =
+        std::pair<Parser, boost::optional<multiversion::FeatureCompatibilityVersion>>;
+    static void registerParser(
+        std::string functionName,
+        Parser parser,
+        boost::optional<multiversion::FeatureCompatibilityVersion> requiredMinVersion);
 
     /**
      * Is this a function that the parser knows about?
@@ -181,7 +198,7 @@ protected:
      * In these cases, this field is ignored.
      */
     WindowBounds _bounds;
-    static StringMap<Parser> parserMap;
+    static StringMap<ExpressionParserRegistration> parserMap;
 };
 
 template <typename NonRemovableType>