intrusive_ptr<DocumentSource> DocumentSourceLookUp::createFromBson(
    BSONElement elem, const boost::intrusive_ptr<ExpressionContext>& pExpCtx) {
    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();

        if (argName == "pipeline") {
            auto result = AggregationRequest::parsePipelineFromBSON(argument);
            if (!result.isOK()) {
                uasserted(ErrorCodes::FailedToParse,
                          str::stream() << "invalid $lookup pipeline definition: "
                                        << result.getStatus().toString());
            }
            pipeline = std::move(result.getValue());
            hasPipeline = true;
            continue;
        }

        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 '" << argName << "' must be a string, found "
                              << argument << ": " << argument.type(),
                argument.type() == BSONType::String);

        if (argName == "from") {
            fromNs = NamespaceString(pExpCtx->ns.db().toString() + '.' + argument.String());
        } else if (argName == "as") {
            as = argument.String();
        } else if (argName == "localField") {
            localField = argument.String();
        } else if (argName == "foreignField") {
            foreignField = argument.String();
        } else {
            uasserted(ErrorCodes::FailedToParse,
                      str::stream() << "unknown argument to $lookup: " << argument.fieldName());
        }
    }

    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(ErrorCodes::FailedToParse,
                "$lookup with 'pipeline' may not specify 'localField' or 'foreignField'",
                localField.empty() && foreignField.empty());

        return new DocumentSourceLookUp(std::move(fromNs),
                                        std::move(as),
                                        std::move(pipeline),
                                        std::move(letVariables),
                                        pExpCtx);
    } else {
        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),
                                        std::move(localField),
                                        std::move(foreignField),
                                        pExpCtx);
    }
}