diff options
Diffstat (limited to 'src/mongo/db/commands/run_aggregate.cpp')
| -rw-r--r-- | src/mongo/db/commands/run_aggregate.cpp | 383 |
1 files changed, 116 insertions, 267 deletions
diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp index 66d68f4413c..9172103b493 100644 --- a/src/mongo/db/commands/run_aggregate.cpp +++ b/src/mongo/db/commands/run_aggregate.cpp @@ -74,9 +74,6 @@ #include "mongo/db/query/query_feature_flags_gen.h" #include "mongo/db/query/query_knobs_gen.h" #include "mongo/db/query/query_planner_common.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/read_concern.h" #include "mongo/db/repl/oplog.h" #include "mongo/db/repl/read_concern_args.h" @@ -107,8 +104,6 @@ namespace { ServerStatusMetricField<Counter64> allowDiskUseMetric{"query.allowDiskUseFalse", &allowDiskUseFalseCounter}; -MONGO_FAIL_POINT_DEFINE(hangAfterCreatingAggregationPlan); - /** * If a pipeline is empty (assuming that a $cursor stage hasn't been created yet), it could mean * that we were able to absorb all pipeline stages and pull them into a single PlanExecutor. So, @@ -226,6 +221,7 @@ bool handleCursorCommand(OperationContext* opCtx, auto&& [stats, _] = explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats); LOGV2_WARNING(23799, + "Aggregate command executor error: {error}, stats: {stats}, cmd: {cmd}", "Aggregate command executor error", "error"_attr = exception.toStatus(), "stats"_attr = redact(stats), @@ -630,6 +626,7 @@ std::vector<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> createLegacyEx getSearchHelpers(expCtx->opCtx->getServiceContext()) ->injectSearchShardFiltererIfNeeded(pipeline.get()); + // Complete creation of the initial $cursor stage, if needed. PipelineD::attachInnerQueryExecutorToPipeline(collections, attachExecutorCallback.first, @@ -642,6 +639,7 @@ std::vector<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> createLegacyEx // There are separate ExpressionContexts for each exchange pipeline, so make sure to // pass the pipeline's ExpressionContext to the plan executor factory. auto pipelineExpCtx = pipelineIt->getContext(); + execs.emplace_back( plan_executor_factory::make(std::move(pipelineExpCtx), std::move(pipelineIt), @@ -659,222 +657,25 @@ std::vector<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> createLegacyEx return execs; } -Status runAggregateOnView(OperationContext* opCtx, - const NamespaceString& origNss, - const AggregateCommandRequest& request, - const MultipleCollectionAccessor& collections, - boost::optional<std::unique_ptr<CollatorInterface>> collatorToUse, - const ViewDefinition* view, - std::shared_ptr<const CollectionCatalog> catalog, - const PrivilegeVector& privileges, - rpc::ReplyBuilderInterface* result, - const std::function<void(void)>& resetContextFn) { - auto nss = request.getNamespace(); - - uassert(ErrorCodes::CommandNotSupportedOnView, - "mapReduce on a view is not supported", - !request.getIsMapReduceCommand()); - - // Check that the default collation of 'view' is compatible with the operation's - // collation. The check is skipped if the request did not specify a collation. - if (!request.getCollation().get_value_or(BSONObj()).isEmpty()) { - invariant(collatorToUse); // Should already be resolved at this point. - if (!CollatorInterface::collatorsMatch(view->defaultCollator(), collatorToUse->get()) && - !view->timeseries()) { - - return {ErrorCodes::OptionNotSupportedOnView, - "Cannot override a view's default collation"}; - } - } - - // Queries on timeseries views may specify non-default collation whereas queries - // on all other types of views must match the default collator (the collation use - // to originally create that collections). Thus in the case of operations on TS - // views, we use the request's collation. - auto timeSeriesCollator = view->timeseries() ? request.getCollation() : boost::none; - - auto resolvedView = - uassertStatusOK(view_catalog_helpers::resolveView(opCtx, catalog, nss, timeSeriesCollator)); - - // With the view & collation resolved, we can relinquish locks. - resetContextFn(); - - // Set this operation's shard version for the underlying collection to unsharded. - // This is prerequisite for future shard versioning checks. - ScopedSetShardRole scopedSetShardRole(opCtx, - resolvedView.getNamespace(), - ChunkVersion::UNSHARDED() /* shardVersion */, - boost::none /* databaseVersion */); - - uassert(std::move(resolvedView), - "Explain of a resolved view must be executed by mongos", - !ShardingState::get(opCtx)->enabled() || !request.getExplain()); - - // Parse the resolved view into a new aggregation request. - auto newRequest = resolvedView.asExpandedViewAggregation(request); - auto newCmd = aggregation_request_helper::serializeToCommandObj(newRequest); - - auto status{Status::OK()}; - try { - status = runAggregate(opCtx, newRequest, newCmd, privileges, result, resolvedView, request); - } catch (const ExceptionForCat<ErrorCategory::StaleShardVersionError>& ex) { - // Since we expect the view to be UNSHARDED, if we reached to this point there are - // two possibilities: - // 1. The shard doesn't know what its shard version/state is and needs to recover - // it (in which case we throw so that the shard can run recovery) - // 2. The collection references by the view is actually SHARDED, in which case the - // router must execute it - if (const auto staleInfo{ex.extraInfo<StaleConfigInfo>()}) { - uassert(std::move(resolvedView), - "Resolved views on sharded collections must be executed by mongos", - !staleInfo->getVersionWanted()); - } - throw; - } - - { - // Set the namespace of the curop back to the view namespace so ctx records - // stats on this view namespace on destruction. - stdx::lock_guard<Client> lk(*opCtx->getClient()); - CurOp::get(opCtx)->setNS_inlock(nss.ns()); - } - - return status; -} - -/** - * Determines the collection type of the query by precedence of various configurations. The order - * of these checks is critical since there may be overlap (e.g., a view over a virtual collection - * is classified as a view). - */ -query_shape::CollectionType determineCollectionType( - const boost::optional<AutoGetCollectionForReadCommandMaybeLockFree>& ctx, - boost::optional<const ResolvedView&> resolvedView, - bool hasChangeStream, - bool isCollectionless) { - if (resolvedView.has_value()) { - if (resolvedView->timeseries()) { - return query_shape::CollectionType::kTimeseries; - } - return query_shape::CollectionType::kView; - } - if (isCollectionless) { - return query_shape::CollectionType::kVirtual; - } - if (hasChangeStream) { - return query_shape::CollectionType::kChangeStream; - } - return ctx ? ctx->getCollectionType() : query_shape::CollectionType::kUnknown; -} - -std::unique_ptr<Pipeline, PipelineDeleter> parsePipelineAndRegisterQueryStats( - OperationContext* opCtx, - const NamespaceString& origNss, - const AggregateCommandRequest& request, - const boost::optional<AutoGetCollectionForReadCommandMaybeLockFree>& ctx, - std::unique_ptr<CollatorInterface> collator, - boost::optional<UUID> uuid, - ExpressionContext::CollationMatchesDefault collationMatchesDefault, - const MultipleCollectionAccessor& collections, - stdx::unordered_set<NamespaceString> pipelineInvolvedNamespaces, - const LiteParsedPipeline& liteParsedPipeline, - bool isCollectionless, - boost::optional<const ResolvedView&> resolvedView, - boost::optional<const AggregateCommandRequest&> origRequest) { - // If we're operating over a view, we first parse just the original user-given request - // for the sake of registering query stats. Then, we'll parse the view pipeline and stitch - // the two pipelines together below. - auto expCtx = - makeExpressionContext(opCtx, request, std::move(collator), uuid, collationMatchesDefault); - // If any involved collection contains extended-range data, set a flag which individual - // DocumentSource parsers can check. - collections.forEach([&](const CollectionPtr& coll) { - if (coll->getRequiresTimeseriesExtendedRangeSupport()) - expCtx->setRequiresTimeseriesExtendedRangeSupport(true); - }); - - const bool hasChangeStream = liteParsedPipeline.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; - } - - auto requestForQueryStats = origRequest.has_value() ? *origRequest : request; - expCtx->startExpressionCounters(); - auto pipeline = Pipeline::parse(requestForQueryStats.getPipeline(), expCtx); - expCtx->stopExpressionCounters(); - - // Register query stats with the pre-optimized pipeline. Exclude queries against collections - // with encrypted fields. We still collect query stats on collection-less aggregations. - bool hasEncryptedFields = ctx && ctx->getCollection() && - ctx->getCollection()->getCollectionOptions().encryptedFieldConfig; - if (!hasEncryptedFields) { - // If this is a query over a resolved view, we want to register query stats with the - // original user-given request and pipeline, rather than the new request generated when - // resolving the view. - auto collectionType = - determineCollectionType(ctx, resolvedView, hasChangeStream, isCollectionless); - - query_stats::registerRequest(opCtx, - origNss, - [&]() { - return std::make_unique<query_stats::AggKey>( - requestForQueryStats, - *pipeline, - expCtx, - pipelineInvolvedNamespaces, - origNss, - collectionType); - }, - hasChangeStream); - } - - if (resolvedView.has_value()) { - expCtx->startExpressionCounters(); - - if (resolvedView->timeseries()) { - // For timeseries, there may have been rewrites done on the raw BSON pipeline - // during view resolution. We must parse the request's full resolved pipeline - // which will account for those rewrites. - // TODO SERVER-82101 Re-organize timeseries rewrites so timeseries can follow the - // same pattern here as other views - pipeline = Pipeline::parse(request.getPipeline(), expCtx); - } else { - // Parse the view pipeline, then stitch the user pipeline and view pipeline together - // to build the total aggregation pipeline. - auto userPipeline = std::move(pipeline); - pipeline = Pipeline::parse(resolvedView->getPipeline(), expCtx); - pipeline->appendPipeline(std::move(userPipeline)); - } - - expCtx->stopExpressionCounters(); - } - - return pipeline; -} } // namespace Status runAggregate(OperationContext* opCtx, + const NamespaceString& nss, AggregateCommandRequest& request, const BSONObj& cmdObj, const PrivilegeVector& privileges, - rpc::ReplyBuilderInterface* result, - boost::optional<const ResolvedView&> resolvedView, - boost::optional<const AggregateCommandRequest&> origRequest) { - return runAggregate( - opCtx, request, {request}, cmdObj, privileges, result, resolvedView, origRequest); + rpc::ReplyBuilderInterface* result) { + return runAggregate(opCtx, nss, request, {request}, cmdObj, privileges, result); } Status runAggregate(OperationContext* opCtx, + const NamespaceString& origNss, AggregateCommandRequest& request, const LiteParsedPipeline& liteParsedPipeline, const BSONObj& cmdObj, const PrivilegeVector& privileges, - rpc::ReplyBuilderInterface* result, - boost::optional<const ResolvedView&> resolvedView, - boost::optional<const AggregateCommandRequest&> origRequest) { - auto origNss = origRequest.has_value() ? origRequest->getNamespace() : request.getNamespace(); + rpc::ReplyBuilderInterface* result) { + // Perform some validations on the LiteParsedPipeline and request before continuing with the // aggregation command. performValidationChecks(opCtx, request, liteParsedPipeline); @@ -944,8 +745,7 @@ Status runAggregate(OperationContext* opCtx, boost::optional<AutoStatsTracker> statsTracker; // If this is a change stream, perform special checks and change the execution namespace. - const auto hasChangeStream = liteParsedPipeline.hasChangeStream(); - if (hasChangeStream) { + if (liteParsedPipeline.hasChangeStream()) { uassert(4928900, str::stream() << AggregateCommandRequest::kCollectionUUIDFieldName << " is not supported for a change stream", @@ -959,6 +759,7 @@ Status runAggregate(OperationContext* opCtx, // Raise an error if 'origNss' is a view. We do not need to check this if we are opening // a stream on an entire db or across the cluster. + const TenantDatabaseName origTenantDbName(boost::none, origNss.db()); if (!origNss.isCollectionlessAggregateNS()) { auto view = catalog->lookupView(opCtx, origNss); uassert(ErrorCodes::CommandNotSupportedOnView, @@ -992,7 +793,7 @@ Status runAggregate(OperationContext* opCtx, nss, Top::LockType::NotLocked, AutoStatsTracker::LogMode::kUpdateTopAndCurOp, - catalog->getDatabaseProfileLevel(nss.db())); + 0); auto [collator, match] = PipelineD::resolveCollator( opCtx, request.getCollation().get_value_or(BSONObj()), nullptr); collatorToUse.emplace(std::move(collator)); @@ -1014,53 +815,108 @@ Status runAggregate(OperationContext* opCtx, } } - // If collectionUUID was provided, verify the collection exists and has the expected UUID. - checkCollectionUUIDMismatch(opCtx, - nss, - collections.getMainCollection(), - request.getCollectionUUID(), - false /* checkFeatureFlag */); - // If this is a view, resolve it by finding the underlying collection and stitching view // pipelines and this request's pipeline together. We then release our locks before // recursively calling runAggregate(), which will re-acquire locks on the underlying // collection. (The lock must be released because recursively acquiring locks on the // database will prohibit yielding.) - // We do not need to expand the view pipeline when there is a $collStats stage, as - // $collStats is supported on a view namespace. For a time-series collection, however, the - // view is abstracted out for the users, so we needed to resolve the namespace to get the - // underlying bucket collection. - if (ctx && ctx->getView() && - (!liteParsedPipeline.startsWithCollStats() || ctx->getView()->timeseries())) { - return runAggregateOnView(opCtx, - origNss, - request, - collections, - std::move(collatorToUse), - ctx->getView(), - catalog, - privileges, - result, - resetContext); + if (ctx && ctx->getView() && !liteParsedPipeline.startsWithCollStats()) { + invariant(nss != NamespaceString::kRsOplogNamespace); + invariant(!nss.isCollectionlessAggregateNS()); + + checkCollectionUUIDMismatch(opCtx, + nss, + collections.getMainCollection(), + request.getCollectionUUID(), + false /* checkFeatureFlag */); + + uassert(ErrorCodes::CommandNotSupportedOnView, + "mapReduce on a view is not supported", + !request.getIsMapReduceCommand()); + + // Check that the default collation of 'view' is compatible with the operation's + // collation. The check is skipped if the request did not specify a collation. + if (!request.getCollation().get_value_or(BSONObj()).isEmpty()) { + invariant(collatorToUse); // Should already be resolved at this point. + if (!CollatorInterface::collatorsMatch(ctx->getView()->defaultCollator(), + collatorToUse->get()) && + !ctx->getView()->timeseries()) { + + return {ErrorCodes::OptionNotSupportedOnView, + "Cannot override a view's default collation"}; + } + } + + // Queries on timeseries views may specify non-default collation whereas queries + // on all other types of views must match the default collator (the collation use + // to originally create that collections). Thus in the case of operations on TS + // views, we use the request's collation. + auto timeSeriesCollator = + ctx->getView()->timeseries() ? request.getCollation() : boost::none; + + auto resolvedView = uassertStatusOK( + view_catalog_helpers::resolveView(opCtx, catalog, nss, timeSeriesCollator)); + + // With the view & collation resolved, we can relinquish locks. + resetContext(); + + // Set this operation's shard version for the underlying collection to unsharded. + // This is prerequisite for future shard versioning checks. + ScopedSetShardRole scopedSetShardRole(opCtx, + resolvedView.getNamespace(), + ChunkVersion::UNSHARDED() /* shardVersion */, + boost::none /* databaseVersion */); + + uassert(std::move(resolvedView), + "Explain of a resolved view must be executed by mongos", + !ShardingState::get(opCtx)->enabled() || !request.getExplain()); + + // Parse the resolved view into a new aggregation request. + auto newRequest = resolvedView.asExpandedViewAggregation(request); + auto newCmd = aggregation_request_helper::serializeToCommandObj(newRequest); + + auto status{Status::OK()}; + try { + status = runAggregate(opCtx, origNss, newRequest, newCmd, privileges, result); + } catch (const ExceptionForCat<ErrorCategory::StaleShardVersionError>& ex) { + // Since we expect the view to be UNSHARDED, if we reached to this point there are + // two possibilities: + // 1. The shard doesn't know what its shard version/state is and needs to recover + // it (in which case we throw so that the shard can run recovery) + // 2. The collection references by the view is actually SHARDED, in which case the + // router must execute it + if (const auto staleInfo{ex.extraInfo<StaleConfigInfo>()}) { + uassert(std::move(resolvedView), + "Resolved views on sharded collections must be executed by mongos", + !staleInfo->getVersionWanted()); + } + throw; + } + + { + // Set the namespace of the curop back to the view namespace so ctx records + // stats on this view namespace on destruction. + stdx::lock_guard<Client> lk(*opCtx->getClient()); + curOp->setNS_inlock(nss.ns()); + } + + return status; } + // If collectionUUID was provided, verify the collection exists and has the expected UUID. + checkCollectionUUIDMismatch(opCtx, + nss, + collections.getMainCollection(), + request.getCollectionUUID(), + false /* checkFeatureFlag */); + invariant(collatorToUse); - auto pipeline = parsePipelineAndRegisterQueryStats(opCtx, - origNss, - request, - ctx, - std::move(*collatorToUse), - uuid, - collatorToUseMatchesDefault, - collections, - pipelineInvolvedNamespaces, - liteParsedPipeline, - nss.isCollectionlessAggregateNS(), - resolvedView, - origRequest); - expCtx = pipeline->getContext(); - - CurOp::get(opCtx)->beginQueryPlanningTimer(); + expCtx = makeExpressionContext( + opCtx, request, std::move(*collatorToUse), uuid, collatorToUseMatchesDefault); + + expCtx->startExpressionCounters(); + auto pipeline = Pipeline::parse(request.getPipeline(), expCtx); + expCtx->stopExpressionCounters(); if (!request.getAllowDiskUse().value_or(true)) { allowDiskUseFalseCounter.increment(); @@ -1126,9 +982,6 @@ Status runAggregate(OperationContext* opCtx, // cursor manager. The global cursor manager does not deliver invalidations or kill // notifications; the underlying PlanExecutor(s) used by the pipeline will be receiving // invalidations and kill notifications themselves, not the cursor we create here. - hangAfterCreatingAggregationPlan.executeIf( - [](const auto&) { hangAfterCreatingAggregationPlan.pauseWhileSet(); }, - [&](const BSONObj& data) { return uuid && UUID::parse(data["uuid"]) == *uuid; }); std::vector<ClientCursorPin> pins; std::vector<ClientCursor*> cursors; @@ -1139,7 +992,6 @@ Status runAggregate(OperationContext* opCtx, } }); for (auto&& exec : execs) { - // TODO SERVER-79373: Do not create a cursor if results can fit in a single batch. ClientCursorParams cursorParams( std::move(exec), origNss, @@ -1183,7 +1035,6 @@ Status runAggregate(OperationContext* opCtx, cmdObj, &bodyBuilder); } - collectQueryStatsMongod(opCtx, std::move(curOp->debug().queryStatsInfo.key)); } else { // Cursor must be specified, if explain is not. const bool keepCursor = handleCursorCommand( @@ -1196,15 +1047,13 @@ Status runAggregate(OperationContext* opCtx, PlanSummaryStats stats; planExplainer.getSummaryStats(&stats); curOp->debug().setPlanSummaryMetrics(stats); - curOp->setEndOfOpMetrics(stats.nReturned); + curOp->debug().nreturned = stats.nReturned; - collectQueryStatsMongod(opCtx, pins[0]); - - // For an optimized away pipeline, signal the cache that a query operation has - // completed. For normal pipelines this is done in DocumentSourceCursor. + // For an optimized away pipeline, signal the cache that a query operation has completed. + // For normal pipelines this is done in DocumentSourceCursor. if (ctx) { - // Due to yielding, the collection pointers saved in MultipleCollectionAccessor - // might have become invalid. We will need to refresh them here. + // Due to yielding, the collection pointers saved in MultipleCollectionAccessor might + // have become invalid. We will need to refresh them here. collections = MultipleCollectionAccessor(opCtx, &ctx->getCollection(), ctx->getNss(), @@ -1227,11 +1076,10 @@ Status runAggregate(OperationContext* opCtx, } } - // The aggregation pipeline may change the namespace of the curop and we need to set it back - // to the original namespace to correctly report command stats. One example when the - // namespace can be changed is when the pipeline contains an $out stage, which executes an - // internal command to create a temp collection, changing the curop namespace to the name of - // this temp collection. + // The aggregation pipeline may change the namespace of the curop and we need to set it back to + // the original namespace to correctly report command stats. One example when the namespace can + // be changed is when the pipeline contains an $out stage, which executes an internal command to + // create a temp collection, changing the curop namespace to the name of this temp collection. { stdx::lock_guard<Client> lk(*opCtx->getClient()); curOp->setNS_inlock(origNss.ns()); @@ -1239,4 +1087,5 @@ Status runAggregate(OperationContext* opCtx, return Status::OK(); } + } // namespace mongo |