Update upstream source from tag 'upstream/6.0.0'master

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

1 files changed, 155 insertions, 177 deletions

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>