SERVER-29073 Allow variable definition within $lookup

6 files changed, 274 insertions, 61 deletions

diff --git a/src/mongo/db/pipeline/document_source_lookup.cpp b/src/mongo/db/pipeline/document_source_lookup.cpp
index 57f0d6ecb66..0f09f98a7ea 100644
--- a/src/mongo/db/pipeline/document_source_lookup.cpp
+++ b/src/mongo/db/pipeline/document_source_lookup.cpp
@@ -67,11 +67,14 @@ std::string pipelineToString(const vector<BSONObj>& pipeline) {
 DocumentSourceLookUp::DocumentSourceLookUp(NamespaceString fromNs,
                                            std::string as,
                                            const boost::intrusive_ptr<ExpressionContext>& pExpCtx)
-    : DocumentSourceNeedsMongod(pExpCtx), _fromNs(std::move(fromNs)), _as(std::move(as)) {
+    : DocumentSourceNeedsMongod(pExpCtx),
+      _fromNs(std::move(fromNs)),
+      _as(std::move(as)),
+      _variablesParseState(_variables.useIdGenerator()) {
     const auto& resolvedNamespace = pExpCtx->getResolvedNamespace(_fromNs);
-    _fromExpCtx = pExpCtx->copyWith(resolvedNamespace.ns);
-    _fromPipeline = resolvedNamespace.pipeline;
-    _resolvedNs = std::move(resolvedNamespace.ns);
+    _resolvedNs = resolvedNamespace.ns;
+    _resolvedPipeline = resolvedNamespace.pipeline;
+    _fromExpCtx = pExpCtx->copyWith(_resolvedNs);
 }
 
 DocumentSourceLookUp::DocumentSourceLookUp(NamespaceString fromNs,
@@ -82,45 +85,58 @@ DocumentSourceLookUp::DocumentSourceLookUp(NamespaceString fromNs,
     : DocumentSourceLookUp(fromNs, as, pExpCtx) {
     _localField = std::move(localField);
     _foreignField = std::move(foreignField);
-    // We append an additional BSONObj to '_fromPipeline' as a placeholder for the $match stage
+    // We append an additional BSONObj to '_resolvedPipeline' as a placeholder for the $match stage
     // we'll eventually construct from the input document.
-    _fromPipeline.reserve(_fromPipeline.size() + 1);
-    _fromPipeline.push_back(BSONObj());
+    _resolvedPipeline.reserve(_resolvedPipeline.size() + 1);
+    _resolvedPipeline.push_back(BSONObj());
 }
 
 DocumentSourceLookUp::DocumentSourceLookUp(NamespaceString fromNs,
                                            std::string as,
                                            std::vector<BSONObj> pipeline,
+                                           BSONObj letVariables,
                                            const boost::intrusive_ptr<ExpressionContext>& pExpCtx)
     : DocumentSourceLookUp(fromNs, as, pExpCtx) {
-    // '_fromPipeline' will first be initialized by the constructor delegated to within this
+    // '_resolvedPipeline' will first be initialized by the constructor delegated to within this
     // constructor's initializer list. It will be populated with view pipeline prefix if 'fromNs'
-    // represents a view. We append the user 'pipeline' to the end of '_fromPipeline' to ensure any
-    // view prefix is not overwritten.
-    _fromPipeline.insert(_fromPipeline.end(), pipeline.begin(), pipeline.end());
+    // represents a view. We append the user 'pipeline' to the end of '_resolvedPipeline' to ensure
+    // any view prefix is not overwritten.
+    _resolvedPipeline.insert(_resolvedPipeline.end(), pipeline.begin(), pipeline.end());
 
     _userPipeline = std::move(pipeline);
+
+    for (auto&& varElem : letVariables) {
+        const auto varName = varElem.fieldNameStringData();
+        Variables::uassertValidNameForUserWrite(varName);
+
+        _letVariables.emplace_back(
+            varName.toString(),
+            Expression::parseOperand(pExpCtx, varElem, pExpCtx->variablesParseState),
+            _variablesParseState.defineVariable(varName));
+    }
 }
 
 std::unique_ptr<LiteParsedDocumentSourceOneForeignCollection> DocumentSourceLookUp::liteParse(
     const AggregationRequest& request, const BSONElement& spec) {
-    uassert(40319,
+    uassert(ErrorCodes::FailedToParse,
             str::stream() << "the $lookup stage specification must be an object, but found "
                           << typeName(spec.type()),
             spec.type() == BSONType::Object);
 
     auto specObj = spec.Obj();
     auto fromElement = specObj["from"];
-    uassert(40320,
+    uassert(ErrorCodes::FailedToParse,
             str::stream() << "missing 'from' option to $lookup stage specification: " << specObj,
             fromElement);
-    uassert(40321,
+    uassert(ErrorCodes::FailedToParse,
             str::stream() << "'from' option to $lookup must be a string, but was type "
                           << typeName(specObj["from"].type()),
             fromElement.type() == BSONType::String);
 
     NamespaceString nss(request.getNamespaceString().db(), fromElement.valueStringData());
-    uassert(40322, str::stream() << "invalid $lookup namespace: " << nss.ns(), nss.isValid());
+    uassert(ErrorCodes::InvalidNamespace,
+            str::stream() << "invalid $lookup namespace: " << nss.ns(),
+            nss.isValid());
     return stdx::make_unique<LiteParsedDocumentSourceOneForeignCollection>(std::move(nss));
 }
 
@@ -166,7 +182,10 @@ DocumentSource::GetNextResult DocumentSourceLookUp::getNext() {
     if (!nextInput.isAdvanced()) {
         return nextInput;
     }
+
     auto inputDoc = nextInput.releaseDocument();
+    copyVariablesToExpCtx(_variables, _variablesParseState, _fromExpCtx.get());
+    resolveLetVariables(inputDoc, &_fromExpCtx->variables);
 
     // If we have not absorbed a $unwind, we cannot absorb a $match. If we have absorbed a $unwind,
     // '_unwindSrc' would be non-null, and we would not have made it here.
@@ -175,11 +194,11 @@ DocumentSource::GetNextResult DocumentSourceLookUp::getNext() {
     if (!wasConstructedWithPipelineSyntax()) {
         auto matchStage =
             makeMatchStageFromInput(inputDoc, *_localField, _foreignField->fullPath(), BSONObj());
-        // We've already allocated space for the trailing $match stage in '_fromPipeline'.
-        _fromPipeline.back() = matchStage;
+        // We've already allocated space for the trailing $match stage in '_resolvedPipeline'.
+        _resolvedPipeline.back() = matchStage;
     }
 
-    auto pipeline = uassertStatusOK(_mongod->makePipeline(_fromPipeline, _fromExpCtx));
+    auto pipeline = uassertStatusOK(_mongod->makePipeline(_resolvedPipeline, _fromExpCtx));
 
     std::vector<Value> results;
     int objsize = 0;
@@ -319,7 +338,7 @@ Pipeline::SourceContainer::iterator DocumentSourceLookUp::doOptimizeAt(
 
     if (wasConstructedWithPipelineSyntax()) {
         auto matchObj = BSON("$match" << *_additionalFilter);
-        _fromPipeline.push_back(matchObj);
+        _resolvedPipeline.push_back(matchObj);
     }
 
     return itr;
@@ -330,7 +349,7 @@ std::string DocumentSourceLookUp::getUserPipelineDefinition() {
         return pipelineToString(_userPipeline);
     }
 
-    return _fromPipeline.back().toString();
+    return _resolvedPipeline.back().toString();
 }
 
 void DocumentSourceLookUp::doDispose() {
@@ -351,7 +370,7 @@ BSONObj DocumentSourceLookUp::makeMatchStageFromInput(const Document& input,
     bool containsRegex = false;
     document_path_support::visitAllValuesAtPath(input, localFieldPath, [&](const Value& nextValue) {
         arrBuilder << nextValue;
-        if (!containsRegex && nextValue.getType() == RegEx) {
+        if (!containsRegex && nextValue.getType() == BSONType::RegEx) {
             containsRegex = true;
         }
     });
@@ -442,14 +461,17 @@ DocumentSource::GetNextResult DocumentSourceLookUp::unwindResult() {
             BSONObj filter = _additionalFilter.value_or(BSONObj());
             auto matchStage =
                 makeMatchStageFromInput(*_input, *_localField, _foreignField->fullPath(), filter);
-            // We've already allocated space for the trailing $match stage in '_fromPipeline'.
-            _fromPipeline.back() = matchStage;
+            // We've already allocated space for the trailing $match stage in '_resolvedPipeline'.
+            _resolvedPipeline.back() = matchStage;
         }
 
         if (_pipeline) {
             _pipeline->dispose(pExpCtx->opCtx);
         }
-        _pipeline = uassertStatusOK(_mongod->makePipeline(_fromPipeline, _fromExpCtx));
+
+        copyVariablesToExpCtx(_variables, _variablesParseState, _fromExpCtx.get());
+        resolveLetVariables(*_input, &_fromExpCtx->variables);
+        _pipeline = uassertStatusOK(_mongod->makePipeline(_resolvedPipeline, _fromExpCtx));
 
         // The $lookup stage takes responsibility for disposing of its Pipeline, since it will
         // potentially be used by multiple OperationContexts, and the $lookup stage is part of an
@@ -489,14 +511,37 @@ DocumentSource::GetNextResult DocumentSourceLookUp::unwindResult() {
     return output.freeze();
 }
 
+void DocumentSourceLookUp::copyVariablesToExpCtx(const Variables& vars,
+                                                 const VariablesParseState& vps,
+                                                 ExpressionContext* expCtx) {
+    expCtx->variables = vars;
+    expCtx->variablesParseState = vps.copyWith(expCtx->variables.useIdGenerator());
+}
+
+void DocumentSourceLookUp::resolveLetVariables(const Document& localDoc, Variables* variables) {
+    invariant(variables);
+
+    for (auto& letVar : _letVariables) {
+        auto value = letVar.expression->evaluate(localDoc);
+        variables->setValue(letVar.id, value);
+    }
+}
+
 void DocumentSourceLookUp::serializeToArray(
     std::vector<Value>& array, boost::optional<ExplainOptions::Verbosity> explain) const {
     Document doc;
     if (wasConstructedWithPipelineSyntax()) {
+        MutableDocument exprList;
+        for (auto letVar : _letVariables) {
+            exprList.addField(letVar.name,
+                              letVar.expression->serialize(static_cast<bool>(explain)));
+        }
+
         doc = Document{{getSourceName(),
                         Document{{"from", _resolvedNs.coll()},
                                  {"as", _as.fullPath()},
-                                 {"pipeline", _fromPipeline}}}};
+                                 {"let", exprList.freeze()},
+                                 {"pipeline", _resolvedPipeline}}}};
     } else {
         doc = Document{{getSourceName(),
                         {Document{{"from", _fromNs.coll()},
@@ -543,8 +588,11 @@ void DocumentSourceLookUp::serializeToArray(
 }
 
 DocumentSource::GetDepsReturn DocumentSourceLookUp::getDependencies(DepsTracker* deps) const {
-    // As current pipeline syntax only supports non-correlated join, it precludes dependencies.
-    if (!wasConstructedWithPipelineSyntax()) {
+    if (wasConstructedWithPipelineSyntax()) {
+        for (auto&& letVar : _letVariables) {
+            letVar.expression->addDependencies(deps);
+        }
+    } else {
         deps->fields.insert(_localField->fullPath());
     }
     return SEE_NEXT;
@@ -556,7 +604,7 @@ void DocumentSourceLookUp::doDetachFromOperationContext() {
         // use Pipeline::detachFromOperationContext() to take care of updating '_fromExpCtx->opCtx'.
         _pipeline->detachFromOperationContext();
         invariant(_fromExpCtx->opCtx == nullptr);
-    } else {
+    } else if (_fromExpCtx) {
         _fromExpCtx->opCtx = nullptr;
     }
 }
@@ -567,21 +615,27 @@ void DocumentSourceLookUp::doReattachToOperationContext(OperationContext* opCtx)
         // use Pipeline::reattachToOperationContext() to take care of updating '_fromExpCtx->opCtx'.
         _pipeline->reattachToOperationContext(opCtx);
         invariant(_fromExpCtx->opCtx == opCtx);
-    } else {
+    } else if (_fromExpCtx) {
         _fromExpCtx->opCtx = opCtx;
     }
 }
 
 intrusive_ptr<DocumentSource> DocumentSourceLookUp::createFromBson(
     BSONElement elem, const boost::intrusive_ptr<ExpressionContext>& pExpCtx) {
-    uassert(4569, "the $lookup specification must be an Object", elem.type() == Object);
+    uassert(ErrorCodes::FailedToParse,
+            "the $lookup specification must be an Object",
+            elem.type() == BSONType::Object);
 
     NamespaceString fromNs;
     std::string as;
+
     std::string localField;
     std::string foreignField;
+
+    BSONObj letVariables;
     std::vector<BSONObj> pipeline;
     bool hasPipeline = false;
+    bool hasLet = false;
 
     for (auto&& argument : elem.Obj()) {
         const auto argName = argument.fieldNameStringData();
@@ -589,7 +643,7 @@ intrusive_ptr<DocumentSource> DocumentSourceLookUp::createFromBson(
         if (argName == "pipeline") {
             auto result = AggregationRequest::parsePipelineFromBSON(argument);
             if (!result.isOK()) {
-                uasserted(40447,
+                uasserted(ErrorCodes::FailedToParse,
                           str::stream() << "invalid $lookup pipeline definition: "
                                         << result.getStatus().toString());
             }
@@ -598,10 +652,21 @@ intrusive_ptr<DocumentSource> DocumentSourceLookUp::createFromBson(
             continue;
         }
 
-        uassert(4570,
-                str::stream() << "$lookup '" << argument << "' must be a string, is type "
+        if (argName == "let") {
+            uassert(ErrorCodes::FailedToParse,
+                    str::stream() << "$lookup argument '" << argument
+                                  << "' must be an object, is type "
+                                  << argument.type(),
+                    argument.type() == BSONType::Object);
+            letVariables = argument.Obj();
+            hasLet = true;
+            continue;
+        }
+
+        uassert(ErrorCodes::FailedToParse,
+                str::stream() << "$lookup argument '" << argument << "' must be a string, is type "
                               << argument.type(),
-                argument.type() == String);
+                argument.type() == BSONType::String);
 
         if (argName == "from") {
             fromNs = NamespaceString(pExpCtx->ns.db().toString() + '.' + argument.String());
@@ -612,26 +677,33 @@ intrusive_ptr<DocumentSource> DocumentSourceLookUp::createFromBson(
         } else if (argName == "foreignField") {
             foreignField = argument.String();
         } else {
-            uasserted(4571,
+            uasserted(ErrorCodes::FailedToParse,
                       str::stream() << "unknown argument to $lookup: " << argument.fieldName());
         }
     }
 
-    uassert(40451, "must specify 'from' field for a $lookup", !fromNs.ns().empty());
-    uassert(40449, "must specify 'as' field for a $lookup", !as.empty());
+    uassert(
+        ErrorCodes::FailedToParse, "must specify 'from' field for a $lookup", !fromNs.ns().empty());
+    uassert(ErrorCodes::FailedToParse, "must specify 'as' field for a $lookup", !as.empty());
 
     if (hasPipeline) {
-        uassert(40450,
+        uassert(ErrorCodes::FailedToParse,
                 "$lookup with 'pipeline' may not specify 'localField' or 'foreignField'",
-                localField.empty() || foreignField.empty());
+                localField.empty() && foreignField.empty());
 
-        return new DocumentSourceLookUp(
-            std::move(fromNs), std::move(as), std::move(pipeline), pExpCtx);
+        return new DocumentSourceLookUp(std::move(fromNs),
+                                        std::move(as),
+                                        std::move(pipeline),
+                                        std::move(letVariables),
+                                        pExpCtx);
     } else {
-        uassert(4572,
+        uassert(ErrorCodes::FailedToParse,
                 "$lookup requires either 'pipeline' or both 'localField' and 'foreignField' to be "
                 "specified",
                 !localField.empty() && !foreignField.empty());
+        uassert(ErrorCodes::FailedToParse,
+                "$lookup with a 'let' argument must also specify 'pipeline'",
+                !hasLet);
 
         return new DocumentSourceLookUp(std::move(fromNs),
                                         std::move(as),
diff --git a/src/mongo/db/pipeline/document_source_lookup.h b/src/mongo/db/pipeline/document_source_lookup.h
index 50b02501189..b0f953738c7 100644
--- a/src/mongo/db/pipeline/document_source_lookup.h
+++ b/src/mongo/db/pipeline/document_source_lookup.h
@@ -126,6 +126,15 @@ protected:
                                                      Pipeline::SourceContainer* container) final;
 
 private:
+    struct LetVariable {
+        LetVariable(std::string name, boost::intrusive_ptr<Expression> expression, Variables::Id id)
+            : name(std::move(name)), expression(std::move(expression)), id(id) {}
+
+        std::string name;
+        boost::intrusive_ptr<Expression> expression;
+        Variables::Id id;
+    };
+
     /**
      * Target constructor. Handles common-field initialization for the syntax-specific delegating
      * constructors.
@@ -151,6 +160,7 @@ private:
     DocumentSourceLookUp(NamespaceString fromNs,
                          std::string as,
                          std::vector<BSONObj> pipeline,
+                         BSONObj letVariables,
                          const boost::intrusive_ptr<ExpressionContext>& pExpCtx);
 
     /**
@@ -163,6 +173,19 @@ private:
     GetNextResult unwindResult();
 
     /**
+     * Copies 'vars' and 'vps' to the Variables and VariablesParseState objects in 'expCtx'. These
+     * copies provide access to 'let' defined variables in sub-pipeline execution.
+     */
+    static void copyVariablesToExpCtx(const Variables& vars,
+                                      const VariablesParseState& vps,
+                                      ExpressionContext* expCtx);
+
+    /**
+     * Resolves let defined variables against 'localDoc' and stores the results in 'variables'.
+     */
+    void resolveLetVariables(const Document& localDoc, Variables* variables);
+
+    /**
      * The pipeline supplied via the $lookup 'pipeline' argument. This may differ from pipeline that
      * is executed in that it will not include optimizations or resolved views.
      */
@@ -177,16 +200,24 @@ private:
     boost::optional<FieldPath> _localField;
     boost::optional<FieldPath> _foreignField;
 
-    // The ExpressionContext used when performing aggregation pipelines against the '_fromNs'
+    // Holds 'let' defined variables defined both in this stage and in parent pipelines. These are
+    // copied to the '_fromExpCtx' ExpressionContext's 'variables' and 'variablesParseState' for use
+    // in foreign pipeline execution.
+    Variables _variables;
+    VariablesParseState _variablesParseState;
+
+    // The ExpressionContext used when performing aggregation pipelines against the '_resolvedNs'
     // namespace.
     boost::intrusive_ptr<ExpressionContext> _fromExpCtx;
 
-    // The aggregation pipeline to perform against the '_fromNs' namespace.
-    std::vector<BSONObj> _fromPipeline;
+    // The aggregation pipeline to perform against the '_resolvedNs' namespace.
+    std::vector<BSONObj> _resolvedPipeline;
     // The pipeline defined with the user request, prior to optimization and addition of any view
     // definitions.
     std::vector<BSONObj> _userPipeline;
 
+    std::vector<LetVariable> _letVariables;
+
     boost::intrusive_ptr<DocumentSourceMatch> _matchSrc;
     boost::intrusive_ptr<DocumentSourceUnwind> _unwindSrc;
 
diff --git a/src/mongo/db/pipeline/document_source_lookup_test.cpp b/src/mongo/db/pipeline/document_source_lookup_test.cpp
index e43d8efb5b6..9398473da5a 100644
--- a/src/mongo/db/pipeline/document_source_lookup_test.cpp
+++ b/src/mongo/db/pipeline/document_source_lookup_test.cpp
@@ -117,6 +117,29 @@ TEST_F(DocumentSourceLookUpTest, AcceptsPipelineSyntax) {
     ASSERT_TRUE(lookup->wasConstructedWithPipelineSyntax());
 }
 
+TEST_F(DocumentSourceLookUpTest, AcceptsPipelineWithLetSyntax) {
+    auto expCtx = getExpCtx();
+    NamespaceString fromNs("test", "coll");
+    expCtx->setResolvedNamespace(fromNs, {fromNs, std::vector<BSONObj>{}});
+
+    auto docSource = DocumentSourceLookUp::createFromBson(
+        BSON("$lookup" << BSON("from"
+                               << "coll"
+                               << "let"
+                               << BSON("var1"
+                                       << "$x")
+                               << "pipeline"
+                               << BSON_ARRAY(BSON("$project" << BSON("hasX"
+                                                                     << "$$var1"))
+                                             << BSON("$match" << BSON("$hasX" << true)))
+                               << "as"
+                               << "as"))
+            .firstElement(),
+        expCtx);
+    auto lookup = static_cast<DocumentSourceLookUp*>(docSource.get());
+    ASSERT_TRUE(lookup->wasConstructedWithPipelineSyntax());
+}
+
 TEST_F(DocumentSourceLookUpTest, RejectsLocalFieldForeignFieldWhenPipelineIsSpecified) {
     auto expCtx = getExpCtx();
     NamespaceString fromNs("test", "coll");
@@ -142,10 +165,83 @@ TEST_F(DocumentSourceLookUpTest, RejectsLocalFieldForeignFieldWhenPipelineIsSpec
              << lookupStage->getSourceName()
              << " stage to uassert on mix of localField/foreignField and pipeline options");
     } catch (const UserException& ex) {
-        ASSERT_EQ(40450, ex.getCode());
+        ASSERT_EQ(ErrorCodes::FailedToParse, ex.getCode());
     }
 }
 
+TEST_F(DocumentSourceLookUpTest, RejectsLocalFieldForeignFieldWhenLetIsSpecified) {
+    auto expCtx = getExpCtx();
+    NamespaceString fromNs("test", "coll");
+    expCtx->setResolvedNamespace(fromNs, {fromNs, std::vector<BSONObj>{}});
+
+    ASSERT_THROWS_CODE(DocumentSourceLookUp::createFromBson(BSON("$lookup" << BSON("from"
+                                                                                   << "coll"
+                                                                                   << "let"
+                                                                                   << BSON("var1"
+                                                                                           << "$a")
+                                                                                   << "localField"
+                                                                                   << "a"
+                                                                                   << "foreignField"
+                                                                                   << "b"
+                                                                                   << "as"
+                                                                                   << "as"))
+                                                                .firstElement(),
+                                                            expCtx),
+                       UserException,
+                       ErrorCodes::FailedToParse);
+}
+
+TEST_F(DocumentSourceLookUpTest, RejectsInvalidLetVariableName) {
+    auto expCtx = getExpCtx();
+    NamespaceString fromNs("test", "coll");
+    expCtx->setResolvedNamespace(fromNs, {fromNs, std::vector<BSONObj>{}});
+
+    ASSERT_THROWS_CODE(DocumentSourceLookUp::createFromBson(
+                           BSON("$lookup" << BSON("from"
+                                                  << "coll"
+                                                  << "let"
+                                                  << BSON(""  // Empty variable name.
+                                                          << "$a")
+                                                  << "pipeline"
+                                                  << BSON_ARRAY(BSON("$match" << BSON("x" << 1)))
+                                                  << "as"
+                                                  << "as"))
+                               .firstElement(),
+                           expCtx),
+                       UserException,
+                       16866);
+
+    ASSERT_THROWS_CODE(DocumentSourceLookUp::createFromBson(
+                           BSON("$lookup" << BSON("from"
+                                                  << "coll"
+                                                  << "let"
+                                                  << BSON("^invalidFirstChar"
+                                                          << "$a")
+                                                  << "pipeline"
+                                                  << BSON_ARRAY(BSON("$match" << BSON("x" << 1)))
+                                                  << "as"
+                                                  << "as"))
+                               .firstElement(),
+                           expCtx),
+                       UserException,
+                       16867);
+
+    ASSERT_THROWS_CODE(DocumentSourceLookUp::createFromBson(
+                           BSON("$lookup" << BSON("from"
+                                                  << "coll"
+                                                  << "let"
+                                                  << BSON("contains.invalidChar"
+                                                          << "$a")
+                                                  << "pipeline"
+                                                  << BSON_ARRAY(BSON("$match" << BSON("x" << 1)))
+                                                  << "as"
+                                                  << "as"))
+                               .firstElement(),
+                           expCtx),
+                       UserException,
+                       16868);
+}
+
 TEST_F(DocumentSourceLookUpTest, ShouldBeAbleToReParseSerializedStage) {
     auto expCtx = getExpCtx();
     NamespaceString fromNs("test", "coll");
@@ -154,6 +250,9 @@ TEST_F(DocumentSourceLookUpTest, ShouldBeAbleToReParseSerializedStage) {
     auto lookupStage = DocumentSourceLookUp::createFromBson(
         BSON("$lookup" << BSON("from"
                                << "coll"
+                               << "let"
+                               << BSON("local_x"
+                                       << "$x")
                                << "pipeline"
                                << BSON_ARRAY(BSON("$match" << BSON("x" << 1)))
                                << "as"
@@ -174,10 +273,13 @@ TEST_F(DocumentSourceLookUpTest, ShouldBeAbleToReParseSerializedStage) {
     ASSERT_EQ(serializedDoc["$lookup"].getType(), BSONType::Object);
 
     auto serializedStage = serializedDoc["$lookup"].getDocument();
-    ASSERT_EQ(serializedStage.size(), 3UL);
+    ASSERT_EQ(serializedStage.size(), 4UL);
     ASSERT_VALUE_EQ(serializedStage["from"], Value(std::string("coll")));
     ASSERT_VALUE_EQ(serializedStage["as"], Value(std::string("as")));
 
+    ASSERT_DOCUMENT_EQ(serializedStage["let"].getDocument(),
+                       Document(fromjson("{local_x: \"$x\"}")));
+
     ASSERT_EQ(serializedStage["pipeline"].getType(), BSONType::Array);
     ASSERT_EQ(serializedStage["pipeline"].getArrayLength(), 1UL);
 
diff --git a/src/mongo/db/pipeline/expression_context.h b/src/mongo/db/pipeline/expression_context.h
index 1d0f04dd1a0..e17128ec8a6 100644
--- a/src/mongo/db/pipeline/expression_context.h
+++ b/src/mongo/db/pipeline/expression_context.h
@@ -123,7 +123,6 @@ public:
 
 protected:
     static const int kInterruptCheckPeriod = 128;
-
     ExpressionContext() : variablesParseState(variables.useIdGenerator()) {}
 
     /**
diff --git a/src/mongo/db/pipeline/pipeline_test.cpp b/src/mongo/db/pipeline/pipeline_test.cpp
index 03d0f137612..ce5437cc454 100644
--- a/src/mongo/db/pipeline/pipeline_test.cpp
+++ b/src/mongo/db/pipeline/pipeline_test.cpp
@@ -229,14 +229,14 @@ TEST(PipelineOptimizationTest, LookupShouldCoalesceWithUnwindOnAs) {
 
 TEST(PipelineOptimizationTest, LookupWithPipelineSyntaxShouldCoalesceWithUnwindOnAs) {
     string inputPipe =
-        "[{$lookup: {from : 'lookupColl', as : 'same', pipeline: []}}"
+        "[{$lookup: {from : 'lookupColl', as : 'same', let: {}, pipeline: []}}"
         ",{$unwind: {path: '$same'}}"
         "]";
     string outputPipe =
-        "[{$lookup: {from : 'lookupColl', as : 'same', pipeline: [], "
+        "[{$lookup: {from : 'lookupColl', as : 'same', let: {}, pipeline: [], "
         "unwinding: {preserveNullAndEmptyArrays: false}}}]";
     string serializedPipe =
-        "[{$lookup: {from : 'lookupColl', as : 'same', pipeline: []}}"
+        "[{$lookup: {from : 'lookupColl', as : 'same', let: {}, pipeline: []}}"
         ",{$unwind: {path: '$same'}}"
         "]";
     assertPipelineOptimizesAndSerializesTo(inputPipe, outputPipe, serializedPipe);
@@ -297,7 +297,7 @@ TEST(PipelineOptimizationTest, LookupWithPipelineSyntaxShouldNotCoalesceWithUnwi
         ",{$unwind: {path: '$from'}}"
         "]";
     string outputPipe =
-        "[{$lookup: {from : 'lookupColl', as : 'same', pipeline: []}}"
+        "[{$lookup: {from : 'lookupColl', as : 'same', let: {}, pipeline: []}}"
         ",{$unwind: {path: '$from'}}"
         "]";
     assertPipelineOptimizesTo(inputPipe, outputPipe);
@@ -321,7 +321,7 @@ TEST(PipelineOptimizationTest, LookupWithPipelineSyntaxShouldSwapWithMatch) {
         " {$match: {'independent': 0}}]";
     string outputPipe =
         "[{$match: {independent: 0}}, "
-        " {$lookup: {from: 'lookupColl', as: 'asField', pipeline: []}}]";
+        " {$lookup: {from: 'lookupColl', as: 'asField', let: {}, pipeline: []}}]";
     assertPipelineOptimizesTo(inputPipe, outputPipe);
 }
 
@@ -374,12 +374,12 @@ TEST(PipelineOptimizationTest, LookupWithPipelineSyntaxShouldAbsorbUnwindMatch)
         "{$unwind: '$asField'}, "
         "{$match: {'asField.subfield': {$eq: 1}}}]";
     string outputPipe =
-        "[{$lookup: {from: 'lookupColl', as: 'asField', pipeline: [{$match: {subfield: {$eq: "
-        "1}}}], "
+        "[{$lookup: {from: 'lookupColl', as: 'asField', let: {}, "
+        "pipeline: [{$match: {subfield: {$eq: 1}}}], "
         "unwinding: {preserveNullAndEmptyArrays: false} } } ]";
     string serializedPipe =
-        "[{$lookup: {from: 'lookupColl', as: 'asField', pipeline: [{$match: {subfield: {$eq: "
-        "1}}}]}}, "
+        "[{$lookup: {from: 'lookupColl', as: 'asField', let: {}, "
+        "pipeline: [{$match: {subfield: {$eq: 1}}}]}}, "
         "{$unwind: {path: '$asField'}}]";
     assertPipelineOptimizesAndSerializesTo(inputPipe, outputPipe, serializedPipe);
 }
diff --git a/src/mongo/db/pipeline/variables.h b/src/mongo/db/pipeline/variables.h
index 444686551fb..efbf115269f 100644
--- a/src/mongo/db/pipeline/variables.h
+++ b/src/mongo/db/pipeline/variables.h
@@ -35,10 +35,10 @@
 
 namespace mongo {
 
-/// The state used as input and working space for Expressions.
+/**
+ * The state used as input and working space for Expressions.
+ */
 class Variables {
-    MONGO_DISALLOW_COPYING(Variables);
-
 public:
     // Each unique variable is assigned a unique id of this type. Negative ids are reserved for
     // system variables and non-negative ids are allocated for user variables.
@@ -93,7 +93,6 @@ public:
         return &_idGenerator;
     }
 
-
 private:
     IdGenerator _idGenerator;
     std::vector<Value> _valueList;
@@ -128,6 +127,16 @@ public:
      */
     Variables::Id getVariable(StringData name) const;
 
+    /**
+     * Return a copy of this VariablesParseState. Will replace the copy's '_idGenerator' pointer
+     * with 'idGenerator'.
+     */
+    VariablesParseState copyWith(Variables::IdGenerator* idGenerator) const {
+        VariablesParseState vps = *this;
+        vps._idGenerator = idGenerator;
+        return vps;
+    }
+
 private:
     // Not owned here.
     Variables::IdGenerator* _idGenerator;