Update upstream source from tag 'upstream/6.0.0'master

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

16 files changed, 322 insertions, 456 deletions

diff --git a/src/mongo/db/pipeline/process_interface/SConscript b/src/mongo/db/pipeline/process_interface/SConscript
index dc09fa5d114..44b0afe1591 100644
--- a/src/mongo/db/pipeline/process_interface/SConscript
+++ b/src/mongo/db/pipeline/process_interface/SConscript
@@ -47,7 +47,6 @@ env.Library(
         '$BUILD_DIR/mongo/db/catalog/catalog_helpers',
         '$BUILD_DIR/mongo/db/catalog/database_holder',
         '$BUILD_DIR/mongo/db/collection_index_usage_tracker',
-        '$BUILD_DIR/mongo/db/concurrency/exception_util',
         '$BUILD_DIR/mongo/db/concurrency/flow_control_ticketholder',
         '$BUILD_DIR/mongo/db/dbhelpers',
         '$BUILD_DIR/mongo/db/index_builds_coordinator_mongod',
@@ -56,7 +55,6 @@ env.Library(
         '$BUILD_DIR/mongo/db/session_catalog',
         '$BUILD_DIR/mongo/db/stats/fill_locker_info',
         '$BUILD_DIR/mongo/db/storage/backup_cursor_hooks',
-        '$BUILD_DIR/mongo/db/storage/durable_catalog_impl',
         '$BUILD_DIR/mongo/scripting/scripting_common',
     ],
 )
diff --git a/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp b/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp
index 3dfb002c54d..469ce5821aa 100644
--- a/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp
+++ b/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp
@@ -37,7 +37,6 @@
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_uuid_mismatch.h"
 #include "mongo/db/catalog/create_collection.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/document_validation.h"
@@ -46,7 +45,7 @@
 #include "mongo/db/catalog/list_indexes.h"
 #include "mongo/db/catalog/rename_collection.h"
 #include "mongo/db/concurrency/d_concurrency.h"
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/cursor_manager.h"
 #include "mongo/db/db_raii.h"
@@ -201,8 +200,7 @@ std::vector<Document> CommonMongodProcessInterface::getIndexStats(OperationConte
         auto idxCatalog = collection->getIndexCatalog();
         auto idx = idxCatalog->findIndexByName(opCtx,
                                                indexName,
-                                               IndexCatalog::InclusionPolicy::kReady |
-                                                   IndexCatalog::InclusionPolicy::kUnfinished);
+                                               /* includeUnfinishedIndexes */ true);
         uassert(ErrorCodes::IndexNotFound,
                 "Could not find entry in IndexCatalog for index " + indexName,
                 idx);
@@ -310,33 +308,34 @@ std::deque<BSONObj> CommonMongodProcessInterface::listCatalog(OperationContext*
 }
 
 boost::optional<BSONObj> CommonMongodProcessInterface::getCatalogEntry(
-    OperationContext* opCtx,
-    const NamespaceString& ns,
-    const boost::optional<UUID>& collUUID) const {
-
-    // Perform an AutoGetCollection. This will verify that the collection still exists at the given
-    // read concern. If it doesn't and the aggregation has specified a UUID then this acquisition
-    // will fail.
-    AutoGetCollectionForRead coll{opCtx, ns};
-    const auto& collPtr = coll.getCollection();
-    checkCollectionUUIDMismatch(opCtx, ns, collPtr, collUUID);
+    OperationContext* opCtx, const NamespaceString& ns) const {
+    Lock::GlobalLock globalLock{opCtx, MODE_IS};
 
-    if (!collPtr) {
+    auto rs = DurableCatalog::get(opCtx)->getRecordStore();
+    if (!rs) {
         return boost::none;
     }
 
-    auto obj = DurableCatalog::get(opCtx)->getCatalogEntry(opCtx, collPtr->getCatalogId());
+    auto cursor = rs->getCursor(opCtx);
+    while (auto record = cursor->next()) {
+        auto obj = record->data.toBson();
+        if (NamespaceString{obj.getStringField("ns")} != ns) {
+            continue;
+        }
 
-    BSONObjBuilder builder;
-    builder.append("db", ns.db());
-    builder.append("name", ns.coll());
-    builder.append("type", "collection");
-    if (auto shardName = getShardName(opCtx); !shardName.empty()) {
-        builder.append("shard", shardName);
+        BSONObjBuilder builder;
+        builder.append("db", ns.db());
+        builder.append("name", ns.coll());
+        builder.append("type", "collection");
+        if (auto shardName = getShardName(opCtx); !shardName.empty()) {
+            builder.append("shard", shardName);
+        }
+        builder.appendElements(obj);
+
+        return builder.obj();
     }
-    builder.appendElements(obj);
 
-    return builder.obj();
+    return boost::none;
 }
 
 void CommonMongodProcessInterface::appendLatencyStats(OperationContext* opCtx,
@@ -605,8 +604,7 @@ bool CommonMongodProcessInterface::fieldsHaveSupportingUniqueIndex(
         return fieldPaths == std::set<FieldPath>{"_id"};
     }
 
-    auto indexIterator = collection->getIndexCatalog()->getIndexIterator(
-        opCtx, IndexCatalog::InclusionPolicy::kReady);
+    auto indexIterator = collection->getIndexCatalog()->getIndexIterator(opCtx, false);
     while (indexIterator->more()) {
         const IndexCatalogEntry* entry = indexIterator->next();
         if (supportsUniqueKey(expCtx, entry, fieldPaths)) {
@@ -747,6 +745,59 @@ CommonMongodProcessInterface::ensureFieldsUniqueOrResolveDocumentKey(
     return {*fieldPaths, targetCollectionVersion};
 }
 
+write_ops::InsertCommandRequest CommonMongodProcessInterface::buildInsertOp(
+    const NamespaceString& nss, std::vector<BSONObj>&& objs, bool bypassDocValidation) {
+    write_ops::InsertCommandRequest insertOp(nss);
+    insertOp.setDocuments(std::move(objs));
+    insertOp.setWriteCommandRequestBase([&] {
+        write_ops::WriteCommandRequestBase wcb;
+        wcb.setOrdered(false);
+        wcb.setBypassDocumentValidation(bypassDocValidation);
+        return wcb;
+    }());
+    return insertOp;
+}
+
+write_ops::UpdateCommandRequest CommonMongodProcessInterface::buildUpdateOp(
+    const boost::intrusive_ptr<ExpressionContext>& expCtx,
+    const NamespaceString& nss,
+    BatchedObjects&& batch,
+    UpsertType upsert,
+    bool multi) {
+    write_ops::UpdateCommandRequest updateOp(nss);
+    updateOp.setUpdates([&] {
+        std::vector<write_ops::UpdateOpEntry> updateEntries;
+        for (auto&& obj : batch) {
+            updateEntries.push_back([&] {
+                write_ops::UpdateOpEntry entry;
+                auto&& [q, u, c] = obj;
+                entry.setQ(std::move(q));
+                entry.setU(std::move(u));
+                entry.setC(std::move(c));
+                entry.setUpsert(upsert != UpsertType::kNone);
+                entry.setUpsertSupplied(
+                    {{entry.getUpsert(), upsert == UpsertType::kInsertSuppliedDoc}});
+                entry.setMulti(multi);
+                return entry;
+            }());
+        }
+        return updateEntries;
+    }());
+    updateOp.setWriteCommandRequestBase([&] {
+        write_ops::WriteCommandRequestBase wcb;
+        wcb.setOrdered(false);
+        wcb.setBypassDocumentValidation(expCtx->bypassDocumentValidation);
+        return wcb;
+    }());
+    auto [constants, letParams] =
+        expCtx->variablesParseState.transitionalCompatibilitySerialize(expCtx->variables);
+    updateOp.setLegacyRuntimeConstants(std::move(constants));
+    if (!letParams.isEmpty()) {
+        updateOp.setLet(std::move(letParams));
+    }
+    return updateOp;
+}
+
 BSONObj CommonMongodProcessInterface::_convertRenameToInternalRename(
     OperationContext* opCtx,
     const BSONObj& renameCommandObj,
diff --git a/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.h b/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.h
index 62edd6ca7d9..4a02ecce883 100644
--- a/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.h
+++ b/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.h
@@ -60,8 +60,7 @@ public:
     std::deque<BSONObj> listCatalog(OperationContext* opCtx) const final;
 
     boost::optional<BSONObj> getCatalogEntry(OperationContext* opCtx,
-                                             const NamespaceString& ns,
-                                             const boost::optional<UUID>& collUUID) const final;
+                                             const NamespaceString& ns) const final;
 
     void appendLatencyStats(OperationContext* opCtx,
                             const NamespaceString& nss,
@@ -143,6 +142,23 @@ protected:
         const Document& documentKey,
         MakePipelineOptions opts);
 
+    /**
+     * Builds an ordered insert op on namespace 'nss' and documents to be written 'objs'.
+     */
+    write_ops::InsertCommandRequest buildInsertOp(const NamespaceString& nss,
+                                                  std::vector<BSONObj>&& objs,
+                                                  bool bypassDocValidation);
+
+    /**
+     * Builds an ordered update op on namespace 'nss' with update entries contained in 'batch'.
+     */
+    write_ops::UpdateCommandRequest buildUpdateOp(
+        const boost::intrusive_ptr<ExpressionContext>& expCtx,
+        const NamespaceString& nss,
+        BatchedObjects&& batch,
+        UpsertType upsert,
+        bool multi);
+
     BSONObj _reportCurrentOpForClient(OperationContext* opCtx,
                                       Client* client,
                                       CurrentOpTruncateMode truncateOps,
diff --git a/src/mongo/db/pipeline/process_interface/common_process_interface.h b/src/mongo/db/pipeline/process_interface/common_process_interface.h
index 55dc54837b1..513edd5a6f4 100644
--- a/src/mongo/db/pipeline/process_interface/common_process_interface.h
+++ b/src/mongo/db/pipeline/process_interface/common_process_interface.h
@@ -32,7 +32,6 @@
 #include <vector>
 
 #include "mongo/bson/bsonobj.h"
-#include "mongo/db/pipeline/expression_context.h"
 #include "mongo/db/pipeline/process_interface/mongo_process_interface.h"
 
 namespace mongo {
@@ -47,70 +46,6 @@ public:
     virtual ~CommonProcessInterface() = default;
 
     /**
-     * Estimates the size of writes that will be executed on the current node. Note that this
-     * does not account for the full size of an update statement because in the case of local
-     * writes, we will not have to serialize to BSON and are therefore not subject to the 16MB
-     * BSONObj size limit.
-     */
-    class LocalWriteSizeEstimator final : public WriteSizeEstimator {
-    public:
-        int estimateInsertHeaderSize(
-            const write_ops::InsertCommandRequest& insertReq) const override {
-            return 0;
-        }
-
-        int estimateUpdateHeaderSize(
-            const write_ops::UpdateCommandRequest& insertReq) const override {
-            return 0;
-        }
-
-        int estimateInsertSizeBytes(const BSONObj& insert) const override {
-            return insert.objsize();
-        }
-
-        int estimateUpdateSizeBytes(const BatchObject& batchObject,
-                                    UpsertType type) const override {
-            int size = std::get<write_ops::UpdateModification>(batchObject).objsize();
-            if (auto vars = std::get<boost::optional<BSONObj>>(batchObject)) {
-                size += vars->objsize();
-            }
-            return size;
-        }
-    };
-
-    /**
-     * Estimate the size of writes that will be sent to the replica set primary.
-     */
-    class TargetPrimaryWriteSizeEstimator final : public WriteSizeEstimator {
-    public:
-        int estimateInsertHeaderSize(
-            const write_ops::InsertCommandRequest& insertReq) const override {
-            return write_ops::getInsertHeaderSizeEstimate(insertReq);
-        }
-
-        int estimateUpdateHeaderSize(
-            const write_ops::UpdateCommandRequest& updateReq) const override {
-            return write_ops::getUpdateHeaderSizeEstimate(updateReq);
-        }
-
-        int estimateInsertSizeBytes(const BSONObj& insert) const override {
-            return insert.objsize() + write_ops::kWriteCommandBSONArrayPerElementOverheadBytes;
-        }
-
-        int estimateUpdateSizeBytes(const BatchObject& batchObject,
-                                    UpsertType type) const override {
-            return getUpdateSizeEstimate(std::get<BSONObj>(batchObject),
-                                         std::get<write_ops::UpdateModification>(batchObject),
-                                         std::get<boost::optional<BSONObj>>(batchObject),
-                                         type != UpsertType::kNone /* includeUpsertSupplied */,
-                                         boost::none /* collation */,
-                                         boost::none /* arrayFilters */,
-                                         BSONObj() /* hint*/) +
-                write_ops::kWriteCommandBSONArrayPerElementOverheadBytes;
-        }
-    };
-
-    /**
      * Returns true if the field names of 'keyPattern' are exactly those in 'uniqueKeyPaths', and
      * each of the elements of 'keyPattern' is numeric, i.e. not "text", "$**", or any other special
      * type of index.
@@ -129,7 +64,6 @@ public:
     virtual std::vector<FieldPath> collectDocumentKeyFieldsActingAsRouter(
         OperationContext*, const NamespaceString&) const override;
 
-
     virtual void updateClientOperationTime(OperationContext* opCtx) const final;
 
     boost::optional<ChunkVersion> refreshAndGetCollectionVersion(
diff --git a/src/mongo/db/pipeline/process_interface/mongo_process_interface.h b/src/mongo/db/pipeline/process_interface/mongo_process_interface.h
index 9a980594737..19477adf8c9 100644
--- a/src/mongo/db/pipeline/process_interface/mongo_process_interface.h
+++ b/src/mongo/db/pipeline/process_interface/mongo_process_interface.h
@@ -78,13 +78,13 @@ class MongoProcessInterface {
 public:
     /**
      * Storage for a batch of BSON Objects to be updated in the write namespace. For each element
-     * in the batch we store a tuple of the following elements:
+     * in the batch we store a tuple of the folliwng elements:
      *   1. BSONObj - specifies the query that identifies a document in the to collection to be
      *      updated.
      *   2. write_ops::UpdateModification - either the new document we want to upsert or insert into
      *      the collection (i.e. a 'classic' replacement update), or the pipeline to run to compute
      *      the new document.
-     *   3. boost::optional<BSONObj> - for pipeline-style updates, specifies variables that can be
+     *   3. boost::optional<BSONObj> - for pipeline-style updated, specifies variables that can be
      *      referred to in the pipeline performing the custom update.
      */
     using BatchObject =
@@ -106,30 +106,6 @@ public:
     enum class CurrentOpBacktraceMode { kIncludeBacktrace, kExcludeBacktrace };
 
     /**
-     * Interface which estimates the size of a given write operation.
-     */
-    class WriteSizeEstimator {
-    public:
-        virtual ~WriteSizeEstimator() = default;
-
-        /**
-         * Set of functions which estimate the entire size of a write command except for the array
-         * of write statements themselves.
-         */
-        virtual int estimateInsertHeaderSize(
-            const write_ops::InsertCommandRequest& insertReq) const = 0;
-        virtual int estimateUpdateHeaderSize(
-            const write_ops::UpdateCommandRequest& updateReq) const = 0;
-
-        /**
-         * Set of functions which estimate the size of a single write statement.
-         */
-        virtual int estimateInsertSizeBytes(const BSONObj& insert) const = 0;
-        virtual int estimateUpdateSizeBytes(const BatchObject& batchObject,
-                                            UpsertType type) const = 0;
-    };
-
-    /**
      * Factory function to create MongoProcessInterface of the right type. The implementation will
      * be installed by a lib higher up in the link graph depending on the application type.
      */
@@ -151,12 +127,6 @@ public:
     virtual ~MongoProcessInterface(){};
 
     /**
-     * Returns an instance of a 'WriteSizeEstimator' interface.
-     */
-    virtual std::unique_ptr<WriteSizeEstimator> getWriteSizeEstimator(
-        OperationContext* opCtx, const NamespaceString& ns) const = 0;
-
-    /**
      * Creates a new TransactionHistoryIterator object. Only applicable in processes which support
      * locally traversing the oplog.
      */
@@ -172,20 +142,6 @@ public:
     virtual bool isSharded(OperationContext* opCtx, const NamespaceString& ns) = 0;
 
     /**
-     * TODO SERVER-79508 validate callers of this function remain correct.
-     *
-     * Returns false if the current request only handles parsing and validating queries. In other
-     * words, we are not executing queries. Examples include query analysis for queryable
-     * encryption, executing pipeline-style operations in the Update system, and creating a Query
-     * Shape. This function only returns false when the process interface is of type
-     * 'StubMongoProcessInterface'.
-     *
-     */
-    virtual bool isExpectedToExecuteQueries() {
-        return true;
-    }
-
-    /**
      * Advances the proxied write time associated with the client in ReplClientInfo to
      * be at least as high as the one tracked by the OperationTimeTracker associated with the
      * given operation context.
@@ -193,30 +149,29 @@ public:
     virtual void updateClientOperationTime(OperationContext* opCtx) const = 0;
 
     /**
-     * Executes 'insertCommand' against 'ns' and returns an error Status if the insert fails. If
-     * 'targetEpoch' is set, throws ErrorCodes::StaleEpoch if the targeted collection does not have
-     * the same epoch or the epoch changes during the course of the insert.
+     * Inserts 'objs' into 'ns' and returns an error Status if the insert fails. If 'targetEpoch' is
+     * set, throws ErrorCodes::StaleEpoch if the targeted collection does not have the same epoch or
+     * the epoch changes during the course of the insert.
      */
     virtual Status insert(const boost::intrusive_ptr<ExpressionContext>& expCtx,
                           const NamespaceString& ns,
-                          std::unique_ptr<write_ops::InsertCommandRequest> insertCommand,
+                          std::vector<BSONObj>&& objs,
                           const WriteConcernOptions& wc,
                           boost::optional<OID> targetEpoch) = 0;
 
     /**
-     * Executes the updates described by 'updateCommand'. Returns an error Status if any of the
-     * updates fail, otherwise returns an 'UpdateResult' objects with the details of the update
-     * operation.  If 'targetEpoch' is set, throws ErrorCodes::StaleEpoch if the targeted collection
-     * does not have the same epoch, or if the epoch changes during the update.
+     * Updates the documents matching 'queries' with the objects 'updates'. Returns an error Status
+     * if any of the updates fail, otherwise returns an 'UpdateResult' objects with the details of
+     * the update operation.  If 'targetEpoch' is set, throws ErrorCodes::StaleEpoch if the targeted
+     * collection does not have the same epoch, or if the epoch changes during the update.
      */
-    virtual StatusWith<UpdateResult> update(
-        const boost::intrusive_ptr<ExpressionContext>& expCtx,
-        const NamespaceString& ns,
-        std::unique_ptr<write_ops::UpdateCommandRequest> updateCommand,
-        const WriteConcernOptions& wc,
-        UpsertType upsert,
-        bool multi,
-        boost::optional<OID> targetEpoch) = 0;
+    virtual StatusWith<UpdateResult> update(const boost::intrusive_ptr<ExpressionContext>& expCtx,
+                                            const NamespaceString& ns,
+                                            BatchedObjects&& batch,
+                                            const WriteConcernOptions& wc,
+                                            UpsertType upsert,
+                                            bool multi,
+                                            boost::optional<OID> targetEpoch) = 0;
 
     /**
      * Returns index usage statistics for each index on collection 'ns' along with additional
@@ -243,10 +198,8 @@ public:
     /**
      * Returns the catalog entry for the given namespace, if it exists.
      */
-    virtual boost::optional<BSONObj> getCatalogEntry(
-        OperationContext* opCtx,
-        const NamespaceString& ns,
-        const boost::optional<UUID>& collUUID = boost::none) const = 0;
+    virtual boost::optional<BSONObj> getCatalogEntry(OperationContext* opCtx,
+                                                     const NamespaceString& ns) const = 0;
 
     /**
      * Appends operation latency statistics for collection "nss" to "builder"
diff --git a/src/mongo/db/pipeline/process_interface/mongos_process_interface.cpp b/src/mongo/db/pipeline/process_interface/mongos_process_interface.cpp
index fb0890f5d61..6f17c7a0121 100644
--- a/src/mongo/db/pipeline/process_interface/mongos_process_interface.cpp
+++ b/src/mongo/db/pipeline/process_interface/mongos_process_interface.cpp
@@ -99,12 +99,6 @@ bool supportsUniqueKey(const boost::intrusive_ptr<ExpressionContext>& expCtx,
 
 }  // namespace
 
-std::unique_ptr<MongoProcessInterface::WriteSizeEstimator>
-MongosProcessInterface::getWriteSizeEstimator(OperationContext* opCtx,
-                                              const NamespaceString& ns) const {
-    return std::make_unique<TargetPrimaryWriteSizeEstimator>();
-}
-
 std::unique_ptr<Pipeline, PipelineDeleter> MongosProcessInterface::attachCursorSourceToPipeline(
     Pipeline* ownedPipeline,
     ShardTargetingPolicy shardTargetingPolicy,
@@ -181,15 +175,8 @@ boost::optional<Document> MongosProcessInterface::lookupSingleDocument(
                 // single shard will be targeted here; however, in certain cases where only the _id
                 // is present, we may need to scatter-gather the query to all shards in order to
                 // find the document.
-                auto requests =
-                    getVersionedRequestsForTargetedShards(expCtx->opCtx,
-                                                          nss,
-                                                          cm,
-                                                          findCmd,
-                                                          filterObj,
-                                                          CollationSpec::kSimpleSpec,
-                                                          boost::none /*letParameters*/,
-                                                          boost::none /*runtimeConstants*/);
+                auto requests = getVersionedRequestsForTargetedShards(
+                    expCtx->opCtx, nss, cm, findCmd, filterObj, CollationSpec::kSimpleSpec);
 
                 // Dispatch the requests. The 'establishCursors' method conveniently prepares the
                 // result into a vector of cursor responses for us.
diff --git a/src/mongo/db/pipeline/process_interface/mongos_process_interface.h b/src/mongo/db/pipeline/process_interface/mongos_process_interface.h
index c3740948f93..82eedfa6dec 100644
--- a/src/mongo/db/pipeline/process_interface/mongos_process_interface.h
+++ b/src/mongo/db/pipeline/process_interface/mongos_process_interface.h
@@ -45,9 +45,6 @@ public:
 
     virtual ~MongosProcessInterface() = default;
 
-    std::unique_ptr<WriteSizeEstimator> getWriteSizeEstimator(
-        OperationContext* opCtx, const NamespaceString& ns) const final;
-
     boost::optional<Document> lookupSingleDocument(
         const boost::intrusive_ptr<ExpressionContext>& expCtx,
         const NamespaceString& nss,
@@ -72,7 +69,7 @@ public:
 
     Status insert(const boost::intrusive_ptr<ExpressionContext>& expCtx,
                   const NamespaceString& ns,
-                  std::unique_ptr<write_ops::InsertCommandRequest> insertCommand,
+                  std::vector<BSONObj>&& objs,
                   const WriteConcernOptions& wc,
                   boost::optional<OID>) final {
         MONGO_UNREACHABLE;
@@ -80,7 +77,7 @@ public:
 
     StatusWith<UpdateResult> update(const boost::intrusive_ptr<ExpressionContext>& expCtx,
                                     const NamespaceString& ns,
-                                    std::unique_ptr<write_ops::UpdateCommandRequest> updateCommand,
+                                    BatchedObjects&& batch,
                                     const WriteConcernOptions& wc,
                                     UpsertType upsert,
                                     bool multi,
@@ -106,8 +103,7 @@ public:
     }
 
     boost::optional<BSONObj> getCatalogEntry(OperationContext* opCtx,
-                                             const NamespaceString& ns,
-                                             const boost::optional<UUID>& collUUID) const final {
+                                             const NamespaceString& ns) const final {
         MONGO_UNREACHABLE;
     }
 
diff --git a/src/mongo/db/pipeline/process_interface/non_shardsvr_process_interface.cpp b/src/mongo/db/pipeline/process_interface/non_shardsvr_process_interface.cpp
index 2a1ce64792a..682c0075340 100644
--- a/src/mongo/db/pipeline/process_interface/non_shardsvr_process_interface.cpp
+++ b/src/mongo/db/pipeline/process_interface/non_shardsvr_process_interface.cpp
@@ -36,7 +36,7 @@
 #include "mongo/db/catalog/list_indexes.h"
 #include "mongo/db/catalog/rename_collection.h"
 #include "mongo/db/concurrency/d_concurrency.h"
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/index_builds_coordinator.h"
 #include "mongo/db/pipeline/document_source_cursor.h"
@@ -96,13 +96,13 @@ boost::optional<Document> NonShardServerProcessInterface::lookupSingleDocument(
     return lookedUpDocument;
 }
 
-Status NonShardServerProcessInterface::insert(
-    const boost::intrusive_ptr<ExpressionContext>& expCtx,
-    const NamespaceString& ns,
-    std::unique_ptr<write_ops::InsertCommandRequest> insertCommand,
-    const WriteConcernOptions& wc,
-    boost::optional<OID> targetEpoch) {
-    auto writeResults = write_ops_exec::performInserts(expCtx->opCtx, *insertCommand);
+Status NonShardServerProcessInterface::insert(const boost::intrusive_ptr<ExpressionContext>& expCtx,
+                                              const NamespaceString& ns,
+                                              std::vector<BSONObj>&& objs,
+                                              const WriteConcernOptions& wc,
+                                              boost::optional<OID> targetEpoch) {
+    auto writeResults = write_ops_exec::performInserts(
+        expCtx->opCtx, buildInsertOp(ns, std::move(objs), expCtx->bypassDocumentValidation));
 
     // Need to check each result in the batch since the writes are unordered.
     for (const auto& result : writeResults.results) {
@@ -116,12 +116,13 @@ Status NonShardServerProcessInterface::insert(
 StatusWith<MongoProcessInterface::UpdateResult> NonShardServerProcessInterface::update(
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
     const NamespaceString& ns,
-    std::unique_ptr<write_ops::UpdateCommandRequest> updateCommand,
+    BatchedObjects&& batch,
     const WriteConcernOptions& wc,
     UpsertType upsert,
     bool multi,
     boost::optional<OID> targetEpoch) {
-    auto writeResults = write_ops_exec::performUpdates(expCtx->opCtx, *updateCommand);
+    auto writeResults = write_ops_exec::performUpdates(
+        expCtx->opCtx, buildUpdateOp(expCtx, ns, std::move(batch), upsert, multi));
 
     // Need to check each result in the batch since the writes are unordered.
     UpdateResult updateResult;
@@ -183,10 +184,9 @@ void NonShardServerProcessInterface::renameIfOptionsAndIndexesHaveNotChanged(
     RenameCollectionOptions options;
     options.dropTarget = renameCommandObj["dropTarget"].trueValue();
     options.stayTemp = renameCommandObj["stayTemp"].trueValue();
-    options.originalCollectionOptions = originalCollectionOptions;
-    options.originalIndexes = originalIndexes;
     // skip sharding validation on non sharded servers
-    doLocalRenameIfOptionsAndIndexesHaveNotChanged(opCtx, sourceNs, targetNs, options);
+    doLocalRenameIfOptionsAndIndexesHaveNotChanged(
+        opCtx, sourceNs, targetNs, options, originalIndexes, originalCollectionOptions);
 }
 
 void NonShardServerProcessInterface::createCollection(OperationContext* opCtx,
@@ -205,7 +205,6 @@ BSONObj NonShardServerProcessInterface::preparePipelineAndExplain(
     Pipeline* ownedPipeline, ExplainOptions::Verbosity verbosity) {
     std::vector<Value> pipelineVec;
     auto firstStage = ownedPipeline->peekFront();
-    auto opts = SerializationOptions{verbosity};
     // If the pipeline already has a cursor explain with that one, otherwise attach a new one like
     // we would for a normal execution and explain that.
     if (firstStage && typeid(*firstStage) == typeid(DocumentSourceCursor)) {
@@ -213,7 +212,7 @@ BSONObj NonShardServerProcessInterface::preparePipelineAndExplain(
         // extracted the necessary information and won't need it again.
         std::unique_ptr<Pipeline, PipelineDeleter> managedPipeline(
             ownedPipeline, PipelineDeleter(ownedPipeline->getContext()->opCtx));
-        pipelineVec = managedPipeline->writeExplainOps(opts);
+        pipelineVec = managedPipeline->writeExplainOps(verbosity);
         ownedPipeline = nullptr;
     } else {
         auto pipelineWithCursor = attachCursorSourceToPipelineForLocalRead(ownedPipeline);
@@ -222,7 +221,7 @@ BSONObj NonShardServerProcessInterface::preparePipelineAndExplain(
             while (pipelineWithCursor->getNext()) {
             }
         }
-        pipelineVec = pipelineWithCursor->writeExplainOps(opts);
+        pipelineVec = pipelineWithCursor->writeExplainOps(verbosity);
     }
     BSONArrayBuilder bab;
     for (auto&& stage : pipelineVec) {
diff --git a/src/mongo/db/pipeline/process_interface/non_shardsvr_process_interface.h b/src/mongo/db/pipeline/process_interface/non_shardsvr_process_interface.h
index 9b96e83a1a8..ccbe90205c9 100644
--- a/src/mongo/db/pipeline/process_interface/non_shardsvr_process_interface.h
+++ b/src/mongo/db/pipeline/process_interface/non_shardsvr_process_interface.h
@@ -88,13 +88,13 @@ public:
 
     Status insert(const boost::intrusive_ptr<ExpressionContext>& expCtx,
                   const NamespaceString& ns,
-                  std::unique_ptr<write_ops::InsertCommandRequest> insertCommand,
+                  std::vector<BSONObj>&& objs,
                   const WriteConcernOptions& wc,
                   boost::optional<OID> targetEpoch) override;
 
     StatusWith<UpdateResult> update(const boost::intrusive_ptr<ExpressionContext>& expCtx,
                                     const NamespaceString& ns,
-                                    std::unique_ptr<write_ops::UpdateCommandRequest> updateCommand,
+                                    BatchedObjects&& batch,
                                     const WriteConcernOptions& wc,
                                     UpsertType upsert,
                                     bool multi,
diff --git a/src/mongo/db/pipeline/process_interface/replica_set_node_process_interface.cpp b/src/mongo/db/pipeline/process_interface/replica_set_node_process_interface.cpp
index 694038eff96..01db33c2337 100644
--- a/src/mongo/db/pipeline/process_interface/replica_set_node_process_interface.cpp
+++ b/src/mongo/db/pipeline/process_interface/replica_set_node_process_interface.cpp
@@ -35,6 +35,7 @@
 #include "mongo/db/catalog/drop_collection.h"
 #include "mongo/db/catalog/rename_collection.h"
 #include "mongo/db/concurrency/d_concurrency.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/index_builds_coordinator.h"
 #include "mongo/db/logical_session_id_helpers.h"
@@ -69,27 +70,26 @@ void ReplicaSetNodeProcessInterface::setReplicaSetNodeExecutor(
     replicaSetNodeExecutor(service) = std::move(executor);
 }
 
-Status ReplicaSetNodeProcessInterface::insert(
-    const boost::intrusive_ptr<ExpressionContext>& expCtx,
-    const NamespaceString& ns,
-    std::unique_ptr<write_ops::InsertCommandRequest> insertCommand,
-    const WriteConcernOptions& wc,
-    boost::optional<OID> targetEpoch) {
+Status ReplicaSetNodeProcessInterface::insert(const boost::intrusive_ptr<ExpressionContext>& expCtx,
+                                              const NamespaceString& ns,
+                                              std::vector<BSONObj>&& objs,
+                                              const WriteConcernOptions& wc,
+                                              boost::optional<OID> targetEpoch) {
     auto&& opCtx = expCtx->opCtx;
     if (_canWriteLocally(opCtx, ns)) {
-        return NonShardServerProcessInterface::insert(
-            expCtx, ns, std::move(insertCommand), wc, targetEpoch);
+        return NonShardServerProcessInterface::insert(expCtx, ns, std::move(objs), wc, targetEpoch);
     }
 
-    BatchedCommandRequest batchInsertCommand(std::move(insertCommand));
+    BatchedCommandRequest insertCommand(
+        buildInsertOp(ns, std::move(objs), expCtx->bypassDocumentValidation));
 
-    return _executeCommandOnPrimary(opCtx, ns, batchInsertCommand.toBSON()).getStatus();
+    return _executeCommandOnPrimary(opCtx, ns, std::move(insertCommand.toBSON())).getStatus();
 }
 
 StatusWith<MongoProcessInterface::UpdateResult> ReplicaSetNodeProcessInterface::update(
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
     const NamespaceString& ns,
-    std::unique_ptr<write_ops::UpdateCommandRequest> updateCommand,
+    BatchedObjects&& batch,
     const WriteConcernOptions& wc,
     UpsertType upsert,
     bool multi,
@@ -97,11 +97,11 @@ StatusWith<MongoProcessInterface::UpdateResult> ReplicaSetNodeProcessInterface::
     auto&& opCtx = expCtx->opCtx;
     if (_canWriteLocally(opCtx, ns)) {
         return NonShardServerProcessInterface::update(
-            expCtx, ns, std::move(updateCommand), wc, upsert, multi, targetEpoch);
+            expCtx, ns, std::move(batch), wc, upsert, multi, targetEpoch);
     }
-    BatchedCommandRequest batchUpdateCommand(std::move(updateCommand));
 
-    auto result = _executeCommandOnPrimary(opCtx, ns, batchUpdateCommand.toBSON());
+    BatchedCommandRequest updateCommand(buildUpdateOp(expCtx, ns, std::move(batch), upsert, multi));
+    auto result = _executeCommandOnPrimary(opCtx, ns, std::move(updateCommand.toBSON()));
     if (!result.isOK()) {
         return result.getStatus();
     }
diff --git a/src/mongo/db/pipeline/process_interface/replica_set_node_process_interface.h b/src/mongo/db/pipeline/process_interface/replica_set_node_process_interface.h
index 55b645c59aa..c61f654e844 100644
--- a/src/mongo/db/pipeline/process_interface/replica_set_node_process_interface.h
+++ b/src/mongo/db/pipeline/process_interface/replica_set_node_process_interface.h
@@ -43,15 +43,6 @@ class ReplicaSetNodeProcessInterface final : public NonShardServerProcessInterfa
 public:
     using NonShardServerProcessInterface::NonShardServerProcessInterface;
 
-    std::unique_ptr<WriteSizeEstimator> getWriteSizeEstimator(
-        OperationContext* opCtx, const NamespaceString& ns) const override {
-        if (_canWriteLocally(opCtx, ns)) {
-            return std::make_unique<LocalWriteSizeEstimator>();
-        } else {
-            return std::make_unique<TargetPrimaryWriteSizeEstimator>();
-        }
-    }
-
     static std::shared_ptr<executor::TaskExecutor> getReplicaSetNodeExecutor(
         ServiceContext* service);
 
@@ -68,13 +59,12 @@ public:
 
     Status insert(const boost::intrusive_ptr<ExpressionContext>& expCtx,
                   const NamespaceString& ns,
-                  std::unique_ptr<write_ops::InsertCommandRequest> insertCommand,
+                  std::vector<BSONObj>&& objs,
                   const WriteConcernOptions& wc,
                   boost::optional<OID> targetEpoch) final;
-
     StatusWith<UpdateResult> update(const boost::intrusive_ptr<ExpressionContext>& expCtx,
                                     const NamespaceString& ns,
-                                    std::unique_ptr<write_ops::UpdateCommandRequest> updateCommand,
+                                    BatchedObjects&& batch,
                                     const WriteConcernOptions& wc,
                                     UpsertType upsert,
                                     bool multi,
diff --git a/src/mongo/db/pipeline/process_interface/shardsvr_process_interface.cpp b/src/mongo/db/pipeline/process_interface/shardsvr_process_interface.cpp
index aa6914f88c3..69b5a111e2b 100644
--- a/src/mongo/db/pipeline/process_interface/shardsvr_process_interface.cpp
+++ b/src/mongo/db/pipeline/process_interface/shardsvr_process_interface.cpp
@@ -29,6 +29,8 @@
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kQuery
 
+#include "mongo/platform/basic.h"
+
 #include "mongo/db/pipeline/process_interface/shardsvr_process_interface.h"
 
 #include <fmt/format.h>
@@ -49,7 +51,7 @@
 #include "mongo/s/cluster_commands_helpers.h"
 #include "mongo/s/cluster_write.h"
 #include "mongo/s/query/document_source_merge_cursors.h"
-#include "mongo/s/router_role.h"
+#include "mongo/s/router.h"
 #include "mongo/s/stale_shard_version_helpers.h"
 
 namespace mongo {
@@ -74,11 +76,14 @@ void ShardServerProcessInterface::checkRoutingInfoEpochOrThrow(
     catalogCache->invalidateShardOrEntireCollectionEntryForShardedCollection(
         nss, targetCollectionVersion, shardId);
 
-    const auto cm = uassertStatusOK(catalogCache->getCollectionRoutingInfo(expCtx->opCtx, nss));
-    auto foundVersion = cm.isSharded() ? cm.getVersion() : ChunkVersion::UNSHARDED();
+    const auto routingInfo =
+        uassertStatusOK(catalogCache->getCollectionRoutingInfo(expCtx->opCtx, nss));
+
+    const auto foundVersion =
+        routingInfo.isSharded() ? routingInfo.getVersion() : ChunkVersion::UNSHARDED();
 
-    uassert(StaleEpochInfo(nss, targetCollectionVersion, foundVersion),
-            str::stream() << "Could not act as router for " << nss.ns() << ", received "
+    uassert(StaleEpochInfo(nss),
+            str::stream() << "Could not act as router for " << nss.ns() << ", wanted "
                           << targetCollectionVersion.toString() << ", but found "
                           << foundVersion.toString(),
             foundVersion.isSameCollection(targetCollectionVersion));
@@ -99,22 +104,20 @@ boost::optional<Document> ShardServerProcessInterface::lookupSingleDocument(
     return doLookupSingleDocument(expCtx, nss, collectionUUID, documentKey, std::move(opts));
 }
 
-Status ShardServerProcessInterface::insert(
-    const boost::intrusive_ptr<ExpressionContext>& expCtx,
-    const NamespaceString& ns,
-    std::unique_ptr<write_ops::InsertCommandRequest> insertCommand,
-    const WriteConcernOptions& wc,
-    boost::optional<OID> targetEpoch) {
+Status ShardServerProcessInterface::insert(const boost::intrusive_ptr<ExpressionContext>& expCtx,
+                                           const NamespaceString& ns,
+                                           std::vector<BSONObj>&& objs,
+                                           const WriteConcernOptions& wc,
+                                           boost::optional<OID> targetEpoch) {
     BatchedCommandResponse response;
     BatchWriteExecStats stats;
 
-    BatchedCommandRequest batchInsertCommand(std::move(insertCommand));
+    BatchedCommandRequest insertCommand(
+        buildInsertOp(ns, std::move(objs), expCtx->bypassDocumentValidation));
 
-    const auto originalWC = expCtx->opCtx->getWriteConcern();
-    ScopeGuard resetWCGuard([&] { expCtx->opCtx->setWriteConcern(originalWC); });
-    expCtx->opCtx->setWriteConcern(wc);
+    insertCommand.setWriteConcern(wc.toBSON());
 
-    cluster::write(expCtx->opCtx, batchInsertCommand, &stats, &response, targetEpoch);
+    cluster::write(expCtx->opCtx, insertCommand, &stats, &response, targetEpoch);
 
     return response.toStatus();
 }
@@ -122,7 +125,7 @@ Status ShardServerProcessInterface::insert(
 StatusWith<MongoProcessInterface::UpdateResult> ShardServerProcessInterface::update(
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
     const NamespaceString& ns,
-    std::unique_ptr<write_ops::UpdateCommandRequest> updateCommand,
+    BatchedObjects&& batch,
     const WriteConcernOptions& wc,
     UpsertType upsert,
     bool multi,
@@ -130,13 +133,11 @@ StatusWith<MongoProcessInterface::UpdateResult> ShardServerProcessInterface::upd
     BatchedCommandResponse response;
     BatchWriteExecStats stats;
 
-    BatchedCommandRequest batchUpdateCommand(std::move(updateCommand));
+    BatchedCommandRequest updateCommand(buildUpdateOp(expCtx, ns, std::move(batch), upsert, multi));
 
-    const auto originalWC = expCtx->opCtx->getWriteConcern();
-    ScopeGuard resetWCGuard([&] { expCtx->opCtx->setWriteConcern(originalWC); });
-    expCtx->opCtx->setWriteConcern(wc);
+    updateCommand.setWriteConcern(wc.toBSON());
 
-    cluster::write(expCtx->opCtx, batchUpdateCommand, &stats, &response, targetEpoch);
+    cluster::write(expCtx->opCtx, updateCommand, &stats, &response, targetEpoch);
 
     if (auto status = response.toStatus(); status != Status::OK()) {
         return status;
@@ -173,35 +174,29 @@ void ShardServerProcessInterface::renameIfOptionsAndIndexesHaveNotChanged(
     const NamespaceString& destinationNs,
     const BSONObj& originalCollectionOptions,
     const std::list<BSONObj>& originalIndexes) {
-    sharding::router::DBPrimaryRouter router(opCtx->getServiceContext(), destinationNs.db());
-    router.route(opCtx,
-                 "ShardServerProcessInterface::renameIfOptionsAndIndexesHaveNotChanged",
-                 [&](OperationContext* opCtx, const CachedDatabaseInfo& cdb) {
-                     auto newCmdObj = CommonMongodProcessInterface::_convertRenameToInternalRename(
-                         opCtx, renameCommandObj, originalCollectionOptions, originalIndexes);
-                     BSONObjBuilder newCmdWithWriteConcernBuilder(std::move(newCmdObj));
-                     newCmdWithWriteConcernBuilder.append(WriteConcernOptions::kWriteConcernField,
-                                                          opCtx->getWriteConcern().toBSON());
-                     newCmdObj = newCmdWithWriteConcernBuilder.done();
-                     auto response = executeCommandAgainstDatabasePrimary(
-                         opCtx,
-                         // internalRenameIfOptionsAndIndexesMatch is adminOnly.
-                         NamespaceString::kAdminDb,
-                         cdb,
-                         newCmdObj,
-                         ReadPreferenceSetting(ReadPreference::PrimaryOnly),
-                         Shard::RetryPolicy::kNoRetry);
-                     uassertStatusOKWithContext(response.swResponse,
-                                                str::stream() << "failed while running command "
-                                                              << newCmdObj);
-                     auto result = response.swResponse.getValue().data;
-                     uassertStatusOKWithContext(getStatusFromCommandResult(result),
-                                                str::stream() << "failed while running command "
-                                                              << newCmdObj);
-                     uassertStatusOKWithContext(getWriteConcernStatusFromCommandResult(result),
-                                                str::stream() << "failed while running command "
-                                                              << newCmdObj);
-                 });
+    auto cachedDbInfo =
+        uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, destinationNs.db()));
+    auto newCmdObj = CommonMongodProcessInterface::_convertRenameToInternalRename(
+        opCtx, renameCommandObj, originalCollectionOptions, originalIndexes);
+    BSONObjBuilder newCmdWithWriteConcernBuilder(std::move(newCmdObj));
+    newCmdWithWriteConcernBuilder.append(WriteConcernOptions::kWriteConcernField,
+                                         opCtx->getWriteConcern().toBSON());
+    newCmdObj = newCmdWithWriteConcernBuilder.done();
+    auto response =
+        executeCommandAgainstDatabasePrimary(opCtx,
+                                             // internalRenameIfOptionsAndIndexesMatch is adminOnly.
+                                             NamespaceString::kAdminDb,
+                                             std::move(cachedDbInfo),
+                                             newCmdObj,
+                                             ReadPreferenceSetting(ReadPreference::PrimaryOnly),
+                                             Shard::RetryPolicy::kNoRetry);
+    uassertStatusOKWithContext(response.swResponse,
+                               str::stream() << "failed while running command " << newCmdObj);
+    auto result = response.swResponse.getValue().data;
+    uassertStatusOKWithContext(getStatusFromCommandResult(result),
+                               str::stream() << "failed while running command " << newCmdObj);
+    uassertStatusOKWithContext(getWriteConcernStatusFromCommandResult(result),
+                               str::stream() << "failed while running command " << newCmdObj);
 }
 
 BSONObj ShardServerProcessInterface::getCollectionOptions(OperationContext* opCtx,
@@ -210,62 +205,58 @@ BSONObj ShardServerProcessInterface::getCollectionOptions(OperationContext* opCt
         return getCollectionOptionsLocally(opCtx, nss);
     }
 
-    sharding::router::DBPrimaryRouter router(opCtx->getServiceContext(), nss.db());
-    return router.route(
-        opCtx,
-        "ShardServerProcessInterface::getCollectionOptions",
-        [&](OperationContext* opCtx, const CachedDatabaseInfo& cdb) {
-            const BSONObj filterObj = BSON("name" << nss.coll());
-            const BSONObj cmdObj = BSON("listCollections" << 1 << "filter" << filterObj);
-
-            const auto shard = uassertStatusOK(
-                Grid::get(opCtx)->shardRegistry()->getShard(opCtx, cdb->getPrimary()));
-            Shard::QueryResponse resultCollections;
-
-            try {
-                resultCollections = uassertStatusOK(shard->runExhaustiveCursorCommand(
-                    opCtx,
-                    ReadPreferenceSetting(ReadPreference::PrimaryOnly),
-                    nss.db().toString(),
-                    appendDbVersionIfPresent(cmdObj, cdb),
-                    Milliseconds(-1)));
-            } catch (const ExceptionFor<ErrorCodes::NamespaceNotFound>&) {
-                return BSONObj{};
-            }
+    auto cachedDbInfo =
+        uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, nss.db()));
+    auto shard = uassertStatusOK(
+        Grid::get(opCtx)->shardRegistry()->getShard(opCtx, cachedDbInfo->getPrimary()));
+
+    const BSONObj filterObj = BSON("name" << nss.coll());
+    const BSONObj cmdObj = BSON("listCollections" << 1 << "filter" << filterObj);
+
+    Shard::QueryResponse resultCollections;
+    try {
+        resultCollections = uassertStatusOK(
+            shard->runExhaustiveCursorCommand(opCtx,
+                                              ReadPreferenceSetting(ReadPreference::PrimaryOnly),
+                                              nss.db().toString(),
+                                              appendDbVersionIfPresent(cmdObj, cachedDbInfo),
+                                              Milliseconds(-1)));
+    } catch (const ExceptionFor<ErrorCodes::NamespaceNotFound>&) {
+        return BSONObj{};
+    }
 
-            if (resultCollections.docs.empty()) {
-                return BSONObj{};
-            }
+    if (resultCollections.docs.empty()) {
+        return BSONObj{};
+    }
 
-            for (const BSONObj& bsonObj : resultCollections.docs) {
-                // Return first element which matches on name and has options.
-                const BSONElement nameElement = bsonObj["name"];
-                if (!nameElement || nameElement.valueStringDataSafe() != nss.coll()) {
-                    continue;
-                }
-
-                const BSONElement optionsElement = bsonObj["options"];
-                if (optionsElement) {
-                    auto optionObj = optionsElement.Obj();
-
-                    // If the BSON object has field 'info' and the BSON element 'info' has field
-                    // 'uuid', then extract the uuid and add to the BSON object to be return. This
-                    // will ensure that the BSON object is complaint with the BSON object returned
-                    // for non-sharded namespace.
-                    if (auto infoElement = bsonObj["info"]; infoElement && infoElement["uuid"]) {
-                        return optionObj.addField(infoElement["uuid"]);
-                    }
-
-                    return optionObj.getOwned();
-                }
-
-                tassert(5983900,
-                        str::stream() << "Expected at most one collection with the name " << nss
-                                      << ": " << resultCollections.docs.size(),
-                        resultCollections.docs.size() <= 1);
+    for (const BSONObj& bsonObj : resultCollections.docs) {
+        // Return first element which matches on name and has options.
+        const BSONElement nameElement = bsonObj["name"];
+        if (!nameElement || nameElement.valueStringDataSafe() != nss.coll()) {
+            continue;
+        }
+
+        const BSONElement optionsElement = bsonObj["options"];
+        if (optionsElement) {
+            auto optionObj = optionsElement.Obj();
+
+            // If the BSON object has field 'info' and the BSON element 'info' has field 'uuid',
+            // then extract the uuid and add to the BSON object to be return. This will ensure that
+            // the BSON object is complaint with the BSON object returned for non-sharded namespace.
+            if (auto infoElement = bsonObj["info"]; infoElement && infoElement["uuid"]) {
+                return optionObj.addField(infoElement["uuid"]);
             }
-            return BSONObj{};
-        });
+
+            return optionObj.getOwned();
+        }
+
+        tassert(5983900,
+                str::stream() << "Expected at most one collection with the name " << nss << ": "
+                              << resultCollections.docs.size(),
+                resultCollections.docs.size() <= 1);
+    }
+
+    return BSONObj{};
 }
 
 std::list<BSONObj> ShardServerProcessInterface::getIndexSpecs(OperationContext* opCtx,
@@ -273,58 +264,49 @@ std::list<BSONObj> ShardServerProcessInterface::getIndexSpecs(OperationContext*
                                                               bool includeBuildUUIDs) {
     // Note that 'ns' must be an unsharded collection. The indexes for a sharded collection must be
     // read from a shard with a chunk instead of the primary shard.
-    sharding::router::DBPrimaryRouter router(opCtx->getServiceContext(), ns.db());
-    return router.route(opCtx,
-                        "ShardServerProcessInterface::getIndexSpecs",
-                        [&](OperationContext* opCtx, const CachedDatabaseInfo& cdb) {
-                            auto shard =
-                                uassertStatusOK(Grid::get(opCtx)->shardRegistry()->getShard(
-                                    opCtx, cdb->getPrimary()));
-                            auto cmdObj = BSON("listIndexes" << ns.coll());
-                            try {
-                                auto indexes = uassertStatusOK(shard->runExhaustiveCursorCommand(
-                                    opCtx,
-                                    ReadPreferenceSetting(ReadPreference::PrimaryOnly),
-                                    ns.db().toString(),
-                                    appendDbVersionIfPresent(cmdObj, cdb),
-                                    Milliseconds(-1)));
-                                return std::list<BSONObj>(indexes.docs.begin(), indexes.docs.end());
-                            } catch (ExceptionFor<ErrorCodes::NamespaceNotFound>&) {
-                                return std::list<BSONObj>();
-                            }
-                        });
+    auto cachedDbInfo =
+        uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, ns.db()));
+    auto shard = uassertStatusOK(
+        Grid::get(opCtx)->shardRegistry()->getShard(opCtx, cachedDbInfo->getPrimary()));
+    auto cmdObj = BSON("listIndexes" << ns.coll());
+    Shard::QueryResponse indexes;
+    try {
+        indexes = uassertStatusOK(
+            shard->runExhaustiveCursorCommand(opCtx,
+                                              ReadPreferenceSetting(ReadPreference::PrimaryOnly),
+                                              ns.db().toString(),
+                                              appendDbVersionIfPresent(cmdObj, cachedDbInfo),
+                                              Milliseconds(-1)));
+    } catch (ExceptionFor<ErrorCodes::NamespaceNotFound>&) {
+        return std::list<BSONObj>();
+    }
+    return std::list<BSONObj>(indexes.docs.begin(), indexes.docs.end());
 }
 
 void ShardServerProcessInterface::createCollection(OperationContext* opCtx,
                                                    const std::string& dbName,
                                                    const BSONObj& cmdObj) {
-    sharding::router::DBPrimaryRouter router(opCtx->getServiceContext(), dbName);
-    router.route(opCtx,
-                 "ShardServerProcessInterface::createCollection",
-                 [&](OperationContext* opCtx, const CachedDatabaseInfo& cdb) {
-                     BSONObjBuilder finalCmdBuilder(cmdObj);
-                     finalCmdBuilder.append(WriteConcernOptions::kWriteConcernField,
-                                            opCtx->getWriteConcern().toBSON());
-                     BSONObj finalCmdObj = finalCmdBuilder.obj();
-                     auto response = executeCommandAgainstDatabasePrimary(
-                         opCtx,
-                         dbName,
-                         cdb,
-                         finalCmdObj,
-                         ReadPreferenceSetting(ReadPreference::PrimaryOnly),
-                         Shard::RetryPolicy::kIdempotent);
-                     uassertStatusOKWithContext(response.swResponse,
-                                                str::stream() << "failed while running command "
-                                                              << finalCmdObj);
-                     auto result = response.swResponse.getValue().data;
-                     uassertStatusOKWithContext(getStatusFromCommandResult(result),
-                                                str::stream() << "failed while running command "
-                                                              << finalCmdObj);
-                     uassertStatusOKWithContext(getWriteConcernStatusFromCommandResult(result),
-                                                str::stream()
-                                                    << "write concern failed while running command "
-                                                    << finalCmdObj);
-                 });
+    auto cachedDbInfo =
+        uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName));
+    BSONObjBuilder finalCmdBuilder(cmdObj);
+    finalCmdBuilder.append(WriteConcernOptions::kWriteConcernField,
+                           opCtx->getWriteConcern().toBSON());
+    BSONObj finalCmdObj = finalCmdBuilder.obj();
+    auto response =
+        executeCommandAgainstDatabasePrimary(opCtx,
+                                             dbName,
+                                             std::move(cachedDbInfo),
+                                             finalCmdObj,
+                                             ReadPreferenceSetting(ReadPreference::PrimaryOnly),
+                                             Shard::RetryPolicy::kIdempotent);
+    uassertStatusOKWithContext(response.swResponse,
+                               str::stream() << "failed while running command " << finalCmdObj);
+    auto result = response.swResponse.getValue().data;
+    uassertStatusOKWithContext(getStatusFromCommandResult(result),
+                               str::stream() << "failed while running command " << finalCmdObj);
+    uassertStatusOKWithContext(getWriteConcernStatusFromCommandResult(result),
+                               str::stream()
+                                   << "write concern failed while running command " << finalCmdObj);
 }
 
 void ShardServerProcessInterface::createIndexesOnEmptyCollection(
@@ -370,32 +352,28 @@ void ShardServerProcessInterface::dropCollection(OperationContext* opCtx,
                                                  const NamespaceString& ns) {
     // Build and execute the dropCollection command against the primary shard of the given
     // database.
-    sharding::router::DBPrimaryRouter router(opCtx->getServiceContext(), ns.db());
-    router.route(
-        opCtx,
-        "ShardServerProcessInterface::dropCollection",
-        [&](OperationContext* opCtx, const CachedDatabaseInfo& cdb) {
-            BSONObjBuilder newCmdBuilder;
-            newCmdBuilder.append("drop", ns.coll());
-            newCmdBuilder.append(WriteConcernOptions::kWriteConcernField,
-                                 opCtx->getWriteConcern().toBSON());
-            auto cmdObj = newCmdBuilder.done();
-            auto response = executeCommandAgainstDatabasePrimary(
-                opCtx,
-                ns.db(),
-                cdb,
-                cmdObj,
-                ReadPreferenceSetting(ReadPreference::PrimaryOnly),
-                Shard::RetryPolicy::kIdempotent);
-            uassertStatusOKWithContext(response.swResponse,
-                                       str::stream() << "failed while running command " << cmdObj);
-            auto result = response.swResponse.getValue().data;
-            uassertStatusOKWithContext(getStatusFromCommandResult(result),
-                                       str::stream() << "failed while running command " << cmdObj);
-            uassertStatusOKWithContext(
-                getWriteConcernStatusFromCommandResult(result),
-                str::stream() << "write concern failed while running command " << cmdObj);
-        });
+    auto cachedDbInfo =
+        uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, ns.db()));
+    BSONObjBuilder newCmdBuilder;
+    newCmdBuilder.append("drop", ns.coll());
+    newCmdBuilder.append(WriteConcernOptions::kWriteConcernField,
+                         opCtx->getWriteConcern().toBSON());
+    auto cmdObj = newCmdBuilder.done();
+    auto response =
+        executeCommandAgainstDatabasePrimary(opCtx,
+                                             ns.db(),
+                                             std::move(cachedDbInfo),
+                                             cmdObj,
+                                             ReadPreferenceSetting(ReadPreference::PrimaryOnly),
+                                             Shard::RetryPolicy::kIdempotent);
+    uassertStatusOKWithContext(response.swResponse,
+                               str::stream() << "failed while running command " << cmdObj);
+    auto result = response.swResponse.getValue().data;
+    uassertStatusOKWithContext(getStatusFromCommandResult(result),
+                               str::stream() << "failed while running command " << cmdObj);
+    uassertStatusOKWithContext(getWriteConcernStatusFromCommandResult(result),
+                               str::stream()
+                                   << "write concern failed while running command " << cmdObj);
 }
 
 std::unique_ptr<Pipeline, PipelineDeleter>
diff --git a/src/mongo/db/pipeline/process_interface/shardsvr_process_interface.h b/src/mongo/db/pipeline/process_interface/shardsvr_process_interface.h
index a08aa23777f..f6026f6ef3a 100644
--- a/src/mongo/db/pipeline/process_interface/shardsvr_process_interface.h
+++ b/src/mongo/db/pipeline/process_interface/shardsvr_process_interface.h
@@ -53,11 +53,6 @@ public:
                                       const NamespaceString& nss,
                                       ChunkVersion targetCollectionVersion) const final;
 
-    std::unique_ptr<WriteSizeEstimator> getWriteSizeEstimator(
-        OperationContext* opCtx, const NamespaceString& ns) const final {
-        return std::make_unique<TargetPrimaryWriteSizeEstimator>();
-    }
-
     std::vector<FieldPath> collectDocumentKeyFieldsActingAsRouter(
         OperationContext*, const NamespaceString&) const final {
         // We don't expect anyone to use this method on the shard itself (yet). This is currently
@@ -76,15 +71,23 @@ public:
         const Document& documentKey,
         boost::optional<BSONObj> readConcern) final;
 
+    /**
+     * Inserts the documents 'objs' into the namespace 'ns' using the ClusterWriter for locking,
+     * routing, stale config handling, etc.
+     */
     Status insert(const boost::intrusive_ptr<ExpressionContext>& expCtx,
                   const NamespaceString& ns,
-                  std::unique_ptr<write_ops::InsertCommandRequest> insertCommand,
+                  std::vector<BSONObj>&& objs,
                   const WriteConcernOptions& wc,
                   boost::optional<OID> targetEpoch) final;
 
+    /**
+     * Replaces the documents matching 'queries' with 'updates' using the ClusterWriter for locking,
+     * routing, stale config handling, etc.
+     */
     StatusWith<UpdateResult> update(const boost::intrusive_ptr<ExpressionContext>& expCtx,
                                     const NamespaceString& ns,
-                                    std::unique_ptr<write_ops::UpdateCommandRequest> updateCommand,
+                                    BatchedObjects&& batch,
                                     const WriteConcernOptions& wc,
                                     UpsertType upsert,
                                     bool multi,
diff --git a/src/mongo/db/pipeline/process_interface/shardsvr_process_interface_test.cpp b/src/mongo/db/pipeline/process_interface/shardsvr_process_interface_test.cpp
index d60af845ccc..5c3f7eebf97 100644
--- a/src/mongo/db/pipeline/process_interface/shardsvr_process_interface_test.cpp
+++ b/src/mongo/db/pipeline/process_interface/shardsvr_process_interface_test.cpp
@@ -28,11 +28,9 @@
  */
 
 #include "mongo/db/concurrency/lock_state.h"
-#include "mongo/db/cursor_id.h"
 #include "mongo/db/pipeline/document_source_out.h"
 #include "mongo/db/pipeline/document_source_queue.h"
 #include "mongo/db/pipeline/process_interface/shardsvr_process_interface.h"
-#include "mongo/db/query/cursor_response.h"
 #include "mongo/s/query/sharded_agg_test_fixture.h"
 #include "mongo/unittest/unittest.h"
 
diff --git a/src/mongo/db/pipeline/process_interface/standalone_process_interface.h b/src/mongo/db/pipeline/process_interface/standalone_process_interface.h
index dc562b9089e..aceff8e6928 100644
--- a/src/mongo/db/pipeline/process_interface/standalone_process_interface.h
+++ b/src/mongo/db/pipeline/process_interface/standalone_process_interface.h
@@ -41,11 +41,6 @@ public:
     StandaloneProcessInterface(std::shared_ptr<executor::TaskExecutor> exec)
         : NonShardServerProcessInterface(std::move(exec)) {}
 
-    std::unique_ptr<MongoProcessInterface::WriteSizeEstimator> getWriteSizeEstimator(
-        OperationContext* opCtx, const NamespaceString& ns) const final {
-        return std::make_unique<LocalWriteSizeEstimator>();
-    }
-
     virtual ~StandaloneProcessInterface() = default;
 };
 
diff --git a/src/mongo/db/pipeline/process_interface/stub_mongo_process_interface.h b/src/mongo/db/pipeline/process_interface/stub_mongo_process_interface.h
index d69d5af7809..3fe1430ac72 100644
--- a/src/mongo/db/pipeline/process_interface/stub_mongo_process_interface.h
+++ b/src/mongo/db/pipeline/process_interface/stub_mongo_process_interface.h
@@ -52,37 +52,6 @@ public:
         MONGO_UNREACHABLE;
     }
 
-    class StubWriteSizeEstimator final : public WriteSizeEstimator {
-    public:
-        int estimateInsertHeaderSize(
-            const write_ops::InsertCommandRequest& insertReq) const override {
-            return 0;
-        }
-
-        int estimateUpdateHeaderSize(
-            const write_ops::UpdateCommandRequest& insertReq) const override {
-            return 0;
-        }
-
-        int estimateInsertSizeBytes(const BSONObj& insert) const override {
-            MONGO_UNREACHABLE;
-        }
-
-        int estimateUpdateSizeBytes(const BatchObject& batchObject,
-                                    UpsertType type) const override {
-            MONGO_UNREACHABLE;
-        }
-    };
-
-    std::unique_ptr<WriteSizeEstimator> getWriteSizeEstimator(
-        OperationContext* opCtx, const NamespaceString& ns) const override {
-        return std::make_unique<StubWriteSizeEstimator>();
-    }
-
-    bool isExpectedToExecuteQueries() override {
-        return false;
-    }
-
     bool isSharded(OperationContext* opCtx, const NamespaceString& ns) override {
         return false;
     }
@@ -91,7 +60,7 @@ public:
 
     Status insert(const boost::intrusive_ptr<ExpressionContext>& expCtx,
                   const NamespaceString& ns,
-                  std::unique_ptr<write_ops::InsertCommandRequest> insertCommand,
+                  std::vector<BSONObj>&& objs,
                   const WriteConcernOptions& wc,
                   boost::optional<OID>) override {
         MONGO_UNREACHABLE;
@@ -99,7 +68,7 @@ public:
 
     StatusWith<UpdateResult> update(const boost::intrusive_ptr<ExpressionContext>& expCtx,
                                     const NamespaceString& ns,
-                                    std::unique_ptr<write_ops::UpdateCommandRequest> updateCommand,
+                                    BatchedObjects&& batch,
                                     const WriteConcernOptions& wc,
                                     UpsertType upsert,
                                     bool multi,
@@ -125,8 +94,7 @@ public:
     }
 
     boost::optional<BSONObj> getCatalogEntry(OperationContext* opCtx,
-                                             const NamespaceString& ns,
-                                             const boost::optional<UUID>& collUUID) const override {
+                                             const NamespaceString& ns) const override {
         MONGO_UNREACHABLE;
     }
 
@@ -259,11 +227,11 @@ public:
         return BackupCursorState{UUID::gen(), boost::none, nullptr, {}};
     }
 
-    void closeBackupCursor(OperationContext* opCtx, const UUID& backupId) override {}
+    void closeBackupCursor(OperationContext* opCtx, const UUID& backupId) final {}
 
     BackupCursorExtendState extendBackupCursor(OperationContext* opCtx,
                                                const UUID& backupId,
-                                               const Timestamp& extendTo) override {
+                                               const Timestamp& extendTo) final {
         return {{}};
     }