SERVER-6491 Prevent dropping shard key index when alternative index doesn't exist

28 files changed, 793 insertions, 263 deletions

diff --git a/etc/backports_required_for_multiversion_tests.yml b/etc/backports_required_for_multiversion_tests.yml
index ef036616336..4ad2f8d3019 100644
--- a/etc/backports_required_for_multiversion_tests.yml
+++ b/etc/backports_required_for_multiversion_tests.yml
@@ -158,6 +158,8 @@ last-continuous:
       test_file: jstests/replsets/apply_ops_dropDatabase.js
     - ticket: SERVER-63732
       test_file: jstests/sharding/shard_collection_basic.js
+    - ticket: SERVER-6491
+      test_file: jstests/sharding/shard_key_index_must_exist.js
 
   # Tests that should only be excluded from particular suites should be listed under that suite.
   suites:
@@ -455,6 +457,8 @@ last-lts:
       test_file: jstests/replsets/apply_ops_dropDatabase.js
     - ticket: SERVER-63732
       test_file: jstests/sharding/shard_collection_basic.js
+    - ticket: SERVER-6491
+      test_file: jstests/sharding/shard_key_index_must_exist.js
 
   # Tests that should only be excluded from particular suites should be listed under that suite.
   suites:
diff --git a/jstests/core/timeseries/timeseries_index_spec.js b/jstests/core/timeseries/timeseries_index_spec.js
index 109be8f856a..f13c64dd80b 100644
--- a/jstests/core/timeseries/timeseries_index_spec.js
+++ b/jstests/core/timeseries/timeseries_index_spec.js
@@ -50,47 +50,61 @@ TimeseriesTest.run(() => {
         }
     };
 
-    const verifyAndDropIndex = function(isDowngradeCompatible) {
+    const verifyAndDropIndex = function(isDowngradeCompatible, indexName) {
+        let sawIndex = false;
+
         let userIndexes = coll.getIndexes();
         for (const index of userIndexes) {
-            checkIndexSpec(index, /*userIndex=*/true, isDowngradeCompatible);
+            if (index.name === indexName) {
+                sawIndex = true;
+                checkIndexSpec(index, /*userIndex=*/true, isDowngradeCompatible);
+            }
         }
 
         let bucketIndexes = bucketsColl.getIndexes();
         for (const index of bucketIndexes) {
-            checkIndexSpec(index, /*userIndex=*/false, isDowngradeCompatible);
+            if (index.name === indexName) {
+                sawIndex = true;
+                checkIndexSpec(index, /*userIndex=*/false, isDowngradeCompatible);
+            }
         }
 
-        assert.commandWorked(coll.dropIndexes("*"));
+        assert(sawIndex,
+               `Index with name: ${indexName} is missing: ${tojson({userIndexes, bucketIndexes})}`);
+
+        assert.commandWorked(coll.dropIndexes(indexName));
     };
 
-    assert.commandWorked(coll.createIndex({[timeFieldName]: 1}));
-    verifyAndDropIndex(/*isDowngradeCompatible=*/true);
+    assert.commandWorked(coll.createIndex({[timeFieldName]: 1}, {name: "timefield_downgradable"}));
+    verifyAndDropIndex(/*isDowngradeCompatible=*/true, "timefield_downgradable");
 
-    assert.commandWorked(coll.createIndex({[metaFieldName]: 1}));
-    verifyAndDropIndex(/*isDowngradeCompatible=*/true);
+    assert.commandWorked(coll.createIndex({[metaFieldName]: 1}, {name: "metafield_downgradable"}));
+    verifyAndDropIndex(/*isDowngradeCompatible=*/true, "metafield_downgradable");
 
-    assert.commandWorked(coll.createIndex({[timeFieldName]: 1, [metaFieldName]: 1}));
-    verifyAndDropIndex(/*isDowngradeCompatible=*/true);
+    assert.commandWorked(coll.createIndex({[timeFieldName]: 1, [metaFieldName]: 1},
+                                          {name: "time_meta_field_downgradable"}));
+    verifyAndDropIndex(/*isDowngradeCompatible=*/true, "time_meta_field_downgradable");
 
     if (TimeseriesTest.timeseriesMetricIndexesEnabled(db.getMongo())) {
-        assert.commandWorked(coll.createIndex({x: 1}));
-        verifyAndDropIndex(/*isDowngradeCompatible=*/false);
-
-        assert.commandWorked(coll.createIndex({x: 1}, {partialFilterExpression: {x: {$gt: 5}}}));
-        verifyAndDropIndex(/*isDowngradeCompatible=*/false);
+        assert.commandWorked(coll.createIndex({x: 1}, {name: "x_1"}));
+        verifyAndDropIndex(/*isDowngradeCompatible=*/false, "x_1");
 
         assert.commandWorked(
-            coll.createIndex({[timeFieldName]: 1}, {partialFilterExpression: {x: {$gt: 5}}}));
-        verifyAndDropIndex(/*isDowngradeCompatible=*/false);
+            coll.createIndex({x: 1}, {name: "x_partial", partialFilterExpression: {x: {$gt: 5}}}));
+        verifyAndDropIndex(/*isDowngradeCompatible=*/false, "x_partial");
 
-        assert.commandWorked(
-            coll.createIndex({[metaFieldName]: 1}, {partialFilterExpression: {x: {$gt: 5}}}));
-        verifyAndDropIndex(/*isDowngradeCompatible=*/false);
+        assert.commandWorked(coll.createIndex(
+            {[timeFieldName]: 1}, {name: "time_partial", partialFilterExpression: {x: {$gt: 5}}}));
+        verifyAndDropIndex(/*isDowngradeCompatible=*/false, "time_partial");
+
+        assert.commandWorked(coll.createIndex(
+            {[metaFieldName]: 1}, {name: "meta_partial", partialFilterExpression: {x: {$gt: 5}}}));
+        verifyAndDropIndex(/*isDowngradeCompatible=*/false, "meta_partial");
 
         assert.commandWorked(
-            coll.createIndex({[metaFieldName]: 1, x: 1}, {partialFilterExpression: {x: {$gt: 5}}}));
-        verifyAndDropIndex(/*isDowngradeCompatible=*/false);
+            coll.createIndex({[metaFieldName]: 1, x: 1},
+                             {name: "meta_x_partial", partialFilterExpression: {x: {$gt: 5}}}));
+        verifyAndDropIndex(/*isDowngradeCompatible=*/false, "meta_x_partial");
     }
 
     // Creating an index directly on the buckets collection is permitted. However, these types of
diff --git a/jstests/core/timeseries/timeseries_metric_index_ascending_descending.js b/jstests/core/timeseries/timeseries_metric_index_ascending_descending.js
index 3185a13e63d..bf1dc10e625 100644
--- a/jstests/core/timeseries/timeseries_metric_index_ascending_descending.js
+++ b/jstests/core/timeseries/timeseries_metric_index_ascending_descending.js
@@ -14,6 +14,7 @@
 "use strict";
 
 load("jstests/core/timeseries/libs/timeseries.js");
+load("jstests/libs/fixture_helpers.js");
 
 if (!TimeseriesTest.timeseriesMetricIndexesEnabled(db.getMongo())) {
     jsTestLog(
@@ -99,12 +100,11 @@ TimeseriesTest.run((insert) => {
     assert.eq({x: -1}, userIndexes[userIndexes.length - 1].key);
 
     bucketIndexes = bucketsColl.getIndexes();
-    assert.eq({"control.max.x": -1, "control.min.x": -1},
-              bucketIndexes[bucketIndexes.length - 1].key);
-    testHint(bucketIndexes[bucketIndexes.length - 1].name);
+    let bucketIndex = bucketIndexes[bucketIndexes.length - 1];
+    assert.eq({"control.max.x": -1, "control.min.x": -1}, bucketIndex.key);
+    testHint(bucketIndex.name);
 
-    // Drop index by name.
-    assert.commandWorked(coll.dropIndex(bucketIndexes[0].name));
+    assert.commandWorked(coll.dropIndex(bucketIndex.name));
     bucketIndexes = bucketsColl.getIndexes();
 
     // Test an index on dotted and sub document fields.
@@ -113,11 +113,11 @@ TimeseriesTest.run((insert) => {
     assert.eq({"x.y": 1}, userIndexes[userIndexes.length - 1].key);
 
     bucketIndexes = bucketsColl.getIndexes();
-    assert.eq({"control.min.x.y": 1, "control.max.x.y": 1},
-              bucketIndexes[bucketIndexes.length - 1].key);
-    testHint(bucketIndexes[bucketIndexes.length - 1].name);
+    bucketIndex = bucketIndexes[bucketIndexes.length - 1];
+    assert.eq({"control.min.x.y": 1, "control.max.x.y": 1}, bucketIndex.key);
+    testHint(bucketIndex.name);
 
-    assert.commandWorked(coll.dropIndex(bucketIndexes[0].name));
+    assert.commandWorked(coll.dropIndex(bucketIndex.name));
     bucketIndexes = bucketsColl.getIndexes();
 
     // Test bad input.
@@ -156,10 +156,17 @@ TimeseriesTest.run((insert) => {
     assert.commandWorked(
         bucketsColl.createIndex({"control.max.x.y": -1, "control.min.x.y": -1, "data.x": 1}));
 
-    userIndexes = coll.getIndexes();
-    assert.eq(0, userIndexes.length);
-
-    bucketIndexes = bucketsColl.getIndexes();
-    assert.eq(13, bucketIndexes.length);
+    if (FixtureHelpers.isSharded(bucketsColl)) {
+        // There are more indexes for sharded collections because it includes the shard key index.
+        userIndexes = coll.getIndexes();
+        assert.eq(1, userIndexes.length);
+        bucketIndexes = bucketsColl.getIndexes();
+        assert.eq(14, bucketIndexes.length);
+    } else {
+        userIndexes = coll.getIndexes();
+        assert.eq(0, userIndexes.length);
+        bucketIndexes = bucketsColl.getIndexes();
+        assert.eq(13, bucketIndexes.length);
+    }
 });
 }());
diff --git a/jstests/sharding/drop_indexes_with_stale_config_error.js b/jstests/sharding/drop_indexes_with_stale_config_error.js
index 20d969fd865..a52f83f0983 100644
--- a/jstests/sharding/drop_indexes_with_stale_config_error.js
+++ b/jstests/sharding/drop_indexes_with_stale_config_error.js
@@ -25,13 +25,13 @@ assert.commandWorked(st.s.adminCommand({split: ns, middle: {x: 0}}));
 assert.commandWorked(st.s.adminCommand({moveChunk: ns, find: {x: 100}, to: st.shard1.shardName}));
 flushRoutersAndRefreshShardMetadata(st, {ns});
 
-assert.commandWorked(mongos0Coll.createIndex({x: 1}));
+assert.commandWorked(mongos0Coll.createIndex({y: 1}));
 
 // Move chunk without refreshing the recipient so that the recipient shard throws a
 // StaleShardVersion error upon receiving the drop index command.
 ShardVersioningUtil.moveChunkNotRefreshRecipient(st.s1, ns, st.shard1, st.shard0, {x: 100});
 
-assert.commandWorked(mongos0Coll.dropIndexes({x: 1}));
+assert.commandWorked(mongos0Coll.dropIndexes({y: 1}));
 
 st.stop();
 })();
diff --git a/jstests/sharding/shard_key_index_must_exist.js b/jstests/sharding/shard_key_index_must_exist.js
new file mode 100644
index 00000000000..5c5b5e7f0ca
--- /dev/null
+++ b/jstests/sharding/shard_key_index_must_exist.js
@@ -0,0 +1,76 @@
+(function() {
+'use strict';
+
+const kCantDropShardKeyIndexErrors = [649100, 649101];
+
+let st = new ShardingTest({shards: 1});
+
+assert.commandWorked(st.s.adminCommand({enableSharding: 'test'}));
+
+let testDB = st.getDB('test');
+
+let checkIndex = function(collName, expectedIndexNames) {
+    let indexNotSeen = expectedIndexNames;
+
+    testDB.getCollection(collName).getIndexes().forEach((index) => {
+        assert(expectedIndexNames.includes(index.name),
+               'index should not expected to exist: ' + tojson(index));
+
+        indexNotSeen = indexNotSeen.filter((name) => {
+            return name == index.name;
+        });
+    });
+
+    assert.eq([], indexNotSeen);
+};
+
+(() => {
+    assert.commandWorked(st.s.adminCommand({shardCollection: 'test.user', key: {x: 1}}));
+    assert.commandFailedWithCode(testDB.runCommand({dropIndexes: 'user', index: {x: 1}}),
+                                 kCantDropShardKeyIndexErrors);
+
+    assert.commandWorked(testDB.runCommand({
+        createIndexes: 'user',
+        indexes: [
+            {key: {x: 1, y: 1}, name: 'xy'},
+            {key: {x: 1, z: 1}, name: 'xz'},
+            {key: {a: 1}, name: 'a'}
+        ]
+    }));
+
+    assert.commandWorked(testDB.runCommand({dropIndexes: 'user', index: '*'}));
+    checkIndex('user', ['_id_', 'x_1', 'xy', 'xz']);
+
+    assert.commandWorked(testDB.runCommand({dropIndexes: 'user', index: {x: 1}}));
+    assert.commandWorked(testDB.runCommand({dropIndexes: 'user', index: {x: 1, z: 1}}));
+
+    assert.commandFailedWithCode(testDB.runCommand({dropIndexes: 'user', index: {x: 1, y: 1}}),
+                                 kCantDropShardKeyIndexErrors);
+    assert.commandFailedWithCode(testDB.runCommand({dropIndexes: 'user', index: 'xy'}),
+                                 kCantDropShardKeyIndexErrors);
+
+    checkIndex('user', ['_id_', 'xy']);
+
+    assert.commandWorked(testDB.runCommand({drop: 'user'}));
+})();
+
+(() => {
+    assert.commandWorked(st.s.adminCommand({shardCollection: 'test.hashed', key: {x: 'hashed'}}));
+
+    assert.commandWorked(
+        testDB.runCommand({createIndexes: 'hashed', indexes: [{key: {x: 1, y: 1}, name: 'xy'}]}));
+
+    assert.commandWorked(testDB.runCommand({dropIndexes: 'hashed', index: '*'}));
+
+    checkIndex('hashed', ['_id_', 'x_hashed']);
+
+    assert.commandFailedWithCode(testDB.runCommand({dropIndexes: 'hashed', index: 'x_hashed'}),
+                                 kCantDropShardKeyIndexErrors);
+
+    checkIndex('hashed', ['_id_', 'x_hashed']);
+
+    assert.commandWorked(testDB.runCommand({drop: 'hashed'}));
+})();
+
+st.stop();
+})();
diff --git a/jstests/sharding/timeseries_multiple_mongos.js b/jstests/sharding/timeseries_multiple_mongos.js
index 3bb0a6ea929..761d5f404ee 100644
--- a/jstests/sharding/timeseries_multiple_mongos.js
+++ b/jstests/sharding/timeseries_multiple_mongos.js
@@ -171,10 +171,9 @@ runTest({
     shardKey: {[metaField]: 1},
     cmdObj: {
         dropIndexes: collName,
-        index: {[metaField]: 1},
+        index: "*",
     },
-    numProfilerEntries:
-        {sharded: 2, unsharded: 0 /* command fails when trying to drop a missing index */}
+    numProfilerEntries: {sharded: 2, unsharded: 1}
 });
 
 runTest({
@@ -229,10 +228,9 @@ runTest({
     shardKey: {[metaField]: 1},
     cmdObj: {
         dropIndexes: bucketsCollName,
-        index: {meta: 1},
+        index: "*",
     },
-    numProfilerEntries:
-        {sharded: 2, unsharded: 0 /* command fails when trying to drop a missing index */},
+    numProfilerEntries: {sharded: 2, unsharded: 1},
 });
 
 runTest({
diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp
index 92aec2083c5..ec887ea1d3b 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -1685,7 +1685,7 @@ Status CollectionImpl::truncate(OperationContext* opCtx) {
     }
 
     // 2) drop indexes
-    _indexCatalog->dropAllIndexes(opCtx, this, true);
+    _indexCatalog->dropAllIndexes(opCtx, this, true, {});
 
     // 3) truncate record store
     auto status = _shared->_recordStore->truncate(opCtx);
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index c46c8f76235..73589163649 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -621,7 +621,7 @@ void DatabaseImpl::_dropCollectionIndexes(OperationContext* opCtx,
     invariant(_name.dbName() == nss.db());
     LOGV2_DEBUG(
         20316, 1, "dropCollection: {namespace} - dropAllIndexes start", "namespace"_attr = nss);
-    collection->getIndexCatalog()->dropAllIndexes(opCtx, collection, true);
+    collection->getIndexCatalog()->dropAllIndexes(opCtx, collection, true, {});
 
     invariant(collection->getTotalIndexCount() == 0);
     LOGV2_DEBUG(
diff --git a/src/mongo/db/catalog/drop_indexes.cpp b/src/mongo/db/catalog/drop_indexes.cpp
index 596615bbead..8f3f0655d73 100644
--- a/src/mongo/db/catalog/drop_indexes.cpp
+++ b/src/mongo/db/catalog/drop_indexes.cpp
@@ -49,6 +49,7 @@
 #include "mongo/db/repl_set_member_in_standalone_mode.h"
 #include "mongo/db/s/collection_sharding_state.h"
 #include "mongo/db/s/database_sharding_state.h"
+#include "mongo/db/s/shard_key_index_util.h"
 #include "mongo/db/service_context.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/visit_helper.h"
@@ -289,22 +290,63 @@ void dropReadyIndexes(OperationContext* opCtx,
     }
 
     IndexCatalog* indexCatalog = collection->getIndexCatalog();
+    auto collDescription =
+        CollectionShardingState::get(opCtx, collection->ns())->getCollectionDescription(opCtx);
+
     if (indexNames.front() == "*") {
-        indexCatalog->dropAllIndexes(
-            opCtx, collection, false, [opCtx, collection](const IndexDescriptor* desc) {
-                opCtx->getServiceContext()->getOpObserver()->onDropIndex(opCtx,
-                                                                         collection->ns(),
-                                                                         collection->uuid(),
-                                                                         desc->indexName(),
-                                                                         desc->infoObj());
-            });
-
-        reply->setMsg("non-_id indexes dropped for collection"_sd);
+        if (collDescription.isSharded()) {
+            indexCatalog->dropIndexes(
+                opCtx,
+                collection,
+                [&](const IndexDescriptor* desc) {
+                    if (desc->isIdIndex()) {
+                        return false;
+                    }
+
+                    if (isCompatibleWithShardKey(opCtx,
+                                                 CollectionPtr(collection),
+                                                 desc->getEntry(),
+                                                 collDescription.getKeyPattern(),
+                                                 false /* requiresSingleKey */)) {
+                        return false;
+                    }
+
+                    return true;
+                },
+                [opCtx, collection](const IndexDescriptor* desc) {
+                    opCtx->getServiceContext()->getOpObserver()->onDropIndex(opCtx,
+                                                                             collection->ns(),
+                                                                             collection->uuid(),
+                                                                             desc->indexName(),
+                                                                             desc->infoObj());
+                });
+
+            reply->setMsg("non-_id indexes and non-shard key indexes dropped for collection"_sd);
+        } else {
+            indexCatalog->dropAllIndexes(
+                opCtx, collection, false, [opCtx, collection](const IndexDescriptor* desc) {
+                    opCtx->getServiceContext()->getOpObserver()->onDropIndex(opCtx,
+                                                                             collection->ns(),
+                                                                             collection->uuid(),
+                                                                             desc->indexName(),
+                                                                             desc->infoObj());
+                });
+
+            reply->setMsg("non-_id indexes dropped for collection"_sd);
+        }
         return;
     }
 
     bool includeUnfinished = true;
     for (const auto& indexName : indexNames) {
+        if (collDescription.isSharded()) {
+            uassert(
+                649101,
+                "Cannot drop the only compatible index for this collection's shard key",
+                !isLastShardKeyIndex(
+                    opCtx, collection, indexCatalog, indexName, collDescription.getKeyPattern()));
+        }
+
         auto desc = indexCatalog->findIndexByName(opCtx, indexName, includeUnfinished);
         if (!desc) {
             uasserted(ErrorCodes::IndexNotFound,
@@ -466,6 +508,19 @@ DropIndexesReply dropIndexes(OperationContext* opCtx,
             auto indexCatalog = collection->getWritableCollection(opCtx)->getIndexCatalog();
             const bool includeUnfinished = false;
             for (const auto& indexName : indexNames) {
+                auto collDescription =
+                    CollectionShardingState::get(opCtx, nss)->getCollectionDescription(opCtx);
+
+                if (collDescription.isSharded()) {
+                    uassert(649100,
+                            "Cannot drop the only compatible index for this collection's shard key",
+                            !isLastShardKeyIndex(opCtx,
+                                                 collection->getCollection(),
+                                                 indexCatalog,
+                                                 indexName,
+                                                 collDescription.getKeyPattern()));
+                }
+
                 auto desc = indexCatalog->findIndexByName(opCtx, indexName, includeUnfinished);
                 if (!desc) {
                     // A similar index wasn't created while we yielded the locks during abort.
diff --git a/src/mongo/db/catalog/index_catalog.cpp b/src/mongo/db/catalog/index_catalog.cpp
index 90209b6bdc1..5d0a88b0a9b 100644
--- a/src/mongo/db/catalog/index_catalog.cpp
+++ b/src/mongo/db/catalog/index_catalog.cpp
@@ -39,7 +39,6 @@ namespace mongo {
 using IndexIterator = IndexCatalog::IndexIterator;
 using ReadyIndexesIterator = IndexCatalog::ReadyIndexesIterator;
 using AllIndexesIterator = IndexCatalog::AllIndexesIterator;
-using ShardKeyIndex = IndexCatalog::ShardKeyIndex;
 
 bool IndexIterator::more() {
     if (_start) {
@@ -102,25 +101,6 @@ const IndexCatalogEntry* AllIndexesIterator::_advance() {
     return entry;
 }
 
-ShardKeyIndex::ShardKeyIndex(const IndexDescriptor* indexDescriptor)
-    : _indexDescriptor(indexDescriptor) {
-    tassert(6012300,
-            "The indexDescriptor for ShardKeyIndex(const IndexDescriptor* indexDescripto) must not "
-            "be a nullptr",
-            indexDescriptor != nullptr);
-}
-
-ShardKeyIndex::ShardKeyIndex(const ClusteredIndexSpec& clusteredIndexSpec)
-    : _indexDescriptor(nullptr),
-      _clusteredIndexKeyPattern(clusteredIndexSpec.getKey().getOwned()) {}
-
-const BSONObj& ShardKeyIndex::keyPattern() const {
-    if (_indexDescriptor != nullptr) {
-        return _indexDescriptor->keyPattern();
-    }
-    return _clusteredIndexKeyPattern;
-}
-
 StringData toString(IndexBuildMethod method) {
     switch (method) {
         case IndexBuildMethod::kHybrid:
diff --git a/src/mongo/db/catalog/index_catalog.h b/src/mongo/db/catalog/index_catalog.h
index 2f922b0bcc3..4dc3d500181 100644
--- a/src/mongo/db/catalog/index_catalog.h
+++ b/src/mongo/db/catalog/index_catalog.h
@@ -190,31 +190,6 @@ public:
         std::unique_ptr<std::vector<IndexCatalogEntry*>> _ownedContainer;
     };
 
-    class ShardKeyIndex {
-    public:
-        /**
-         * Wraps information pertaining to the 'index' used as the shard key.
-         *
-         * A clustered index is not tied to an IndexDescriptor whereas all other types of indexes
-         * are. Either the 'index' is a clustered index and '_clusteredIndexKeyPattern' is
-         * non-empty, or '_indexDescriptor' is non-null and a standard index exists.
-         */
-        ShardKeyIndex(const IndexDescriptor* indexDescriptor);
-        ShardKeyIndex(const ClusteredIndexSpec& clusteredIndexSpec);
-
-        const BSONObj& keyPattern() const;
-        const IndexDescriptor* descriptor() const {
-            return _indexDescriptor;
-        }
-
-    private:
-        const IndexDescriptor* _indexDescriptor;
-
-        // Stores the keyPattern when the index is a clustered index and there is no
-        // IndexDescriptor. Empty otherwise.
-        BSONObj _clusteredIndexKeyPattern;
-    };
-
     IndexCatalog() = default;
     virtual ~IndexCatalog() = default;
 
@@ -281,26 +256,6 @@ public:
                                          const BSONObj& key,
                                          bool includeUnfinishedIndexes,
                                          std::vector<const IndexDescriptor*>* matches) const = 0;
-
-    /**
-     * Returns an index suitable for shard key range scans.
-     *
-     * This index:
-     * - must be prefixed by 'shardKey', and
-     * - must not be a partial index.
-     * - must have the simple collation.
-     *
-     * If the parameter 'requireSingleKey' is true, then this index additionally must not be
-     * multi-key.
-     *
-     * If no such index exists, returns NULL.
-     */
-    virtual const boost::optional<ShardKeyIndex> findShardKeyPrefixedIndex(
-        OperationContext* opCtx,
-        const CollectionPtr& collection,
-        const BSONObj& shardKey,
-        bool requireSingleKey) const = 0;
-
     virtual void findIndexByType(OperationContext* opCtx,
                                  const std::string& type,
                                  std::vector<const IndexDescriptor*>& matches,
@@ -419,6 +374,16 @@ public:
         const std::vector<BSONObj>& indexSpecsToBuild) const = 0;
 
     /**
+     * Drops indexes in the index catalog that returns true when it's descriptor returns true for
+     * 'matchFn'. If 'onDropFn' is provided, it will be called before each index is dropped to
+     * allow timestamping each individual drop.
+     */
+    virtual void dropIndexes(OperationContext* opCtx,
+                             Collection* collection,
+                             std::function<bool(const IndexDescriptor*)> matchFn,
+                             std::function<void(const IndexDescriptor*)> onDropFn) = 0;
+
+    /**
      * Drops all indexes in the index catalog, optionally dropping the id index depending on the
      * 'includingIdIndex' parameter value. If 'onDropFn' is provided, it will be called before each
      * index is dropped to allow timestamping each individual drop.
@@ -427,9 +392,6 @@ public:
                                 Collection* collection,
                                 bool includingIdIndex,
                                 std::function<void(const IndexDescriptor*)> onDropFn) = 0;
-    virtual void dropAllIndexes(OperationContext* opCtx,
-                                Collection* collection,
-                                bool includingIdIndex) = 0;
 
     /**
      * Drops the index given its descriptor.
diff --git a/src/mongo/db/catalog/index_catalog_impl.cpp b/src/mongo/db/catalog/index_catalog_impl.cpp
index 115c664d0fc..0104b6bd2d2 100644
--- a/src/mongo/db/catalog/index_catalog_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_impl.cpp
@@ -335,8 +335,7 @@ Status IndexCatalogImpl::_isNonIDIndexAndNotAllowedToBuild(OperationContext* opC
 void IndexCatalogImpl::_logInternalState(OperationContext* opCtx,
                                          const CollectionPtr& collection,
                                          long long numIndexesInCollectionCatalogEntry,
-                                         const std::vector<std::string>& indexNamesToDrop,
-                                         bool haveIdIndex) {
+                                         const std::vector<std::string>& indexNamesToDrop) {
     invariant(opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_X));
 
     LOGV2_ERROR(20365,
@@ -345,8 +344,7 @@ void IndexCatalogImpl::_logInternalState(OperationContext* opCtx,
                 "numIndexesInCollectionCatalogEntry"_attr = numIndexesInCollectionCatalogEntry,
                 "numReadyIndexes"_attr = _readyIndexes.size(),
                 "numBuildingIndexes"_attr = _buildingIndexes.size(),
-                "indexNamesToDrop"_attr = indexNamesToDrop,
-                "haveIdIndex"_attr = haveIdIndex);
+                "indexNamesToDrop"_attr = indexNamesToDrop);
 
     // Report the ready indexes.
     for (const auto& entry : _readyIndexes) {
@@ -1092,15 +1090,15 @@ BSONObj IndexCatalogImpl::getDefaultIdIndexSpec(const CollectionPtr& collection)
     return b.obj();
 }
 
-void IndexCatalogImpl::dropAllIndexes(OperationContext* opCtx,
-                                      Collection* collection,
-                                      bool includingIdIndex,
-                                      std::function<void(const IndexDescriptor*)> onDropFn) {
+void IndexCatalogImpl::dropIndexes(OperationContext* opCtx,
+                                   Collection* collection,
+                                   std::function<bool(const IndexDescriptor*)> matchFn,
+                                   std::function<void(const IndexDescriptor*)> onDropFn) {
     uassert(ErrorCodes::BackgroundOperationInProgressForNamespace,
             str::stream() << "cannot perform operation: an index build is currently running",
             !haveAnyIndexesInProgress());
 
-    bool haveIdIndex = false;
+    bool didExclude = false;
 
     invariant(_buildingIndexes.size() == 0);
     vector<string> indexNamesToDrop;
@@ -1110,11 +1108,11 @@ void IndexCatalogImpl::dropAllIndexes(OperationContext* opCtx,
         while (ii->more()) {
             seen++;
             const IndexDescriptor* desc = ii->next()->descriptor();
-            if (desc->isIdIndex() && includingIdIndex == false) {
-                haveIdIndex = true;
-                continue;
+            if (matchFn(desc)) {
+                indexNamesToDrop.push_back(desc->indexName());
+            } else {
+                didExclude = true;
             }
-            indexNamesToDrop.push_back(desc->indexName());
         }
         invariant(seen == numIndexesTotal(opCtx));
     }
@@ -1139,18 +1137,10 @@ void IndexCatalogImpl::dropAllIndexes(OperationContext* opCtx,
 
     long long numIndexesInCollectionCatalogEntry = collection->getTotalIndexCount();
 
-    if (haveIdIndex) {
-        fassert(17324, numIndexesTotal(opCtx) == 1);
-        fassert(17325, numIndexesReady(opCtx) == 1);
-        fassert(17326, numIndexesInCollectionCatalogEntry == 1);
-        fassert(17336, _readyIndexes.size() == 1);
-    } else {
+    if (!didExclude) {
         if (numIndexesTotal(opCtx) || numIndexesInCollectionCatalogEntry || _readyIndexes.size()) {
-            _logInternalState(opCtx,
-                              collection,
-                              numIndexesInCollectionCatalogEntry,
-                              indexNamesToDrop,
-                              haveIdIndex);
+            _logInternalState(
+                opCtx, collection, numIndexesInCollectionCatalogEntry, indexNamesToDrop);
         }
         fassert(17327, numIndexesTotal(opCtx) == 0);
         fassert(17328, numIndexesInCollectionCatalogEntry == 0);
@@ -1160,8 +1150,18 @@ void IndexCatalogImpl::dropAllIndexes(OperationContext* opCtx,
 
 void IndexCatalogImpl::dropAllIndexes(OperationContext* opCtx,
                                       Collection* collection,
-                                      bool includingIdIndex) {
-    dropAllIndexes(opCtx, collection, includingIdIndex, {});
+                                      bool includingIdIndex,
+                                      std::function<void(const IndexDescriptor*)> onDropFn) {
+    dropIndexes(opCtx,
+                collection,
+                [includingIdIndex](const IndexDescriptor* indexDescriptor) {
+                    if (includingIdIndex) {
+                        return true;
+                    }
+
+                    return !indexDescriptor->isIdIndex();
+                },
+                onDropFn);
 }
 
 Status IndexCatalogImpl::dropIndex(OperationContext* opCtx,
@@ -1375,46 +1375,6 @@ void IndexCatalogImpl::findIndexesByKeyPattern(OperationContext* opCtx,
     }
 }
 
-const boost::optional<IndexCatalog::ShardKeyIndex> IndexCatalogImpl::findShardKeyPrefixedIndex(
-    OperationContext* opCtx,
-    const CollectionPtr& collection,
-    const BSONObj& shardKey,
-    bool requireSingleKey) const {
-    if (collection->isClustered() &&
-        clustered_util::matchesClusterKey(shardKey, collection->getClusteredInfo())) {
-        auto clusteredIndexSpec = collection->getClusteredInfo()->getIndexSpec();
-        return IndexCatalog::ShardKeyIndex(clusteredIndexSpec);
-    }
-
-    const IndexDescriptor* best = nullptr;
-
-    std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, false);
-    while (ii->more()) {
-        const IndexCatalogEntry* entry = ii->next();
-        const IndexDescriptor* desc = entry->descriptor();
-        bool hasSimpleCollation = desc->collation().isEmpty();
-
-        if (desc->isPartial() || desc->isSparse())
-            continue;
-
-        if (!shardKey.isPrefixOf(desc->keyPattern(), SimpleBSONElementComparator::kInstance))
-            continue;
-
-        if (!entry->isMultikey(opCtx, collection) && hasSimpleCollation) {
-            return IndexCatalog::ShardKeyIndex(desc);
-        }
-
-        if (!requireSingleKey && hasSimpleCollation)
-            best = desc;
-    }
-
-    if (best != nullptr) {
-        return IndexCatalog::ShardKeyIndex(best);
-    }
-
-    return boost::none;
-}
-
 void IndexCatalogImpl::findIndexByType(OperationContext* opCtx,
                                        const string& type,
                                        vector<const IndexDescriptor*>& matches,
diff --git a/src/mongo/db/catalog/index_catalog_impl.h b/src/mongo/db/catalog/index_catalog_impl.h
index bcdfd51fcbb..259732da61c 100644
--- a/src/mongo/db/catalog/index_catalog_impl.h
+++ b/src/mongo/db/catalog/index_catalog_impl.h
@@ -122,26 +122,6 @@ public:
                                  const BSONObj& key,
                                  bool includeUnfinishedIndexes,
                                  std::vector<const IndexDescriptor*>* matches) const override;
-
-    /**
-     * Returns an index suitable for shard key range scans.
-     *
-     * This index:
-     * - must be prefixed by 'shardKey', and
-     * - must not be a partial index.
-     * - must have the simple collation.
-     *
-     * If the parameter 'requireSingleKey' is true, then this index additionally must not be
-     * multi-key.
-     *
-     * If no such index exists, returns NULL.
-     */
-    const boost::optional<ShardKeyIndex> findShardKeyPrefixedIndex(
-        OperationContext* opCtx,
-        const CollectionPtr& collection,
-        const BSONObj& shardKey,
-        bool requireSingleKey) const override;
-
     void findIndexByType(OperationContext* opCtx,
                          const std::string& type,
                          std::vector<const IndexDescriptor*>& matches,
@@ -207,19 +187,14 @@ public:
         const CollectionPtr& collection,
         const std::vector<BSONObj>& indexSpecsToBuild) const override;
 
-    /**
-     * Drops all indexes in the index catalog, optionally dropping the id index depending on the
-     * 'includingIdIndex' parameter value. If the 'droppedIndexes' parameter is not null,
-     * it is filled with the names and index info of the dropped indexes.
-     */
+    void dropIndexes(OperationContext* opCtx,
+                     Collection* collection,
+                     std::function<bool(const IndexDescriptor*)> matchFn,
+                     std::function<void(const IndexDescriptor*)> onDropFn) override;
     void dropAllIndexes(OperationContext* opCtx,
                         Collection* collection,
                         bool includingIdIndex,
                         std::function<void(const IndexDescriptor*)> onDropFn) override;
-    void dropAllIndexes(OperationContext* opCtx,
-                        Collection* collection,
-                        bool includingIdIndex) override;
-
 
     Status dropIndex(OperationContext* opCtx,
                      Collection* collection,
@@ -414,8 +389,7 @@ private:
     void _logInternalState(OperationContext* opCtx,
                            const CollectionPtr& collection,
                            long long numIndexesInCollectionCatalogEntry,
-                           const std::vector<std::string>& indexNamesToDrop,
-                           bool haveIdIndex);
+                           const std::vector<std::string>& indexNamesToDrop);
 
     IndexCatalogEntryContainer _readyIndexes;
     IndexCatalogEntryContainer _buildingIndexes;
diff --git a/src/mongo/db/commands/dbcommands.cpp b/src/mongo/db/commands/dbcommands.cpp
index 2267c619692..05f4f383fe9 100644
--- a/src/mongo/db/commands/dbcommands.cpp
+++ b/src/mongo/db/commands/dbcommands.cpp
@@ -67,7 +67,6 @@
 #include "mongo/db/drop_gen.h"
 #include "mongo/db/exec/working_set_common.h"
 #include "mongo/db/index/index_access_method.h"
-#include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/introspect.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/json.h"
@@ -90,6 +89,7 @@
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/request_execution_context.h"
 #include "mongo/db/s/collection_sharding_state.h"
+#include "mongo/db/s/shard_key_index_util.h"
 #include "mongo/db/stats/storage_stats.h"
 #include "mongo/db/storage/storage_engine_init.h"
 #include "mongo/db/timeseries/timeseries_collmod.h"
@@ -387,11 +387,11 @@ public:
                 keyPattern = Helpers::inferKeyPattern(min);
             }
 
-            auto catalog = collection->getIndexCatalog();
-            auto shardKeyIdx = catalog->findShardKeyPrefixedIndex(opCtx,
-                                                                  *collection,
-                                                                  keyPattern,
-                                                                  /*requireSingleKey=*/true);
+            auto shardKeyIdx = findShardKeyPrefixedIndex(opCtx,
+                                                         *collection,
+                                                         collection->getIndexCatalog(),
+                                                         keyPattern,
+                                                         /*requireSingleKey=*/true);
 
             if (!shardKeyIdx) {
                 errmsg = "couldn't find valid index containing key pattern";
diff --git a/src/mongo/db/commands/drop_indexes.cpp b/src/mongo/db/commands/drop_indexes.cpp
index 3f60b169975..bfd6739f36c 100644
--- a/src/mongo/db/commands/drop_indexes.cpp
+++ b/src/mongo/db/commands/drop_indexes.cpp
@@ -231,7 +231,7 @@ public:
         writeConflictRetry(opCtx, "dropAllIndexes", toReIndexNss.ns(), [&] {
             WriteUnitOfWork wunit(opCtx);
             collection.getWritableCollection()->getIndexCatalog()->dropAllIndexes(
-                opCtx, collection.getWritableCollection(), true);
+                opCtx, collection.getWritableCollection(), true, {});
 
             swIndexesToRebuild =
                 indexer->init(opCtx, collection, all, MultiIndexBlock::kNoopOnInitFn);
diff --git a/src/mongo/db/query/internal_plans.cpp b/src/mongo/db/query/internal_plans.cpp
index 425d2857eb5..81ffba55a4e 100644
--- a/src/mongo/db/query/internal_plans.cpp
+++ b/src/mongo/db/query/internal_plans.cpp
@@ -316,7 +316,7 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::deleteWith
 std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::shardKeyIndexScan(
     OperationContext* opCtx,
     const CollectionPtr* collection,
-    const IndexCatalog::ShardKeyIndex& shardKeyIdx,
+    const ShardKeyIndex& shardKeyIdx,
     const BSONObj& startKey,
     const BSONObj& endKey,
     BoundInclusion boundInclusion,
@@ -344,7 +344,7 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::deleteWith
     OperationContext* opCtx,
     const CollectionPtr* coll,
     std::unique_ptr<DeleteStageParams> params,
-    const IndexCatalog::ShardKeyIndex& shardKeyIdx,
+    const ShardKeyIndex& shardKeyIdx,
     const BSONObj& startKey,
     const BSONObj& endKey,
     BoundInclusion boundInclusion,
diff --git a/src/mongo/db/query/internal_plans.h b/src/mongo/db/query/internal_plans.h
index 781b336526f..967f9c3b073 100644
--- a/src/mongo/db/query/internal_plans.h
+++ b/src/mongo/db/query/internal_plans.h
@@ -35,6 +35,7 @@
 #include "mongo/db/query/index_bounds.h"
 #include "mongo/db/query/plan_executor.h"
 #include "mongo/db/record_id.h"
+#include "mongo/db/s/shard_key_index_util.h"
 
 namespace mongo {
 
@@ -135,7 +136,7 @@ public:
     static std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> shardKeyIndexScan(
         OperationContext* opCtx,
         const CollectionPtr* collection,
-        const IndexCatalog::ShardKeyIndex& shardKeyIdx,
+        const ShardKeyIndex& shardKeyIdx,
         const BSONObj& startKey,
         const BSONObj& endKey,
         BoundInclusion boundInclusion,
@@ -153,7 +154,7 @@ public:
         OperationContext* opCtx,
         const CollectionPtr* collection,
         std::unique_ptr<DeleteStageParams> params,
-        const IndexCatalog::ShardKeyIndex& shardKeyIdx,
+        const ShardKeyIndex& shardKeyIdx,
         const BSONObj& startKey,
         const BSONObj& endKey,
         BoundInclusion boundInclusion,
diff --git a/src/mongo/db/s/SConscript b/src/mongo/db/s/SConscript
index c2d83bdaedc..11ff08d61e8 100644
--- a/src/mongo/db/s/SConscript
+++ b/src/mongo/db/s/SConscript
@@ -17,6 +17,7 @@ env.Library(
         'database_sharding_state.cpp',
         'global_user_write_block_state.cpp',
         'operation_sharding_state.cpp',
+        'shard_key_index_util.cpp',
         'sharding_api_d_params.idl',
         'sharding_migration_critical_section.cpp',
         'sharding_state.cpp',
@@ -28,6 +29,7 @@ env.Library(
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/base',
+        '$BUILD_DIR/mongo/db/catalog/index_catalog',
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
         '$BUILD_DIR/mongo/db/range_arithmetic',
         '$BUILD_DIR/mongo/s/grid',
@@ -555,6 +557,7 @@ env.CppUnitTest(
         'session_catalog_migration_destination_test.cpp',
         'session_catalog_migration_source_test.cpp',
         'shard_local_test.cpp',
+        'shard_key_index_util_test.cpp',
         'shard_metadata_util_test.cpp',
         'shard_server_catalog_cache_loader_test.cpp',
         'sharding_data_transform_cumulative_metrics_test.cpp',
@@ -579,6 +582,7 @@ env.CppUnitTest(
         '$BUILD_DIR/mongo/client/remote_command_targeter_mock',
         '$BUILD_DIR/mongo/db/auth/authmocks',
         '$BUILD_DIR/mongo/db/catalog/catalog_helpers',
+        '$BUILD_DIR/mongo/db/catalog/catalog_test_fixture',
         '$BUILD_DIR/mongo/db/commands/server_status',
         '$BUILD_DIR/mongo/db/exec/document_value/document_value_test_util',
         '$BUILD_DIR/mongo/db/keys_collection_client_direct',
diff --git a/src/mongo/db/s/auto_split_vector.cpp b/src/mongo/db/s/auto_split_vector.cpp
index ecb2b568c38..41e1149a5f2 100644
--- a/src/mongo/db/s/auto_split_vector.cpp
+++ b/src/mongo/db/s/auto_split_vector.cpp
@@ -35,7 +35,6 @@
 
 #include "mongo/base/status_with.h"
 #include "mongo/db/bson/dotted_path_support.h"
-#include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/dbhelpers.h"
 #include "mongo/db/exec/working_set_common.h"
@@ -44,6 +43,8 @@
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/query/internal_plans.h"
 #include "mongo/db/query/plan_executor.h"
+#include "mongo/db/s/shard_key_index_util.h"
+
 #include "mongo/logv2/log.h"
 
 namespace mongo {
@@ -62,8 +63,9 @@ BSONObj prettyKey(const BSONObj& keyPattern, const BSONObj& key) {
  * object extended to cover the entire shardKey. See KeyPattern::extendRangeBound documentation for
  * some examples.
  */
-const std::tuple<BSONObj, BSONObj> getMinMaxExtendedBounds(
-    const IndexCatalog::ShardKeyIndex& shardKeyIdx, const BSONObj& min, const BSONObj& max) {
+const std::tuple<BSONObj, BSONObj> getMinMaxExtendedBounds(const ShardKeyIndex& shardKeyIdx,
+                                                           const BSONObj& min,
+                                                           const BSONObj& max) {
     KeyPattern kp(shardKeyIdx.keyPattern());
 
     // Extend min to get (min, MinKey, MinKey, ....)
@@ -122,9 +124,11 @@ std::pair<std::vector<BSONObj>, bool> autoSplitVector(OperationContext* opCtx,
 
         // Allow multiKey based on the invariant that shard keys must be single-valued. Therefore,
         // any multi-key index prefixed by shard key cannot be multikey over the shard key fields.
-        auto catalog = collection->getIndexCatalog();
-        auto shardKeyIdx = catalog->findShardKeyPrefixedIndex(
-            opCtx, *collection, keyPattern, /*requireSingleKey=*/false);
+        auto shardKeyIdx = findShardKeyPrefixedIndex(opCtx,
+                                                     *collection,
+                                                     collection->getIndexCatalog(),
+                                                     keyPattern,
+                                                     /*requireSingleKey=*/false);
         uassert(ErrorCodes::IndexNotFound,
                 str::stream() << "couldn't find index over splitting key "
                               << keyPattern.clientReadable().toString(),
diff --git a/src/mongo/db/s/check_sharding_index_command.cpp b/src/mongo/db/s/check_sharding_index_command.cpp
index 36f86f31908..16dbbf8a2fe 100644
--- a/src/mongo/db/s/check_sharding_index_command.cpp
+++ b/src/mongo/db/s/check_sharding_index_command.cpp
@@ -37,6 +37,7 @@
 #include "mongo/db/commands.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/keypattern.h"
+#include "mongo/db/s/shard_key_index_util.h"
 
 namespace mongo {
 namespace {
@@ -93,9 +94,11 @@ public:
             return false;
         }
 
-        auto catalog = collection->getIndexCatalog();
-        auto shardKeyIdx = catalog->findShardKeyPrefixedIndex(
-            opCtx, *collection, keyPattern, /*requireSingleKey=*/true);
+        auto shardKeyIdx = findShardKeyPrefixedIndex(opCtx,
+                                                     *collection,
+                                                     collection->getIndexCatalog(),
+                                                     keyPattern,
+                                                     /*requireSingleKey=*/true);
         if (!shardKeyIdx) {
             errmsg = "couldn't find valid index for shard key";
             return false;
diff --git a/src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp b/src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp
index 96077785587..d8ab0893eb6 100644
--- a/src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp
+++ b/src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp
@@ -40,11 +40,11 @@
 #include "mongo/db/dbhelpers.h"
 #include "mongo/db/exec/working_set_common.h"
 #include "mongo/db/index/index_access_method.h"
-#include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/repl/optime.h"
 #include "mongo/db/repl/replication_process.h"
 #include "mongo/db/s/collection_sharding_runtime.h"
 #include "mongo/db/s/migration_source_manager.h"
+#include "mongo/db/s/shard_key_index_util.h"
 #include "mongo/db/s/sharding_runtime_d_params_gen.h"
 #include "mongo/db/s/sharding_statistics.h"
 #include "mongo/db/s/start_chunk_clone_request.h"
@@ -844,9 +844,11 @@ MigrationChunkClonerSourceLegacy::_getIndexScanExecutor(
     InternalPlanner::IndexScanOptions scanOption) {
     // Allow multiKey based on the invariant that shard keys must be single-valued. Therefore, any
     // multi-key index prefixed by shard key cannot be multikey over the shard key fields.
-    auto catalog = collection->getIndexCatalog();
-    auto shardKeyIdx = catalog->findShardKeyPrefixedIndex(
-        opCtx, collection, _shardKeyPattern.toBSON(), /*requireSingleKey=*/false);
+    auto shardKeyIdx = findShardKeyPrefixedIndex(opCtx,
+                                                 collection,
+                                                 collection->getIndexCatalog(),
+                                                 _shardKeyPattern.toBSON(),
+                                                 /*requireSingleKey=*/false);
     if (!shardKeyIdx) {
         return {ErrorCodes::IndexNotFound,
                 str::stream() << "can't find index with prefix " << _shardKeyPattern.toBSON()
diff --git a/src/mongo/db/s/range_deletion_util.cpp b/src/mongo/db/s/range_deletion_util.cpp
index f0b406a45ea..196097126c8 100644
--- a/src/mongo/db/s/range_deletion_util.cpp
+++ b/src/mongo/db/s/range_deletion_util.cpp
@@ -38,7 +38,6 @@
 
 #include <boost/optional.hpp>
 
-#include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/client.h"
 #include "mongo/db/concurrency/write_conflict_exception.h"
@@ -56,6 +55,7 @@
 #include "mongo/db/repl/repl_client_info.h"
 #include "mongo/db/repl/wait_for_majority_service.h"
 #include "mongo/db/s/migration_util.h"
+#include "mongo/db/s/shard_key_index_util.h"
 #include "mongo/db/s/sharding_statistics.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/storage/remove_saver.h"
@@ -133,9 +133,8 @@ StatusWith<int> deleteNextBatch(OperationContext* opCtx,
 
     // The IndexChunk has a keyPattern that may apply to more than one index - we need to
     // select the index and get the full index keyPattern here.
-    auto catalog = collection->getIndexCatalog();
-    auto shardKeyIdx = catalog->findShardKeyPrefixedIndex(
-        opCtx, collection, keyPattern, /*requireSingleKey=*/false);
+    auto shardKeyIdx = findShardKeyPrefixedIndex(
+        opCtx, collection, collection->getIndexCatalog(), keyPattern, /*requireSingleKey=*/false);
     if (!shardKeyIdx) {
         LOGV2_ERROR_OPTIONS(23765,
                             {logv2::UserAssertAfterLog(ErrorCodes::InternalError)},
@@ -644,9 +643,11 @@ void setOrphanCountersOnRangeDeletionTasks(OperationContext* opCtx) {
 
             KeyPattern keyPattern;
             uassertStatusOK(deletionTask.getRange().extractKeyPattern(&keyPattern));
-            auto catalog = collection->getIndexCatalog();
-            auto shardKeyIdx = catalog->findShardKeyPrefixedIndex(
-                opCtx, *collection, keyPattern.toBSON(), /*requireSingleKey=*/false);
+            auto shardKeyIdx = findShardKeyPrefixedIndex(opCtx,
+                                                         *collection,
+                                                         collection->getIndexCatalog(),
+                                                         keyPattern.toBSON(),
+                                                         /*requireSingleKey=*/false);
 
             uassert(ErrorCodes::IndexNotFound,
                     str::stream() << "couldn't find index over shard key " << keyPattern.toBSON()
diff --git a/src/mongo/db/s/shard_key_index_util.cpp b/src/mongo/db/s/shard_key_index_util.cpp
new file mode 100644
index 00000000000..56d6ca08656
--- /dev/null
+++ b/src/mongo/db/s/shard_key_index_util.cpp
@@ -0,0 +1,150 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kSharding
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/bson/simple_bsonelement_comparator.h"
+#include "mongo/db/catalog/clustered_collection_util.h"
+#include "mongo/db/catalog/collection.h"
+#include "mongo/db/catalog/index_catalog.h"
+#include "mongo/db/s/shard_key_index_util.h"
+
+namespace mongo {
+
+namespace {
+const boost::optional<ShardKeyIndex> _findShardKeyPrefixedIndex(
+    OperationContext* opCtx,
+    const CollectionPtr& collection,
+    const IndexCatalog* indexCatalog,
+    const boost::optional<std::string>& excludeName,
+    const BSONObj& shardKey,
+    bool requireSingleKey) {
+    if (collection->isClustered() &&
+        clustered_util::matchesClusterKey(shardKey, collection->getClusteredInfo())) {
+        auto clusteredIndexSpec = collection->getClusteredInfo()->getIndexSpec();
+        return ShardKeyIndex(clusteredIndexSpec);
+    }
+
+    const IndexDescriptor* best = nullptr;
+
+    auto indexIterator = indexCatalog->getIndexIterator(opCtx, false);
+    while (indexIterator->more()) {
+        auto indexEntry = indexIterator->next();
+        auto indexDescriptor = indexEntry->descriptor();
+
+        if (excludeName && indexDescriptor->indexName() == *excludeName) {
+            continue;
+        }
+
+        if (isCompatibleWithShardKey(opCtx, collection, indexEntry, shardKey, requireSingleKey)) {
+            if (!indexEntry->isMultikey(opCtx, collection)) {
+                return ShardKeyIndex(indexDescriptor);
+            }
+
+            best = indexDescriptor;
+        }
+    }
+
+    if (best != nullptr) {
+        return ShardKeyIndex(best);
+    }
+
+    return boost::none;
+}
+
+}  // namespace
+
+ShardKeyIndex::ShardKeyIndex(const IndexDescriptor* indexDescriptor)
+    : _indexDescriptor(indexDescriptor) {
+    tassert(6012300,
+            "The indexDescriptor for ShardKeyIndex(const IndexDescriptor* indexDescripto) must not "
+            "be a nullptr",
+            indexDescriptor != nullptr);
+}
+
+ShardKeyIndex::ShardKeyIndex(const ClusteredIndexSpec& clusteredIndexSpec)
+    : _indexDescriptor(nullptr),
+      _clusteredIndexKeyPattern(clusteredIndexSpec.getKey().getOwned()) {}
+
+const BSONObj& ShardKeyIndex::keyPattern() const {
+    if (_indexDescriptor != nullptr) {
+        return _indexDescriptor->keyPattern();
+    }
+    return _clusteredIndexKeyPattern;
+}
+
+bool isCompatibleWithShardKey(OperationContext* opCtx,
+                              const CollectionPtr& collection,
+                              const IndexCatalogEntry* indexEntry,
+                              const BSONObj& shardKey,
+                              bool requireSingleKey) {
+    auto desc = indexEntry->descriptor();
+    bool hasSimpleCollation = desc->collation().isEmpty();
+
+    if (desc->isPartial() || desc->isSparse()) {
+        return false;
+    }
+
+    if (!shardKey.isPrefixOf(desc->keyPattern(), SimpleBSONElementComparator::kInstance)) {
+        return false;
+    }
+
+    if (!indexEntry->isMultikey(opCtx, collection) && hasSimpleCollation) {
+        return true;
+    }
+
+    if (!requireSingleKey && hasSimpleCollation) {
+        return true;
+    }
+
+    return false;
+}
+
+bool isLastShardKeyIndex(OperationContext* opCtx,
+                         const CollectionPtr& collection,
+                         const IndexCatalog* indexCatalog,
+                         const std::string& indexName,
+                         const BSONObj& shardKey) {
+    return !_findShardKeyPrefixedIndex(
+                opCtx, collection, indexCatalog, indexName, shardKey, false /* requireSingleKey */)
+                .is_initialized();
+}
+
+const boost::optional<ShardKeyIndex> findShardKeyPrefixedIndex(OperationContext* opCtx,
+                                                               const CollectionPtr& collection,
+                                                               const IndexCatalog* indexCatalog,
+                                                               const BSONObj& shardKey,
+                                                               bool requireSingleKey) {
+    return _findShardKeyPrefixedIndex(
+        opCtx, collection, indexCatalog, boost::none, shardKey, requireSingleKey);
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/s/shard_key_index_util.h b/src/mongo/db/s/shard_key_index_util.h
new file mode 100644
index 00000000000..db59c57ceca
--- /dev/null
+++ b/src/mongo/db/s/shard_key_index_util.h
@@ -0,0 +1,104 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/bson/bsonobj.h"
+#include "mongo/db/catalog/clustered_collection_options_gen.h"
+#include "mongo/db/catalog/index_catalog.h"
+
+namespace mongo {
+
+class Collection;
+class CollectionPtr;
+
+class IndexDescriptor;
+
+class ShardKeyIndex {
+public:
+    /**
+     * Wraps information pertaining to the 'index' used as the shard key.
+     *
+     * A clustered index is not tied to an IndexDescriptor whereas all other types of indexes
+     * are. Either the 'index' is a clustered index and '_clusteredIndexKeyPattern' is
+     * non-empty, or '_indexDescriptor' is non-null and a standard index exists.
+     */
+    ShardKeyIndex(const IndexDescriptor* indexDescriptor);
+    ShardKeyIndex(const ClusteredIndexSpec& clusteredIndexSpec);
+
+    const BSONObj& keyPattern() const;
+    const IndexDescriptor* descriptor() const {
+        return _indexDescriptor;
+    }
+
+private:
+    const IndexDescriptor* _indexDescriptor;
+
+    // Stores the keyPattern when the index is a clustered index and there is no
+    // IndexDescriptor. Empty otherwise.
+    BSONObj _clusteredIndexKeyPattern;
+};
+
+/**
+ * Returns true if the given index is compatible with the shard key pattern.
+ */
+bool isCompatibleWithShardKey(OperationContext* opCtx,
+                              const CollectionPtr& collection,
+                              const IndexCatalogEntry* indexEntry,
+                              const BSONObj& shardKey,
+                              bool requireSingleKey);
+
+/**
+ * Returns an index suitable for shard key range scans if it exists.
+ *
+ * This index:
+ * - must be prefixed by 'shardKey', and
+ * - must not be a partial index.
+ * - must have the simple collation.
+ *
+ * If the parameter 'requireSingleKey' is true, then this index additionally must not be
+ * multi-key.
+ */
+const boost::optional<ShardKeyIndex> findShardKeyPrefixedIndex(OperationContext* opCtx,
+                                                               const CollectionPtr& collection,
+                                                               const IndexCatalog* indexCatalog,
+                                                               const BSONObj& shardKey,
+                                                               bool requireSingleKey);
+
+/**
+ * Returns true if the given index name is the last remaining index that is compatible with the
+ * shard key index.
+ */
+bool isLastShardKeyIndex(OperationContext* opCtx,
+                         const CollectionPtr& collection,
+                         const IndexCatalog* indexCatalog,
+                         const std::string& indexName,
+                         const BSONObj& shardKey);
+
+}  // namespace mongo
diff --git a/src/mongo/db/s/shard_key_index_util_test.cpp b/src/mongo/db/s/shard_key_index_util_test.cpp
new file mode 100644
index 00000000000..c9c1b4fdc23
--- /dev/null
+++ b/src/mongo/db/s/shard_key_index_util_test.cpp
@@ -0,0 +1,227 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/catalog/catalog_test_fixture.h"
+#include "mongo/db/catalog/collection.h"
+#include "mongo/db/catalog/index_catalog_entry_impl.h"
+#include "mongo/db/catalog_raii.h"
+#include "mongo/db/dbdirectclient.h"
+#include "mongo/db/index/index_descriptor.h"
+#include "mongo/db/s/shard_key_index_util.h"
+
+namespace mongo {
+namespace {
+
+constexpr int kIndexVersion = 2;
+
+class ShardKeyIndexUtilTest : public CatalogTestFixture {
+public:
+    ShardKeyIndexUtilTest() : CatalogTestFixture() {}
+
+    void createIndex(const BSONObj& spec) {
+        // Build the specified index on the collection.
+        WriteUnitOfWork wuow(opCtx());
+        auto* indexCatalog = _coll->getWritableCollection(opCtx())->getIndexCatalog();
+        uassertStatusOK(indexCatalog->createIndexOnEmptyCollection(
+            opCtx(), _coll->getWritableCollection(opCtx()), spec));
+        wuow.commit();
+    }
+
+    const NamespaceString& nss() const {
+        return _nss;
+    }
+
+    const CollectionPtr& coll() const {
+        return (*_coll).getCollection();
+    }
+
+    OperationContext* opCtx() {
+        return operationContext();
+    }
+
+protected:
+    void setUp() override {
+        CatalogTestFixture::setUp();
+        _coll.emplace(opCtx(), _nss, MODE_X);
+        ASSERT_OK(storageInterface()->createCollection(opCtx(), _nss, {}));
+    }
+
+    void tearDown() override {
+        _coll.reset();
+        CatalogTestFixture::tearDown();
+    }
+
+private:
+    const NamespaceString _nss{"test.user"};
+    boost::optional<AutoGetCollection> _coll;
+};
+
+TEST_F(ShardKeyIndexUtilTest, SimpleKeyPattern) {
+    createIndex(BSON("key" << BSON("y" << 1) << "name"
+                           << "y"
+                           << "v" << kIndexVersion));
+    createIndex(BSON("key" << BSON("x" << 1) << "name"
+                           << "x"
+                           << "v" << kIndexVersion));
+
+    auto index = findShardKeyPrefixedIndex(
+        opCtx(), coll(), coll()->getIndexCatalog(), BSON("x" << 1), true /* requireSingleKey */);
+
+    ASSERT_TRUE(index);
+    ASSERT_EQ("x", index->descriptor()->indexName());
+}
+
+TEST_F(ShardKeyIndexUtilTest, HashedKeyPattern) {
+    createIndex(BSON("key" << BSON("y" << 1) << "name"
+                           << "y"
+                           << "v" << kIndexVersion));
+    createIndex(BSON("key" << BSON("x" << 1) << "name"
+                           << "x"
+                           << "v" << kIndexVersion));
+    createIndex(BSON("key" << BSON("x"
+                                   << "hashed")
+                           << "name"
+                           << "xhashed"
+                           << "v" << kIndexVersion));
+
+    auto index = findShardKeyPrefixedIndex(opCtx(),
+                                           coll(),
+                                           coll()->getIndexCatalog(),
+                                           BSON("x"
+                                                << "hashed"),
+                                           true /* requireSingleKey */);
+
+    ASSERT_TRUE(index);
+    ASSERT_EQ("xhashed", index->descriptor()->indexName());
+}
+
+TEST_F(ShardKeyIndexUtilTest, PrefixKeyPattern) {
+    createIndex(BSON("key" << BSON("y" << 1) << "name"
+                           << "y"
+                           << "v" << kIndexVersion));
+    createIndex(BSON("key" << BSON("x" << 1 << "y" << 1 << "z" << 1) << "name"
+                           << "xyz"
+                           << "v" << kIndexVersion));
+
+    auto index = findShardKeyPrefixedIndex(opCtx(),
+                                           coll(),
+                                           coll()->getIndexCatalog(),
+                                           BSON("x" << 1 << "y" << 1),
+                                           true /* requireSingleKey */);
+
+    ASSERT_TRUE(index);
+    ASSERT_EQ("xyz", index->descriptor()->indexName());
+}
+
+TEST_F(ShardKeyIndexUtilTest, ExcludesIncompatibleIndexes) {
+    createIndex(BSON("key" << BSON("x" << 1) << "name"
+                           << "sparse"
+                           << "sparse" << true << "v" << kIndexVersion));
+    createIndex(BSON("key" << BSON("x" << 1) << "name"
+                           << "partial"
+                           << "partialFilterExpression" << BSON("x" << BSON("$exists" << true))
+                           << "v" << kIndexVersion));
+    createIndex(BSON("key" << BSON("x" << 1) << "name"
+                           << "collation"
+                           << "collation"
+                           << BSON("locale"
+                                   << "fr")
+                           << "v" << kIndexVersion));
+    createIndex(BSON("key" << BSON("x" << 1) << "name"
+                           << "x"
+                           << "v" << kIndexVersion));
+
+
+    auto index = findShardKeyPrefixedIndex(
+        opCtx(), coll(), coll()->getIndexCatalog(), BSON("x" << 1), true /* requireSingleKey */);
+
+    ASSERT_TRUE(index);
+    ASSERT_EQ("x", index->descriptor()->indexName());
+}
+
+TEST_F(ShardKeyIndexUtilTest, ExcludesMultiKeyIfRequiresSingleKey) {
+    createIndex(BSON("key" << BSON("x" << 1) << "name"
+                           << "x"
+                           << "v" << kIndexVersion));
+
+    DBDirectClient client(opCtx());
+    client.insert(nss().ns(), BSON("x" << BSON_ARRAY(1 << 2)));
+
+    auto index = findShardKeyPrefixedIndex(
+        opCtx(), coll(), coll()->getIndexCatalog(), BSON("x" << 1), true /* requireSingleKey */);
+
+    ASSERT_FALSE(index);
+}
+
+TEST_F(ShardKeyIndexUtilTest, IncludesMultiKeyIfSingleKeyNotRequired) {
+    createIndex(BSON("key" << BSON("x" << 1) << "name"
+                           << "x"
+                           << "v" << kIndexVersion));
+
+    DBDirectClient client(opCtx());
+    client.insert(nss().ns(), BSON("x" << BSON_ARRAY(1 << 2)));
+
+    auto index = findShardKeyPrefixedIndex(
+        opCtx(), coll(), coll()->getIndexCatalog(), BSON("x" << 1), false /* requireSingleKey */);
+
+    ASSERT_TRUE(index);
+    ASSERT_EQ("x", index->descriptor()->indexName());
+}
+
+TEST_F(ShardKeyIndexUtilTest, LastShardIndexWithSingleCandidate) {
+    createIndex(BSON("key" << BSON("y" << 1) << "name"
+                           << "y"
+                           << "v" << kIndexVersion));
+    createIndex(BSON("key" << BSON("x" << 1) << "name"
+                           << "x"
+                           << "v" << kIndexVersion));
+
+    ASSERT_TRUE(
+        isLastShardKeyIndex(opCtx(), coll(), coll()->getIndexCatalog(), "x", BSON("x" << 1)));
+}
+
+TEST_F(ShardKeyIndexUtilTest, LastShardIndexWithMultipleCandidates) {
+    createIndex(BSON("key" << BSON("y" << 1) << "name"
+                           << "y"
+                           << "v" << kIndexVersion));
+    createIndex(BSON("key" << BSON("x" << 1) << "name"
+                           << "x"
+                           << "v" << kIndexVersion));
+    createIndex(BSON("key" << BSON("x" << 1 << "y" << 1) << "name"
+                           << "xy"
+                           << "v" << kIndexVersion));
+
+    ASSERT_FALSE(
+        isLastShardKeyIndex(opCtx(), coll(), coll()->getIndexCatalog(), "x", BSON("x" << 1)));
+}
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/db/s/split_chunk.cpp b/src/mongo/db/s/split_chunk.cpp
index 651dc68f04a..cd3f7d15322 100644
--- a/src/mongo/db/s/split_chunk.cpp
+++ b/src/mongo/db/s/split_chunk.cpp
@@ -35,7 +35,6 @@
 
 #include "mongo/base/status_with.h"
 #include "mongo/bson/util/bson_extract.h"
-#include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/dbhelpers.h"
@@ -46,6 +45,7 @@
 #include "mongo/db/s/active_migrations_registry.h"
 #include "mongo/db/s/collection_sharding_runtime.h"
 #include "mongo/db/s/shard_filtering_metadata_refresh.h"
+#include "mongo/db/s/shard_key_index_util.h"
 #include "mongo/db/s/sharding_state.h"
 #include "mongo/logv2/log.h"
 #include "mongo/s/catalog/type_chunk.h"
@@ -60,7 +60,7 @@ const ReadPreferenceSetting kPrimaryOnlyReadPreference{ReadPreference::PrimaryOn
 
 bool checkIfSingleDoc(OperationContext* opCtx,
                       const CollectionPtr& collection,
-                      const IndexCatalog::ShardKeyIndex& idx,
+                      const ShardKeyIndex& idx,
                       const ChunkType* chunk) {
     KeyPattern kp(idx.keyPattern());
     BSONObj newmin = Helpers::toKeyFormat(kp.extendRangeBound(chunk->getMin(), false));
@@ -231,9 +231,11 @@ StatusWith<boost::optional<ChunkRange>> splitChunk(OperationContext* opCtx,
 
     // Allow multiKey based on the invariant that shard keys must be single-valued. Therefore,
     // any multi-key index prefixed by shard key cannot be multikey over the shard key fields.
-    auto catalog = collection->getIndexCatalog();
-    auto shardKeyIdx = catalog->findShardKeyPrefixedIndex(
-        opCtx, *collection, keyPatternObj, /*requireSingleKey=*/false);
+    auto shardKeyIdx = findShardKeyPrefixedIndex(opCtx,
+                                                 *collection,
+                                                 collection->getIndexCatalog(),
+                                                 keyPatternObj,
+                                                 /*requireSingleKey=*/false);
     if (!shardKeyIdx) {
         return boost::optional<ChunkRange>(boost::none);
     }
diff --git a/src/mongo/db/s/split_vector.cpp b/src/mongo/db/s/split_vector.cpp
index 78bc2aa9d8e..276b91951e7 100644
--- a/src/mongo/db/s/split_vector.cpp
+++ b/src/mongo/db/s/split_vector.cpp
@@ -35,7 +35,6 @@
 
 #include "mongo/base/status_with.h"
 #include "mongo/db/bson/dotted_path_support.h"
-#include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/dbhelpers.h"
 #include "mongo/db/exec/working_set_common.h"
@@ -45,6 +44,7 @@
 #include "mongo/db/query/internal_plans.h"
 #include "mongo/db/query/plan_executor.h"
 #include "mongo/db/s/collection_sharding_runtime.h"
+#include "mongo/db/s/shard_key_index_util.h"
 #include "mongo/logv2/log.h"
 
 namespace mongo {
@@ -85,9 +85,11 @@ std::vector<BSONObj> splitVector(OperationContext* opCtx,
 
         // Allow multiKey based on the invariant that shard keys must be single-valued. Therefore,
         // any multi-key index prefixed by shard key cannot be multikey over the shard key fields.
-        auto catalog = collection->getIndexCatalog();
-        auto shardKeyIdx = catalog->findShardKeyPrefixedIndex(
-            opCtx, *collection, keyPattern, /*requireSingleKey=*/false);
+        auto shardKeyIdx = findShardKeyPrefixedIndex(opCtx,
+                                                     *collection,
+                                                     collection->getIndexCatalog(),
+                                                     keyPattern,
+                                                     /*requireSingleKey=*/false);
         uassert(ErrorCodes::IndexNotFound,
                 str::stream() << "couldn't find index over splitting key "
                               << keyPattern.clientReadable().toString(),
diff --git a/src/mongo/dbtests/indexupdatetests.cpp b/src/mongo/dbtests/indexupdatetests.cpp
index 81de6756956..e44981aa502 100644
--- a/src/mongo/dbtests/indexupdatetests.cpp
+++ b/src/mongo/dbtests/indexupdatetests.cpp
@@ -250,7 +250,7 @@ public:
                 WriteUnitOfWork wunit(_opCtx);
                 // Drop all indexes including id index.
                 coll.getWritableCollection()->getIndexCatalog()->dropAllIndexes(
-                    _opCtx, coll.getWritableCollection(), true);
+                    _opCtx, coll.getWritableCollection(), true, {});
                 // Insert some documents.
                 int32_t nDocs = 1000;
                 OpDebug* const nullOpDebug = nullptr;
@@ -300,7 +300,7 @@ public:
             options.capped = true;
             options.cappedSize = 10 * 1024;
             Collection* coll = db->createCollection(_opCtx, _nss, options);
-            coll->getIndexCatalog()->dropAllIndexes(_opCtx, coll, true);
+            coll->getIndexCatalog()->dropAllIndexes(_opCtx, coll, true, {});
             // Insert some documents.
             int32_t nDocs = 1000;
             OpDebug* const nullOpDebug = nullptr;