Update upstream source from tag 'upstream/6.0.0'master

Update to upstream version '6.0.0'
with Debian dir 5604a80ec1c96ca76f25f40d78e6ef855abec322

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,