SERVER-47398 Start sending 'let' for aggregates.

Replaces usages of 'runtimeConstants' with 'let' parameters. This does
not replace usages for the write commands and for $merge. We will need
follow-up work for that.

21 files changed, 301 insertions, 175 deletions

diff --git a/jstests/aggregation/expressions/filter.js b/jstests/aggregation/expressions/filter.js
index 70f3deca9b2..0f1489e434d 100644
--- a/jstests/aggregation/expressions/filter.js
+++ b/jstests/aggregation/expressions/filter.js
@@ -238,7 +238,7 @@ assertErrorCode(coll, [{$project: {b: {$filter: filterDoc}}}], 28650);
 
 // 'as' is not a valid variable name.
 filterDoc = {input: '$a', as: '$x', cond: true};
-assertErrorCode(coll, [{$project: {b: {$filter: filterDoc}}}], 16867);
+assertErrorCode(coll, [{$project: {b: {$filter: filterDoc}}}], ErrorCodes.FailedToParse);
 
 // 'input' is not an array.
 filterDoc = {input: 'string', as: 'x', cond: true};
diff --git a/jstests/aggregation/expressions/let.js b/jstests/aggregation/expressions/let.js
index a4044b7c0dc..bd6e83d3e9f 100644
--- a/jstests/aggregation/expressions/let.js
+++ b/jstests/aggregation/expressions/let.js
@@ -114,13 +114,18 @@ assert.eq(coll.aggregate({$group: {_id: 0, objs: {$push: '$$CURRENT'}}}).toArray
           [{_id: 0, objs: [{_id: 'obj'}]}]);
 
 // Check name validity checks.
-assertErrorCode(coll, {$project: {a: {$let: {vars: {ROOT: 1}, in : '$$ROOT'}}}}, 16867);
-assertErrorCode(coll, {$project: {a: {$let: {vars: {FOO: 1}, in : '$$FOO'}}}}, 16867);
-assertErrorCode(coll, {$project: {a: {$let: {vars: {_underbar: 1}, in : '$$FOO'}}}}, 16867);
-assertErrorCode(coll, {$project: {a: {$let: {vars: {'a.b': 1}, in : '$$FOO'}}}}, 16868);
-assertErrorCode(coll, {$project: {a: {$let: {vars: {'a b': 1}, in : '$$FOO'}}}}, 16868);
-assertErrorCode(coll, {$project: {a: '$$_underbar'}}, 16867);
-assertErrorCode(coll, {$project: {a: '$$with spaces'}}, 16868);
+assertErrorCode(
+    coll, {$project: {a: {$let: {vars: {ROOT: 1}, in : '$$ROOT'}}}}, ErrorCodes.FailedToParse);
+assertErrorCode(
+    coll, {$project: {a: {$let: {vars: {FOO: 1}, in : '$$FOO'}}}}, ErrorCodes.FailedToParse);
+assertErrorCode(
+    coll, {$project: {a: {$let: {vars: {_underbar: 1}, in : '$$FOO'}}}}, ErrorCodes.FailedToParse);
+assertErrorCode(
+    coll, {$project: {a: {$let: {vars: {'a.b': 1}, in : '$$FOO'}}}}, ErrorCodes.FailedToParse);
+assertErrorCode(
+    coll, {$project: {a: {$let: {vars: {'a b': 1}, in : '$$FOO'}}}}, ErrorCodes.FailedToParse);
+assertErrorCode(coll, {$project: {a: '$$_underbar'}}, ErrorCodes.FailedToParse);
+assertErrorCode(coll, {$project: {a: '$$with spaces'}}, ErrorCodes.FailedToParse);
 
 // Verify that variables defined in '$let' cannot be used to initialize other variables.
 assertErrorCode(
diff --git a/jstests/aggregation/bugs/server9841.js b/jstests/aggregation/expressions/map.js
index 3c3fd1a1369..d8b3794aea9 100644
--- a/jstests/aggregation/bugs/server9841.js
+++ b/jstests/aggregation/expressions/map.js
@@ -2,7 +2,8 @@
 // @tags: [
 //   sbe_incompatible,
 // ]
-
+(function() {
+"use strict";
 load('jstests/aggregation/extras/utils.js');
 var t = db.server9841;
 t.drop();
@@ -32,7 +33,9 @@ test({$map: {input: "$mixed", as: "var", in : '$$var.a'}},
 test({$map: {input: "$null", as: "var", in : '$$var'}}, null);
 
 // can't set ROOT
-assertErrorCode(t, {$project: {a: {$map: {input: "$simple", as: "ROOT", in : '$$ROOT'}}}}, 16867);
+assertErrorCode(t,
+                {$project: {a: {$map: {input: "$simple", as: "ROOT", in : '$$ROOT'}}}},
+                ErrorCodes.FailedToParse);
 
 // error on non-array
 assertErrorCode(t, {$project: {a: {$map: {input: "$notArray", as: "var", in : '$$var'}}}}, 16883);
@@ -45,3 +48,4 @@ assertErrorCode(t, {$project: {a: {$map: {input: "$simple", as: "var"}}}}, 16882
 
 // 'in' uses undefined variable name.
 assertErrorCode(t, {$project: {a: {$map: {input: "$simple", in : '$$var'}}}}, 17276);
+}());
diff --git a/jstests/aggregation/variables/remove_system_variable.js b/jstests/aggregation/variables/remove_system_variable.js
index 43de402a8ed..04341180aad 100644
--- a/jstests/aggregation/variables/remove_system_variable.js
+++ b/jstests/aggregation/variables/remove_system_variable.js
@@ -63,7 +63,7 @@ assert.commandFailedWithCode(db.runCommand({
         {$project: {_id: 0, a: {$let: {vars: {"REMOVE": 3}, in : {b: "$$REMOVE", c: 2}}}}}
     ]
 }),
-                             16867);
+                             ErrorCodes.FailedToParse);
 
 // Test that $$REMOVE, $$CURRENT, $$ROOT, and user-defined variables can all be used together.
 assert.eq(
diff --git a/jstests/aggregation/variables/runtime_constants.js b/jstests/aggregation/variables/runtime_constants.js
index c01dd20f223..49d305cb4c7 100644
--- a/jstests/aggregation/variables/runtime_constants.js
+++ b/jstests/aggregation/variables/runtime_constants.js
@@ -16,11 +16,11 @@ assert.commandWorked(coll.insert({x: true}));
 assert.commandFailedWithCode(
     db.runCommand(
         {aggregate: coll.getName(), pipeline: [{$addFields: {testField: "$$IS_MR"}}], cursor: {}}),
-    4631101);
+    51144);
 
 // Runtime constant $$JS_SCOPE is unable to be retrieved by users.
 assert.commandFailedWithCode(
     db.runCommand(
         {aggregate: coll.getName(), pipeline: [{$addFields: {field: "$$JS_SCOPE"}}], cursor: {}}),
-    4631100);
+    51144);
 })();
diff --git a/jstests/core/command_let_variables.js b/jstests/core/command_let_variables.js
index 08fd250c40d..0540e782218 100644
--- a/jstests/core/command_let_variables.js
+++ b/jstests/core/command_let_variables.js
@@ -290,31 +290,31 @@ assert.eq(result.length, 0);
 assert.commandFailedWithCode(
     testDB.runCommand(
         {aggregate: coll.getName(), pipeline: [], cursor: {}, let : {Reserved: "failure"}}),
-    16867);
+    ErrorCodes.FailedToParse);
 assert.commandFailedWithCode(
     testDB.runCommand(
         {aggregate: coll.getName(), pipeline: [], cursor: {}, let : {NOW: "failure"}}),
-    16867);
+    4738901);
 assert.commandFailedWithCode(
     testDB.runCommand(
         {aggregate: coll.getName(), pipeline: [], cursor: {}, let : {CLUSTER_TIME: "failure"}}),
-    16867);
+    4738901);
 assert.commandFailedWithCode(
     testDB.runCommand(
         {aggregate: coll.getName(), pipeline: [], cursor: {}, let : {IS_MR: "failure"}}),
-    16867);
+    4738901);
 assert.commandFailedWithCode(
     testDB.runCommand(
         {aggregate: coll.getName(), pipeline: [], cursor: {}, let : {JS_SCOPE: "failure"}}),
-    16867);
+    4738901);
 assert.commandFailedWithCode(
     testDB.runCommand(
         {aggregate: coll.getName(), pipeline: [], cursor: {}, let : {ROOT: "failure"}}),
-    16867);
+    ErrorCodes.FailedToParse);
 assert.commandFailedWithCode(
     testDB.runCommand(
         {aggregate: coll.getName(), pipeline: [], cursor: {}, let : {REMOVE: "failure"}}),
-    16867);
+    ErrorCodes.FailedToParse);
 
 // Test that let variables can be used within views.
 assert.commandWorked(testDB.runCommand({
diff --git a/src/mongo/db/exec/document_value/value.h b/src/mongo/db/exec/document_value/value.h
index 4e99aeab9e9..9b8e795a4b0 100644
--- a/src/mongo/db/exec/document_value/value.h
+++ b/src/mongo/db/exec/document_value/value.h
@@ -193,6 +193,10 @@ public:
         return Date == getType() || bsonTimestamp == getType() || jstOID == getType();
     }
 
+    bool isObject() const {
+        return getType() == BSONType::Object;
+    }
+
     /// Get the BSON type of the field.
     BSONType getType() const {
         return _storage.bsonType();
diff --git a/src/mongo/db/pipeline/document_source_lookup_test.cpp b/src/mongo/db/pipeline/document_source_lookup_test.cpp
index 5d8bedf72b4..66c35191000 100644
--- a/src/mongo/db/pipeline/document_source_lookup_test.cpp
+++ b/src/mongo/db/pipeline/document_source_lookup_test.cpp
@@ -341,7 +341,7 @@ TEST_F(DocumentSourceLookUpTest, RejectsInvalidLetVariableName) {
                 .firstElement(),
             expCtx),
         AssertionException,
-        16866);
+        ErrorCodes::FailedToParse);
 
     ASSERT_THROWS_CODE(
         DocumentSourceLookUp::createFromBson(
@@ -356,7 +356,7 @@ TEST_F(DocumentSourceLookUpTest, RejectsInvalidLetVariableName) {
                 .firstElement(),
             expCtx),
         AssertionException,
-        16867);
+        ErrorCodes::FailedToParse);
 
     ASSERT_THROWS_CODE(
         DocumentSourceLookUp::createFromBson(
@@ -371,7 +371,7 @@ TEST_F(DocumentSourceLookUpTest, RejectsInvalidLetVariableName) {
                 .firstElement(),
             expCtx),
         AssertionException,
-        16868);
+        ErrorCodes::FailedToParse);
 }
 
 TEST_F(DocumentSourceLookUpTest, ShouldBeAbleToReParseSerializedStage) {
diff --git a/src/mongo/db/pipeline/expression_context.h b/src/mongo/db/pipeline/expression_context.h
index 4366ae30dcd..c6a20a28ab9 100644
--- a/src/mongo/db/pipeline/expression_context.h
+++ b/src/mongo/db/pipeline/expression_context.h
@@ -266,10 +266,6 @@ public:
         _resolvedNamespaces = std::move(resolvedNamespaces);
     }
 
-    const LegacyRuntimeConstants& getLegacyRuntimeConstants() const {
-        return variables.getLegacyRuntimeConstants();
-    }
-
     /**
      * Retrieves the Javascript Scope for the current thread or creates a new one if it has not been
      * created yet. Initializes the Scope with the 'jsScope' variables from the runtimeConstants.
@@ -282,15 +278,17 @@ public:
         uassert(31264,
                 "Cannot run server-side javascript without the javascript engine enabled",
                 getGlobalScriptEngine());
-        const auto& runtimeConstants = getLegacyRuntimeConstants();
-        const boost::optional<bool> isMapReduceCommand = runtimeConstants.getIsMapReduce();
+        const auto isMapReduce =
+            (variables.hasValue(Variables::kIsMapReduceId) &&
+             variables.getValue(Variables::kIsMapReduceId).getType() == BSONType::Bool &&
+             variables.getValue(Variables::kIsMapReduceId).coerceToBool());
         if (inMongos) {
             invariant(!forceLoadOfStoredProcedures);
-            invariant(!isMapReduceCommand);
+            invariant(!isMapReduce);
         }
 
         // Stored procedures are only loaded for the $where expression and MapReduce command.
-        const bool loadStoredProcedures = forceLoadOfStoredProcedures || isMapReduceCommand;
+        const bool loadStoredProcedures = forceLoadOfStoredProcedures || isMapReduce;
 
         if (hasWhereClause && !loadStoredProcedures) {
             uasserted(4649200,
@@ -298,9 +296,13 @@ public:
                       "$where.");
         }
 
-        const boost::optional<mongo::BSONObj>& scope = runtimeConstants.getJsScope();
-        return JsExecution::get(
-            opCtx, scope.get_value_or(BSONObj()), ns.db(), loadStoredProcedures, jsHeapLimitMB);
+        auto scopeObj = BSONObj();
+        if (variables.hasValue(Variables::kJsScopeId)) {
+            auto scopeVar = variables.getValue(Variables::kJsScopeId);
+            invariant(scopeVar.isObject());
+            scopeObj = scopeVar.getDocument().toBson();
+        }
+        return JsExecution::get(opCtx, scopeObj, ns.db(), loadStoredProcedures, jsHeapLimitMB);
     }
 
     // The explain verbosity requested by the user, or boost::none if no explain was requested.
diff --git a/src/mongo/db/pipeline/expression_context_test.cpp b/src/mongo/db/pipeline/expression_context_test.cpp
index a4b23f75752..b105dce531b 100644
--- a/src/mongo/db/pipeline/expression_context_test.cpp
+++ b/src/mongo/db/pipeline/expression_context_test.cpp
@@ -229,7 +229,7 @@ TEST_F(ExpressionContextTest, ParametersCauseGracefulFailuresIfUppercase) {
                                             BSON("A" << 12),
                                             false}),
         DBException,
-        16867);
+        ErrorCodes::FailedToParse);
 }
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp b/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp
index 072f57bb9b7..102d9bd74ee 100644
--- a/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp
+++ b/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp
@@ -633,10 +633,11 @@ Update CommonMongodProcessInterface::buildUpdateOp(
         wcb.setBypassDocumentValidation(expCtx->bypassDocumentValidation);
         return wcb;
     }());
-    updateOp.setLegacyRuntimeConstants(expCtx->getLegacyRuntimeConstants());
-    if (auto letParams = expCtx->variablesParseState.serializeUserVariables(expCtx->variables);
-        !letParams.isEmpty()) {
-        updateOp.setLet(letParams);
+    auto [constants, letParams] =
+        expCtx->variablesParseState.transitionalCompatibilitySerialize(expCtx->variables);
+    updateOp.setLegacyRuntimeConstants(std::move(constants));
+    if (!letParams.isEmpty()) {
+        updateOp.setLet(std::move(letParams));
     }
     return updateOp;
 }
diff --git a/src/mongo/db/pipeline/sharded_agg_helpers.cpp b/src/mongo/db/pipeline/sharded_agg_helpers.cpp
index 7412ac0b00c..93d4c5dadc8 100644
--- a/src/mongo/db/pipeline/sharded_agg_helpers.cpp
+++ b/src/mongo/db/pipeline/sharded_agg_helpers.cpp
@@ -125,11 +125,29 @@ RemoteCursor openChangeStreamNewShardMonitor(const boost::intrusive_ptr<Expressi
 BSONObj genericTransformForShards(MutableDocument&& cmdForShards,
                                   const boost::intrusive_ptr<ExpressionContext>& expCtx,
                                   boost::optional<ExplainOptions::Verbosity> explainVerbosity,
-                                  const boost::optional<LegacyRuntimeConstants>& constants,
                                   BSONObj collationObj) {
-    if (constants) {
+    // Serialize the variables.
+    // Check whether we are in a mixed-version cluster and so must use the old serialization format.
+    // This is only tricky in the case we are sending an aggregate from shard to shard. For this
+    // scenario we can rely on feature compatibility version to detect when this is safe. Feature
+    // compatibility version is not generally accurate on mongos since it was intended to guard
+    // changes to data format and mongos has no persisted state. However, mongos is upgraded last
+    // after all the shards, so any recipient will understand the 'let' parameter.
+    // TODO SERVER-52900 This code can be remove when we branch for the next LTS release.
+    if (serverGlobalParams.clusterRole == ClusterRole::ShardServer &&
+        !serverGlobalParams.featureCompatibility.isGreaterThanOrEqualTo(
+            ServerGlobalParams::FeatureCompatibility::Version::kVersion49)) {
+        // A mixed version cluster. Use the old format to be sure it is understood.
+        auto [legacyRuntimeConstants, unusedSerializedVariables] =
+            expCtx->variablesParseState.transitionalCompatibilitySerialize(expCtx->variables);
+
         cmdForShards[AggregationRequest::kLegacyRuntimeConstantsName] =
-            Value(constants.get().toBSON());
+            Value(legacyRuntimeConstants.toBSON());
+    } else {
+        // Either this is a "modern" cluster or we are a mongos and can assume the shards are
+        // "modern" and will understand the 'let' parameter.
+        cmdForShards[AggregationRequest::kLetName] =
+            Value(expCtx->variablesParseState.serialize(expCtx->variables));
     }
 
     cmdForShards[AggregationRequest::kFromMongosName] = Value(expCtx->inMongos);
@@ -703,7 +721,6 @@ BSONObj createPassthroughCommandForShard(
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
     Document serializedCommand,
     boost::optional<ExplainOptions::Verbosity> explainVerbosity,
-    const boost::optional<LegacyRuntimeConstants>& constants,
     Pipeline* pipeline,
     BSONObj collationObj) {
     // Create the command for the shards.
@@ -713,7 +730,7 @@ BSONObj createPassthroughCommandForShard(
     }
 
     return genericTransformForShards(
-        std::move(targetedCmd), expCtx, explainVerbosity, constants, collationObj);
+        std::move(targetedCmd), expCtx, explainVerbosity, collationObj);
 }
 
 BSONObj createCommandForTargetedShards(const boost::intrusive_ptr<ExpressionContext>& expCtx,
@@ -751,11 +768,8 @@ BSONObj createCommandForTargetedShards(const boost::intrusive_ptr<ExpressionCont
     targetedCmd[AggregationRequest::kExchangeName] =
         exchangeSpec ? Value(exchangeSpec->exchangeSpec.toBSON()) : Value();
 
-    return genericTransformForShards(std::move(targetedCmd),
-                                     expCtx,
-                                     expCtx->explain,
-                                     expCtx->getLegacyRuntimeConstants(),
-                                     expCtx->getCollatorBSON());
+    return genericTransformForShards(
+        std::move(targetedCmd), expCtx, expCtx->explain, expCtx->getCollatorBSON());
 }
 
 /**
@@ -833,14 +847,11 @@ DispatchShardPipelineResults dispatchShardPipeline(
         opCtx,
         true,             /* appendRC */
         !expCtx->explain, /* appendWC */
-        splitPipelines ? createCommandForTargetedShards(
-                             expCtx, serializedCommand, *splitPipelines, exchangeSpec, true)
-                       : createPassthroughCommandForShard(expCtx,
-                                                          serializedCommand,
-                                                          expCtx->explain,
-                                                          expCtx->getLegacyRuntimeConstants(),
-                                                          pipeline.get(),
-                                                          collationObj));
+        splitPipelines
+            ? createCommandForTargetedShards(
+                  expCtx, serializedCommand, *splitPipelines, exchangeSpec, true)
+            : createPassthroughCommandForShard(
+                  expCtx, serializedCommand, expCtx->explain, pipeline.get(), collationObj));
 
     // A $changeStream pipeline must run on all shards, and will also open an extra cursor on the
     // config server in order to monitor for new shards. To guarantee that we do not miss any
diff --git a/src/mongo/db/pipeline/sharded_agg_helpers.h b/src/mongo/db/pipeline/sharded_agg_helpers.h
index db689349a5f..4cd8d62a6f2 100644
--- a/src/mongo/db/pipeline/sharded_agg_helpers.h
+++ b/src/mongo/db/pipeline/sharded_agg_helpers.h
@@ -130,7 +130,6 @@ BSONObj createPassthroughCommandForShard(
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
     Document serializedCommand,
     boost::optional<ExplainOptions::Verbosity> explainVerbosity,
-    const boost::optional<LegacyRuntimeConstants>& constants,
     Pipeline* pipeline,
     BSONObj collationObj);
 
diff --git a/src/mongo/db/pipeline/variable_validation.cpp b/src/mongo/db/pipeline/variable_validation.cpp
index a0b29b2f59e..db42567723a 100644
--- a/src/mongo/db/pipeline/variable_validation.cpp
+++ b/src/mongo/db/pipeline/variable_validation.cpp
@@ -28,42 +28,63 @@
  */
 
 #include "mongo/base/error_codes.h"
+#include "mongo/util/stacktrace.h"
 #include "mongo/util/str.h"
 
 namespace mongo::variableValidation {
+
+namespace {
+Status isValidName(StringData varName,
+                   std::function<bool(char)> prefixPred,
+                   std::function<bool(char)> suffixPred,
+                   int prefixLen) {
+    if (varName.empty()) {
+        return Status{ErrorCodes::FailedToParse, "empty variable names are not allowed"};
+    }
+    for (int i = 0; i < prefixLen; ++i)
+        if (!prefixPred(varName[i])) {
+            return Status{ErrorCodes::FailedToParse,
+                          str::stream()
+                              << "'" << varName
+                              << "' starts with an invalid character for a user variable name"};
+        }
+
+    for (size_t i = prefixLen; i < varName.size(); i++)
+        if (!suffixPred(varName[i])) {
+            return Status{ErrorCodes::FailedToParse,
+                          str::stream() << "'" << varName << "' contains an invalid character "
+                                        << "for a variable name: '" << varName[i] << "'"};
+        }
+    return Status::OK();
+}
+}  // namespace
+
 void validateName(StringData varName,
                   std::function<bool(char)> prefixPred,
                   std::function<bool(char)> suffixPred,
                   int prefixLen) {
-    uassert(16866, "empty variable names are not allowed", !varName.empty());
-    for (int i = 0; i < prefixLen; ++i)
-        if (!prefixPred(varName[i]))
-            uasserted(16867,
-                      str::stream()
-                          << "'" << varName
-                          << "' starts with an invalid character for a user variable name");
-
-    for (size_t i = prefixLen; i < varName.size(); i++)
-        if (!suffixPred(varName[i]))
-            uasserted(16868,
-                      str::stream() << "'" << varName << "' contains an invalid character "
-                                    << "for a variable name: '" << varName[i] << "'");
+    uassertStatusOK(isValidName(varName, prefixPred, suffixPred, prefixLen));
 }
 
-void validateNameForUserWrite(StringData varName) {
+Status isValidNameForUserWrite(StringData varName) {
     // System variables users allowed to write to (currently just one)
     if (varName == "CURRENT") {
-        return;
+        return Status::OK();
     }
-    validateName(varName,
-                 [](char ch) -> bool {
-                     return (ch >= 'a' && ch <= 'z') || (ch & '\x80');  // non-ascii
-                 },
-                 [](char ch) -> bool {
-                     return (ch >= 'a' && ch <= 'z') || (ch >= 'A' && ch <= 'Z') ||
-                         (ch >= '0' && ch <= '9') || (ch == '_') || (ch & '\x80');  // non-ascii
-                 },
-                 1);
+    return isValidName(varName,
+                       [](char ch) -> bool {
+                           return (ch >= 'a' && ch <= 'z') || (ch & '\x80');  // non-ascii
+                       },
+                       [](char ch) -> bool {
+                           return (ch >= 'a' && ch <= 'z') || (ch >= 'A' && ch <= 'Z') ||
+                               (ch >= '0' && ch <= '9') || (ch == '_') ||
+                               (ch & '\x80');  // non-ascii
+                       },
+                       1);
+}
+
+void validateNameForUserWrite(StringData varName) {
+    uassertStatusOK(isValidNameForUserWrite(varName));
 }
 
 void validateNameForUserRead(StringData varName) {
diff --git a/src/mongo/db/pipeline/variable_validation.h b/src/mongo/db/pipeline/variable_validation.h
index a11b1df49cd..baee73180a4 100644
--- a/src/mongo/db/pipeline/variable_validation.h
+++ b/src/mongo/db/pipeline/variable_validation.h
@@ -31,6 +31,7 @@
 
 
 namespace mongo::variableValidation {
+Status isValidNameForUserWrite(StringData varName);
 void validateNameForUserWrite(StringData varName);
 void validateNameForUserRead(StringData varName);
 void validateName(StringData varName,
diff --git a/src/mongo/db/pipeline/variables.cpp b/src/mongo/db/pipeline/variables.cpp
index 0773c321815..8cabf0f3c52 100644
--- a/src/mongo/db/pipeline/variables.cpp
+++ b/src/mongo/db/pipeline/variables.cpp
@@ -45,44 +45,52 @@ using namespace std::string_literals;
 constexpr Variables::Id Variables::kRootId;
 constexpr Variables::Id Variables::kRemoveId;
 
-const StringMap<Variables::Id> Variables::kBuiltinVarNameToId = {{"ROOT", kRootId},
-                                                                 {"REMOVE", kRemoveId},
-                                                                 {"NOW", kNowId},
-                                                                 {"CLUSTER_TIME", kClusterTimeId},
-                                                                 {"JS_SCOPE", kJsScopeId},
-                                                                 {"IS_MR", kIsMapReduceId}};
+constexpr StringData kRootName = "ROOT"_sd;
+constexpr StringData kRemoveName = "REMOVE"_sd;
+constexpr StringData kNowName = "NOW"_sd;
+constexpr StringData kClusterTimeName = "CLUSTER_TIME"_sd;
+constexpr StringData kJsScopeName = "JS_SCOPE"_sd;
+constexpr StringData kIsMapReduceName = "IS_MR"_sd;
+
+const StringMap<Variables::Id> Variables::kBuiltinVarNameToId = {
+    {kRootName.rawData(), kRootId},
+    {kRemoveName.rawData(), kRemoveId},
+    {kNowName.rawData(), kNowId},
+    {kClusterTimeName.rawData(), kClusterTimeId},
+    {kJsScopeName.rawData(), kJsScopeId},
+    {kIsMapReduceName.rawData(), kIsMapReduceId}};
 
 const std::map<Variables::Id, std::string> Variables::kIdToBuiltinVarName = {
-    {kRootId, "ROOT"},
-    {kRemoveId, "REMOVE"},
-    {kNowId, "NOW"},
-    {kClusterTimeId, "CLUSTER_TIME"},
-    {kJsScopeId, "JS_SCOPE"},
-    {kIsMapReduceId, "IS_MR"}};
+    {kRootId, kRootName.rawData()},
+    {kRemoveId, kRemoveName.rawData()},
+    {kNowId, kNowName.rawData()},
+    {kClusterTimeId, kClusterTimeName.rawData()},
+    {kJsScopeId, kJsScopeName.rawData()},
+    {kIsMapReduceId, kIsMapReduceName.rawData()}};
 
 const std::map<StringData, std::function<void(const Value&)>> Variables::kSystemVarValidators = {
-    {"NOW"_sd,
+    {kNowName,
      [](const auto& value) {
          uassert(ErrorCodes::TypeMismatch,
                  str::stream() << "$$NOW must have a date value, found "
                                << typeName(value.getType()),
                  value.getType() == BSONType::Date);
      }},
-    {"CLUSTER_TIME"_sd,
+    {kClusterTimeName,
      [](const auto& value) {
          uassert(ErrorCodes::TypeMismatch,
                  str::stream() << "$$CLUSTER_TIME must have a timestamp value, found "
                                << typeName(value.getType()),
                  value.getType() == BSONType::bsonTimestamp);
      }},
-    {"JS_SCOPE"_sd,
+    {kJsScopeName,
      [](const auto& value) {
          uassert(ErrorCodes::TypeMismatch,
                  str::stream() << "$$JS_SCOPE must have an object value, found "
                                << typeName(value.getType()),
                  value.getType() == BSONType::Object);
      }},
-    {"IS_MR"_sd, [](const auto& value) {
+    {kIsMapReduceName, [](const auto& value) {
          uassert(ErrorCodes::TypeMismatch,
                  str::stream() << "$$IS_MR must have a bool value, found "
                                << typeName(value.getType()),
@@ -95,7 +103,7 @@ void Variables::setValue(Id id, const Value& value, bool isConstant) {
     // If a value has already been set for 'id', and that value was marked as constant, then it
     // is illegal to modify.
     invariant(!hasConstantValue(id));
-    _letParametersMap[id] = {value, isConstant};
+    _definitions[id] = {value, isConstant};
 }
 
 void Variables::setValue(Variables::Id id, const Value& value) {
@@ -111,8 +119,8 @@ void Variables::setConstantValue(Variables::Id id, const Value& value) {
 Value Variables::getUserDefinedValue(Variables::Id id) const {
     invariant(isUserDefinedVariable(id));
 
-    auto it = _letParametersMap.find(id);
-    uassert(40434, str::stream() << "Undefined variable id: " << id, it != _letParametersMap.end());
+    auto it = _definitions.find(id);
+    uassert(40434, str::stream() << "Undefined variable id: " << id, it != _definitions.end());
     return it->second.value;
 }
 
@@ -126,17 +134,14 @@ Value Variables::getValue(Id id, const Document& root) const {
                 return Value();
             case Variables::kNowId:
             case Variables::kClusterTimeId:
-                if (auto it = _runtimeConstantsMap.find(id); it != _runtimeConstantsMap.end()) {
-                    return it->second;
+            case Variables::kJsScopeId:
+            case Variables::kIsMapReduceId:
+                if (auto it = _definitions.find(id); it != _definitions.end()) {
+                    return it->second.value;
                 }
                 uasserted(51144,
                           str::stream() << "Builtin variable '$$" << getBuiltinVariableName(id)
                                         << "' is not available");
-                MONGO_UNREACHABLE;
-            case Variables::kJsScopeId:
-                uasserted(4631100, "Use of undefined variable '$$JS_SCOPE'.");
-            case Variables::kIsMapReduceId:
-                uasserted(4631101, "Use of undefined variable '$$IS_MR'.");
             default:
                 MONGO_UNREACHABLE;
         }
@@ -158,38 +163,72 @@ Document Variables::getDocument(Id id, const Document& root) const {
     return Document();
 }
 
-const LegacyRuntimeConstants& Variables::getLegacyRuntimeConstants() const {
-    invariant(_legacyRuntimeConstants);
-    return *_legacyRuntimeConstants;
-}
-
 void Variables::setLegacyRuntimeConstants(const LegacyRuntimeConstants& constants) {
-    invariant(!_legacyRuntimeConstants);
-    _runtimeConstantsMap[kNowId] = Value(constants.getLocalNow());
+    const bool constant = true;
+    _definitions[kNowId] = {Value(constants.getLocalNow()), constant};
     // We use a null Timestamp to indicate that the clusterTime is not available; this can happen if
     // the logical clock is not running. We do not use boost::optional because this would allow the
     // IDL to serialize a RuntimConstants without clusterTime, which should always be an error.
     if (!constants.getClusterTime().isNull()) {
-        _runtimeConstantsMap[kClusterTimeId] = Value(constants.getClusterTime());
+        _definitions[kClusterTimeId] = {Value(constants.getClusterTime()), constant};
     }
 
     if (constants.getJsScope()) {
-        _runtimeConstantsMap[kJsScopeId] = Value(constants.getJsScope().get());
+        _definitions[kJsScopeId] = {Value(*constants.getJsScope()), constant};
     }
     if (constants.getIsMapReduce()) {
-        _runtimeConstantsMap[kIsMapReduceId] = Value(constants.getIsMapReduce().get());
+        _definitions[kIsMapReduceId] = {Value(*constants.getIsMapReduce()), constant};
     }
-    _legacyRuntimeConstants = constants;
 }
 
 void Variables::setDefaultRuntimeConstants(OperationContext* opCtx) {
     setLegacyRuntimeConstants(Variables::generateRuntimeConstants(opCtx));
 }
 
+void Variables::appendSystemVariables(BSONObjBuilder& bob) const {
+    for (auto&& [name, id] : kBuiltinVarNameToId) {
+        if (hasValue(id)) {
+            bob << name << getValue(id);
+        }
+    }
+}
+
+namespace {
+
+/**
+ * Returns a callback function which can be used to verify the value conforms to expectations if
+ * 'varName' is a reserved system variable. Throws an exception if 'varName' is a reserved name
+ * (e.g. capital letter) but not one of the known variables. Returns boost::none for normal
+ * variables.
+ */
+boost::optional<std::function<void(const Value&)>> validateVariable(OperationContext* opCtx,
+                                                                    StringData varName) {
+    auto validateStatus = variableValidation::isValidNameForUserWrite(varName);
+    if (validateStatus.isOK()) {
+        return boost::none;
+    }
+    // Reserved field name. It may be an internal propogation of a constant. Otherwise we need to
+    // reject it.
+    const auto& knownConstantIt = Variables::kSystemVarValidators.find(varName);
+    if (knownConstantIt == Variables::kSystemVarValidators.end()) {
+        uassertStatusOKWithContext(validateStatus, "Invalid 'let' parameter");
+    }
+
+    uassert(4738901,
+            str::stream() << "Attempt to set internal constant: " << varName,
+            opCtx->getClient()->session() &&
+                (opCtx->getClient()->session()->getTags() & transport::Session::kInternalClient));
+
+    return knownConstantIt->second;
+}
+
+}  // namespace
+
 void Variables::seedVariablesWithLetParameters(ExpressionContext* const expCtx,
                                                const BSONObj letParams) {
     for (auto&& elem : letParams) {
-        variableValidation::validateNameForUserWrite(elem.fieldName());
+        const auto fieldName = elem.fieldNameStringData();
+        auto maybeSystemVarValidator = validateVariable(expCtx->opCtx, fieldName);
         auto expr = Expression::parseOperand(expCtx, elem, expCtx->variablesParseState);
 
         uassert(4890500,
@@ -198,23 +237,14 @@ void Variables::seedVariablesWithLetParameters(ExpressionContext* const expCtx,
                 expr->getDependencies().hasNoRequirements());
         Value value = expr->evaluate(Document{}, &expCtx->variables);
 
-        const auto sysVarName = [&]() -> boost::optional<StringData> {
-            // ROOT and REMOVE are excluded since they're not constants.
-            auto name = elem.fieldNameStringData();
-            if (auto it = kSystemVarValidators.find(name); it != kSystemVarValidators.end()) {
-                auto&& [ignore, validator] = *it;
-                validator(value);
-                return name;
-            }
-            return boost::none;
-        }();
-        if (sysVarName) {
-            if (!(sysVarName == "CLUSTER_TIME"_sd && value.getTimestamp().isNull())) {
+        if (maybeSystemVarValidator) {
+            (*maybeSystemVarValidator)(value);
+            if (!(fieldName == kClusterTimeName && value.getTimestamp().isNull())) {
                 // Avoid populating a value for CLUSTER_TIME if the value is null.
-                _runtimeConstantsMap[kBuiltinVarNameToId.at(*sysVarName)] = value;
+                _definitions[kBuiltinVarNameToId.at(fieldName)] = {value, true};
             }
         } else {
-            setConstantValue(expCtx->variablesParseState.defineVariable(elem.fieldName()), value);
+            setConstantValue(expCtx->variablesParseState.defineVariable(fieldName), value);
         }
     }
 }
@@ -239,6 +269,41 @@ void Variables::copyToExpCtx(const VariablesParseState& vps, ExpressionContext*
     expCtx->variablesParseState = vps.copyWith(expCtx->variables.useIdGenerator());
 }
 
+LegacyRuntimeConstants Variables::transitionalExtractRuntimeConstants() const {
+    LegacyRuntimeConstants extracted;
+    for (auto&& [builtinName, ignoredValidator] : kSystemVarValidators) {
+        const auto builtinId = kBuiltinVarNameToId.at(builtinName);
+        if (auto it = _definitions.find(builtinId); it != _definitions.end()) {
+            const auto& [value, unusedIsConstant] = it->second;
+            switch (builtinId) {
+                case kNowId: {
+                    invariant(value.getType() == BSONType::Date);
+                    extracted.setLocalNow(value.getDate());
+                    break;
+                }
+                case kClusterTimeId: {
+                    invariant(value.getType() == BSONType::bsonTimestamp);
+                    extracted.setClusterTime(value.getTimestamp());
+                    break;
+                }
+                case kJsScopeId: {
+                    invariant(value.getType() == BSONType::Object);
+                    extracted.setJsScope(value.getDocument().toBson());
+                    break;
+                }
+                case kIsMapReduceId: {
+                    invariant(value.getType() == BSONType::Bool);
+                    extracted.setIsMapReduce(value.getBool());
+                    break;
+                }
+                default:
+                    MONGO_UNREACHABLE;
+            }
+        }
+    }
+    return extracted;
+}
+
 Variables::Id VariablesParseState::defineVariable(StringData name) {
     // Caller should have validated before hand by using variableValidationvalidateNameForUserWrite.
     massert(17275,
@@ -281,11 +346,25 @@ std::set<Variables::Id> VariablesParseState::getDefinedVariableIDs() const {
     return ids;
 }
 
-BSONObj VariablesParseState::serializeUserVariables(const Variables& vars) const {
+BSONObj VariablesParseState::serialize(const Variables& vars) const {
     auto bob = BSONObjBuilder{};
     for (auto&& [var_name, id] : _variables)
         if (vars.hasValue(id))
             bob << var_name << vars.getValue(id);
+
+    // System variables have to be added separately since the variable IDs are reserved and not
+    // allocated like normal variables, and so not present in '_variables'.
+    vars.appendSystemVariables(bob);
     return bob.obj();
 }
+
+std::pair<LegacyRuntimeConstants, BSONObj> VariablesParseState::transitionalCompatibilitySerialize(
+    const Variables& vars) const {
+    auto bob = BSONObjBuilder{};
+    for (auto&& [var_name, id] : _variables)
+        if (vars.hasValue(id))
+            bob << var_name << vars.getValue(id);
+
+    return {vars.transitionalExtractRuntimeConstants(), bob.obj()};
+}
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/variables.h b/src/mongo/db/pipeline/variables.h
index 704c22bf6f3..d1fcadea9f7 100644
--- a/src/mongo/db/pipeline/variables.h
+++ b/src/mongo/db/pipeline/variables.h
@@ -126,11 +126,8 @@ public:
      * Returns whether a constant value for 'id' has been defined using setConstantValue().
      */
     bool hasConstantValue(Variables::Id id) const {
-        if (auto it = _letParametersMap.find(id);
-            it != _letParametersMap.end() && it->second.isConstant) {
-            return true;
-        }
-        return false;
+        auto it = _definitions.find(id);
+        return (it != _definitions.end() && it->second.isConstant);
     }
 
     /**
@@ -144,13 +141,6 @@ public:
     }
 
     /**
-     * Return a reference to an object which represents the variables which are considered "runtime
-     * constants." It is a programming error to call this function without having called
-     * setLegacyRuntimeConstants().
-     */
-    const LegacyRuntimeConstants& getLegacyRuntimeConstants() const;
-
-    /**
      * Set the runtime constants. It is a programming error to call this more than once.
      */
     void setLegacyRuntimeConstants(const LegacyRuntimeConstants& constants);
@@ -167,13 +157,11 @@ public:
                                         const BSONObj letParameters);
 
     bool hasValue(Variables::Id id) const {
-        if (id < 0)  // system variables.
-            return true;
-        if (auto it = _letParametersMap.find(id); it != _letParametersMap.end())
-            return true;
-        return false;
+        return _definitions.find(id) != _definitions.end();
     };
 
+    void appendSystemVariables(BSONObjBuilder& bob) const;
+
     /**
      * Copies this Variables and 'vps' to the Variables and VariablesParseState objects in 'expCtx'.
      * The VariablesParseState's 'idGenerator' in 'expCtx' is replaced with the pointer to the
@@ -185,6 +173,8 @@ public:
      */
     void copyToExpCtx(const VariablesParseState& vps, ExpressionContext* expCtx) const;
 
+    LegacyRuntimeConstants transitionalExtractRuntimeConstants() const;
+
 private:
     struct ValueAndState {
         ValueAndState() = default;
@@ -207,11 +197,7 @@ private:
     }
 
     IdGenerator _idGenerator;
-    stdx::unordered_map<Id, ValueAndState> _letParametersMap;
-    stdx::unordered_map<Id, Value> _runtimeConstantsMap;
-
-    // Populated after construction. Should not be set more than once.
-    boost::optional<LegacyRuntimeConstants> _legacyRuntimeConstants;
+    stdx::unordered_map<Id, ValueAndState> _definitions;
 };
 
 /**
@@ -256,10 +242,16 @@ public:
     std::set<Variables::Id> getDefinedVariableIDs() const;
 
     /**
-     * Serializes the IDs and associated values of user-defined variables that are currently in
-     * scope.
+     * Serializes a map from name to values of defined variables.
+     */
+    BSONObj serialize(const Variables& vars) const;
+
+    /**
+     * Splits defined variables into runtime constants and "the rest" as a transitional tool while
+     * we phase out using LegacyRuntimeConstants.
      */
-    BSONObj serializeUserVariables(const Variables& vars) const;
+    std::pair<LegacyRuntimeConstants, BSONObj> transitionalCompatibilitySerialize(
+        const Variables& vars) const;
 
     /**
      * Return a copy of this VariablesParseState. Will replace the copy's '_idGenerator' pointer
diff --git a/src/mongo/db/update/pipeline_executor_test.cpp b/src/mongo/db/update/pipeline_executor_test.cpp
index 2fbcb6e2d7c..c650e06749d 100644
--- a/src/mongo/db/update/pipeline_executor_test.cpp
+++ b/src/mongo/db/update/pipeline_executor_test.cpp
@@ -351,15 +351,21 @@ TEST_F(PipelineExecutorTest, RejectsInvalidConstantNames) {
 
     // Empty name.
     auto constants = BSON("" << 10);
-    ASSERT_THROWS_CODE(PipelineExecutor(expCtx, pipeline, constants), AssertionException, 16866);
+    ASSERT_THROWS_CODE(PipelineExecutor(expCtx, pipeline, constants),
+                       AssertionException,
+                       ErrorCodes::FailedToParse);
 
     // Invalid first character.
     constants = BSON("^invalidFirstChar" << 10);
-    ASSERT_THROWS_CODE(PipelineExecutor(expCtx, pipeline, constants), AssertionException, 16867);
+    ASSERT_THROWS_CODE(PipelineExecutor(expCtx, pipeline, constants),
+                       AssertionException,
+                       ErrorCodes::FailedToParse);
 
     // Contains invalid character.
     constants = BSON("contains*InvalidChar" << 10);
-    ASSERT_THROWS_CODE(PipelineExecutor(expCtx, pipeline, constants), AssertionException, 16868);
+    ASSERT_THROWS_CODE(PipelineExecutor(expCtx, pipeline, constants),
+                       AssertionException,
+                       ErrorCodes::FailedToParse);
 }
 
 TEST_F(PipelineExecutorTest, CanUseConstants) {
diff --git a/src/mongo/s/commands/cluster_map_reduce_agg.cpp b/src/mongo/s/commands/cluster_map_reduce_agg.cpp
index 2c483d02d97..63cd6dfbd39 100644
--- a/src/mongo/s/commands/cluster_map_reduce_agg.cpp
+++ b/src/mongo/s/commands/cluster_map_reduce_agg.cpp
@@ -118,8 +118,8 @@ Document serializeToCommand(BSONObj originalCmd, const MapReduce& parsedMr, Pipe
         Value(Document{{"batchSize", std::numeric_limits<long long>::max()}});
     translatedCmd[AggregationRequest::kAllowDiskUseName] = Value(true);
     translatedCmd[AggregationRequest::kFromMongosName] = Value(true);
-    translatedCmd[AggregationRequest::kLegacyRuntimeConstantsName] =
-        Value(pipeline->getContext()->getLegacyRuntimeConstants().toBSON());
+    translatedCmd[AggregationRequest::kLetName] = Value(
+        pipeline->getContext()->variablesParseState.serialize(pipeline->getContext()->variables));
     translatedCmd[AggregationRequest::kIsMapReduceCommandName] = Value(true);
 
     if (shouldBypassDocumentValidationForCommand(originalCmd)) {
diff --git a/src/mongo/s/query/cluster_aggregate.cpp b/src/mongo/s/query/cluster_aggregate.cpp
index d10d4133e2a..44e6fc0317e 100644
--- a/src/mongo/s/query/cluster_aggregate.cpp
+++ b/src/mongo/s/query/cluster_aggregate.cpp
@@ -289,7 +289,8 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx,
         // passthrough, we only need a bare minimum expression context anyway.
         invariant(targeter.policy ==
                   cluster_aggregation_planner::AggregationTargeter::kPassthrough);
-        expCtx = make_intrusive<ExpressionContext>(opCtx, nullptr, namespaces.executionNss);
+        expCtx = make_intrusive<ExpressionContext>(
+            opCtx, nullptr, namespaces.executionNss, boost::none, request.getLetParameters());
     }
 
     if (request.getExplain()) {
diff --git a/src/mongo/s/query/cluster_aggregation_planner.cpp b/src/mongo/s/query/cluster_aggregation_planner.cpp
index 3e9a48552b1..120767dd590 100644
--- a/src/mongo/s/query/cluster_aggregation_planner.cpp
+++ b/src/mongo/s/query/cluster_aggregation_planner.cpp
@@ -126,8 +126,8 @@ BSONObj createCommandForMergingShard(Document serializedCommand,
     mergeCmd["pipeline"] = Value(pipelineForMerging->serialize());
     mergeCmd[AggregationRequest::kFromMongosName] = Value(true);
 
-    mergeCmd[AggregationRequest::kLegacyRuntimeConstantsName] =
-        Value(mergeCtx->getLegacyRuntimeConstants().toBSON());
+    mergeCmd[AggregationRequest::kLetName] =
+        Value(mergeCtx->variablesParseState.serialize(mergeCtx->variables));
 
     // If the user didn't specify a collation already, make sure there's a collation attached to
     // the merge command, since the merging shard may not have the collection metadata.
@@ -607,13 +607,13 @@ Status runPipelineOnPrimaryShard(const boost::intrusive_ptr<ExpressionContext>&
 
     // Format the command for the shard. This adds the 'fromMongos' field, wraps the command as an
     // explain if necessary, and rewrites the result into a format safe to forward to shards.
-    BSONObj cmdObj = applyReadWriteConcern(
-        opCtx,
-        true,     /* appendRC */
-        !explain, /* appendWC */
-        CommandHelpers::filterCommandRequestForPassthrough(
-            sharded_agg_helpers::createPassthroughCommandForShard(
-                expCtx, serializedCommand, explain, boost::none, nullptr, BSONObj())));
+    BSONObj cmdObj =
+        applyReadWriteConcern(opCtx,
+                              true,     /* appendRC */
+                              !explain, /* appendWC */
+                              CommandHelpers::filterCommandRequestForPassthrough(
+                                  sharded_agg_helpers::createPassthroughCommandForShard(
+                                      expCtx, serializedCommand, explain, nullptr, BSONObj())));
 
     const auto shardId = cm.dbPrimary();
     const auto cmdObjWithShardVersion = (shardId != ShardRegistry::kConfigServerShardId)