SERVER-38472 Do not check if collection sharded from config server

(cherry picked from commit c3e78c91c3a86fd6aba44a0b3c97062f55512f56)

4 files changed, 95 insertions, 61 deletions

diff --git a/src/mongo/db/s/config/configsvr_shard_collection_command.cpp b/src/mongo/db/s/config/configsvr_shard_collection_command.cpp
index 5df7dbdee5b..81b01a612d3 100644
--- a/src/mongo/db/s/config/configsvr_shard_collection_command.cpp
+++ b/src/mongo/db/s/config/configsvr_shard_collection_command.cpp
@@ -243,57 +243,6 @@ void validateAndDeduceFullRequestOptions(OperationContext* opCtx,
 }
 
 /**
- * Throws an exception if the collection is already sharded with different options.
- *
- * If the collection is already sharded with the same options, returns the existing collection's
- * full spec, else returns boost::none.
- */
-boost::optional<CollectionType> checkIfAlreadyShardedWithSameOptions(
-    OperationContext* opCtx,
-    const NamespaceString& nss,
-    const ConfigsvrShardCollectionRequest& request) {
-    auto existingColls =
-        uassertStatusOK(Grid::get(opCtx)->shardRegistry()->getConfigShard()->exhaustiveFindOnConfig(
-                            opCtx,
-                            ReadPreferenceSetting{ReadPreference::PrimaryOnly},
-                            repl::ReadConcernLevel::kLocalReadConcern,
-                            CollectionType::ConfigNS,
-                            BSON("_id" << nss.ns() << "dropped" << false),
-                            BSONObj(),
-                            1))
-            .docs;
-
-    if (!existingColls.empty()) {
-        auto existingOptions = uassertStatusOK(CollectionType::fromBSON(existingColls.front()));
-
-        CollectionType requestedOptions;
-        requestedOptions.setNs(nss);
-        requestedOptions.setKeyPattern(KeyPattern(request.getKey()));
-        requestedOptions.setDefaultCollation(*request.getCollation());
-        requestedOptions.setUnique(request.getUnique());
-
-        // If the collection is already sharded, fail if the deduced options in this request do not
-        // match the options the collection was originally sharded with.
-        uassert(ErrorCodes::AlreadyInitialized,
-                str::stream() << "sharding already enabled for collection " << nss.ns()
-                              << " with options "
-                              << existingOptions.toString(),
-                requestedOptions.hasSameOptions(existingOptions));
-
-        // We did a local read of the collection entry above and found that this shardCollection
-        // request was already satisfied. However, the data may not be majority committed (a
-        // previous shardCollection attempt may have failed with a writeConcern error).
-        // Since the current Client doesn't know the opTime of the last write to the collection
-        // entry, make it wait for the last opTime in the system when we wait for writeConcern.
-        repl::ReplClientInfo::forClient(opCtx->getClient()).setLastOpToSystemLastOpTime(opCtx);
-        return existingOptions;
-    }
-
-    // Not currently sharded.
-    return boost::none;
-}
-
-/**
  * Compares the proposed shard key with the collection's existing indexes on the primary shard to
  * ensure they are a legal combination.
  *
@@ -745,16 +694,6 @@ public:
         // validated.
         invariant(request.getCollation());
 
-        // Step 2.
-        if (auto existingColl = checkIfAlreadyShardedWithSameOptions(opCtx, nss, request)) {
-            result << "collectionsharded" << nss.ns();
-            if (existingColl->getUUID()) {
-                result << "collectionUUID" << *existingColl->getUUID();
-            }
-            return true;
-        }
-
-        bool isEmpty = (conn->count(nss.ns()) == 0);
         boost::optional<UUID> uuid;
 
         // The primary shard will read the config.tags collection so we need to lock the zone
@@ -800,6 +739,23 @@ public:
 
             return true;
         } else {
+            // Step 2.
+            if (auto existingColl =
+                    InitialSplitPolicy::checkIfCollectionAlreadyShardedWithSameOptions(
+                        opCtx,
+                        nss,
+                        shardsvrShardCollectionRequest,
+                        repl::ReadConcernLevel::kLocalReadConcern)) {
+                result << "collectionsharded" << nss.ns();
+                if (existingColl->getUUID()) {
+                    result << "collectionUUID" << *existingColl->getUUID();
+                }
+                repl::ReplClientInfo::forClient(opCtx->getClient())
+                    .setLastOpToSystemLastOpTime(opCtx);
+                return true;
+            }
+
+            bool isEmpty = (conn->count(nss.ns()) == 0);
 
             // Step 3.
             validateShardKeyAgainstExistingIndexes(
diff --git a/src/mongo/db/s/config/initial_split_policy.cpp b/src/mongo/db/s/config/initial_split_policy.cpp
index 5d6a4b2b5bc..d18663a6872 100644
--- a/src/mongo/db/s/config/initial_split_policy.cpp
+++ b/src/mongo/db/s/config/initial_split_policy.cpp
@@ -375,4 +375,37 @@ void InitialSplitPolicy::writeFirstChunksToConfig(
     }
 }
 
+boost::optional<CollectionType> InitialSplitPolicy::checkIfCollectionAlreadyShardedWithSameOptions(
+    OperationContext* opCtx,
+    const NamespaceString& nss,
+    const ShardsvrShardCollection& request,
+    repl::ReadConcernLevel readConcernLevel) {
+    auto const catalogClient = Grid::get(opCtx)->catalogClient();
+
+    auto collStatus = catalogClient->getCollection(opCtx, nss, readConcernLevel);
+    if (collStatus == ErrorCodes::NamespaceNotFound) {
+        // Not currently sharded.
+        return boost::none;
+    }
+
+    uassertStatusOK(collStatus);
+    auto existingOptions = collStatus.getValue().value;
+
+    CollectionType requestedOptions;
+    requestedOptions.setNs(nss);
+    requestedOptions.setKeyPattern(KeyPattern(request.getKey()));
+    requestedOptions.setDefaultCollation(*request.getCollation());
+    requestedOptions.setUnique(request.getUnique());
+
+    // If the collection is already sharded, fail if the deduced options in this request do not
+    // match the options the collection was originally sharded with.
+    uassert(ErrorCodes::AlreadyInitialized,
+            str::stream() << "sharding already enabled for collection " << nss.ns()
+                          << " with options "
+                          << existingOptions.toString(),
+            requestedOptions.hasSameOptions(existingOptions));
+
+    return existingOptions;
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/s/config/initial_split_policy.h b/src/mongo/db/s/config/initial_split_policy.h
index 16cb2e82b24..67492c24a98 100644
--- a/src/mongo/db/s/config/initial_split_policy.h
+++ b/src/mongo/db/s/config/initial_split_policy.h
@@ -34,8 +34,11 @@
 
 #include "mongo/bson/bsonobj.h"
 #include "mongo/db/namespace_string.h"
+#include "mongo/db/repl/read_concern_level.h"
 #include "mongo/s/catalog/type_chunk.h"
+#include "mongo/s/catalog/type_collection.h"
 #include "mongo/s/catalog/type_tags.h"
+#include "mongo/s/request_types/shard_collection_gen.h"
 #include "mongo/s/shard_id.h"
 #include "mongo/s/shard_key_pattern.h"
 
@@ -127,5 +130,17 @@ public:
      */
     static void writeFirstChunksToConfig(
         OperationContext* opCtx, const InitialSplitPolicy::ShardCollectionConfig& initialChunks);
+
+    /**
+     * Throws an exception if the collection is already sharded with different options.
+     *
+     * If the collection is already sharded with the same options, returns the existing collection's
+     * full spec, else returns boost::none.
+     */
+    static boost::optional<CollectionType> checkIfCollectionAlreadyShardedWithSameOptions(
+        OperationContext* opCtx,
+        const NamespaceString& nss,
+        const ShardsvrShardCollection& request,
+        repl::ReadConcernLevel readConcernLevel);
 };
 }  // namespace mongo
diff --git a/src/mongo/db/s/shardsvr_shard_collection.cpp b/src/mongo/db/s/shardsvr_shard_collection.cpp
index 829ffe23499..0e9344eb1db 100644
--- a/src/mongo/db/s/shardsvr_shard_collection.cpp
+++ b/src/mongo/db/s/shardsvr_shard_collection.cpp
@@ -135,6 +135,22 @@ BSONObj makeCreateIndexesCmd(const NamespaceString& nss,
     return appendAllowImplicitCreate(createIndexes.obj(), true);
 }
 
+bool checkIfCollectionAlreadyShardedWithSameOptions(OperationContext* opCtx,
+                                                    const NamespaceString& nss,
+                                                    const ShardsvrShardCollection& request,
+                                                    BSONObjBuilder& result) {
+    if (auto existingColl = InitialSplitPolicy::checkIfCollectionAlreadyShardedWithSameOptions(
+            opCtx, nss, request, repl::ReadConcernLevel::kMajorityReadConcern)) {
+        result << "collectionsharded" << nss.ns();
+        if (existingColl->getUUID()) {
+            result << "collectionUUID" << *existingColl->getUUID();
+        }
+
+        return true;
+    }
+    return false;
+}
+
 /**
  * Compares the proposed shard key with the collection's existing indexes on the primary shard to
  * ensure they are a legal combination.
@@ -607,9 +623,23 @@ public:
             result << "collectionsharded" << nss.ns();
         } else {
             try {
+                if (checkIfCollectionAlreadyShardedWithSameOptions(opCtx, nss, request, result)) {
+                    status = Status::OK();
+                    scopedShardCollection.signalComplete(status);
+
+                    return true;
+                }
+
                 // Take the collection critical section so that no writes can happen.
                 CollectionCriticalSection critSec(opCtx, nss);
 
+                if (checkIfCollectionAlreadyShardedWithSameOptions(opCtx, nss, request, result)) {
+                    status = Status::OK();
+                    scopedShardCollection.signalComplete(status);
+
+                    return true;
+                }
+
                 auto proposedKey(request.getKey().getOwned());
                 ShardKeyPattern shardKeyPattern(proposedKey);