diff options
Diffstat (limited to 'src/mongo/db/commands/run_aggregate.cpp')
| -rw-r--r-- | src/mongo/db/commands/run_aggregate.cpp | 204 |
1 files changed, 121 insertions, 83 deletions
diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp index 9172103b493..ac43ccf8b36 100644 --- a/src/mongo/db/commands/run_aggregate.cpp +++ b/src/mongo/db/commands/run_aggregate.cpp @@ -221,7 +221,6 @@ 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), @@ -657,6 +656,93 @@ 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, + const boost::intrusive_ptr<ExpressionContext>& expCtx, + std::shared_ptr<const CollectionCatalog> catalog, + const PrivilegeVector& privileges, + CurOp* curOp, + rpc::ReplyBuilderInterface* result, + const std::function<void(void)>& resetContextFn) { + auto nss = request.getNamespace(); + checkCollectionUUIDMismatch( + opCtx, nss, collections.getMainCollection(), request.getCollectionUUID()); + + 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, 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; +} + } // namespace Status runAggregate(OperationContext* opCtx, @@ -745,7 +831,8 @@ Status runAggregate(OperationContext* opCtx, boost::optional<AutoStatsTracker> statsTracker; // If this is a change stream, perform special checks and change the execution namespace. - if (liteParsedPipeline.hasChangeStream()) { + const auto hasChangeStream = liteParsedPipeline.hasChangeStream(); + if (hasChangeStream) { uassert(4928900, str::stream() << AggregateCommandRequest::kCollectionUUIDFieldName << " is not supported for a change stream", @@ -820,87 +907,24 @@ Status runAggregate(OperationContext* opCtx, // 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.) - 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; + // 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(), + expCtx, + catalog, + privileges, + curOp, + result, + resetContext); } // If collectionUUID was provided, verify the collection exists and has the expected UUID. @@ -914,6 +938,20 @@ Status runAggregate(OperationContext* opCtx, expCtx = makeExpressionContext( opCtx, request, std::move(*collatorToUse), uuid, collatorToUseMatchesDefault); + // 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); + }); + + // 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; + expCtx->ignoreTokenVersionOnResume = true; + } + expCtx->startExpressionCounters(); auto pipeline = Pipeline::parse(request.getPipeline(), expCtx); expCtx->stopExpressionCounters(); |