New upstream version 6.0.20upstream

62 files changed, 1276 insertions, 529 deletions

diff --git a/src/mongo/s/SConscript b/src/mongo/s/SConscript
index e443f02fa0a..e24b7998cb4 100644
--- a/src/mongo/s/SConscript
+++ b/src/mongo/s/SConscript
@@ -60,9 +60,9 @@ env.Library(
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/commands/txn_cmd_request',
-        '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/logical_session_id_helpers',
         '$BUILD_DIR/mongo/db/pipeline/process_interface/mongo_process_interface',
+        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
         '$BUILD_DIR/mongo/db/repl/read_concern_args',
         '$BUILD_DIR/mongo/db/session_catalog',
         '$BUILD_DIR/mongo/db/shared_request_handling',
@@ -239,6 +239,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/index_commands_idl',
         '$BUILD_DIR/mongo/db/namespace_string',
         '$BUILD_DIR/mongo/db/query/query_request',
+        '$BUILD_DIR/mongo/db/query/query_shape/query_shape',
         '$BUILD_DIR/mongo/db/repl/optime',
         '$BUILD_DIR/mongo/db/server_options',
         '$BUILD_DIR/mongo/idl/feature_flag',
@@ -462,7 +463,6 @@ env.Library(
         '$BUILD_DIR/mongo/db/commands/server_status',
         '$BUILD_DIR/mongo/db/commands/server_status_core',
         '$BUILD_DIR/mongo/db/commands/server_status_servers',
-        '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/dbdirectclient',
         '$BUILD_DIR/mongo/db/ftdc/ftdc_mongos',
         '$BUILD_DIR/mongo/db/logical_session_cache',
@@ -470,6 +470,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/logical_time_metadata_hook',
         '$BUILD_DIR/mongo/db/pipeline/process_interface/mongos_process_interface_factory',
         '$BUILD_DIR/mongo/db/process_health/fault_manager',
+        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
         '$BUILD_DIR/mongo/db/read_write_concern_defaults',
         '$BUILD_DIR/mongo/db/server_options',
         '$BUILD_DIR/mongo/db/server_options_base',
diff --git a/src/mongo/s/catalog/sharding_catalog_client.h b/src/mongo/s/catalog/sharding_catalog_client.h
index 6b4ea0f4d7d..41057194e3a 100644
--- a/src/mongo/s/catalog/sharding_catalog_client.h
+++ b/src/mongo/s/catalog/sharding_catalog_client.h
@@ -315,19 +315,6 @@ public:
                                         const WriteConcernOptions& writeConcern) = 0;
 
     /**
-     * Directly inserts documents in the specified namespace on the config server. Inserts said
-     * documents using a retryable write. Underneath, a session is created and destroyed -- this
-     * ad-hoc session creation strategy should never be used outside of specific, non-performant
-     * code paths.
-     *
-     * Must only be used for insertions in the 'config' database.
-     */
-    virtual void insertConfigDocumentsAsRetryableWrite(OperationContext* opCtx,
-                                                       const NamespaceString& nss,
-                                                       std::vector<BSONObj> docs,
-                                                       const WriteConcernOptions& writeConcern) = 0;
-
-    /**
      * Updates a single document in the specified namespace on the config server. Must only be used
      * for updates to the 'config' database.
      *
diff --git a/src/mongo/s/catalog/sharding_catalog_client_impl.cpp b/src/mongo/s/catalog/sharding_catalog_client_impl.cpp
index 9720eeef27c..38949060bf1 100644
--- a/src/mongo/s/catalog/sharding_catalog_client_impl.cpp
+++ b/src/mongo/s/catalog/sharding_catalog_client_impl.cpp
@@ -102,37 +102,6 @@ void toBatchError(const Status& status, BatchedCommandResponse* response) {
     response->setStatus(status);
 }
 
-void sendRetryableWriteBatchRequestToConfig(OperationContext* opCtx,
-                                            const NamespaceString& nss,
-                                            std::vector<BSONObj>& docs,
-                                            TxnNumber txnNumber,
-                                            const WriteConcernOptions& writeConcern) {
-    auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
-
-    BatchedCommandRequest request([&] {
-        write_ops::InsertCommandRequest insertOp(nss);
-        insertOp.setDocuments(docs);
-        return insertOp;
-    }());
-    request.setWriteConcern(writeConcern.toBSON());
-
-    BSONObj cmdObj = request.toBSON();
-    BSONObjBuilder bob(cmdObj);
-    bob.append(OperationSessionInfo::kTxnNumberFieldName, txnNumber);
-
-    BatchedCommandResponse batchResponse;
-    auto response = configShard->runCommand(opCtx,
-                                            ReadPreferenceSetting{ReadPreference::PrimaryOnly},
-                                            nss.db().toString(),
-                                            bob.obj(),
-                                            Shard::kDefaultConfigCommandTimeout,
-                                            Shard::RetryPolicy::kIdempotent);
-
-    auto writeStatus = Shard::CommandResponse::processBatchWriteResponse(response, &batchResponse);
-
-    uassertStatusOK(batchResponse.toStatus());
-    uassertStatusOK(writeStatus);
-}
 
 AggregateCommandRequest makeCollectionAndChunksAggregation(OperationContext* opCtx,
                                                            const NamespaceString& nss,
@@ -1042,12 +1011,14 @@ Status ShardingCatalogClientImpl::insertConfigDocument(OperationContext* opCtx,
         insertOp.setDocuments({doc});
         return insertOp;
     }());
-    request.setWriteConcern(writeConcern.toBSON());
 
     auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
     for (int retry = 1; retry <= kMaxWriteRetry; retry++) {
-        auto response = configShard->runBatchWriteCommand(
-            opCtx, Shard::kDefaultConfigCommandTimeout, request, Shard::RetryPolicy::kNoRetry);
+        auto response = configShard->runBatchWriteCommand(opCtx,
+                                                          Shard::kDefaultConfigCommandTimeout,
+                                                          request,
+                                                          writeConcern,
+                                                          Shard::RetryPolicy::kNoRetry);
 
         Status status = response.toStatus();
 
@@ -1102,49 +1073,6 @@ Status ShardingCatalogClientImpl::insertConfigDocument(OperationContext* opCtx,
     MONGO_UNREACHABLE;
 }
 
-void ShardingCatalogClientImpl::insertConfigDocumentsAsRetryableWrite(
-    OperationContext* opCtx,
-    const NamespaceString& nss,
-    std::vector<BSONObj> docs,
-    const WriteConcernOptions& writeConcern) {
-    invariant(nss.db() == NamespaceString::kAdminDb || nss.db() == NamespaceString::kConfigDb);
-
-    AlternativeSessionRegion asr(opCtx);
-    TxnNumber currentTxnNumber = 0;
-
-    std::vector<BSONObj> workingBatch;
-    size_t workingBatchItemSize = 0;
-    int workingBatchDocSize = 0;
-
-    while (!docs.empty()) {
-        BSONObj toAdd = docs.back();
-        docs.pop_back();
-
-        const int docSizePlusOverhead =
-            toAdd.objsize() + write_ops::kRetryableAndTxnBatchWriteBSONSizeOverhead;
-        // Check if pushing this object will exceed the batch size limit or the max object size
-        if ((workingBatchItemSize + 1 > write_ops::kMaxWriteBatchSize) ||
-            (workingBatchDocSize + docSizePlusOverhead > BSONObjMaxUserSize)) {
-            sendRetryableWriteBatchRequestToConfig(
-                asr.opCtx(), nss, workingBatch, currentTxnNumber, writeConcern);
-            ++currentTxnNumber;
-
-            workingBatch.clear();
-            workingBatchItemSize = 0;
-            workingBatchDocSize = 0;
-        }
-
-        workingBatch.push_back(toAdd);
-        ++workingBatchItemSize;
-        workingBatchDocSize += docSizePlusOverhead;
-    }
-
-    if (!workingBatch.empty()) {
-        sendRetryableWriteBatchRequestToConfig(
-            asr.opCtx(), nss, workingBatch, currentTxnNumber, writeConcern);
-    }
-}
-
 StatusWith<bool> ShardingCatalogClientImpl::updateConfigDocument(
     OperationContext* opCtx,
     const NamespaceString& nss,
@@ -1189,11 +1117,10 @@ StatusWith<bool> ShardingCatalogClientImpl::_updateConfigDocument(
         }()});
         return updateOp;
     }());
-    request.setWriteConcern(writeConcern.toBSON());
 
     auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
     auto response = configShard->runBatchWriteCommand(
-        opCtx, maxTimeMs, request, Shard::RetryPolicy::kIdempotent);
+        opCtx, maxTimeMs, request, writeConcern, Shard::RetryPolicy::kIdempotent);
 
     Status status = response.toStatus();
     if (!status.isOK()) {
@@ -1225,11 +1152,13 @@ Status ShardingCatalogClientImpl::removeConfigDocuments(OperationContext* opCtx,
         }()});
         return deleteOp;
     }());
-    request.setWriteConcern(writeConcern.toBSON());
 
     auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
-    auto response = configShard->runBatchWriteCommand(
-        opCtx, Shard::kDefaultConfigCommandTimeout, request, Shard::RetryPolicy::kIdempotent);
+    auto response = configShard->runBatchWriteCommand(opCtx,
+                                                      Shard::kDefaultConfigCommandTimeout,
+                                                      request,
+                                                      writeConcern,
+                                                      Shard::RetryPolicy::kIdempotent);
     return response.toStatus();
 }
 
diff --git a/src/mongo/s/catalog/sharding_catalog_client_impl.h b/src/mongo/s/catalog/sharding_catalog_client_impl.h
index 874c58ec526..e0aa93ad3b6 100644
--- a/src/mongo/s/catalog/sharding_catalog_client_impl.h
+++ b/src/mongo/s/catalog/sharding_catalog_client_impl.h
@@ -144,11 +144,6 @@ public:
                                 const BSONObj& doc,
                                 const WriteConcernOptions& writeConcern) override;
 
-    void insertConfigDocumentsAsRetryableWrite(OperationContext* opCtx,
-                                               const NamespaceString& nss,
-                                               std::vector<BSONObj> docs,
-                                               const WriteConcernOptions& writeConcern) override;
-
     StatusWith<bool> updateConfigDocument(OperationContext* opCtx,
                                           const NamespaceString& nss,
                                           const BSONObj& query,
diff --git a/src/mongo/s/catalog/sharding_catalog_client_mock.cpp b/src/mongo/s/catalog/sharding_catalog_client_mock.cpp
index e3011cb17d7..7f19157811b 100644
--- a/src/mongo/s/catalog/sharding_catalog_client_mock.cpp
+++ b/src/mongo/s/catalog/sharding_catalog_client_mock.cpp
@@ -152,12 +152,6 @@ Status ShardingCatalogClientMock::insertConfigDocument(OperationContext* opCtx,
     return {ErrorCodes::InternalError, "Method not implemented"};
 }
 
-void ShardingCatalogClientMock::insertConfigDocumentsAsRetryableWrite(
-    OperationContext* opCtx,
-    const NamespaceString& nss,
-    std::vector<BSONObj> docs,
-    const WriteConcernOptions& writeConcern) {}
-
 StatusWith<bool> ShardingCatalogClientMock::updateConfigDocument(
     OperationContext* opCtx,
     const NamespaceString& nss,
diff --git a/src/mongo/s/catalog/sharding_catalog_client_mock.h b/src/mongo/s/catalog/sharding_catalog_client_mock.h
index 9cd6096a358..4cbfd759604 100644
--- a/src/mongo/s/catalog/sharding_catalog_client_mock.h
+++ b/src/mongo/s/catalog/sharding_catalog_client_mock.h
@@ -121,11 +121,6 @@ public:
                                 const BSONObj& doc,
                                 const WriteConcernOptions& writeConcern) override;
 
-    void insertConfigDocumentsAsRetryableWrite(OperationContext* opCtx,
-                                               const NamespaceString& nss,
-                                               std::vector<BSONObj> docs,
-                                               const WriteConcernOptions& writeConcern) override;
-
     StatusWith<bool> updateConfigDocument(OperationContext* opCtx,
                                           const NamespaceString& nss,
                                           const BSONObj& query,
diff --git a/src/mongo/s/catalog_cache_refresh_test.cpp b/src/mongo/s/catalog_cache_refresh_test.cpp
index c958dec98fb..39c8e4b13cd 100644
--- a/src/mongo/s/catalog_cache_refresh_test.cpp
+++ b/src/mongo/s/catalog_cache_refresh_test.cpp
@@ -33,6 +33,7 @@
 
 #include "mongo/db/concurrency/locker_noop.h"
 #include "mongo/db/pipeline/aggregation_request_helper.h"
+#include "mongo/db/query/cursor_response.h"
 #include "mongo/s/catalog/type_chunk.h"
 #include "mongo/s/catalog/type_collection.h"
 #include "mongo/s/catalog/type_database_gen.h"
diff --git a/src/mongo/s/catalog_cache_test.cpp b/src/mongo/s/catalog_cache_test.cpp
index b894d6cdee6..61a6ca2b067 100644
--- a/src/mongo/s/catalog_cache_test.cpp
+++ b/src/mongo/s/catalog_cache_test.cpp
@@ -33,6 +33,8 @@
 
 #include <boost/optional/optional_io.hpp>
 
+#include "mongo/db/cursor_id.h"
+#include "mongo/db/query/cursor_response.h"
 #include "mongo/s/catalog/type_database_gen.h"
 #include "mongo/s/catalog_cache.h"
 #include "mongo/s/catalog_cache_loader_mock.h"
diff --git a/src/mongo/s/catalog_cache_test_fixture.cpp b/src/mongo/s/catalog_cache_test_fixture.cpp
index 6e66a30d6b2..9edd553f704 100644
--- a/src/mongo/s/catalog_cache_test_fixture.cpp
+++ b/src/mongo/s/catalog_cache_test_fixture.cpp
@@ -38,7 +38,9 @@
 #include "mongo/client/remote_command_targeter_factory_mock.h"
 #include "mongo/client/remote_command_targeter_mock.h"
 #include "mongo/db/client.h"
+#include "mongo/db/cursor_id.h"
 #include "mongo/db/query/collation/collator_factory_mock.h"
+#include "mongo/db/query/cursor_response.h"
 #include "mongo/s/catalog/type_chunk.h"
 #include "mongo/s/catalog/type_collection.h"
 #include "mongo/s/catalog/type_database_gen.h"
diff --git a/src/mongo/s/chunk.cpp b/src/mongo/s/chunk.cpp
index 35132c6d82c..e141cfea226 100644
--- a/src/mongo/s/chunk.cpp
+++ b/src/mongo/s/chunk.cpp
@@ -120,7 +120,7 @@ BSONObj ChunkInfo::toBSON() const {
     bob.append("maxKeyString", _maxKeyString);
     bob.append("shardId", _shardId);
     _lastmod.serializeToBSON("lastmod", &bob);
-    bob.append("jumbo", _jumbo);
+    bob.append("jumbo", _jumbo.load());
     bob.append("bytesWritten", (long long)_writesTracker->getBytesWritten());
 
     BSONArrayBuilder historyArr{bob.subarrayStart("history")};
@@ -132,7 +132,7 @@ BSONObj ChunkInfo::toBSON() const {
 }
 
 void ChunkInfo::markAsJumbo() {
-    _jumbo = true;
+    _jumbo.store(true);
 }
 
 void Chunk::throwIfMoved() const {
diff --git a/src/mongo/s/chunk.h b/src/mongo/s/chunk.h
index 2c0dabe7a3f..16f5909a72a 100644
--- a/src/mongo/s/chunk.h
+++ b/src/mongo/s/chunk.h
@@ -29,6 +29,7 @@
 
 #pragma once
 
+#include "mongo/platform/atomic_word.h"
 #include "mongo/s/catalog/type_chunk.h"
 #include "mongo/s/chunk_version.h"
 #include "mongo/s/shard_id.h"
@@ -92,7 +93,7 @@ public:
     }
 
     bool isJumbo() const {
-        return _jumbo;
+        return _jumbo.load();
     }
 
     /**
@@ -132,7 +133,7 @@ private:
 
     // Indicates whether this chunk should be treated as jumbo and not attempted to be moved or
     // split
-    mutable bool _jumbo;
+    AtomicWord<bool> _jumbo;
 
     // Used for tracking writes to this chunk, to estimate its size for the autosplitter. Since
     // ChunkInfo objects are always treated as const, and this contains metadata about the chunk
diff --git a/src/mongo/s/chunk_manager_targeter.cpp b/src/mongo/s/chunk_manager_targeter.cpp
index f750554ea56..f7380a00920 100644
--- a/src/mongo/s/chunk_manager_targeter.cpp
+++ b/src/mongo/s/chunk_manager_targeter.cpp
@@ -674,7 +674,7 @@ void ChunkManagerTargeter::noteStaleShardResponse(OperationContext* opCtx,
         Grid::get(opCtx)
             ->catalogCache()
             ->invalidateShardOrEntireCollectionEntryForShardedCollection(
-                _nss, staleInfo.getVersionWanted(), endpoint.shardName);
+                _nss, boost::none, endpoint.shardName);
     }
 
     _lastError = LastErrorType::kStaleShardVersion;
diff --git a/src/mongo/s/client/shard.cpp b/src/mongo/s/client/shard.cpp
index ac360694af4..1408b227b1e 100644
--- a/src/mongo/s/client/shard.cpp
+++ b/src/mongo/s/client/shard.cpp
@@ -41,7 +41,6 @@ namespace mongo {
 namespace {
 
 const int kOnErrorNumRetries = 3;
-
 }  // namespace
 
 Status Shard::CommandResponse::getEffectiveStatus(
@@ -194,36 +193,6 @@ StatusWith<Shard::QueryResponse> Shard::runExhaustiveCursorCommand(
     MONGO_UNREACHABLE;
 }
 
-BatchedCommandResponse Shard::runBatchWriteCommand(OperationContext* opCtx,
-                                                   const Milliseconds maxTimeMS,
-                                                   const BatchedCommandRequest& batchRequest,
-                                                   RetryPolicy retryPolicy) {
-    const StringData dbname = batchRequest.getNS().db();
-    const BSONObj cmdObj = batchRequest.toBSON();
-
-    for (int retry = 1; retry <= kOnErrorNumRetries; ++retry) {
-        // Note: write commands can only be issued against a primary.
-        auto swResponse = _runCommand(
-            opCtx, ReadPreferenceSetting{ReadPreference::PrimaryOnly}, dbname, maxTimeMS, cmdObj);
-
-        BatchedCommandResponse batchResponse;
-        auto writeStatus = CommandResponse::processBatchWriteResponse(swResponse, &batchResponse);
-        if (retry < kOnErrorNumRetries && isRetriableError(writeStatus.code(), retryPolicy)) {
-            LOGV2_DEBUG(22721,
-                        2,
-                        "Batch write command to shard {shardId} failed with retryable error "
-                        "and will be retried. Caused by {error}",
-                        "Batch write command failed with retryable error and will be retried",
-                        "shardId"_attr = getId(),
-                        "error"_attr = redact(writeStatus));
-            continue;
-        }
-
-        return batchResponse;
-    }
-    MONGO_UNREACHABLE;
-}
-
 StatusWith<Shard::QueryResponse> Shard::exhaustiveFindOnConfig(
     OperationContext* opCtx,
     const ReadPreferenceSetting& readPref,
@@ -250,4 +219,34 @@ StatusWith<Shard::QueryResponse> Shard::exhaustiveFindOnConfig(
     MONGO_UNREACHABLE;
 }
 
+BatchedCommandResponse Shard::_submitBatchWriteCommand(OperationContext* opCtx,
+                                                       const BSONObj& serialisedBatchRequest,
+                                                       StringData dbName,
+                                                       Milliseconds maxTimeMS,
+                                                       RetryPolicy retryPolicy) {
+    for (int retry = 1; retry <= kOnErrorNumRetries; ++retry) {
+        // Note: write commands can only be issued against a primary.
+        auto swResponse = _runCommand(opCtx,
+                                      ReadPreferenceSetting{ReadPreference::PrimaryOnly},
+                                      dbName,
+                                      maxTimeMS,
+                                      serialisedBatchRequest);
+
+        BatchedCommandResponse batchResponse;
+        auto writeStatus = CommandResponse::processBatchWriteResponse(swResponse, &batchResponse);
+        if (retry < kOnErrorNumRetries && isRetriableError(writeStatus.code(), retryPolicy)) {
+            LOGV2_DEBUG(22721,
+                        2,
+                        "Batch write command failed with retryable error and will be retried",
+                        "shardId"_attr = getId(),
+                        "error"_attr = redact(writeStatus));
+            continue;
+        }
+
+        return batchResponse;
+    }
+    MONGO_UNREACHABLE;
+}
+
+
 }  // namespace mongo
diff --git a/src/mongo/s/client/shard.h b/src/mongo/s/client/shard.h
index f690341fcb2..e545f0b2f6b 100644
--- a/src/mongo/s/client/shard.h
+++ b/src/mongo/s/client/shard.h
@@ -226,10 +226,11 @@ public:
      * commands return errors in a different format than regular commands do, so checking for
      * retriable errors must be done differently.
      */
-    BatchedCommandResponse runBatchWriteCommand(OperationContext* opCtx,
-                                                Milliseconds maxTimeMS,
-                                                const BatchedCommandRequest& batchRequest,
-                                                RetryPolicy retryPolicy);
+    virtual BatchedCommandResponse runBatchWriteCommand(OperationContext* opCtx,
+                                                        Milliseconds maxTimeMS,
+                                                        const BatchedCommandRequest& batchRequest,
+                                                        const WriteConcernOptions& writeConcern,
+                                                        RetryPolicy retryPolicy) = 0;
 
     /**
      * Warning: This method exhausts the cursor and pulls all data into memory.
@@ -292,10 +293,22 @@ public:
 protected:
     Shard(const ShardId& id);
 
+    /**
+     * Submits the batch request applying the specified retry policy and timeout and using the
+     * machinery provided by each implementation.
+     * Callers of this function must ensure to have configured the write concern settings
+     * accordingly to their specific semantics.
+     */
+    BatchedCommandResponse _submitBatchWriteCommand(OperationContext* opCtx,
+                                                    const BSONObj& serialisedBatchRequest,
+                                                    StringData dbName,
+                                                    Milliseconds maxTimeMS,
+                                                    RetryPolicy retryPolicy);
+
 private:
     /**
-     * Runs the specified command against the shard backed by this object with a timeout set to the
-     * minimum of maxTimeMSOverride or the timeout of the OperationContext.
+     * Runs the specified command against the shard backed by this object with a timeout set to
+     * the minimum of maxTimeMSOverride or the timeout of the OperationContext.
      *
      * The return value exposes RemoteShard's host for calls to updateReplSetMonitor.
      *
diff --git a/src/mongo/s/client/shard_remote.cpp b/src/mongo/s/client/shard_remote.cpp
index 9c9241ec7e9..156dad7171a 100644
--- a/src/mongo/s/client/shard_remote.cpp
+++ b/src/mongo/s/client/shard_remote.cpp
@@ -540,6 +540,23 @@ Status ShardRemote::runAggregation(
 }
 
 
+BatchedCommandResponse ShardRemote::runBatchWriteCommand(OperationContext* opCtx,
+                                                         const Milliseconds maxTimeMS,
+                                                         const BatchedCommandRequest& batchRequest,
+                                                         const WriteConcernOptions& writeConcern,
+                                                         RetryPolicy retryPolicy) {
+    const auto dbName = batchRequest.getNS().db();
+    const BSONObj cmdObj = [&] {
+        BSONObjBuilder cmdObjBuilder;
+        batchRequest.serialize(&cmdObjBuilder);
+        cmdObjBuilder.append(WriteConcernOptions::kWriteConcernField, writeConcern.toBSON());
+        return cmdObjBuilder.obj();
+    }();
+
+    return _submitBatchWriteCommand(opCtx, cmdObj, dbName, maxTimeMS, retryPolicy);
+}
+
+
 StatusWith<ShardRemote::AsyncCmdHandle> ShardRemote::_scheduleCommand(
     OperationContext* opCtx,
     const ReadPreferenceSetting& readPref,
diff --git a/src/mongo/s/client/shard_remote.h b/src/mongo/s/client/shard_remote.h
index 6c99a8a5247..6aec3f87da5 100644
--- a/src/mongo/s/client/shard_remote.h
+++ b/src/mongo/s/client/shard_remote.h
@@ -91,6 +91,12 @@ public:
         std::function<bool(const std::vector<BSONObj>& batch,
                            const boost::optional<BSONObj>& postBatchResumeToken)> callback);
 
+    BatchedCommandResponse runBatchWriteCommand(OperationContext* opCtx,
+                                                Milliseconds maxTimeMS,
+                                                const BatchedCommandRequest& batchRequest,
+                                                const WriteConcernOptions& writeConcern,
+                                                RetryPolicy retryPolicy) final;
+
 private:
     struct AsyncCmdHandle {
         HostAndPort hostTargetted;
diff --git a/src/mongo/s/cluster_commands_helpers.cpp b/src/mongo/s/cluster_commands_helpers.cpp
index 416dbec7324..1a84121bf9e 100644
--- a/src/mongo/s/cluster_commands_helpers.cpp
+++ b/src/mongo/s/cluster_commands_helpers.cpp
@@ -662,7 +662,7 @@ bool appendEmptyResultSet(OperationContext* opCtx,
                           const std::string& ns) {
     invariant(!status.isOK());
 
-    CurOp::get(opCtx)->debug().nreturned = 0;
+    CurOp::get(opCtx)->debug().additiveMetrics.nreturned = 0;
     CurOp::get(opCtx)->debug().nShards = 0;
 
     if (status == ErrorCodes::NamespaceNotFound) {
diff --git a/src/mongo/s/commands/SConscript b/src/mongo/s/commands/SConscript
index 33bd413b4f7..9f9c43c079a 100644
--- a/src/mongo/s/commands/SConscript
+++ b/src/mongo/s/commands/SConscript
@@ -132,6 +132,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/query/command_request_response',
         '$BUILD_DIR/mongo/db/query/cursor_response_idl',
         '$BUILD_DIR/mongo/db/query/map_reduce_output_format',
+        '$BUILD_DIR/mongo/db/query/query_shape/query_shape',
         '$BUILD_DIR/mongo/db/read_write_concern_defaults',
         '$BUILD_DIR/mongo/db/repl/hello_auth',
         '$BUILD_DIR/mongo/db/repl/hello_command',
@@ -240,6 +241,7 @@ env.CppUnitTest(
         "cluster_command_test_fixture.cpp",
         "cluster_delete_test.cpp",
         "cluster_distinct_test.cpp",
+        "cluster_explain_test.cpp",
         "cluster_find_and_modify_test.cpp",
         "cluster_find_test.cpp",
         "cluster_insert_test.cpp",
diff --git a/src/mongo/s/commands/cluster_db_stats_cmd.cpp b/src/mongo/s/commands/cluster_db_stats_cmd.cpp
index 9154d5f40d8..9439e967521 100644
--- a/src/mongo/s/commands/cluster_db_stats_cmd.cpp
+++ b/src/mongo/s/commands/cluster_db_stats_cmd.cpp
@@ -131,11 +131,12 @@ public:
                 opCtx, this, CommandHelpers::filterCommandRequestForPassthrough(cmdObj)),
             ReadPreferenceSetting::get(opCtx),
             Shard::RetryPolicy::kIdempotent);
-        if (!appendRawResponses(opCtx, &errmsg, &output, shardResponses).responseOK) {
+        auto appendResult = appendRawResponses(opCtx, &errmsg, &output, shardResponses);
+        if (!appendResult.responseOK) {
             return false;
         }
 
-        aggregateResults(scale, shardResponses, output);
+        aggregateResults(scale, appendResult.successResponses, output);
         return true;
     }
 
diff --git a/src/mongo/s/commands/cluster_explain.cpp b/src/mongo/s/commands/cluster_explain.cpp
index 759ce8ed3d4..dd85d94d57f 100644
--- a/src/mongo/s/commands/cluster_explain.cpp
+++ b/src/mongo/s/commands/cluster_explain.cpp
@@ -30,7 +30,6 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/bson/bsonmisc.h"
-#include "mongo/db/commands.h"
 #include "mongo/db/query/explain_common.h"
 #include "mongo/idl/command_generic_argument.h"
 #include "mongo/rpc/get_status_from_command_result.h"
@@ -115,18 +114,36 @@ void throwOnBadARSResponse(const AsyncRequestsSender::Response& arsResponse) {
 
 // static
 BSONObj ClusterExplain::wrapAsExplain(const BSONObj& cmdObj, ExplainOptions::Verbosity verbosity) {
-    auto filtered = CommandHelpers::filterCommandRequestForPassthrough(cmdObj);
     BSONObjBuilder out;
-    out.append("explain", filtered);
-    out.append("verbosity", ExplainOptions::verbosityString(verbosity));
-
-    // Propagate all generic arguments out of the inner command since the shards will only process
-    // them at the top level.
-    for (auto elem : filtered) {
-        if (isGenericArgument(elem.fieldNameStringData())) {
-            out.append(elem);
+    // Prune generic arguments out of the inner command since any relevant ones should already
+    // be provided to the outer explain command. The shards will only process them at the top level.
+    // As an exception, the "comment" parameter will be propagated out of the inner command to
+    // maintain the behavior in our documentation:
+    // https://www.mongodb.com/docs/manual/reference/command/explain/.
+    // The "readConcern" parameter will also be propagated out of the inner command as the final
+    // explain command inherits readConcern from the inner command invocation.
+    BSONObjBuilder explainBuilder = out.subobjStart("explain");
+    BSONElement commentField;
+    BSONElement readConcernField;
+    for (auto&& elem : cmdObj) {
+        const auto& fieldName = elem.fieldNameStringData();
+        if (!isGenericArgument(fieldName)) {
+            explainBuilder.append(elem);
+        } else if (fieldName == "comment"_sd) {
+            commentField = elem;
+        } else if (fieldName == "readConcern"_sd) {
+            readConcernField = elem;
         }
     }
+    explainBuilder.done();
+
+    out.append("verbosity", ExplainOptions::verbosityString(verbosity));
+    if (commentField) {
+        out.append(commentField);
+    }
+    if (readConcernField) {
+        out.append(readConcernField);
+    }
 
     return out.obj();
 }
diff --git a/src/mongo/s/commands/cluster_explain.h b/src/mongo/s/commands/cluster_explain.h
index f1ff6716de8..9be0ccb960a 100644
--- a/src/mongo/s/commands/cluster_explain.h
+++ b/src/mongo/s/commands/cluster_explain.h
@@ -46,7 +46,8 @@ class ClusterExplain {
 public:
     /**
      * Returns an explain command request wrapping the passed in command at the given verbosity
-     * level, propagating generic top-level command arguments.
+     * level, pruning any generic arguments in the inner command as they should already be provided
+     * on the top-level outer commmand.
      */
     static BSONObj wrapAsExplain(const BSONObj& cmdObj, ExplainOptions::Verbosity verbosity);
 
diff --git a/src/mongo/s/commands/cluster_explain_test.cpp b/src/mongo/s/commands/cluster_explain_test.cpp
new file mode 100644
index 00000000000..298ed85ca61
--- /dev/null
+++ b/src/mongo/s/commands/cluster_explain_test.cpp
@@ -0,0 +1,103 @@
+/**
+ *    Copyright (C) 2024-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/db/query/explain_verbosity_gen.h"
+#include "mongo/idl/command_generic_argument.h"
+#include "mongo/s/commands/cluster_explain.h"
+#include "mongo/unittest/bson_test_util.h"
+#include "mongo/unittest/unittest.h"
+#include "mongo/util/assert_util.h"
+
+namespace mongo {
+namespace {
+void testPruneGenericArgs(const std::string& genericArg) {
+    auto internalCmd = fromjson("{find: 'test', filter: {a: 1}, " + genericArg + "}");
+    auto verbosity = explain::VerbosityEnum::kQueryPlanner;
+    auto expected =
+        fromjson("{explain: {find: 'test', filter: {a: 1}}, verbosity: 'queryPlanner'}");
+    ASSERT_BSONOBJ_EQ(ClusterExplain::wrapAsExplain(internalCmd, verbosity), expected);
+}
+
+void testPropagateGenericArgs(const std::string& genericArg) {
+    auto internalCmd = fromjson("{find: 'test', filter: {a: 1}, " + genericArg + "}");
+    auto verbosity = explain::VerbosityEnum::kQueryPlanner;
+    auto expected = fromjson(
+        "{explain: {find: 'test', filter: {a: 1}}, verbosity: 'queryPlanner', " + genericArg + "}");
+    ASSERT_BSONOBJ_EQ(ClusterExplain::wrapAsExplain(internalCmd, verbosity), expected);
+}
+
+TEST(ClusterExplainTest, PruneMaxTimeMS) {
+    std::string maxTimeMS = "maxTimeMS: 1";
+    testPruneGenericArgs(maxTimeMS);
+}
+
+TEST(ClusterExplainTest, PruneWriteConcern) {
+    std::string writeConcern = "writeConcern: {w: 1}";
+    testPruneGenericArgs(writeConcern);
+}
+
+TEST(ClusterExplainTest, PruneLsid) {
+    auto internalCmd = BSON("find"
+                            << "test"
+                            << "filter" << BSON("a" << 1) << "lsid"
+                            << BSON("id" << mongo::UUID::gen()));
+    auto verbosity = explain::VerbosityEnum::kQueryPlanner;
+    auto expected =
+        fromjson("{explain: {find: 'test', filter: {a: 1}}, verbosity: 'queryPlanner'}");
+    ASSERT_BSONOBJ_EQ(ClusterExplain::wrapAsExplain(internalCmd, verbosity), expected);
+}
+
+TEST(ClusterExplainTest, PruneReadPreference) {
+    std::string readPreference = "$queryOptions: {$readPreference: 'secondary'}";
+    testPruneGenericArgs(readPreference);
+}
+
+TEST(ClusterExplainTest, PruneClusterTime) {
+    auto internalCmd = BSON("find"
+                            << "test"
+                            << "filter" << BSON("a" << 1) << "$clusterTime"
+                            << BSON("clusterTime" << Timestamp(2, 2)) << "$configTime"
+                            << Timestamp(2, 2) << "$topologyTime" << Timestamp(2, 2));
+    auto verbosity = explain::VerbosityEnum::kQueryPlanner;
+    auto expected =
+        fromjson("{explain: {find: 'test', filter: {a: 1}}, verbosity: 'queryPlanner'}");
+    ASSERT_BSONOBJ_EQ(ClusterExplain::wrapAsExplain(internalCmd, verbosity), expected);
+}
+
+TEST(ClusterExplainTest, PropagateComment) {
+    std::string comment = "comment: 'Quetzlcoatl'";
+    testPropagateGenericArgs(comment);
+}
+
+TEST(ClusterExplainTest, PropagateReadConcern) {
+    std::string readConcern = "readConcern: {level: 'linearizable'}";
+    testPropagateGenericArgs(readConcern);
+}
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/s/commands/cluster_find_cmd.h b/src/mongo/s/commands/cluster_find_cmd.h
index 7d58f7ab0c4..6476d883200 100644
--- a/src/mongo/s/commands/cluster_find_cmd.h
+++ b/src/mongo/s/commands/cluster_find_cmd.h
@@ -38,6 +38,9 @@
 #include "mongo/db/fle_crud.h"
 #include "mongo/db/matcher/extensions_callback_noop.h"
 #include "mongo/db/query/cursor_response.h"
+#include "mongo/db/query/query_shape/query_shape.h"
+#include "mongo/db/query/query_stats/find_key.h"
+#include "mongo/db/query/query_stats/query_stats.h"
 #include "mongo/db/stats/counters.h"
 #include "mongo/db/views/resolved_view.h"
 #include "mongo/rpc/get_status_from_command_result.h"
@@ -201,16 +204,23 @@ public:
 
             Impl::checkCanRunHere(opCtx);
 
-            auto findCommand = _parseCmdObjectToFindCommandRequest(opCtx, ns(), _request.body);
-
-            const boost::intrusive_ptr<ExpressionContext> expCtx;
-            auto cq = uassertStatusOK(
-                CanonicalQuery::canonicalize(opCtx,
-                                             std::move(findCommand),
-                                             false, /* isExplain */
-                                             expCtx,
-                                             ExtensionsCallbackNoop(),
-                                             MatchExpressionParser::kAllowAllSpecialFeatures));
+            auto&& parsedFindResult = uassertStatusOK(parsed_find_command::parse(
+                opCtx,
+                _parseCmdObjectToFindCommandRequest(opCtx, ns(), _request.body),
+                ExtensionsCallbackNoop(),
+                MatchExpressionParser::kAllowAllSpecialFeatures));
+            auto& expCtx = parsedFindResult.first;
+            auto& parsedFind = parsedFindResult.second;
+
+            if (!_didDoFLERewrite) {
+                query_stats::registerRequest(opCtx, expCtx->ns, [&]() {
+                    // This callback is either never invoked or invoked
+                    // immediately within registerRequest, so
+                    // use-after-move of parsedFind isn't an issue.
+                    return std::make_unique<query_stats::FindKey>(expCtx, *parsedFind);
+                });
+            }
+            auto cq = uassertStatusOK(CanonicalQuery::canonicalize(expCtx, std::move(parsedFind)));
 
             try {
                 // Do the work to generate the first batch of results. This blocks waiting to get
@@ -264,7 +274,7 @@ public:
          * were supplied with the command, and sets the constant runtime values that will be
          * forwarded to each shard.
          */
-        static std::unique_ptr<FindCommandRequest> _parseCmdObjectToFindCommandRequest(
+        std::unique_ptr<FindCommandRequest> _parseCmdObjectToFindCommandRequest(
             OperationContext* opCtx, NamespaceString nss, BSONObj cmdObj) {
             auto findCommand = query_request_helper::makeFromFindCommand(
                 std::move(cmdObj),
@@ -291,6 +301,7 @@ public:
                 invariant(findCommand->getNamespaceOrUUID().nss());
                 processFLEFindS(
                     opCtx, findCommand->getNamespaceOrUUID().nss().get(), findCommand.get());
+                _didDoFLERewrite = true;
             }
 
             return findCommand;
@@ -298,6 +309,7 @@ public:
 
         const OpMsgRequest& _request;
         const StringData _dbName;
+        bool _didDoFLERewrite{false};
     };
 };
 
diff --git a/src/mongo/s/commands/cluster_fle2_compact_cmd.cpp b/src/mongo/s/commands/cluster_fle2_compact_cmd.cpp
index b1cecbf555e..7abae202f17 100644
--- a/src/mongo/s/commands/cluster_fle2_compact_cmd.cpp
+++ b/src/mongo/s/commands/cluster_fle2_compact_cmd.cpp
@@ -32,6 +32,7 @@
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/fle2_compact_gen.h"
+#include "mongo/db/curop.h"
 #include "mongo/s/cluster_commands_helpers.h"
 #include "mongo/s/grid.h"
 
diff --git a/src/mongo/s/commands/cluster_write_cmd.cpp b/src/mongo/s/commands/cluster_write_cmd.cpp
index 637ad92085d..8590832acc6 100644
--- a/src/mongo/s/commands/cluster_write_cmd.cpp
+++ b/src/mongo/s/commands/cluster_write_cmd.cpp
@@ -145,11 +145,6 @@ boost::optional<WouldChangeOwningShardInfo> getWouldChangeOwningShardErrorInfo(
 void handleWouldChangeOwningShardErrorRetryableWrite(OperationContext* opCtx,
                                                      BatchedCommandRequest* request,
                                                      BatchedCommandResponse* response) {
-    // Strip write concern because this command will be sent as part of a
-    // transaction and the write concern has already been loaded onto the opCtx and
-    // will be picked up by the transaction API.
-    request->unsetWriteConcern();
-
     // Strip runtime constants because they will be added again when the API sends this command
     // through the service entry point.
     request->unsetLegacyRuntimeConstants();
@@ -318,11 +313,6 @@ bool handleWouldChangeOwningShardError(OperationContext* opCtx,
                 auto& readConcernArgs = repl::ReadConcernArgs::get(opCtx);
                 readConcernArgs = repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern);
 
-                // Ensure the retried operation does not include WC inside the transaction.  The
-                // transaction commit will still use the WC, because it uses the WC from the opCtx
-                // (which has been set previously in Strategy).
-                request->unsetWriteConcern();
-
                 documentShardKeyUpdateUtil::startTransactionForShardKeyUpdate(opCtx);
                 // Clear the error details from the response object before sending the write again
                 response->unsetErrDetails();
@@ -507,22 +497,6 @@ bool ClusterWriteCmd::InvocationBase::runImpl(OperationContext* opCtx,
     BatchWriteExecStats stats;
     BatchedCommandResponse response;
 
-    // Append the write concern from the opCtx extracted during command setup.
-    if (!batchedRequest.hasWriteConcern()) {
-        batchedRequest.setWriteConcern(opCtx->getWriteConcern().toBSON());
-    }
-
-    // Write ops are never allowed to have writeConcern inside transactions. Normally
-    // disallowing WC on non-terminal commands in a transaction is handled earlier, during
-    // command dispatch. However, if this is a regular write operation being automatically
-    // retried inside a transaction (such as changing a document's shard key across shards),
-    // then batchedRequest will have a writeConcern (added by the if() above) from when it was
-    // initially run outside a transaction. Thus it's necessary to unconditionally clear the
-    // writeConcern when in a transaction.
-    if (TransactionRouter::get(opCtx)) {
-        batchedRequest.unsetWriteConcern();
-    }
-
     cluster::write(opCtx, batchedRequest, &stats, &response);
 
     bool updatedShardKey = false;
diff --git a/src/mongo/s/mongos_main.cpp b/src/mongo/s/mongos_main.cpp
index dd78e7772d1..665f3fe08d4 100644
--- a/src/mongo/s/mongos_main.cpp
+++ b/src/mongo/s/mongos_main.cpp
@@ -316,7 +316,10 @@ void cleanupTask(const ShutdownTaskArgs& shutdownArgs) {
 
         ReplicaSetMonitor::shutdown();
 
-        opCtx->setIsExecutingShutdown();
+        {
+            stdx::lock_guard lg(client);
+            opCtx->setIsExecutingShutdown();
+        }
 
         if (serviceContext) {
             serviceContext->setKillAllOperations();
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,
diff --git a/src/mongo/s/resharding/common_types.idl b/src/mongo/s/resharding/common_types.idl
index fb0547b7091..456f96bf811 100644
--- a/src/mongo/s/resharding/common_types.idl
+++ b/src/mongo/s/resharding/common_types.idl
@@ -271,10 +271,13 @@ structs:
         description: "A struct representing the information needed for a resharding pipeline to
                       determine which documents belong to a particular shard."
         strict: true
+        query_shape_component: true
         fields:
             recipientShardId:
                 type: shard_id
                 description: "The id of the recipient shard."
+                query_shape: anonymize
             reshardingKey:
                 type: KeyPattern
                 description: "The index specification document to use as the new shard key."
+                query_shape: custom
diff --git a/src/mongo/s/service_entry_point_mongos.cpp b/src/mongo/s/service_entry_point_mongos.cpp
index 12ebf1299bd..7c1e38d008c 100644
--- a/src/mongo/s/service_entry_point_mongos.cpp
+++ b/src/mongo/s/service_entry_point_mongos.cpp
@@ -203,9 +203,17 @@ Future<DbResponse> HandleRequest::run() {
 }
 
 Future<DbResponse> ServiceEntryPointMongos::handleRequestImpl(OperationContext* opCtx,
-                                                              const Message& message) noexcept {
+                                                              const Message& message) try {
     auto hr = std::make_shared<HandleRequest>(opCtx, message);
     return hr->run();
+} catch (const DBException& ex) {
+    auto status = ex.toStatus();
+    LOGV2(9431602, "Failed to handle request", "error"_attr = redact(status));
+    return status;
+} catch (...) {
+    auto error = exceptionToStatus();
+    LOGV2_FATAL(
+        9431601, "Request handling produced unhandled exception", "error"_attr = redact(error));
 }
 
 Future<DbResponse> ServiceEntryPointMongos::handleRequest(OperationContext* opCtx,
@@ -219,7 +227,7 @@ void ServiceEntryPointMongos::onClientConnect(Client* client) {
     }
 }
 
-void ServiceEntryPointMongos::onClientDisconnect(Client* client) {
+void ServiceEntryPointMongos::onClientDisconnect(Client* client) try {
     if (load_balancer_support::isFromLoadBalancer(client)) {
         _loadBalancedConnections.decrement();
 
@@ -259,6 +267,11 @@ void ServiceEntryPointMongos::onClientDisconnect(Client* client) {
                  "aborting in-progress transaction because load-balanced client disconnected"});
         }
     }
+} catch (const DBException& ex) {
+    LOGV2_DEBUG(8969800,
+                2,
+                "Encountered error while performing client connection cleanup",
+                "error"_attr = ex.toStatus());
 }
 
 void ServiceEntryPointMongos::appendStats(BSONObjBuilder* bob) const {
diff --git a/src/mongo/s/service_entry_point_mongos.h b/src/mongo/s/service_entry_point_mongos.h
index c5c6530d2a9..ccda290e475 100644
--- a/src/mongo/s/service_entry_point_mongos.h
+++ b/src/mongo/s/service_entry_point_mongos.h
@@ -45,8 +45,7 @@ class ServiceEntryPointMongos final : public ServiceEntryPointImpl {
 public:
     using ServiceEntryPointImpl::ServiceEntryPointImpl;
 
-    static Future<DbResponse> handleRequestImpl(OperationContext* opCtx,
-                                                const Message& request) noexcept;
+    static Future<DbResponse> handleRequestImpl(OperationContext* opCtx, const Message& request);
 
     Future<DbResponse> handleRequest(OperationContext* opCtx,
                                      const Message& request) noexcept override;
diff --git a/src/mongo/s/sessions_collection_sharded_test.cpp b/src/mongo/s/sessions_collection_sharded_test.cpp
index d82a4edfbfa..b8cd82617e5 100644
--- a/src/mongo/s/sessions_collection_sharded_test.cpp
+++ b/src/mongo/s/sessions_collection_sharded_test.cpp
@@ -34,6 +34,7 @@
 #include "mongo/client/remote_command_targeter_mock.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/logical_session_id.h"
+#include "mongo/db/query/cursor_response.h"
 #include "mongo/s/catalog/type_shard.h"
 #include "mongo/s/catalog_cache_test_fixture.h"
 #include "mongo/s/client/shard_registry.h"
diff --git a/src/mongo/s/transaction_router_test.cpp b/src/mongo/s/transaction_router_test.cpp
index 74ce1210978..d72d1aea6ff 100644
--- a/src/mongo/s/transaction_router_test.cpp
+++ b/src/mongo/s/transaction_router_test.cpp
@@ -56,15 +56,6 @@
 #include "mongo/util/net/socket_utils.h"
 #include "mongo/util/tick_source_mock.h"
 
-#define ASSERT_DOES_NOT_THROW(EXPRESSION)                                          \
-    try {                                                                          \
-        EXPRESSION;                                                                \
-    } catch (const AssertionException& e) {                                        \
-        str::stream err;                                                           \
-        err << "Threw an exception incorrectly: " << e.toString();                 \
-        ::mongo::unittest::TestAssertionFailure(__FILE__, __LINE__, err).stream(); \
-    }
-
 namespace mongo {
 namespace {
 
diff --git a/src/mongo/s/write_ops/batch_write_exec.cpp b/src/mongo/s/write_ops/batch_write_exec.cpp
index 45b89579dbb..980d376b607 100644
--- a/src/mongo/s/write_ops/batch_write_exec.cpp
+++ b/src/mongo/s/write_ops/batch_write_exec.cpp
@@ -104,6 +104,25 @@ bool hasTransientTransactionError(const BatchedCommandResponse& response) {
 // applies when no writes are occurring and metadata is not changing on reload.
 const int kMaxRoundsWithoutProgress(5);
 
+/**
+ * Provides the write concern with which child batches have to be internally submitted.
+ */
+boost::optional<WriteConcernOptions> getWriteConcernForChildBatch(OperationContext* opCtx) {
+    // Per-operation write concern is not supported in transactions.
+    if (TransactionRouter::get(opCtx)) {
+        return boost::none;
+    }
+
+    // Retrieve the WC specified by the remote client; in case of "fire and forget" request, the WC
+    // needs to be upgraded to "w: 1" for the sharding protocol to correctly handle internal
+    // writeErrors.
+    auto wc = opCtx->getWriteConcern();
+    if (!wc.requiresWriteAcknowledgement()) {
+        wc.w = 1;
+    }
+
+    return wc;
+}
 }  // namespace
 
 void BatchWriteExec::executeBatch(OperationContext* opCtx,
@@ -195,6 +214,7 @@ void BatchWriteExec::executeBatch(OperationContext* opCtx,
             //
 
             std::vector<AsyncRequestsSender::Request> requests;
+            const auto wcSettingForChildBatch = getWriteConcernForChildBatch(opCtx);
 
             // Get as many batches as we can at once
             for (auto&& childBatch : childBatches) {
@@ -217,6 +237,10 @@ void BatchWriteExec::executeBatch(OperationContext* opCtx,
                     BSONObjBuilder requestBuilder;
                     shardBatchRequest.serialize(&requestBuilder);
                     logical_session_id_helpers::serializeLsidAndTxnNumber(opCtx, &requestBuilder);
+                    if (wcSettingForChildBatch) {
+                        requestBuilder.append(WriteConcernOptions::kWriteConcernField,
+                                              wcSettingForChildBatch->toBSON());
+                    }
 
                     return requestBuilder.obj();
                 }();
diff --git a/src/mongo/s/write_ops/batch_write_exec_test.cpp b/src/mongo/s/write_ops/batch_write_exec_test.cpp
index a2dc99859cc..3e90e5b65d3 100644
--- a/src/mongo/s/write_ops/batch_write_exec_test.cpp
+++ b/src/mongo/s/write_ops/batch_write_exec_test.cpp
@@ -397,7 +397,6 @@ TEST_F(BatchWriteExecTest, SingleOpUnordered) {
         insertOp.setDocuments({BSON("x" << 1)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     // Do single-target, single doc batch write op
     auto future = launchAsync([&] {
@@ -436,7 +435,6 @@ TEST_F(BatchWriteExecTest, SingleUpdateTargetsShardWithLet) {
                                                                        << "100")))});
         return updateOp;
     }());
-    updateRequest.setWriteConcern(BSONObj());
 
     const static auto epoch = OID::gen();
     const static Timestamp timestamp(2);
@@ -522,7 +520,6 @@ TEST_F(BatchWriteExecTest, SingleDeleteTargetsShardWithLet) {
         deleteOp.setDeletes(std::vector{write_ops::DeleteOpEntry(q, false)});
         return deleteOp;
     }());
-    deleteRequest.setWriteConcern(BSONObj());
 
     const static auto epoch = OID::gen();
 
@@ -608,7 +605,6 @@ TEST_F(BatchWriteExecTest, MultiOpLargeOrdered) {
         insertOp.setDocuments(docsToInsert);
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     auto future = launchAsync([&] {
         BatchedCommandResponse response;
@@ -641,7 +637,6 @@ TEST_F(BatchWriteExecTest, SingleOpUnorderedError) {
         insertOp.setDocuments({BSON("x" << 1)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     auto future = launchAsync([&] {
         BatchedCommandResponse response;
@@ -682,7 +677,6 @@ TEST_F(BatchWriteExecTest, MultiOpLargeUnorderedWithStaleShardVersionError) {
         insertOp.setDocuments(docsToInsert);
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     auto future = launchAsync([&] {
         BatchedCommandResponse response;
@@ -714,7 +708,6 @@ TEST_F(BatchWriteExecTest, StaleShardVersionReturnedFromBatchWithSingleMultiWrit
             write_ops::UpdateModification::parseFromClassicUpdate(BSON("Key" << 100)))});
         return updateOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     const static auto epoch = OID::gen();
     const static Timestamp timestamp(2);
@@ -811,7 +804,6 @@ TEST_F(BatchWriteExecTest, MultiOpLargeUnorderedWithCannotRefreshError) {
         insertOp.setDocuments(docsToInsert);
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     auto future = launchAsync([&] {
         BatchedCommandResponse response;
@@ -848,7 +840,6 @@ TEST_F(BatchWriteExecTest,
                 write_ops::UpdateModification::parseFromClassicUpdate(BSON("y" << 2)))});
         return updateOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     const static auto epoch = OID::gen();
     const static Timestamp timestamp(2);
@@ -952,7 +943,6 @@ TEST_F(BatchWriteExecTest, RetryableErrorReturnedFromMultiWriteWithShard1Firs) {
                 write_ops::UpdateModification::parseFromClassicUpdate(BSON("y" << 2)))});
         return updateOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     const static auto epoch = OID::gen();
     const static Timestamp timestamp(2);
@@ -1066,7 +1056,6 @@ TEST_F(BatchWriteExecTest, RetryableErrorReturnedFromMultiWriteWithShard1FirstOK
                 write_ops::UpdateModification::parseFromClassicUpdate(BSON("y" << 2)))});
         return updateOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     const static auto epoch = OID::gen();
     const static Timestamp timestamp(2);
@@ -1176,7 +1165,6 @@ TEST_F(BatchWriteExecTest, RetryableErrorReturnedFromWriteWithShard1SSVShard2OK)
             write_ops::UpdateModification::parseFromClassicUpdate(BSON("x" << 1)))});
         return updateOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     const static auto epoch = OID::gen();
     const static Timestamp timestamp(2);
@@ -1276,7 +1264,6 @@ TEST_F(BatchWriteExecTest, StaleShardOp) {
         insertOp.setDocuments({BSON("x" << 1)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     // Execute request
     auto future = launchAsync([&] {
@@ -1308,7 +1295,6 @@ TEST_F(BatchWriteExecTest, MultiStaleShardOp) {
         insertOp.setDocuments({BSON("x" << 1)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     auto future = launchAsync([&] {
         BatchedCommandResponse response;
@@ -1346,7 +1332,6 @@ TEST_F(BatchWriteExecTest, TooManyStaleShardOp) {
         insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     auto future = launchAsync([&] {
         BatchedCommandResponse response;
@@ -1381,7 +1366,6 @@ TEST_F(BatchWriteExecTest, StaleDbOp) {
         insertOp.setDocuments({BSON("x" << 1)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     // Execute request
     auto future = launchAsync([&] {
@@ -1413,7 +1397,6 @@ TEST_F(BatchWriteExecTest, MultiStaleDbOp) {
         insertOp.setDocuments({BSON("x" << 1)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     auto future = launchAsync([&] {
         BatchedCommandResponse response;
@@ -1451,7 +1434,6 @@ TEST_F(BatchWriteExecTest, TooManyStaleDbOp) {
         insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     auto future = launchAsync([&] {
         BatchedCommandResponse response;
@@ -1486,7 +1468,6 @@ TEST_F(BatchWriteExecTest, MultiCannotRefreshShardOp) {
         insertOp.setDocuments({BSON("x" << 1)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     auto future = launchAsync([&] {
         BatchedCommandResponse response;
@@ -1522,7 +1503,6 @@ TEST_F(BatchWriteExecTest, TooManyCannotRefreshShardOp) {
         insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     auto future = launchAsync([&] {
         BatchedCommandResponse response;
@@ -1566,7 +1546,6 @@ TEST_F(BatchWriteExecTest, RetryableWritesLargeBatch) {
         insertOp.setDocuments(docsToInsert);
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     operationContext()->setLogicalSessionId(makeLogicalSessionIdForTest());
     operationContext()->setTxnNumber(5);
@@ -1601,7 +1580,6 @@ TEST_F(BatchWriteExecTest, RetryableErrorNoTxnNumber) {
         insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     BatchedCommandResponse retryableErrResponse;
     retryableErrResponse.setStatus({ErrorCodes::NotWritablePrimary, "mock retryable error"});
@@ -1640,7 +1618,6 @@ TEST_F(BatchWriteExecTest, RetryableErrorTxnNumber) {
         insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     operationContext()->setLogicalSessionId(makeLogicalSessionIdForTest());
     operationContext()->setTxnNumber(5);
@@ -1678,7 +1655,6 @@ TEST_F(BatchWriteExecTest, NonRetryableErrorTxnNumber) {
         insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     operationContext()->setLogicalSessionId(makeLogicalSessionIdForTest());
     operationContext()->setTxnNumber(5);
@@ -1720,7 +1696,6 @@ TEST_F(BatchWriteExecTest, StaleEpochIsNotRetryable) {
         insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     operationContext()->setLogicalSessionId(makeLogicalSessionIdForTest());
     operationContext()->setTxnNumber(5);
@@ -1748,6 +1723,249 @@ TEST_F(BatchWriteExecTest, StaleEpochIsNotRetryable) {
     future.default_timed_get();
 }
 
+TEST_F(BatchWriteExecTest, FireAndForgetBatchInsertGetsReplyWithOnlyOkStatus) {
+    const int kNumDocsToInsert = 5;
+    const std::string kDocValue("sample");
+
+    std::vector<BSONObj> docsToInsert;
+    docsToInsert.reserve(kNumDocsToInsert);
+    for (int i = 0; i < kNumDocsToInsert; i++) {
+        docsToInsert.push_back(BSON("_id" << i << "otherField" << kDocValue));
+    }
+
+    BatchedCommandRequest request([&] {
+        write_ops::InsertCommandRequest insertOp(nss);
+        insertOp.setWriteCommandRequestBase([] {
+            write_ops::WriteCommandRequestBase writeCommandBase;
+            writeCommandBase.setOrdered(true);
+            return writeCommandBase;
+        }());
+        insertOp.setDocuments(docsToInsert);
+        return insertOp;
+    }());
+
+    auto future = launchAsync([&] {
+        BatchedCommandResponse response;
+        BatchWriteExecStats stats;
+        // Set Unacknowledged WC for a "fire & forget" request
+        auto opCtx = operationContext();
+        opCtx->setWriteConcern(
+            WriteConcernOptions::parse(WriteConcernOptions::Unacknowledged).getValue());
+        BatchWriteExec::executeBatch(opCtx, singleShardNSTargeter, request, &response, &stats);
+
+        // The reply should only contain an OK status, without any further detail on
+        // the ops actually executed across the cluster.
+        BatchedCommandResponse expectedReplyToFireAndForgetRequest;
+        expectedReplyToFireAndForgetRequest.setStatus(Status::OK());
+        ASSERT_EQUALS(response.toBSON().woCompare(expectedReplyToFireAndForgetRequest.toBSON()), 0);
+    });
+
+    expectInsertsReturnSuccess(docsToInsert.begin(), docsToInsert.end());
+
+    future.default_timed_get();
+}
+
+TEST_F(BatchWriteExecTest, FireAndForgetBatchUpdateGetsReplyWithOnlyOkStatus) {
+    BatchedCommandRequest request([&] {
+        write_ops::UpdateCommandRequest updateOp(nss);
+        updateOp.setWriteCommandRequestBase([] {
+            write_ops::WriteCommandRequestBase writeCommandBase;
+            writeCommandBase.setOrdered(false);
+            return writeCommandBase;
+        }());
+        updateOp.setUpdates(std::vector{write_ops::UpdateOpEntry(
+            BSON("_id" << 100),
+            write_ops::UpdateModification::parseFromClassicUpdate(BSON("Key" << 100)))});
+        return updateOp;
+    }());
+
+    const static auto epoch = OID::gen();
+    const static Timestamp timestamp(2);
+
+    // This allows the batch to target each write operation to perform this test
+    class MultiShardTargeter : public MockNSTargeter {
+    public:
+        using MockNSTargeter::MockNSTargeter;
+
+        std::vector<ShardEndpoint> targetUpdate(OperationContext* opCtx,
+                                                const BatchItemRef& itemRef) const override {
+            if (targetAll) {
+                return std::vector{
+                    ShardEndpoint(
+                        kShardName1, ChunkVersion(100, 200, epoch, timestamp), boost::none),
+                    ShardEndpoint(
+                        kShardName2, ChunkVersion(101, 200, epoch, timestamp), boost::none)};
+            } else {
+                return std::vector{ShardEndpoint(
+                    kShardName2, ChunkVersion(101, 200, epoch, timestamp), boost::none)};
+            }
+        }
+
+        bool targetAll = true;
+    };
+
+    MultiShardTargeter multiShardNSTargeter(
+        nss,
+        {MockRange(
+             ShardEndpoint(kShardName1, ChunkVersion(100, 200, epoch, timestamp), boost::none),
+             BSON("sk" << MINKEY),
+             BSON("sk" << 10)),
+         MockRange(
+             ShardEndpoint(kShardName2, ChunkVersion(101, 200, epoch, timestamp), boost::none),
+             BSON("sk" << 10),
+             BSON("sk" << MAXKEY))});
+    auto future = launchAsync([&] {
+        // Set Unacknowledged WC for a "fire & forget" request
+        auto opCtx = operationContext();
+        opCtx->setWriteConcern(
+            WriteConcernOptions::parse(WriteConcernOptions::Unacknowledged).getValue());
+
+        BatchedCommandResponse response;
+        BatchWriteExecStats stats;
+        BatchWriteExec::executeBatch(opCtx, multiShardNSTargeter, request, &response, &stats);
+        return response;
+    });
+
+    onCommandForPoolExecutor([&](const RemoteCommandRequest& request) {
+        ASSERT_EQ(kTestShardHost1, request.target);
+
+        BatchedCommandResponse response;
+        response.setStatus(Status::OK());
+        response.setNModified(1);
+
+        return response.toBSON();
+    });
+
+    onCommandForPoolExecutor([&](const RemoteCommandRequest& request) {
+        ASSERT_EQ(kTestShardHost2, request.target);
+
+        BatchedCommandResponse response;
+        response.setStatus(Status::OK());
+        response.setNModified(0);
+        response.addToErrDetails(
+            write_ops::WriteError(0,
+                                  Status(StaleConfigInfo(nss,
+                                                         ChunkVersion(101, 200, epoch, timestamp),
+                                                         ChunkVersion(105, 200, epoch, timestamp),
+                                                         ShardId(kShardName2)),
+                                         "Stale error")));
+        return response.toBSON();
+    });
+
+    onCommandForPoolExecutor([&](const RemoteCommandRequest& request) {
+        ASSERT_EQ(kTestShardHost2, request.target);
+
+        BatchedCommandResponse response;
+        response.setStatus(Status::OK());
+        response.setNModified(2);
+
+        return response.toBSON();
+    });
+
+    // The reply should only contain an OK status, without any further detail on
+    // the ops actually executed across the cluster.
+    // (Despite of a "fire and forget" request, child batches still need to be internally processed
+    // before returning a reply).
+    auto response = future.default_timed_get();
+    BatchedCommandResponse expectedReplyToFireAndForgetRequest;
+    expectedReplyToFireAndForgetRequest.setStatus(Status::OK());
+    ASSERT_EQUALS(response.toBSON().woCompare(expectedReplyToFireAndForgetRequest.toBSON()), 0);
+}
+
+TEST_F(BatchWriteExecTest, FireAndForgetBatchDeleteGetsReplyWithOnlyOkStatus) {
+    // Try to update the single doc where a let param is used in the shard key.
+    const auto let = BSON("y" << 100);
+    const auto rtc = LegacyRuntimeConstants{Date_t::now(), Timestamp(1, 1)};
+    const auto q = BSON("x"
+                        << "$$y");
+    BatchedCommandRequest deleteRequest([&] {
+        write_ops::DeleteCommandRequest deleteOp(nss);
+        deleteOp.setWriteCommandRequestBase([] {
+            write_ops::WriteCommandRequestBase writeCommandBase;
+            writeCommandBase.setOrdered(false);
+            return writeCommandBase;
+        }());
+        deleteOp.setLet(let);
+        deleteOp.setLegacyRuntimeConstants(rtc);
+        deleteOp.setDeletes(std::vector{write_ops::DeleteOpEntry(q, false)});
+        return deleteOp;
+    }());
+
+    const static auto epoch = OID::gen();
+
+    class MultiShardTargeter : public MockNSTargeter {
+    public:
+        using MockNSTargeter::MockNSTargeter;
+
+    protected:
+        std::vector<ShardEndpoint> targetDelete(OperationContext* opCtx,
+                                                const BatchItemRef& itemRef) const override {
+            return std::vector{ShardEndpoint(
+                kShardName2, ChunkVersion(101, 200, epoch, Timestamp(1, 1)), boost::none)};
+        }
+    };
+
+    MultiShardTargeter multiShardNSTargeter(
+        nss,
+        {MockRange(ShardEndpoint(
+                       kShardName1, ChunkVersion(100, 200, epoch, Timestamp(1, 1)), boost::none),
+                   BSON("x" << MINKEY),
+                   BSON("x" << 0)),
+         MockRange(ShardEndpoint(
+                       kShardName2, ChunkVersion(101, 200, epoch, Timestamp(1, 1)), boost::none),
+                   BSON("x" << 0),
+                   BSON("x" << MAXKEY))});
+
+    auto future = launchAsync([&] {
+        BatchedCommandResponse response;
+        BatchWriteExecStats stats;
+
+        // Set Unacknowledged WC for a "fire & forget" request
+        auto opCtx = operationContext();
+        opCtx->setWriteConcern(
+            WriteConcernOptions::parse(WriteConcernOptions::Unacknowledged).getValue());
+
+        BatchWriteExec::executeBatch(opCtx, multiShardNSTargeter, deleteRequest, &response, &stats);
+
+        return response;
+    });
+
+    // The update will hit the first shard.
+    onCommandForPoolExecutor(
+        [&](const RemoteCommandRequest& request) {
+            ASSERT_EQ(kTestShardHost2, request.target);
+
+            BatchedCommandResponse response;
+            response.setStatus(Status::OK());
+
+            // Check that let params are propagated to shards.
+            const auto opMsgRequest(OpMsgRequest::fromDBAndBody(request.dbname, request.cmdObj));
+            const auto actualBatchedUpdate(BatchedCommandRequest::parseDelete(opMsgRequest));
+            ASSERT_BSONOBJ_EQ(let, actualBatchedUpdate.getLet().value_or(BSONObj()));
+            ASSERT_EQUALS(actualBatchedUpdate.getLegacyRuntimeConstants()->getLocalNow(),
+                          rtc.getLocalNow());
+            ASSERT_EQUALS(actualBatchedUpdate.getLegacyRuntimeConstants()->getClusterTime(),
+                          rtc.getClusterTime());
+
+            // Check that let params are only forwarded and not evaluated.
+            auto expectedQ = BSON("x"
+                                  << "$$y");
+            for (auto&& u : actualBatchedUpdate.getDeleteRequest().getDeletes())
+                ASSERT_BSONOBJ_EQ(expectedQ, u.getQ());
+
+            return response.toBSON();
+        });
+
+    // The reply should only contain an OK status, without any further detail on
+    // the ops actually executed across the cluster.
+    // (Despite of a "fire and forget" request, child batches still need to be internally processed
+    // before returning a reply).
+    auto response = future.default_timed_get();
+    BatchedCommandResponse expectedReplyToFireAndForgetRequest;
+    expectedReplyToFireAndForgetRequest.setStatus(Status::OK());
+    ASSERT_EQUALS(response.toBSON().woCompare(expectedReplyToFireAndForgetRequest.toBSON()), 0);
+}
+
 TEST_F(BatchWriteExecTest, TenantMigrationAbortedErrorOrderedOp) {
     const std::vector<BSONObj> expected{BSON("x" << 1), BSON("x" << 2), BSON("x" << 3)};
     BatchedCommandRequest request([&] {
@@ -1760,7 +1978,6 @@ TEST_F(BatchWriteExecTest, TenantMigrationAbortedErrorOrderedOp) {
         insertOp.setDocuments(expected);
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     // Execute request
     auto future = launchAsync([&] {
@@ -1791,7 +2008,6 @@ TEST_F(BatchWriteExecTest, TenantMigrationAbortedErrorUnorderedOp) {
         insertOp.setDocuments(expected);
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     // Execute request
     auto future = launchAsync([&] {
@@ -1822,7 +2038,6 @@ TEST_F(BatchWriteExecTest, MultipleTenantMigrationAbortedErrorUnorderedOp) {
         insertOp.setDocuments(expected);
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     const int numTenantMigrationAbortedErrors = 3;
 
@@ -1857,7 +2072,6 @@ TEST_F(BatchWriteExecTest, MultipleTenantMigrationAbortedErrorOrderedOp) {
         insertOp.setDocuments(expected);
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     const int numTenantMigrationAbortedErrors = 3;
 
@@ -1892,7 +2106,6 @@ TEST_F(BatchWriteExecTest, PartialTenantMigrationAbortedErrorOrderedOp) {
         insertOp.setDocuments(expected);
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     // Execute request
     auto future = launchAsync([&] {
@@ -1925,7 +2138,6 @@ TEST_F(BatchWriteExecTest, PartialTenantMigrationErrorUnorderedOp) {
         insertOp.setDocuments(expected);
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     // Execute request
     auto future = launchAsync([&] {
@@ -2010,7 +2222,6 @@ TEST_F(BatchWriteExecTargeterErrorTest, TargetedFailedAndErrorResponse) {
             write_ops::UpdateModification::parseFromClassicUpdate(BSON("Key" << 100)))});
         return updateOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     const static auto epoch = OID::gen();
     const static Timestamp timestamp(2);
@@ -2146,7 +2357,6 @@ TEST_F(BatchWriteExecTransactionTargeterErrorTest, TargetedFailedAndErrorRespons
             write_ops::UpdateModification::parseFromClassicUpdate(BSON("Key" << 100)))});
         return updateOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     const static auto epoch = OID::gen();
     const static Timestamp timestamp(2);
@@ -2290,7 +2500,6 @@ TEST_F(BatchWriteExecTransactionMultiShardTest, TargetedSucceededAndErrorRespons
             write_ops::UpdateModification::parseFromClassicUpdate(BSON("Key" << 100)))});
         return updateOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     const static auto epoch = OID::gen();
     const static Timestamp timestamp(2);
@@ -2481,7 +2690,6 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchThrows_CommandError) {
         insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     auto future = launchAsync([&] {
         BatchedCommandResponse response;
@@ -2513,7 +2721,6 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_WriteError) {
         insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     auto future = launchAsync([&] {
         BatchedCommandResponse response;
@@ -2543,7 +2750,6 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_WriteErrorOrdered) {
         insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     auto future = launchAsync([&] {
         BatchedCommandResponse response;
@@ -2573,7 +2779,6 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_WriteErrorFromBusyCache)
         insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     auto future = launchAsync([&] {
         BatchedCommandResponse response;
@@ -2603,7 +2808,6 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_WriteErrorOrderedFromBusy
         insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     auto future = launchAsync([&] {
         BatchedCommandResponse response;
@@ -2633,7 +2837,6 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_TransientTxnError) {
         insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     auto future = launchAsync([&] {
         BatchedCommandResponse response;
@@ -2659,7 +2862,6 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_DispatchError) {
         insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     auto future = launchAsync([&] {
         BatchedCommandResponse response;
@@ -2691,7 +2893,6 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_TransientDispatchError) {
         insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
         return insertOp;
     }());
-    request.setWriteConcern(BSONObj());
 
     auto future = launchAsync([&] {
         BatchedCommandResponse response;
diff --git a/src/mongo/s/write_ops/batch_write_op.cpp b/src/mongo/s/write_ops/batch_write_op.cpp
index f93e5302e4f..ed67f6c0088 100644
--- a/src/mongo/s/write_ops/batch_write_op.cpp
+++ b/src/mongo/s/write_ops/batch_write_op.cpp
@@ -568,6 +568,9 @@ BatchedCommandRequest BatchWriteOp::buildBatchRequest(const TargetedWriteBatch&
             wcb.setStmtIds(std::move(stmtIdsForOp));
         }
 
+        wcb.setBypassEmptyTsReplacement(
+            _clientRequest.getWriteCommandRequestBase().getBypassEmptyTsReplacement());
+
         return wcb;
     }());
 
@@ -580,19 +583,6 @@ BatchedCommandRequest BatchWriteOp::buildBatchRequest(const TargetedWriteBatch&
     if (dbVersion)
         request.setDbVersion(*dbVersion);
 
-    if (_clientRequest.hasWriteConcern()) {
-        if (_clientRequest.requiresWriteAcknowledgement()) {
-            request.setWriteConcern(_clientRequest.getWriteConcern());
-        } else {
-            // Mongos needs to send to the shard with w > 0 so it will be able to see the
-            // writeErrors
-            request.setWriteConcern(upgradeWriteConcern(_clientRequest.getWriteConcern()));
-        }
-    } else if (!TransactionRouter::get(_opCtx)) {
-        // Apply the WC from the opCtx (except if in a transaction).
-        request.setWriteConcern(_opCtx->getWriteConcern().toBSON());
-    }
-
     return request;
 }
 
diff --git a/src/mongo/s/write_ops/batch_write_op_test.cpp b/src/mongo/s/write_ops/batch_write_op_test.cpp
index 80ea3818494..b21e64c6d53 100644
--- a/src/mongo/s/write_ops/batch_write_op_test.cpp
+++ b/src/mongo/s/write_ops/batch_write_op_test.cpp
@@ -235,7 +235,6 @@ TEST_F(BatchWriteOpTest, SingleWriteConcernErrorOrdered) {
         insertOp.setDocuments({BSON("x" << 1)});
         return insertOp;
     }());
-    request.setWriteConcern(BSON("w" << 3));
 
     BatchWriteOp batchOp(_opCtx, request);
 
@@ -247,7 +246,6 @@ TEST_F(BatchWriteOpTest, SingleWriteConcernErrorOrdered) {
 
     BatchedCommandRequest targetBatch =
         batchOp.buildBatchRequest(*targeted.begin()->second, targeter);
-    ASSERT(targetBatch.getWriteConcern().woCompare(request.getWriteConcern()) == 0);
 
     BatchedCommandResponse response;
     buildResponse(1, &response);
@@ -1029,7 +1027,6 @@ TEST_F(BatchWriteOpTest, MultiOpErrorAndWriteConcernErrorUnordered) {
         insertOp.setDocuments({BSON("x" << 1), BSON("x" << 1)});
         return insertOp;
     }());
-    request.setWriteConcern(BSON("w" << 3));
 
     BatchWriteOp batchOp(_opCtx, request);
 
@@ -1072,7 +1069,6 @@ TEST_F(BatchWriteOpTest, SingleOpErrorAndWriteConcernErrorOrdered) {
         updateOp.setUpdates({buildUpdate(BSON("x" << GTE << -1 << LT << 2), true)});
         return updateOp;
     }());
-    request.setWriteConcern(BSON("w" << 3));
 
     BatchWriteOp batchOp(_opCtx, request);
 
@@ -1415,7 +1411,6 @@ TEST_F(BatchWriteOpTest, MultiOpTwoWCErrors) {
         insertOp.setDocuments({BSON("x" << -1), BSON("x" << 2)});
         return insertOp;
     }());
-    request.setWriteConcern(BSON("w" << 3));
 
     BatchWriteOp batchOp(_opCtx, request);
 
diff --git a/src/mongo/s/write_ops/batched_command_request.cpp b/src/mongo/s/write_ops/batched_command_request.cpp
index e6b3c970592..185857d6acc 100644
--- a/src/mongo/s/write_ops/batched_command_request.cpp
+++ b/src/mongo/s/write_ops/batched_command_request.cpp
@@ -38,8 +38,6 @@
 namespace mongo {
 namespace {
 
-const auto kWriteConcern = "writeConcern"_sd;
-
 template <class T>
 BatchedCommandRequest constructBatchedCommandRequest(const OpMsgRequest& request) {
     auto batchRequest = BatchedCommandRequest{T::parse(request)};
@@ -53,16 +51,6 @@ BatchedCommandRequest constructBatchedCommandRequest(const OpMsgRequest& request
         batchRequest.setShardVersion(shardVersion);
     }
 
-    auto writeConcernField = request.body[kWriteConcern];
-    if (!writeConcernField.eoo()) {
-        auto wcObj = writeConcernField.Obj();
-        // Client write concerns without 'w' fields should be filled with the default write concern,
-        // which should be populated later to the operation context during the command setup phase.
-        if (wcObj.hasElement("w")) {
-            batchRequest.setWriteConcern(wcObj);
-        }
-    }
-
     // The 'isTimeseriesNamespace' is an internal parameter used for communication between mongos
     // and mongod.
     auto isTimeseriesNamespace =
@@ -178,19 +166,9 @@ const boost::optional<BSONObj>& BatchedCommandRequest::getLet() const {
     return _visit(Visitor{});
 };
 
-bool BatchedCommandRequest::requiresWriteAcknowledgement() const {
-    if (!hasWriteConcern()) {
-        return true;
-    }
-
-    BSONObj writeConcern = getWriteConcern();
-    BSONElement wElem = writeConcern["w"];
-    if (!wElem.isNumber() || wElem.Number() != 0) {
-        return true;
-    }
-
-    return false;
-}
+const OptionalBool& BatchedCommandRequest::getBypassEmptyTsReplacement() const {
+    return _visit([](auto&& op) -> decltype(auto) { return op.getBypassEmptyTsReplacement(); });
+};
 
 const write_ops::WriteCommandRequestBase& BatchedCommandRequest::getWriteCommandRequestBase()
     const {
@@ -211,10 +189,6 @@ void BatchedCommandRequest::serialize(BSONObjBuilder* builder) const {
     if (_dbVersion) {
         builder->append("databaseVersion", _dbVersion->toBSON());
     }
-
-    if (_writeConcern) {
-        builder->append(kWriteConcern, *_writeConcern);
-    }
 }
 
 BSONObj BatchedCommandRequest::toBSON() const {
diff --git a/src/mongo/s/write_ops/batched_command_request.h b/src/mongo/s/write_ops/batched_command_request.h
index 1834fe4286b..0bcb51a3556 100644
--- a/src/mongo/s/write_ops/batched_command_request.h
+++ b/src/mongo/s/write_ops/batched_command_request.h
@@ -115,25 +115,6 @@ public:
 
     std::size_t sizeWriteOps() const;
 
-    void setWriteConcern(const BSONObj& writeConcern) {
-        _writeConcern = writeConcern.getOwned();
-    }
-
-    void unsetWriteConcern() {
-        _writeConcern = boost::none;
-    }
-
-    bool hasWriteConcern() const {
-        return _writeConcern.is_initialized();
-    }
-
-    const BSONObj& getWriteConcern() const {
-        invariant(_writeConcern);
-        return *_writeConcern;
-    }
-
-    bool requiresWriteAcknowledgement() const;
-
     void setShardVersion(ChunkVersion shardVersion) {
         _shardVersion = std::move(shardVersion);
     }
@@ -168,6 +149,7 @@ public:
 
     const boost::optional<LegacyRuntimeConstants>& getLegacyRuntimeConstants() const;
     const boost::optional<BSONObj>& getLet() const;
+    const OptionalBool& getBypassEmptyTsReplacement() const;
 
     const write_ops::WriteCommandRequestBase& getWriteCommandRequestBase() const;
     void setWriteCommandRequestBase(write_ops::WriteCommandRequestBase writeCommandBase);
@@ -250,8 +232,6 @@ private:
 
     boost::optional<ChunkVersion> _shardVersion;
     boost::optional<DatabaseVersion> _dbVersion;
-
-    boost::optional<BSONObj> _writeConcern;
 };
 
 /**
diff --git a/src/mongo/s/write_ops/batched_command_request_test.cpp b/src/mongo/s/write_ops/batched_command_request_test.cpp
index 9a5e968f10d..be0728b1533 100644
--- a/src/mongo/s/write_ops/batched_command_request_test.cpp
+++ b/src/mongo/s/write_ops/batched_command_request_test.cpp
@@ -92,14 +92,12 @@ TEST(BatchedCommandRequest, InsertCloneWithIds) {
         insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
         return insertOp;
     }());
-    batchedRequest.setWriteConcern(BSON("w" << 2));
 
     const auto clonedRequest(BatchedCommandRequest::cloneInsertWithIds(std::move(batchedRequest)));
 
     ASSERT_EQ("xyz.abc", clonedRequest.getNS().ns());
     ASSERT(clonedRequest.getWriteCommandRequestBase().getOrdered());
     ASSERT(clonedRequest.getWriteCommandRequestBase().getBypassDocumentValidation());
-    ASSERT_BSONOBJ_EQ(BSON("w" << 2), clonedRequest.getWriteConcern());
 
     const auto& insertDocs = clonedRequest.getInsertRequest().getDocuments();
     ASSERT_EQ(2u, insertDocs.size());
diff --git a/src/mongo/s/write_ops/write_op.cpp b/src/mongo/s/write_ops/write_op.cpp
index 236c7efbc94..6b33d3b9312 100644
--- a/src/mongo/s/write_ops/write_op.cpp
+++ b/src/mongo/s/write_ops/write_op.cpp
@@ -29,7 +29,24 @@
 
 #include "mongo/s/write_ops/write_op.h"
 
+
+#include <absl/container/flat_hash_set.h>
+#include <algorithm>
+#include <boost/move/utility_core.hpp>
+#include <boost/none.hpp>
+#include <boost/optional/optional.hpp>
+#include <ostream>
+#include <string>
+
+#include "mongo/base/error_codes.h"
+#include "mongo/base/status.h"
+#include "mongo/bson/bsonobjbuilder.h"
+#include "mongo/db/catalog/collection_uuid_mismatch_info.h"
+#include "mongo/db/stats/counters.h"
+#include "mongo/s/sharding_feature_flags_gen.h"
 #include "mongo/s/transaction_router.h"
+#include "mongo/s/write_ops/batch_write_op.h"
+#include "mongo/s/write_ops/batched_command_request.h"
 #include "mongo/util/assert_util.h"
 
 namespace mongo {
@@ -79,6 +96,16 @@ write_ops::WriteError combineOpErrors(const std::vector<ChildWriteOp const*>& er
                                  Status(MultipleErrorsOccurredInfo(errB.arr()), msg.str()));
 }
 
+bool isSafeToIgnoreErrorInPartiallyAppliedOp(write_ops::WriteError& error) {
+    // UUID mismatch errors are safe to ignore if the actualCollection is null in conjuntion with
+    // other successful operations. This is true because it means we wrongly targeted a non-owning
+    // shard with the operation and we wouldn't have applied any modifications anyway.
+    //
+    // Note this is only safe if we're using ShardVersion::IGNORED since we're ignoring any
+    // placement concern and broadcasting to all shards.
+    return error.getStatus().code() == ErrorCodes::CollectionUUIDMismatch &&
+        !error.getStatus().extraInfo<CollectionUUIDMismatchInfo>()->actualCollection();
+}
 }  // namespace
 
 const BatchItemRef& WriteOp::getWriteItem() const {
@@ -182,7 +209,29 @@ void WriteOp::_updateOpState() {
         _state = WriteOpState_Ready;
     } else if (!childErrors.empty()) {
         _error = combineOpErrors(childErrors);
-        _state = WriteOpState_Error;
+        bool isTargetingAllShardsWithSVIgnored =
+            childErrors.front()
+                ->endpoint->shardVersion
+                .map([&](const auto& cv) { return ChunkVersion::isIgnoredVersion(cv); })
+                .get_value_or(false);
+        // There are errors that are safe to ignore if they were correctly applied to other shards
+        // and we're using ShardVersion::IGNORED. They are safe to ignore as they can be interpreted
+        // as no-ops if the shard response had been instead a successful result since they wouldn't
+        // have modified any data. As a result, we can swallow the errors and treat them as a
+        // successful operation.
+        if (isTargetingAllShardsWithSVIgnored && isSafeToIgnoreErrorInPartiallyAppliedOp(*_error) &&
+            !_successfulShardSet.empty()) {
+            if (!hasPendingChild) {
+                _error.reset();
+                _state = WriteOpState_Completed;
+            } else {
+                // As this error is acceptable we wait until all other operations finish to take a
+                // decision.
+                return;
+            }
+        } else {
+            _state = WriteOpState_Error;
+        }
     } else if (hasPendingChild && _inTxn) {
         // Return early here since this means that there were no errors while in txn
         // but there are still ops that have not yet finished.