summaryrefslogtreecommitdiff
path: root/src/mongo/db/pipeline/pipeline.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/mongo/db/pipeline/pipeline.cpp')
-rw-r--r--src/mongo/db/pipeline/pipeline.cpp164
1 files changed, 38 insertions, 126 deletions
diff --git a/src/mongo/db/pipeline/pipeline.cpp b/src/mongo/db/pipeline/pipeline.cpp
index 30dbcba6290..97a896e5898 100644
--- a/src/mongo/db/pipeline/pipeline.cpp
+++ b/src/mongo/db/pipeline/pipeline.cpp
@@ -40,7 +40,6 @@
#include "mongo/db/jsobj.h"
#include "mongo/db/operation_context.h"
#include "mongo/db/pipeline/accumulator.h"
-#include "mongo/db/pipeline/change_stream_helpers.h"
#include "mongo/db/pipeline/document_source.h"
#include "mongo/db/pipeline/document_source_match.h"
#include "mongo/db/pipeline/document_source_merge.h"
@@ -106,32 +105,14 @@ void validateTopLevelPipeline(const Pipeline& pipeline) {
// If the first stage is a $changeStream stage, then all stages in the pipeline must be
// either $changeStream stages or allowlisted as being able to run in a change stream.
- const bool isChangeStream = firstStageConstraints.isChangeStreamStage();
- // Record whether any of the stages in the pipeline is a $changeStreamSplitLargeEvent.
- bool hasChangeStreamSplitLargeEventStage = false;
- for (auto&& source : sources) {
- uassert(ErrorCodes::IllegalOperation,
- str::stream() << source->getSourceName()
- << " is not permitted in a $changeStream pipeline",
- !(isChangeStream && !source->constraints().isAllowedInChangeStream()));
- // Check whether any stages must only be run in a change stream pipeline.
- uassert(ErrorCodes::IllegalOperation,
- str::stream() << source->getSourceName()
- << " can only be used in a $changeStream pipeline",
- !(source->constraints().requiresChangeStream() && !isChangeStream));
- // Check whether this is a change stream split stage.
- if ("$changeStreamSplitLargeEvent"_sd == source->getSourceName()) {
- hasChangeStreamSplitLargeEventStage = true;
+ if (firstStageConstraints.isChangeStreamStage()) {
+ for (auto&& source : sources) {
+ uassert(ErrorCodes::IllegalOperation,
+ str::stream() << source->getSourceName()
+ << " is not permitted in a $changeStream pipeline",
+ source->constraints().isAllowedInChangeStream());
}
}
- auto expCtx = pipeline.getContext();
- auto spec = isChangeStream ? expCtx->changeStreamSpec : boost::none;
- auto hasSplitEventResumeToken = spec &&
- change_stream::resolveResumeTokenFromSpec(expCtx, *spec).fragmentNum.has_value();
- uassert(ErrorCodes::ChangeStreamFatalError,
- "To resume from a split event, the $changeStream pipeline must include a "
- "$changeStreamSplitLargeEvent stage",
- !(hasSplitEventResumeToken && !hasChangeStreamSplitLargeEventStage));
}
// Verify that usage of $searchMeta and $search is legal.
@@ -174,12 +155,11 @@ Pipeline::~Pipeline() {
std::unique_ptr<Pipeline, PipelineDeleter> Pipeline::clone(
const boost::intrusive_ptr<ExpressionContext>& newExpCtx) const {
- auto expCtx = newExpCtx ? newExpCtx : getContext();
SourceContainer clonedStages;
for (auto&& stage : _sources) {
- clonedStages.push_back(stage->clone(expCtx));
+ clonedStages.push_back(stage->clone(newExpCtx));
}
- return create(clonedStages, expCtx);
+ return create(clonedStages, newExpCtx ? newExpCtx : getContext());
}
template <class T>
@@ -251,43 +231,39 @@ std::unique_ptr<Pipeline, PipelineDeleter> Pipeline::create(
}
void Pipeline::validateCommon(bool alreadyOptimized) const {
+ size_t i = 0;
+
uassert(ErrorCodes::FailedToParse,
str::stream() << "Pipeline length must be no longer than "
<< internalPipelineLengthLimit << " stages",
static_cast<int>(_sources.size()) <= internalPipelineLengthLimit);
- checkValidOperationContext();
-
- // Keep track of stages which can only appear once.
- std::set<StringData> singleUseStages;
-
- for (auto sourceIter = _sources.begin(); sourceIter != _sources.end(); ++sourceIter) {
- auto& stage = *sourceIter;
+ for (auto&& stage : _sources) {
auto constraints = stage->constraints(_splitState);
// Verify that all stages adhere to their PositionRequirement constraints.
uassert(40602,
str::stream() << stage->getSourceName()
<< " is only valid as the first stage in a pipeline",
- !(constraints.requiredPosition == PositionRequirement::kFirst &&
- sourceIter != _sources.begin()));
+ !(constraints.requiredPosition == PositionRequirement::kFirst && i != 0));
+ uassert(40603,
+ str::stream() << stage->getSourceName()
+ << " is only valid as the first stage in an optimized pipeline",
+ !(alreadyOptimized &&
+ constraints.requiredPosition == PositionRequirement::kFirstAfterOptimization &&
+ i != 0));
- // TODO SERVER-73790: use PositionRequirement::kCustom to validate $match.
auto matchStage = dynamic_cast<DocumentSourceMatch*>(stage.get());
uassert(17313,
"$match with $text is only allowed as the first pipeline stage",
- !(sourceIter != _sources.begin() && matchStage && matchStage->isTextQuery()));
+ !(i != 0 && matchStage && matchStage->isTextQuery()));
uassert(40601,
str::stream() << stage->getSourceName()
<< " can only be the final stage in the pipeline",
!(constraints.requiredPosition == PositionRequirement::kLast &&
- std::next(sourceIter) != _sources.end()));
-
- // If the stage has a special requirement about its position, validate it.
- if (constraints.requiredPosition == PositionRequirement::kCustom) {
- stage->validatePipelinePosition(alreadyOptimized, sourceIter, _sources);
- }
+ i != _sources.size() - 1));
+ ++i;
// Verify that we are not attempting to run a mongoS-only stage on mongoD.
uassert(40644,
@@ -299,17 +275,6 @@ void Pipeline::validateCommon(bool alreadyOptimized) const {
str::stream() << "Stage not supported inside of a multi-document transaction: "
<< stage->getSourceName(),
!(pCtx->opCtx->inMultiDocumentTransaction() && !constraints.isAllowedInTransaction()));
-
- // Verify that a stage which can only appear once doesn't appear more than that.
- uassert(7183900,
- str::stream() << stage->getSourceName() << " can only be used once in the pipeline",
- !(constraints.canAppearOnlyOnceInPipeline &&
- !singleUseStages.insert(stage->getSourceName()).second));
-
- tassert(7355707,
- "If a stage is broadcast to all shard servers then it must be a data source.",
- constraints.hostRequirement != HostTypeRequirement::kAllShardServers ||
- !constraints.requiresInputDocSource);
}
}
@@ -318,29 +283,21 @@ void Pipeline::optimizePipeline() {
if (MONGO_unlikely(disablePipelineOptimization.shouldFail())) {
return;
}
+
optimizeContainer(&_sources);
- optimizeEachStage(&_sources);
}
void Pipeline::optimizeContainer(SourceContainer* container) {
+ SourceContainer optimizedSources;
+
SourceContainer::iterator itr = container->begin();
try {
while (itr != container->end()) {
invariant((*itr).get());
itr = (*itr).get()->optimizeAt(itr, container);
}
- } catch (DBException& ex) {
- ex.addContext("Failed to optimize pipeline");
- throw;
- }
- stitch(container);
-}
-
-void Pipeline::optimizeEachStage(SourceContainer* container) {
- SourceContainer optimizedSources;
- try {
- // We should have our final number of stages. Optimize each individually.
+ // Once we have reached our final number of stages, optimize each individually.
for (auto&& source : *container) {
if (auto out = source->optimize()) {
optimizedSources.push_back(out);
@@ -381,9 +338,6 @@ void Pipeline::detachFromOperationContext() {
for (auto&& source : _sources) {
source->detachFromOperationContext();
}
-
- // Check for a null operation context to make sure that all children detached correctly.
- checkValidOperationContext();
}
void Pipeline::reattachToOperationContext(OperationContext* opCtx) {
@@ -392,24 +346,6 @@ void Pipeline::reattachToOperationContext(OperationContext* opCtx) {
for (auto&& source : _sources) {
source->reattachToOperationContext(opCtx);
}
-
- checkValidOperationContext();
-}
-
-bool Pipeline::validateOperationContext(const OperationContext* opCtx) const {
- return std::all_of(_sources.begin(), _sources.end(), [this, opCtx](const auto& s) {
- // All sources in a pipeline must share its expression context. Subpipelines may have a
- // different expression context, but must point to the same operation context. Let the
- // sources validate this themselves since they don't all have the same subpipelines, etc.
- return s->getContext() == getContext() && s->validateOperationContext(opCtx);
- });
-}
-
-void Pipeline::checkValidOperationContext() const {
- tassert(7406000,
- str::stream()
- << "All DocumentSources and subpipelines must have the same operation context",
- validateOperationContext(getContext()->opCtx));
}
void Pipeline::dispose(OperationContext* opCtx) {
@@ -461,19 +397,11 @@ bool Pipeline::needsMongosMerger() const {
});
}
-bool Pipeline::needsAllShardServers() const {
- return std::any_of(_sources.begin(), _sources.end(), [&](const auto& stage) {
- return stage->constraints().resolvedHostTypeRequirement(pCtx) ==
- HostTypeRequirement::kAllShardServers;
- });
-}
-
bool Pipeline::needsShard() const {
return std::any_of(_sources.begin(), _sources.end(), [&](const auto& stage) {
auto hostType = stage->constraints().resolvedHostTypeRequirement(pCtx);
return (hostType == HostTypeRequirement::kAnyShard ||
- hostType == HostTypeRequirement::kPrimaryShard ||
- hostType == HostTypeRequirement::kAllShardServers);
+ hostType == HostTypeRequirement::kPrimaryShard);
});
}
@@ -519,20 +447,20 @@ stdx::unordered_set<NamespaceString> Pipeline::getInvolvedCollections() const {
}
vector<Value> Pipeline::serializeContainer(const SourceContainer& container,
- boost::optional<const SerializationOptions&> opts) {
+ boost::optional<ExplainOptions::Verbosity> explain) {
vector<Value> serializedSources;
for (auto&& source : container) {
- source->serializeToArray(serializedSources, opts ? opts.get() : SerializationOptions());
+ source->serializeToArray(serializedSources, explain);
}
return serializedSources;
}
-
-vector<Value> Pipeline::serialize(boost::optional<const SerializationOptions&> opts) const {
- return serializeContainer(_sources, opts);
+vector<Value> Pipeline::serialize(boost::optional<ExplainOptions::Verbosity> explain) const {
+ return serializeContainer(_sources, explain);
}
-vector<BSONObj> Pipeline::serializeToBson(boost::optional<const SerializationOptions&> opts) const {
- const auto serialized = serialize(opts);
+vector<BSONObj> Pipeline::serializeToBson(
+ boost::optional<ExplainOptions::Verbosity> explain) const {
+ const auto serialized = serialize(explain);
std::vector<BSONObj> asBson;
asBson.reserve(serialized.size());
for (auto&& stage : serialized) {
@@ -573,16 +501,16 @@ boost::optional<Document> Pipeline::getNext() {
: boost::optional<Document>{nextResult.releaseDocument()};
}
-vector<Value> Pipeline::writeExplainOps(const SerializationOptions& opts) const {
+vector<Value> Pipeline::writeExplainOps(ExplainOptions::Verbosity verbosity) const {
vector<Value> array;
for (auto&& stage : _sources) {
auto beforeSize = array.size();
- stage->serializeToArray(array, opts);
+ stage->serializeToArray(array, verbosity);
auto afterSize = array.size();
// Append execution stats to the serialized stage if the specified verbosity is
// 'executionStats' or 'allPlansExecution'.
invariant(afterSize - beforeSize == 1u);
- if (*opts.verbosity >= ExplainOptions::Verbosity::kExecStats) {
+ if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
auto serializedStage = array.back();
array.back() = appendCommonExecStats(serializedStage, stage->getCommonStats());
}
@@ -673,8 +601,7 @@ Status Pipeline::_pipelineCanRunOnMongoS() const {
auto hostRequirement = constraints.resolvedHostTypeRequirement(pCtx);
const bool needsShard = (hostRequirement == HostTypeRequirement::kAnyShard ||
- hostRequirement == HostTypeRequirement::kPrimaryShard ||
- hostRequirement == HostTypeRequirement::kAllShardServers);
+ hostRequirement == HostTypeRequirement::kPrimaryShard);
const bool mustWriteToDisk =
(constraints.diskRequirement == DiskUseRequirement::kWritesPersistentData);
@@ -761,31 +688,17 @@ boost::intrusive_ptr<DocumentSource> Pipeline::popFrontWithNameAndCriteria(
return popFront();
}
-void Pipeline::appendPipeline(std::unique_ptr<Pipeline, PipelineDeleter> otherPipeline) {
- auto& otherPipelineSources = otherPipeline->getSources();
- while (!otherPipelineSources.empty()) {
- _sources.push_back(std::move(otherPipelineSources.front()));
- otherPipelineSources.pop_front();
- }
- constexpr bool alreadyOptimized = false;
- validateCommon(alreadyOptimized);
- stitch();
-}
-
-
std::unique_ptr<Pipeline, PipelineDeleter> Pipeline::makePipeline(
const std::vector<BSONObj>& rawPipeline,
const boost::intrusive_ptr<ExpressionContext>& expCtx,
const MakePipelineOptions opts) {
auto pipeline = Pipeline::parse(rawPipeline, expCtx, opts.validator);
- bool alreadyOptimized = opts.alreadyOptimized;
-
if (opts.optimize) {
pipeline->optimizePipeline();
- alreadyOptimized = true;
}
+ constexpr bool alreadyOptimized = true;
pipeline->validateCommon(alreadyOptimized);
if (opts.attachCursorSource) {
@@ -802,7 +715,6 @@ Pipeline::SourceContainer::iterator Pipeline::optimizeEndOfPipeline(
// optimize, since otherwise calls to optimizeAt() will overrun these limits.
auto endOfPipeline = Pipeline::SourceContainer(std::next(itr), container->end());
Pipeline::optimizeContainer(&endOfPipeline);
- Pipeline::optimizeEachStage(&endOfPipeline);
container->erase(std::next(itr), container->end());
container->splice(std::next(itr), endOfPipeline);
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 11:48:09 +0000