New upstream version 6.0.20upstream

23 files changed, 669 insertions, 196 deletions

diff --git a/src/mongo/s/query/SConscript b/src/mongo/s/query/SConscript
index 94bfbde4878..f8179e37eeb 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',
+        "cluster_query_knobs.idl",
+        "store_possible_cursor.cpp",
     ],
     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,6 +37,7 @@ 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',
@@ -93,24 +94,11 @@ 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",
@@ -119,12 +107,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',
     ],
 )
 
@@ -171,7 +159,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..3ea786af9ae 100644
--- a/src/mongo/s/query/async_results_merger.cpp
+++ b/src/mongo/s/query/async_results_merger.cpp
@@ -161,6 +161,10 @@ AsyncResultsMerger::~AsyncResultsMerger() {
     invariant(_remotesExhausted(lk) || _lifecycleState == kKillComplete);
 }
 
+const AsyncResultsMergerParams& AsyncResultsMerger::params() const {
+    return _params;
+}
+
 bool AsyncResultsMerger::remotesExhausted() const {
     stdx::lock_guard<Latch> lk(_mutex);
     return _remotesExhausted(lk);
diff --git a/src/mongo/s/query/async_results_merger.h b/src/mongo/s/query/async_results_merger.h
index 3fde29e141e..518a129b0ec 100644
--- a/src/mongo/s/query/async_results_merger.h
+++ b/src/mongo/s/query/async_results_merger.h
@@ -109,6 +109,11 @@ public:
     ~AsyncResultsMerger();
 
     /**
+     * Returns a const reference to the parameters.
+     */
+    const AsyncResultsMergerParams& params() const;
+
+    /**
      * Returns true if all of the remote cursors are exhausted.
      */
     bool remotesExhausted() const;
@@ -485,7 +490,7 @@ private:
     OperationContext* _opCtx;
     std::shared_ptr<executor::TaskExecutor> _executor;
     TailableModeEnum _tailableMode;
-    AsyncResultsMergerParams _params;
+    const AsyncResultsMergerParams _params;
 
     // Must be acquired before accessing any data members (other than _params, which is read-only).
     mutable Mutex _mutex = MONGO_MAKE_LATCH("AsyncResultsMerger::_mutex");
diff --git a/src/mongo/s/query/async_results_merger_params.idl b/src/mongo/s/query/async_results_merger_params.idl
index e3c4d03bdd3..5382c9cc718 100644
--- a/src/mongo/s/query/async_results_merger_params.idl
+++ b/src/mongo/s/query/async_results_merger_params.idl
@@ -50,15 +50,19 @@ 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.
 
@@ -66,35 +70,46 @@ 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: array<RemoteCursor>
+            remotes:
+                type: array<RemoteCursor>
+                query_shape: literal
             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.
-            nss: namespacestring
+                query_shape: literal
+            nss:
+                type: namespacestring
+                query_shape: custom
             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.
diff --git a/src/mongo/s/query/async_results_merger_test.cpp b/src/mongo/s/query/async_results_merger_test.cpp
index 80600272e63..cc1e1e473f3 100644
--- a/src/mongo/s/query/async_results_merger_test.cpp
+++ b/src/mongo/s/query/async_results_merger_test.cpp
@@ -2022,5 +2022,27 @@ TEST_F(AsyncResultsMergerTest, ShouldNotScheduleGetMoresWithoutAnOperationContex
     killFuture.wait();
 }
 
+TEST_F(AsyncResultsMergerTest, CanAccessParams) {
+    std::vector<RemoteCursor> cursors;
+    cursors.push_back(
+        makeRemoteCursor(kTestShardIds[0], kTestShardHosts[0], CursorResponse(kTestNss, 5, {})));
+    auto arm = makeARMFromExistingCursors(std::move(cursors));
+
+    // Check actual parameters.
+    ASSERT_EQ(kTestNss, arm->params().getNss());
+    ASSERT_EQ(1, arm->params().getRemotes().size());
+
+    // Schedule requests. We need to do this because the dtor of AsyncResultsMerger fires an
+    // assertion if the remotes are not exhausted and the AsyncResultsMerger hasn't been killed.
+    auto readyEvent = unittest::assertGet(arm->nextEvent());
+    std::vector<CursorResponse> responses;
+    std::vector<BSONObj> batch = {fromjson("{_id: 1}"), fromjson("{_id: 2}"), fromjson("{_id: 3}")};
+    responses.emplace_back(kTestNss, CursorId(0), batch);
+    scheduleNetworkResponses(std::move(responses));
+
+    // Now the AsyncResultsMerger can go out of scope without triggering the assertion failure.
+    ASSERT_TRUE(arm->remotesExhausted());
+}
+
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/s/query/blocking_results_merger.cpp b/src/mongo/s/query/blocking_results_merger.cpp
index fc56a0d9e3b..7cb19e3eb91 100644
--- a/src/mongo/s/query/blocking_results_merger.cpp
+++ b/src/mongo/s/query/blocking_results_merger.cpp
@@ -46,6 +46,10 @@ BlockingResultsMerger::BlockingResultsMerger(OperationContext* opCtx,
       _arm(opCtx, std::move(executor), std::move(armParams)),
       _resourceYielder(std::move(resourceYielder)) {}
 
+const AsyncResultsMergerParams& BlockingResultsMerger::asyncResultsMergerParams() const {
+    return _arm.params();
+}
+
 StatusWith<stdx::cv_status> BlockingResultsMerger::doWaiting(
     OperationContext* opCtx, const std::function<StatusWith<stdx::cv_status>()>& waitFn) noexcept {
 
diff --git a/src/mongo/s/query/blocking_results_merger.h b/src/mongo/s/query/blocking_results_merger.h
index c05cecc5da8..9c0e78f9ba5 100644
--- a/src/mongo/s/query/blocking_results_merger.h
+++ b/src/mongo/s/query/blocking_results_merger.h
@@ -48,6 +48,11 @@ public:
                           std::unique_ptr<ResourceYielder> resourceYielder);
 
     /**
+     * Returns a const reference to the AsyncResultsMergerParams owned by the AsyncResultsMerger.
+     */
+    const AsyncResultsMergerParams& asyncResultsMergerParams() const;
+
+    /**
      * Blocks until the next result is available or an error is detected.
      */
     StatusWith<ClusterQueryResult> next(OperationContext*);
diff --git a/src/mongo/s/query/blocking_results_merger_test.cpp b/src/mongo/s/query/blocking_results_merger_test.cpp
index 15e37b0460d..41dd31b895e 100644
--- a/src/mongo/s/query/blocking_results_merger_test.cpp
+++ b/src/mongo/s/query/blocking_results_merger_test.cpp
@@ -292,5 +292,20 @@ TEST_F(ResultsMergerTestFixture, ShouldBeAbleToHandleExceptionWhenUnyielding) {
     future.default_timed_get();
 }
 
+TEST_F(ResultsMergerTestFixture, CanAccessAsyncResultsMergerParams) {
+    std::vector<RemoteCursor> cursors;
+    cursors.emplace_back(
+        makeRemoteCursor(kTestShardIds[0], kTestShardHosts[0], CursorResponse(kTestNss, 1, {})));
+    auto params = makeARMParamsFromExistingCursors(std::move(cursors));
+    BlockingResultsMerger blockingMerger(
+        operationContext(), std::move(params), executor(), nullptr);
+
+    ASSERT_EQ(kTestNss, blockingMerger.asyncResultsMergerParams().getNss());
+    ASSERT_EQ(1, blockingMerger.asyncResultsMergerParams().getRemotes().size());
+
+    // Kill merger because otherwise it will run into an assertion in its dtor.
+    blockingMerger.kill(operationContext());
+}
+
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/s/query/cluster_aggregate.cpp b/src/mongo/s/query/cluster_aggregate.cpp
index 3d6e9b5c2af..6374bcfd494 100644
--- a/src/mongo/s/query/cluster_aggregate.cpp
+++ b/src/mongo/s/query/cluster_aggregate.cpp
@@ -27,6 +27,7 @@
  *    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"
@@ -56,6 +57,9 @@
 #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"
@@ -95,7 +99,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(stdx::unordered_set<NamespaceString> involvedNamespaces) {
+auto resolveInvolvedNamespaces(const stdx::unordered_set<NamespaceString>& involvedNamespaces) {
     StringMap<ExpressionContext::ResolvedNamespace> resolvedNamespaces;
     for (auto&& nss : involvedNamespaces) {
         resolvedNamespaces.try_emplace(nss.coll(), nss, std::vector<BSONObj>{});
@@ -258,6 +262,68 @@ 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,
@@ -351,39 +417,15 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx,
 
     boost::intrusive_ptr<ExpressionContext> expCtx;
     const auto pipelineBuilder = [&]() {
-        // 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);
-
-        // 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);
+        auto pipeline = parsePipelineAndRegisterQueryStats(opCtx,
+                                                           involvedNamespaces,
+                                                           namespaces.executionNss,
+                                                           request,
+                                                           cm,
+                                                           liteParsedPipeline,
+                                                           hasChangeStream,
+                                                           shouldDoFLERewrite);
+        expCtx = pipeline->getContext();
 
         // If the aggregate command supports encrypted collections, do rewrites of the pipeline to
         // support querying against encrypted fields.
@@ -429,15 +471,48 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx,
                 cluster_aggregation_planner::AggregationTargeter::TargetingPolicy::kMongosRequired);
 
     if (!expCtx) {
-        // 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.
+        // 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.
         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()) {
@@ -465,10 +540,11 @@ 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(*expCtx->explain)}};
+                                        {"stages", targeter.pipeline->writeExplainOps(opts)}};
                     return Status::OK();
                 }
 
@@ -537,11 +613,12 @@ 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 e124b63c4f9..70adf70404a 100644
--- a/src/mongo/s/query/cluster_aggregation_planner.cpp
+++ b/src/mongo/s/query/cluster_aggregation_planner.cpp
@@ -345,12 +345,27 @@ 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 (cursorState == ClusterCursorManager::CursorState::NotExhausted) {
+    if (!exhausted) {
         auto authUsers = AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames();
         clusterCursorId = uassertStatusOK(Grid::get(opCtx)->getCursorManager()->registerCursor(
             opCtx,
@@ -359,16 +374,9 @@ 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();
@@ -599,12 +607,13 @@ AggregationTargeter AggregationTargeter::make(
     }();
 
     // Determine whether this aggregation must be dispatched to all shards in the cluster.
-    const bool mustRunOnAll =
-        sharded_agg_helpers::mustRunOnAllShards(executionNss, hasChangeStream, startsWithDocuments);
+    const bool mustRunOnAllShards = sharded_agg_helpers::checkIfMustRunOnAllShards(
+        executionNss, hasChangeStream, startsWithDocuments);
 
     // 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 || (mustRunOnAll && hasChangeStream) || (startsWithDocuments && !mustRunOnAll));
+    invariant(cm || (mustRunOnAllShards && hasChangeStream) ||
+              (startsWithDocuments && !mustRunOnAllShards));
 
     // A pipeline is allowed to passthrough to the primary shard iff the following conditions are
     // met:
@@ -616,7 +625,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() && !mustRunOnAll && allowedToPassthrough &&
+    if (cm && !cm->isSharded() && !mustRunOnAllShards && allowedToPassthrough &&
         !involvesShardedCollections) {
         return AggregationTargeter{TargetingPolicy::kPassthrough, nullptr, cm};
     } else {
@@ -858,6 +867,7 @@ 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_client_cursor.h b/src/mongo/s/query/cluster_client_cursor.h
index 8ff611eb308..23a3367416d 100644
--- a/src/mongo/s/query/cluster_client_cursor.h
+++ b/src/mongo/s/query/cluster_client_cursor.h
@@ -211,15 +211,30 @@ 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.
      */
-    virtual std::uint64_t getNBatches() const = 0;
+    std::uint64_t getNBatches() const {
+        return _metrics.nBatches.value_or(0);
+    }
 
     /**
      * Increment the number of batches returned so far by one.
      */
-    virtual void incNBatches() = 0;
+    void incNBatches() {
+        _metrics.incrementNBatches();
+    }
+
+    void incrementCursorMetrics(OpDebug::AdditiveMetrics newMetrics) {
+        _metrics.add(newMetrics);
+        if (!_firstResponseExecutionTime) {
+            _firstResponseExecutionTime = _metrics.executionTime;
+        }
+    }
 
     //
     // maxTimeMS support.
@@ -245,6 +260,20 @@ 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 73be5a7512a..6b094a604f4 100644
--- a/src/mongo/s/query/cluster_client_cursor_impl.cpp
+++ b/src/mongo/s/query/cluster_client_cursor_impl.cpp
@@ -27,6 +27,8 @@
  *    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"
@@ -34,6 +36,8 @@
 #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"
@@ -75,7 +79,10 @@ ClusterClientCursorImpl::ClusterClientCursorImpl(OperationContext* opCtx,
       _opCtx(opCtx),
       _createdDate(opCtx->getServiceContext()->getPreciseClockSource()->now()),
       _lastUseDate(_createdDate),
-      _queryHash(CurOp::get(opCtx)->debug().queryHash) {
+      _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) {
     dassert(!_params.compareWholeSortKeyOnRouter ||
             SimpleBSONObjComparator::kInstance.evaluate(
                 _params.sortToApplyOnRouter == AsyncResultsMerger::kWholeSortKeySortPattern));
@@ -92,7 +99,11 @@ ClusterClientCursorImpl::ClusterClientCursorImpl(OperationContext* opCtx,
       _opCtx(opCtx),
       _createdDate(opCtx->getServiceContext()->getPreciseClockSource()->now()),
       _lastUseDate(_createdDate),
-      _queryHash(CurOp::get(opCtx)->debug().queryHash) {
+      _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)) {
     dassert(!_params.compareWholeSortKeyOnRouter ||
             SimpleBSONObjComparator::kInstance.evaluate(
                 _params.sortToApplyOnRouter == AsyncResultsMerger::kWholeSortKeySortPattern));
@@ -100,7 +111,7 @@ ClusterClientCursorImpl::ClusterClientCursorImpl(OperationContext* opCtx,
 }
 
 ClusterClientCursorImpl::~ClusterClientCursorImpl() {
-    if (_nBatchesReturned > 1)
+    if (_metrics.nBatches && *_metrics.nBatches > 1)
         mongosCursorStatsMoreThanOneBatch.increment();
 }
 
@@ -128,7 +139,25 @@ 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) {
@@ -217,12 +246,12 @@ boost::optional<uint32_t> ClusterClientCursorImpl::getQueryHash() const {
     return _queryHash;
 }
 
-std::uint64_t ClusterClientCursorImpl::getNBatches() const {
-    return _nBatchesReturned;
+boost::optional<std::size_t> ClusterClientCursorImpl::getQueryStatsKeyHash() const {
+    return _queryStatsKeyHash;
 }
 
-void ClusterClientCursorImpl::incNBatches() {
-    ++_nBatchesReturned;
+bool ClusterClientCursorImpl::getQueryStatsWillNeverExhaust() const {
+    return _queryStatsWillNeverExhaust;
 }
 
 APIParameters ClusterClientCursorImpl::getAPIParameters() const {
@@ -265,4 +294,7 @@ 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 2529254cfce..8064a45595b 100644
--- a/src/mongo/s/query/cluster_client_cursor_impl.h
+++ b/src/mongo/s/query/cluster_client_cursor_impl.h
@@ -32,6 +32,7 @@
 #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"
@@ -116,9 +117,11 @@ public:
 
     boost::optional<uint32_t> getQueryHash() const final;
 
-    std::uint64_t getNBatches() const final;
+    boost::optional<std::size_t> getQueryStatsKeyHash() const final;
 
-    void incNBatches() final;
+    bool getQueryStatsWillNeverExhaust() const final;
+
+    std::unique_ptr<query_stats::Key> takeKey() final;
 
 public:
     /**
@@ -175,8 +178,19 @@ private:
     // The hash of the query shape to be used for slow query logging;
     boost::optional<uint32_t> _queryHash;
 
-    // The number of batches returned by this cursor.
-    std::uint64_t _nBatchesReturned = 0;
+    // 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;
 };
 
 }  // 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 567f3450499..103951694a0 100644
--- a/src/mongo/s/query/cluster_client_cursor_mock.cpp
+++ b/src/mongo/s/query/cluster_client_cursor_mock.cpp
@@ -89,14 +89,6 @@ 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;
 }
@@ -113,6 +105,14 @@ 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,4 +168,8 @@ 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 bc2991ecf89..64ec06d750f 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/logical_session_id.h"
+#include "mongo/db/query/query_stats/key.h"
 #include "mongo/s/query/cluster_client_cursor.h"
 
 namespace mongo {
@@ -106,9 +106,9 @@ public:
 
     boost::optional<uint32_t> getQueryHash() const final;
 
-    std::uint64_t getNBatches() const final;
+    boost::optional<std::size_t> getQueryStatsKeyHash() const final;
 
-    void incNBatches() final;
+    bool getQueryStatsWillNeverExhaust() const final;
 
     /**
      * Returns false unless the mock cursor has been fully iterated.
@@ -120,6 +120,8 @@ 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 1209d709e22..4452e6a8811 100644
--- a/src/mongo/s/query/cluster_cursor_manager.cpp
+++ b/src/mongo/s/query/cluster_cursor_manager.cpp
@@ -41,6 +41,7 @@
 #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"
@@ -246,6 +247,7 @@ 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);
 }
@@ -574,4 +576,60 @@ 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 be10b0d60bd..73d07d91476 100644
--- a/src/mongo/s/query/cluster_cursor_manager.h
+++ b/src/mongo/s/query/cluster_cursor_manager.h
@@ -599,4 +599,19 @@ 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 e27d4174e99..01c78f9c3c5 100644
--- a/src/mongo/s/query/cluster_find.cpp
+++ b/src/mongo/s/query/cluster_find.cpp
@@ -33,6 +33,7 @@
 
 #include "mongo/s/query/cluster_find.h"
 
+#include "mongo/db/query/query_stats/query_stats.h"
 #include <fmt/format.h>
 
 #include <memory>
@@ -54,6 +55,7 @@
 #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"
@@ -373,23 +375,26 @@ CursorId runQueryWithoutRetrying(OperationContext* opCtx,
         cursorState = ClusterCursorManager::CursorState::Exhausted;
     }
 
+    auto&& opDebug = CurOp::get(opCtx)->debug();
     // Fill out query exec properties.
-    CurOp::get(opCtx)->debug().nShards = ccc->getNumRemotes();
-    CurOp::get(opCtx)->debug().nreturned = results->size();
+    opDebug.nShards = ccc->getNumRemotes();
+    opDebug.additiveMetrics.nBatches = 1;
 
     // 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) {
-        CurOp::get(opCtx)->debug().cursorExhausted = true;
+        opDebug.cursorExhausted = true;
 
         if (shardIds.size() > 0) {
             updateNumHostsTargetedMetrics(opCtx, cm, shardIds.size());
         }
+        collectQueryStatsMongos(opCtx, ccc->takeKey());
         return CursorId(0);
     }
 
@@ -400,13 +405,13 @@ CursorId runQueryWithoutRetrying(OperationContext* opCtx,
         ? ClusterCursorManager::CursorLifetime::Immortal
         : ClusterCursorManager::CursorLifetime::Mortal;
     auto authUsers = AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames();
-    ccc->incNBatches();
+    collectQueryStatsMongos(opCtx, ccc);
 
     auto cursorId = uassertStatusOK(cursorManager->registerCursor(
         opCtx, ccc.releaseCursor(), query.nss(), cursorType, cursorLifetime, authUsers));
 
     // Record the cursorID in CurOp.
-    CurOp::get(opCtx)->debug().cursorid = cursorId;
+    opDebug.cursorid = cursorId;
 
     if (shardIds.size() > 0) {
         updateNumHostsTargetedMetrics(opCtx, cm, shardIds.size());
@@ -466,6 +471,19 @@ 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;
@@ -506,16 +524,9 @@ CursorId ClusterFind::runQuery(OperationContext* opCtx,
     for (size_t retries = 1; retries <= kMaxRetries; ++retries) {
         auto swCM = getCollectionRoutingInfoForTxnCmd(opCtx, query.nss());
         if (swCM == ErrorCodes::NamespaceNotFound) {
-            uassert(CollectionUUIDMismatchInfo(query.nss().db().toString(),
-                                               *findCommand.getCollectionUUID(),
-                                               query.nss().coll().toString(),
-                                               boost::none),
-                    "Database does not exist",
-                    !findCommand.getCollectionUUID());
-
             // If the database doesn't exist, we successfully return an empty result set without
             // creating a cursor.
-            return CursorId(0);
+            return earlyExitWithNoResults(opCtx, query, findCommand);
         }
 
         const auto cm = uassertStatusOK(std::move(swCM));
@@ -842,17 +853,20 @@ 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());
-    pinnedCursor.getValue()->incNBatches();
+    collectQueryStatsMongos(opCtx, pinnedCursor.getValue());
+
     // 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 f3af2bf0d99..02e4a1d24d5 100644
--- a/src/mongo/s/query/document_source_merge_cursors.cpp
+++ b/src/mongo/s/query/document_source_merge_cursors.cpp
@@ -58,10 +58,10 @@ DocumentSourceMergeCursors::DocumentSourceMergeCursors(
 }
 
 std::size_t DocumentSourceMergeCursors::getNumRemotes() const {
-    if (_armParams) {
-        return _armParams->getRemotes().size();
+    if (_blockingResultsMerger) {
+        return _blockingResultsMerger->getNumRemotes();
     }
-    return _blockingResultsMerger->getNumRemotes();
+    return _armParams->getRemotes().size();
 }
 
 BSONObj DocumentSourceMergeCursors::getHighWaterMark() {
@@ -72,16 +72,34 @@ BSONObj DocumentSourceMergeCursors::getHighWaterMark() {
 }
 
 bool DocumentSourceMergeCursors::remotesExhausted() const {
-    if (_armParams) {
+    if (!_blockingResultsMerger) {
         // We haven't started iteration yet.
         return false;
     }
     return _blockingResultsMerger->remotesExhausted();
 }
 
+Status DocumentSourceMergeCursors::setAwaitDataTimeout(Milliseconds awaitDataTimeout) {
+    if (!_blockingResultsMerger) {
+        // In cases where a cursor was established with a batchSize of 0, the first getMore
+        // might specify a custom maxTimeMS (AKA await data timeout). In these cases we will not
+        // have iterated the cursor yet so will not have populated the merger, but need to
+        // remember/track the custom await data timeout. We will soon iterate the cursor, so we
+        // just populate the merger now and let it track the await data timeout itself.
+        populateMerger();
+    }
+    return _blockingResultsMerger->setAwaitDataTimeout(awaitDataTimeout);
+}
+
+void DocumentSourceMergeCursors::addNewShardCursors(std::vector<RemoteCursor>&& newCursors) {
+    tassert(9535000, "_blockingResultsMerger must be set", _blockingResultsMerger);
+    recordRemoteCursorShardIds(newCursors);
+    _blockingResultsMerger->addNewShardCursors(std::move(newCursors));
+}
+
 void DocumentSourceMergeCursors::populateMerger() {
-    invariant(!_blockingResultsMerger);
-    invariant(_armParams);
+    tassert(9535001, "_blockingResultsMerger must not yet be set", !_blockingResultsMerger);
+    tassert(9535002, "_armParams must be set", _armParams);
 
     _blockingResultsMerger.emplace(
         pExpCtx->opCtx,
@@ -97,7 +115,7 @@ void DocumentSourceMergeCursors::populateMerger() {
 }
 
 std::unique_ptr<RouterStageMerge> DocumentSourceMergeCursors::convertToRouterStage() {
-    invariant(!_blockingResultsMerger, "Expected conversion to happen before execution");
+    tassert(9535003, "Expected conversion to happen before execution", !_blockingResultsMerger);
     return std::make_unique<RouterStageMerge>(
         pExpCtx->opCtx, pExpCtx->mongoProcessInterface->taskExecutor, std::move(*_armParams));
 }
@@ -114,11 +132,13 @@ DocumentSource::GetNextResult DocumentSourceMergeCursors::doGetNext() {
     return Document::fromBsonWithMetaData(*next.getResult());
 }
 
-Value DocumentSourceMergeCursors::serialize(
-    boost::optional<ExplainOptions::Verbosity> explain) const {
-    invariant(!_blockingResultsMerger);
-    invariant(_armParams);
-    return Value(Document{{kStageName, _armParams->toBSON()}});
+Value DocumentSourceMergeCursors::serialize(const SerializationOptions& opts) const {
+    if (_blockingResultsMerger) {
+        return Value(Document{
+            {kStageName, _blockingResultsMerger->asyncResultsMergerParams().toBSON(opts)}});
+    }
+    tassert(9535004, "_armParams must be set", _armParams);
+    return Value(Document{{kStageName, _armParams->toBSON(opts)}});
 }
 
 boost::intrusive_ptr<DocumentSource> DocumentSourceMergeCursors::createFromBson(
@@ -150,7 +170,7 @@ void DocumentSourceMergeCursors::reattachToOperationContext(OperationContext* op
 
 void DocumentSourceMergeCursors::doDispose() {
     if (_blockingResultsMerger) {
-        invariant(!_ownCursors);
+        tassert(9535005, "_ownCursors must not be set", !_ownCursors);
         _blockingResultsMerger->kill(pExpCtx->opCtx);
     } else if (_ownCursors) {
         populateMerger();
@@ -158,7 +178,6 @@ void DocumentSourceMergeCursors::doDispose() {
     }
 }
 
-
 void DocumentSourceMergeCursors::recordRemoteCursorShardIds(
     const std::vector<RemoteCursor>& remoteCursors) {
     for (const auto& remoteCursor : remoteCursors) {
diff --git a/src/mongo/s/query/document_source_merge_cursors.h b/src/mongo/s/query/document_source_merge_cursors.h
index 33050bf45ab..925010afa42 100644
--- a/src/mongo/s/query/document_source_merge_cursors.h
+++ b/src/mongo/s/query/document_source_merge_cursors.h
@@ -30,11 +30,15 @@
 #pragma once
 
 #include <memory>
+#include <set>
+#include <variant>
+#include <vector>
 
 #include "mongo/db/pipeline/document_source.h"
 #include "mongo/executor/task_executor.h"
 #include "mongo/s/query/blocking_results_merger.h"
 #include "mongo/s/query/router_stage_merge.h"
+#include "mongo/util/duration.h"
 
 namespace mongo {
 
@@ -80,7 +84,7 @@ public:
     /**
      * Serializes this stage to be sent to perform the merging on a different host.
      */
-    Value serialize(boost::optional<ExplainOptions::Verbosity> explain = boost::none) const final;
+    Value serialize(const SerializationOptions& opts = SerializationOptions{}) const final override;
 
     StageConstraints constraints(Pipeline::SplitState pipeState) const final {
         StageConstraints constraints(StreamType::kStreaming,
@@ -118,27 +122,13 @@ public:
 
     bool remotesExhausted() const;
 
-    Status setAwaitDataTimeout(Milliseconds awaitDataTimeout) {
-        if (!_blockingResultsMerger) {
-            // In cases where a cursor was established with a batchSize of 0, the first getMore
-            // might specify a custom maxTimeMS (AKA await data timeout). In these cases we will not
-            // have iterated the cursor yet so will not have populated the merger, but need to
-            // remember/track the custom await data timeout. We will soon iterate the cursor, so we
-            // just populate the merger now and let it track the await data timeout itself.
-            populateMerger();
-        }
-        return _blockingResultsMerger->setAwaitDataTimeout(awaitDataTimeout);
-    }
+    Status setAwaitDataTimeout(Milliseconds awaitDataTimeout);
 
     /**
      * Adds the specified shard cursors to the set of cursors to be merged. The results from the
      * new cursors will be returned as normal through getNext().
      */
-    void addNewShardCursors(std::vector<RemoteCursor>&& newCursors) {
-        invariant(_blockingResultsMerger);
-        recordRemoteCursorShardIds(newCursors);
-        _blockingResultsMerger->addNewShardCursors(std::move(newCursors));
-    }
+    void addNewShardCursors(std::vector<RemoteCursor>&& newCursors);
 
     /**
      * Marks the remote cursors as unowned, meaning that they won't be killed upon disposing of this
@@ -170,7 +160,7 @@ private:
 
     // When we have parsed the params out of a BSONObj, the object needs to stay around while the
     // params are in use. We store them here.
-    boost::optional<BSONObj> _armParamsObj;
+    const boost::optional<BSONObj> _armParamsObj;
 
     // '_blockingResultsMerger' is lazily populated. Until we need to use it, '_armParams' will be
     // populated with the parameters. Once we start using '_blockingResultsMerger', '_armParams'
@@ -180,7 +170,11 @@ private:
     // cursors within '_blockingResultsMerger' to be killed prematurely. For example, if this stage
     // is parsed on mongos then forwarded to the shards, it should not kill the cursors when it goes
     // out of scope on mongos.
+    // Note that there is a single case in which neither _armParams nor _blockingResultsMerger are
+    // set, and this after convertToRouterStage() is called. After that call the DocumentSource will
+    // remain in an unusable state.
     boost::optional<AsyncResultsMergerParams> _armParams;
+    // Can only be populated if _armParams is not set. Not populated initially.
     boost::optional<BlockingResultsMerger> _blockingResultsMerger;
 
     // Indicates whether the cursors stored in _armParams are "owned", meaning the cursors should be
diff --git a/src/mongo/s/query/establish_cursors.cpp b/src/mongo/s/query/establish_cursors.cpp
index 82ec1df2809..18196d76751 100644
--- a/src/mongo/s/query/establish_cursors.cpp
+++ b/src/mongo/s/query/establish_cursors.cpp
@@ -42,8 +42,11 @@
 #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"
@@ -102,12 +105,13 @@ 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
@@ -129,22 +133,26 @@ 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) {
-        BSONObjBuilder requestWithOpKey(remote.second);
-        _opKey.appendToBuilder(&requestWithOpKey, "clientOperationKey");
-        requests.emplace_back(remote.first, requestWithOpKey.obj());
+        requests.emplace_back(remote.first, appendOpKey(_opKey, remote.second));
     }
 
     LOGV2_DEBUG(4625502,
@@ -182,11 +190,9 @@ void CursorEstablisher::waitForResponse() noexcept {
 
             hadValidCursor = true;
 
-            RemoteCursor remoteCursor;
-            remoteCursor.setCursorResponse(std::move(cursor.getValue()));
-            remoteCursor.setShardId(response.shardId);
-            remoteCursor.setHostAndPort(*response.shardHostAndPort);
-            _remoteCursors.emplace_back(std::move(remoteCursor));
+            _remoteCursors.emplace_back(RemoteCursor(response.shardId.toString(),
+                                                     *response.shardHostAndPort,
+                                                     std::move(cursor.getValue())));
         }
 
         if (response.shardHostAndPort && !hadValidCursor) {
@@ -199,16 +205,41 @@ 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;
     }
 
-    LOGV2(4625501,
-          "Unable to establish remote cursors",
-          "error"_attr = *_maybeFailure,
-          "nRemotes"_attr = _remotesToClean.size());
+    if (!(_maybeFailure->code() == ErrorCodes::CommandOnShardedViewNotSupportedOnMongod)) {
+        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.
@@ -218,21 +249,8 @@ void CursorEstablisher::checkForFailedRequests() {
     // Filter out duplicate hosts.
     auto remotes = std::set<HostAndPort>(_remotesToClean.begin(), _remotesToClean.end());
 
-    // 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));
-        }));
+    uassertStatusOK(
+        scheduleCursorCleanup(_executor, _opCtx->getServiceContext(), _opKey, std::move(remotes)));
 
     // Throw our failure.
     uassertStatusOK(*_maybeFailure);
@@ -294,10 +312,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();
 
@@ -327,6 +345,16 @@ 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
 
@@ -358,4 +386,101 @@ 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 3a904adcadd..cd19af7eea9 100644
--- a/src/mongo/s/query/establish_cursors.h
+++ b/src/mongo/s/query/establish_cursors.h
@@ -73,6 +73,26 @@ 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/store_possible_cursor.cpp b/src/mongo/s/query/store_possible_cursor.cpp
index c778daa9d84..723cafff2e4 100644
--- a/src/mongo/s/query/store_possible_cursor.cpp
+++ b/src/mongo/s/query/store_possible_cursor.cpp
@@ -88,15 +88,17 @@ StatusWith<BSONObj> storePossibleCursor(OperationContext* opCtx,
         return incomingCursorResponse.getStatus();
     }
 
-    CurOp::get(opCtx)->debug().nreturned = incomingCursorResponse.getValue().getBatch().size();
-
+    auto&& opDebug = CurOp::get(opCtx)->debug();
+    opDebug.additiveMetrics.nBatches = 1;
     // 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.
-    CurOp::get(opCtx)->debug().nShards = std::max(CurOp::get(opCtx)->debug().nShards, 1);
+    opDebug.nShards = std::max(opDebug.nShards, 1);
+    CurOp::get(opCtx)->setEndOfOpMetrics(incomingCursorResponse.getValue().getBatch().size());
 
     if (incomingCursorResponse.getValue().getCursorId() == CursorId(0)) {
-        CurOp::get(opCtx)->debug().cursorExhausted = true;
+        opDebug.cursorExhausted = true;
+        collectQueryStatsMongos(opCtx, std::move(opDebug.queryStatsInfo.key));
         return cmdResult;
     }
 
@@ -128,7 +130,7 @@ StatusWith<BSONObj> storePossibleCursor(OperationContext* opCtx,
     }
 
     auto ccc = ClusterClientCursorImpl::make(opCtx, std::move(executor), std::move(params));
-    ccc->incNBatches();
+    collectQueryStatsMongos(opCtx, ccc);
     // We don't expect to use this cursor until a subsequent getMore, so detach from the current
     // OperationContext until then.
     ccc->detachFromOperationContext();
@@ -144,7 +146,7 @@ StatusWith<BSONObj> storePossibleCursor(OperationContext* opCtx,
         return clusterCursorId.getStatus();
     }
 
-    CurOp::get(opCtx)->debug().cursorid = clusterCursorId.getValue();
+    opDebug.cursorid = clusterCursorId.getValue();
 
     CursorResponse outgoingCursorResponse(
         requestedNss,