/** * Copyright (C) 2018-present MongoDB, Inc. * * This program is free software: you can redistribute it and/or modify * it under the terms of the Server Side Public License, version 1, * as published by MongoDB, Inc. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * Server Side Public License for more details. * * You should have received a copy of the Server Side Public License * along with this program. If not, see * . * * As a special exception, the copyright holders give permission to link the * code of portions of this program with the OpenSSL library under certain * conditions as described in each individual source file and distribute * linked combinations including the program with the OpenSSL library. You * must comply with the Server Side Public License in all respects for * all of the code used other than as permitted herein. If you modify file(s) * with this exception, you may extend this exception to your version of the * file(s), but you are not obligated to do so. If you do not wish to do so, * delete this exception statement from your version. If you delete this * exception statement from all source files in the program, then also delete * it in the license file. */ #include "mongo/db/pipeline/document_source_lookup.h" #include "mongo/base/init.h" #include "mongo/db/catalog_shard_feature_flag_gen.h" #include "mongo/db/exec/document_value/document.h" #include "mongo/db/exec/document_value/value.h" #include "mongo/db/jsobj.h" #include "mongo/db/matcher/expression_algo.h" #include "mongo/db/namespace_string.h" #include "mongo/db/pipeline/aggregation_request_helper.h" #include "mongo/db/pipeline/document_path_support.h" #include "mongo/db/pipeline/document_source_documents.h" #include "mongo/db/pipeline/document_source_merge_gen.h" #include "mongo/db/pipeline/document_source_sort.h" #include "mongo/db/pipeline/expression.h" #include "mongo/db/pipeline/expression_context.h" #include "mongo/db/pipeline/expression_dependencies.h" #include "mongo/db/pipeline/sort_reorder_helpers.h" #include "mongo/db/pipeline/variable_validation.h" #include "mongo/db/query/collation/collator_factory_interface.h" #include "mongo/db/query/query_knobs_gen.h" #include "mongo/db/stats/counters.h" #include "mongo/db/views/resolved_view.h" #include "mongo/logv2/log.h" #include "mongo/platform/overflow_arithmetic.h" #include "mongo/util/fail_point.h" #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kQuery namespace mongo { namespace { /** * Constructs a query of the following shape: * {$or: [ * {'fieldName': {$eq: 'values[0]'}}, * {'fieldName': {$eq: 'values[1]'}}, * ... * ]} */ BSONObj buildEqualityOrQuery(const std::string& fieldName, const BSONArray& values) { BSONObjBuilder orBuilder; { BSONArrayBuilder orPredicatesBuilder(orBuilder.subarrayStart("$or")); for (auto&& value : values) { orPredicatesBuilder.append(BSON(fieldName << BSON("$eq" << value))); } } return orBuilder.obj(); } void lookupPipeValidator(const Pipeline& pipeline) { const auto& sources = pipeline.getSources(); std::for_each(sources.begin(), sources.end(), [](auto& src) { uassert(51047, str::stream() << src->getSourceName() << " is not allowed within a $lookup's sub-pipeline", src->constraints().isAllowedInLookupPipeline()); }); } // Parses $lookup 'from' field. The 'from' field must be a string or one of the following // exceptions: // {from: {db: "config", coll: "cache.chunks.*"}, ...} or // {from: {db: "local", coll: "oplog.rs"}, ...} or // {from: {db: "local", coll: "tenantMigration.oplogView"}, ...} . NamespaceString parseLookupFromAndResolveNamespace(const BSONElement& elem, const DatabaseName& defaultDb) { // The object syntax only works for 'cache.chunks.*', 'local.oplog.rs', and // 'local.tenantMigration.oplogViewwhich' which are not user namespaces so object type is // omitted from the error message below. uassert(ErrorCodes::FailedToParse, str::stream() << "$lookup 'from' field must be a string, but found " << typeName(elem.type()), elem.type() == BSONType::String || elem.type() == BSONType::Object); if (elem.type() == BSONType::String) { return NamespaceStringUtil::parseNamespaceFromRequest(defaultDb, elem.valueStringData()); } // Valdate the db and coll names. auto spec = NamespaceSpec::parse( IDLParserContext{elem.fieldNameStringData(), false /* apiStrict */, defaultDb.tenantId()}, elem.embeddedObject()); auto nss = NamespaceStringUtil::parseNamespaceFromRequest(spec.getDb().value_or(DatabaseName()), spec.getColl().value_or("")); uassert( ErrorCodes::FailedToParse, str::stream() << "$lookup with syntax {from: {db:<>, coll:<>},..} is not supported for db: " << nss.dbName().toStringForErrorMsg() << " and coll: " << nss.coll(), nss.isConfigDotCacheDotChunks() || nss == NamespaceString::kRsOplogNamespace || nss == NamespaceString::kTenantMigrationOplogView || nss == NamespaceString::kConfigsvrCollectionsNamespace); return nss; } // Creates the conditions for joining the local and foreign fields inside of a $match. static BSONObj createMatchStageJoinObj(const Document& input, const FieldPath& localFieldPath, const std::string& foreignFieldName) { // Add the 'localFieldPath' of 'input' into 'localFieldList'. If 'localFieldPath' references a // field with an array in its path, we may need to join on multiple values, so we add each // element to 'localFieldList'. BSONArrayBuilder arrBuilder; bool containsRegex = false; document_path_support::visitAllValuesAtPath(input, localFieldPath, [&](const Value& nextValue) { arrBuilder << nextValue; if (!containsRegex && nextValue.getType() == BSONType::RegEx) { containsRegex = true; } }); if (arrBuilder.arrSize() == 0) { // Missing values are treated as null. arrBuilder << BSONNULL; } // We construct a query of one of the following forms, depending on the contents of // 'localFieldList'. // // {: {$eq: }} // if 'localFieldList' contains a single element. // // {: {$in: [, , ...]}} // if 'localFieldList' contains more than one element but doesn't contain any that are // regular expressions. // // {$or: [{: {$eq: }}, {: {$eq: }}, ...]} // if 'localFieldList' contains more than one element and it contains at least one element // that is a regular expression. const auto localFieldListSize = arrBuilder.arrSize(); const auto localFieldList = arrBuilder.arr(); BSONObjBuilder joinObj; if (localFieldListSize > 1) { // A $lookup on an array value corresponds to finding documents in the foreign collection // that have a value of any of the elements in the array value, rather than finding // documents that have a value equal to the entire array value. These semantics are // automatically provided to us by using the $in query operator. if (containsRegex) { // A regular expression inside the $in query operator will perform pattern matching on // any string values. Since we want regular expressions to only match other RegEx types, // we write the query as a $or of equality comparisons instead. return buildEqualityOrQuery(foreignFieldName, localFieldList); } else { // { : { "$in" : } } BSONObjBuilder subObj(joinObj.subobjStart(foreignFieldName)); subObj << "$in" << localFieldList; subObj.doneFast(); return joinObj.obj(); } } // Otherwise we have a simple $eq. // { : { "$eq" : } } BSONObjBuilder subObj(joinObj.subobjStart(foreignFieldName)); subObj << "$eq" << localFieldList[0]; subObj.doneFast(); return joinObj.obj(); } } // namespace DocumentSourceLookUp::DocumentSourceLookUp( NamespaceString fromNs, std::string as, boost::optional> fromCollator, const boost::intrusive_ptr& expCtx) : DocumentSource(kStageName, expCtx), _fromNs(std::move(fromNs)), _as(std::move(as)), _variables(expCtx->variables), _variablesParseState(expCtx->variablesParseState.copyWith(_variables.useIdGenerator())) { if (!_fromNs.isOnInternalDb()) { globalOpCounters.gotNestedAggregate(); } const auto& resolvedNamespace = expCtx->getResolvedNamespace(_fromNs); _resolvedNs = resolvedNamespace.ns; _resolvedPipeline = resolvedNamespace.pipeline; _fromExpCtx = expCtx->copyForSubPipeline(resolvedNamespace.ns, resolvedNamespace.uuid); _fromExpCtx->inLookup = true; if (fromCollator) { _fromExpCtx->setCollator(std::move(fromCollator.value())); _hasExplicitCollation = true; } } DocumentSourceLookUp::DocumentSourceLookUp( NamespaceString fromNs, std::string as, std::string localField, std::string foreignField, boost::optional> fromCollator, const boost::intrusive_ptr& expCtx) : DocumentSourceLookUp(fromNs, as, std::move(fromCollator), expCtx) { _localField = std::move(localField); _foreignField = std::move(foreignField); // We append an additional BSONObj to '_resolvedPipeline' as a placeholder for the $match stage // we'll eventually construct from the input document. _resolvedPipeline.reserve(_resolvedPipeline.size() + 1); _resolvedPipeline.push_back(BSON("$match" << BSONObj())); _fieldMatchPipelineIdx = _resolvedPipeline.size() - 1; initializeResolvedIntrospectionPipeline(); } std::vector extractSourceStage(const std::vector& pipeline) { if (!pipeline.empty() && (pipeline[0].hasField(DocumentSourceDocuments::kStageName) || pipeline[0].hasField("$search"_sd))) { return {pipeline[0]}; } return {}; } DocumentSourceLookUp::DocumentSourceLookUp( NamespaceString fromNs, std::string as, std::vector pipeline, BSONObj letVariables, boost::optional> fromCollator, boost::optional> localForeignFields, const boost::intrusive_ptr& expCtx) : DocumentSourceLookUp(fromNs, as, std::move(fromCollator), expCtx) { // '_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 will then append stages to ensure any view prefix is not overwritten. if (localForeignFields != boost::none) { std::tie(_localField, _foreignField) = *localForeignFields; // Append a BSONObj to '_resolvedPipeline' as a placeholder for the stage corresponding to // the local/foreignField $match. It must next after $documents if present. auto sourceStages = extractSourceStage(pipeline); _resolvedPipeline.insert(_resolvedPipeline.end(), sourceStages.begin(), sourceStages.end()); _resolvedPipeline.push_back(BSON("$match" << BSONObj())); _fieldMatchPipelineIdx = _resolvedPipeline.size() - 1; // Add the user pipeline to '_resolvedPipeline' after any potential view prefix and $match _resolvedPipeline.insert( _resolvedPipeline.end(), pipeline.begin() + sourceStages.size(), pipeline.end()); } else { // When local/foreignFields are included, we cannot enable the cache because the $match // is a correlated prefix that will not be detected. Here, local/foreignFields are absent, // so we enable the cache. _cache.emplace(internalDocumentSourceLookupCacheSizeBytes.load()); // Add the user pipeline to '_resolvedPipeline' after any potential view prefix and $match _resolvedPipeline.insert(_resolvedPipeline.end(), pipeline.begin(), pipeline.end()); } _userPipeline = std::move(pipeline); for (auto&& varElem : letVariables) { const auto varName = varElem.fieldNameStringData(); variableValidation::validateNameForUserWrite(varName); _letVariables.emplace_back( varName.toString(), Expression::parseOperand(expCtx.get(), varElem, expCtx->variablesParseState), _variablesParseState.defineVariable(varName)); } initializeResolvedIntrospectionPipeline(); } DocumentSourceLookUp::DocumentSourceLookUp(const DocumentSourceLookUp& original, const boost::intrusive_ptr& newExpCtx) : DocumentSource(kStageName, newExpCtx), _fromNs(original._fromNs), _resolvedNs(original._resolvedNs), _as(original._as), _additionalFilter(original._additionalFilter), _localField(original._localField), _foreignField(original._foreignField), _fieldMatchPipelineIdx(original._fieldMatchPipelineIdx), _variables(original._variables), _variablesParseState(original._variablesParseState.copyWith(_variables.useIdGenerator())), _fromExpCtx(original._fromExpCtx->copyWith(_resolvedNs, original._fromExpCtx->uuid)), _hasExplicitCollation(original._hasExplicitCollation), _resolvedPipeline(original._resolvedPipeline), _userPipeline(original._userPipeline), _resolvedIntrospectionPipeline(original._resolvedIntrospectionPipeline->clone()), _letVariables(original._letVariables) { if (!_localField && !_foreignField) { _cache.emplace(internalDocumentSourceCursorBatchSizeBytes.load()); } if (original._matchSrc) { _matchSrc = static_cast(original._matchSrc->clone(pExpCtx).get()); } if (original._unwindSrc) { _unwindSrc = static_cast(original._unwindSrc->clone(pExpCtx).get()); } } boost::intrusive_ptr DocumentSourceLookUp::clone( const boost::intrusive_ptr& newExpCtx) const { return make_intrusive(*this, newExpCtx); } void validateLookupCollectionlessPipeline(const std::vector& pipeline) { uassert(ErrorCodes::FailedToParse, "$lookup stage without explicit collection must have a pipeline with $documents as " "first stage", pipeline.size() > 0 && // TODO SERVER-59628 We should be able to check for any valid data source here, not // just $documents. !pipeline[0].getField(DocumentSourceDocuments::kStageName).eoo()); } void validateLookupCollectionlessPipeline(const BSONElement& pipeline) { uassert(ErrorCodes::FailedToParse, "must specify 'pipeline' when 'from' is empty", pipeline); auto parsedPipeline = parsePipelineFromBSON(pipeline); validateLookupCollectionlessPipeline(parsedPipeline); } std::unique_ptr DocumentSourceLookUp::LiteParsed::parse( const NamespaceString& nss, const BSONElement& spec) { 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"]; auto pipelineElem = specObj["pipeline"]; NamespaceString fromNss; if (!fromElement) { validateLookupCollectionlessPipeline(pipelineElem); fromNss = NamespaceString::makeCollectionlessAggregateNSS(nss.dbName()); } else { fromNss = parseLookupFromAndResolveNamespace(fromElement, nss.dbName()); } uassert(ErrorCodes::InvalidNamespace, str::stream() << "invalid $lookup namespace: " << fromNss.toStringForErrorMsg(), fromNss.isValid()); // Recursively lite parse the nested pipeline, if one exists. boost::optional liteParsedPipeline; if (pipelineElem) { auto pipeline = parsePipelineFromBSON(pipelineElem); liteParsedPipeline = LiteParsedPipeline(fromNss, pipeline); } bool hasInternalCollation = static_cast(specObj["_internalCollation"]); return std::make_unique( spec.fieldName(), std::move(fromNss), std::move(liteParsedPipeline), hasInternalCollation); } PrivilegeVector DocumentSourceLookUp::LiteParsed::requiredPrivileges( bool isMongos, bool bypassDocumentValidation) const { PrivilegeVector requiredPrivileges; invariant(_pipelines.size() <= 1); invariant(_foreignNss); // If no pipeline is specified or the local/foreignField syntax was used, then assume that we're // reading directly from the collection. if (_pipelines.empty() || !_pipelines[0].startsWithInitialSource()) { Privilege::addPrivilegeToPrivilegeVector( &requiredPrivileges, Privilege(ResourcePattern::forExactNamespace(*_foreignNss), ActionType::find)); } // Add the sub-pipeline privileges, if one was specified. if (!_pipelines.empty()) { const LiteParsedPipeline& pipeline = _pipelines[0]; Privilege::addPrivilegesToPrivilegeVector( &requiredPrivileges, std::move(pipeline.requiredPrivileges(isMongos, bypassDocumentValidation))); } return requiredPrivileges; } REGISTER_DOCUMENT_SOURCE(lookup, DocumentSourceLookUp::LiteParsed::parse, DocumentSourceLookUp::createFromBson, AllowedWithApiStrict::kConditionally); const char* DocumentSourceLookUp::getSourceName() const { return kStageName.rawData(); } bool DocumentSourceLookUp::foreignShardedLookupAllowed() const { return !pExpCtx->opCtx->inMultiDocumentTransaction(); } void DocumentSourceLookUp::determineSbeCompatibility() { _sbeCompatibility = pExpCtx->sbeCompatibility; // This stage has the SBE compatibility as least the same as that of the expression context. auto sbeCompatibleByStageConfig = // We currently only support lowering equi-join that uses localField/foreignField // syntax. !_userPipeline && _localField && _foreignField // SBE doesn't support match-like paths with numeric components. (Note: "as" field is a // project-like field and numbers in it are treated as literal names of fields rather // than indexes into arrays, which is compatible with SBE.) && !FieldRef(_localField->fullPath()).hasNumericPathComponents() && !FieldRef(_foreignField->fullPath()).hasNumericPathComponents() // Setting a collator on an individual $lookup stage with _internalCollation isn't supported && !_hasExplicitCollation // We currently don't lower $lookup against views ('_fromNs' does not correspond to a // view). && pExpCtx->getResolvedNamespace(_fromNs).pipeline.empty(); if (!sbeCompatibleByStageConfig) { _sbeCompatibility = SbeCompatibility::notCompatible; } } StageConstraints DocumentSourceLookUp::constraints(Pipeline::SplitState pipeState) const { HostTypeRequirement hostRequirement; if (_fromNs.isConfigDotCacheDotChunks()) { // $lookup from config.cache.chunks* namespaces is permitted to run on each individual // shard, rather than just the primary, since each shard should have an identical copy of // the namespace. hostRequirement = HostTypeRequirement::kAnyShard; } else if (pipeState == Pipeline::SplitState::kSplitForShards) { // This stage will only be on the shards pipeline if $lookup on sharded foreign collections // is allowed. hostRequirement = HostTypeRequirement::kAnyShard; } else if (_fromNs == NamespaceString::kConfigsvrCollectionsNamespace && // (Ignore FCV check): If the config shard feature flag is enabled, the config // server should have the components necessary to handle a merge. Config servers are // upgraded first and downgraded last, so if any server is running the latest binary, // we can assume the conifg servers are too. !gFeatureFlagCatalogShard.isEnabledAndIgnoreFCVUnsafe()) { // This is an unsharded collection, but the primary shard would be the config server, and // the config servers are not prepared to take queries. Instead, we'll merge on any of the // other shards. hostRequirement = HostTypeRequirement::kAnyShard; } else { // If the pipeline is unsplit or this stage is on the merging part of the pipeline, // when $lookup on sharded foreign collections is allowed, the foreign collection is // sharded, and the stage is executing on mongos, the stage can run on mongos or any shard. hostRequirement = (foreignShardedLookupAllowed() && pExpCtx->inMongos && pExpCtx->mongoProcessInterface->isSharded(pExpCtx->opCtx, _fromNs)) ? HostTypeRequirement::kNone : HostTypeRequirement::kPrimaryShard; } // By default, $lookup is allowed in a transaction and does not use disk. StageConstraints constraints(StreamType::kStreaming, PositionRequirement::kNone, hostRequirement, DiskUseRequirement::kNoDiskUse, FacetRequirement::kAllowed, TransactionRequirement::kAllowed, LookupRequirement::kAllowed, UnionRequirement::kAllowed); // However, if $lookup is specified with a pipeline, it inherits the strictest disk use, facet, // transaction, and lookup requirements from the children in its pipeline. if (hasPipeline()) { constraints = StageConstraints::getStrictestConstraints( _resolvedIntrospectionPipeline->getSources(), constraints); } constraints.canSwapWithMatch = true; constraints.canSwapWithSkippingOrLimitingStage = !_unwindSrc; return constraints; } DocumentSource::GetNextResult DocumentSourceLookUp::doGetNext() { if (_unwindSrc) { return unwindResult(); } auto nextInput = pSource->getNext(); if (!nextInput.isAdvanced()) { return nextInput; } auto inputDoc = nextInput.releaseDocument(); // 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. invariant(!_matchSrc); if (hasLocalFieldForeignFieldJoin()) { auto matchStage = makeMatchStageFromInput(inputDoc, *_localField, _foreignField->fullPath(), BSONObj()); // We've already allocated space for the trailing $match stage in '_resolvedPipeline'. _resolvedPipeline[*_fieldMatchPipelineIdx] = matchStage; } std::unique_ptr pipeline; try { pipeline = buildPipeline(inputDoc); } catch (const ExceptionForCat& ex) { // If lookup on a sharded collection is disallowed and the foreign collection is sharded, // throw a custom exception. if (auto staleInfo = ex.extraInfo(); staleInfo && staleInfo->getVersionWanted() && staleInfo->getVersionWanted() != ShardVersion::UNSHARDED()) { uassert(3904800, "Cannot run $lookup with a sharded foreign collection in a transaction", foreignShardedLookupAllowed()); } throw; } std::vector results; long long objsize = 0; const auto maxBytes = internalLookupStageIntermediateDocumentMaxSizeBytes.load(); while (auto result = pipeline->getNext()) { long long safeSum = 0; bool hasOverflowed = overflow::add(objsize, result->getApproximateSize(), &safeSum); uassert(4568, str::stream() << "Total size of documents in " << _fromNs.coll() << " matching pipeline's $lookup stage exceeds " << maxBytes << " bytes", !hasOverflowed && objsize <= maxBytes); objsize = safeSum; results.emplace_back(std::move(*result)); } accumulatePipelinePlanSummaryStats(*pipeline, _stats.planSummaryStats); // Check if pipeline uses disk. _stats.planSummaryStats.usedDisk = _stats.planSummaryStats.usedDisk || pipeline->usedDisk(); MutableDocument output(std::move(inputDoc)); output.setNestedField(_as, Value(std::move(results))); return output.freeze(); } std::unique_ptr DocumentSourceLookUp::buildPipelineFromViewDefinition( std::vector serializedPipeline, ExpressionContext::ResolvedNamespace resolvedNamespace) { // We don't want to optimize or attach a cursor source here because we need to update // _resolvedPipeline so we can reuse it on subsequent calls to getNext(), and we may need to // update _fieldMatchPipelineIdx as well in the case of a field join. MakePipelineOptions opts; opts.optimize = false; opts.attachCursorSource = false; opts.validator = lookupPipeValidator; // Resolve the view definition. auto pipeline = Pipeline::makePipelineFromViewDefinition( _fromExpCtx, resolvedNamespace, serializedPipeline, opts); // Store the pipeline with resolved namespaces so that we only trigger this exception on the // first input document. _resolvedPipeline = pipeline->serializeToBson(); // The index of the field join match stage needs to be set to the length of the view // pipeline, as it is no longer the first stage in the resolved pipeline. if (hasLocalFieldForeignFieldJoin()) { _fieldMatchPipelineIdx = resolvedNamespace.pipeline.size(); } // Update the expression context with any new namespaces the resolved pipeline has introduced. LiteParsedPipeline liteParsedPipeline(resolvedNamespace.ns, resolvedNamespace.pipeline); _fromExpCtx = _fromExpCtx->copyWith(resolvedNamespace.ns, resolvedNamespace.uuid); _fromExpCtx->addResolvedNamespaces(liteParsedPipeline.getInvolvedNamespaces()); return pipeline; } std::unique_ptr DocumentSourceLookUp::buildPipeline( const Document& inputDoc) { // Copy all 'let' variables into the foreign pipeline's expression context. _variables.copyToExpCtx(_variablesParseState, _fromExpCtx.get()); _fromExpCtx->forcePlanCache = true; // Resolve the 'let' variables to values per the given input document. resolveLetVariables(inputDoc, &_fromExpCtx->variables); std::unique_ptr expectUnshardedCollectionInScope; const auto allowForeignShardedColl = foreignShardedLookupAllowed(); if (!allowForeignShardedColl) { // Enforce that the foreign collection must be unsharded for lookup. expectUnshardedCollectionInScope = _fromExpCtx->mongoProcessInterface->expectUnshardedCollectionInScope( _fromExpCtx->opCtx, _fromExpCtx->ns, boost::none); } // If we don't have a cache, build and return the pipeline immediately. if (!_cache || _cache->isAbandoned()) { MakePipelineOptions pipelineOpts; pipelineOpts.optimize = true; pipelineOpts.attachCursorSource = true; pipelineOpts.validator = lookupPipeValidator; // By default, $lookup doesnt support sharded 'from' collections. pipelineOpts.shardTargetingPolicy = allowForeignShardedColl ? ShardTargetingPolicy::kAllowed : ShardTargetingPolicy::kNotAllowed; try { return Pipeline::makePipeline(_resolvedPipeline, _fromExpCtx, pipelineOpts); } catch (const ExceptionFor& e) { // This exception returns the information we need to resolve a sharded view. Update the // pipeline with the resolved view definition. auto pipeline = buildPipelineFromViewDefinition( _resolvedPipeline, ExpressionContext::ResolvedNamespace{e->getNamespace(), e->getPipeline()}); LOGV2_DEBUG(3254800, 3, "$lookup found view definition. ns: {namespace}, pipeline: {pipeline}. New " "$lookup sub-pipeline: {new_pipe}", logAttrs(e->getNamespace()), "pipeline"_attr = Value(e->getPipeline()), "new_pipe"_attr = _resolvedPipeline); // We can now safely optimize and reattempt attaching the cursor source. pipeline = Pipeline::makePipeline(_resolvedPipeline, _fromExpCtx, pipelineOpts); return pipeline; } } // Construct the basic pipeline without a cache stage. Avoid optimizing here since we need to // add the cache first, as detailed below. MakePipelineOptions pipelineOpts; pipelineOpts.optimize = false; pipelineOpts.attachCursorSource = false; pipelineOpts.validator = lookupPipeValidator; auto pipeline = Pipeline::makePipeline(_resolvedPipeline, _fromExpCtx, pipelineOpts); // We can store the unoptimized serialization of the pipeline so that if we need to resolve // a sharded view later on, and we have a local-foreign field join, we will need to update // metadata tracking the position of this join in the _resolvedPipeline. auto serializedPipeline = pipeline->serializeToBson(); addCacheStageAndOptimize(*pipeline); if (!_cache->isServing()) { // The cache has either been abandoned or has not yet been built. Attach a cursor. auto shardTargetingPolicy = allowForeignShardedColl ? ShardTargetingPolicy::kAllowed : ShardTargetingPolicy::kNotAllowed; try { pipeline = pExpCtx->mongoProcessInterface->attachCursorSourceToPipeline( pipeline.release(), shardTargetingPolicy); } catch (const ExceptionFor& e) { // This exception returns the information we need to resolve a sharded view. Update the // pipeline with the resolved view definition. pipeline = buildPipelineFromViewDefinition( serializedPipeline, ExpressionContext::ResolvedNamespace{e->getNamespace(), e->getPipeline()}); // The serialized pipeline does not have a cache stage, so we will add it back to the // pipeline here if the cache has not been abandoned. if (_cache && !_cache->isAbandoned()) { addCacheStageAndOptimize(*pipeline); } LOGV2_DEBUG(3254801, 3, "$lookup found view definition. ns: {namespace}, pipeline: {pipeline}. New " "$lookup sub-pipeline: {new_pipe}", logAttrs(e->getNamespace()), "pipeline"_attr = Value(e->getPipeline()), "new_pipe"_attr = _resolvedPipeline); // Try to attach the cursor source again. pipeline = pExpCtx->mongoProcessInterface->attachCursorSourceToPipeline( pipeline.release(), shardTargetingPolicy); } } // If the cache has been abandoned, release it. if (_cache->isAbandoned()) { _cache.reset(); } invariant(pipeline); return pipeline; } void DocumentSourceLookUp::addCacheStageAndOptimize(Pipeline& pipeline) { // Add the cache stage at the end and optimize. During the optimization process, the cache will // either move itself to the correct position in the pipeline, or will abandon itself if no // suitable cache position exists. Do it only if pipeline optimization is enabled, otherwise // Pipeline::optimizePipeline() will exit early and correct placement of the cache will not // occur. if (auto fp = globalFailPointRegistry().find("disablePipelineOptimization"); fp && fp->shouldFail()) { _cache->abandon(); } else { pipeline.addFinalSource( DocumentSourceSequentialDocumentCache::create(_fromExpCtx, _cache.get_ptr())); } pipeline.optimizePipeline(); } DocumentSource::GetModPathsReturn DocumentSourceLookUp::getModifiedPaths() const { OrderedPathSet modifiedPaths{_as.fullPath()}; if (_unwindSrc) { auto pathsModifiedByUnwind = _unwindSrc->getModifiedPaths(); invariant(pathsModifiedByUnwind.type == GetModPathsReturn::Type::kFiniteSet); modifiedPaths.insert(pathsModifiedByUnwind.paths.begin(), pathsModifiedByUnwind.paths.end()); } return {GetModPathsReturn::Type::kFiniteSet, std::move(modifiedPaths), {}}; } Pipeline::SourceContainer::iterator DocumentSourceLookUp::doOptimizeAt( Pipeline::SourceContainer::iterator itr, Pipeline::SourceContainer* container) { invariant(*itr == this); if (std::next(itr) == container->end()) { return container->end(); } // If the following stage is $sort and there is no internal $unwind, consider pushing it ahead // of $lookup. if (!_unwindSrc) { itr = tryReorderingWithSort(itr, container); if (*itr != this) { return itr; } } auto nextUnwind = dynamic_cast((*std::next(itr)).get()); // If we are not already handling an $unwind stage internally, we can combine with the // following $unwind stage. if (nextUnwind && !_unwindSrc && nextUnwind->getUnwindPath() == _as.fullPath()) { _unwindSrc = std::move(nextUnwind); // We cannot push absorbed $unwind stages into SBE. _sbeCompatibility = SbeCompatibility::notCompatible; container->erase(std::next(itr)); return itr; } // Attempt to internalize any predicates of a $match upon the "_as" field. auto nextMatch = dynamic_cast((*std::next(itr)).get()); if (!nextMatch) { return std::next(itr); } if (!_unwindSrc || _unwindSrc->indexPath() || _unwindSrc->preserveNullAndEmptyArrays()) { // We must be unwinding our result to internalize a $match. For example, consider the // following pipeline: // // Input: {_id: 0} // Foreign Collection: {a: 0, b: 0}, {a: 0, b: 5} // Pipeline: // {$lookup: {localField: "_id", foreignField: "a", as: "foo"}} // {$match: {'foo.b': {$gt: 0}}} // Output: {_id: 0, foo: [{a: 0, b: 0}, {a: 0, b: 5}]} // // If we executed {b: {$gt: 0}} as part of our $lookup, our output would instead be: // {_id: 0, foo: [{a: 0, b: 5}]} // // However, if we are already unwinding 'foo', then we can move the $match inside, since it // will have the same effect as filtering the unwound results, that is, the output will be: // {_id: 0, foo: {a: 0, b: 5}} // // Note that we cannot absorb a $match if the absorbed $unwind has // "preserveNullAndEmptyArrays" set to true, for the following reason: A document that had // an empty output array from $lookup would be preserved by the $unwind, but could be // removed by the $match. However, if we absorb the $match into the $lookup, our joined // query inside the $lookup will output an empty array, which $unwind will then preserve. // Thus, depending on the optimization, the user would see a different output. // // In addition, we must avoid internalizing a $match if an absorbed $unwind has an // "includeArrayIndex" option, since the $match will alter the indices of the returned // values. return std::next(itr); } // We cannot internalize a $match if a collation has been set on the $lookup stage and it // differs from that of the parent pipeline. if (_fromExpCtx->getCollator() && !CollatorInterface::collatorsMatch(pExpCtx->getCollator(), _fromExpCtx->getCollator())) { return std::next(itr); } auto outputPath = _as.fullPath(); // Since $match splitting is handled in a generic way, we expect to have already swapped // portions of the $match that do not depend on the 'as' path or on an internalized $unwind's // index path before ourselves. But due to the early return above, we know there is no // internalized $unwind with an index path. // // Therefore, 'nextMatch' should only depend on the 'as' path. We now try to absorb the match on // the 'as' path in order to push down these predicates into the foreign collection. bool isMatchOnlyOnAs = true; auto computeWhetherMatchOnAs = [&isMatchOnlyOnAs, &outputPath](MatchExpression* expression, std::string path) -> void { // If 'expression' is the child of a $elemMatch, we cannot internalize the $match. For // example, {b: {$elemMatch: {$gt: 1, $lt: 4}}}, where "b" is our "_as" field. This is // because there's no way to modify the expression to be a match just on 'b'--we cannot // change the path to an empty string, or remove the node entirely. if (expression->matchType() == MatchExpression::ELEM_MATCH_VALUE || expression->matchType() == MatchExpression::ELEM_MATCH_OBJECT) { isMatchOnlyOnAs = false; } if (expression->numChildren() == 0) { // 'expression' is a leaf node; examine the path. It is important that 'outputPath' // not equal 'path', because we cannot change the expression {b: {$eq: 3}}, where // 'path' is 'b', to be a match on a subfield, since no subfield exists. isMatchOnlyOnAs = isMatchOnlyOnAs && expression::isPathPrefixOf(outputPath, path); } }; expression::mapOver(nextMatch->getMatchExpression(), computeWhetherMatchOnAs); if (!isMatchOnlyOnAs) { // "nextMatch" does not contain any predicates that can be absorbed into this stage. return std::next(itr); } // We can internalize the $match. This $lookup should already be marked as SBE incompatible // because a $match can only be internalized if an $unwind, which is SBE incompatible, was // absorbed as well. tassert(5843701, "This $lookup cannot be compatible with SBE", _sbeCompatibility == SbeCompatibility::notCompatible); if (!_matchSrc) { _matchSrc = nextMatch; } else { // We have already absorbed a $match. We need to join it with 'dependent'. _matchSrc->joinMatchWith(nextMatch); } // Remove the original $match. container->erase(std::next(itr)); // We have internalized a $match, but have not yet computed the descended $match that should // be applied to our queries. _additionalFilter = DocumentSourceMatch::descendMatchOnPath( _matchSrc->getMatchExpression(), _as.fullPath(), pExpCtx) ->getQuery() .getOwned(); // Add '_additionalFilter' to '_resolvedPipeline' if there is a pipeline. If there is no // pipeline, '_additionalFilter' can safely be added to the local/foreignField $match stage // during 'doGetNext()'. if (hasPipeline()) { auto matchObj = BSON("$match" << *_additionalFilter); _resolvedPipeline.push_back(matchObj); } // There may be further optimization between this $lookup and the new neighbor, so we return an // iterator pointing to ourself. return itr; } bool DocumentSourceLookUp::usedDisk() { if (_pipeline) _stats.planSummaryStats.usedDisk = _stats.planSummaryStats.usedDisk || _pipeline->usedDisk(); return _stats.planSummaryStats.usedDisk; } void DocumentSourceLookUp::doDispose() { if (_pipeline) { accumulatePipelinePlanSummaryStats(*_pipeline, _stats.planSummaryStats); _pipeline->dispose(pExpCtx->opCtx); _pipeline.reset(); } } BSONObj DocumentSourceLookUp::makeMatchStageFromInput(const Document& input, const FieldPath& localFieldPath, const std::string& foreignFieldName, const BSONObj& additionalFilter) { // We wrap the query in a $match so that it can be parsed into a DocumentSourceMatch when // constructing a pipeline to execute. BSONObjBuilder match; BSONObj joinObj = createMatchStageJoinObj(input, localFieldPath, foreignFieldName); // If we have one condition, do not place inside a $and. This BSON could be created many times, // so we want to produce simple queries for the planner if possible. if (additionalFilter.isEmpty()) { match << "$match" << joinObj; } else { BSONObjBuilder query(match.subobjStart("$match")); BSONArrayBuilder andObj(query.subarrayStart("$and")); andObj.append(joinObj); andObj.append(additionalFilter); andObj.doneFast(); query.doneFast(); } return match.obj(); } DocumentSource::GetNextResult DocumentSourceLookUp::unwindResult() { const boost::optional indexPath(_unwindSrc->indexPath()); // Loop until we get a document that has at least one match. // Note we may return early from this loop if our source stage is exhausted or if the unwind // source was asked to return empty arrays and we get a document without a match. while (!_pipeline || !_nextValue) { // Accumulate stats from the pipeline for the previous input, if applicable. This is to // avoid missing the accumulation of stats on an early exit (below) if the input (i.e., left // side of the lookup) is done. if (_pipeline) { accumulatePipelinePlanSummaryStats(*_pipeline, _stats.planSummaryStats); _pipeline->dispose(pExpCtx->opCtx); } auto nextInput = pSource->getNext(); if (!nextInput.isAdvanced()) { return nextInput; } _input = nextInput.releaseDocument(); if (hasLocalFieldForeignFieldJoin()) { // At this point, if there is a pipeline, '_additionalFilter' was added to the end of // '_resolvedPipeline' in doOptimizeAt(). If there is no pipeline, we must add it to the // $match stage created here. BSONObj filter = hasPipeline() ? BSONObj() : _additionalFilter.value_or(BSONObj()); auto matchStage = makeMatchStageFromInput(*_input, *_localField, _foreignField->fullPath(), filter); // We've already allocated space for the trailing $match stage in '_resolvedPipeline'. _resolvedPipeline[*_fieldMatchPipelineIdx] = matchStage; } _pipeline = buildPipeline(*_input); // 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 // outer Pipeline that will propagate dispose() calls before being destroyed. _pipeline.get_deleter().dismissDisposal(); _cursorIndex = 0; _nextValue = _pipeline->getNext(); if (_unwindSrc->preserveNullAndEmptyArrays() && !_nextValue) { // There were no results for this cursor, but the $unwind was asked to preserve empty // arrays, so we should return a document without the array. MutableDocument output(std::move(*_input)); // Note this will correctly create objects in the prefix of '_as', to act as if we had // created an empty array and then removed it. output.setNestedField(_as, Value()); if (indexPath) { output.setNestedField(*indexPath, Value(BSONNULL)); } return output.freeze(); } } invariant(bool(_input) && bool(_nextValue)); auto currentValue = *_nextValue; _nextValue = _pipeline->getNext(); // Move input document into output if this is the last or only result, otherwise perform a copy. MutableDocument output(_nextValue ? *_input : std::move(*_input)); output.setNestedField(_as, Value(currentValue)); if (indexPath) { output.setNestedField(*indexPath, Value(_cursorIndex)); } ++_cursorIndex; return output.freeze(); } void DocumentSourceLookUp::resolveLetVariables(const Document& localDoc, Variables* variables) { invariant(variables); for (auto& letVar : _letVariables) { auto value = letVar.expression->evaluate(localDoc, &pExpCtx->variables); variables->setConstantValue(letVar.id, value); } } void DocumentSourceLookUp::initializeResolvedIntrospectionPipeline() { _variables.copyToExpCtx(_variablesParseState, _fromExpCtx.get()); _fromExpCtx->startExpressionCounters(); _resolvedIntrospectionPipeline = Pipeline::parse(_resolvedPipeline, _fromExpCtx, lookupPipeValidator); _fromExpCtx->stopExpressionCounters(); } void DocumentSourceLookUp::appendSpecificExecStats(MutableDocument& doc) const { const PlanSummaryStats& stats = _stats.planSummaryStats; doc["totalDocsExamined"] = Value(static_cast(stats.totalDocsExamined)); doc["totalKeysExamined"] = Value(static_cast(stats.totalKeysExamined)); doc["collectionScans"] = Value(stats.collectionScans); std::vector indexesUsedVec; std::transform(stats.indexesUsed.begin(), stats.indexesUsed.end(), std::back_inserter(indexesUsedVec), [](std::string idx) -> Value { return Value(idx); }); doc["indexesUsed"] = Value{std::move(indexesUsedVec)}; } void DocumentSourceLookUp::serializeToArray(std::vector& array, SerializationOptions opts) const { // Support alternative $lookup from config.cache.chunks* namespaces. // // Do not include the tenantId in serialized 'from' namespace. auto fromValue = (pExpCtx->ns.db() == _fromNs.db()) ? Value(opts.serializeIdentifier(_fromNs.coll())) : Value(Document{{"db", opts.serializeIdentifier(_fromNs.dbName().db())}, {"coll", opts.serializeIdentifier(_fromNs.coll())}}); MutableDocument output(Document{ {getSourceName(), Document{{"from", fromValue}, {"as", opts.serializeFieldPath(_as)}}}}); if (hasLocalFieldForeignFieldJoin()) { output[getSourceName()]["localField"] = Value(opts.serializeFieldPath(_localField.value())); output[getSourceName()]["foreignField"] = Value(opts.serializeFieldPath(_foreignField.value())); } // Add a pipeline field if only-pipeline syntax was used (to ensure the output is valid $lookup // syntax) or if a $match was absorbed. auto serializedPipeline = [&]() -> std::vector { auto pipeline = _userPipeline.get_value_or(std::vector()); if (opts.applyHmacToIdentifiers || opts.replacementForLiteralArgs) { return Pipeline::parse(pipeline, _fromExpCtx)->serializeToBson(opts); } return pipeline; }(); if (_additionalFilter) { auto serializedFilter = [&]() -> BSONObj { if (opts.applyHmacToIdentifiers || opts.replacementForLiteralArgs) { auto filter = uassertStatusOK(MatchExpressionParser::parse(*_additionalFilter, pExpCtx)); return filter->serialize(opts); } return *_additionalFilter; }(); serializedPipeline.emplace_back(BSON("$match" << serializedFilter)); } if (!hasLocalFieldForeignFieldJoin() || serializedPipeline.size() > 0) { MutableDocument exprList; for (const auto& letVar : _letVariables) { exprList.addField(opts.serializeFieldPathFromString(letVar.name), letVar.expression->serialize(opts)); } output[getSourceName()]["let"] = Value(exprList.freeze()); output[getSourceName()]["pipeline"] = Value(serializedPipeline); } if (_hasExplicitCollation) { output[getSourceName()]["_internalCollation"] = Value(_fromExpCtx->getCollatorBSON()); } if (opts.verbosity) { if (_unwindSrc) { const boost::optional indexPath = _unwindSrc->indexPath(); output[getSourceName()]["unwinding"] = Value(DOC("preserveNullAndEmptyArrays" << _unwindSrc->preserveNullAndEmptyArrays() << "includeArrayIndex" << (indexPath ? Value(indexPath->fullPath()) : Value()))); } if (opts.verbosity.value() >= ExplainOptions::Verbosity::kExecStats) { appendSpecificExecStats(output); } array.push_back(output.freezeToValue()); } else { array.push_back(output.freezeToValue()); if (_unwindSrc) { _unwindSrc->serializeToArray(array); } } } DepsTracker::State DocumentSourceLookUp::getDependencies(DepsTracker* deps) const { if (hasPipeline() || _letVariables.size() > 0) { // We will use the introspection pipeline which we prebuilt during construction. invariant(_resolvedIntrospectionPipeline); // We are not attempting to enforce that any referenced metadata are in fact unavailable, // this is done elsewhere. We only need to know what variable dependencies exist in the // subpipeline for the top-level pipeline. So without knowledge of what metadata is in fact // unavailable, we "lie" and say that all metadata is available to avoid tripping any // assertions. DepsTracker subDeps(DepsTracker::kNoMetadata); // Get the subpipeline dependencies. Subpipeline stages may reference both 'let' variables // declared by this $lookup and variables declared externally. for (auto&& source : _resolvedIntrospectionPipeline->getSources()) { source->getDependencies(&subDeps); } // Add the 'let' dependencies to the tracker. for (auto&& letVar : _letVariables) { expression::addDependencies(letVar.expression.get(), deps); } } if (hasLocalFieldForeignFieldJoin()) { const FieldRef ref(_localField->fullPath()); // We need everything up until the first numeric component. Otherwise, a projection could // treat the numeric component as a field name rather than an index into an array. size_t firstNumericIx; for (firstNumericIx = 0; firstNumericIx < ref.numParts(); firstNumericIx++) { // We are lenient with the component, because classic $lookup treats 0-prefixed numeric // fields like "00" as both an index and a field name. Allowing it in a dependency would // restrict the usage to only a field name. if (ref.isNumericPathComponentLenient(firstNumericIx)) { break; } } deps->fields.insert(ref.dottedSubstring(0, firstNumericIx).toString()); } // Purposely ignore '_matchSrc' and '_unwindSrc', since those should only be absorbed if we know // they are only operating on the "as" field which will be generated by this stage. return DepsTracker::State::SEE_NEXT; } void DocumentSourceLookUp::addVariableRefs(std::set* refs) const { // Do not add SEARCH_META as a reference, since it is scoped to one pipeline. if (hasPipeline()) { std::set subPipeRefs; _resolvedIntrospectionPipeline->addVariableRefs(&subPipeRefs); for (auto&& varId : subPipeRefs) { if (varId != Variables::kSearchMetaId) refs->insert(varId); } } // Add the 'let' variable references. Because the caller is only interested in references to // external variables, filter out any subpipeline references to 'let' variables declared by this // $lookup. This step must happen after gathering the sub-pipeline variable references as they // may refer to let variables. for (auto&& letVar : _letVariables) { expression::addVariableRefs(letVar.expression.get(), refs); refs->erase(letVar.id); } } boost::optional DocumentSourceLookUp::distributedPlanLogic() { // If $lookup into a sharded foreign collection is allowed and the foreign namespace is sharded, // top-level $lookup stages can run in parallel on the shards. // // Note that this decision is inherently racy and subject to become stale. This is okay because // either choice will work correctly; we are simply applying a heuristic optimization. if (foreignShardedLookupAllowed() && pExpCtx->subPipelineDepth == 0 && pExpCtx->mongoProcessInterface->isSharded(_fromExpCtx->opCtx, _fromNs)) { return boost::none; } if (_fromExpCtx->ns.isConfigDotCacheDotChunks()) { // When $lookup reads from config.cache.chunks.* namespaces, it should run on each // individual shard in parallel. This is a special case, and atypical for standard $lookup // since a full copy of config.cache.chunks.* collections exists on all shards. return boost::none; } // {shardsStage, mergingStage, sortPattern} return DistributedPlanLogic{nullptr, this, boost::none}; } void DocumentSourceLookUp::detachFromOperationContext() { if (_pipeline) { // We have a pipeline we're going to be executing across multiple calls to getNext(), so we // use Pipeline::detachFromOperationContext() to take care of updating '_fromExpCtx->opCtx'. _pipeline->detachFromOperationContext(); invariant(_fromExpCtx->opCtx == nullptr); } else if (_fromExpCtx) { _fromExpCtx->opCtx = nullptr; } } void DocumentSourceLookUp::reattachToOperationContext(OperationContext* opCtx) { if (_pipeline) { // We have a pipeline we're going to be executing across multiple calls to getNext(), so we // use Pipeline::reattachToOperationContext() to take care of updating '_fromExpCtx->opCtx'. _pipeline->reattachToOperationContext(opCtx); invariant(_fromExpCtx->opCtx == opCtx); } else if (_fromExpCtx) { _fromExpCtx->opCtx = opCtx; } } bool DocumentSourceLookUp::validateOperationContext(const OperationContext* opCtx) const { if (getContext()->opCtx != opCtx || (_fromExpCtx && _fromExpCtx->opCtx != opCtx)) { return false; } if (_pipeline) { const auto& sources = _pipeline->getSources(); return std::all_of(sources.begin(), sources.end(), [opCtx](const auto& s) { return s->validateOperationContext(opCtx); }); } return true; } boost::intrusive_ptr DocumentSourceLookUp::createFromBson( BSONElement elem, const boost::intrusive_ptr& 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 pipeline; bool hasPipeline = false; bool hasLet = false; boost::optional> fromCollator; for (auto&& argument : elem.Obj()) { const auto argName = argument.fieldNameStringData(); if (argName == kPipelineField) { pipeline = parsePipelineFromBSON(argument); hasPipeline = true; continue; } if (argName == "let"_sd) { 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; } if (argName == kFromField) { fromNs = parseLookupFromAndResolveNamespace(argument, pExpCtx->ns.dbName()); continue; } if (argName == "_internalCollation"_sd) { const auto& collationSpec = argument.Obj(); if (!collationSpec.isEmpty()) { fromCollator.emplace(uassertStatusOK( CollatorFactoryInterface::get(pExpCtx->opCtx->getServiceContext()) ->makeFromBSON(collationSpec))); } continue; } uassert(ErrorCodes::FailedToParse, str::stream() << "$lookup argument '" << argName << "' must be a string, found " << argument << ": " << argument.type(), argument.type() == BSONType::String); if (argName == kAsField) { as = argument.String(); } else if (argName == kLocalField) { localField = argument.String(); } else if (argName == kForeignField) { foreignField = argument.String(); } else { uasserted(ErrorCodes::FailedToParse, str::stream() << "unknown argument to $lookup: " << argument.fieldName()); } } if (fromNs.ns().empty()) { validateLookupCollectionlessPipeline(pipeline); fromNs = NamespaceString::makeCollectionlessAggregateNSS(pExpCtx->ns.dbName()); } uassert(ErrorCodes::FailedToParse, "must specify 'as' field for a $lookup", !as.empty()); boost::intrusive_ptr lookupStage = nullptr; if (hasPipeline) { if (localField.empty() && foreignField.empty()) { // $lookup specified with only pipeline syntax. lookupStage = new DocumentSourceLookUp(std::move(fromNs), std::move(as), std::move(pipeline), std::move(letVariables), std::move(fromCollator), boost::none, pExpCtx); } else { // $lookup specified with pipeline syntax and local/foreignField syntax. uassert(ErrorCodes::FailedToParse, "$lookup requires both or neither of 'localField' and 'foreignField' to be " "specified", !localField.empty() && !foreignField.empty()); lookupStage = new DocumentSourceLookUp(std::move(fromNs), std::move(as), std::move(pipeline), std::move(letVariables), std::move(fromCollator), std::pair(std::move(localField), std::move(foreignField)), pExpCtx); } } else { // $lookup specified with only local/foreignField syntax. 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); lookupStage = new DocumentSourceLookUp(std::move(fromNs), std::move(as), std::move(localField), std::move(foreignField), std::move(fromCollator), pExpCtx); } lookupStage->determineSbeCompatibility(); return lookupStage; } void DocumentSourceLookUp::addInvolvedCollections( stdx::unordered_set* collectionNames) const { collectionNames->insert(_resolvedNs); for (auto&& stage : _resolvedIntrospectionPipeline->getSources()) { stage->addInvolvedCollections(collectionNames); } } } // namespace mongo