summaryrefslogtreecommitdiff
path: root/src/mongo/db/pipeline/document_source_lookup.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/mongo/db/pipeline/document_source_lookup.cpp')
-rw-r--r--src/mongo/db/pipeline/document_source_lookup.cpp176
1 files changed, 37 insertions, 139 deletions
diff --git a/src/mongo/db/pipeline/document_source_lookup.cpp b/src/mongo/db/pipeline/document_source_lookup.cpp
index ca9be5509bf..cfb885e94c5 100644
--- a/src/mongo/db/pipeline/document_source_lookup.cpp
+++ b/src/mongo/db/pipeline/document_source_lookup.cpp
@@ -35,12 +35,10 @@
#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_queue.h"
#include "mongo/db/pipeline/document_source_sort.h"
#include "mongo/db/pipeline/expression.h"
#include "mongo/db/pipeline/expression_context.h"
@@ -110,8 +108,7 @@ NamespaceString parseLookupFromAndResolveNamespace(const BSONElement& elem, Stri
str::stream() << "$lookup with syntax {from: {db:<>, coll:<>},..} is not supported for db: "
<< nss.db() << " and coll: " << nss.coll(),
nss.isConfigDotCacheDotChunks() || nss == NamespaceString::kRsOplogNamespace ||
- nss == NamespaceString::kTenantMigrationOplogView ||
- nss == NamespaceString::kConfigsvrCollectionsNamespace);
+ nss == NamespaceString::kTenantMigrationOplogView);
return nss;
}
@@ -180,20 +177,15 @@ DocumentSourceLookUp::DocumentSourceLookUp(
// 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);
-
- // Initialize the introspection pipeline before we insert the $match. This is okay because we do
- // not use the introspection pipeline during/after query execution, which is when the $match is
- // necessary.
- initializeResolvedIntrospectionPipeline();
-
_resolvedPipeline.push_back(BSON("$match" << BSONObj()));
_fieldMatchPipelineIdx = _resolvedPipeline.size() - 1;
+
+ initializeResolvedIntrospectionPipeline();
}
std::vector<BSONObj> extractSourceStage(const std::vector<BSONObj>& pipeline) {
if (!pipeline.empty() &&
(pipeline[0].hasField(DocumentSourceDocuments::kStageName) ||
- pipeline[0].hasField(DocumentSourceQueue::kStageName) ||
pipeline[0].hasField("$search"_sd))) {
return {pipeline[0]};
}
@@ -220,9 +212,8 @@ DocumentSourceLookUp::DocumentSourceLookUp(
// the local/foreignField $match. It must next after $documents if present.
auto sourceStages = extractSourceStage(pipeline);
_resolvedPipeline.insert(_resolvedPipeline.end(), sourceStages.begin(), sourceStages.end());
- // Save the correct position of the $match, but wait to insert it until we have finished
- // constructing the pipeline and created the introspection pipeline below.
- _fieldMatchPipelineIdx = _resolvedPipeline.size();
+ _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());
@@ -247,25 +238,15 @@ DocumentSourceLookUp::DocumentSourceLookUp(
_variablesParseState.defineVariable(varName));
}
- // Initialize the introspection pipeline before we insert the $match (if applicable). This is
- // okay because we only use the introspection pipeline for reference while doing query analysis
- // and analyzing involved dependencies/variables/collections/constraints. We do not use the
- // introspection pipeline during/after query execution, which is when the $match is necessary.
- // It wouldn't hurt anything to include the $match in this pipeline, but we also use the
- // introspection pipeline in serialization, so it would be a bit odd to include an extra empty
- // $match.
initializeResolvedIntrospectionPipeline();
-
- // Finally, insert the $match placeholder if we need it.
- if (_fieldMatchPipelineIdx) {
- _resolvedPipeline.insert(_resolvedPipeline.begin() + *_fieldMatchPipelineIdx,
- BSON("$match" << BSONObj()));
- }
}
DocumentSourceLookUp::DocumentSourceLookUp(const DocumentSourceLookUp& original,
const boost::intrusive_ptr<ExpressionContext>& newExpCtx)
- : DocumentSource(kStageName, newExpCtx),
+ : DocumentSource(
+ kStageName,
+ newExpCtx ? newExpCtx
+ : original.pExpCtx->copyWith(original.pExpCtx->ns, original.pExpCtx->uuid)),
_fromNs(original._fromNs),
_resolvedNs(original._resolvedNs),
_as(original._as),
@@ -285,10 +266,10 @@ DocumentSourceLookUp::DocumentSourceLookUp(const DocumentSourceLookUp& original,
_cache.emplace(internalDocumentSourceCursorBatchSizeBytes.load());
}
if (original._matchSrc) {
- _matchSrc = static_cast<DocumentSourceMatch*>(original._matchSrc->clone(pExpCtx).get());
+ _matchSrc = static_cast<DocumentSourceMatch*>(original._matchSrc->clone().get());
}
if (original._unwindSrc) {
- _unwindSrc = static_cast<DocumentSourceUnwind*>(original._unwindSrc->clone(pExpCtx).get());
+ _unwindSrc = static_cast<DocumentSourceUnwind*>(original._unwindSrc->clone().get());
}
}
@@ -418,11 +399,6 @@ StageConstraints DocumentSourceLookUp::constraints(Pipeline::SplitState pipeStat
// 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) {
- // 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
@@ -578,7 +554,6 @@ std::unique_ptr<Pipeline, PipelineDeleter> DocumentSourceLookUp::buildPipeline(
// If we don't have a cache, build and return the pipeline immediately.
if (!_cache || _cache->isAbandoned()) {
MakePipelineOptions pipelineOpts;
- pipelineOpts.alreadyOptimized = false;
pipelineOpts.optimize = true;
pipelineOpts.attachCursorSource = true;
pipelineOpts.validator = lookupPipeValidator;
@@ -613,7 +588,6 @@ std::unique_ptr<Pipeline, PipelineDeleter> DocumentSourceLookUp::buildPipeline(
// 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.alreadyOptimized = false;
pipelineOpts.optimize = false;
pipelineOpts.attachCursorSource = false;
pipelineOpts.validator = lookupPipeValidator;
@@ -669,59 +643,25 @@ std::unique_ptr<Pipeline, PipelineDeleter> DocumentSourceLookUp::buildPipeline(
return pipeline;
}
-/**
- * Method that looks for a DocumentSourceSequentialDocumentCache stage and calls optimizeAt() on
- * it if it has yet to be optimized.
- */
-void findAndOptimizeSequentialDocumentCache(Pipeline& pipeline) {
- auto& container = pipeline.getSources();
- auto itr = (&container)->begin();
- while (itr != (&container)->end()) {
- if (dynamic_cast<DocumentSourceSequentialDocumentCache*>(itr->get())) {
- auto sequentialCache = dynamic_cast<DocumentSourceSequentialDocumentCache*>(itr->get());
- if (!sequentialCache->hasOptimizedPos()) {
- sequentialCache->optimizeAt(itr, &container);
- }
- }
- itr = std::next(itr);
- }
-}
-
void DocumentSourceLookUp::addCacheStageAndOptimize(Pipeline& pipeline) {
- // Adds the cache to the end of the pipeline and calls optimizeContainer which will ensure the
- // stages of the pipeline are in the correct and optimal order, before the cache runs
- // doOptimizeAt. During the optimization process, the cache will either move itself to the
- // correct position in the pipeline, or abandon itself if no suitable cache position exists.
- // Once the cache is finished optimizing, the entire pipeline is optimized.
- //
- // When pipeline optimization is disabled, 'Pipeline::optimizePipeline()' exits early and so the
- // cache would not be placed correctly. So we only add the cache when pipeline optimization is
- // enabled.
+ // 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 {
- // The cache needs to see the full pipeline in its correct order in order to properly place
- // itself, therefore we are adding it to the end of the pipeline, and calling
- // optimizeContainer on the pipeline to ensure the rest of the pipeline is in its correct
- // order before optimizing the cache.
pipeline.addFinalSource(
DocumentSourceSequentialDocumentCache::create(_fromExpCtx, _cache.get_ptr()));
-
- auto& container = pipeline.getSources();
-
- Pipeline::optimizeContainer(&container);
-
- // We want to ensure the cache has been optimized prior to any calls to optimize().
- findAndOptimizeSequentialDocumentCache(pipeline);
-
- // Optimize the pipeline, with the cache in its correct position if it exists.
- Pipeline::optimizeEachStage(&container);
}
+
+ pipeline.optimizePipeline();
}
DocumentSource::GetModPathsReturn DocumentSourceLookUp::getModifiedPaths() const {
- OrderedPathSet modifiedPaths{_as.fullPath()};
+ std::set<std::string> modifiedPaths{_as.fullPath()};
if (_unwindSrc) {
auto pathsModifiedByUnwind = _unwindSrc->getModifiedPaths();
invariant(pathsModifiedByUnwind.type == GetModPathsReturn::Type::kFiniteSet);
@@ -1079,68 +1019,44 @@ void DocumentSourceLookUp::appendSpecificExecStats(MutableDocument& doc) const {
doc["indexesUsed"] = Value{std::move(indexesUsedVec)};
}
-void DocumentSourceLookUp::serializeToArray(std::vector<Value>& array,
- const SerializationOptions& opts) const {
+void DocumentSourceLookUp::serializeToArray(
+ std::vector<Value>& array, boost::optional<ExplainOptions::Verbosity> explain) const {
+
// Support alternative $lookup from config.cache.chunks* namespaces.
auto fromValue = (pExpCtx->ns.db() == _fromNs.db())
- ? Value(opts.serializeIdentifier(_fromNs.coll()))
- : Value(Document{{"db", opts.serializeIdentifier(_fromNs.db())},
- {"coll", opts.serializeIdentifier(_fromNs.coll())}});
+ ? Value(_fromNs.coll())
+ : Value(Document{{"db", _fromNs.db()}, {"coll", _fromNs.coll()}});
- MutableDocument output(Document{
- {getSourceName(), Document{{"from", fromValue}, {"as", opts.serializeFieldPath(_as)}}}});
+ MutableDocument output(
+ Document{{getSourceName(), Document{{"from", fromValue}, {"as", _as.fullPath()}}}});
if (hasLocalFieldForeignFieldJoin()) {
- output[getSourceName()]["localField"] = Value(opts.serializeFieldPath(_localField.value()));
- output[getSourceName()]["foreignField"] =
- Value(opts.serializeFieldPath(_foreignField.value()));
+ output[getSourceName()]["localField"] = Value(_localField->fullPath());
+ output[getSourceName()]["foreignField"] = Value(_foreignField->fullPath());
}
// 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<BSONObj> {
- if (!_userPipeline) {
- return std::vector<BSONObj>{};
- }
- if (opts.transformIdentifiers ||
- opts.literalPolicy != LiteralSerializationPolicy::kUnchanged) {
- return Pipeline::parse(*_userPipeline, _fromExpCtx)->serializeToBson(opts);
- }
- if (opts.serializeForQueryAnalysis) {
- // If we are in query analysis, encrypted fields will have been marked in the
- // introspection pipeline, so we need to serialize that here.
- return _resolvedIntrospectionPipeline->serializeToBson(opts);
- }
- return *_userPipeline;
- }();
+ auto pipeline = _userPipeline.get_value_or(std::vector<BSONObj>());
if (_additionalFilter) {
- auto serializedFilter = [&]() -> BSONObj {
- if (opts.transformIdentifiers ||
- opts.literalPolicy != LiteralSerializationPolicy::kUnchanged) {
- auto filter =
- uassertStatusOK(MatchExpressionParser::parse(*_additionalFilter, pExpCtx));
- return filter->serialize(opts);
- }
- return *_additionalFilter;
- }();
- serializedPipeline.emplace_back(BSON("$match" << serializedFilter));
+ pipeline.emplace_back(BSON("$match" << *_additionalFilter));
}
- if (!hasLocalFieldForeignFieldJoin() || serializedPipeline.size() > 0) {
+ if (!hasLocalFieldForeignFieldJoin() || pipeline.size() > 0) {
MutableDocument exprList;
for (auto letVar : _letVariables) {
- exprList.addField(opts.serializeFieldPathFromString(letVar.name),
- letVar.expression->serialize(opts));
+ exprList.addField(letVar.name,
+ letVar.expression->serialize(static_cast<bool>(explain)));
}
output[getSourceName()]["let"] = Value(exprList.freeze());
- output[getSourceName()]["pipeline"] = Value(serializedPipeline);
+ output[getSourceName()]["pipeline"] = Value(pipeline);
}
if (_hasExplicitCollation) {
output[getSourceName()]["_internalCollation"] = Value(_fromExpCtx->getCollatorBSON());
}
- if (opts.verbosity) {
+ if (explain) {
if (_unwindSrc) {
const boost::optional<FieldPath> indexPath = _unwindSrc->indexPath();
output[getSourceName()]["unwinding"] =
@@ -1148,11 +1064,8 @@ void DocumentSourceLookUp::serializeToArray(std::vector<Value>& array,
<< _unwindSrc->preserveNullAndEmptyArrays() << "includeArrayIndex"
<< (indexPath ? Value(indexPath->fullPath()) : Value())));
}
- // Conflict is .get() vs .value(), changed in 6.1 during
- // https://jira.mongodb.org/browse/SERVER-68246. Going to stick with 7.0's version of
- // '.value()', I think there is no meaningful difference here since line 1080 check that the
- // optional is set.
- if (opts.verbosity.value() >= ExplainOptions::Verbosity::kExecStats) {
+
+ if (explain.get() >= ExplainOptions::Verbosity::kExecStats) {
appendSpecificExecStats(output);
}
@@ -1255,21 +1168,6 @@ void DocumentSourceLookUp::reattachToOperationContext(OperationContext* 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<DocumentSource> DocumentSourceLookUp::createFromBson(
BSONElement elem, const boost::intrusive_ptr<ExpressionContext>& pExpCtx) {
uassert(ErrorCodes::FailedToParse,
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 13:30:04 +0000