diff options
Diffstat (limited to 'src/mongo/s/query')
23 files changed, 203 insertions, 711 deletions
diff --git a/src/mongo/s/query/SConscript b/src/mongo/s/query/SConscript index f8179e37eeb..f6f1ac53b05 100644 --- a/src/mongo/s/query/SConscript +++ b/src/mongo/s/query/SConscript @@ -8,17 +8,17 @@ env.Library( target="cluster_query", source=[ "cluster_find.cpp", - "cluster_query_knobs.idl", - "store_possible_cursor.cpp", + 'cluster_query_knobs.idl', ], LIBDEPS=[ '$BUILD_DIR/mongo/db/commands', + '$BUILD_DIR/mongo/db/curop', '$BUILD_DIR/mongo/db/curop_failpoint_helpers', '$BUILD_DIR/mongo/db/query/query_common', - '$BUILD_DIR/mongo/db/query/query_stats/query_stats', '$BUILD_DIR/mongo/s/sharding_router_api', "cluster_client_cursor", "cluster_cursor_cleanup_job", + "store_possible_cursor", ], LIBDEPS_PRIVATE=[ '$BUILD_DIR/mongo/db/catalog/collection_uuid_mismatch_info', @@ -37,14 +37,12 @@ env.Library( '$BUILD_DIR/mongo/db/pipeline/pipeline', '$BUILD_DIR/mongo/db/pipeline/process_interface/mongos_process_interface', '$BUILD_DIR/mongo/db/pipeline/sharded_agg_helpers', - '$BUILD_DIR/mongo/db/query/query_shape/query_shape', '$BUILD_DIR/mongo/db/views/view_catalog_helpers', '$BUILD_DIR/mongo/db/views/views', '$BUILD_DIR/mongo/s/query/cluster_client_cursor', 'cluster_query', ], LIBDEPS_PRIVATE=[ - '$BUILD_DIR/mongo/db/catalog/collection_uuid_mismatch_info', '$BUILD_DIR/mongo/db/timeseries/timeseries_options', ] ) @@ -94,11 +92,24 @@ env.Library( ], LIBDEPS_PRIVATE=[ '$BUILD_DIR/mongo/db/catalog/collection_uuid_mismatch_info', - '$BUILD_DIR/mongo/executor/async_multicaster', ] ) env.Library( + target="store_possible_cursor", + source=[ + "store_possible_cursor.cpp" + ], + LIBDEPS=[ + "$BUILD_DIR/mongo/base", + "$BUILD_DIR/mongo/db/curop", + "$BUILD_DIR/mongo/db/query/command_request_response", + "cluster_client_cursor", + "cluster_cursor_manager", + ], +) + +env.Library( target="cluster_cursor_manager", source=[ "cluster_cursor_manager.cpp", @@ -107,12 +118,12 @@ env.Library( '$BUILD_DIR/mongo/base', '$BUILD_DIR/mongo/db/auth/auth', '$BUILD_DIR/mongo/db/auth/authprivilege', + '$BUILD_DIR/mongo/db/curop', '$BUILD_DIR/mongo/db/generic_cursor', '$BUILD_DIR/mongo/db/kill_sessions', '$BUILD_DIR/mongo/db/logical_session_cache', '$BUILD_DIR/mongo/db/logical_session_id', '$BUILD_DIR/mongo/db/query/query_knobs', - '$BUILD_DIR/mongo/db/query/query_stats/query_stats', ], ) @@ -159,7 +170,7 @@ env.CppUnitTest( "cluster_aggregate", "cluster_client_cursor", "cluster_cursor_manager", - "cluster_query", "router_exec_stage", + "store_possible_cursor", ], ) diff --git a/src/mongo/s/query/async_results_merger.cpp b/src/mongo/s/query/async_results_merger.cpp index 50fb6310888..363e151fbfd 100644 --- a/src/mongo/s/query/async_results_merger.cpp +++ b/src/mongo/s/query/async_results_merger.cpp @@ -846,8 +846,7 @@ void AsyncResultsMerger::_scheduleKillCursors(WithLock, OperationContext* opCtx) invariant(_killCompleteInfo); for (const auto& remote : _remotes) { - if ((remote.status.isOK() || remote.status == ErrorCodes::MaxTimeMSExpired) && - remote.cursorId && !remote.exhausted()) { + if (remote.status.isOK() && remote.cursorId && !remote.exhausted()) { BSONObj cmdObj = KillCursorsCommandRequest(_params.getNss(), {remote.cursorId}).toBSON(BSONObj{}); diff --git a/src/mongo/s/query/async_results_merger_params.idl b/src/mongo/s/query/async_results_merger_params.idl index 5382c9cc718..2b5857e6c9b 100644 --- a/src/mongo/s/query/async_results_merger_params.idl +++ b/src/mongo/s/query/async_results_merger_params.idl @@ -50,19 +50,15 @@ types: structs: RemoteCursor: description: A description of a cursor opened on a remote server. - query_shape_component: true fields: shardId: type: string description: The shardId of the shard on which the cursor resides. - query_shape: anonymize hostAndPort: type: HostAndPort description: The exact host (within the shard) on which the cursor resides. - query_shape: anonymize cursorResponse: type: CursorResponse - query_shape: literal description: The response after establishing a cursor on the remote shard, including the first batch. @@ -70,47 +66,33 @@ structs: description: The parameters needed to establish an AsyncResultsMerger. chained_structs: OperationSessionInfoFromClient : OperationSessionInfo - query_shape_component: true fields: sort: type: object description: The sort requested on the merging operation. Empty if there is no sort. optional: true - query_shape: literal compareWholeSortKey: type: bool default: false - query_shape: literal description: >- When 'compareWholeSortKey' is true, $sortKey is a scalar value, rather than an object. We extract the sort key {$sortKey: <value>}. The sort key pattern is verified to be {$sortKey: 1}. - remotes: - type: array<RemoteCursor> - query_shape: literal + remotes: array<RemoteCursor> tailableMode: type: TailableMode optional: true description: If set, the tailability mode of this cursor. - query_shape: parameter batchSize: type: safeInt64 optional: true description: The batch size for this cursor. - query_shape: literal - nss: - type: namespacestring - query_shape: custom + nss: namespacestring allowPartialResults: type: bool default: false description: If set, error responses are ignored. - query_shape: parameter recordRemoteOpWaitTime: type: bool default: false - query_shape: parameter - description: >- - This parameter is not used anymore but should stay for a while for backward - compatibility. - TODO SERVER-73120 Remove this parameter when releasing 8.0. + description: If set, records the total time spent waiting for remote operations to complete. diff --git a/src/mongo/s/query/blocking_results_merger.cpp b/src/mongo/s/query/blocking_results_merger.cpp index fc56a0d9e3b..c1a957a311f 100644 --- a/src/mongo/s/query/blocking_results_merger.cpp +++ b/src/mongo/s/query/blocking_results_merger.cpp @@ -42,6 +42,7 @@ BlockingResultsMerger::BlockingResultsMerger(OperationContext* opCtx, std::shared_ptr<executor::TaskExecutor> executor, std::unique_ptr<ResourceYielder> resourceYielder) : _tailableMode(armParams.getTailableMode().value_or(TailableModeEnum::kNormal)), + _recordRemoteOpWaitTime(armParams.getRecordRemoteOpWaitTime()), _executor(executor), _arm(opCtx, std::move(executor), std::move(armParams)), _resourceYielder(std::move(resourceYielder)) {} @@ -145,7 +146,9 @@ StatusWith<ClusterQueryResult> BlockingResultsMerger::blockUntilNext(OperationCo return _arm.nextReady(); } StatusWith<ClusterQueryResult> BlockingResultsMerger::next(OperationContext* opCtx) { - CurOp::get(opCtx)->ensureRecordRemoteOpWait(); + if (_recordRemoteOpWaitTime) { + CurOp::get(opCtx)->enableRecordRemoteOpWait(); + } // Non-tailable and tailable non-awaitData cursors always block until ready(). AwaitData // cursors wait for ready() only until a specified time limit is exceeded. diff --git a/src/mongo/s/query/blocking_results_merger.h b/src/mongo/s/query/blocking_results_merger.h index c05cecc5da8..4a368d9471a 100644 --- a/src/mongo/s/query/blocking_results_merger.h +++ b/src/mongo/s/query/blocking_results_merger.h @@ -118,6 +118,7 @@ private: const std::function<StatusWith<stdx::cv_status>()>& waitFn) noexcept; TailableModeEnum _tailableMode; + bool _recordRemoteOpWaitTime; std::shared_ptr<executor::TaskExecutor> _executor; // In a case where we have a tailable, awaitData cursor, a call to 'next()' will block waiting 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; diff --git a/src/mongo/s/query/cluster_aggregation_planner.cpp b/src/mongo/s/query/cluster_aggregation_planner.cpp index 70adf70404a..3f970f2f8c1 100644 --- a/src/mongo/s/query/cluster_aggregation_planner.cpp +++ b/src/mongo/s/query/cluster_aggregation_planner.cpp @@ -65,7 +65,6 @@ namespace cluster_aggregation_planner { MONGO_FAIL_POINT_DEFINE(shardedAggregateFailToDispatchExchangeConsumerPipeline); MONGO_FAIL_POINT_DEFINE(shardedAggregateFailToEstablishMergingShardCursor); -MONGO_FAIL_POINT_DEFINE(shardedAggregateHangBeforeDispatchMergingPipeline); using sharded_agg_helpers::DispatchShardPipelineResults; using sharded_agg_helpers::SplitPipeline; @@ -172,13 +171,9 @@ Status dispatchMergingPipeline(const boost::intrusive_ptr<ExpressionContext>& ex const PrivilegeVector& privileges, bool hasChangeStream) { // We should never be in a situation where we call this function on a non-merge pipeline. - tassert(6525900, - "tried to dispatch merge pipeline but the pipeline was not split", - shardDispatchResults.splitPipeline); + invariant(shardDispatchResults.splitPipeline); auto* mergePipeline = shardDispatchResults.splitPipeline->mergePipeline.get(); - tassert(6525901, - "tried to dispatch merge pipeline but there was no merge portion of the split pipeline", - mergePipeline); + invariant(mergePipeline); auto* opCtx = expCtx->opCtx; std::vector<ShardId> targetedShards; @@ -237,13 +232,9 @@ Status dispatchMergingPipeline(const boost::intrusive_ptr<ExpressionContext>& ex privileges, expCtx->tailableMode)); - // If the mergingShard returned an error and did not accept ownership it is our responsibility - // to kill the cursors. - uassertStatusOK(getStatusFromCommandResult(mergeResponse.swResponse.getValue().data)); - - // If we didn't get an error from the merging shard, ownership for the shard cursors has been - // transferred to the merging shard. Dismiss the ownership in the current merging pipeline such - // that when it goes out of scope it does not attempt to kill the cursors. + // Ownership for the shard cursors has been transferred to the merging shard. Dismiss the + // ownership in the current merging pipeline such that when it goes out of scope it does not + // attempt to kill the cursors. auto mergeCursors = static_cast<DocumentSourceMergeCursors*>(mergePipeline->peekFront()); mergeCursors->dismissCursorOwnership(); @@ -345,27 +336,12 @@ BSONObj establishMergingMongosCursor(OperationContext* opCtx, responseBuilder.setPostBatchResumeToken(ccc->getPostBatchResumeToken()); } - bool exhausted = cursorState != ClusterCursorManager::CursorState::NotExhausted; - int nShards = ccc->getNumRemotes(); - - auto&& opDebug = CurOp::get(opCtx)->debug(); - // Fill out the aggregation metrics in CurOp, and record queryStats metrics, before detaching - // the cursor from its opCtx. - opDebug.nShards = std::max(opDebug.nShards, nShards); - opDebug.cursorExhausted = exhausted; - opDebug.additiveMetrics.nBatches = 1; - CurOp::get(opCtx)->setEndOfOpMetrics(responseBuilder.numDocs()); - if (exhausted) { - collectQueryStatsMongos(opCtx, ccc->takeKey()); - } else { - collectQueryStatsMongos(opCtx, ccc); - } - ccc->detachFromOperationContext(); + int nShards = ccc->getNumRemotes(); CursorId clusterCursorId = 0; - if (!exhausted) { + if (cursorState == ClusterCursorManager::CursorState::NotExhausted) { auto authUsers = AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames(); clusterCursorId = uassertStatusOK(Grid::get(opCtx)->getCursorManager()->registerCursor( opCtx, @@ -374,9 +350,16 @@ BSONObj establishMergingMongosCursor(OperationContext* opCtx, ClusterCursorManager::CursorType::MultiTarget, ClusterCursorManager::CursorLifetime::Mortal, authUsers)); - opDebug.cursorid = clusterCursorId; } + // Fill out the aggregation metrics in CurOp. + if (clusterCursorId > 0) { + CurOp::get(opCtx)->debug().cursorid = clusterCursorId; + } + CurOp::get(opCtx)->debug().nShards = std::max(CurOp::get(opCtx)->debug().nShards, nShards); + CurOp::get(opCtx)->debug().cursorExhausted = (clusterCursorId == 0); + CurOp::get(opCtx)->debug().nreturned = responseBuilder.numDocs(); + responseBuilder.done(clusterCursorId, requestedNss.ns()); auto bodyBuilder = replyBuilder.getBodyBuilder(); @@ -391,9 +374,6 @@ DispatchShardPipelineResults dispatchExchangeConsumerPipeline( const NamespaceString& executionNss, Document serializedCommand, DispatchShardPipelineResults* shardDispatchResults) { - tassert(7163600, - "dispatchExchangeConsumerPipeline() must not be called for explain operation", - !expCtx->explain); auto opCtx = expCtx->opCtx; if (MONGO_unlikely(shardedAggregateFailToDispatchExchangeConsumerPipeline.shouldFail())) { @@ -430,8 +410,7 @@ DispatchShardPipelineResults dispatchExchangeConsumerPipeline( serializedCommand, consumerPipelines.back(), boost::none, /* exchangeSpec */ - false /* needsMerge */, - boost::none /* explain */); + false /* needsMerge */); requests.emplace_back(shardDispatchResults->exchangeSpec->consumerShards[idx], consumerCmdObj); @@ -456,11 +435,8 @@ DispatchShardPipelineResults dispatchExchangeConsumerPipeline( SplitPipeline splitPipeline{nullptr, std::move(mergePipeline), boost::none}; - // Relinquish ownership of the consumer pipelines' cursors. These cursors are now set up to be - // merged by a set of $mergeCursors pipelines that we just dispatched to the shards above. Now - // that we've established those pipelines on the shards, we are no longer responsible for - // ensuring they are cleaned up. If there was a problem establishing the cursors then - // establishCursors() would have thrown and mongos would kill all the consumer cursors itself. + // Relinquish ownership of the local consumer pipelines' cursors as each shard is now + // responsible for its own producer cursors. for (const auto& pipeline : consumerPipelines) { const auto& mergeCursors = static_cast<DocumentSourceMergeCursors*>(pipeline.shardsPipeline->peekFront()); @@ -587,7 +563,6 @@ AggregationTargeter AggregationTargeter::make( boost::optional<ChunkManager> cm, stdx::unordered_set<NamespaceString> involvedNamespaces, bool hasChangeStream, - bool startsWithDocuments, bool allowedToPassthrough, bool perShardCursor) { if (perShardCursor) { @@ -607,13 +582,11 @@ AggregationTargeter AggregationTargeter::make( }(); // Determine whether this aggregation must be dispatched to all shards in the cluster. - const bool mustRunOnAllShards = sharded_agg_helpers::checkIfMustRunOnAllShards( - executionNss, hasChangeStream, startsWithDocuments); + const bool mustRunOnAll = + sharded_agg_helpers::mustRunOnAllShards(executionNss, hasChangeStream); - // If we don't have a routing table, then this is either a $changeStream which must run on all - // shards or a $documents stage which must not. - invariant(cm || (mustRunOnAllShards && hasChangeStream) || - (startsWithDocuments && !mustRunOnAllShards)); + // If we don't have a routing table, then this is a $changeStream which must run on all shards. + invariant(cm || (mustRunOnAll && hasChangeStream)); // A pipeline is allowed to passthrough to the primary shard iff the following conditions are // met: @@ -625,7 +598,7 @@ AggregationTargeter AggregationTargeter::make( // $currentOp. // 4. Doesn't need transformation via DocumentSource::serialize(). For example, list sessions // needs to include information about users that can only be deduced on mongos. - if (cm && !cm->isSharded() && !mustRunOnAllShards && allowedToPassthrough && + if (cm && !cm->isSharded() && !mustRunOnAll && allowedToPassthrough && !involvesShardedCollections) { return AggregationTargeter{TargetingPolicy::kPassthrough, nullptr, cm}; } else { @@ -690,16 +663,11 @@ Status dispatchPipelineAndMerge(OperationContext* opCtx, const ClusterAggregate::Namespaces& namespaces, const PrivilegeVector& privileges, BSONObjBuilder* result, - bool hasChangeStream, - bool startsWithDocuments) { + bool hasChangeStream) { auto expCtx = targeter.pipeline->getContext(); // If not, split the pipeline as necessary and dispatch to the relevant shards. - auto shardDispatchResults = - sharded_agg_helpers::dispatchShardPipeline(serializedCommand, - hasChangeStream, - startsWithDocuments, - std::move(targeter.pipeline), - expCtx->explain); + auto shardDispatchResults = sharded_agg_helpers::dispatchShardPipeline( + serializedCommand, hasChangeStream, std::move(targeter.pipeline)); // If the operation is an explain, then we verify that it succeeded on all targeted // shards, write the results to the output builder, and return immediately. @@ -733,8 +701,6 @@ Status dispatchPipelineAndMerge(OperationContext* opCtx, expCtx, namespaces.executionNss, serializedCommand, &shardDispatchResults); } - shardedAggregateHangBeforeDispatchMergingPipeline.pauseWhileSet(); - // If we reach here, we have a merge pipeline to dispatch. return dispatchMergingPipeline(expCtx, namespaces, @@ -867,7 +833,6 @@ Status runPipelineOnSpecificShardOnly(const boost::intrusive_ptr<ExpressionConte if (explain) { // If this was an explain, then we get back an explain result object rather than a cursor. result = response.swResponse.getValue().data; - collectQueryStatsMongos(opCtx, std::move(CurOp::get(opCtx)->debug().queryStatsInfo.key)); } else { result = uassertStatusOK(storePossibleCursor( opCtx, diff --git a/src/mongo/s/query/cluster_aggregation_planner.h b/src/mongo/s/query/cluster_aggregation_planner.h index 78c919192db..046d8eab6ca 100644 --- a/src/mongo/s/query/cluster_aggregation_planner.h +++ b/src/mongo/s/query/cluster_aggregation_planner.h @@ -82,7 +82,6 @@ struct AggregationTargeter { boost::optional<ChunkManager> cm, stdx::unordered_set<NamespaceString> involvedNamespaces, bool hasChangeStream, - bool startsWithDocuments, bool allowedToPassthrough, bool perShardCursor); @@ -126,8 +125,7 @@ Status dispatchPipelineAndMerge(OperationContext* opCtx, const ClusterAggregate::Namespaces& namespaces, const PrivilegeVector& privileges, BSONObjBuilder* result, - bool hasChangeStream, - bool startsWithDocuments); + bool hasChangeStream); /** * Similar to runPipelineOnPrimaryShard but allows $changeStreams. Intended for use by per shard diff --git a/src/mongo/s/query/cluster_client_cursor.h b/src/mongo/s/query/cluster_client_cursor.h index 23a3367416d..8ff611eb308 100644 --- a/src/mongo/s/query/cluster_client_cursor.h +++ b/src/mongo/s/query/cluster_client_cursor.h @@ -211,30 +211,15 @@ public: */ virtual boost::optional<uint32_t> getQueryHash() const = 0; - virtual boost::optional<std::size_t> getQueryStatsKeyHash() const = 0; - - virtual bool getQueryStatsWillNeverExhaust() const = 0; - /** * Returns the number of batches returned by this cursor. */ - std::uint64_t getNBatches() const { - return _metrics.nBatches.value_or(0); - } + virtual std::uint64_t getNBatches() const = 0; /** * Increment the number of batches returned so far by one. */ - void incNBatches() { - _metrics.incrementNBatches(); - } - - void incrementCursorMetrics(OpDebug::AdditiveMetrics newMetrics) { - _metrics.add(newMetrics); - if (!_firstResponseExecutionTime) { - _firstResponseExecutionTime = _metrics.executionTime; - } - } + virtual void incNBatches() = 0; // // maxTimeMS support. @@ -260,20 +245,6 @@ public: _leftoverMaxTimeMicros = leftoverMaxTimeMicros; } - /** - * Returns and releases ownership of the Key associated with the request this - * cursor is handling. - */ - virtual std::unique_ptr<query_stats::Key> takeKey() = 0; - -protected: - // Metrics that are accumulated over the lifetime of the cursor, incremented with each getMore. - // Useful for diagnostics like queryStats. - OpDebug::AdditiveMetrics _metrics; - - // The execution time collected from the initial operation prior to any getMore requests. - boost::optional<Microseconds> _firstResponseExecutionTime; - private: // Unused maxTime budget for this cursor. Microseconds _leftoverMaxTimeMicros = Microseconds::max(); diff --git a/src/mongo/s/query/cluster_client_cursor_impl.cpp b/src/mongo/s/query/cluster_client_cursor_impl.cpp index 6b094a604f4..73be5a7512a 100644 --- a/src/mongo/s/query/cluster_client_cursor_impl.cpp +++ b/src/mongo/s/query/cluster_client_cursor_impl.cpp @@ -27,8 +27,6 @@ * it in the license file. */ -#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kQuery - #include "mongo/platform/basic.h" #include "mongo/s/query/cluster_client_cursor_impl.h" @@ -36,8 +34,6 @@ #include <memory> #include "mongo/db/curop.h" -#include "mongo/db/query/query_stats/query_stats.h" -#include "mongo/logv2/log.h" #include "mongo/s/query/router_stage_limit.h" #include "mongo/s/query/router_stage_merge.h" #include "mongo/s/query/router_stage_remove_metadata_fields.h" @@ -79,10 +75,7 @@ ClusterClientCursorImpl::ClusterClientCursorImpl(OperationContext* opCtx, _opCtx(opCtx), _createdDate(opCtx->getServiceContext()->getPreciseClockSource()->now()), _lastUseDate(_createdDate), - _queryHash(CurOp::get(opCtx)->debug().queryHash), - _queryStatsKeyHash(CurOp::get(opCtx)->debug().queryStatsInfo.keyHash), - _queryStatsKey(std::move(CurOp::get(opCtx)->debug().queryStatsInfo.key)), - _queryStatsWillNeverExhaust(CurOp::get(opCtx)->debug().queryStatsInfo.willNeverExhaust) { + _queryHash(CurOp::get(opCtx)->debug().queryHash) { dassert(!_params.compareWholeSortKeyOnRouter || SimpleBSONObjComparator::kInstance.evaluate( _params.sortToApplyOnRouter == AsyncResultsMerger::kWholeSortKeySortPattern)); @@ -99,11 +92,7 @@ ClusterClientCursorImpl::ClusterClientCursorImpl(OperationContext* opCtx, _opCtx(opCtx), _createdDate(opCtx->getServiceContext()->getPreciseClockSource()->now()), _lastUseDate(_createdDate), - _queryHash(CurOp::get(opCtx)->debug().queryHash), - _queryStatsKeyHash(CurOp::get(opCtx)->debug().queryStatsInfo.keyHash), - _queryStatsKey(std::move(CurOp::get(opCtx)->debug().queryStatsInfo.key)), - _queryStatsWillNeverExhaust( - std::move(CurOp::get(opCtx)->debug().queryStatsInfo.willNeverExhaust)) { + _queryHash(CurOp::get(opCtx)->debug().queryHash) { dassert(!_params.compareWholeSortKeyOnRouter || SimpleBSONObjComparator::kInstance.evaluate( _params.sortToApplyOnRouter == AsyncResultsMerger::kWholeSortKeySortPattern)); @@ -111,7 +100,7 @@ ClusterClientCursorImpl::ClusterClientCursorImpl(OperationContext* opCtx, } ClusterClientCursorImpl::~ClusterClientCursorImpl() { - if (_metrics.nBatches && *_metrics.nBatches > 1) + if (_nBatchesReturned > 1) mongosCursorStatsMoreThanOneBatch.increment(); } @@ -139,25 +128,7 @@ StatusWith<ClusterQueryResult> ClusterClientCursorImpl::next() { } void ClusterClientCursorImpl::kill(OperationContext* opCtx) { - if (_hasBeenKilled) { - LOGV2_DEBUG(7372700, - 3, - "Kill called on cluster client cursor after cursor has already been killed, so " - "ignoring"); - return; - } - - query_stats::writeQueryStatsOnCursorDisposeOrKill( - opCtx, - _queryStatsKeyHash, - std::move(_queryStatsKey), - _queryStatsWillNeverExhaust, - _metrics.executionTime.value_or(Microseconds{0}).count(), - _firstResponseExecutionTime.value_or(Microseconds{0}).count(), - _metrics.nreturned.value_or(0)); - _root->kill(opCtx); - _hasBeenKilled = true; } void ClusterClientCursorImpl::reattachToOperationContext(OperationContext* opCtx) { @@ -246,12 +217,12 @@ boost::optional<uint32_t> ClusterClientCursorImpl::getQueryHash() const { return _queryHash; } -boost::optional<std::size_t> ClusterClientCursorImpl::getQueryStatsKeyHash() const { - return _queryStatsKeyHash; +std::uint64_t ClusterClientCursorImpl::getNBatches() const { + return _nBatchesReturned; } -bool ClusterClientCursorImpl::getQueryStatsWillNeverExhaust() const { - return _queryStatsWillNeverExhaust; +void ClusterClientCursorImpl::incNBatches() { + ++_nBatchesReturned; } APIParameters ClusterClientCursorImpl::getAPIParameters() const { @@ -294,7 +265,4 @@ std::unique_ptr<RouterExecStage> ClusterClientCursorImpl::buildMergerPlan( return root; } -std::unique_ptr<query_stats::Key> ClusterClientCursorImpl::takeKey() { - return std::move(_queryStatsKey); -} } // namespace mongo diff --git a/src/mongo/s/query/cluster_client_cursor_impl.h b/src/mongo/s/query/cluster_client_cursor_impl.h index 8064a45595b..2529254cfce 100644 --- a/src/mongo/s/query/cluster_client_cursor_impl.h +++ b/src/mongo/s/query/cluster_client_cursor_impl.h @@ -32,7 +32,6 @@ #include <memory> #include <queue> -#include "mongo/bson/bsonobj.h" #include "mongo/executor/task_executor.h" #include "mongo/s/query/cluster_client_cursor.h" #include "mongo/s/query/cluster_client_cursor_guard.h" @@ -117,11 +116,9 @@ public: boost::optional<uint32_t> getQueryHash() const final; - boost::optional<std::size_t> getQueryStatsKeyHash() const final; + std::uint64_t getNBatches() const final; - bool getQueryStatsWillNeverExhaust() const final; - - std::unique_ptr<query_stats::Key> takeKey() final; + void incNBatches() final; public: /** @@ -178,19 +175,8 @@ private: // The hash of the query shape to be used for slow query logging; boost::optional<uint32_t> _queryHash; - // If boost::none, queryStats should not be collected for this cursor. - boost::optional<std::size_t> _queryStatsKeyHash; - - // The Key used by query stats to generate the query stats store key. - std::unique_ptr<query_stats::Key> _queryStatsKey; - - bool _queryStatsWillNeverExhaust = false; - - // Tracks if kill() has been called on the cursor. Multiple calls to kill() are treated as a - // noop. - // TODO SERVER-74482 investigate where kill() is called multiple times and remove unnecessary - // calls - bool _hasBeenKilled = false; + // The number of batches returned by this cursor. + std::uint64_t _nBatchesReturned = 0; }; } // namespace mongo diff --git a/src/mongo/s/query/cluster_client_cursor_mock.cpp b/src/mongo/s/query/cluster_client_cursor_mock.cpp index 103951694a0..567f3450499 100644 --- a/src/mongo/s/query/cluster_client_cursor_mock.cpp +++ b/src/mongo/s/query/cluster_client_cursor_mock.cpp @@ -89,6 +89,14 @@ long long ClusterClientCursorMock::getNumReturnedSoFar() const { return _numReturnedSoFar; } +std::uint64_t ClusterClientCursorMock::getNBatches() const { + return _nBatchesReturned; +} + +void ClusterClientCursorMock::incNBatches() { + ++_nBatchesReturned; +} + Date_t ClusterClientCursorMock::getCreatedDate() const { return _createdDate; } @@ -105,14 +113,6 @@ boost::optional<uint32_t> ClusterClientCursorMock::getQueryHash() const { return boost::none; } -boost::optional<std::size_t> ClusterClientCursorMock::getQueryStatsKeyHash() const { - return boost::none; -} - -bool ClusterClientCursorMock::getQueryStatsWillNeverExhaust() const { - return false; -} - void ClusterClientCursorMock::kill(OperationContext* opCtx) { _killed = true; if (_killCallback) { @@ -168,8 +168,4 @@ boost::optional<repl::ReadConcernArgs> ClusterClientCursorMock::getReadConcern() return boost::none; } -std::unique_ptr<query_stats::Key> ClusterClientCursorMock::takeKey() { - return nullptr; -} - } // namespace mongo diff --git a/src/mongo/s/query/cluster_client_cursor_mock.h b/src/mongo/s/query/cluster_client_cursor_mock.h index 64ec06d750f..bc2991ecf89 100644 --- a/src/mongo/s/query/cluster_client_cursor_mock.h +++ b/src/mongo/s/query/cluster_client_cursor_mock.h @@ -33,7 +33,7 @@ #include <functional> #include <queue> -#include "mongo/db/query/query_stats/key.h" +#include "mongo/db/logical_session_id.h" #include "mongo/s/query/cluster_client_cursor.h" namespace mongo { @@ -106,9 +106,9 @@ public: boost::optional<uint32_t> getQueryHash() const final; - boost::optional<std::size_t> getQueryStatsKeyHash() const final; + std::uint64_t getNBatches() const final; - bool getQueryStatsWillNeverExhaust() const final; + void incNBatches() final; /** * Returns false unless the mock cursor has been fully iterated. @@ -120,8 +120,6 @@ public: */ void queueError(Status status); - std::unique_ptr<query_stats::Key> takeKey() final; - private: bool _killed = false; std::queue<StatusWith<ClusterQueryResult>> _resultsQueue; diff --git a/src/mongo/s/query/cluster_cursor_manager.cpp b/src/mongo/s/query/cluster_cursor_manager.cpp index 4452e6a8811..1209d709e22 100644 --- a/src/mongo/s/query/cluster_cursor_manager.cpp +++ b/src/mongo/s/query/cluster_cursor_manager.cpp @@ -41,7 +41,6 @@ #include "mongo/db/kill_sessions_common.h" #include "mongo/db/logical_session_cache.h" #include "mongo/db/query/query_knobs_gen.h" -#include "mongo/db/query/query_stats/query_stats.h" #include "mongo/logv2/log.h" #include "mongo/util/clock_source.h" #include "mongo/util/str.h" @@ -247,7 +246,6 @@ StatusWith<ClusterCursorManager::PinnedCursor> ClusterCursorManager::checkOutCur cursorGuard->reattachToOperationContext(opCtx); CurOp::get(opCtx)->debug().queryHash = cursorGuard->getQueryHash(); - CurOp::get(opCtx)->debug().queryStatsInfo.keyHash = cursorGuard->getQueryStatsKeyHash(); return PinnedCursor(this, std::move(cursorGuard), entry->getNamespace(), cursorId); } @@ -576,60 +574,4 @@ StatusWith<ClusterClientCursorGuard> ClusterCursorManager::_detachCursor(WithLoc return std::move(cursor); } - -void collectQueryStatsMongos(OperationContext* opCtx, std::unique_ptr<query_stats::Key> key) { - // If we haven't registered a cursor to prepare for getMore requests, we record - // queryStats directly. - auto&& opDebug = CurOp::get(opCtx)->debug(); - int64_t execTime = opDebug.additiveMetrics.executionTime.value_or(Microseconds{0}).count(); - query_stats::writeQueryStats(opCtx, - opDebug.queryStatsInfo.keyHash, - std::move(key), - execTime, - execTime, - opDebug.additiveMetrics.nreturned.value_or(0)); -} - -void collectQueryStatsMongos(OperationContext* opCtx, ClusterClientCursorGuard& cursor) { - cursor->incrementCursorMetrics(CurOp::get(opCtx)->debug().additiveMetrics); - - // For a change stream query that never ends, we want to collect query stats on the initial - // query and each getMore. Here we record the initial query. - // TODO SERVER-89058 Modify comment to include tailable cursors. - if (cursor->getQueryStatsWillNeverExhaust()) { - auto& opDebug = CurOp::get(opCtx)->debug(); - - int64_t execTime = opDebug.additiveMetrics.executionTime.value_or(Microseconds{0}).count(); - - query_stats::writeQueryStats(opCtx, - opDebug.queryStatsInfo.keyHash, - cursor->takeKey(), - execTime, - execTime, - opDebug.additiveMetrics.nreturned.value_or(0), - cursor->getQueryStatsWillNeverExhaust()); - } -} - -void collectQueryStatsMongos(OperationContext* opCtx, ClusterCursorManager::PinnedCursor& cursor) { - cursor->incrementCursorMetrics(CurOp::get(opCtx)->debug().additiveMetrics); - - // For a change stream query that never ends, we want to update query stats for every getMore on - // the cursor. - // TODO SERVER-89058 Modify comment to include tailable cursors. - if (cursor->getQueryStatsWillNeverExhaust()) { - auto& opDebug = CurOp::get(opCtx)->debug(); - - int64_t execTime = opDebug.additiveMetrics.executionTime.value_or(Microseconds{0}).count(); - - query_stats::writeQueryStats(opCtx, - opDebug.queryStatsInfo.keyHash, - nullptr, - execTime, - execTime, - opDebug.additiveMetrics.nreturned.value_or(0), - cursor->getQueryStatsWillNeverExhaust()); - } -} - } // namespace mongo diff --git a/src/mongo/s/query/cluster_cursor_manager.h b/src/mongo/s/query/cluster_cursor_manager.h index 73d07d91476..be10b0d60bd 100644 --- a/src/mongo/s/query/cluster_cursor_manager.h +++ b/src/mongo/s/query/cluster_cursor_manager.h @@ -599,19 +599,4 @@ private: size_t _cursorsTimedOut = 0; }; -/** - * Record metrics for the current operation on opDebug and aggregates those metrics for queryStats - * use. If a cursor is provided (via ClusterClientCursorGuard or - * ClusterCursorManager::PinnedCursor), metrics are aggregated on the cursor; otherwise, metrics are - * written directly to the queryStats store. - * NOTE: Metrics are taken from opDebug.additiveMetrics, so CurOp::setEndOfOpMetrics must be called - * *prior* to calling these. - * - * Currently, queryStats is only collected for find and aggregate requests (and their subsequent - * getMore requests), so these should only be called from those request paths. - */ -void collectQueryStatsMongos(OperationContext* opCtx, std::unique_ptr<query_stats::Key> key); -void collectQueryStatsMongos(OperationContext* opCtx, ClusterClientCursorGuard& cursor); -void collectQueryStatsMongos(OperationContext* opCtx, ClusterCursorManager::PinnedCursor& cursor); - } // namespace mongo diff --git a/src/mongo/s/query/cluster_find.cpp b/src/mongo/s/query/cluster_find.cpp index 01c78f9c3c5..bd186639684 100644 --- a/src/mongo/s/query/cluster_find.cpp +++ b/src/mongo/s/query/cluster_find.cpp @@ -33,7 +33,6 @@ #include "mongo/s/query/cluster_find.h" -#include "mongo/db/query/query_stats/query_stats.h" #include <fmt/format.h> #include <memory> @@ -55,7 +54,6 @@ #include "mongo/db/query/find_common.h" #include "mongo/db/query/getmore_command_gen.h" #include "mongo/db/query/query_planner_common.h" -#include "mongo/db/query/query_stats/query_stats.h" #include "mongo/executor/task_executor_pool.h" #include "mongo/logv2/log.h" #include "mongo/platform/overflow_arithmetic.h" @@ -63,7 +61,6 @@ #include "mongo/s/client/num_hosts_targeted_metrics.h" #include "mongo/s/client/shard_registry.h" #include "mongo/s/cluster_commands_helpers.h" -#include "mongo/s/collection_uuid_mismatch.h" #include "mongo/s/grid.h" #include "mongo/s/query/async_results_merger.h" #include "mongo/s/query/cluster_client_cursor_impl.h" @@ -85,6 +82,11 @@ static const BSONObj kSortKeyMetaProjection = BSON("$meta" << "sortKey"); static const BSONObj kGeoNearDistanceMetaProjection = BSON("$meta" << "geoNearDistance"); +// We must allow some amount of overhead per result document, since when we make a cursor response +// the documents are elements of a BSONArray. The overhead is 1 byte/doc for the type + 1 byte/doc +// for the field name's null terminator + 1 byte per digit in the array index. The index can be no +// more than 8 decimal digits since the response is at most 16MB, and 16 * 1024 * 1024 < 1 * 10^8. +static const int kPerDocumentOverheadBytesUpperBound = 10; const char kFindCmdName[] = "find"; @@ -301,9 +303,12 @@ CursorId runQueryWithoutRetrying(OperationContext* opCtx, if (ex.code() == ErrorCodes::CollectionUUIDMismatch && !ex.extraInfo<CollectionUUIDMismatchInfo>()->actualCollection() && !shardIds.count(cm.dbPrimary())) { - // We received CollectionUUIDMismatch but it does not contain the actual namespace, and - // we did not attempt to establish a cursor on the primary shard. - uassertStatusOK(populateCollectionUUIDMismatch(opCtx, ex.toStatus())); + // We received CollectionUUIDMismatchInfo but it does not contain the actual + // namespace, and we did not attempt to establish a cursor on the primary shard. + // Attempt to do so now in case the collection corresponding to the provided UUID is + // unsharded. This should throw CollectionUUIDMismatchInfo, StaleShardVersion, or + // StaleDbVersion. + establishCursorsOnShards({cm.dbPrimary()}); MONGO_UNREACHABLE; } @@ -336,7 +341,7 @@ CursorId runQueryWithoutRetrying(OperationContext* opCtx, FindCommon::waitInFindBeforeMakingBatch(opCtx, query); auto cursorState = ClusterCursorManager::CursorState::NotExhausted; - FindCommon::BSONArrayResponseSizeTracker responseSizeTracker; + size_t bytesBuffered = 0; // This loop will not result in actually calling getMore against shards, but just loading // results from the initial batches (that were obtained while establishing cursors) into @@ -359,13 +364,14 @@ CursorId runQueryWithoutRetrying(OperationContext* opCtx, // If adding this object will cause us to exceed the message size limit, then we stash it // for later. - if (!responseSizeTracker.haveSpaceForNext(nextObj)) { + if (!FindCommon::haveSpaceForNext(nextObj, results->size(), bytesBuffered)) { ccc->queueResult(nextObj); break; } - // Add doc to the batch. - responseSizeTracker.add(nextObj); + // Add doc to the batch. Account for the space overhead associated with returning this doc + // inside a BSON array. + bytesBuffered += (nextObj.objsize() + kPerDocumentOverheadBytesUpperBound); results->push_back(std::move(nextObj)); } @@ -375,26 +381,23 @@ CursorId runQueryWithoutRetrying(OperationContext* opCtx, cursorState = ClusterCursorManager::CursorState::Exhausted; } - auto&& opDebug = CurOp::get(opCtx)->debug(); // Fill out query exec properties. - opDebug.nShards = ccc->getNumRemotes(); - opDebug.additiveMetrics.nBatches = 1; + CurOp::get(opCtx)->debug().nShards = ccc->getNumRemotes(); + CurOp::get(opCtx)->debug().nreturned = results->size(); // If the caller wants to know whether the cursor returned partial results, set it here. if (partialResultsReturned) { *partialResultsReturned = ccc->partialResultsReturned(); } - CurOp::get(opCtx)->setEndOfOpMetrics(results->size()); // If the cursor is exhausted, then there are no more results to return and we don't need to // allocate a cursor id. if (cursorState == ClusterCursorManager::CursorState::Exhausted) { - opDebug.cursorExhausted = true; + CurOp::get(opCtx)->debug().cursorExhausted = true; if (shardIds.size() > 0) { updateNumHostsTargetedMetrics(opCtx, cm, shardIds.size()); } - collectQueryStatsMongos(opCtx, ccc->takeKey()); return CursorId(0); } @@ -405,13 +408,13 @@ CursorId runQueryWithoutRetrying(OperationContext* opCtx, ? ClusterCursorManager::CursorLifetime::Immortal : ClusterCursorManager::CursorLifetime::Mortal; auto authUsers = AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames(); - collectQueryStatsMongos(opCtx, ccc); + ccc->incNBatches(); auto cursorId = uassertStatusOK(cursorManager->registerCursor( opCtx, ccc.releaseCursor(), query.nss(), cursorType, cursorLifetime, authUsers)); // Record the cursorID in CurOp. - opDebug.cursorid = cursorId; + CurOp::get(opCtx)->debug().cursorid = cursorId; if (shardIds.size() > 0) { updateNumHostsTargetedMetrics(opCtx, cm, shardIds.size()); @@ -471,19 +474,6 @@ Status setUpOperationContextStateForGetMore(OperationContext* opCtx, return Status::OK(); } -CursorId earlyExitWithNoResults(OperationContext* opCtx, - const CanonicalQuery& query, - const FindCommandRequest& findCommand) { - uassert(CollectionUUIDMismatchInfo(query.nss().db().toString(), - *findCommand.getCollectionUUID(), - query.nss().coll().toString(), - boost::none), - "Database does not exist", - !findCommand.getCollectionUUID()); - collectQueryStatsMongos(opCtx, std::move(CurOp::get(opCtx)->debug().queryStatsInfo.key)); - - return CursorId(0); -} } // namespace const size_t ClusterFind::kMaxRetries = 10; @@ -526,7 +516,7 @@ CursorId ClusterFind::runQuery(OperationContext* opCtx, if (swCM == ErrorCodes::NamespaceNotFound) { // If the database doesn't exist, we successfully return an empty result set without // creating a cursor. - return earlyExitWithNoResults(opCtx, query, findCommand); + return CursorId(0); } const auto cm = uassertStatusOK(std::move(swCM)); @@ -773,7 +763,7 @@ StatusWith<CursorResponse> ClusterFind::runGetMore(OperationContext* opCtx, } std::vector<BSONObj> batch; - FindCommon::BSONArrayResponseSizeTracker responseSizeTracker; + size_t bytesBuffered = 0; long long batchSize = cmd.getBatchSize().value_or(0); auto cursorState = ClusterCursorManager::CursorState::NotExhausted; BSONObj postBatchResumeToken; @@ -826,7 +816,8 @@ StatusWith<CursorResponse> ClusterFind::runGetMore(OperationContext* opCtx, break; } - if (!responseSizeTracker.haveSpaceForNext(*next.getValue().getResult())) { + if (!FindCommon::haveSpaceForNext( + *next.getValue().getResult(), batch.size(), bytesBuffered)) { pinnedCursor.getValue()->queueResult(*next.getValue().getResult()); stashedResult = true; break; @@ -835,8 +826,10 @@ StatusWith<CursorResponse> ClusterFind::runGetMore(OperationContext* opCtx, // As soon as we get a result, this operation no longer waits. awaitDataState(opCtx).shouldWaitForInserts = false; - // Add doc to the batch. - responseSizeTracker.add(*next.getValue().getResult()); + // Add doc to the batch. Account for the space overhead associated with returning this doc + // inside a BSON array. + bytesBuffered += + (next.getValue().getResult()->objsize() + kPerDocumentOverheadBytesUpperBound); batch.push_back(std::move(*next.getValue().getResult())); // Update the postBatchResumeToken. For non-$changeStream aggregations, this will be empty. @@ -853,20 +846,17 @@ StatusWith<CursorResponse> ClusterFind::runGetMore(OperationContext* opCtx, postBatchResumeToken = pinnedCursor.getValue()->getPostBatchResumeToken(); } - auto&& opDebug = CurOp::get(opCtx)->debug(); - // Set nReturned and whether the cursor has been exhausted. - opDebug.cursorExhausted = (idToReturn == 0); - opDebug.additiveMetrics.nBatches = 1; - CurOp::get(opCtx)->setEndOfOpMetrics(batch.size()); - const bool partialResultsReturned = pinnedCursor.getValue()->partialResultsReturned(); pinnedCursor.getValue()->setLeftoverMaxTimeMicros(opCtx->getRemainingMaxTimeMicros()); - collectQueryStatsMongos(opCtx, pinnedCursor.getValue()); - + pinnedCursor.getValue()->incNBatches(); // Upon successful completion, transfer ownership of the cursor back to the cursor manager. If // the cursor has been exhausted, the cursor manager will clean it up for us. pinnedCursor.getValue().returnCursor(cursorState); + // Set nReturned and whether the cursor has been exhausted. + CurOp::get(opCtx)->debug().cursorExhausted = (idToReturn == 0); + CurOp::get(opCtx)->debug().nreturned = batch.size(); + if (MONGO_unlikely(waitBeforeUnpinningOrDeletingCursorAfterGetMoreBatch.shouldFail())) { CurOpFailpointHelpers::waitWhileFailPointEnabled( &waitBeforeUnpinningOrDeletingCursorAfterGetMoreBatch, diff --git a/src/mongo/s/query/document_source_merge_cursors.cpp b/src/mongo/s/query/document_source_merge_cursors.cpp index 34f6b1f11e6..c6f8f3fbaf6 100644 --- a/src/mongo/s/query/document_source_merge_cursors.cpp +++ b/src/mongo/s/query/document_source_merge_cursors.cpp @@ -52,6 +52,7 @@ DocumentSourceMergeCursors::DocumentSourceMergeCursors( : DocumentSource(kStageName, expCtx), _armParamsObj(std::move(ownedParamsSpec)), _armParams(std::move(armParams)) { + _armParams->setRecordRemoteOpWaitTime(true); // Populate the shard ids from the 'RemoteCursor'. recordRemoteCursorShardIds(_armParams->getRemotes()); @@ -82,6 +83,7 @@ bool DocumentSourceMergeCursors::remotesExhausted() const { void DocumentSourceMergeCursors::populateMerger() { invariant(!_blockingResultsMerger); invariant(_armParams); + invariant(_armParams->getRecordRemoteOpWaitTime()); _blockingResultsMerger.emplace( pExpCtx->opCtx, @@ -114,10 +116,11 @@ DocumentSource::GetNextResult DocumentSourceMergeCursors::doGetNext() { return Document::fromBsonWithMetaData(*next.getResult()); } -Value DocumentSourceMergeCursors::serialize(const SerializationOptions& opts) const { +Value DocumentSourceMergeCursors::serialize( + boost::optional<ExplainOptions::Verbosity> explain) const { invariant(!_blockingResultsMerger); invariant(_armParams); - return Value(Document{{kStageName, _armParams->toBSON(opts)}}); + return Value(Document{{kStageName, _armParams->toBSON()}}); } boost::intrusive_ptr<DocumentSource> DocumentSourceMergeCursors::createFromBson( diff --git a/src/mongo/s/query/document_source_merge_cursors.h b/src/mongo/s/query/document_source_merge_cursors.h index 4eb75bd34bd..33050bf45ab 100644 --- a/src/mongo/s/query/document_source_merge_cursors.h +++ b/src/mongo/s/query/document_source_merge_cursors.h @@ -80,7 +80,7 @@ public: /** * Serializes this stage to be sent to perform the merging on a different host. */ - Value serialize(const SerializationOptions& opts = SerializationOptions{}) const final override; + Value serialize(boost::optional<ExplainOptions::Verbosity> explain = boost::none) const final; StageConstraints constraints(Pipeline::SplitState pipeState) const final { StageConstraints constraints(StreamType::kStreaming, diff --git a/src/mongo/s/query/establish_cursors.cpp b/src/mongo/s/query/establish_cursors.cpp index 18196d76751..82ec1df2809 100644 --- a/src/mongo/s/query/establish_cursors.cpp +++ b/src/mongo/s/query/establish_cursors.cpp @@ -42,11 +42,8 @@ #include "mongo/db/cursor_id.h" #include "mongo/db/query/cursor_response.h" #include "mongo/db/query/kill_cursors_gen.h" -#include "mongo/db/query/query_knobs_gen.h" -#include "mongo/executor/async_multicaster.h" #include "mongo/executor/remote_command_request.h" #include "mongo/executor/remote_command_response.h" -#include "mongo/executor/task_executor.h" #include "mongo/logv2/log.h" #include "mongo/s/grid.h" #include "mongo/s/multi_statement_transaction_requests_sender.h" @@ -105,13 +102,12 @@ public: return std::exchange(_remoteCursors, {}); }; - static void killOpOnShards(ServiceContext* srvCtx, - std::shared_ptr<executor::TaskExecutor> executor, - OperationKey opKey, - std::set<HostAndPort> remotes) noexcept; - private: void _handleFailure(const AsyncRequestsSender::Response& response, Status status) noexcept; + static void _killOpOnShards(ServiceContext* srvCtx, + std::shared_ptr<executor::TaskExecutor> executor, + OperationKey opKey, + std::set<HostAndPort> remotes) noexcept; /** * Favors the status with 'CollectionUUIDMismatch' error to be saved in '_maybeFailure' to be @@ -133,26 +129,22 @@ private: std::vector<HostAndPort> _remotesToClean; }; -// Attach our OperationKey to a request. This will allow us to kill any outstanding -// requests in case we're interrupted or one of the remotes returns an error. Note that although -// the opCtx may have an OperationKey set on it already, do not inherit it here because we may -// target ourselves which implies the same node receiving multiple operations with the same -// opKey. -BSONObj appendOpKey(const OperationKey& opKey, const BSONObj& request) { - BSONObjBuilder newCmd(request); - opKey.appendToBuilder(&newCmd, "clientOperationKey"); - return newCmd.obj(); -} - void CursorEstablisher::sendRequests(const ReadPreferenceSetting& readPref, const std::vector<std::pair<ShardId, BSONObj>>& remotes, Shard::RetryPolicy retryPolicy) { // Construct the requests std::vector<AsyncRequestsSender::Request> requests; + // Attach our OperationKey to each remote request. This will allow us to kill any outstanding + // requests in case we're interrupted or one of the remotes returns an error. Note that although + // the opCtx may have an OperationKey set on it already, do not inherit it here because we may + // target ourselves which implies the same node receiving multiple operations with the same + // opKey. // TODO SERVER-47261 management of the opKey should move to the ARS. for (const auto& remote : remotes) { - requests.emplace_back(remote.first, appendOpKey(_opKey, remote.second)); + BSONObjBuilder requestWithOpKey(remote.second); + _opKey.appendToBuilder(&requestWithOpKey, "clientOperationKey"); + requests.emplace_back(remote.first, requestWithOpKey.obj()); } LOGV2_DEBUG(4625502, @@ -190,9 +182,11 @@ void CursorEstablisher::waitForResponse() noexcept { hadValidCursor = true; - _remoteCursors.emplace_back(RemoteCursor(response.shardId.toString(), - *response.shardHostAndPort, - std::move(cursor.getValue()))); + RemoteCursor remoteCursor; + remoteCursor.setCursorResponse(std::move(cursor.getValue())); + remoteCursor.setShardId(response.shardId); + remoteCursor.setHostAndPort(*response.shardHostAndPort); + _remoteCursors.emplace_back(std::move(remoteCursor)); } if (response.shardHostAndPort && !hadValidCursor) { @@ -205,41 +199,16 @@ void CursorEstablisher::waitForResponse() noexcept { } } -// Schedule killOperations against all cursors that were established. Make sure to -// capture arguments by value since the cleanup work may get scheduled after -// returning from this function. -StatusWith<executor::TaskExecutor::CallbackHandle> scheduleCursorCleanup( - std::shared_ptr<executor::TaskExecutor> executor, - ServiceContext* svcCtx, - OperationKey opKey, - std::set<HostAndPort>&& remotesToClean) { - return executor->scheduleWork([svcCtx = svcCtx, - executor = executor, - opKey = opKey, - remotesToClean = std::move(remotesToClean)]( - const executor::TaskExecutor::CallbackArgs& args) mutable { - if (!args.status.isOK()) { - LOGV2_WARNING( - 7355702, "Failed to schedule remote cursor cleanup", "error"_attr = args.status); - return; - } - CursorEstablisher::killOpOnShards( - svcCtx, std::move(executor), std::move(opKey), std::move(remotesToClean)); - }); -} - void CursorEstablisher::checkForFailedRequests() { if (!_maybeFailure) { // If we saw no failures, there is nothing to do. return; } - if (!(_maybeFailure->code() == ErrorCodes::CommandOnShardedViewNotSupportedOnMongod)) { - LOGV2(4625501, - "Unable to establish remote cursors", - "error"_attr = *_maybeFailure, - "nRemotes"_attr = _remotesToClean.size()); - } + LOGV2(4625501, + "Unable to establish remote cursors", + "error"_attr = *_maybeFailure, + "nRemotes"_attr = _remotesToClean.size()); if (_remotesToClean.empty()) { // If we don't have any remotes to clean, throw early. @@ -249,8 +218,21 @@ void CursorEstablisher::checkForFailedRequests() { // Filter out duplicate hosts. auto remotes = std::set<HostAndPort>(_remotesToClean.begin(), _remotesToClean.end()); - uassertStatusOK( - scheduleCursorCleanup(_executor, _opCtx->getServiceContext(), _opKey, std::move(remotes))); + // Schedule killOperations against all cursors that were established. Make sure to + // capture arguments by value since the cleanup work may get scheduled after + // returning from this function. + uassertStatusOK(_executor->scheduleWork( + [svcCtx = _opCtx->getServiceContext(), + executor = _executor, + opKey = _opKey, + remotes = std::move(remotes)](const executor::TaskExecutor::CallbackArgs& args) mutable { + if (!args.status.isOK()) { + LOGV2_WARNING( + 48038, "Failed to schedule remote cursor cleanup", "error"_attr = args.status); + return; + } + _killOpOnShards(svcCtx, std::move(executor), std::move(opKey), std::move(remotes)); + })); // Throw our failure. uassertStatusOK(*_maybeFailure); @@ -312,10 +294,10 @@ void CursorEstablisher::_handleFailure(const AsyncRequestsSender::Response& resp _maybeFailure = std::move(status); } -void CursorEstablisher::killOpOnShards(ServiceContext* srvCtx, - std::shared_ptr<executor::TaskExecutor> executor, - OperationKey opKey, - std::set<HostAndPort> remotes) noexcept try { +void CursorEstablisher::_killOpOnShards(ServiceContext* srvCtx, + std::shared_ptr<executor::TaskExecutor> executor, + OperationKey opKey, + std::set<HostAndPort> remotes) noexcept try { ThreadClient tc("establishCursors cleanup", srvCtx); auto opCtx = tc->makeOperationContext(); @@ -345,16 +327,6 @@ void CursorEstablisher::killOpOnShards(ServiceContext* srvCtx, } catch (const AssertionException& ex) { LOGV2_DEBUG(4625503, 2, "Failed to cleanup remote operations", "error"_attr = ex.toStatus()); } -/** - * Returns a copy of 'cmdObj' with the $readPreference mode set to secondaryPreferred. - */ -BSONObj appendReadPreferenceNearest(BSONObj cmdObj) { - BSONObjBuilder cmdWithReadPrefBob(std::move(cmdObj)); - cmdWithReadPrefBob.append("$readPreference", - BSON("mode" - << "nearest")); - return cmdWithReadPrefBob.obj(); -} } // namespace @@ -386,101 +358,4 @@ void killRemoteCursor(OperationContext* opCtx, executor->scheduleRemoteCommand(request, [](auto const&) {}).getStatus().ignore(); } -std::pair<std::vector<HostAndPort>, StringMap<ShardId>> getHostInfos( - OperationContext* opCtx, const std::set<ShardId>& shardIds) { - std::vector<HostAndPort> servers; - StringMap<ShardId> hostToShardId; - - // Get the host/port of every node in each shard. - auto registry = Grid::get(opCtx)->shardRegistry(); - for (const auto& shardId : shardIds) { - auto shard = uassertStatusOK(registry->getShard(opCtx, shardId)); - auto cs = shard->getConnString(); - auto shardServers = cs.getServers(); - for (auto& host : shardServers) { - hostToShardId.emplace(host.toString(), shardId); - } - servers.insert(servers.end(), shardServers.begin(), shardServers.end()); - } - return {std::move(servers), hostToShardId}; -} - -std::vector<RemoteCursor> establishCursorsOnAllHosts( - OperationContext* opCtx, - std::shared_ptr<executor::TaskExecutor> executor, - const NamespaceString& nss, - const std::set<ShardId>& shardIds, - BSONObj cmdObj, - bool allowPartialResults, - Shard::RetryPolicy retryPolicy) { - auto [servers, hostToShardId] = getHostInfos(opCtx, shardIds); - OperationKey opKey = UUID::gen(); - - // Operation key will allow us to kill any outstanding requests in case we're interrupted. - // Secondaries will reject aggregation commands with a default read preference (primary). The - // actual semantics of read preference don't make as much sense when broadcasting to all - // shards, but we will set read preference to 'nearest' since it does not imply preference for - // primary or secondary. - BSONObj cmd = appendOpKey(opKey, appendReadPreferenceNearest(cmdObj)); - - executor::AsyncMulticaster::Options options; - options.maxConcurrency = internalQueryAggMulticastMaxConcurrency; - auto results = executor::AsyncMulticaster(executor, options) - .multicast(servers, - nss.db().toString(), - cmd, - opCtx, - Milliseconds(internalQueryAggMulticastTimeoutMS)); - std::vector<RemoteCursor> remoteCursors; - std::set<HostAndPort> remotesToClean; - - boost::optional<Status> failure; - - for (auto&& [hostAndPort, result] : results) { - if (result.isOK()) { - auto cursors = CursorResponse::parseFromBSONMany(result.data); - bool hadValidCursor = false; - - auto it = hostToShardId.find(hostAndPort.toString()); - tassert(7355701, "Host must have shard ID.", it != hostToShardId.end()); - auto shardId = it->second; - - for (auto& cursor : cursors) { - if (!cursor.isOK()) { - failure = cursor.getStatus(); - continue; - } - hadValidCursor = true; - - remoteCursors.emplace_back( - RemoteCursor(shardId.toString(), hostAndPort, std::move(cursor.getValue()))); - } - - if (hadValidCursor) { - remotesToClean.insert(hostAndPort); - } - } else { - LOGV2_DEBUG(7355700, - 3, - "Experienced a failure while establishing cursors", - "error"_attr = result.status); - failure = result.status; - } - } - if (failure.has_value() && !allowPartialResults) { - LOGV2(7355705, - "Unable to establish remote cursors", - "error"_attr = *failure, - "nRemotes"_attr = remoteCursors.size()); - - if (!remotesToClean.empty()) { - uassertStatusOK(scheduleCursorCleanup( - executor, opCtx->getServiceContext(), opKey, std::move(remotesToClean))); - } - - uassertStatusOK(failure.value()); - } - return remoteCursors; -} - } // namespace mongo diff --git a/src/mongo/s/query/establish_cursors.h b/src/mongo/s/query/establish_cursors.h index cd19af7eea9..3a904adcadd 100644 --- a/src/mongo/s/query/establish_cursors.h +++ b/src/mongo/s/query/establish_cursors.h @@ -73,26 +73,6 @@ std::vector<RemoteCursor> establishCursors( Shard::RetryPolicy retryPolicy = Shard::RetryPolicy::kIdempotent); /** - * Establishes cursors on every host in the remote shards by issuing requests in parallel with the - * AsyncMulticaster. - * - * If any of the cursors fail to be established, this function performs cleanup by sending - * killCursors to any cursors that were established, then throws the error. If the namespace - * represents a view, an exception containing a ResolvedView is thrown. - * - * On success, the ownership of the cursors is transferred to the caller. This means the caller is - * now responsible for either exhausting the cursors or sending killCursors to them. - */ -std::vector<RemoteCursor> establishCursorsOnAllHosts( - OperationContext* opCtx, - std::shared_ptr<executor::TaskExecutor> executor, - const NamespaceString& nss, - const std::set<ShardId>& shardIds, - BSONObj cmdObj, - bool allowPartialResults, - Shard::RetryPolicy retryPolicy = Shard::RetryPolicy::kIdempotent); - -/** * Schedules a remote killCursor command for 'cursor'. * * Note that this method is optimistic and does not check the return status for the killCursors diff --git a/src/mongo/s/query/router_stage_remove_metadata_fields.cpp b/src/mongo/s/query/router_stage_remove_metadata_fields.cpp index 7f3d89beef1..320c441ef46 100644 --- a/src/mongo/s/query/router_stage_remove_metadata_fields.cpp +++ b/src/mongo/s/query/router_stage_remove_metadata_fields.cpp @@ -53,14 +53,9 @@ StatusWith<ClusterQueryResult> RouterStageRemoveMetadataFields::next() { } BSONObjIterator iterator(*childResult.getValue().getResult()); - // Find the first field that we need to remove. - for (; iterator.more(); ++iterator) { - // To save some time, we ensure that the current field name starts with a $ - // before checking if it's actually a metadata field in the map. - if ((*iterator).fieldName()[0] == '$' && _metaFields.contains((*iterator).fieldName())) { - break; - } + while (iterator.more() && (*iterator).fieldName()[0] != '$') { + ++iterator; } if (!iterator.more()) { diff --git a/src/mongo/s/query/router_stage_remove_metadata_fields_test.cpp b/src/mongo/s/query/router_stage_remove_metadata_fields_test.cpp index 9fa199a2ac9..a18aa0cbb31 100644 --- a/src/mongo/s/query/router_stage_remove_metadata_fields_test.cpp +++ b/src/mongo/s/query/router_stage_remove_metadata_fields_test.cpp @@ -194,60 +194,6 @@ TEST(RouterStageRemoveMetadataFieldsTest, ForwardsAwaitDataTimeout) { ASSERT_EQ(789, durationCount<Milliseconds>(awaitDataTimeout.getValue())); } -// Grabs the next document from a stage and ensure it matches expectedDoc. -// Assumes that remotes are exhausted after one document. -void verifyNextDocument(RouterExecStage* stage, const BSONObj& expectedDoc) { - auto result = stage->next(); - ASSERT_OK(result.getStatus()); - ASSERT(result.getValue().getResult()); - ASSERT_BSONOBJ_EQ(*result.getValue().getResult(), expectedDoc); - ASSERT_TRUE(stage->remotesExhausted()); -} - -TEST(RouterStageRemoveMetadataFieldsTest, AllowsNonMetaDataDollars) { - auto mockStage = std::make_unique<RouterStageMock>(opCtx); - mockStage->queueResult(BSON("$a" << 1 << "$sortKey" << 1 << "b" << 1)); - mockStage->queueResult(BSON("a" << 2 << "$sortKey" << 1 << "$b" << 2)); - mockStage->markRemotesExhausted(); - - auto sortKeyStage = std::make_unique<RouterStageRemoveMetadataFields>( - opCtx, std::move(mockStage), StringDataSet{"$sortKey"_sd}); - ASSERT_TRUE(sortKeyStage->remotesExhausted()); - - verifyNextDocument(sortKeyStage.get(), BSON("$a" << 1 << "b" << 1)); - verifyNextDocument(sortKeyStage.get(), BSON("a" << 2 << "$b" << 2)); - - auto endResult = sortKeyStage->next(); - ASSERT_OK(endResult.getStatus()); - ASSERT(endResult.getValue().isEOF()); - ASSERT_TRUE(sortKeyStage->remotesExhausted()); -} - -// For every keyword, ensure it's removed if it's in the first, middle, or -// last position, and that the remainder of the document is undisturbed. -TEST(RouterStageRemoveMetadataFieldsTest, RemovesAllMetaDataDollars) { - for (auto& keyword : Document::allMetadataFieldNames) { - auto mockStage = std::make_unique<RouterStageMock>(opCtx); - mockStage->queueResult(BSON(keyword << 1 << "$a" << 1)); - mockStage->queueResult(BSON("$a" << 1 << keyword << 1)); - mockStage->queueResult(BSON("$a" << 1 << keyword << 1 << "$b" << 1)); - mockStage->markRemotesExhausted(); - - auto sortKeyStage = std::make_unique<RouterStageRemoveMetadataFields>( - opCtx, std::move(mockStage), Document::allMetadataFieldNames); - ASSERT_TRUE(sortKeyStage->remotesExhausted()); - - verifyNextDocument(sortKeyStage.get(), BSON("$a" << 1)); - verifyNextDocument(sortKeyStage.get(), BSON("$a" << 1)); - verifyNextDocument(sortKeyStage.get(), BSON("$a" << 1 << "$b" << 1)); - - auto endResult = sortKeyStage->next(); - ASSERT_OK(endResult.getStatus()); - ASSERT(endResult.getValue().isEOF()); - ASSERT_TRUE(sortKeyStage->remotesExhausted()); - } -} - } // namespace } // namespace mongo diff --git a/src/mongo/s/query/store_possible_cursor.cpp b/src/mongo/s/query/store_possible_cursor.cpp index 723cafff2e4..c778daa9d84 100644 --- a/src/mongo/s/query/store_possible_cursor.cpp +++ b/src/mongo/s/query/store_possible_cursor.cpp @@ -88,17 +88,15 @@ StatusWith<BSONObj> storePossibleCursor(OperationContext* opCtx, return incomingCursorResponse.getStatus(); } - auto&& opDebug = CurOp::get(opCtx)->debug(); - opDebug.additiveMetrics.nBatches = 1; + CurOp::get(opCtx)->debug().nreturned = incomingCursorResponse.getValue().getBatch().size(); + // If nShards has already been set, then we are storing the forwarding $mergeCursors cursor from // a split aggregation pipeline, and the shards half of that pipeline may have targeted multiple // shards. In that case, leave the current value as-is. - opDebug.nShards = std::max(opDebug.nShards, 1); - CurOp::get(opCtx)->setEndOfOpMetrics(incomingCursorResponse.getValue().getBatch().size()); + CurOp::get(opCtx)->debug().nShards = std::max(CurOp::get(opCtx)->debug().nShards, 1); if (incomingCursorResponse.getValue().getCursorId() == CursorId(0)) { - opDebug.cursorExhausted = true; - collectQueryStatsMongos(opCtx, std::move(opDebug.queryStatsInfo.key)); + CurOp::get(opCtx)->debug().cursorExhausted = true; return cmdResult; } @@ -130,7 +128,7 @@ StatusWith<BSONObj> storePossibleCursor(OperationContext* opCtx, } auto ccc = ClusterClientCursorImpl::make(opCtx, std::move(executor), std::move(params)); - collectQueryStatsMongos(opCtx, ccc); + ccc->incNBatches(); // We don't expect to use this cursor until a subsequent getMore, so detach from the current // OperationContext until then. ccc->detachFromOperationContext(); @@ -146,7 +144,7 @@ StatusWith<BSONObj> storePossibleCursor(OperationContext* opCtx, return clusterCursorId.getStatus(); } - opDebug.cursorid = clusterCursorId.getValue(); + CurOp::get(opCtx)->debug().cursorid = clusterCursorId.getValue(); CursorResponse outgoingCursorResponse( requestedNss, |