diff options
Diffstat (limited to 'src/mongo/s/query/cluster_aggregate.cpp')
| -rw-r--r-- | src/mongo/s/query/cluster_aggregate.cpp | 172 |
1 files changed, 36 insertions, 136 deletions
diff --git a/src/mongo/s/query/cluster_aggregate.cpp b/src/mongo/s/query/cluster_aggregate.cpp index 6374bcfd494..b3df71582da 100644 --- a/src/mongo/s/query/cluster_aggregate.cpp +++ b/src/mongo/s/query/cluster_aggregate.cpp @@ -27,7 +27,6 @@ * it in the license file. */ -#include "mongo/s/chunk_manager.h" #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kCommand #include "mongo/platform/basic.h" @@ -38,7 +37,6 @@ #include "mongo/db/api_parameters.h" #include "mongo/db/auth/authorization_session.h" -#include "mongo/db/catalog/collection_uuid_mismatch_info.h" #include "mongo/db/client.h" #include "mongo/db/commands.h" #include "mongo/db/curop.h" @@ -57,9 +55,6 @@ #include "mongo/db/query/explain_common.h" #include "mongo/db/query/find_common.h" #include "mongo/db/query/fle/server_rewrite.h" -#include "mongo/db/query/query_stats/agg_key.h" -#include "mongo/db/query/query_stats/key.h" -#include "mongo/db/query/query_stats/query_stats.h" #include "mongo/db/timeseries/timeseries_options.h" #include "mongo/db/views/resolved_view.h" #include "mongo/db/views/view.h" @@ -99,7 +94,7 @@ namespace { // definition. It's okay that this is incorrect, we will repopulate the real namespace map on the // mongod. Note that this function must be called before forwarding an aggregation command on an // unsharded collection, in order to verify that the involved namespaces are allowed to be sharded. -auto resolveInvolvedNamespaces(const stdx::unordered_set<NamespaceString>& involvedNamespaces) { +auto resolveInvolvedNamespaces(stdx::unordered_set<NamespaceString> involvedNamespaces) { StringMap<ExpressionContext::ResolvedNamespace> resolvedNamespaces; for (auto&& nss : involvedNamespaces) { resolvedNamespaces.try_emplace(nss.coll(), nss, std::vector<BSONObj>{}); @@ -262,68 +257,6 @@ std::vector<BSONObj> rebuildPipelineWithTimeSeriesGranularity(const std::vector< return newPipeline; } -/** - * Builds an expCtx with which to parse the request's pipeline, then parses the pipeline and - * registers the pre-optimized pipeline with query stats collection. - */ -std::unique_ptr<Pipeline, PipelineDeleter> parsePipelineAndRegisterQueryStats( - OperationContext* opCtx, - const stdx::unordered_set<NamespaceString>& involvedNamespaces, - const NamespaceString& executionNss, - AggregateCommandRequest& request, - const boost::optional<ChunkManager>& cm, - const LiteParsedPipeline& liteParsedPipeline, - bool hasChangeStream, - bool shouldDoFLERewrite) { - // Populate the collection UUID and the appropriate collation to use. - auto [collationObj, uuid] = [&]() -> std::pair<BSONObj, boost::optional<UUID>> { - // If this is a change stream, take the user-defined collation if one exists, or an - // empty BSONObj otherwise. Change streams never inherit the collection's default - // collation, and since collectionless aggregations generally run on the 'admin' - // database, the standard logic would attempt to resolve its non-existent UUID and - // collation by sending a specious 'listCollections' command to the config servers. - if (hasChangeStream) { - return {request.getCollation().value_or(BSONObj()), boost::none}; - } - - return cluster_aggregation_planner::getCollationAndUUID( - opCtx, cm, executionNss, request.getCollation().value_or(BSONObj())); - }(); - - // Build an ExpressionContext for the pipeline. This instantiates an appropriate collator, - // resolves all involved namespaces, and creates a shared MongoProcessInterface for use by the - // pipeline's stages. - boost::intrusive_ptr<ExpressionContext> expCtx = - makeExpressionContext(opCtx, - request, - collationObj, - uuid, - resolveInvolvedNamespaces(involvedNamespaces), - hasChangeStream); - - // A pipeline with $changeStreamSplitLargeEvent requires the use of resume token format v2, - // since the 'fragmentNum' field only exists in this version and later. - if (hasChangeStream && liteParsedPipeline.endsWithChangeStreamSplitLargeEvent()) { - expCtx->changeStreamTokenVersion = 2; - } - - // Parse and optimize the full pipeline. - auto pipeline = Pipeline::parse(request.getPipeline(), expCtx); - - // Skip query stats recording for queryable encryption queries. - if (!shouldDoFLERewrite) { - query_stats::registerRequest( - opCtx, - executionNss, - [&]() { - return std::make_unique<query_stats::AggKey>( - request, *pipeline, expCtx, involvedNamespaces, executionNss); - }, - hasChangeStream); - } - return pipeline; -} - } // namespace Status ClusterAggregate::runAggregate(OperationContext* opCtx, @@ -376,7 +309,6 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx, auto hasChangeStream = liteParsedPipeline.hasChangeStream(); auto involvedNamespaces = liteParsedPipeline.getInvolvedNamespaces(); auto shouldDoFLERewrite = ::mongo::shouldDoFLERewrite(request); - auto startsWithDocuments = liteParsedPipeline.startsWithDocuments(); // If the routing table is not already taken by the higher level, fill it now. if (!cm) { @@ -391,14 +323,6 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx, sharded_agg_helpers::getExecutionNsRoutingInfo(opCtx, namespaces.executionNss); if (!executionNsRoutingInfoStatus.isOK()) { - uassert(CollectionUUIDMismatchInfo(request.getDbName().toString(), - *request.getCollectionUUID(), - request.getNamespace().coll().toString(), - boost::none), - "Database does not exist", - executionNsRoutingInfoStatus != ErrorCodes::NamespaceNotFound || - !request.getCollectionUUID()); - if (liteParsedPipeline.startsWithCollStats()) { uassertStatusOKWithContext(executionNsRoutingInfoStatus, "Unable to retrieve information for $collStats stage"); @@ -407,7 +331,7 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx, if (executionNsRoutingInfoStatus.isOK()) { cm = std::move(executionNsRoutingInfoStatus.getValue()); - } else if (!((hasChangeStream || startsWithDocuments) && + } else if (!(hasChangeStream && executionNsRoutingInfoStatus == ErrorCodes::NamespaceNotFound)) { appendEmptyResultSetWithStatus( opCtx, namespaces.requestedNss, executionNsRoutingInfoStatus.getStatus(), result); @@ -417,15 +341,33 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx, boost::intrusive_ptr<ExpressionContext> expCtx; const auto pipelineBuilder = [&]() { - auto pipeline = parsePipelineAndRegisterQueryStats(opCtx, - involvedNamespaces, - namespaces.executionNss, - request, - cm, - liteParsedPipeline, - hasChangeStream, - shouldDoFLERewrite); - expCtx = pipeline->getContext(); + // Populate the collection UUID and the appropriate collation to use. + auto [collationObj, uuid] = [&]() -> std::pair<BSONObj, boost::optional<UUID>> { + // If this is a change stream, take the user-defined collation if one exists, or an + // empty BSONObj otherwise. Change streams never inherit the collection's default + // collation, and since collectionless aggregations generally run on the 'admin' + // database, the standard logic would attempt to resolve its non-existent UUID and + // collation by sending a specious 'listCollections' command to the config servers. + if (hasChangeStream) { + return {request.getCollation().value_or(BSONObj()), boost::none}; + } + + return cluster_aggregation_planner::getCollationAndUUID( + opCtx, cm, namespaces.executionNss, request.getCollation().value_or(BSONObj())); + }(); + + // Build an ExpressionContext for the pipeline. This instantiates an appropriate collator, + // resolves all involved namespaces, and creates a shared MongoProcessInterface for use by + // the pipeline's stages. + expCtx = makeExpressionContext(opCtx, + request, + collationObj, + uuid, + resolveInvolvedNamespaces(involvedNamespaces), + hasChangeStream); + + // Parse and optimize the full pipeline. + auto pipeline = Pipeline::parse(request.getPipeline(), expCtx); // If the aggregate command supports encrypted collections, do rewrites of the pipeline to // support querying against encrypted fields. @@ -439,11 +381,6 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx, } pipeline->optimizePipeline(); - - // Validate the pipeline post-optimization. - const bool alreadyOptimized = true; - pipeline->validateCommon(alreadyOptimized); - return pipeline; }; @@ -458,7 +395,6 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx, cm, involvedNamespaces, hasChangeStream, - startsWithDocuments, allowedToPassthrough, request.getPassthroughToShard().has_value()); @@ -471,48 +407,15 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx, cluster_aggregation_planner::AggregationTargeter::TargetingPolicy::kMongosRequired); if (!expCtx) { - // When the AggregationTargeter chooses a "passthrough" or "specific shard only" policy, it - // does not call the 'pipelineBuilder' function, so we've yet to construct an expression - // context or register query stats. Because this is a passthrough, we only need a bare - // minimum expression context on mongos. + // When the AggregationTargeter chooses a "passthrough" policy, it does not call the + // 'pipelineBuilder' function, so we never get an expression context. Because this is a + // passthrough, we only need a bare minimum expression context anyway. invariant(targeter.policy == cluster_aggregation_planner::AggregationTargeter::kPassthrough || targeter.policy == cluster_aggregation_planner::AggregationTargeter::kSpecificShardOnly); - expCtx = make_intrusive<ExpressionContext>( opCtx, nullptr, namespaces.executionNss, boost::none, request.getLet()); - expCtx->addResolvedNamespaces(involvedNamespaces); - - - // We might need 'inMongos' temporarily set to true for query stats parsing, but we don't - // want to modify the value of 'expCtx' for future code execution so we will set it back to - // its original value. - ON_BLOCK_EXIT([&expCtx, originalInMongosVal = expCtx->inMongos]() { - expCtx->inMongos = originalInMongosVal; - }); - - // In order to parse a change stream request for query stats, 'inMongos' needs - // to be set to true. - if (hasChangeStream) { - expCtx->inMongos = true; - } - - // Skip query stats recording for queryable encryption queries. - if (!shouldDoFLERewrite) { - // We want to hold off parsing the pipeline until it's clear we must. Because of that, - // we wait to parse the pipeline until this callback is invoked within - // query_stats::registerRequest. - query_stats::registerRequest( - opCtx, - namespaces.executionNss, - [&]() { - auto pipeline = Pipeline::parse(request.getPipeline(), expCtx); - return std::make_unique<query_stats::AggKey>( - request, *pipeline, expCtx, involvedNamespaces, namespaces.executionNss); - }, - hasChangeStream); - } } if (request.getExplain()) { @@ -540,11 +443,10 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx, // If this is an explain write the explain output and return. auto expCtx = targeter.pipeline->getContext(); if (expCtx->explain) { - auto opts = SerializationOptions{}; - opts.verbosity = boost::make_optional(ExplainOptions::Verbosity::kQueryPlanner); *result << "splitPipeline" << BSONNULL << "mongos" << Document{{"host", getHostNameCachedAndPort()}, - {"stages", targeter.pipeline->writeExplainOps(opts)}}; + {"stages", + targeter.pipeline->writeExplainOps(*expCtx->explain)}}; return Status::OK(); } @@ -567,8 +469,7 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx, namespaces, privileges, result, - hasChangeStream, - startsWithDocuments); + hasChangeStream); } case cluster_aggregation_planner::AggregationTargeter::TargetingPolicy:: kSpecificShardOnly: { @@ -613,12 +514,11 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx, updateHostsTargetedMetrics(opCtx, namespaces.executionNss, cm, involvedNamespaces); // Report usage statistics for each stage in the pipeline. liteParsedPipeline.tickGlobalStageCounters(); + // Add 'command' object to explain output. if (expCtx->explain) { explain_common::appendIfRoom( aggregation_request_helper::serializeToCommandObj(request), "command", result); - collectQueryStatsMongos(opCtx, - std::move(CurOp::get(opCtx)->debug().queryStatsInfo.key)); } } return status; |