SERVER-41819 Move methods from KVCollectionCatalogEntry to KVCatalog

73 files changed, 1354 insertions, 1249 deletions

diff --git a/src/mongo/db/catalog/capped_utils.cpp b/src/mongo/db/catalog/capped_utils.cpp
index 0f9562d653c..e16f19cc2d0 100644
--- a/src/mongo/db/catalog/capped_utils.cpp
+++ b/src/mongo/db/catalog/capped_utils.cpp
@@ -51,6 +51,7 @@
 #include "mongo/db/query/plan_yield_policy.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/views/view_catalog.h"
 #include "mongo/util/scopeguard.h"
 
@@ -146,7 +147,7 @@ void cloneCollectionAsCapped(OperationContext* opCtx,
 
     // create new collection
     {
-        auto options = fromCollection->getCatalogEntry()->getCollectionOptions(opCtx);
+        auto options = DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, fromNss);
         // The capped collection will get its own new unique id, as the conversion isn't reversible,
         // so it can't be rolled back.
         options.uuid.reset();
diff --git a/src/mongo/db/catalog/capped_utils_test.cpp b/src/mongo/db/catalog/capped_utils_test.cpp
index 444a4c73fdc..ce2afb79b1d 100644
--- a/src/mongo/db/catalog/capped_utils_test.cpp
+++ b/src/mongo/db/catalog/capped_utils_test.cpp
@@ -39,6 +39,7 @@
 #include "mongo/db/repl/replication_coordinator_mock.h"
 #include "mongo/db/repl/storage_interface_impl.h"
 #include "mongo/db/service_context_d_test_fixture.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/unittest/unittest.h"
 
 namespace {
@@ -98,8 +99,7 @@ CollectionOptions getCollectionOptions(OperationContext* opCtx, const NamespaceS
     auto collection = autoColl.getCollection();
     ASSERT_TRUE(collection) << "Unable to get collections options for " << nss
                             << " because collection does not exist.";
-    auto catalogEntry = collection->getCatalogEntry();
-    return catalogEntry->getCollectionOptions(opCtx);
+    return DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, nss);
 }
 
 // Size of capped collection to be passed to convertToCapped() which accepts a double.
diff --git a/src/mongo/db/catalog/catalog_control_test.cpp b/src/mongo/db/catalog/catalog_control_test.cpp
index 961e1a83997..59afeb15a25 100644
--- a/src/mongo/db/catalog/catalog_control_test.cpp
+++ b/src/mongo/db/catalog/catalog_control_test.cpp
@@ -108,10 +108,10 @@ public:
     const KVEngine* getEngine() const {
         return nullptr;
     }
-    KVCatalog* getCatalog() {
+    DurableCatalog* getCatalog() {
         return nullptr;
     }
-    const KVCatalog* getCatalog() const {
+    const DurableCatalog* getCatalog() const {
         return nullptr;
     }
 };
diff --git a/src/mongo/db/catalog/coll_mod.cpp b/src/mongo/db/catalog/coll_mod.cpp
index 5f081288ea9..f801ee42d29 100644
--- a/src/mongo/db/catalog/coll_mod.cpp
+++ b/src/mongo/db/catalog/coll_mod.cpp
@@ -52,6 +52,7 @@
 #include "mongo/db/s/sharding_state.h"
 #include "mongo/db/server_options.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/recovery_unit.h"
 #include "mongo/db/views/view_catalog.h"
 #include "mongo/util/fail_point_service.h"
@@ -329,7 +330,8 @@ Status _collModInternal(OperationContext* opCtx,
     // provide to the OpObserver. TTL index updates aren't a part of collection options so we
     // save the relevant TTL index data in a separate object.
 
-    CollectionOptions oldCollOptions = coll->getCatalogEntry()->getCollectionOptions(opCtx);
+    CollectionOptions oldCollOptions = DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, nss);
+
     boost::optional<TTLCollModInfo> ttlInfo;
 
     // Handle collMod operation type appropriately.
@@ -343,8 +345,8 @@ Status _collModInternal(OperationContext* opCtx,
             result->appendAs(oldExpireSecs, "expireAfterSeconds_old");
 
             // Change the value of "expireAfterSeconds" on disk.
-            coll->getCatalogEntry()->updateTTLSetting(
-                opCtx, cmr.idx->indexName(), newExpireSecs.safeNumberLong());
+            DurableCatalog::get(opCtx)->updateTTLSetting(
+                opCtx, coll->ns(), cmr.idx->indexName(), newExpireSecs.safeNumberLong());
 
             // Notify the index catalog that the definition of this index changed.
             cmr.idx = coll->getIndexCatalog()->refreshEntry(opCtx, cmr.idx);
@@ -377,7 +379,7 @@ Status _collModInternal(OperationContext* opCtx,
         // upgrade collMod.
         invariant(cmdObj.nFields() == 1);
         std::vector<std::string> indexNames;
-        coll->getCatalogEntry()->getAllUniqueIndexes(opCtx, &indexNames);
+        DurableCatalog::get(opCtx)->getAllUniqueIndexes(opCtx, nss, &indexNames);
 
         for (size_t i = 0; i < indexNames.size(); i++) {
             const IndexDescriptor* desc =
@@ -385,7 +387,7 @@ Status _collModInternal(OperationContext* opCtx,
             invariant(desc);
 
             // Update index metadata in storage engine.
-            coll->getCatalogEntry()->updateIndexMetadata(opCtx, desc);
+            DurableCatalog::get(opCtx)->updateIndexMetadata(opCtx, nss, desc);
 
             // Refresh the in-memory instance of the index.
             desc = coll->getIndexCatalog()->refreshEntry(opCtx, desc);
diff --git a/src/mongo/db/catalog/collection.h b/src/mongo/db/catalog/collection.h
index d3b71a184fb..dd2af83ba7d 100644
--- a/src/mongo/db/catalog/collection.h
+++ b/src/mongo/db/catalog/collection.h
@@ -178,7 +178,9 @@ public:
          * only constructs an in-memory representation of what already exists on disk.
          */
         virtual std::unique_ptr<Collection> make(
-            OperationContext* opCtx, CollectionCatalogEntry* collectionCatalogEntry) const = 0;
+            OperationContext* opCtx,
+            CollectionUUID uuid,
+            CollectionCatalogEntry* collectionCatalogEntry) const = 0;
     };
 
     /**
diff --git a/src/mongo/db/catalog/collection_catalog_entry.h b/src/mongo/db/catalog/collection_catalog_entry.h
index 5802d50d96c..cbf53669e47 100644
--- a/src/mongo/db/catalog/collection_catalog_entry.h
+++ b/src/mongo/db/catalog/collection_catalog_entry.h
@@ -81,153 +81,6 @@ public:
 
     // ------- indexes ----------
 
-    virtual CollectionOptions getCollectionOptions(OperationContext* opCtx) const = 0;
-
-    virtual int getTotalIndexCount(OperationContext* opCtx) const = 0;
-
-    virtual int getCompletedIndexCount(OperationContext* opCtx) const = 0;
-
-    virtual int getMaxAllowedIndexes() const = 0;
-
-    virtual void getAllIndexes(OperationContext* opCtx, std::vector<std::string>* names) const = 0;
-
-    virtual void getReadyIndexes(OperationContext* opCtx,
-                                 std::vector<std::string>* names) const = 0;
-
-    virtual void getAllUniqueIndexes(OperationContext* opCtx,
-                                     std::vector<std::string>* names) const {}
-
-    virtual BSONObj getIndexSpec(OperationContext* opCtx, StringData idxName) const = 0;
-
-    /**
-     * Returns true if the index identified by 'indexName' is multikey, and returns false otherwise.
-     *
-     * If the 'multikeyPaths' pointer is non-null, then it must point to an empty vector. If this
-     * index supports tracking path-level multikey information, then this function sets
-     * 'multikeyPaths' as the path components that cause this index to be multikey.
-     *
-     * In particular, if this function returns false and the index supports tracking path-level
-     * multikey information, then 'multikeyPaths' is initialized as a vector with size equal to the
-     * number of elements in the index key pattern of empty sets.
-     */
-    virtual bool isIndexMultikey(OperationContext* opCtx,
-                                 StringData indexName,
-                                 MultikeyPaths* multikeyPaths) const = 0;
-
-    /**
-     * Sets the index identified by 'indexName' to be multikey.
-     *
-     * If 'multikeyPaths' is non-empty, then it must be a vector with size equal to the number of
-     * elements in the index key pattern. Additionally, at least one path component of the indexed
-     * fields must cause this index to be multikey.
-     *
-     * This function returns true if the index metadata has changed, and returns false otherwise.
-     */
-    virtual bool setIndexIsMultikey(OperationContext* opCtx,
-                                    StringData indexName,
-                                    const MultikeyPaths& multikeyPaths) = 0;
-
-    virtual bool isIndexReady(OperationContext* opCtx, StringData indexName) const = 0;
-
-    virtual bool isIndexPresent(OperationContext* opCtx, StringData indexName) const = 0;
-
-    virtual KVPrefix getIndexPrefix(OperationContext* opCtx, StringData indexName) const = 0;
-
-    virtual Status removeIndex(OperationContext* opCtx, StringData indexName) = 0;
-
-    virtual Status prepareForIndexBuild(OperationContext* opCtx,
-                                        const IndexDescriptor* spec,
-                                        IndexBuildProtocol indexBuildProtocol,
-                                        bool isBackgroundSecondaryBuild) = 0;
-
-    /**
-     * Returns whether or not the index is being built with the two-phase index build procedure.
-     */
-    virtual bool isTwoPhaseIndexBuild(OperationContext* opCtx, StringData indexName) const = 0;
-
-    /**
-     * Returns the server-compatibility version of the index build procedure.
-     */
-    virtual long getIndexBuildVersion(OperationContext* opCtx, StringData indexName) const = 0;
-
-    /**
-     * Indicate that a build index is now in the "scanning" phase of a hybrid index build. The
-     * 'constraintViolationsIdent' is only used for unique indexes.
-     *
-     * It is only valid to call this when the index is using the kTwoPhase IndexBuildProtocol.
-     */
-    virtual void setIndexBuildScanning(OperationContext* opCtx,
-                                       StringData indexName,
-                                       std::string sideWritesIdent,
-                                       boost::optional<std::string> constraintViolationsIdent) = 0;
-
-    /**
-     * Returns whether or not this index is building in the "scanning" phase.
-     */
-    virtual bool isIndexBuildScanning(OperationContext* opCtx, StringData indexName) const = 0;
-
-    /**
-     * Indicate that a build index is now in the "draining" phase of a hybrid index build.
-     *
-     * It is only valid to call this when the index is using the kTwoPhase IndexBuildProtocol.
-     */
-    virtual void setIndexBuildDraining(OperationContext* opCtx, StringData indexName) = 0;
-
-    /**
-     * Returns whether or not this index is building in the "draining" phase.
-     */
-    virtual bool isIndexBuildDraining(OperationContext* opCtx, StringData indexName) const = 0;
-
-    /**
-     * Indicate that an index build is completed and the index is ready to use.
-     */
-    virtual void indexBuildSuccess(OperationContext* opCtx, StringData indexName) = 0;
-
-    virtual boost::optional<std::string> getSideWritesIdent(OperationContext* opCtx,
-                                                            StringData indexName) const = 0;
-
-    virtual boost::optional<std::string> getConstraintViolationsIdent(
-        OperationContext* opCtx, StringData indexName) const = 0;
-
-    /* Updates the expireAfterSeconds field of the given index to the value in newExpireSecs.
-     * The specified index must already contain an expireAfterSeconds field, and the value in
-     * that field and newExpireSecs must both be numeric.
-     */
-    virtual void updateTTLSetting(OperationContext* opCtx,
-                                  StringData idxName,
-                                  long long newExpireSeconds) = 0;
-
-    virtual void updateIndexMetadata(OperationContext* opCtx, const IndexDescriptor* desc) {}
-
-    /**
-     * Updates the validator for this collection.
-     *
-     * An empty validator removes all validation.
-     */
-    virtual void updateValidator(OperationContext* opCtx,
-                                 const BSONObj& validator,
-                                 StringData validationLevel,
-                                 StringData validationAction) = 0;
-
-    /**
-     * Updates the 'temp' setting for this collection.
-     */
-    virtual void setIsTemp(OperationContext* opCtx, bool isTemp) = 0;
-
-    /**
-     * Compare the UUID argument to the UUID obtained from the metadata. Return true if they
-     * are equal, false otherwise. uuid can become a CollectionUUID once MMAPv1 is removed.
-     */
-    virtual bool isEqualToMetadataUUID(OperationContext* opCtx, OptionalCollectionUUID uuid) = 0;
-
-    /**
-     * Updates size of a capped Collection.
-     */
-    virtual void updateCappedSize(OperationContext* opCtx, long long size) = 0;
-
-    // TODO SERVER-36385 Remove this function: we don't set the feature tracker bit in 4.4 because
-    // 4.4 can only downgrade to 4.2 which can read long TypeBits.
-    virtual void setIndexKeyStringWithLongTypeBitsExistsOnDisk(OperationContext* opCtx) = 0;
 
     virtual RecordStore* getRecordStore() = 0;
     virtual const RecordStore* getRecordStore() const = 0;
diff --git a/src/mongo/db/catalog/collection_catalog_entry_mock.h b/src/mongo/db/catalog/collection_catalog_entry_mock.h
index c9096059e5f..d3951a43f57 100644
--- a/src/mongo/db/catalog/collection_catalog_entry_mock.h
+++ b/src/mongo/db/catalog/collection_catalog_entry_mock.h
@@ -40,125 +40,6 @@ public:
         return CollectionOptions();
     }
 
-    int getTotalIndexCount(OperationContext* opCtx) const {
-        return 0;
-    }
-
-    int getCompletedIndexCount(OperationContext* opCtx) const {
-        return 0;
-    }
-
-    int getMaxAllowedIndexes() const {
-        return 0;
-    }
-
-    void getAllIndexes(OperationContext* opCtx, std::vector<std::string>* names) const {}
-
-    void getReadyIndexes(OperationContext* opCtx, std::vector<std::string>* names) const {}
-
-    void getAllUniqueIndexes(OperationContext* opCtx, std::vector<std::string>* names) const {}
-
-    BSONObj getIndexSpec(OperationContext* opCtx, StringData idxName) const {
-        return BSONObj();
-    }
-
-    bool isIndexMultikey(OperationContext* opCtx,
-                         StringData indexName,
-                         MultikeyPaths* multikeyPaths) const {
-        return false;
-    }
-
-    bool setIndexIsMultikey(OperationContext* opCtx,
-                            StringData indexName,
-                            const MultikeyPaths& multikeyPaths) {
-        return false;
-    }
-
-    RecordId getIndexHead(OperationContext* opCtx, StringData indexName) const {
-        return RecordId(0);
-    }
-
-    void setIndexHead(OperationContext* opCtx, StringData indexName, const RecordId& newHead) {}
-
-    bool isIndexReady(OperationContext* opCtx, StringData indexName) const {
-        return false;
-    }
-
-    bool isIndexPresent(OperationContext* opCtx, StringData indexName) const {
-        return false;
-    }
-
-    KVPrefix getIndexPrefix(OperationContext* opCtx, StringData indexName) const {
-        MONGO_UNREACHABLE;
-    }
-
-    Status removeIndex(OperationContext* opCtx, StringData indexName) {
-        return Status::OK();
-    }
-
-    Status prepareForIndexBuild(OperationContext* opCtx,
-                                const IndexDescriptor* spec,
-                                IndexBuildProtocol indexBuildProtocol,
-                                bool isBackgroundSecondaryBuild) {
-        return Status::OK();
-    }
-
-    bool isTwoPhaseIndexBuild(OperationContext* opCtx, StringData indexName) const {
-        return false;
-    }
-
-    long getIndexBuildVersion(OperationContext* opCtx, StringData indexName) const {
-        return 0;
-    }
-
-    void setIndexBuildScanning(OperationContext* opCtx,
-                               StringData indexName,
-                               std::string sideWritesIdent,
-                               boost::optional<std::string> constraintViolationsIdent) {}
-
-    bool isIndexBuildScanning(OperationContext* opCtx, StringData indexName) const {
-        return false;
-    }
-
-    void setIndexBuildDraining(OperationContext* opCtx, StringData indexName) {}
-
-    bool isIndexBuildDraining(OperationContext* opCtx, StringData indexName) const {
-        return false;
-    }
-
-    void indexBuildSuccess(OperationContext* opCtx, StringData indexName) {}
-
-    boost::optional<std::string> getSideWritesIdent(OperationContext* opCtx,
-                                                    StringData indexName) const {
-        return boost::none;
-    }
-
-    boost::optional<std::string> getConstraintViolationsIdent(OperationContext* opCtx,
-                                                              StringData indexName) const {
-        return boost::none;
-    }
-
-    void updateTTLSetting(OperationContext* opCtx, StringData idxName, long long newExpireSeconds) {
-    }
-
-    void updateIndexMetadata(OperationContext* opCtx, const IndexDescriptor* desc) {}
-
-    void updateFlags(OperationContext* opCtx, int newValue) {}
-
-    void updateValidator(OperationContext* opCtx,
-                         const BSONObj& validator,
-                         StringData validationLevel,
-                         StringData validationAction) {}
-
-    void setIsTemp(OperationContext* opCtx, bool isTemp) {}
-
-    bool isEqualToMetadataUUID(OperationContext* opCtx, OptionalCollectionUUID uuid) {
-        return false;
-    }
-
-    void updateCappedSize(OperationContext* opCtx, long long size) {}
-
-    void setIndexKeyStringWithLongTypeBitsExistsOnDisk(OperationContext* opCtx) {}
 
     RecordStore* getRecordStore() {
         return nullptr;
diff --git a/src/mongo/db/catalog/collection_catalog_test.cpp b/src/mongo/db/catalog/collection_catalog_test.cpp
index b4629639497..1c335aa4d71 100644
--- a/src/mongo/db/catalog/collection_catalog_test.cpp
+++ b/src/mongo/db/catalog/collection_catalog_test.cpp
@@ -37,6 +37,7 @@
 #include "mongo/db/catalog/collection_mock.h"
 #include "mongo/db/concurrency/lock_manager_defs.h"
 #include "mongo/db/operation_context_noop.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/unittest/death_test.h"
 #include "mongo/unittest/unittest.h"
 
@@ -705,7 +706,9 @@ TEST_F(ForEachCollectionFromDbTest, ForEachCollectionFromDbWithPredicate) {
             [&](const Collection* collection, const CollectionCatalogEntry* catalogEntry) {
                 ASSERT_TRUE(
                     opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_NONE));
-                return catalogEntry->getCollectionOptions(opCtx).temp;
+                return DurableCatalog::get(opCtx)
+                    ->getCollectionOptions(opCtx, collection->ns())
+                    .temp;
             });
 
         ASSERT_EQUALS(numCollectionsTraversed, 2);
@@ -727,7 +730,9 @@ TEST_F(ForEachCollectionFromDbTest, ForEachCollectionFromDbWithPredicate) {
             [&](const Collection* collection, const CollectionCatalogEntry* catalogEntry) {
                 ASSERT_TRUE(
                     opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_NONE));
-                return !catalogEntry->getCollectionOptions(opCtx).temp;
+                return !DurableCatalog::get(opCtx)
+                            ->getCollectionOptions(opCtx, collection->ns())
+                            .temp;
             });
 
         ASSERT_EQUALS(numCollectionsTraversed, 1);
diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp
index 4356a9f2dfc..99d7394478d 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -67,6 +67,7 @@
 #include "mongo/db/repl/oplog.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/key_string.h"
 #include "mongo/db/storage/record_store.h"
 #include "mongo/db/update/update_driver.h"
@@ -203,8 +204,7 @@ CollectionImpl::CollectionImpl(OperationContext* opCtx,
       _needCappedLock(supportsDocLocking() && _recordStore && _recordStore->isCapped() &&
                       _ns.db() != "local"),
       _infoCache(std::make_unique<CollectionInfoCacheImpl>(this, _ns)),
-      _indexCatalog(
-          std::make_unique<IndexCatalogImpl>(this, getCatalogEntry()->getMaxAllowedIndexes())),
+      _indexCatalog(std::make_unique<IndexCatalogImpl>(this)),
       _cappedNotifier(_recordStore && _recordStore->isCapped()
                           ? std::make_unique<CappedInsertNotifier>()
                           : nullptr) {
@@ -227,22 +227,22 @@ CollectionImpl::~CollectionImpl() {
 }
 
 std::unique_ptr<Collection> CollectionImpl::FactoryImpl::make(
-    OperationContext* opCtx, CollectionCatalogEntry* collectionCatalogEntry) const {
+    OperationContext* opCtx,
+    CollectionUUID uuid,
+    CollectionCatalogEntry* collectionCatalogEntry) const {
     auto rs = collectionCatalogEntry->getRecordStore();
-    const auto uuid = collectionCatalogEntry->getCollectionOptions(opCtx).uuid;
     const auto nss = collectionCatalogEntry->ns();
     return std::make_unique<CollectionImpl>(opCtx, nss.ns(), uuid, collectionCatalogEntry, rs);
 }
 
 void CollectionImpl::init(OperationContext* opCtx) {
-    _collator = parseCollation(opCtx, _ns, _details->getCollectionOptions(opCtx).collation);
-    _validatorDoc = _details->getCollectionOptions(opCtx).validator.getOwned();
+    auto collectionOptions = DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, _ns);
+    _collator = parseCollation(opCtx, _ns, collectionOptions.collation);
+    _validatorDoc = collectionOptions.validator.getOwned();
     _validator = uassertStatusOK(
         parseValidator(opCtx, _validatorDoc, MatchExpressionParser::kAllowAllSpecialFeatures));
-    _validationAction = uassertStatusOK(
-        _parseValidationAction(_details->getCollectionOptions(opCtx).validationAction));
-    _validationLevel = uassertStatusOK(
-        _parseValidationLevel(_details->getCollectionOptions(opCtx).validationLevel));
+    _validationAction = uassertStatusOK(_parseValidationAction(collectionOptions.validationAction));
+    _validationLevel = uassertStatusOK(_parseValidationLevel(collectionOptions.validationLevel));
 
     getIndexCatalog()->init(opCtx).transitional_ignore();
     infoCache()->init(opCtx);
@@ -740,7 +740,7 @@ StatusWith<RecordData> CollectionImpl::updateDocumentWithDamages(
 }
 
 bool CollectionImpl::isTemporary(OperationContext* opCtx) const {
-    return _details->getCollectionOptions(opCtx).temp;
+    return DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, _ns).temp;
 }
 
 bool CollectionImpl::isCapped() const {
@@ -851,7 +851,8 @@ Status CollectionImpl::setValidator(OperationContext* opCtx, BSONObj validatorDo
     if (!statusWithMatcher.isOK())
         return statusWithMatcher.getStatus();
 
-    _details->updateValidator(opCtx, validatorDoc, getValidationLevel(), getValidationAction());
+    DurableCatalog::get(opCtx)->updateValidator(
+        opCtx, ns(), validatorDoc, getValidationLevel(), getValidationAction());
 
     opCtx->recoveryUnit()->onRollback([
         this,
@@ -899,7 +900,8 @@ Status CollectionImpl::setValidationLevel(OperationContext* opCtx, StringData ne
     auto oldValidationLevel = _validationLevel;
     _validationLevel = levelSW.getValue();
 
-    _details->updateValidator(opCtx, _validatorDoc, getValidationLevel(), getValidationAction());
+    DurableCatalog::get(opCtx)->updateValidator(
+        opCtx, ns(), _validatorDoc, getValidationLevel(), getValidationAction());
     opCtx->recoveryUnit()->onRollback(
         [this, oldValidationLevel]() { this->_validationLevel = oldValidationLevel; });
 
@@ -917,7 +919,9 @@ Status CollectionImpl::setValidationAction(OperationContext* opCtx, StringData n
     auto oldValidationAction = _validationAction;
     _validationAction = actionSW.getValue();
 
-    _details->updateValidator(opCtx, _validatorDoc, getValidationLevel(), getValidationAction());
+
+    DurableCatalog::get(opCtx)->updateValidator(
+        opCtx, ns(), _validatorDoc, getValidationLevel(), getValidationAction());
     opCtx->recoveryUnit()->onRollback(
         [this, oldValidationAction]() { this->_validationAction = oldValidationAction; });
 
@@ -943,7 +947,7 @@ Status CollectionImpl::updateValidator(OperationContext* opCtx,
         this->_validationAction = oldValidationAction;
     });
 
-    _details->updateValidator(opCtx, newValidator, newLevel, newAction);
+    DurableCatalog::get(opCtx)->updateValidator(opCtx, ns(), newValidator, newLevel, newAction);
     _validatorDoc = std::move(newValidator);
 
     auto validatorSW =
@@ -1280,7 +1284,7 @@ void _validateCatalogEntry(OperationContext* opCtx,
                            CollectionImpl* coll,
                            BSONObj validatorDoc,
                            ValidateResults* results) {
-    CollectionOptions options = coll->getCatalogEntry()->getCollectionOptions(opCtx);
+    CollectionOptions options = DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, coll->ns());
     addErrorIfUnequal(options.uuid, coll->uuid(), "UUID", results);
     const CollatorInterface* collation = coll->getDefaultCollator();
     addErrorIfUnequal(options.collation.isEmpty(), !collation, "simple collation", results);
@@ -1437,7 +1441,7 @@ void CollectionImpl::setNs(NamespaceString nss) {
 }
 
 void CollectionImpl::indexBuildSuccess(OperationContext* opCtx, IndexCatalogEntry* index) {
-    _details->indexBuildSuccess(opCtx, index->descriptor()->indexName());
+    DurableCatalog::get(opCtx)->indexBuildSuccess(opCtx, ns(), index->descriptor()->indexName());
     _indexCatalog->indexBuildSuccess(opCtx, index);
 }
 
diff --git a/src/mongo/db/catalog/collection_impl.h b/src/mongo/db/catalog/collection_impl.h
index 25c85a660fb..b91028a847f 100644
--- a/src/mongo/db/catalog/collection_impl.h
+++ b/src/mongo/db/catalog/collection_impl.h
@@ -57,7 +57,9 @@ public:
     class FactoryImpl : public Factory {
     public:
         std::unique_ptr<Collection> make(
-            OperationContext* opCtx, CollectionCatalogEntry* collectionCatalogEntry) const final;
+            OperationContext* opCtx,
+            CollectionUUID uuid,
+            CollectionCatalogEntry* collectionCatalogEntry) const final;
     };
 
     bool ok() const final {
diff --git a/src/mongo/db/catalog/create_collection_test.cpp b/src/mongo/db/catalog/create_collection_test.cpp
index cad2aa82562..c9ebc485480 100644
--- a/src/mongo/db/catalog/create_collection_test.cpp
+++ b/src/mongo/db/catalog/create_collection_test.cpp
@@ -41,6 +41,7 @@
 #include "mongo/db/repl/replication_coordinator_mock.h"
 #include "mongo/db/repl/storage_interface_impl.h"
 #include "mongo/db/service_context_d_test_fixture.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/uuid.h"
 
@@ -101,8 +102,7 @@ CollectionOptions getCollectionOptions(OperationContext* opCtx, const NamespaceS
     auto collection = autoColl.getCollection();
     ASSERT_TRUE(collection) << "Unable to get collections options for " << nss
                             << " because collection does not exist.";
-    auto catalogEntry = collection->getCatalogEntry();
-    return catalogEntry->getCollectionOptions(opCtx);
+    return DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, nss);
 }
 
 /**
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index 3e981742a6e..f4eb17d7b13 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -65,7 +65,7 @@
 #include "mongo/db/server_options.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/stats/top.h"
-#include "mongo/db/storage/kv/kv_catalog.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/recovery_unit.h"
 #include "mongo/db/storage/storage_engine.h"
 #include "mongo/db/storage/storage_engine_init.h"
@@ -191,7 +191,7 @@ void DatabaseImpl::clearTmpCollections(OperationContext* opCtx) const {
 
     CollectionCatalog::CollectionInfoFn predicate =
         [&](const Collection* collection, const CollectionCatalogEntry* catalogEntry) {
-            return catalogEntry->getCollectionOptions(opCtx).temp;
+            return DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, collection->ns()).temp;
         };
 
     catalog::forEachCollectionFromDb(opCtx, name(), MODE_X, callback, predicate);
@@ -464,7 +464,7 @@ void DatabaseImpl::_dropCollectionIndexes(OperationContext* opCtx,
     LOG(1) << "dropCollection: " << nss << " - dropAllIndexes start";
     collection->getIndexCatalog()->dropAllIndexes(opCtx, true);
 
-    invariant(collection->getCatalogEntry()->getTotalIndexCount(opCtx) == 0);
+    invariant(DurableCatalog::get(opCtx)->getTotalIndexCount(opCtx, nss) == 0);
     LOG(1) << "dropCollection: " << nss << " - dropAllIndexes done";
 }
 
@@ -474,8 +474,7 @@ Status DatabaseImpl::_finishDropCollection(OperationContext* opCtx,
     UUID uuid = *collection->uuid();
     log() << "Finishing collection drop for " << nss << " (" << uuid << ").";
 
-    auto storageEngine = opCtx->getServiceContext()->getStorageEngine();
-    auto status = storageEngine->getCatalog()->dropCollection(opCtx, nss);
+    auto status = DurableCatalog::get(opCtx)->dropCollection(opCtx, nss);
     if (!status.isOK())
         return status;
 
@@ -532,8 +531,7 @@ Status DatabaseImpl::renameCollection(OperationContext* opCtx,
 
     Top::get(opCtx->getServiceContext()).collectionDropped(fromNss);
 
-    auto storageEngine = opCtx->getServiceContext()->getStorageEngine();
-    Status status = storageEngine->getCatalog()->renameCollection(opCtx, fromNss, toNss, stayTemp);
+    Status status = DurableCatalog::get(opCtx)->renameCollection(opCtx, fromNss, toNss, stayTemp);
 
     // Set the namespace of 'collToRename' from within the CollectionCatalog. This is necessary
     // because
@@ -661,16 +659,15 @@ Collection* DatabaseImpl::createCollection(OperationContext* opCtx,
           << " UUID: " << optionsWithUUID.uuid.get() << " and options: " << options.toBSON();
 
     // Create CollectionCatalogEntry
-    auto storageEngine = opCtx->getServiceContext()->getStorageEngine();
-    auto statusWithCatalogEntry = storageEngine->getCatalog()->createCollection(
+    auto statusWithCatalogEntry = DurableCatalog::get(opCtx)->createCollection(
         opCtx, nss, optionsWithUUID, true /*allocateDefaultSpace*/);
     massertStatusOK(statusWithCatalogEntry.getStatus());
     std::unique_ptr<CollectionCatalogEntry> ownedCatalogEntry =
         std::move(statusWithCatalogEntry.getValue());
 
     // Create Collection object
-    std::unique_ptr<Collection> ownedCollection =
-        Collection::Factory::get(opCtx)->make(opCtx, ownedCatalogEntry.get());
+    std::unique_ptr<Collection> ownedCollection = Collection::Factory::get(opCtx)->make(
+        opCtx, optionsWithUUID.uuid.get(), ownedCatalogEntry.get());
     auto collection = ownedCollection.get();
     ownedCollection->init(opCtx);
 
diff --git a/src/mongo/db/catalog/database_test.cpp b/src/mongo/db/catalog/database_test.cpp
index e1e6b8ecca1..981edc0aca4 100644
--- a/src/mongo/db/catalog/database_test.cpp
+++ b/src/mongo/db/catalog/database_test.cpp
@@ -57,6 +57,7 @@
 #include "mongo/db/repl/storage_interface_mock.h"
 #include "mongo/db/s/op_observer_sharding_impl.h"
 #include "mongo/db/service_context_d_test_fixture.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/scopeguard.h"
 
@@ -369,8 +370,8 @@ TEST_F(DatabaseTest, RenameCollectionPreservesUuidOfSourceCollectionAndUpdatesUu
         auto toCollection = db->getCollection(opCtx, toNss);
         ASSERT_TRUE(toCollection);
 
-        auto catalogEntry = toCollection->getCatalogEntry();
-        auto toCollectionOptions = catalogEntry->getCollectionOptions(opCtx);
+        auto toCollectionOptions =
+            DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, toCollection->ns());
 
         auto toUuid = toCollectionOptions.uuid;
         ASSERT_TRUE(toUuid);
diff --git a/src/mongo/db/catalog/index_build_block.cpp b/src/mongo/db/catalog/index_build_block.cpp
index 98ce154ca7e..d66dd8fbb5f 100644
--- a/src/mongo/db/catalog/index_build_block.cpp
+++ b/src/mongo/db/catalog/index_build_block.cpp
@@ -42,6 +42,7 @@
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/logical_clock.h"
 #include "mongo/db/operation_context.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/log.h"
 
@@ -81,8 +82,8 @@ Status IndexCatalogImpl::IndexBuildBlock::init(OperationContext* opCtx, Collecti
 
     // Setup on-disk structures.
     const auto protocol = IndexBuildProtocol::kSinglePhase;
-    Status status = collection->getCatalogEntry()->prepareForIndexBuild(
-        opCtx, descriptor.get(), protocol, isBackgroundSecondaryBuild);
+    Status status = DurableCatalog::get(opCtx)->prepareForIndexBuild(
+        opCtx, _nss, descriptor.get(), protocol, isBackgroundSecondaryBuild);
     if (!status.isOK())
         return status;
 
@@ -103,8 +104,8 @@ Status IndexCatalogImpl::IndexBuildBlock::init(OperationContext* opCtx, Collecti
                       _indexBuildInterceptor->getConstraintViolationsTableIdent())
                 : boost::none;
 
-            collection->getCatalogEntry()->setIndexBuildScanning(
-                opCtx, _entry->descriptor()->indexName(), sideWritesIdent, constraintsIdent);
+            DurableCatalog::get(opCtx)->setIndexBuildScanning(
+                opCtx, _nss, _entry->descriptor()->indexName(), sideWritesIdent, constraintsIdent);
         }
     }
 
diff --git a/src/mongo/db/catalog/index_catalog_entry.h b/src/mongo/db/catalog/index_catalog_entry.h
index 46b35196513..de034469453 100644
--- a/src/mongo/db/catalog/index_catalog_entry.h
+++ b/src/mongo/db/catalog/index_catalog_entry.h
@@ -126,12 +126,6 @@ public:
      */
     virtual void setMultikey(OperationContext* const opCtx, const MultikeyPaths& multikeyPaths) = 0;
 
-    /**
-     * TODO SERVER-36385 Remove this function: we don't set the feature tracker bit in 4.4
-     * because 4.4 can only downgrade to 4.2 which can read long TypeBits.
-     */
-    virtual void setIndexKeyStringWithLongTypeBitsExistsOnDisk(OperationContext* const opCtx) = 0;
-
     // if this ready is ready for queries
     virtual bool isReady(OperationContext* const opCtx) const = 0;
 
diff --git a/src/mongo/db/catalog/index_catalog_entry_impl.cpp b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
index 7599297b5ca..d3d7c79a5a7 100644
--- a/src/mongo/db/catalog/index_catalog_entry_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
@@ -49,6 +49,8 @@
 #include "mongo/db/operation_context.h"
 #include "mongo/db/query/collation/collator_factory_interface.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/storage/durable_catalog.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/transaction_participant.h"
 #include "mongo/util/log.h"
 #include "mongo/util/scopeguard.h"
@@ -68,7 +70,8 @@ IndexCatalogEntryImpl::IndexCatalogEntryImpl(OperationContext* const opCtx,
       _infoCache(infoCache),
       _ordering(Ordering::make(_descriptor->keyPattern())),
       _isReady(false),
-      _prefix(collection->getIndexPrefix(opCtx, _descriptor->indexName())) {
+      _prefix(DurableCatalog::get(opCtx)->getIndexPrefix(
+          opCtx, collection->ns(), _descriptor->indexName())) {
     _descriptor->_cachedEntry = this;
 
     _isReady = _catalogIsReady(opCtx);
@@ -264,8 +267,8 @@ void IndexCatalogEntryImpl::setMultikey(OperationContext* opCtx,
     // CollectionCatalogEntry::setIndexIsMultikey() requires that we discard the path-level
     // multikey information in order to avoid unintentionally setting path-level multikey
     // information on an index created before 3.4.
-    const bool indexMetadataHasChanged =
-        _collection->setIndexIsMultikey(opCtx, _descriptor->indexName(), paths);
+    const bool indexMetadataHasChanged = DurableCatalog::get(opCtx)->setIndexIsMultikey(
+        opCtx, _collection->ns(), _descriptor->indexName(), paths);
 
     // When the recovery unit commits, update the multikey paths if needed and clear the plan cache
     // if the index metadata has changed.
@@ -295,10 +298,6 @@ void IndexCatalogEntryImpl::setMultikey(OperationContext* opCtx,
     }
 }
 
-void IndexCatalogEntryImpl::setIndexKeyStringWithLongTypeBitsExistsOnDisk(OperationContext* opCtx) {
-    _collection->setIndexKeyStringWithLongTypeBitsExistsOnDisk(opCtx);
-}
-
 void IndexCatalogEntryImpl::setNs(NamespaceString ns) {
     _ns = ns;
     _descriptor->setNs(std::move(ns));
@@ -307,20 +306,24 @@ void IndexCatalogEntryImpl::setNs(NamespaceString ns) {
 // ----
 
 bool IndexCatalogEntryImpl::_catalogIsReady(OperationContext* opCtx) const {
-    return _collection->isIndexReady(opCtx, _descriptor->indexName());
+    return DurableCatalog::get(opCtx)->isIndexReady(
+        opCtx, _collection->ns(), _descriptor->indexName());
 }
 
 bool IndexCatalogEntryImpl::_catalogIsPresent(OperationContext* opCtx) const {
-    return _collection->isIndexPresent(opCtx, _descriptor->indexName());
+    return DurableCatalog::get(opCtx)->isIndexPresent(
+        opCtx, _collection->ns(), _descriptor->indexName());
 }
 
 bool IndexCatalogEntryImpl::_catalogIsMultikey(OperationContext* opCtx,
                                                MultikeyPaths* multikeyPaths) const {
-    return _collection->isIndexMultikey(opCtx, _descriptor->indexName(), multikeyPaths);
+    return DurableCatalog::get(opCtx)->isIndexMultikey(
+        opCtx, _collection->ns(), _descriptor->indexName(), multikeyPaths);
 }
 
 KVPrefix IndexCatalogEntryImpl::_catalogGetPrefix(OperationContext* opCtx) const {
-    return _collection->getIndexPrefix(opCtx, _descriptor->indexName());
+    return DurableCatalog::get(opCtx)->getIndexPrefix(
+        opCtx, _collection->ns(), _descriptor->indexName());
 }
 
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/index_catalog_entry_impl.h b/src/mongo/db/catalog/index_catalog_entry_impl.h
index 541f573cae3..f7da6b660a5 100644
--- a/src/mongo/db/catalog/index_catalog_entry_impl.h
+++ b/src/mongo/db/catalog/index_catalog_entry_impl.h
@@ -156,10 +156,6 @@ public:
      */
     void setMultikey(OperationContext* opCtx, const MultikeyPaths& multikeyPaths) final;
 
-    // TODO SERVER-36385 Remove this function: we don't set the feature tracker bit in 4.4 because
-    // 4.4 can only downgrade to 4.2 which can read long TypeBits.
-    void setIndexKeyStringWithLongTypeBitsExistsOnDisk(OperationContext* opCtx) final;
-
     // if this ready is ready for queries
     bool isReady(OperationContext* opCtx) const final;
 
@@ -224,7 +220,7 @@ private:
     bool _indexTracksPathLevelMultikeyInfo = false;
 
     // Set to true if this index is multikey. '_isMultikey' serves as a cache of the information
-    // stored in the NamespaceDetails or KVCatalog.
+    // stored in the NamespaceDetails or DurableCatalog.
     AtomicWord<bool> _isMultikey;
 
     // Controls concurrent access to '_indexMultikeyPaths'. We acquire this mutex rather than the
diff --git a/src/mongo/db/catalog/index_catalog_impl.cpp b/src/mongo/db/catalog/index_catalog_impl.cpp
index 0c3fc6e56c9..8239708c505 100644
--- a/src/mongo/db/catalog/index_catalog_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_impl.cpp
@@ -65,7 +65,7 @@
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/server_options.h"
 #include "mongo/db/service_context.h"
-#include "mongo/db/storage/kv/kv_catalog.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/kv/kv_engine.h"
 #include "mongo/db/storage/storage_engine_init.h"
 #include "mongo/util/assert_util.h"
@@ -89,10 +89,8 @@ const BSONObj IndexCatalogImpl::_idObj = BSON("_id" << 1);
 
 // -------------
 
-IndexCatalogImpl::IndexCatalogImpl(Collection* collection, int maxNumIndexesAllowed)
-    : _magic(INDEX_CATALOG_UNINIT),
-      _collection(collection),
-      _maxNumIndexesAllowed(maxNumIndexesAllowed) {}
+IndexCatalogImpl::IndexCatalogImpl(Collection* collection)
+    : _magic(INDEX_CATALOG_UNINIT), _collection(collection) {}
 
 IndexCatalogImpl::~IndexCatalogImpl() {
     if (_magic != INDEX_CATALOG_UNINIT) {
@@ -104,13 +102,13 @@ IndexCatalogImpl::~IndexCatalogImpl() {
 
 Status IndexCatalogImpl::init(OperationContext* opCtx) {
     vector<string> indexNames;
-    _collection->getCatalogEntry()->getAllIndexes(opCtx, &indexNames);
+    auto durableCatalog = DurableCatalog::get(opCtx);
+    durableCatalog->getAllIndexes(opCtx, _collection->ns(), &indexNames);
 
     for (size_t i = 0; i < indexNames.size(); i++) {
         const string& indexName = indexNames[i];
-        BSONObj spec = _collection->getCatalogEntry()->getIndexSpec(opCtx, indexName).getOwned();
-
-        invariant(_collection->getCatalogEntry()->isIndexReady(opCtx, indexName));
+        BSONObj spec = durableCatalog->getIndexSpec(opCtx, _collection->ns(), indexName).getOwned();
+        invariant(durableCatalog->isIndexReady(opCtx, _collection->ns(), indexName));
 
         BSONObj keyPattern = spec.getObjectField("key");
         auto descriptor =
@@ -148,10 +146,10 @@ IndexCatalogEntry* IndexCatalogImpl::_setupInMemoryStructures(
 
     IndexDescriptor* desc = entry->descriptor();
 
-    auto engine = opCtx->getServiceContext()->getStorageEngine();
     std::string ident =
-        engine->getCatalog()->getIndexIdent(opCtx, _collection->ns(), desc->indexName());
+        DurableCatalog::get(opCtx)->getIndexIdent(opCtx, _collection->ns(), desc->indexName());
 
+    auto engine = opCtx->getServiceContext()->getStorageEngine();
     std::unique_ptr<SortedDataInterface> sdi =
         engine->getEngine()->getGroupedSortedDataInterface(opCtx, ident, desc, entry->getPrefix());
 
@@ -311,25 +309,26 @@ void IndexCatalogImpl::_logInternalState(OperationContext* opCtx,
     std::vector<std::string> allIndexes;
     std::vector<std::string> readyIndexes;
 
-    _collection->getCatalogEntry()->getAllIndexes(opCtx, &allIndexes);
-    _collection->getCatalogEntry()->getReadyIndexes(opCtx, &readyIndexes);
+    auto durableCatalog = DurableCatalog::get(opCtx);
+    durableCatalog->getAllIndexes(opCtx, _collection->ns(), &allIndexes);
+    durableCatalog->getReadyIndexes(opCtx, _collection->ns(), &readyIndexes);
 
     error() << "All indexes:";
     for (const auto& index : allIndexes) {
         error() << "Index '" << index << "' with specification: "
-                << redact(_collection->getCatalogEntry()->getIndexSpec(opCtx, index));
+                << redact(durableCatalog->getIndexSpec(opCtx, _collection->ns(), index));
     }
 
     error() << "Ready indexes:";
     for (const auto& index : readyIndexes) {
         error() << "Index '" << index << "' with specification: "
-                << redact(_collection->getCatalogEntry()->getIndexSpec(opCtx, index));
+                << redact(durableCatalog->getIndexSpec(opCtx, _collection->ns(), index));
     }
 
     error() << "Index names to drop:";
     for (const auto& indexNameToDrop : indexNamesToDrop) {
         error() << "Index '" << indexNameToDrop << "' with specification: "
-                << redact(_collection->getCatalogEntry()->getIndexSpec(opCtx, indexNameToDrop));
+                << redact(durableCatalog->getIndexSpec(opCtx, _collection->ns(), indexNameToDrop));
     }
 }
 
@@ -449,12 +448,16 @@ StatusWith<BSONObj> IndexCatalogImpl::createIndexOnEmptyCollection(OperationCont
     indexBuildBlock.success(opCtx, _collection);
 
     // sanity check
-    invariant(_collection->getCatalogEntry()->isIndexReady(opCtx, descriptor->indexName()));
+    invariant(DurableCatalog::get(opCtx)->isIndexReady(
+        opCtx, _collection->ns(), descriptor->indexName()));
 
     return spec;
 }
 
 namespace {
+
+constexpr int kMaxNumIndexesAllowed = 64;
+
 // While technically recursive, only current possible with 2 levels.
 Status _checkValidFilterExpressions(MatchExpression* expression, int level = 0) {
     if (!expression)
@@ -805,7 +808,7 @@ Status IndexCatalogImpl::_doesSpecConflictWithExisting(OperationContext* opCtx,
         }
     }
 
-    if (numIndexesTotal(opCtx) >= _maxNumIndexesAllowed) {
+    if (numIndexesTotal(opCtx) >= kMaxNumIndexesAllowed) {
         string s = str::stream() << "add index fails, too many indexes for " << _collection->ns()
                                  << " key:" << key;
         log() << s;
@@ -898,7 +901,7 @@ void IndexCatalogImpl::dropAllIndexes(OperationContext* opCtx,
     // verify state is sane post cleaning
 
     long long numIndexesInCollectionCatalogEntry =
-        _collection->getCatalogEntry()->getTotalIndexCount(opCtx);
+        DurableCatalog::get(opCtx)->getTotalIndexCount(opCtx, _collection->ns());
 
     if (haveIdIndex) {
         fassert(17324, numIndexesTotal(opCtx) == 1);
@@ -1012,7 +1015,7 @@ Status IndexCatalogImpl::_dropIndex(OperationContext* opCtx, IndexCatalogEntry*
 void IndexCatalogImpl::_deleteIndexFromDisk(OperationContext* opCtx,
                                             const string& indexName,
                                             const string& indexNamespace) {
-    Status status = _collection->getCatalogEntry()->removeIndex(opCtx, indexName);
+    Status status = DurableCatalog::get(opCtx)->removeIndex(opCtx, _collection->ns(), indexName);
     if (status.code() == ErrorCodes::NamespaceNotFound) {
         // this is ok, as we may be partially through index creation
     } else if (!status.isOK()) {
@@ -1057,19 +1060,20 @@ bool IndexCatalogImpl::haveAnyIndexesInProgress() const {
 
 int IndexCatalogImpl::numIndexesTotal(OperationContext* opCtx) const {
     int count = _readyIndexes.size() + _buildingIndexes.size();
-    dassert(_collection->getCatalogEntry()->getTotalIndexCount(opCtx) == count);
+    dassert(DurableCatalog::get(opCtx)->getTotalIndexCount(opCtx, _collection->ns()) == count);
     return count;
 }
 
 int IndexCatalogImpl::numIndexesReady(OperationContext* opCtx) const {
     std::vector<const IndexDescriptor*> itIndexes;
     std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, /*includeUnfinished*/ false);
+    auto durableCatalog = DurableCatalog::get(opCtx);
     while (ii->more()) {
         itIndexes.push_back(ii->next()->descriptor());
     }
     DEV {
         std::vector<std::string> completedIndexes;
-        _collection->getCatalogEntry()->getReadyIndexes(opCtx, &completedIndexes);
+        durableCatalog->getReadyIndexes(opCtx, _collection->ns(), &completedIndexes);
 
         // There is a potential inconistency where the index information in the collection catalog
         // entry and the index catalog differ. Log as much information as possible here.
@@ -1222,7 +1226,8 @@ const IndexDescriptor* IndexCatalogImpl::refreshEntry(OperationContext* opCtx,
     invariant(_buildingIndexes.size() == 0);
 
     const std::string indexName = oldDesc->indexName();
-    invariant(_collection->getCatalogEntry()->isIndexReady(opCtx, indexName));
+    auto durableCatalog = DurableCatalog::get(opCtx);
+    invariant(durableCatalog->isIndexReady(opCtx, _collection->ns(), indexName));
 
     // Delete the IndexCatalogEntry that owns this descriptor.  After deletion, 'oldDesc' is
     // invalid and should not be dereferenced. Also, invalidate the index from the
@@ -1234,7 +1239,7 @@ const IndexDescriptor* IndexCatalogImpl::refreshEntry(OperationContext* opCtx,
     _collection->infoCache()->droppedIndex(opCtx, indexName);
 
     // Ask the CollectionCatalogEntry for the new index spec.
-    BSONObj spec = _collection->getCatalogEntry()->getIndexSpec(opCtx, indexName).getOwned();
+    BSONObj spec = durableCatalog->getIndexSpec(opCtx, _collection->ns(), indexName).getOwned();
     BSONObj keyPattern = spec.getObjectField("key");
 
     // Re-register this index in the index catalog with the new spec. Also, add the new index
diff --git a/src/mongo/db/catalog/index_catalog_impl.h b/src/mongo/db/catalog/index_catalog_impl.h
index 8c31d34721e..b95110b6dd1 100644
--- a/src/mongo/db/catalog/index_catalog_impl.h
+++ b/src/mongo/db/catalog/index_catalog_impl.h
@@ -56,7 +56,7 @@ struct InsertDeleteOptions;
  */
 class IndexCatalogImpl : public IndexCatalog {
 public:
-    explicit IndexCatalogImpl(Collection* collection, int maxNumIndexesAllowed);
+    explicit IndexCatalogImpl(Collection* collection);
     ~IndexCatalogImpl() override;
 
     // must be called before used
@@ -483,7 +483,6 @@ private:
 
     int _magic;
     Collection* const _collection;
-    const int _maxNumIndexesAllowed;
 
     IndexCatalogEntryContainer _readyIndexes;
     IndexCatalogEntryContainer _buildingIndexes;
diff --git a/src/mongo/db/catalog/index_consistency.cpp b/src/mongo/db/catalog/index_consistency.cpp
index 1ff0f14d65a..0c61691fa7d 100644
--- a/src/mongo/db/catalog/index_consistency.cpp
+++ b/src/mongo/db/catalog/index_consistency.cpp
@@ -40,6 +40,7 @@
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/index_names.h"
 #include "mongo/db/server_options.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/key_string.h"
 #include "mongo/db/storage/record_store.h"
 #include "mongo/db/storage/sorted_data_interface.h"
@@ -87,7 +88,7 @@ IndexConsistency::IndexConsistency(OperationContext* opCtx,
 
         indexInfo.indexName = indexName;
         indexInfo.keyPattern = descriptor->keyPattern();
-        indexInfo.isReady = _collection->getCatalogEntry()->isIndexReady(opCtx, indexName);
+        indexInfo.isReady = DurableCatalog::get(opCtx)->isIndexReady(opCtx, nss, indexName);
 
         uint32_t indexNameHash;
         MurmurHash3_x86_32(indexName.c_str(), indexName.size(), 0, &indexNameHash);
diff --git a/src/mongo/db/catalog/rename_collection.cpp b/src/mongo/db/catalog/rename_collection.cpp
index 7f1de22bab6..be9779d409a 100644
--- a/src/mongo/db/catalog/rename_collection.cpp
+++ b/src/mongo/db/catalog/rename_collection.cpp
@@ -56,6 +56,7 @@
 #include "mongo/db/s/database_sharding_state.h"
 #include "mongo/db/server_options.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/views/view_catalog.h"
 #include "mongo/util/fail_point_service.h"
 #include "mongo/util/log.h"
@@ -561,7 +562,8 @@ Status renameBetweenDBs(OperationContext* opCtx,
 
     Collection* tmpColl = nullptr;
     {
-        auto collectionOptions = sourceColl->getCatalogEntry()->getCollectionOptions(opCtx);
+        auto collectionOptions =
+            DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, sourceColl->ns());
 
         // Renaming across databases will result in a new UUID.
         collectionOptions.uuid = UUID::gen();
diff --git a/src/mongo/db/catalog/rename_collection_test.cpp b/src/mongo/db/catalog/rename_collection_test.cpp
index cc90a4a83a2..539c80f08f6 100644
--- a/src/mongo/db/catalog/rename_collection_test.cpp
+++ b/src/mongo/db/catalog/rename_collection_test.cpp
@@ -57,6 +57,7 @@
 #include "mongo/db/repl/replication_coordinator_mock.h"
 #include "mongo/db/repl/storage_interface_mock.h"
 #include "mongo/db/service_context_d_test_fixture.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/unittest/death_test.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/assert_util.h"
@@ -370,8 +371,7 @@ CollectionOptions _getCollectionOptions(OperationContext* opCtx, const Namespace
     auto collection = autoColl.getCollection();
     ASSERT_TRUE(collection) << "Unable to get collections options for " << nss
                             << " because collection does not exist.";
-    auto catalogEntry = collection->getCatalogEntry();
-    return catalogEntry->getCollectionOptions(opCtx);
+    return DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, nss);
 }
 
 /**
@@ -399,8 +399,7 @@ bool _isTempCollection(OperationContext* opCtx, const NamespaceString& nss) {
     auto collection = autoColl.getCollection();
     ASSERT_TRUE(collection) << "Unable to check if " << nss
                             << " is a temporary collection because collection does not exist.";
-    auto catalogEntry = collection->getCatalogEntry();
-    auto options = catalogEntry->getCollectionOptions(opCtx);
+    auto options = _getCollectionOptions(opCtx, nss);
     return options.temp;
 }
 
diff --git a/src/mongo/db/cloner.cpp b/src/mongo/db/cloner.cpp
index f9cc6d2830a..cf2c445a66b 100644
--- a/src/mongo/db/cloner.cpp
+++ b/src/mongo/db/cloner.cpp
@@ -60,6 +60,7 @@
 #include "mongo/db/repl/isself.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/storage_options.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/fail_point_service.h"
@@ -611,8 +612,9 @@ Status Cloner::createCollectionsForDb(
                     // exists on the target, we check if the existing collection's options and
                     // UUID match those of the one we're trying to create. If they do, we treat
                     // the create as a no-op; if they don't match, we return an error.
-                    auto existingOpts =
-                        collection->getCatalogEntry()->getCollectionOptions(opCtx).toBSON();
+                    auto existingOpts = DurableCatalog::get(opCtx)
+                                            ->getCollectionOptions(opCtx, collection->ns())
+                                            .toBSON();
                     UnorderedFieldsBSONObjComparator bsonCmp;
 
                     optionsBuilder.append(params.collectionInfo["info"]["uuid"]);
diff --git a/src/mongo/db/commands/drop_indexes.cpp b/src/mongo/db/commands/drop_indexes.cpp
index b6ca4a72a6a..4e93324f7d4 100644
--- a/src/mongo/db/commands/drop_indexes.cpp
+++ b/src/mongo/db/commands/drop_indexes.cpp
@@ -52,6 +52,7 @@
 #include "mongo/db/logical_clock.h"
 #include "mongo/db/op_observer.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/views/view_catalog.h"
 #include "mongo/util/log.h"
 #include "mongo/util/quick_exit.h"
@@ -157,12 +158,13 @@ public:
         vector<BSONObj> all;
         {
             vector<string> indexNames;
-            collection->getCatalogEntry()->getAllIndexes(opCtx, &indexNames);
+            DurableCatalog::get(opCtx)->getAllIndexes(opCtx, collection->ns(), &indexNames);
             all.reserve(indexNames.size());
 
             for (size_t i = 0; i < indexNames.size(); i++) {
                 const string& name = indexNames[i];
-                BSONObj spec = collection->getCatalogEntry()->getIndexSpec(opCtx, name);
+                BSONObj spec =
+                    DurableCatalog::get(opCtx)->getIndexSpec(opCtx, collection->ns(), name);
 
                 {
                     BSONObjBuilder bob;
diff --git a/src/mongo/db/commands/list_collections.cpp b/src/mongo/db/commands/list_collections.cpp
index 29cd4b02a29..cb0a65f607f 100644
--- a/src/mongo/db/commands/list_collections.cpp
+++ b/src/mongo/db/commands/list_collections.cpp
@@ -57,6 +57,7 @@
 #include "mongo/db/query/cursor_response.h"
 #include "mongo/db/query/find_common.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/storage_engine.h"
 #include "mongo/db/storage/storage_options.h"
 #include "mongo/db/views/view_catalog.h"
@@ -186,7 +187,7 @@ BSONObj buildCollectionBson(OperationContext* opCtx,
         return b.obj();
     }
 
-    CollectionOptions options = collection->getCatalogEntry()->getCollectionOptions(opCtx);
+    CollectionOptions options = DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, nss);
 
     // While the UUID is stored as a collection option, from the user's perspective it is an
     // unsettable read-only property, so put it in the 'info' section.
diff --git a/src/mongo/db/commands/list_indexes.cpp b/src/mongo/db/commands/list_indexes.cpp
index 59721e91ce4..bc9d164b65f 100644
--- a/src/mongo/db/commands/list_indexes.cpp
+++ b/src/mongo/db/commands/list_indexes.cpp
@@ -48,6 +48,7 @@
 #include "mongo/db/query/cursor_response.h"
 #include "mongo/db/query/find_common.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/storage_engine.h"
 #include "mongo/util/uuid.h"
 
@@ -154,8 +155,7 @@ public:
                     str::stream() << "ns does not exist: " << ctx.getNss().ns(),
                     collection);
 
-            const CollectionCatalogEntry* cce = collection->getCatalogEntry();
-            invariant(cce);
+            auto durableCatalog = DurableCatalog::get(opCtx);
 
             nss = ctx.getNss();
 
@@ -165,7 +165,7 @@ public:
             vector<string> indexNames;
             writeConflictRetry(opCtx, "listIndexes", nss.ns(), [&] {
                 indexNames.clear();
-                cce->getAllIndexes(opCtx, &indexNames);
+                durableCatalog->getAllIndexes(opCtx, nss, &indexNames);
             });
 
             auto ws = std::make_unique<WorkingSet>();
@@ -173,16 +173,18 @@ public:
 
             for (size_t i = 0; i < indexNames.size(); i++) {
                 auto indexSpec = writeConflictRetry(opCtx, "listIndexes", nss.ns(), [&] {
-                    if (includeBuildUUIDs && !cce->isIndexReady(opCtx, indexNames[i])) {
+                    if (includeBuildUUIDs &&
+                        !durableCatalog->isIndexReady(opCtx, nss, indexNames[i])) {
                         BSONObjBuilder builder;
-                        builder.append("spec"_sd, cce->getIndexSpec(opCtx, indexNames[i]));
+                        builder.append("spec"_sd,
+                                       durableCatalog->getIndexSpec(opCtx, nss, indexNames[i]));
 
                         // TODO(SERVER-37980): Replace with index build UUID.
                         auto indexBuildUUID = UUID::gen();
                         indexBuildUUID.appendToBuilder(&builder, "buildUUID"_sd);
                         return builder.obj();
                     }
-                    return cce->getIndexSpec(opCtx, indexNames[i]);
+                    return durableCatalog->getIndexSpec(opCtx, nss, indexNames[i]);
                 });
 
                 WorkingSetID id = ws->allocate();
diff --git a/src/mongo/db/commands/mr.cpp b/src/mongo/db/commands/mr.cpp
index d4a8b7da1b3..19276846563 100644
--- a/src/mongo/db/commands/mr.cpp
+++ b/src/mongo/db/commands/mr.cpp
@@ -65,6 +65,7 @@
 #include "mongo/db/s/sharding_state.h"
 #include "mongo/db/server_options.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/s/catalog_cache.h"
 #include "mongo/s/client/parallel.h"
 #include "mongo/s/client/shard_connection.h"
@@ -550,7 +551,8 @@ void State::prepTempCollection() {
 
         auto const finalColl = autoGetFinalColl.getCollection();
         if (finalColl) {
-            finalOptions = finalColl->getCatalogEntry()->getCollectionOptions(_opCtx);
+            finalOptions =
+                DurableCatalog::get(_opCtx)->getCollectionOptions(_opCtx, finalColl->ns());
 
             std::unique_ptr<IndexCatalog::IndexIterator> ii =
                 finalColl->getIndexCatalog()->getIndexIterator(_opCtx, true);
diff --git a/src/mongo/db/commands/resize_oplog.cpp b/src/mongo/db/commands/resize_oplog.cpp
index cdae013bf44..45331d2609e 100644
--- a/src/mongo/db/commands/resize_oplog.cpp
+++ b/src/mongo/db/commands/resize_oplog.cpp
@@ -41,6 +41,7 @@
 #include "mongo/db/db_raii.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/operation_context.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/util/log.h"
 #include "mongo/util/scopeguard.h"
 
@@ -115,8 +116,7 @@ public:
         WriteUnitOfWork wunit(opCtx);
         Status status = coll->getRecordStore()->updateCappedSize(opCtx, size);
         uassertStatusOK(status);
-        CollectionCatalogEntry* entry = coll->getCatalogEntry();
-        entry->updateCappedSize(opCtx, size);
+        DurableCatalog::get(opCtx)->updateCappedSize(opCtx, coll->ns(), size);
         wunit.commit();
         LOG(0) << "replSetResizeOplog success, currentSize:" << size;
         return true;
diff --git a/src/mongo/db/commands/validate.cpp b/src/mongo/db/commands/validate.cpp
index d30c820e747..4a57e81f477 100644
--- a/src/mongo/db/commands/validate.cpp
+++ b/src/mongo/db/commands/validate.cpp
@@ -37,6 +37,7 @@
 #include "mongo/db/commands.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/query/internal_plans.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/record_store.h"
 #include "mongo/db/views/view_catalog.h"
 #include "mongo/util/fail_point_service.h"
@@ -169,8 +170,8 @@ public:
             return CommandHelpers::appendCommandStatusNoThrow(result, status);
         }
 
-        CollectionCatalogEntry* catalogEntry = collection->getCatalogEntry();
-        CollectionOptions opts = catalogEntry->getCollectionOptions(opCtx);
+        CollectionOptions opts =
+            DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, collection->ns());
 
         // All collections must have a UUID.
         if (!opts.uuid) {
diff --git a/src/mongo/db/index/index_access_method.cpp b/src/mongo/db/index/index_access_method.cpp
index fc415013018..e57b22b758d 100644
--- a/src/mongo/db/index/index_access_method.cpp
+++ b/src/mongo/db/index/index_access_method.cpp
@@ -50,6 +50,7 @@
 #include "mongo/db/operation_context.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/repl/timestamp_block.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/storage_options.h"
 #include "mongo/util/log.h"
 #include "mongo/util/progress_meter.h"
@@ -243,7 +244,8 @@ Status AbstractIndexAccessMethod::insertKeys(OperationContext* opCtx,
                 }
 
                 if (status.isOK() && ret.getValue() == SpecialFormatInserted::LongTypeBitsInserted)
-                    _btreeState->setIndexKeyStringWithLongTypeBitsExistsOnDisk(opCtx);
+                    DurableCatalog::get(opCtx)->setIndexKeyStringWithLongTypeBitsExistsOnDisk(
+                        opCtx);
             }
             if (isFatalError(opCtx, status, key)) {
                 return status;
@@ -487,7 +489,8 @@ Status AbstractIndexAccessMethod::update(OperationContext* opCtx,
                     _newInterface->insert(opCtx, key, recordId, ticket.dupsAllowed);
                 status = ret.getStatus();
                 if (status.isOK() && ret.getValue() == SpecialFormatInserted::LongTypeBitsInserted)
-                    _btreeState->setIndexKeyStringWithLongTypeBitsExistsOnDisk(opCtx);
+                    DurableCatalog::get(opCtx)->setIndexKeyStringWithLongTypeBitsExistsOnDisk(
+                        opCtx);
             }
             if (isFatalError(opCtx, status, key)) {
                 return status;
@@ -688,7 +691,7 @@ Status AbstractIndexAccessMethod::commitBulk(OperationContext* opCtx,
             StatusWith<SpecialFormatInserted> ret = builder->addKey(data.first, data.second);
             status = ret.getStatus();
             if (status.isOK() && ret.getValue() == SpecialFormatInserted::LongTypeBitsInserted)
-                _btreeState->setIndexKeyStringWithLongTypeBitsExistsOnDisk(opCtx);
+                DurableCatalog::get(opCtx)->setIndexKeyStringWithLongTypeBitsExistsOnDisk(opCtx);
         }
 
         if (!status.isOK()) {
@@ -726,7 +729,7 @@ Status AbstractIndexAccessMethod::commitBulk(OperationContext* opCtx,
     // tracker bit so that downgrade binary which cannot read the long TypeBits fails to
     // start up.
     if (specialFormatInserted == SpecialFormatInserted::LongTypeBitsInserted)
-        _btreeState->setIndexKeyStringWithLongTypeBitsExistsOnDisk(opCtx);
+        DurableCatalog::get(opCtx)->setIndexKeyStringWithLongTypeBitsExistsOnDisk(opCtx);
     wunit.commit();
     return Status::OK();
 }
diff --git a/src/mongo/db/index_builds_coordinator.cpp b/src/mongo/db/index_builds_coordinator.cpp
index c9bb9bca276..7d23602ff20 100644
--- a/src/mongo/db/index_builds_coordinator.cpp
+++ b/src/mongo/db/index_builds_coordinator.cpp
@@ -49,6 +49,7 @@
 #include "mongo/db/s/collection_sharding_state.h"
 #include "mongo/db/s/database_sharding_state.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/s/shard_key_pattern.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/log.h"
@@ -193,7 +194,7 @@ StatusWith<std::pair<long long, long long>> IndexBuildsCoordinator::startIndexRe
                 if (!descriptor) {
                     // If it's unfinished index, drop it directly via removeIndex.
                     Status status =
-                        collection->getCatalogEntry()->removeIndex(opCtx, indexNames[i]);
+                        DurableCatalog::get(opCtx)->removeIndex(opCtx, nss, indexNames[i]);
                     continue;
                 }
                 Status s = indexCatalog->dropIndex(opCtx, descriptor);
diff --git a/src/mongo/db/op_observer_impl.cpp b/src/mongo/db/op_observer_impl.cpp
index 597030d47c4..8f9152d0ee0 100644
--- a/src/mongo/db/op_observer_impl.cpp
+++ b/src/mongo/db/op_observer_impl.cpp
@@ -54,6 +54,7 @@
 #include "mongo/db/s/collection_sharding_state.h"
 #include "mongo/db/server_options.h"
 #include "mongo/db/session_catalog_mongod.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/transaction_participant.h"
 #include "mongo/db/transaction_participant_gen.h"
 #include "mongo/db/views/durable_view_catalog.h"
@@ -764,8 +765,7 @@ void OpObserverImpl::onCollMod(OperationContext* opCtx,
 
     invariant(coll->uuid());
     invariant(coll->uuid() == uuid);
-    CollectionCatalogEntry* entry = coll->getCatalogEntry();
-    invariant(entry->isEqualToMetadataUUID(opCtx, uuid));
+    invariant(DurableCatalog::get(opCtx)->isEqualToMetadataUUID(opCtx, nss, uuid));
 }
 
 void OpObserverImpl::onDropDatabase(OperationContext* opCtx, const std::string& dbName) {
diff --git a/src/mongo/db/repair_database.cpp b/src/mongo/db/repair_database.cpp
index 61c90a722e5..95cdae1bf1a 100644
--- a/src/mongo/db/repair_database.cpp
+++ b/src/mongo/db/repair_database.cpp
@@ -53,6 +53,7 @@
 #include "mongo/db/index_builds_coordinator.h"
 #include "mongo/db/logical_clock.h"
 #include "mongo/db/query/query_knobs_gen.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/storage_engine.h"
 #include "mongo/util/log.h"
 #include "mongo/util/scopeguard.h"
@@ -65,9 +66,10 @@ StatusWith<IndexNameObjs> getIndexNameObjs(OperationContext* opCtx,
     IndexNameObjs ret;
     std::vector<std::string>& indexNames = ret.first;
     std::vector<BSONObj>& indexSpecs = ret.second;
+    auto durableCatalog = DurableCatalog::get(opCtx);
     {
         // Fetch all indexes
-        cce->getAllIndexes(opCtx, &indexNames);
+        durableCatalog->getAllIndexes(opCtx, cce->ns(), &indexNames);
         auto newEnd =
             std::remove_if(indexNames.begin(),
                            indexNames.end(),
@@ -76,8 +78,9 @@ StatusWith<IndexNameObjs> getIndexNameObjs(OperationContext* opCtx,
 
         indexSpecs.reserve(indexNames.size());
 
+
         for (const auto& name : indexNames) {
-            BSONObj spec = cce->getIndexSpec(opCtx, name);
+            BSONObj spec = durableCatalog->getIndexSpec(opCtx, cce->ns(), name);
             using IndexVersion = IndexDescriptor::IndexVersion;
             IndexVersion indexVersion = IndexVersion::kV1;
             if (auto indexVersionElem = spec[IndexDescriptor::kIndexVersionFieldName]) {
diff --git a/src/mongo/db/repair_database_and_check_version.cpp b/src/mongo/db/repair_database_and_check_version.cpp
index 2c56a5877dd..1bc6616cbfa 100644
--- a/src/mongo/db/repair_database_and_check_version.cpp
+++ b/src/mongo/db/repair_database_and_check_version.cpp
@@ -53,6 +53,7 @@
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/repl_set_member_in_standalone_mode.h"
 #include "mongo/db/server_options.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/storage_repair_observer.h"
 #include "mongo/util/exit.h"
 #include "mongo/util/fail_point.h"
@@ -138,9 +139,10 @@ Status restoreMissingFeatureCompatibilityVersionDocument(OperationContext* opCtx
  * the _id field
  */
 bool checkIdIndexExists(OperationContext* opCtx, const CollectionCatalogEntry* catalogEntry) {
-    auto indexCount = catalogEntry->getTotalIndexCount(opCtx);
+    auto durableCatalog = DurableCatalog::get(opCtx);
+    auto indexCount = durableCatalog->getTotalIndexCount(opCtx, catalogEntry->ns());
     auto indexNames = std::vector<std::string>(indexCount);
-    catalogEntry->getAllIndexes(opCtx, &indexNames);
+    durableCatalog->getAllIndexes(opCtx, catalogEntry->ns(), &indexNames);
 
     for (auto name : indexNames) {
         if (name == "_id_") {
@@ -215,7 +217,7 @@ Status ensureCollectionProperties(OperationContext* opCtx,
             // All user-created replicated collections created since MongoDB 4.0 have _id indexes.
             auto requiresIndex = coll->requiresIdIndex() && coll->ns().isReplicated();
             auto catalogEntry = coll->getCatalogEntry();
-            auto collOptions = catalogEntry->getCollectionOptions(opCtx);
+            auto collOptions = DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, coll->ns());
             auto hasAutoIndexIdField = collOptions.autoIndexId == CollectionOptions::YES;
 
             // Even if the autoIndexId field is not YES, the collection may still have an _id index
diff --git a/src/mongo/db/repl/dbcheck.cpp b/src/mongo/db/repl/dbcheck.cpp
index 714bbc4e335..31071a1af71 100644
--- a/src/mongo/db/repl/dbcheck.cpp
+++ b/src/mongo/db/repl/dbcheck.cpp
@@ -45,6 +45,7 @@
 #include "mongo/db/repl/dbcheck_gen.h"
 #include "mongo/db/repl/oplog.h"
 #include "mongo/db/repl/optime.h"
+#include "mongo/db/storage/durable_catalog.h"
 
 namespace mongo {
 
@@ -374,14 +375,12 @@ std::vector<BSONObj> collectionIndexInfo(OperationContext* opCtx, Collection* co
     std::vector<std::string> names;
 
     // List the indices,
-    const auto* cce = collection->getCatalogEntry();
-    invariant(cce);
-
-    cce->getAllIndexes(opCtx, &names);
+    auto durableCatalog = DurableCatalog::get(opCtx);
+    durableCatalog->getAllIndexes(opCtx, collection->ns(), &names);
 
     // and get the info for each one.
     for (const auto& name : names) {
-        result.push_back(cce->getIndexSpec(opCtx, name));
+        result.push_back(durableCatalog->getIndexSpec(opCtx, collection->ns(), name));
     }
 
     auto comp = std::make_unique<SimpleBSONObjComparator>();
@@ -392,7 +391,7 @@ std::vector<BSONObj> collectionIndexInfo(OperationContext* opCtx, Collection* co
 }
 
 BSONObj collectionOptions(OperationContext* opCtx, Collection* collection) {
-    return collection->getCatalogEntry()->getCollectionOptions(opCtx).toBSON();
+    return DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, collection->ns()).toBSON();
 }
 
 AutoGetDbForDbCheck::AutoGetDbForDbCheck(OperationContext* opCtx, const NamespaceString& nss)
diff --git a/src/mongo/db/repl/idempotency_test_fixture.cpp b/src/mongo/db/repl/idempotency_test_fixture.cpp
index ac0a2510718..cf367c53a30 100644
--- a/src/mongo/db/repl/idempotency_test_fixture.cpp
+++ b/src/mongo/db/repl/idempotency_test_fixture.cpp
@@ -57,6 +57,7 @@
 #include "mongo/db/repl/replication_consistency_markers_mock.h"
 #include "mongo/db/repl/replication_coordinator_mock.h"
 #include "mongo/db/repl/storage_interface.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/util/md5.hpp"
 
 namespace mongo {
@@ -620,13 +621,13 @@ CollectionState IdempotencyTest::validate(const NamespaceString& nss) {
 
     std::string dataHash = computeDataHash(collection);
 
-    auto collectionCatalog = collection->getCatalogEntry();
-    auto collectionOptions = collectionCatalog->getCollectionOptions(_opCtx.get());
+    auto durableCatalog = DurableCatalog::get(_opCtx.get());
+    auto collectionOptions = durableCatalog->getCollectionOptions(_opCtx.get(), collection->ns());
     std::vector<std::string> allIndexes;
     BSONObjSet indexSpecs = SimpleBSONObjComparator::kInstance.makeBSONObjSet();
-    collectionCatalog->getAllIndexes(_opCtx.get(), &allIndexes);
+    durableCatalog->getAllIndexes(_opCtx.get(), collection->ns(), &allIndexes);
     for (auto const& index : allIndexes) {
-        indexSpecs.insert(collectionCatalog->getIndexSpec(_opCtx.get(), index));
+        indexSpecs.insert(durableCatalog->getIndexSpec(_opCtx.get(), collection->ns(), index));
     }
     ASSERT_EQUALS(indexSpecs.size(), allIndexes.size());
 
diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp
index 5d3a741ee1b..613ae52586f 100644
--- a/src/mongo/db/repl/oplog.cpp
+++ b/src/mongo/db/repl/oplog.cpp
@@ -83,6 +83,7 @@
 #include "mongo/db/service_context.h"
 #include "mongo/db/stats/counters.h"
 #include "mongo/db/stats/server_write_concern_metrics.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/storage_engine.h"
 #include "mongo/db/storage/storage_options.h"
 #include "mongo/db/transaction_participant.h"
@@ -704,7 +705,7 @@ void createOplog(OperationContext* opCtx,
     if (collection) {
         if (replSettings.getOplogSizeBytes() != 0) {
             const CollectionOptions oplogOpts =
-                collection->getCatalogEntry()->getCollectionOptions(opCtx);
+                DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, oplogCollectionName);
 
             int o = (int)(oplogOpts.cappedSize / (1024 * 1024));
             int n = (int)(replSettings.getOplogSizeBytes() / (1024 * 1024));
diff --git a/src/mongo/db/repl/rollback_test_fixture.cpp b/src/mongo/db/repl/rollback_test_fixture.cpp
index d7ac0b18c27..70d500b440d 100644
--- a/src/mongo/db/repl/rollback_test_fixture.cpp
+++ b/src/mongo/db/repl/rollback_test_fixture.cpp
@@ -47,6 +47,7 @@
 #include "mongo/db/repl/replication_process.h"
 #include "mongo/db/repl/replication_recovery.h"
 #include "mongo/db/repl/rs_rollback.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/logger/log_component.h"
 #include "mongo/logger/logger.h"
 #include "mongo/util/str.h"
@@ -321,7 +322,7 @@ void RollbackResyncsCollectionOptionsTest::resyncCollectionOptionsTest(
     // Make sure the collection options are correct.
     AutoGetCollectionForReadCommand autoColl(_opCtx.get(), NamespaceString(nss.toString()));
     auto collAfterRollbackOptions =
-        autoColl.getCollection()->getCatalogEntry()->getCollectionOptions(_opCtx.get());
+        DurableCatalog::get(_opCtx.get())->getCollectionOptions(_opCtx.get(), nss);
 
     BSONObjBuilder expectedOptionsBob;
     if (localCollOptions.uuid) {
diff --git a/src/mongo/db/repl/rs_rollback.cpp b/src/mongo/db/repl/rs_rollback.cpp
index 26491c2728d..c96cc26598d 100644
--- a/src/mongo/db/repl/rs_rollback.cpp
+++ b/src/mongo/db/repl/rs_rollback.cpp
@@ -72,6 +72,7 @@
 #include "mongo/db/repl/rslog.h"
 #include "mongo/db/s/shard_identity_rollback_notifier.h"
 #include "mongo/db/session_catalog_mongod.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/remove_saver.h"
 #include "mongo/db/transaction_participant.h"
 #include "mongo/s/client/shard_registry.h"
@@ -1227,8 +1228,6 @@ void rollback_internal::syncFixUp(OperationContext* opCtx,
             Collection* collection = CollectionCatalog::get(opCtx).lookupCollectionByUUID(uuid);
             invariant(collection);
 
-            auto cce = collection->getCatalogEntry();
-
             auto infoResult = rollbackSource.getCollectionInfoByUUID(nss->db().toString(), uuid);
 
             if (!infoResult.isOK()) {
@@ -1275,7 +1274,7 @@ void rollback_internal::syncFixUp(OperationContext* opCtx,
             WriteUnitOfWork wuow(opCtx);
 
             // Set collection to whatever temp status is on the sync source.
-            cce->setIsTemp(opCtx, options.temp);
+            DurableCatalog::get(opCtx)->setIsTemp(opCtx, *nss, options.temp);
 
             // Set any document validation options. We update the validator fields without
             // parsing/validation, since we fetched the options object directly from the sync
@@ -1295,8 +1294,9 @@ void rollback_internal::syncFixUp(OperationContext* opCtx,
             wuow.commit();
 
             LOG(1) << "Resynced collection metadata for collection: " << *nss << ", UUID: " << uuid
-                   << ", with: " << redact(info)
-                   << ", to: " << redact(cce->getCollectionOptions(opCtx).toBSON());
+                   << ", with: " << redact(info) << ", to: "
+                   << redact(
+                          DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, *nss).toBSON());
         }
 
         // Since we read from the sync source to retrieve the metadata of the
diff --git a/src/mongo/db/repl/rs_rollback_test.cpp b/src/mongo/db/repl/rs_rollback_test.cpp
index b669cb65536..26262101061 100644
--- a/src/mongo/db/repl/rs_rollback_test.cpp
+++ b/src/mongo/db/repl/rs_rollback_test.cpp
@@ -57,6 +57,7 @@
 #include "mongo/db/repl/rollback_test_fixture.h"
 #include "mongo/db/repl/rs_rollback.h"
 #include "mongo/db/s/shard_identity_rollback_notifier.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/unittest/death_test.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/net/hostandport.h"
@@ -1114,7 +1115,8 @@ TEST_F(RSRollbackTest, RollbackRenameCollectionInSameDatabaseCommand) {
 
         // Remote collection options should have been empty.
         auto collAfterRollbackOptions =
-            oldCollName.getCollection()->getCatalogEntry()->getCollectionOptions(_opCtx.get());
+            DurableCatalog::get(_opCtx.get())
+                ->getCollectionOptions(_opCtx.get(), oldCollName.getCollection()->ns());
         ASSERT_BSONOBJ_EQ(BSON("uuid" << *options.uuid), collAfterRollbackOptions.toBSON());
     }
 }
@@ -1169,9 +1171,8 @@ TEST_F(RSRollbackTest,
     ASSERT_TRUE(rollbackSource.getCollectionInfoCalled);
 
     AutoGetCollectionForReadCommand autoColl(_opCtx.get(), NamespaceString(renameFromNss));
-    auto collAfterRollback = autoColl.getCollection();
     auto collAfterRollbackOptions =
-        collAfterRollback->getCatalogEntry()->getCollectionOptions(_opCtx.get());
+        DurableCatalog::get(_opCtx.get())->getCollectionOptions(_opCtx.get(), renameFromNss);
     ASSERT_TRUE(collAfterRollbackOptions.temp);
     ASSERT_BSONOBJ_EQ(BSON("uuid" << *options.uuid << "temp" << true),
                       collAfterRollbackOptions.toBSON());
@@ -1815,7 +1816,8 @@ TEST_F(RSRollbackTest, RollbackCollectionModificationCommand) {
     // Make sure the collection options are correct.
     AutoGetCollectionForReadCommand autoColl(_opCtx.get(), NamespaceString("test.t"));
     auto collAfterRollbackOptions =
-        autoColl.getCollection()->getCatalogEntry()->getCollectionOptions(_opCtx.get());
+        DurableCatalog::get(_opCtx.get())
+            ->getCollectionOptions(_opCtx.get(), NamespaceString("test.t"));
     ASSERT_BSONOBJ_EQ(BSON("uuid" << *options.uuid), collAfterRollbackOptions.toBSON());
 }
 
diff --git a/src/mongo/db/repl/storage_interface_impl.cpp b/src/mongo/db/repl/storage_interface_impl.cpp
index 5675b38179a..1085568ca1d 100644
--- a/src/mongo/db/repl/storage_interface_impl.cpp
+++ b/src/mongo/db/repl/storage_interface_impl.cpp
@@ -75,6 +75,7 @@
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/repl/rollback_gen.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/log.h"
 #include "mongo/util/str.h"
@@ -423,9 +424,8 @@ StatusWith<size_t> StorageInterfaceImpl::getOplogMaxSize(OperationContext* opCtx
     if (!collectionResult.isOK()) {
         return collectionResult.getStatus();
     }
-    auto collection = collectionResult.getValue();
 
-    const auto options = collection->getCatalogEntry()->getCollectionOptions(opCtx);
+    const auto options = DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, nss);
     if (!options.capped)
         return {ErrorCodes::BadValue, str::stream() << nss.ns() << " isn't capped"};
 
diff --git a/src/mongo/db/repl/storage_interface_impl_test.cpp b/src/mongo/db/repl/storage_interface_impl_test.cpp
index cd4240a234c..248142c5bc6 100644
--- a/src/mongo/db/repl/storage_interface_impl_test.cpp
+++ b/src/mongo/db/repl/storage_interface_impl_test.cpp
@@ -52,6 +52,7 @@
 #include "mongo/db/repl/storage_interface_impl.h"
 #include "mongo/db/repl/sync_tail_test_fixture.h"
 #include "mongo/db/service_context_d_test_fixture.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/stdx/thread.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/assert_util.h"
@@ -802,7 +803,7 @@ TEST_F(StorageInterfaceImplTest, RenameCollectionWithStayTempFalseMakesItNotTemp
 
     AutoGetCollectionForReadCommand autoColl2(opCtx, toNss);
     ASSERT_TRUE(autoColl2.getCollection());
-    ASSERT_FALSE(autoColl2.getCollection()->getCatalogEntry()->getCollectionOptions(opCtx).temp);
+    ASSERT_FALSE(DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, toNss).temp);
 }
 
 TEST_F(StorageInterfaceImplTest, RenameCollectionWithStayTempTrueMakesItTemp) {
@@ -821,7 +822,7 @@ TEST_F(StorageInterfaceImplTest, RenameCollectionWithStayTempTrueMakesItTemp) {
 
     AutoGetCollectionForReadCommand autoColl2(opCtx, toNss);
     ASSERT_TRUE(autoColl2.getCollection());
-    ASSERT_TRUE(autoColl2.getCollection()->getCatalogEntry()->getCollectionOptions(opCtx).temp);
+    ASSERT_TRUE(DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, toNss).temp);
 }
 
 TEST_F(StorageInterfaceImplTest, RenameCollectionFailsBetweenDatabases) {
diff --git a/src/mongo/db/storage/biggie/biggie_kv_engine.cpp b/src/mongo/db/storage/biggie/biggie_kv_engine.cpp
index 3aeefdf59c2..0437c5b160d 100644
--- a/src/mongo/db/storage/biggie/biggie_kv_engine.cpp
+++ b/src/mongo/db/storage/biggie/biggie_kv_engine.cpp
@@ -105,6 +105,7 @@ bool KVEngine::trySwapMaster(StringStore& newMaster, uint64_t version) {
 
 
 Status KVEngine::createSortedDataInterface(OperationContext* opCtx,
+                                           const CollectionOptions& collOptions,
                                            StringData ident,
                                            const IndexDescriptor* desc) {
     _idents[ident.toString()] = false;
diff --git a/src/mongo/db/storage/biggie/biggie_kv_engine.h b/src/mongo/db/storage/biggie/biggie_kv_engine.h
index e41f334a329..e5af4004816 100644
--- a/src/mongo/db/storage/biggie/biggie_kv_engine.h
+++ b/src/mongo/db/storage/biggie/biggie_kv_engine.h
@@ -67,6 +67,7 @@ public:
                                                                          StringData ident) override;
 
     virtual Status createSortedDataInterface(OperationContext* opCtx,
+                                             const CollectionOptions& collOptions,
                                              StringData ident,
                                              const IndexDescriptor* desc);
 
diff --git a/src/mongo/db/storage/bson_collection_catalog_entry.cpp b/src/mongo/db/storage/bson_collection_catalog_entry.cpp
index 1ff70e21d04..8b74fed91b4 100644
--- a/src/mongo/db/storage/bson_collection_catalog_entry.cpp
+++ b/src/mongo/db/storage/bson_collection_catalog_entry.cpp
@@ -106,114 +106,6 @@ const StringData BSONCollectionCatalogEntry::kIndexBuildDraining = "draining"_sd
 BSONCollectionCatalogEntry::BSONCollectionCatalogEntry(StringData ns)
     : CollectionCatalogEntry(ns) {}
 
-CollectionOptions BSONCollectionCatalogEntry::getCollectionOptions(OperationContext* opCtx) const {
-    MetaData md = _getMetaData(opCtx);
-    return md.options;
-}
-
-int BSONCollectionCatalogEntry::getTotalIndexCount(OperationContext* opCtx) const {
-    MetaData md = _getMetaData(opCtx);
-
-    return static_cast<int>(md.indexes.size());
-}
-
-int BSONCollectionCatalogEntry::getCompletedIndexCount(OperationContext* opCtx) const {
-    MetaData md = _getMetaData(opCtx);
-
-    int num = 0;
-    for (unsigned i = 0; i < md.indexes.size(); i++) {
-        if (md.indexes[i].ready)
-            num++;
-    }
-    return num;
-}
-
-BSONObj BSONCollectionCatalogEntry::getIndexSpec(OperationContext* opCtx,
-                                                 StringData indexName) const {
-    MetaData md = _getMetaData(opCtx);
-
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-
-    BSONObj spec = md.indexes[offset].spec.getOwned();
-    if (spec.hasField("ns") || disableIndexSpecNamespaceGeneration.load()) {
-        return spec;
-    }
-
-    BSONObj nsObj = BSON("ns" << ns().ns());
-    spec = spec.addField(nsObj.firstElement());
-    return spec;
-}
-
-
-void BSONCollectionCatalogEntry::getAllIndexes(OperationContext* opCtx,
-                                               std::vector<std::string>* names) const {
-    MetaData md = _getMetaData(opCtx);
-
-    for (unsigned i = 0; i < md.indexes.size(); i++) {
-        names->push_back(md.indexes[i].spec["name"].String());
-    }
-}
-
-void BSONCollectionCatalogEntry::getReadyIndexes(OperationContext* opCtx,
-                                                 std::vector<std::string>* names) const {
-    MetaData md = _getMetaData(opCtx);
-
-    for (unsigned i = 0; i < md.indexes.size(); i++) {
-        if (md.indexes[i].ready)
-            names->push_back(md.indexes[i].spec["name"].String());
-    }
-}
-
-void BSONCollectionCatalogEntry::getAllUniqueIndexes(OperationContext* opCtx,
-                                                     std::vector<std::string>* names) const {
-    MetaData md = _getMetaData(opCtx);
-
-    for (unsigned i = 0; i < md.indexes.size(); i++) {
-        if (md.indexes[i].spec["unique"]) {
-            std::string indexName = md.indexes[i].spec["name"].String();
-            names->push_back(indexName);
-        }
-    }
-}
-
-bool BSONCollectionCatalogEntry::isIndexMultikey(OperationContext* opCtx,
-                                                 StringData indexName,
-                                                 MultikeyPaths* multikeyPaths) const {
-    MetaData md = _getMetaData(opCtx);
-
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-
-    if (multikeyPaths && !md.indexes[offset].multikeyPaths.empty()) {
-        *multikeyPaths = md.indexes[offset].multikeyPaths;
-    }
-
-    return md.indexes[offset].multikey;
-}
-
-bool BSONCollectionCatalogEntry::isIndexPresent(OperationContext* opCtx,
-                                                StringData indexName) const {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    return offset >= 0;
-}
-
-bool BSONCollectionCatalogEntry::isIndexReady(OperationContext* opCtx, StringData indexName) const {
-    MetaData md = _getMetaData(opCtx);
-
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    return md.indexes[offset].ready;
-}
-
-KVPrefix BSONCollectionCatalogEntry::getIndexPrefix(OperationContext* opCtx,
-                                                    StringData indexName) const {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    return md.indexes[offset].prefix;
-}
 
 // --------------------------
 
diff --git a/src/mongo/db/storage/bson_collection_catalog_entry.h b/src/mongo/db/storage/bson_collection_catalog_entry.h
index 5f9ae3274bb..d33b895d237 100644
--- a/src/mongo/db/storage/bson_collection_catalog_entry.h
+++ b/src/mongo/db/storage/bson_collection_catalog_entry.h
@@ -51,31 +51,6 @@ public:
 
     virtual ~BSONCollectionCatalogEntry() {}
 
-    virtual CollectionOptions getCollectionOptions(OperationContext* opCtx) const;
-
-    virtual int getTotalIndexCount(OperationContext* opCtx) const;
-
-    virtual int getCompletedIndexCount(OperationContext* opCtx) const;
-
-    virtual BSONObj getIndexSpec(OperationContext* opCtx, StringData idxName) const;
-
-    virtual void getAllIndexes(OperationContext* opCtx, std::vector<std::string>* names) const;
-
-    virtual void getReadyIndexes(OperationContext* opCtx, std::vector<std::string>* names) const;
-
-    virtual void getAllUniqueIndexes(OperationContext* opCtx,
-                                     std::vector<std::string>* names) const;
-
-    virtual bool isIndexMultikey(OperationContext* opCtx,
-                                 StringData indexName,
-                                 MultikeyPaths* multikeyPaths) const;
-
-    virtual bool isIndexReady(OperationContext* opCtx, StringData indexName) const;
-
-    virtual bool isIndexPresent(OperationContext* opCtx, StringData indexName) const;
-
-    virtual KVPrefix getIndexPrefix(OperationContext* opCtx, StringData indexName) const;
-
     // ------ for implementors
 
     struct IndexMetaData {
@@ -128,8 +103,5 @@ public:
         std::vector<IndexMetaData> indexes;
         KVPrefix prefix = KVPrefix::kNotPrefixed;
     };
-
-protected:
-    virtual MetaData _getMetaData(OperationContext* opCtx) const = 0;
 };
 }
diff --git a/src/mongo/db/storage/devnull/devnull_kv_engine.h b/src/mongo/db/storage/devnull/devnull_kv_engine.h
index 33fe29d38f4..7fcf3968f5c 100644
--- a/src/mongo/db/storage/devnull/devnull_kv_engine.h
+++ b/src/mongo/db/storage/devnull/devnull_kv_engine.h
@@ -65,6 +65,7 @@ public:
                                                                   StringData ident) override;
 
     virtual Status createSortedDataInterface(OperationContext* opCtx,
+                                             const CollectionOptions& collOptions,
                                              StringData ident,
                                              const IndexDescriptor* desc) {
         return Status::OK();
diff --git a/src/mongo/db/storage/durable_catalog.h b/src/mongo/db/storage/durable_catalog.h
new file mode 100644
index 00000000000..a7c85f38e3b
--- /dev/null
+++ b/src/mongo/db/storage/durable_catalog.h
@@ -0,0 +1,306 @@
+/**
+ *    Copyright (C) 2019-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/base/string_data.h"
+#include "mongo/db/catalog/collection_catalog_entry.h"
+#include "mongo/db/catalog/collection_options.h"
+#include "mongo/db/namespace_string.h"
+#include "mongo/db/operation_context.h"
+#include "mongo/db/storage/bson_collection_catalog_entry.h"
+#include "mongo/db/storage/kv/kv_prefix.h"
+#include "mongo/db/storage/storage_engine.h"
+
+namespace mongo {
+
+/**
+ * An interface to modify the on-disk catalog metadata.
+ */
+class DurableCatalog {
+    DurableCatalog(const DurableCatalog&) = delete;
+    DurableCatalog& operator=(const DurableCatalog&) = delete;
+    DurableCatalog(DurableCatalog&&) = delete;
+    DurableCatalog& operator=(DurableCatalog&&) = delete;
+
+protected:
+    DurableCatalog() = default;
+
+public:
+    virtual ~DurableCatalog() {}
+
+    static DurableCatalog* get(OperationContext* opCtx) {
+        return opCtx->getServiceContext()->getStorageEngine()->getCatalog();
+    }
+
+    virtual void init(OperationContext* opCtx) = 0;
+
+    virtual std::vector<NamespaceString> getAllCollections() const = 0;
+
+    virtual std::string getCollectionIdent(const NamespaceString& nss) const = 0;
+
+    virtual std::string getIndexIdent(OperationContext* opCtx,
+                                      const NamespaceString& nss,
+                                      StringData idName) const = 0;
+
+    virtual BSONCollectionCatalogEntry::MetaData getMetaData(OperationContext* opCtx,
+                                                             const NamespaceString& nss) const = 0;
+    virtual void putMetaData(OperationContext* opCtx,
+                             const NamespaceString& nss,
+                             BSONCollectionCatalogEntry::MetaData& md) = 0;
+
+    virtual std::vector<std::string> getAllIdentsForDB(StringData db) const = 0;
+    virtual std::vector<std::string> getAllIdents(OperationContext* opCtx) const = 0;
+
+    virtual bool isUserDataIdent(StringData ident) const = 0;
+
+    virtual bool isInternalIdent(StringData ident) const = 0;
+
+    virtual bool isCollectionIdent(StringData ident) const = 0;
+
+    virtual RecordStore* getRecordStore() = 0;
+
+    /**
+     * Create an entry in the catalog for an orphaned collection found in the
+     * storage engine. Return the generated ns of the collection.
+     * Note that this function does not recreate the _id index on the collection because it does not
+     * have access to index catalog.
+     */
+    virtual StatusWith<std::string> newOrphanedIdent(OperationContext* opCtx,
+                                                     std::string ident) = 0;
+
+    virtual std::string getFilesystemPathForDb(const std::string& dbName) const = 0;
+
+    /**
+     * Generate an internal ident name.
+     */
+    virtual std::string newInternalIdent() = 0;
+
+    virtual std::unique_ptr<CollectionCatalogEntry> makeCollectionCatalogEntry(
+        OperationContext* opCtx, const NamespaceString& nss, bool forRepair) = 0;
+
+    virtual StatusWith<std::unique_ptr<CollectionCatalogEntry>> createCollection(
+        OperationContext* opCtx,
+        const NamespaceString& nss,
+        const CollectionOptions& options,
+        bool allocateDefaultSpace) = 0;
+
+    virtual Status renameCollection(OperationContext* opCtx,
+                                    const NamespaceString& fromNss,
+                                    const NamespaceString& toNss,
+                                    bool stayTemp) = 0;
+
+    virtual Status dropCollection(OperationContext* opCtx, const NamespaceString& nss) = 0;
+
+    /**
+     * Updates size of a capped Collection.
+     */
+    virtual void updateCappedSize(OperationContext* opCtx, NamespaceString ns, long long size) = 0;
+
+    /*
+     * Updates the expireAfterSeconds field of the given index to the value in newExpireSecs.
+     * The specified index must already contain an expireAfterSeconds field, and the value in
+     * that field and newExpireSecs must both be numeric.
+     */
+    virtual void updateTTLSetting(OperationContext* opCtx,
+                                  NamespaceString ns,
+                                  StringData idxName,
+                                  long long newExpireSeconds) = 0;
+
+    /**
+     * Compare the UUID argument to the UUID obtained from the metadata. Return true if they
+     * are equal, false otherwise. uuid can become a CollectionUUID once MMAPv1 is removed.
+     */
+    virtual bool isEqualToMetadataUUID(OperationContext* opCtx,
+                                       NamespaceString ns,
+                                       OptionalCollectionUUID uuid) = 0;
+
+    /**
+     * Updates the 'temp' setting for this collection.
+     */
+    virtual void setIsTemp(OperationContext* opCtx, NamespaceString ns, bool isTemp) = 0;
+
+    virtual boost::optional<std::string> getSideWritesIdent(OperationContext* opCtx,
+                                                            NamespaceString ns,
+                                                            StringData indexName) const = 0;
+
+    // TODO SERVER-36385 Remove this function: we don't set the feature tracker bit in 4.4 because
+    // 4.4 can only downgrade to 4.2 which can read long TypeBits.
+    virtual void setIndexKeyStringWithLongTypeBitsExistsOnDisk(OperationContext* opCtx) = 0;
+
+    /**
+     * Updates the validator for this collection.
+     *
+     * An empty validator removes all validation.
+     */
+    virtual void updateValidator(OperationContext* opCtx,
+                                 NamespaceString ns,
+                                 const BSONObj& validator,
+                                 StringData validationLevel,
+                                 StringData validationAction) = 0;
+
+    virtual void updateIndexMetadata(OperationContext* opCtx,
+                                     NamespaceString ns,
+                                     const IndexDescriptor* desc) = 0;
+
+    virtual Status removeIndex(OperationContext* opCtx,
+                               NamespaceString ns,
+                               StringData indexName) = 0;
+
+    virtual Status prepareForIndexBuild(OperationContext* opCtx,
+                                        NamespaceString ns,
+                                        const IndexDescriptor* spec,
+                                        IndexBuildProtocol indexBuildProtocol,
+                                        bool isBackgroundSecondaryBuild) = 0;
+
+    /**
+     * Returns whether or not the index is being built with the two-phase index build procedure.
+     */
+    virtual bool isTwoPhaseIndexBuild(OperationContext* opCtx,
+                                      NamespaceString ns,
+                                      StringData indexName) const = 0;
+
+    /**
+     * Indicate that a build index is now in the "scanning" phase of a hybrid index build. The
+     * 'constraintViolationsIdent' is only used for unique indexes.
+     *
+     * It is only valid to call this when the index is using the kTwoPhase IndexBuildProtocol.
+     */
+    virtual void setIndexBuildScanning(OperationContext* opCtx,
+                                       NamespaceString ns,
+                                       StringData indexName,
+                                       std::string sideWritesIdent,
+                                       boost::optional<std::string> constraintViolationsIdent) = 0;
+
+
+    /**
+     * Returns whether or not this index is building in the "scanning" phase.
+     */
+    virtual bool isIndexBuildScanning(OperationContext* opCtx,
+                                      NamespaceString ns,
+                                      StringData indexName) const = 0;
+
+    /**
+     * Indicate that a build index is now in the "draining" phase of a hybrid index build.
+     *
+     * It is only valid to call this when the index is using the kTwoPhase IndexBuildProtocol.
+     */
+    virtual void setIndexBuildDraining(OperationContext* opCtx,
+                                       NamespaceString ns,
+                                       StringData indexName) = 0;
+
+    /**
+     * Returns whether or not this index is building in the "draining" phase.
+     */
+    virtual bool isIndexBuildDraining(OperationContext* opCtx,
+                                      NamespaceString ns,
+                                      StringData indexName) const = 0;
+
+    /**
+     * Indicate that an index build is completed and the index is ready to use.
+     */
+    virtual void indexBuildSuccess(OperationContext* opCtx,
+                                   NamespaceString ns,
+                                   StringData indexName) = 0;
+
+    /**
+     * Returns true if the index identified by 'indexName' is multikey, and returns false otherwise.
+     *
+     * If the 'multikeyPaths' pointer is non-null, then it must point to an empty vector. If this
+     * index supports tracking path-level multikey information, then this function sets
+     * 'multikeyPaths' as the path components that cause this index to be multikey.
+     *
+     * In particular, if this function returns false and the index supports tracking path-level
+     * multikey information, then 'multikeyPaths' is initialized as a vector with size equal to the
+     * number of elements in the index key pattern of empty sets.
+     */
+    virtual bool isIndexMultikey(OperationContext* opCtx,
+                                 NamespaceString ns,
+                                 StringData indexName,
+                                 MultikeyPaths* multikeyPaths) const = 0;
+
+    /**
+     * Sets the index identified by 'indexName' to be multikey.
+     *
+     * If 'multikeyPaths' is non-empty, then it must be a vector with size equal to the number of
+     * elements in the index key pattern. Additionally, at least one path component of the indexed
+     * fields must cause this index to be multikey.
+     *
+     * This function returns true if the index metadata has changed, and returns false otherwise.
+     */
+    virtual bool setIndexIsMultikey(OperationContext* opCtx,
+                                    NamespaceString ns,
+                                    StringData indexName,
+                                    const MultikeyPaths& multikeyPaths) = 0;
+
+    virtual boost::optional<std::string> getConstraintViolationsIdent(
+        OperationContext* opCtx, NamespaceString ns, StringData indexName) const = 0;
+
+    /**
+     * Returns the server-compatibility version of the index build procedure.
+     */
+    virtual long getIndexBuildVersion(OperationContext* opCtx,
+                                      NamespaceString ns,
+                                      StringData indexName) const = 0;
+
+    virtual CollectionOptions getCollectionOptions(OperationContext* opCtx,
+                                                   NamespaceString ns) const = 0;
+
+    virtual int getTotalIndexCount(OperationContext* opCtx, NamespaceString ns) const = 0;
+
+    virtual int getCompletedIndexCount(OperationContext* opCtx, NamespaceString ns) const = 0;
+
+    virtual BSONObj getIndexSpec(OperationContext* opCtx,
+                                 NamespaceString ns,
+                                 StringData indexName) const = 0;
+
+    virtual void getAllIndexes(OperationContext* opCtx,
+                               NamespaceString ns,
+                               std::vector<std::string>* names) const = 0;
+
+    virtual void getReadyIndexes(OperationContext* opCtx,
+                                 NamespaceString ns,
+                                 std::vector<std::string>* names) const = 0;
+    virtual void getAllUniqueIndexes(OperationContext* opCtx,
+                                     NamespaceString ns,
+                                     std::vector<std::string>* names) const = 0;
+
+    virtual bool isIndexPresent(OperationContext* opCtx,
+                                NamespaceString ns,
+                                StringData indexName) const = 0;
+
+    virtual bool isIndexReady(OperationContext* opCtx,
+                              NamespaceString ns,
+                              StringData indexName) const = 0;
+
+    virtual KVPrefix getIndexPrefix(OperationContext* opCtx,
+                                    NamespaceString ns,
+                                    StringData indexName) const = 0;
+};
+}  // namespace mongo
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_engine.cpp b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_engine.cpp
index 0cb25117917..209af0253fa 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_engine.cpp
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_engine.cpp
@@ -83,6 +83,7 @@ std::unique_ptr<RecordStore> EphemeralForTestEngine::makeTemporaryRecordStore(
 }
 
 Status EphemeralForTestEngine::createSortedDataInterface(OperationContext* opCtx,
+                                                         const CollectionOptions& collOptions,
                                                          StringData ident,
                                                          const IndexDescriptor* desc) {
     // Register the ident in `_dataMap` (for `getAllIdents`). Remainder of work done in
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_engine.h b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_engine.h
index a499bba6b67..3eabcae6654 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_engine.h
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_engine.h
@@ -58,6 +58,7 @@ public:
                                                                   StringData ident) override;
 
     virtual Status createSortedDataInterface(OperationContext* opCtx,
+                                             const CollectionOptions& collOptions,
                                              StringData ident,
                                              const IndexDescriptor* desc);
 
diff --git a/src/mongo/db/storage/kv/SConscript b/src/mongo/db/storage/kv/SConscript
index 058d45096d0..4f70282fda2 100644
--- a/src/mongo/db/storage/kv/SConscript
+++ b/src/mongo/db/storage/kv/SConscript
@@ -18,7 +18,7 @@ env.Library(
     source=[
         'kv_catalog.cpp',
         'kv_collection_catalog_entry.cpp',
-        ],
+    ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/bson/util/bson_extract',
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
@@ -29,8 +29,11 @@ env.Library(
         '$BUILD_DIR/mongo/db/catalog/collection_catalog',
         'kv_drop_pending_ident_reaper',
         'kv_prefix',
-        ],
-    )
+    ],
+    LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/db/catalog/disable_index_spec_namespace_generation',
+    ],
+)
 
 # Should not be referenced outside this SConscript file.
 env.Library(
diff --git a/src/mongo/db/storage/kv/kv_catalog.cpp b/src/mongo/db/storage/kv/kv_catalog.cpp
index dcecb8c429c..105f1dbfd58 100644
--- a/src/mongo/db/storage/kv/kv_catalog.cpp
+++ b/src/mongo/db/storage/kv/kv_catalog.cpp
@@ -36,7 +36,9 @@
 
 #include "mongo/bson/util/bson_extract.h"
 #include "mongo/bson/util/builder.h"
+#include "mongo/db/catalog/disable_index_spec_namespace_generation_gen.h"
 #include "mongo/db/concurrency/d_concurrency.h"
+#include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/storage/kv/kv_catalog_feature_tracker.h"
@@ -132,6 +134,10 @@ std::string escapeDbName(StringData dbname) {
     return escaped;
 }
 
+bool indexTypeSupportsPathLevelMultikeyTracking(StringData accessMethod) {
+    return accessMethod == IndexNames::BTREE || accessMethod == IndexNames::GEO_2DSPHERE;
+}
+
 }  // namespace
 
 using std::unique_ptr;
@@ -168,6 +174,61 @@ public:
     const Entry _entry;
 };
 
+class KVCatalog::AddIndexChange : public RecoveryUnit::Change {
+public:
+    AddIndexChange(OperationContext* opCtx, StorageEngineInterface* engine, StringData ident)
+        : _opCtx(opCtx), _engine(engine), _ident(ident.toString()) {}
+
+    virtual void commit(boost::optional<Timestamp>) {}
+    virtual void rollback() {
+        // Intentionally ignoring failure.
+        auto kvEngine = _engine->getEngine();
+        MONGO_COMPILER_VARIABLE_UNUSED auto status = kvEngine->dropIdent(_opCtx, _ident);
+    }
+
+    OperationContext* const _opCtx;
+    StorageEngineInterface* _engine;
+    const std::string _ident;
+};
+
+class KVCatalog::RemoveIndexChange : public RecoveryUnit::Change {
+public:
+    RemoveIndexChange(OperationContext* opCtx,
+                      StorageEngineInterface* engine,
+                      OptionalCollectionUUID uuid,
+                      const NamespaceString& indexNss,
+                      StringData indexName,
+                      StringData ident)
+        : _opCtx(opCtx),
+          _engine(engine),
+          _uuid(uuid),
+          _indexNss(indexNss),
+          _indexName(indexName),
+          _ident(ident.toString()) {}
+
+    virtual void rollback() {}
+    virtual void commit(boost::optional<Timestamp> commitTimestamp) {
+        // Intentionally ignoring failure here. Since we've removed the metadata pointing to the
+        // index, we should never see it again anyway.
+        if (_engine->getStorageEngine()->supportsPendingDrops() && commitTimestamp) {
+            log() << "Deferring table drop for index '" << _indexName << "' on collection '"
+                  << _indexNss << (_uuid ? " (" + _uuid->toString() + ")'" : "") << ". Ident: '"
+                  << _ident << "', commit timestamp: '" << commitTimestamp << "'";
+            _engine->addDropPendingIdent(*commitTimestamp, _indexNss, _ident);
+        } else {
+            auto kvEngine = _engine->getEngine();
+            MONGO_COMPILER_VARIABLE_UNUSED auto status = kvEngine->dropIdent(_opCtx, _ident);
+        }
+    }
+
+    OperationContext* const _opCtx;
+    StorageEngineInterface* _engine;
+    OptionalCollectionUUID _uuid;
+    const NamespaceString _indexNss;
+    const std::string _indexName;
+    const std::string _ident;
+};
+
 bool KVCatalog::FeatureTracker::isFeatureDocument(BSONObj obj) {
     BSONElement firstElem = obj.firstElement();
     if (firstElem.fieldNameStringData() == kIsFeatureDocumentFieldName) {
@@ -726,8 +787,7 @@ std::unique_ptr<CollectionCatalogEntry> KVCatalog::makeCollectionCatalogEntry(
         invariant(rs);
     }
 
-    return std::make_unique<KVCollectionCatalogEntry>(
-        _engine, this, nss.ns(), ident, std::move(rs));
+    return std::make_unique<KVCollectionCatalogEntry>(_engine, nss.ns(), ident, std::move(rs));
 }
 
 StatusWith<std::unique_ptr<CollectionCatalogEntry>> KVCatalog::createCollection(
@@ -774,8 +834,7 @@ StatusWith<std::unique_ptr<CollectionCatalogEntry>> KVCatalog::createCollection(
     auto rs = _engine->getEngine()->getGroupedRecordStore(opCtx, nss.ns(), ident, options, prefix);
     invariant(rs);
 
-    return std::make_unique<KVCollectionCatalogEntry>(
-        _engine, this, nss.ns(), ident, std::move(rs));
+    return std::make_unique<KVCollectionCatalogEntry>(_engine, nss.ns(), ident, std::move(rs));
 }
 
 Status KVCatalog::renameCollection(OperationContext* opCtx,
@@ -814,17 +873,17 @@ Status KVCatalog::dropCollection(OperationContext* opCtx, const NamespaceString&
     auto& catalog = CollectionCatalog::get(opCtx);
     auto uuid = catalog.lookupUUIDByNSS(nss);
 
-    invariant(entry->getTotalIndexCount(opCtx) == entry->getCompletedIndexCount(opCtx));
+    invariant(getTotalIndexCount(opCtx, nss) == getCompletedIndexCount(opCtx, nss));
 
     {
         std::vector<std::string> indexNames;
-        entry->getAllIndexes(opCtx, &indexNames);
+        getAllIndexes(opCtx, nss, &indexNames);
         for (size_t i = 0; i < indexNames.size(); i++) {
-            entry->removeIndex(opCtx, indexNames[i]).transitional_ignore();
+            Status status = removeIndex(opCtx, nss, indexNames[i]);
         }
     }
 
-    invariant(entry->getTotalIndexCount(opCtx) == 0);
+    invariant(getTotalIndexCount(opCtx, nss) == 0);
 
     const std::string ident = getCollectionIdent(nss);
 
@@ -854,4 +913,402 @@ Status KVCatalog::dropCollection(OperationContext* opCtx, const NamespaceString&
 
     return Status::OK();
 }
+
+void KVCatalog::updateCappedSize(OperationContext* opCtx, NamespaceString ns, long long size) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    md.options.cappedSize = size;
+    putMetaData(opCtx, ns, md);
+}
+
+void KVCatalog::updateTTLSetting(OperationContext* opCtx,
+                                 NamespaceString ns,
+                                 StringData idxName,
+                                 long long newExpireSeconds) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(idxName);
+    invariant(offset >= 0);
+    md.indexes[offset].updateTTLSetting(newExpireSeconds);
+    putMetaData(opCtx, ns, md);
+}
+
+bool KVCatalog::isEqualToMetadataUUID(OperationContext* opCtx,
+                                      NamespaceString ns,
+                                      OptionalCollectionUUID uuid) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    return md.options.uuid && md.options.uuid == uuid;
+}
+
+void KVCatalog::setIsTemp(OperationContext* opCtx, NamespaceString ns, bool isTemp) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    md.options.temp = isTemp;
+    putMetaData(opCtx, ns, md);
+}
+
+boost::optional<std::string> KVCatalog::getSideWritesIdent(OperationContext* opCtx,
+                                                           NamespaceString ns,
+                                                           StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    return md.indexes[offset].sideWritesIdent;
+}
+
+void KVCatalog::setIndexKeyStringWithLongTypeBitsExistsOnDisk(OperationContext* opCtx) {
+    const auto feature = FeatureTracker::RepairableFeature::kIndexKeyStringWithLongTypeBits;
+    if (!getFeatureTracker()->isRepairableFeatureInUse(opCtx, feature)) {
+        getFeatureTracker()->markRepairableFeatureAsInUse(opCtx, feature);
+    }
+}
+
+void KVCatalog::updateValidator(OperationContext* opCtx,
+                                NamespaceString ns,
+                                const BSONObj& validator,
+                                StringData validationLevel,
+                                StringData validationAction) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    md.options.validator = validator;
+    md.options.validationLevel = validationLevel.toString();
+    md.options.validationAction = validationAction.toString();
+    putMetaData(opCtx, ns, md);
+}
+
+void KVCatalog::updateIndexMetadata(OperationContext* opCtx,
+                                    NamespaceString ns,
+                                    const IndexDescriptor* desc) {
+    // Update any metadata Ident has for this index
+    const string ident = getIndexIdent(opCtx, ns, desc->indexName());
+    auto kvEngine = _engine->getEngine();
+    kvEngine->alterIdentMetadata(opCtx, ident, desc);
+}
+
+Status KVCatalog::removeIndex(OperationContext* opCtx, NamespaceString ns, StringData indexName) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    if (md.findIndexOffset(indexName) < 0)
+        return Status::OK();  // never had the index so nothing to do.
+
+    const string ident = getIndexIdent(opCtx, ns, indexName);
+
+    md.eraseIndex(indexName);
+    putMetaData(opCtx, ns, md);
+
+    // Lazily remove to isolate underlying engine from rollback.
+    opCtx->recoveryUnit()->registerChange(new RemoveIndexChange(
+        opCtx, _engine, md.options.uuid, ns.makeIndexNamespace(indexName), indexName, ident));
+    return Status::OK();
+}
+
+Status KVCatalog::prepareForIndexBuild(OperationContext* opCtx,
+                                       NamespaceString ns,
+                                       const IndexDescriptor* spec,
+                                       IndexBuildProtocol indexBuildProtocol,
+                                       bool isBackgroundSecondaryBuild) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    KVPrefix prefix = KVPrefix::getNextPrefix(ns);
+    BSONCollectionCatalogEntry::IndexMetaData imd;
+    imd.spec = spec->infoObj();
+    imd.ready = false;
+    imd.multikey = false;
+    imd.prefix = prefix;
+    imd.isBackgroundSecondaryBuild = isBackgroundSecondaryBuild;
+    imd.runTwoPhaseBuild = indexBuildProtocol == IndexBuildProtocol::kTwoPhase;
+
+    if (indexTypeSupportsPathLevelMultikeyTracking(spec->getAccessMethodName())) {
+        const auto feature = FeatureTracker::RepairableFeature::kPathLevelMultikeyTracking;
+        if (!getFeatureTracker()->isRepairableFeatureInUse(opCtx, feature)) {
+            getFeatureTracker()->markRepairableFeatureAsInUse(opCtx, feature);
+        }
+        imd.multikeyPaths = MultikeyPaths{static_cast<size_t>(spec->keyPattern().nFields())};
+    }
+
+    // Mark collation feature as in use if the index has a non-simple collation.
+    if (imd.spec["collation"]) {
+        const auto feature = KVCatalog::FeatureTracker::NonRepairableFeature::kCollation;
+        if (!getFeatureTracker()->isNonRepairableFeatureInUse(opCtx, feature)) {
+            getFeatureTracker()->markNonRepairableFeatureAsInUse(opCtx, feature);
+        }
+    }
+
+    md.indexes.push_back(imd);
+    putMetaData(opCtx, ns, md);
+
+    string ident = getIndexIdent(opCtx, ns, spec->indexName());
+
+    auto kvEngine = _engine->getEngine();
+    const Status status = kvEngine->createGroupedSortedDataInterface(
+        opCtx, getCollectionOptions(opCtx, ns), ident, spec, prefix);
+    if (status.isOK()) {
+        opCtx->recoveryUnit()->registerChange(new AddIndexChange(opCtx, _engine, ident));
+    }
+
+    return status;
+}
+
+bool KVCatalog::isTwoPhaseIndexBuild(OperationContext* opCtx,
+                                     NamespaceString ns,
+                                     StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    return md.indexes[offset].runTwoPhaseBuild;
+}
+
+void KVCatalog::setIndexBuildScanning(OperationContext* opCtx,
+                                      NamespaceString ns,
+                                      StringData indexName,
+                                      std::string sideWritesIdent,
+                                      boost::optional<std::string> constraintViolationsIdent) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    invariant(!md.indexes[offset].ready);
+    invariant(!md.indexes[offset].buildPhase);
+    invariant(md.indexes[offset].runTwoPhaseBuild);
+
+    md.indexes[offset].buildPhase = BSONCollectionCatalogEntry::kIndexBuildScanning.toString();
+    md.indexes[offset].sideWritesIdent = sideWritesIdent;
+    md.indexes[offset].constraintViolationsIdent = constraintViolationsIdent;
+    putMetaData(opCtx, ns, md);
+}
+
+bool KVCatalog::isIndexBuildScanning(OperationContext* opCtx,
+                                     NamespaceString ns,
+                                     StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    return md.indexes[offset].buildPhase ==
+        BSONCollectionCatalogEntry::kIndexBuildScanning.toString();
+}
+
+void KVCatalog::setIndexBuildDraining(OperationContext* opCtx,
+                                      NamespaceString ns,
+                                      StringData indexName) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    invariant(!md.indexes[offset].ready);
+    invariant(md.indexes[offset].runTwoPhaseBuild);
+    invariant(md.indexes[offset].buildPhase ==
+              BSONCollectionCatalogEntry::kIndexBuildScanning.toString());
+
+    md.indexes[offset].buildPhase = BSONCollectionCatalogEntry::kIndexBuildDraining.toString();
+    putMetaData(opCtx, ns, md);
+}
+
+bool KVCatalog::isIndexBuildDraining(OperationContext* opCtx,
+                                     NamespaceString ns,
+                                     StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    return md.indexes[offset].buildPhase ==
+        BSONCollectionCatalogEntry::kIndexBuildDraining.toString();
+}
+
+void KVCatalog::indexBuildSuccess(OperationContext* opCtx,
+                                  NamespaceString ns,
+                                  StringData indexName) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    md.indexes[offset].ready = true;
+    md.indexes[offset].runTwoPhaseBuild = false;
+    md.indexes[offset].buildPhase = boost::none;
+    md.indexes[offset].sideWritesIdent = boost::none;
+    md.indexes[offset].constraintViolationsIdent = boost::none;
+    putMetaData(opCtx, ns, md);
+}
+
+bool KVCatalog::isIndexMultikey(OperationContext* opCtx,
+                                NamespaceString ns,
+                                StringData indexName,
+                                MultikeyPaths* multikeyPaths) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+
+    if (multikeyPaths && !md.indexes[offset].multikeyPaths.empty()) {
+        *multikeyPaths = md.indexes[offset].multikeyPaths;
+    }
+
+    return md.indexes[offset].multikey;
+}
+
+bool KVCatalog::setIndexIsMultikey(OperationContext* opCtx,
+                                   NamespaceString ns,
+                                   StringData indexName,
+                                   const MultikeyPaths& multikeyPaths) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+
+    const bool tracksPathLevelMultikeyInfo = !md.indexes[offset].multikeyPaths.empty();
+    if (tracksPathLevelMultikeyInfo) {
+        invariant(!multikeyPaths.empty());
+        invariant(multikeyPaths.size() == md.indexes[offset].multikeyPaths.size());
+    } else {
+        invariant(multikeyPaths.empty());
+
+        if (md.indexes[offset].multikey) {
+            // The index is already set as multikey and we aren't tracking path-level multikey
+            // information for it. We return false to indicate that the index metadata is unchanged.
+            return false;
+        }
+    }
+
+    md.indexes[offset].multikey = true;
+
+    if (tracksPathLevelMultikeyInfo) {
+        bool newPathIsMultikey = false;
+        bool somePathIsMultikey = false;
+
+        // Store new path components that cause this index to be multikey in catalog's index
+        // metadata.
+        for (size_t i = 0; i < multikeyPaths.size(); ++i) {
+            std::set<size_t>& indexMultikeyComponents = md.indexes[offset].multikeyPaths[i];
+            for (const auto multikeyComponent : multikeyPaths[i]) {
+                auto result = indexMultikeyComponents.insert(multikeyComponent);
+                newPathIsMultikey = newPathIsMultikey || result.second;
+                somePathIsMultikey = true;
+            }
+        }
+
+        // If all of the sets in the multikey paths vector were empty, then no component of any
+        // indexed field caused the index to be multikey. setIndexIsMultikey() therefore shouldn't
+        // have been called.
+        invariant(somePathIsMultikey);
+
+        if (!newPathIsMultikey) {
+            // We return false to indicate that the index metadata is unchanged.
+            return false;
+        }
+    }
+
+    putMetaData(opCtx, ns, md);
+    return true;
+}
+
+boost::optional<std::string> KVCatalog::getConstraintViolationsIdent(OperationContext* opCtx,
+                                                                     NamespaceString ns,
+                                                                     StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    return md.indexes[offset].constraintViolationsIdent;
+}
+
+long KVCatalog::getIndexBuildVersion(OperationContext* opCtx,
+                                     NamespaceString ns,
+                                     StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    return md.indexes[offset].versionOfBuild;
+}
+
+CollectionOptions KVCatalog::getCollectionOptions(OperationContext* opCtx,
+                                                  NamespaceString ns) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    return md.options;
+}
+
+int KVCatalog::getTotalIndexCount(OperationContext* opCtx, NamespaceString ns) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    return static_cast<int>(md.indexes.size());
+}
+
+int KVCatalog::getCompletedIndexCount(OperationContext* opCtx, NamespaceString ns) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    int num = 0;
+    for (unsigned i = 0; i < md.indexes.size(); i++) {
+        if (md.indexes[i].ready)
+            num++;
+    }
+    return num;
+}
+
+BSONObj KVCatalog::getIndexSpec(OperationContext* opCtx,
+                                NamespaceString ns,
+                                StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+
+    BSONObj spec = md.indexes[offset].spec.getOwned();
+    if (spec.hasField("ns") || disableIndexSpecNamespaceGeneration.load()) {
+        return spec;
+    }
+
+    BSONObj nsObj = BSON("ns" << ns.ns());
+    spec = spec.addField(nsObj.firstElement());
+    return spec;
+}
+
+void KVCatalog::getAllIndexes(OperationContext* opCtx,
+                              NamespaceString ns,
+                              std::vector<std::string>* names) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    for (unsigned i = 0; i < md.indexes.size(); i++) {
+        names->push_back(md.indexes[i].spec["name"].String());
+    }
+}
+
+void KVCatalog::getReadyIndexes(OperationContext* opCtx,
+                                NamespaceString ns,
+                                std::vector<std::string>* names) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    for (unsigned i = 0; i < md.indexes.size(); i++) {
+        if (md.indexes[i].ready)
+            names->push_back(md.indexes[i].spec["name"].String());
+    }
+}
+
+void KVCatalog::getAllUniqueIndexes(OperationContext* opCtx,
+                                    NamespaceString ns,
+                                    std::vector<std::string>* names) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    for (unsigned i = 0; i < md.indexes.size(); i++) {
+        if (md.indexes[i].spec["unique"]) {
+            std::string indexName = md.indexes[i].spec["name"].String();
+            names->push_back(indexName);
+        }
+    }
+}
+
+bool KVCatalog::isIndexPresent(OperationContext* opCtx,
+                               NamespaceString ns,
+                               StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    return offset >= 0;
+}
+
+bool KVCatalog::isIndexReady(OperationContext* opCtx,
+                             NamespaceString ns,
+                             StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    return md.indexes[offset].ready;
+}
+
+KVPrefix KVCatalog::getIndexPrefix(OperationContext* opCtx,
+                                   NamespaceString ns,
+                                   StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    return md.indexes[offset].prefix;
+}
 }
diff --git a/src/mongo/db/storage/kv/kv_catalog.h b/src/mongo/db/storage/kv/kv_catalog.h
index 957b12c38e1..6d46dcc482b 100644
--- a/src/mongo/db/storage/kv/kv_catalog.h
+++ b/src/mongo/db/storage/kv/kv_catalog.h
@@ -38,6 +38,7 @@
 #include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/record_id.h"
 #include "mongo/db/storage/bson_collection_catalog_entry.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/kv/kv_prefix.h"
 #include "mongo/stdx/mutex.h"
 
@@ -47,7 +48,7 @@ class OperationContext;
 class RecordStore;
 class StorageEngineInterface;
 
-class KVCatalog {
+class KVCatalog : public DurableCatalog {
 public:
     class FeatureTracker;
 
@@ -96,19 +97,10 @@ public:
         return _rs;
     }
 
-    /**
-     * Create an entry in the catalog for an orphaned collection found in the
-     * storage engine. Return the generated ns of the collection.
-     * Note that this function does not recreate the _id index on the collection because it does not
-     * have access to index catalog.
-     */
     StatusWith<std::string> newOrphanedIdent(OperationContext* opCtx, std::string ident);
 
     std::string getFilesystemPathForDb(const std::string& dbName) const;
 
-    /**
-     * Generate an internal ident name.
-     */
     std::string newInternalIdent();
 
     std::unique_ptr<CollectionCatalogEntry> makeCollectionCatalogEntry(OperationContext* opCtx,
@@ -128,9 +120,118 @@ public:
 
     Status dropCollection(OperationContext* opCtx, const NamespaceString& nss);
 
+    void updateCappedSize(OperationContext* opCtx, NamespaceString ns, long long size);
+
+    void updateTTLSetting(OperationContext* opCtx,
+                          NamespaceString ns,
+                          StringData idxName,
+                          long long newExpireSeconds);
+
+    bool isEqualToMetadataUUID(OperationContext* opCtx,
+                               NamespaceString ns,
+                               OptionalCollectionUUID uuid);
+
+    void setIsTemp(OperationContext* opCtx, NamespaceString ns, bool isTemp);
+
+    boost::optional<std::string> getSideWritesIdent(OperationContext* opCtx,
+                                                    NamespaceString ns,
+                                                    StringData indexName) const;
+
+    // TODO SERVER-36385 Remove this function: we don't set the feature tracker bit in 4.4 because
+    // 4.4 can only downgrade to 4.2 which can read long TypeBits.
+    void setIndexKeyStringWithLongTypeBitsExistsOnDisk(OperationContext* opCtx);
+
+    void updateValidator(OperationContext* opCtx,
+                         NamespaceString ns,
+                         const BSONObj& validator,
+                         StringData validationLevel,
+                         StringData validationAction);
+
+    void updateIndexMetadata(OperationContext* opCtx,
+                             NamespaceString ns,
+                             const IndexDescriptor* desc);
+
+    Status removeIndex(OperationContext* opCtx, NamespaceString ns, StringData indexName);
+
+    Status prepareForIndexBuild(OperationContext* opCtx,
+                                NamespaceString ns,
+                                const IndexDescriptor* spec,
+                                IndexBuildProtocol indexBuildProtocol,
+                                bool isBackgroundSecondaryBuild);
+
+    bool isTwoPhaseIndexBuild(OperationContext* opCtx,
+                              NamespaceString ns,
+                              StringData indexName) const;
+
+    void setIndexBuildScanning(OperationContext* opCtx,
+                               NamespaceString ns,
+                               StringData indexName,
+                               std::string sideWritesIdent,
+                               boost::optional<std::string> constraintViolationsIdent);
+
+
+    bool isIndexBuildScanning(OperationContext* opCtx,
+                              NamespaceString ns,
+                              StringData indexName) const;
+
+    void setIndexBuildDraining(OperationContext* opCtx, NamespaceString ns, StringData indexName);
+
+    bool isIndexBuildDraining(OperationContext* opCtx,
+                              NamespaceString ns,
+                              StringData indexName) const;
+
+    void indexBuildSuccess(OperationContext* opCtx, NamespaceString ns, StringData indexName);
+
+    bool isIndexMultikey(OperationContext* opCtx,
+                         NamespaceString ns,
+                         StringData indexName,
+                         MultikeyPaths* multikeyPaths) const;
+
+    bool setIndexIsMultikey(OperationContext* opCtx,
+                            NamespaceString ns,
+                            StringData indexName,
+                            const MultikeyPaths& multikeyPaths);
+
+    boost::optional<std::string> getConstraintViolationsIdent(OperationContext* opCtx,
+                                                              NamespaceString ns,
+                                                              StringData indexName) const;
+
+    long getIndexBuildVersion(OperationContext* opCtx,
+                              NamespaceString ns,
+                              StringData indexName) const;
+
+    CollectionOptions getCollectionOptions(OperationContext* opCtx, NamespaceString ns) const;
+
+    int getTotalIndexCount(OperationContext* opCtx, NamespaceString ns) const;
+
+    int getCompletedIndexCount(OperationContext* opCtx, NamespaceString ns) const;
+
+    BSONObj getIndexSpec(OperationContext* opCtx, NamespaceString ns, StringData indexName) const;
+
+    void getAllIndexes(OperationContext* opCtx,
+                       NamespaceString ns,
+                       std::vector<std::string>* names) const;
+
+    void getReadyIndexes(OperationContext* opCtx,
+                         NamespaceString ns,
+                         std::vector<std::string>* names) const;
+    void getAllUniqueIndexes(OperationContext* opCtx,
+                             NamespaceString ns,
+                             std::vector<std::string>* names) const;
+
+    bool isIndexPresent(OperationContext* opCtx, NamespaceString ns, StringData indexName) const;
+
+    bool isIndexReady(OperationContext* opCtx, NamespaceString ns, StringData indexName) const;
+
+    KVPrefix getIndexPrefix(OperationContext* opCtx,
+                            NamespaceString ns,
+                            StringData indexName) const;
+
 private:
     class AddIdentChange;
     class RemoveIdentChange;
+    class AddIndexChange;
+    class RemoveIndexChange;
 
     friend class StorageEngineImpl;
     friend class KVCatalogTest;
diff --git a/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp b/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp
index 43150b4d6b3..d971924cadd 100644
--- a/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp
+++ b/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp
@@ -37,7 +37,7 @@
 
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/index/index_descriptor.h"
-#include "mongo/db/storage/kv/kv_catalog.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/kv/kv_catalog_feature_tracker.h"
 #include "mongo/db/storage/kv/kv_engine.h"
 #include "mongo/db/storage/kv/storage_engine_impl.h"
@@ -47,351 +47,12 @@ namespace mongo {
 
 using std::string;
 
-namespace {
-
-bool indexTypeSupportsPathLevelMultikeyTracking(StringData accessMethod) {
-    return accessMethod == IndexNames::BTREE || accessMethod == IndexNames::GEO_2DSPHERE;
-}
-
-}  // namespace
-
-class KVCollectionCatalogEntry::AddIndexChange : public RecoveryUnit::Change {
-public:
-    AddIndexChange(OperationContext* opCtx, KVCollectionCatalogEntry* cce, StringData ident)
-        : _opCtx(opCtx), _cce(cce), _ident(ident.toString()) {}
-
-    virtual void commit(boost::optional<Timestamp>) {}
-    virtual void rollback() {
-        // Intentionally ignoring failure.
-        auto kvEngine = _cce->_engine->getEngine();
-        MONGO_COMPILER_VARIABLE_UNUSED auto status = kvEngine->dropIdent(_opCtx, _ident);
-    }
-
-    OperationContext* const _opCtx;
-    KVCollectionCatalogEntry* const _cce;
-    const std::string _ident;
-};
-
-class KVCollectionCatalogEntry::RemoveIndexChange : public RecoveryUnit::Change {
-public:
-    RemoveIndexChange(OperationContext* opCtx,
-                      KVCollectionCatalogEntry* cce,
-                      OptionalCollectionUUID uuid,
-                      const NamespaceString& indexNss,
-                      StringData indexName,
-                      StringData ident)
-        : _opCtx(opCtx),
-          _cce(cce),
-          _uuid(uuid),
-          _indexNss(indexNss),
-          _indexName(indexName),
-          _ident(ident.toString()) {}
-
-    virtual void rollback() {}
-    virtual void commit(boost::optional<Timestamp> commitTimestamp) {
-        // Intentionally ignoring failure here. Since we've removed the metadata pointing to the
-        // index, we should never see it again anyway.
-        auto engine = _cce->_engine;
-        auto storageEngine = engine->getStorageEngine();
-        if (storageEngine->supportsPendingDrops() && commitTimestamp) {
-            log() << "Deferring table drop for index '" << _indexName << "' on collection '"
-                  << _indexNss << (_uuid ? " (" + _uuid->toString() + ")'" : "") << ". Ident: '"
-                  << _ident << "', commit timestamp: '" << commitTimestamp << "'";
-            engine->addDropPendingIdent(*commitTimestamp, _indexNss, _ident);
-        } else {
-            auto kvEngine = engine->getEngine();
-            MONGO_COMPILER_VARIABLE_UNUSED auto status = kvEngine->dropIdent(_opCtx, _ident);
-        }
-    }
-
-    OperationContext* const _opCtx;
-    KVCollectionCatalogEntry* const _cce;
-    OptionalCollectionUUID _uuid;
-    const NamespaceString _indexNss;
-    const std::string _indexName;
-    const std::string _ident;
-};
-
-
 KVCollectionCatalogEntry::KVCollectionCatalogEntry(StorageEngineInterface* engine,
-                                                   KVCatalog* catalog,
                                                    StringData ns,
                                                    StringData ident,
                                                    std::unique_ptr<RecordStore> rs)
     : BSONCollectionCatalogEntry(ns),
       _engine(engine),
-      _catalog(catalog),
       _ident(ident.toString()),
       _recordStore(std::move(rs)) {}
-
-KVCollectionCatalogEntry::~KVCollectionCatalogEntry() {}
-
-bool KVCollectionCatalogEntry::setIndexIsMultikey(OperationContext* opCtx,
-                                                  StringData indexName,
-                                                  const MultikeyPaths& multikeyPaths) {
-    MetaData md = _getMetaData(opCtx);
-
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-
-    const bool tracksPathLevelMultikeyInfo = !md.indexes[offset].multikeyPaths.empty();
-    if (tracksPathLevelMultikeyInfo) {
-        invariant(!multikeyPaths.empty());
-        invariant(multikeyPaths.size() == md.indexes[offset].multikeyPaths.size());
-    } else {
-        invariant(multikeyPaths.empty());
-
-        if (md.indexes[offset].multikey) {
-            // The index is already set as multikey and we aren't tracking path-level multikey
-            // information for it. We return false to indicate that the index metadata is unchanged.
-            return false;
-        }
-    }
-
-    md.indexes[offset].multikey = true;
-
-    if (tracksPathLevelMultikeyInfo) {
-        bool newPathIsMultikey = false;
-        bool somePathIsMultikey = false;
-
-        // Store new path components that cause this index to be multikey in catalog's index
-        // metadata.
-        for (size_t i = 0; i < multikeyPaths.size(); ++i) {
-            std::set<size_t>& indexMultikeyComponents = md.indexes[offset].multikeyPaths[i];
-            for (const auto multikeyComponent : multikeyPaths[i]) {
-                auto result = indexMultikeyComponents.insert(multikeyComponent);
-                newPathIsMultikey = newPathIsMultikey || result.second;
-                somePathIsMultikey = true;
-            }
-        }
-
-        // If all of the sets in the multikey paths vector were empty, then no component of any
-        // indexed field caused the index to be multikey. setIndexIsMultikey() therefore shouldn't
-        // have been called.
-        invariant(somePathIsMultikey);
-
-        if (!newPathIsMultikey) {
-            // We return false to indicate that the index metadata is unchanged.
-            return false;
-        }
-    }
-
-    _catalog->putMetaData(opCtx, ns(), md);
-    return true;
-}
-
-void KVCollectionCatalogEntry::setIndexKeyStringWithLongTypeBitsExistsOnDisk(
-    OperationContext* opCtx) {
-    const auto feature =
-        KVCatalog::FeatureTracker::RepairableFeature::kIndexKeyStringWithLongTypeBits;
-    if (!_catalog->getFeatureTracker()->isRepairableFeatureInUse(opCtx, feature)) {
-        _catalog->getFeatureTracker()->markRepairableFeatureAsInUse(opCtx, feature);
-    }
-}
-
-Status KVCollectionCatalogEntry::removeIndex(OperationContext* opCtx, StringData indexName) {
-    MetaData md = _getMetaData(opCtx);
-
-    if (md.findIndexOffset(indexName) < 0)
-        return Status::OK();  // never had the index so nothing to do.
-
-    const string ident = _catalog->getIndexIdent(opCtx, ns(), indexName);
-
-    md.eraseIndex(indexName);
-    _catalog->putMetaData(opCtx, ns(), md);
-
-    // Lazily remove to isolate underlying engine from rollback.
-    opCtx->recoveryUnit()->registerChange(new RemoveIndexChange(
-        opCtx, this, md.options.uuid, ns().makeIndexNamespace(indexName), indexName, ident));
-    return Status::OK();
-}
-
-Status KVCollectionCatalogEntry::prepareForIndexBuild(OperationContext* opCtx,
-                                                      const IndexDescriptor* spec,
-                                                      IndexBuildProtocol indexBuildProtocol,
-                                                      bool isBackgroundSecondaryBuild) {
-    MetaData md = _getMetaData(opCtx);
-
-    KVPrefix prefix = KVPrefix::getNextPrefix(ns());
-    IndexMetaData imd;
-    imd.spec = spec->infoObj();
-    imd.ready = false;
-    imd.multikey = false;
-    imd.prefix = prefix;
-    imd.isBackgroundSecondaryBuild = isBackgroundSecondaryBuild;
-    imd.runTwoPhaseBuild = indexBuildProtocol == IndexBuildProtocol::kTwoPhase;
-
-    if (indexTypeSupportsPathLevelMultikeyTracking(spec->getAccessMethodName())) {
-        const auto feature =
-            KVCatalog::FeatureTracker::RepairableFeature::kPathLevelMultikeyTracking;
-        if (!_catalog->getFeatureTracker()->isRepairableFeatureInUse(opCtx, feature)) {
-            _catalog->getFeatureTracker()->markRepairableFeatureAsInUse(opCtx, feature);
-        }
-        imd.multikeyPaths = MultikeyPaths{static_cast<size_t>(spec->keyPattern().nFields())};
-    }
-
-    // Mark collation feature as in use if the index has a non-simple collation.
-    if (imd.spec["collation"]) {
-        const auto feature = KVCatalog::FeatureTracker::NonRepairableFeature::kCollation;
-        if (!_catalog->getFeatureTracker()->isNonRepairableFeatureInUse(opCtx, feature)) {
-            _catalog->getFeatureTracker()->markNonRepairableFeatureAsInUse(opCtx, feature);
-        }
-    }
-
-    md.indexes.push_back(imd);
-    _catalog->putMetaData(opCtx, ns(), md);
-
-    string ident = _catalog->getIndexIdent(opCtx, ns(), spec->indexName());
-
-    auto kvEngine = _engine->getEngine();
-    const Status status = kvEngine->createGroupedSortedDataInterface(opCtx, ident, spec, prefix);
-    if (status.isOK()) {
-        opCtx->recoveryUnit()->registerChange(new AddIndexChange(opCtx, this, ident));
-    }
-
-    return status;
-}
-
-bool KVCollectionCatalogEntry::isTwoPhaseIndexBuild(OperationContext* opCtx,
-                                                    StringData indexName) const {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    return md.indexes[offset].runTwoPhaseBuild;
-}
-
-long KVCollectionCatalogEntry::getIndexBuildVersion(OperationContext* opCtx,
-                                                    StringData indexName) const {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    return md.indexes[offset].versionOfBuild;
-}
-
-void KVCollectionCatalogEntry::setIndexBuildScanning(
-    OperationContext* opCtx,
-    StringData indexName,
-    std::string sideWritesIdent,
-    boost::optional<std::string> constraintViolationsIdent) {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    invariant(!md.indexes[offset].ready);
-    invariant(!md.indexes[offset].buildPhase);
-    invariant(md.indexes[offset].runTwoPhaseBuild);
-
-    md.indexes[offset].buildPhase = kIndexBuildScanning.toString();
-    md.indexes[offset].sideWritesIdent = sideWritesIdent;
-    md.indexes[offset].constraintViolationsIdent = constraintViolationsIdent;
-    _catalog->putMetaData(opCtx, ns(), md);
-}
-
-bool KVCollectionCatalogEntry::isIndexBuildScanning(OperationContext* opCtx,
-                                                    StringData indexName) const {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    return md.indexes[offset].buildPhase == kIndexBuildScanning.toString();
-}
-
-void KVCollectionCatalogEntry::setIndexBuildDraining(OperationContext* opCtx,
-                                                     StringData indexName) {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    invariant(!md.indexes[offset].ready);
-    invariant(md.indexes[offset].runTwoPhaseBuild);
-    invariant(md.indexes[offset].buildPhase == kIndexBuildScanning.toString());
-
-    md.indexes[offset].buildPhase = kIndexBuildDraining.toString();
-    _catalog->putMetaData(opCtx, ns(), md);
-}
-
-bool KVCollectionCatalogEntry::isIndexBuildDraining(OperationContext* opCtx,
-                                                    StringData indexName) const {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    return md.indexes[offset].buildPhase == kIndexBuildDraining.toString();
-}
-
-void KVCollectionCatalogEntry::indexBuildSuccess(OperationContext* opCtx, StringData indexName) {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    md.indexes[offset].ready = true;
-    md.indexes[offset].runTwoPhaseBuild = false;
-    md.indexes[offset].buildPhase = boost::none;
-    md.indexes[offset].sideWritesIdent = boost::none;
-    md.indexes[offset].constraintViolationsIdent = boost::none;
-    _catalog->putMetaData(opCtx, ns(), md);
-}
-
-boost::optional<std::string> KVCollectionCatalogEntry::getSideWritesIdent(
-    OperationContext* opCtx, StringData indexName) const {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    return md.indexes[offset].sideWritesIdent;
-}
-
-boost::optional<std::string> KVCollectionCatalogEntry::getConstraintViolationsIdent(
-    OperationContext* opCtx, StringData indexName) const {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    return md.indexes[offset].constraintViolationsIdent;
-}
-
-void KVCollectionCatalogEntry::updateTTLSetting(OperationContext* opCtx,
-                                                StringData idxName,
-                                                long long newExpireSeconds) {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(idxName);
-    invariant(offset >= 0);
-    md.indexes[offset].updateTTLSetting(newExpireSeconds);
-    _catalog->putMetaData(opCtx, ns(), md);
-}
-
-void KVCollectionCatalogEntry::updateIndexMetadata(OperationContext* opCtx,
-                                                   const IndexDescriptor* desc) {
-    // Update any metadata Ident has for this index
-    const string ident = _catalog->getIndexIdent(opCtx, ns(), desc->indexName());
-    auto kvEngine = _engine->getEngine();
-    kvEngine->alterIdentMetadata(opCtx, ident, desc);
-}
-
-bool KVCollectionCatalogEntry::isEqualToMetadataUUID(OperationContext* opCtx,
-                                                     OptionalCollectionUUID uuid) {
-    MetaData md = _getMetaData(opCtx);
-    return md.options.uuid && md.options.uuid == uuid;
-}
-
-void KVCollectionCatalogEntry::updateValidator(OperationContext* opCtx,
-                                               const BSONObj& validator,
-                                               StringData validationLevel,
-                                               StringData validationAction) {
-    MetaData md = _getMetaData(opCtx);
-    md.options.validator = validator;
-    md.options.validationLevel = validationLevel.toString();
-    md.options.validationAction = validationAction.toString();
-    _catalog->putMetaData(opCtx, ns(), md);
-}
-
-void KVCollectionCatalogEntry::setIsTemp(OperationContext* opCtx, bool isTemp) {
-    MetaData md = _getMetaData(opCtx);
-    md.options.temp = isTemp;
-    _catalog->putMetaData(opCtx, ns(), md);
-}
-
-void KVCollectionCatalogEntry::updateCappedSize(OperationContext* opCtx, long long size) {
-    MetaData md = _getMetaData(opCtx);
-    md.options.cappedSize = size;
-    _catalog->putMetaData(opCtx, ns(), md);
-}
-
-BSONCollectionCatalogEntry::MetaData KVCollectionCatalogEntry::_getMetaData(
-    OperationContext* opCtx) const {
-    return _catalog->getMetaData(opCtx, ns());
-}
 }
diff --git a/src/mongo/db/storage/kv/kv_collection_catalog_entry.h b/src/mongo/db/storage/kv/kv_collection_catalog_entry.h
index 644a00f8943..e9133a607f5 100644
--- a/src/mongo/db/storage/kv/kv_collection_catalog_entry.h
+++ b/src/mongo/db/storage/kv/kv_collection_catalog_entry.h
@@ -38,76 +38,16 @@
 
 namespace mongo {
 
-class KVCatalog;
 class StorageEngineInterface;
 
 class KVCollectionCatalogEntry final : public BSONCollectionCatalogEntry {
 public:
     KVCollectionCatalogEntry(StorageEngineInterface* engine,
-                             KVCatalog* catalog,
                              StringData ns,
                              StringData ident,
                              std::unique_ptr<RecordStore> rs);
 
-    ~KVCollectionCatalogEntry() final;
-
-    int getMaxAllowedIndexes() const final {
-        return 64;
-    };
-
-    bool setIndexIsMultikey(OperationContext* opCtx,
-                            StringData indexName,
-                            const MultikeyPaths& multikeyPaths) final;
-
-    // TODO SERVER-36385 Remove this function: we don't set the feature tracker bit in 4.4 because
-    // 4.4 can only downgrade to 4.2 which can read long TypeBits.
-    void setIndexKeyStringWithLongTypeBitsExistsOnDisk(OperationContext* opCtx) final;
-
-    Status removeIndex(OperationContext* opCtx, StringData indexName) final;
-
-    Status prepareForIndexBuild(OperationContext* opCtx,
-                                const IndexDescriptor* spec,
-                                IndexBuildProtocol indexBuildProtocol,
-                                bool isBackgroundSecondaryBuild) final;
-
-    bool isTwoPhaseIndexBuild(OperationContext* opCtx, StringData indexName) const final;
-
-    long getIndexBuildVersion(OperationContext* opCtx, StringData indexName) const final;
-
-    void setIndexBuildScanning(OperationContext* opCtx,
-                               StringData indexName,
-                               std::string sideWritesIdent,
-                               boost::optional<std::string> constraintViolationsIdent) final;
-
-    bool isIndexBuildScanning(OperationContext* opCtx, StringData indexName) const final;
-
-    void setIndexBuildDraining(OperationContext* opCtx, StringData indexName) final;
-
-    bool isIndexBuildDraining(OperationContext* opCtx, StringData indexName) const final;
-
-    void indexBuildSuccess(OperationContext* opCtx, StringData indexName) final;
-
-    boost::optional<std::string> getSideWritesIdent(OperationContext* opCtx,
-                                                    StringData indexName) const final;
-
-    boost::optional<std::string> getConstraintViolationsIdent(OperationContext* opCtx,
-                                                              StringData indexName) const final;
-    void updateTTLSetting(OperationContext* opCtx,
-                          StringData idxName,
-                          long long newExpireSeconds) final;
-
-    void updateIndexMetadata(OperationContext* opCtx, const IndexDescriptor* desc) final;
-
-    void updateValidator(OperationContext* opCtx,
-                         const BSONObj& validator,
-                         StringData validationLevel,
-                         StringData validationAction) final;
-
-    void setIsTemp(OperationContext* opCtx, bool isTemp);
-
-    void updateCappedSize(OperationContext*, long long int) final;
-
-    bool isEqualToMetadataUUID(OperationContext* opCtx, OptionalCollectionUUID uuid) final;
+    ~KVCollectionCatalogEntry() final {}
 
     RecordStore* getRecordStore() {
         return _recordStore.get();
@@ -116,15 +56,8 @@ public:
         return _recordStore.get();
     }
 
-protected:
-    MetaData _getMetaData(OperationContext* opCtx) const final;
-
 private:
-    class AddIndexChange;
-    class RemoveIndexChange;
-
     StorageEngineInterface* const _engine;  // not owned
-    KVCatalog* _catalog;                    // not owned
     std::string _ident;
     std::unique_ptr<RecordStore> _recordStore;  // owned
 };
diff --git a/src/mongo/db/storage/kv/kv_collection_catalog_entry_test.cpp b/src/mongo/db/storage/kv/kv_collection_catalog_entry_test.cpp
index 5c1aa53aef3..67cc5ed1697 100644
--- a/src/mongo/db/storage/kv/kv_collection_catalog_entry_test.cpp
+++ b/src/mongo/db/storage/kv/kv_collection_catalog_entry_test.cpp
@@ -96,9 +96,12 @@ public:
         }
     }
 
-    CollectionCatalogEntry* getCollectionCatalogEntry() {
-        return CollectionCatalog::get(getGlobalServiceContext())
-            .lookupCollectionCatalogEntryByNamespace(_nss);
+    NamespaceString ns() {
+        return _nss;
+    }
+
+    DurableCatalog* getCatalog() {
+        return _storageEngine.getCatalog();
     }
 
     std::string createIndex(BSONObj keyPattern,
@@ -116,8 +119,8 @@ public:
         {
             WriteUnitOfWork wuow(opCtx.get());
             const bool isSecondaryBackgroundIndexBuild = false;
-            ASSERT_OK(getCollectionCatalogEntry()->prepareForIndexBuild(
-                opCtx.get(), &desc, protocol, isSecondaryBackgroundIndexBuild));
+            ASSERT_OK(_storageEngine.getCatalog()->prepareForIndexBuild(
+                opCtx.get(), _nss, &desc, protocol, isSecondaryBackgroundIndexBuild));
             wuow.commit();
         }
 
@@ -159,106 +162,100 @@ private:
 
 TEST_F(KVCollectionCatalogEntryTest, MultikeyPathsForBtreeIndexInitializedToVectorOfEmptySets) {
     std::string indexName = createIndex(BSON("a" << 1 << "b" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
+    DurableCatalog* catalog = getCatalog();
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(!collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(!catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {std::set<size_t>{}, std::set<size_t>{}});
     }
 }
 
 TEST_F(KVCollectionCatalogEntryTest, CanSetIndividualPathComponentOfBtreeIndexAsMultikey) {
     std::string indexName = createIndex(BSON("a" << 1 << "b" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    ASSERT(collEntry->setIndexIsMultikey(opCtx.get(), indexName, {std::set<size_t>{}, {0U}}));
+    DurableCatalog* catalog = getCatalog();
+    ASSERT(catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, {std::set<size_t>{}, {0U}}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {std::set<size_t>{}, {0U}});
     }
 }
 
 TEST_F(KVCollectionCatalogEntryTest, MultikeyPathsAccumulateOnDifferentFields) {
     std::string indexName = createIndex(BSON("a" << 1 << "b" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    ASSERT(collEntry->setIndexIsMultikey(opCtx.get(), indexName, {std::set<size_t>{}, {0U}}));
+    DurableCatalog* catalog = getCatalog();
+    ASSERT(catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, {std::set<size_t>{}, {0U}}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {std::set<size_t>{}, {0U}});
     }
 
-    ASSERT(collEntry->setIndexIsMultikey(opCtx.get(), indexName, {{0U}, std::set<size_t>{}}));
+    ASSERT(catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, {{0U}, std::set<size_t>{}}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {{0U}, {0U}});
     }
 }
 
 TEST_F(KVCollectionCatalogEntryTest, MultikeyPathsAccumulateOnDifferentComponentsOfTheSameField) {
     std::string indexName = createIndex(BSON("a.b" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    ASSERT(collEntry->setIndexIsMultikey(opCtx.get(), indexName, {{0U}}));
+    DurableCatalog* catalog = getCatalog();
+    ASSERT(catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, {{0U}}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {{0U}});
     }
 
-    ASSERT(collEntry->setIndexIsMultikey(opCtx.get(), indexName, {{1U}}));
+    ASSERT(catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, {{1U}}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {{0U, 1U}});
     }
 }
 
 TEST_F(KVCollectionCatalogEntryTest, NoOpWhenSpecifiedPathComponentsAlreadySetAsMultikey) {
     std::string indexName = createIndex(BSON("a" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    ASSERT(collEntry->setIndexIsMultikey(opCtx.get(), indexName, {{0U}}));
+    DurableCatalog* catalog = getCatalog();
+    ASSERT(catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, {{0U}}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {{0U}});
     }
 
-    ASSERT(!collEntry->setIndexIsMultikey(opCtx.get(), indexName, {{0U}}));
+    ASSERT(!catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, {{0U}}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {{0U}});
     }
 }
 
 TEST_F(KVCollectionCatalogEntryTest, CanSetMultipleFieldsAndComponentsAsMultikey) {
     std::string indexName = createIndex(BSON("a.b.c" << 1 << "a.b.d" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    ASSERT(collEntry->setIndexIsMultikey(opCtx.get(), indexName, {{0U, 1U}, {0U, 1U}}));
+    DurableCatalog* catalog = getCatalog();
+    ASSERT(catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, {{0U, 1U}, {0U, 1U}}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {{0U, 1U}, {0U, 1U}});
     }
 }
@@ -267,31 +264,29 @@ DEATH_TEST_F(KVCollectionCatalogEntryTest,
              CannotOmitPathLevelMultikeyInfoWithBtreeIndex,
              "Invariant failure !multikeyPaths.empty()") {
     std::string indexName = createIndex(BSON("a" << 1 << "b" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    collEntry->setIndexIsMultikey(opCtx.get(), indexName, MultikeyPaths{});
+    DurableCatalog* catalog = getCatalog();
+    catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, MultikeyPaths{});
 }
 
 DEATH_TEST_F(KVCollectionCatalogEntryTest,
              AtLeastOnePathComponentMustCauseIndexToBeMultikey,
              "Invariant failure somePathIsMultikey") {
     std::string indexName = createIndex(BSON("a" << 1 << "b" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    collEntry->setIndexIsMultikey(opCtx.get(), indexName, {std::set<size_t>{}, std::set<size_t>{}});
+    DurableCatalog* catalog = getCatalog();
+    catalog->setIndexIsMultikey(
+        opCtx.get(), ns(), indexName, {std::set<size_t>{}, std::set<size_t>{}});
 }
 
 TEST_F(KVCollectionCatalogEntryTest, PathLevelMultikeyTrackingIsSupportedBy2dsphereIndexes) {
     std::string indexType = IndexNames::GEO_2DSPHERE;
     std::string indexName = createIndex(BSON("a" << indexType << "b" << 1), indexType);
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
+    DurableCatalog* catalog = getCatalog();
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(!collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(!catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {std::set<size_t>{}, std::set<size_t>{}});
     }
 }
@@ -302,12 +297,11 @@ TEST_F(KVCollectionCatalogEntryTest, PathLevelMultikeyTrackingIsNotSupportedByAl
 
     for (auto&& indexType : indexTypes) {
         std::string indexName = createIndex(BSON("a" << indexType << "b" << 1), indexType);
-        CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
         auto opCtx = newOperationContext();
+        DurableCatalog* catalog = getCatalog();
         {
             MultikeyPaths multikeyPaths;
-            ASSERT(!collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+            ASSERT(!catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
             ASSERT(multikeyPaths.empty());
         }
     }
@@ -316,14 +310,13 @@ TEST_F(KVCollectionCatalogEntryTest, PathLevelMultikeyTrackingIsNotSupportedByAl
 TEST_F(KVCollectionCatalogEntryTest, CanSetEntireTextIndexAsMultikey) {
     std::string indexType = IndexNames::TEXT;
     std::string indexName = createIndex(BSON("a" << indexType << "b" << 1), indexType);
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    ASSERT(collEntry->setIndexIsMultikey(opCtx.get(), indexName, MultikeyPaths{}));
+    DurableCatalog* catalog = getCatalog();
+    ASSERT(catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, MultikeyPaths{}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         ASSERT(multikeyPaths.empty());
     }
 }
@@ -331,119 +324,115 @@ TEST_F(KVCollectionCatalogEntryTest, CanSetEntireTextIndexAsMultikey) {
 TEST_F(KVCollectionCatalogEntryTest, NoOpWhenEntireIndexAlreadySetAsMultikey) {
     std::string indexType = IndexNames::TEXT;
     std::string indexName = createIndex(BSON("a" << indexType << "b" << 1), indexType);
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    ASSERT(collEntry->setIndexIsMultikey(opCtx.get(), indexName, MultikeyPaths{}));
+    DurableCatalog* catalog = getCatalog();
+    ASSERT(catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, MultikeyPaths{}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         ASSERT(multikeyPaths.empty());
     }
 
-    ASSERT(!collEntry->setIndexIsMultikey(opCtx.get(), indexName, MultikeyPaths{}));
+    ASSERT(!catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, MultikeyPaths{}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         ASSERT(multikeyPaths.empty());
     }
 }
 
 TEST_F(KVCollectionCatalogEntryTest, SinglePhaseIndexBuild) {
     std::string indexName = createIndex(BSON("a" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-
-    ASSERT_EQ(kExpectedVersion, collEntry->getIndexBuildVersion(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexReady(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isTwoPhaseIndexBuild(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildScanning(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildDraining(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->getSideWritesIdent(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->getConstraintViolationsIdent(opCtx.get(), indexName));
-
-    collEntry->indexBuildSuccess(opCtx.get(), indexName);
-
-    ASSERT_EQ(kExpectedVersion, collEntry->getIndexBuildVersion(opCtx.get(), indexName));
-    ASSERT_TRUE(collEntry->isIndexReady(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isTwoPhaseIndexBuild(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildScanning(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildDraining(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->getSideWritesIdent(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->getConstraintViolationsIdent(opCtx.get(), indexName));
+    DurableCatalog* catalog = getCatalog();
+
+    ASSERT_EQ(kExpectedVersion, catalog->getIndexBuildVersion(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexReady(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isTwoPhaseIndexBuild(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildScanning(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildDraining(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->getSideWritesIdent(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->getConstraintViolationsIdent(opCtx.get(), ns(), indexName));
+
+    catalog->indexBuildSuccess(opCtx.get(), ns(), indexName);
+
+    ASSERT_EQ(kExpectedVersion, catalog->getIndexBuildVersion(opCtx.get(), ns(), indexName));
+    ASSERT_TRUE(catalog->isIndexReady(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isTwoPhaseIndexBuild(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildScanning(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildDraining(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->getSideWritesIdent(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->getConstraintViolationsIdent(opCtx.get(), ns(), indexName));
 }
 
 TEST_F(KVCollectionCatalogEntryTest, TwoPhaseIndexBuild) {
     std::string indexName =
         createIndex(BSON("a" << 1), IndexNames::BTREE, IndexBuildProtocol::kTwoPhase);
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-
-    ASSERT_EQ(kExpectedVersion, collEntry->getIndexBuildVersion(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexReady(opCtx.get(), indexName));
-    ASSERT_TRUE(collEntry->isTwoPhaseIndexBuild(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildScanning(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildDraining(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->getSideWritesIdent(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->getConstraintViolationsIdent(opCtx.get(), indexName));
-
-    collEntry->setIndexBuildScanning(
-        opCtx.get(), indexName, kSideWritesTableIdent, kConstraintViolationsTableIdent);
-
-    ASSERT_EQ(kExpectedVersion, collEntry->getIndexBuildVersion(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexReady(opCtx.get(), indexName));
-    ASSERT_TRUE(collEntry->isTwoPhaseIndexBuild(opCtx.get(), indexName));
-    ASSERT_TRUE(collEntry->isIndexBuildScanning(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildDraining(opCtx.get(), indexName));
-    ASSERT_EQ(kSideWritesTableIdent, collEntry->getSideWritesIdent(opCtx.get(), indexName));
+    DurableCatalog* catalog = getCatalog();
+
+    ASSERT_EQ(kExpectedVersion, catalog->getIndexBuildVersion(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexReady(opCtx.get(), ns(), indexName));
+    ASSERT_TRUE(catalog->isTwoPhaseIndexBuild(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildScanning(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildDraining(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->getSideWritesIdent(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->getConstraintViolationsIdent(opCtx.get(), ns(), indexName));
+
+    catalog->setIndexBuildScanning(
+        opCtx.get(), ns(), indexName, kSideWritesTableIdent, kConstraintViolationsTableIdent);
+
+    ASSERT_EQ(kExpectedVersion, catalog->getIndexBuildVersion(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexReady(opCtx.get(), ns(), indexName));
+    ASSERT_TRUE(catalog->isTwoPhaseIndexBuild(opCtx.get(), ns(), indexName));
+    ASSERT_TRUE(catalog->isIndexBuildScanning(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildDraining(opCtx.get(), ns(), indexName));
+    ASSERT_EQ(kSideWritesTableIdent, catalog->getSideWritesIdent(opCtx.get(), ns(), indexName));
     ASSERT_EQ(kConstraintViolationsTableIdent,
-              collEntry->getConstraintViolationsIdent(opCtx.get(), indexName));
+              catalog->getConstraintViolationsIdent(opCtx.get(), ns(), indexName));
 
-    collEntry->setIndexBuildDraining(opCtx.get(), indexName);
+    catalog->setIndexBuildDraining(opCtx.get(), ns(), indexName);
 
-    ASSERT_EQ(kExpectedVersion, collEntry->getIndexBuildVersion(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexReady(opCtx.get(), indexName));
-    ASSERT_TRUE(collEntry->isTwoPhaseIndexBuild(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildScanning(opCtx.get(), indexName));
-    ASSERT_TRUE(collEntry->isIndexBuildDraining(opCtx.get(), indexName));
-    ASSERT_EQ(kSideWritesTableIdent, collEntry->getSideWritesIdent(opCtx.get(), indexName));
+    ASSERT_EQ(kExpectedVersion, catalog->getIndexBuildVersion(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexReady(opCtx.get(), ns(), indexName));
+    ASSERT_TRUE(catalog->isTwoPhaseIndexBuild(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildScanning(opCtx.get(), ns(), indexName));
+    ASSERT_TRUE(catalog->isIndexBuildDraining(opCtx.get(), ns(), indexName));
+    ASSERT_EQ(kSideWritesTableIdent, catalog->getSideWritesIdent(opCtx.get(), ns(), indexName));
     ASSERT_EQ(kConstraintViolationsTableIdent,
-              collEntry->getConstraintViolationsIdent(opCtx.get(), indexName));
+              catalog->getConstraintViolationsIdent(opCtx.get(), ns(), indexName));
 
-    collEntry->indexBuildSuccess(opCtx.get(), indexName);
+    catalog->indexBuildSuccess(opCtx.get(), ns(), indexName);
 
-    ASSERT_EQ(kExpectedVersion, collEntry->getIndexBuildVersion(opCtx.get(), indexName));
-    ASSERT(collEntry->isIndexReady(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildScanning(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildDraining(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isTwoPhaseIndexBuild(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->getSideWritesIdent(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->getConstraintViolationsIdent(opCtx.get(), indexName));
+    ASSERT_EQ(kExpectedVersion, catalog->getIndexBuildVersion(opCtx.get(), ns(), indexName));
+    ASSERT(catalog->isIndexReady(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildScanning(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildDraining(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isTwoPhaseIndexBuild(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->getSideWritesIdent(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->getConstraintViolationsIdent(opCtx.get(), ns(), indexName));
 }
 
 DEATH_TEST_F(KVCollectionCatalogEntryTest,
              SinglePhaseIllegalScanPhase,
              "Invariant failure md.indexes[offset].runTwoPhaseBuild") {
     std::string indexName = createIndex(BSON("a" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    collEntry->setIndexBuildScanning(
-        opCtx.get(), indexName, kSideWritesTableIdent, kConstraintViolationsTableIdent);
+    DurableCatalog* catalog = getCatalog();
+
+    catalog->setIndexBuildScanning(
+        opCtx.get(), ns(), indexName, kSideWritesTableIdent, kConstraintViolationsTableIdent);
 }
 
 DEATH_TEST_F(KVCollectionCatalogEntryTest,
              SinglePhaseIllegalDrainPhase,
              "Invariant failure md.indexes[offset].runTwoPhaseBuild") {
     std::string indexName = createIndex(BSON("a" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    collEntry->setIndexBuildDraining(opCtx.get(), indexName);
+    DurableCatalog* catalog = getCatalog();
+    catalog->setIndexBuildDraining(opCtx.get(), ns(), indexName);
 }
 
 DEATH_TEST_F(KVCollectionCatalogEntryTest,
@@ -451,10 +440,9 @@ DEATH_TEST_F(KVCollectionCatalogEntryTest,
              "Invariant failure multikeyPaths.empty()") {
     std::string indexType = IndexNames::TEXT;
     std::string indexName = createIndex(BSON("a" << indexType << "b" << 1), indexType);
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    collEntry->setIndexIsMultikey(opCtx.get(), indexName, {{0U}, {0U}});
+    DurableCatalog* catalog = getCatalog();
+    catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, {{0U}, {0U}});
 }
 
 }  // namespace
diff --git a/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper_test.cpp b/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper_test.cpp
index 064a5ee66e6..7356d68e685 100644
--- a/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper_test.cpp
+++ b/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper_test.cpp
@@ -86,6 +86,7 @@ public:
         return {};
     }
     Status createSortedDataInterface(OperationContext* opCtx,
+                                     const CollectionOptions& collOptions,
                                      StringData ident,
                                      const IndexDescriptor* desc) override {
         return Status::OK();
diff --git a/src/mongo/db/storage/kv/kv_engine.h b/src/mongo/db/storage/kv/kv_engine.h
index 5a6d6c80791..ab7dd5bc5e5 100644
--- a/src/mongo/db/storage/kv/kv_engine.h
+++ b/src/mongo/db/storage/kv/kv_engine.h
@@ -141,6 +141,7 @@ public:
     }
 
     virtual Status createSortedDataInterface(OperationContext* opCtx,
+                                             const CollectionOptions& collOptions,
                                              StringData ident,
                                              const IndexDescriptor* desc) = 0;
 
@@ -155,11 +156,12 @@ public:
      *        share a table. Sharing indexes belonging to different databases is forbidden.
      */
     virtual Status createGroupedSortedDataInterface(OperationContext* opCtx,
+                                                    const CollectionOptions& collOptions,
                                                     StringData ident,
                                                     const IndexDescriptor* desc,
                                                     KVPrefix prefix) {
         invariant(prefix == KVPrefix::kNotPrefixed);
-        return createSortedDataInterface(opCtx, ident, desc);
+        return createSortedDataInterface(opCtx, collOptions, ident, desc);
     }
 
     virtual int64_t getIdentSize(OperationContext* opCtx, StringData ident) = 0;
diff --git a/src/mongo/db/storage/kv/kv_engine_test_harness.cpp b/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
index 22cb8778836..0418fe9a597 100644
--- a/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
+++ b/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
@@ -33,7 +33,6 @@
 #include "mongo/db/catalog/collection_impl.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/operation_context_noop.h"
-#include "mongo/db/storage/kv/kv_catalog.h"
 #include "mongo/db/storage/kv/kv_catalog_test_fixture.h"
 #include "mongo/db/storage/kv/kv_engine.h"
 #include "mongo/db/storage/kv/kv_prefix.h"
@@ -181,7 +180,7 @@ TEST(KVEngineTestHarness, SimpleSorted1) {
     unique_ptr<SortedDataInterface> sorted;
     {
         MyOperationContext opCtx(engine);
-        ASSERT_OK(engine->createSortedDataInterface(&opCtx, ident, &desc));
+        ASSERT_OK(engine->createSortedDataInterface(&opCtx, CollectionOptions(), ident, &desc));
         sorted = engine->getSortedDataInterface(&opCtx, ident, &desc);
         ASSERT(sorted);
     }
@@ -724,7 +723,7 @@ DEATH_TEST_F(KVCatalogTest, TerminateOnNonNumericIndexVersion, "Fatal Assertion
     unique_ptr<SortedDataInterface> sorted;
     {
         MyOperationContext opCtx(engine);
-        ASSERT_OK(engine->createSortedDataInterface(&opCtx, ident, &desc));
+        ASSERT_OK(engine->createSortedDataInterface(&opCtx, CollectionOptions(), ident, &desc));
         sorted = engine->getSortedDataInterface(&opCtx, ident, &desc);
         ASSERT(sorted);
     }
diff --git a/src/mongo/db/storage/kv/storage_engine_impl.cpp b/src/mongo/db/storage/kv/storage_engine_impl.cpp
index f2734ced75f..bd57044c36e 100644
--- a/src/mongo/db/storage/kv/storage_engine_impl.cpp
+++ b/src/mongo/db/storage/kv/storage_engine_impl.cpp
@@ -92,7 +92,7 @@ void StorageEngineImpl::loadCatalog(OperationContext* opCtx) {
 
         if (status.code() == ErrorCodes::DataModifiedByRepair) {
             warning() << "Catalog data modified by repair: " << status.reason();
-            repairObserver->onModification(str::stream() << "KVCatalog repaired: "
+            repairObserver->onModification(str::stream() << "DurableCatalog repaired: "
                                                          << status.reason());
         } else {
             fassertNoTrace(50926, status);
@@ -239,10 +239,10 @@ void StorageEngineImpl::_initCollection(OperationContext* opCtx,
                                         const NamespaceString& nss,
                                         bool forRepair) {
     auto catalogEntry = _catalog->makeCollectionCatalogEntry(opCtx, nss, forRepair);
-    auto uuid = catalogEntry->getCollectionOptions(opCtx).uuid.get();
+    auto uuid = _catalog->getCollectionOptions(opCtx, nss).uuid.get();
 
     auto collectionFactory = Collection::Factory::get(getGlobalServiceContext());
-    auto collection = collectionFactory->make(opCtx, catalogEntry.get());
+    auto collection = collectionFactory->make(opCtx, uuid, catalogEntry.get());
 
     auto& collectionCatalog = CollectionCatalog::get(getGlobalServiceContext());
     collectionCatalog.registerCollection(uuid, std::move(catalogEntry), std::move(collection));
@@ -291,15 +291,15 @@ Status StorageEngineImpl::_recoverOrphanedCollection(OperationContext* opCtx,
 
 /**
  * This method reconciles differences between idents the KVEngine is aware of and the
- * KVCatalog. There are three differences to consider:
+ * DurableCatalog. There are three differences to consider:
  *
- * First, a KVEngine may know of an ident that the KVCatalog does not. This method will drop
+ * First, a KVEngine may know of an ident that the DurableCatalog does not. This method will drop
  * the ident from the KVEngine.
  *
- * Second, a KVCatalog may have a collection ident that the KVEngine does not. This is an
+ * Second, a DurableCatalog may have a collection ident that the KVEngine does not. This is an
  * illegal state and this method fasserts.
  *
- * Third, a KVCatalog may have an index ident that the KVEngine does not. This method will
+ * Third, a DurableCatalog may have an index ident that the KVEngine does not. This method will
  * rebuild the index.
  */
 StatusWith<std::vector<StorageEngine::CollectionIndexNamePair>>
@@ -968,7 +968,7 @@ int64_t StorageEngineImpl::sizeOnDiskForDb(OperationContext* opCtx, StringData d
             size += catalogEntry->getRecordStore()->storageSize(opCtx);
 
             std::vector<std::string> indexNames;
-            catalogEntry->getAllIndexes(opCtx, &indexNames);
+            _catalog->getAllIndexes(opCtx, collection->ns(), &indexNames);
 
             for (size_t i = 0; i < indexNames.size(); i++) {
                 std::string ident =
diff --git a/src/mongo/db/storage/kv/storage_engine_impl.h b/src/mongo/db/storage/kv/storage_engine_impl.h
index 6c6dd3fdf2c..3d267bcbeeb 100644
--- a/src/mongo/db/storage/kv/storage_engine_impl.h
+++ b/src/mongo/db/storage/kv/storage_engine_impl.h
@@ -38,8 +38,9 @@
 #include "mongo/base/string_data.h"
 #include "mongo/bson/timestamp.h"
 #include "mongo/db/namespace_string.h"
+#include "mongo/db/storage/durable_catalog.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/journal_listener.h"
-#include "mongo/db/storage/kv/kv_catalog.h"
 #include "mongo/db/storage/kv/kv_catalog_feature_tracker.h"
 #include "mongo/db/storage/kv/kv_drop_pending_ident_reaper.h"
 #include "mongo/db/storage/kv/storage_engine_interface.h"
@@ -308,10 +309,10 @@ public:
                              const NamespaceString& nss,
                              StringData ident) override;
 
-    KVCatalog* getCatalog() {
+    DurableCatalog* getCatalog() {
         return _catalog.get();
     }
-    const KVCatalog* getCatalog() const {
+    const DurableCatalog* getCatalog() const {
         return _catalog.get();
     }
 
diff --git a/src/mongo/db/storage/kv/storage_engine_interface.h b/src/mongo/db/storage/kv/storage_engine_interface.h
index 89ab71b83b2..db7201c1492 100644
--- a/src/mongo/db/storage/kv/storage_engine_interface.h
+++ b/src/mongo/db/storage/kv/storage_engine_interface.h
@@ -32,7 +32,7 @@
 namespace mongo {
 
 class KVEngine;
-class KVCatalog;
+class DurableCatalog;
 class StorageEngine;
 
 class StorageEngineInterface {
@@ -44,6 +44,6 @@ public:
     virtual void addDropPendingIdent(const Timestamp& dropTimestamp,
                                      const NamespaceString& nss,
                                      StringData ident) = 0;
-    virtual KVCatalog* getCatalog() = 0;
+    virtual DurableCatalog* getCatalog() = 0;
 };
 }
diff --git a/src/mongo/db/storage/kv/storage_engine_test.cpp b/src/mongo/db/storage/kv/storage_engine_test.cpp
index 40eddbfcef3..9e21339597c 100644
--- a/src/mongo/db/storage/kv/storage_engine_test.cpp
+++ b/src/mongo/db/storage/kv/storage_engine_test.cpp
@@ -43,8 +43,8 @@
 #include "mongo/db/repl/replication_coordinator_mock.h"
 #include "mongo/db/service_context_d_test_fixture.h"
 #include "mongo/db/storage/devnull/devnull_kv_engine.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/ephemeral_for_test/ephemeral_for_test_engine.h"
-#include "mongo/db/storage/kv/kv_catalog.h"
 #include "mongo/db/storage/kv/kv_engine.h"
 #include "mongo/db/storage/kv/storage_engine_impl.h"
 #include "mongo/db/storage/kv/storage_engine_test_fixture.h"
@@ -60,12 +60,12 @@ namespace {
 TEST_F(StorageEngineTest, ReconcileIdentsTest) {
     auto opCtx = cc().makeOperationContext();
 
-    // Add a collection, `db.coll1` to both the KVCatalog and KVEngine. The returned value is the
-    // `ident` name given to the collection.
+    // Add a collection, `db.coll1` to both the DurableCatalog and KVEngine. The returned value is
+    // the `ident` name given to the collection.
     auto swIdentName = createCollection(opCtx.get(), NamespaceString("db.coll1"));
     ASSERT_OK(swIdentName);
-    // Create a table in the KVEngine not reflected in the KVCatalog. This should be dropped when
-    // reconciling.
+    // Create a table in the KVEngine not reflected in the DurableCatalog. This should be dropped
+    // when reconciling.
     ASSERT_OK(createCollTable(opCtx.get(), NamespaceString("db.coll2")));
     ASSERT_OK(reconcile(opCtx.get()).getStatus());
     auto identsVec = getAllKVEngineIdents(opCtx.get());
@@ -87,7 +87,7 @@ TEST_F(StorageEngineTest, ReconcileIdentsTest) {
     ASSERT_EQUALS("db.coll1", toRebuild.first);
     ASSERT_EQUALS("_id", toRebuild.second);
 
-    // Now drop the `db.coll1` table, while leaving the KVCatalog entry.
+    // Now drop the `db.coll1` table, while leaving the DurableCatalog entry.
     ASSERT_OK(dropIdent(opCtx.get(), swIdentName.getValue()));
     ASSERT_EQUALS(static_cast<const unsigned long>(1), getAllKVEngineIdents(opCtx.get()).size());
 
diff --git a/src/mongo/db/storage/kv/storage_engine_test_fixture.h b/src/mongo/db/storage/kv/storage_engine_test_fixture.h
index 7f3bddb4bc8..7c588c991cd 100644
--- a/src/mongo/db/storage/kv/storage_engine_test_fixture.h
+++ b/src/mongo/db/storage/kv/storage_engine_test_fixture.h
@@ -33,7 +33,7 @@
 #include "mongo/db/catalog/collection_mock.h"
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/service_context_d_test_fixture.h"
-#include "mongo/db/storage/kv/kv_catalog.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/kv/kv_engine.h"
 #include "mongo/db/storage/kv/storage_engine_impl.h"
 #include "mongo/db/storage/storage_repair_observer.h"
@@ -68,7 +68,7 @@ public:
     }
 
     /**
-     * Create a collection table in the KVEngine not reflected in the KVCatalog.
+     * Create a collection table in the KVEngine not reflected in the DurableCatalog.
      */
     Status createCollTable(OperationContext* opCtx, NamespaceString collName) {
         const std::string identName = "collection-" + collName.ns();
@@ -134,11 +134,9 @@ public:
         auto descriptor = std::make_unique<IndexDescriptor>(
             collection.get(), IndexNames::findPluginName(spec), spec);
 
-        CollectionCatalogEntry* cce =
-            CollectionCatalog::get(opCtx).lookupCollectionCatalogEntryByNamespace(collNs);
         const auto protocol = IndexBuildProtocol::kTwoPhase;
-        auto ret = cce->prepareForIndexBuild(
-            opCtx, descriptor.get(), protocol, isBackgroundSecondaryBuild);
+        auto ret = DurableCatalog::get(opCtx)->prepareForIndexBuild(
+            opCtx, collNs, descriptor.get(), protocol, isBackgroundSecondaryBuild);
         return ret;
     }
 
@@ -147,25 +145,20 @@ public:
                         std::string key,
                         std::string sideWritesIdent,
                         std::string constraintViolationsIdent) {
-        CollectionCatalogEntry* cce =
-            CollectionCatalog::get(opCtx).lookupCollectionCatalogEntryByNamespace(collNs);
-        cce->setIndexBuildScanning(opCtx, key, sideWritesIdent, constraintViolationsIdent);
+        DurableCatalog::get(opCtx)->setIndexBuildScanning(
+            opCtx, collNs, key, sideWritesIdent, constraintViolationsIdent);
     }
 
     void indexBuildDrain(OperationContext* opCtx, NamespaceString collNs, std::string key) {
-        CollectionCatalogEntry* cce =
-            CollectionCatalog::get(opCtx).lookupCollectionCatalogEntryByNamespace(collNs);
-        cce->setIndexBuildDraining(opCtx, key);
+        DurableCatalog::get(opCtx)->setIndexBuildDraining(opCtx, collNs, key);
     }
 
     void indexBuildSuccess(OperationContext* opCtx, NamespaceString collNs, std::string key) {
-        CollectionCatalogEntry* cce =
-            CollectionCatalog::get(opCtx).lookupCollectionCatalogEntryByNamespace(collNs);
-        cce->indexBuildSuccess(opCtx, key);
+        DurableCatalog::get(opCtx)->indexBuildSuccess(opCtx, collNs, key);
     }
 
-    Status removeEntry(OperationContext* opCtx, StringData ns, KVCatalog* catalog) {
-        return catalog->_removeEntry(opCtx, NamespaceString(ns));
+    Status removeEntry(OperationContext* opCtx, StringData ns, DurableCatalog* catalog) {
+        return dynamic_cast<KVCatalog*>(catalog)->_removeEntry(opCtx, NamespaceString(ns));
     }
 
     StorageEngine* _storageEngine;
diff --git a/src/mongo/db/storage/mobile/mobile_kv_engine.cpp b/src/mongo/db/storage/mobile/mobile_kv_engine.cpp
index c6ff81b12f7..648ac3612ae 100644
--- a/src/mongo/db/storage/mobile/mobile_kv_engine.cpp
+++ b/src/mongo/db/storage/mobile/mobile_kv_engine.cpp
@@ -230,6 +230,7 @@ std::unique_ptr<RecordStore> MobileKVEngine::makeTemporaryRecordStore(OperationC
 
 
 Status MobileKVEngine::createSortedDataInterface(OperationContext* opCtx,
+                                                 const CollectionOptions& collOptions,
                                                  StringData ident,
                                                  const IndexDescriptor* desc) {
     return MobileIndex::create(opCtx, ident.toString());
diff --git a/src/mongo/db/storage/mobile/mobile_kv_engine.h b/src/mongo/db/storage/mobile/mobile_kv_engine.h
index 094bf0674d9..a09d4187200 100644
--- a/src/mongo/db/storage/mobile/mobile_kv_engine.h
+++ b/src/mongo/db/storage/mobile/mobile_kv_engine.h
@@ -65,6 +65,7 @@ public:
                                                           StringData ident) override;
 
     Status createSortedDataInterface(OperationContext* opCtx,
+                                     const CollectionOptions& collOptions,
                                      StringData ident,
                                      const IndexDescriptor* desc) override;
 
diff --git a/src/mongo/db/storage/record_store.h b/src/mongo/db/storage/record_store.h
index d5152e51d23..93f01d8c81a 100644
--- a/src/mongo/db/storage/record_store.h
+++ b/src/mongo/db/storage/record_store.h
@@ -231,8 +231,8 @@ public:
  * an OperationContext may throw a WriteConflictException.
  *
  * This class must be thread-safe for document-level locking storage engines. In addition, for
- * storage engines implementing the KVEngine some methods must be thread safe, see KVCatalog. Only
- * for MMAPv1 is this class not thread-safe.
+ * storage engines implementing the KVEngine some methods must be thread safe, see DurableCatalog.
+ * Only for MMAPv1 is this class not thread-safe.
  */
 class RecordStore {
     RecordStore(const RecordStore&) = delete;
diff --git a/src/mongo/db/storage/storage_engine.h b/src/mongo/db/storage/storage_engine.h
index 38b3d19ebd9..0ef67b009ed 100644
--- a/src/mongo/db/storage/storage_engine.h
+++ b/src/mongo/db/storage/storage_engine.h
@@ -42,7 +42,7 @@
 namespace mongo {
 
 class JournalListener;
-class KVCatalog;
+class DurableCatalog;
 class KVEngine;
 class OperationContext;
 class RecoveryUnit;
@@ -534,8 +534,8 @@ public:
 
     virtual KVEngine* getEngine() = 0;
     virtual const KVEngine* getEngine() const = 0;
-    virtual KVCatalog* getCatalog() = 0;
-    virtual const KVCatalog* getCatalog() const = 0;
+    virtual DurableCatalog* getCatalog() = 0;
+    virtual const DurableCatalog* getCatalog() const = 0;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
index a24d400a419..0b364917a76 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
@@ -1246,6 +1246,7 @@ string WiredTigerKVEngine::_uri(StringData ident) const {
 }
 
 Status WiredTigerKVEngine::createGroupedSortedDataInterface(OperationContext* opCtx,
+                                                            const CollectionOptions& collOptions,
                                                             StringData ident,
                                                             const IndexDescriptor* desc,
                                                             KVPrefix prefix) {
@@ -1257,9 +1258,6 @@ Status WiredTigerKVEngine::createGroupedSortedDataInterface(OperationContext* op
     // Treat 'collIndexOptions' as an empty string when the collection member of 'desc' is NULL in
     // order to allow for unit testing WiredTigerKVEngine::createSortedDataInterface().
     if (collection) {
-        const CollectionCatalogEntry* cce = collection->getCatalogEntry();
-        const CollectionOptions collOptions = cce->getCollectionOptions(opCtx);
-
         if (!collOptions.indexOptionDefaults["storageEngine"].eoo()) {
             BSONObj storageEngineOptions = collOptions.indexOptionDefaults["storageEngine"].Obj();
             collIndexOptions =
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
index bf194231e61..975e153f823 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
@@ -118,9 +118,11 @@ public:
                                                           StringData ident) override;
 
     Status createSortedDataInterface(OperationContext* opCtx,
+                                     const CollectionOptions& collOptions,
                                      StringData ident,
                                      const IndexDescriptor* desc) override {
-        return createGroupedSortedDataInterface(opCtx, ident, desc, KVPrefix::kNotPrefixed);
+        return createGroupedSortedDataInterface(
+            opCtx, collOptions, ident, desc, KVPrefix::kNotPrefixed);
     }
 
     std::unique_ptr<SortedDataInterface> getSortedDataInterface(
@@ -141,6 +143,7 @@ public:
                                                        KVPrefix prefix) override;
 
     Status createGroupedSortedDataInterface(OperationContext* opCtx,
+                                            const CollectionOptions& collOptions,
                                             StringData ident,
                                             const IndexDescriptor* desc,
                                             KVPrefix prefix) override;
diff --git a/src/mongo/db/ttl.cpp b/src/mongo/db/ttl.cpp
index 69df58f9428..e3d14867b05 100644
--- a/src/mongo/db/ttl.cpp
+++ b/src/mongo/db/ttl.cpp
@@ -52,6 +52,7 @@
 #include "mongo/db/query/internal_plans.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/ttl_collection_cache.h"
 #include "mongo/db/ttl_gen.h"
 #include "mongo/util/background.h"
@@ -123,6 +124,7 @@ public:
 private:
     void doTTLPass() {
         const ServiceContext::UniqueOperationContext opCtxPtr = cc().makeOperationContext();
+        auto durableCatalog = DurableCatalog::get(opCtxPtr.get());
         OperationContext& opCtx = *opCtxPtr;
 
         // If part of replSet but not in a readable state (e.g. during initial sync), skip.
@@ -147,11 +149,10 @@ private:
                 continue;
             }
 
-            CollectionCatalogEntry* collEntry = coll->getCatalogEntry();
             std::vector<std::string> indexNames;
-            collEntry->getAllIndexes(&opCtx, &indexNames);
+            durableCatalog->getAllIndexes(&opCtx, coll->ns(), &indexNames);
             for (const std::string& name : indexNames) {
-                BSONObj spec = collEntry->getIndexSpec(&opCtx, name);
+                BSONObj spec = durableCatalog->getIndexSpec(&opCtx, coll->ns(), name);
                 if (spec.hasField(secondsExpireField)) {
                     ttlIndexes.push_back(spec.getOwned());
                 }
diff --git a/src/mongo/dbtests/indexcatalogtests.cpp b/src/mongo/dbtests/indexcatalogtests.cpp
index f5aef397bcd..2c2356d51c5 100644
--- a/src/mongo/dbtests/indexcatalogtests.cpp
+++ b/src/mongo/dbtests/indexcatalogtests.cpp
@@ -35,6 +35,7 @@
 #include "mongo/db/client.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/index/index_descriptor.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/dbtests/dbtests.h"
 
 namespace IndexCatalogTests {
@@ -158,7 +159,8 @@ public:
         // Change value of "expireAfterSeconds" on disk.
         {
             WriteUnitOfWork wuow(&opCtx);
-            _coll->getCatalogEntry()->updateTTLSetting(&opCtx, "x_1", 10);
+            opCtx.getServiceContext()->getStorageEngine()->getCatalog()->updateTTLSetting(
+                &opCtx, _nss, "x_1", 10);
             wuow.commit();
         }
 
diff --git a/src/mongo/dbtests/storage_timestamp_tests.cpp b/src/mongo/dbtests/storage_timestamp_tests.cpp
index 0a71bdea88a..0efe1849674 100644
--- a/src/mongo/dbtests/storage_timestamp_tests.cpp
+++ b/src/mongo/dbtests/storage_timestamp_tests.cpp
@@ -308,11 +308,11 @@ public:
         return optRecord.get().data.toBson();
     }
 
-    BSONCollectionCatalogEntry::MetaData getMetaDataAtTime(KVCatalog* kvCatalog,
+    BSONCollectionCatalogEntry::MetaData getMetaDataAtTime(DurableCatalog* durableCatalog,
                                                            NamespaceString ns,
                                                            const Timestamp& ts) {
         OneOffRead oor(_opCtx, ts);
-        return kvCatalog->getMetaData(_opCtx, ns);
+        return durableCatalog->getMetaData(_opCtx, ns);
     }
 
     StatusWith<BSONObj> doAtomicApplyOps(const std::string& dbName,
@@ -455,21 +455,22 @@ public:
     }
 
     /**
-     * Asserts that the given collection is in (or not in) the KVCatalog's list of idents at the
+     * Asserts that the given collection is in (or not in) the DurableCatalog's list of idents at
+     * the
      * provided timestamp.
      */
     void assertNamespaceInIdents(NamespaceString nss, Timestamp ts, bool shouldExpect) {
         OneOffRead oor(_opCtx, ts);
-        auto kvCatalog = _opCtx->getServiceContext()->getStorageEngine()->getCatalog();
+        auto durableCatalog = DurableCatalog::get(_opCtx);
 
         AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_IS);
 
-        // getCollectionIdent() returns the ident for the given namespace in the KVCatalog.
+        // getCollectionIdent() returns the ident for the given namespace in the DurableCatalog.
         // getAllIdents() actually looks in the RecordStore for a list of all idents, and is thus
         // versioned by timestamp. We can expect a namespace to have a consistent ident across
         // timestamps, provided the collection does not get renamed.
-        auto expectedIdent = kvCatalog->getCollectionIdent(nss);
-        auto idents = kvCatalog->getAllIdents(_opCtx);
+        auto expectedIdent = durableCatalog->getCollectionIdent(nss);
+        auto idents = durableCatalog->getAllIdents(_opCtx);
         auto found = std::find(idents.begin(), idents.end(), expectedIdent);
 
         if (shouldExpect) {
@@ -483,10 +484,10 @@ public:
     /**
      * Use `ts` = Timestamp::min to observe all indexes.
      */
-    std::string getNewIndexIdentAtTime(KVCatalog* kvCatalog,
+    std::string getNewIndexIdentAtTime(DurableCatalog* durableCatalog,
                                        std::vector<std::string>& origIdents,
                                        Timestamp ts) {
-        auto ret = getNewIndexIdentsAtTime(kvCatalog, origIdents, ts);
+        auto ret = getNewIndexIdentsAtTime(durableCatalog, origIdents, ts);
         ASSERT_EQ(static_cast<std::size_t>(1), ret.size()) << " Num idents: " << ret.size();
         return ret[0];
     }
@@ -494,14 +495,14 @@ public:
     /**
      * Use `ts` = Timestamp::min to observe all indexes.
      */
-    std::vector<std::string> getNewIndexIdentsAtTime(KVCatalog* kvCatalog,
+    std::vector<std::string> getNewIndexIdentsAtTime(DurableCatalog* durableCatalog,
                                                      std::vector<std::string>& origIdents,
                                                      Timestamp ts) {
         OneOffRead oor(_opCtx, ts);
 
         // Find the collection and index ident by performing a set difference on the original
         // idents and the current idents.
-        std::vector<std::string> identsWithColl = kvCatalog->getAllIdents(_opCtx);
+        std::vector<std::string> identsWithColl = durableCatalog->getAllIdents(_opCtx);
         std::sort(origIdents.begin(), origIdents.end());
         std::sort(identsWithColl.begin(), identsWithColl.end());
         std::vector<std::string> idxIdents;
@@ -517,10 +518,11 @@ public:
         return idxIdents;
     }
 
-    std::string getDroppedIndexIdent(KVCatalog* kvCatalog, std::vector<std::string>& origIdents) {
+    std::string getDroppedIndexIdent(DurableCatalog* durableCatalog,
+                                     std::vector<std::string>& origIdents) {
         // Find the collection and index ident by performing a set difference on the original
         // idents and the current idents.
-        std::vector<std::string> identsWithColl = kvCatalog->getAllIdents(_opCtx);
+        std::vector<std::string> identsWithColl = durableCatalog->getAllIdents(_opCtx);
         std::sort(origIdents.begin(), origIdents.end());
         std::sort(identsWithColl.begin(), identsWithColl.end());
         std::vector<std::string> collAndIdxIdents;
@@ -534,11 +536,11 @@ public:
         return collAndIdxIdents[0];
     }
 
-    std::vector<std::string> _getIdentDifference(KVCatalog* kvCatalog,
+    std::vector<std::string> _getIdentDifference(DurableCatalog* durableCatalog,
                                                  std::vector<std::string>& origIdents) {
         // Find the ident difference by performing a set difference on the original idents and the
         // current idents.
-        std::vector<std::string> identsWithColl = kvCatalog->getAllIdents(_opCtx);
+        std::vector<std::string> identsWithColl = durableCatalog->getAllIdents(_opCtx);
         std::sort(origIdents.begin(), origIdents.end());
         std::sort(identsWithColl.begin(), identsWithColl.end());
         std::vector<std::string> collAndIdxIdents;
@@ -550,9 +552,9 @@ public:
         return collAndIdxIdents;
     }
     std::tuple<std::string, std::string> getNewCollectionIndexIdent(
-        KVCatalog* kvCatalog, std::vector<std::string>& origIdents) {
+        DurableCatalog* durableCatalog, std::vector<std::string>& origIdents) {
         // Find the collection and index ident difference.
-        auto collAndIdxIdents = _getIdentDifference(kvCatalog, origIdents);
+        auto collAndIdxIdents = _getIdentDifference(durableCatalog, origIdents);
 
         ASSERT(collAndIdxIdents.size() == 1 || collAndIdxIdents.size() == 2);
         if (collAndIdxIdents.size() == 1) {
@@ -570,13 +572,13 @@ public:
     /**
      * Note: expectedNewIndexIdents should include the _id index.
      */
-    void assertRenamedCollectionIdentsAtTimestamp(KVCatalog* kvCatalog,
+    void assertRenamedCollectionIdentsAtTimestamp(DurableCatalog* durableCatalog,
                                                   std::vector<std::string>& origIdents,
                                                   size_t expectedNewIndexIdents,
                                                   Timestamp timestamp) {
         OneOffRead oor(_opCtx, timestamp);
         // Find the collection and index ident difference.
-        auto collAndIdxIdents = _getIdentDifference(kvCatalog, origIdents);
+        auto collAndIdxIdents = _getIdentDifference(durableCatalog, origIdents);
         size_t newNssIdents, newIdxIdents;
         newNssIdents = newIdxIdents = 0;
         for (const auto& ident : collAndIdxIdents) {
@@ -595,13 +597,13 @@ public:
             << ") differ from actual new index idents (" << newIdxIdents << ")";
     }
 
-    void assertIdentsExistAtTimestamp(KVCatalog* kvCatalog,
+    void assertIdentsExistAtTimestamp(DurableCatalog* durableCatalog,
                                       const std::string& collIdent,
                                       const std::string& indexIdent,
                                       Timestamp timestamp) {
         OneOffRead oor(_opCtx, timestamp);
 
-        auto allIdents = kvCatalog->getAllIdents(_opCtx);
+        auto allIdents = durableCatalog->getAllIdents(_opCtx);
         if (collIdent.size() > 0) {
             // Index build test does not pass in a collection ident.
             ASSERT(std::find(allIdents.begin(), allIdents.end(), collIdent) != allIdents.end());
@@ -613,12 +615,12 @@ public:
         }
     }
 
-    void assertIdentsMissingAtTimestamp(KVCatalog* kvCatalog,
+    void assertIdentsMissingAtTimestamp(DurableCatalog* durableCatalog,
                                         const std::string& collIdent,
                                         const std::string& indexIdent,
                                         Timestamp timestamp) {
         OneOffRead oor(_opCtx, timestamp);
-        auto allIdents = kvCatalog->getAllIdents(_opCtx);
+        auto allIdents = durableCatalog->getAllIdents(_opCtx);
         if (collIdent.size() > 0) {
             // Index build test does not pass in a collection ident.
             ASSERT(std::find(allIdents.begin(), allIdents.end(), collIdent) == allIdents.end());
@@ -650,16 +652,18 @@ public:
                              Timestamp ts,
                              bool shouldBeMultikey,
                              const MultikeyPaths& expectedMultikeyPaths) {
-        auto catalog = collection->getCatalogEntry();
+        DurableCatalog* durableCatalog = DurableCatalog::get(opCtx);
 
         OneOffRead oor(_opCtx, ts);
 
         MultikeyPaths actualMultikeyPaths;
         if (!shouldBeMultikey) {
-            ASSERT_FALSE(catalog->isIndexMultikey(opCtx, indexName, &actualMultikeyPaths))
+            ASSERT_FALSE(durableCatalog->isIndexMultikey(
+                opCtx, collection->ns(), indexName, &actualMultikeyPaths))
                 << "index " << indexName << " should not be multikey at timestamp " << ts;
         } else {
-            ASSERT(catalog->isIndexMultikey(opCtx, indexName, &actualMultikeyPaths))
+            ASSERT(durableCatalog->isIndexMultikey(
+                opCtx, collection->ns(), indexName, &actualMultikeyPaths))
                 << "index " << indexName << " should be multikey at timestamp " << ts;
         }
 
@@ -1746,7 +1750,7 @@ public:
             std::make_unique<repl::DropPendingCollectionReaper>(storageInterface));
 
         auto storageEngine = _opCtx->getServiceContext()->getStorageEngine();
-        auto kvCatalog = storageEngine->getCatalog();
+        auto durableCatalog = storageEngine->getCatalog();
 
         // Declare the database to be in a "synced" state, i.e: in steady-state replication.
         Timestamp syncTime = _clock->reserveTicks(1).asTimestamp();
@@ -1771,7 +1775,7 @@ public:
 
             // Save the pre-state idents so we can capture the specific idents related to collection
             // creation.
-            std::vector<std::string> origIdents = kvCatalog->getAllIdents(_opCtx);
+            std::vector<std::string> origIdents = durableCatalog->getAllIdents(_opCtx);
             const auto& nss = std::get<0>(tuple);
 
             // Non-replicated namespaces are wrapped in an unreplicated writes block. This has the
@@ -1787,7 +1791,8 @@ public:
             // Bind the local values to the variables in the parent scope.
             auto& collIdent = std::get<1>(tuple);
             auto& indexIdent = std::get<2>(tuple);
-            std::tie(collIdent, indexIdent) = getNewCollectionIndexIdent(kvCatalog, origIdents);
+            std::tie(collIdent, indexIdent) =
+                getNewCollectionIndexIdent(durableCatalog, origIdents);
         }
 
         AutoGetCollection coll(_opCtx, nss, LockMode::MODE_X);
@@ -1806,10 +1811,10 @@ public:
         // If the storage engine is managing drops internally, the ident should not be visible after
         // a drop.
         if (storageEngine->supportsPendingDrops()) {
-            assertIdentsMissingAtTimestamp(kvCatalog, collIdent, indexIdent, postRenameTime);
+            assertIdentsMissingAtTimestamp(durableCatalog, collIdent, indexIdent, postRenameTime);
         } else {
             // The namespace has changed, but the ident still exists as-is after the rename.
-            assertIdentsExistAtTimestamp(kvCatalog, collIdent, indexIdent, postRenameTime);
+            assertIdentsExistAtTimestamp(durableCatalog, collIdent, indexIdent, postRenameTime);
         }
 
         const Timestamp dropTime = _clock->reserveTicks(1).asTimestamp();
@@ -1823,12 +1828,12 @@ public:
 
         // Assert that the idents do not exist.
         assertIdentsMissingAtTimestamp(
-            kvCatalog, sysProfileIdent, sysProfileIndexIdent, Timestamp::max());
-        assertIdentsMissingAtTimestamp(kvCatalog, collIdent, indexIdent, Timestamp::max());
+            durableCatalog, sysProfileIdent, sysProfileIndexIdent, Timestamp::max());
+        assertIdentsMissingAtTimestamp(durableCatalog, collIdent, indexIdent, Timestamp::max());
 
         // dropDatabase must not timestamp the final write. The collection and index should seem
         // to have never existed.
-        assertIdentsMissingAtTimestamp(kvCatalog, collIdent, indexIdent, syncTime);
+        assertIdentsMissingAtTimestamp(durableCatalog, collIdent, indexIdent, syncTime);
     }
 };
 
@@ -1858,7 +1863,7 @@ public:
         }
 
         auto storageEngine = _opCtx->getServiceContext()->getStorageEngine();
-        auto kvCatalog = storageEngine->getCatalog();
+        auto durableCatalog = storageEngine->getCatalog();
 
         NamespaceString nss("unittests.timestampIndexBuilds");
         reset(nss);
@@ -1878,7 +1883,7 @@ public:
 
         // Save the pre-state idents so we can capture the specific ident related to index
         // creation.
-        std::vector<std::string> origIdents = kvCatalog->getAllIdents(_opCtx);
+        std::vector<std::string> origIdents = durableCatalog->getAllIdents(_opCtx);
 
         // Build an index on `{a: 1}`. This index will be multikey.
         MultiIndexBlock indexer;
@@ -1944,22 +1949,24 @@ public:
         const Timestamp afterIndexBuild = _clock->reserveTicks(1).asTimestamp();
 
         const std::string indexIdent =
-            getNewIndexIdentAtTime(kvCatalog, origIdents, Timestamp::min());
-        assertIdentsMissingAtTimestamp(kvCatalog, "", indexIdent, beforeIndexBuild.asTimestamp());
+            getNewIndexIdentAtTime(durableCatalog, origIdents, Timestamp::min());
+        assertIdentsMissingAtTimestamp(
+            durableCatalog, "", indexIdent, beforeIndexBuild.asTimestamp());
 
         // Assert that the index entry exists after init and `ready: false`.
-        assertIdentsExistAtTimestamp(kvCatalog, "", indexIdent, afterIndexInit.asTimestamp());
+        assertIdentsExistAtTimestamp(durableCatalog, "", indexIdent, afterIndexInit.asTimestamp());
         {
-            ASSERT_FALSE(getIndexMetaData(
-                             getMetaDataAtTime(kvCatalog, nss, afterIndexInit.asTimestamp()), "a_1")
-                             .ready);
+            ASSERT_FALSE(
+                getIndexMetaData(
+                    getMetaDataAtTime(durableCatalog, nss, afterIndexInit.asTimestamp()), "a_1")
+                    .ready);
         }
 
         // After the build completes, assert that the index is `ready: true` and multikey.
-        assertIdentsExistAtTimestamp(kvCatalog, "", indexIdent, afterIndexBuild);
+        assertIdentsExistAtTimestamp(durableCatalog, "", indexIdent, afterIndexBuild);
         {
             auto indexMetaData =
-                getIndexMetaData(getMetaDataAtTime(kvCatalog, nss, afterIndexBuild), "a_1");
+                getIndexMetaData(getMetaDataAtTime(durableCatalog, nss, afterIndexBuild), "a_1");
             ASSERT(indexMetaData.ready);
             ASSERT(indexMetaData.multikey);
 
@@ -2126,7 +2133,7 @@ class TimestampMultiIndexBuilds : public StorageTimestampTest {
 public:
     void run() {
         auto storageEngine = _opCtx->getServiceContext()->getStorageEngine();
-        auto kvCatalog = storageEngine->getCatalog();
+        auto durableCatalog = storageEngine->getCatalog();
 
         NamespaceString nss("unittests.timestampMultiIndexBuilds");
         reset(nss);
@@ -2147,7 +2154,7 @@ public:
 
             // Save the pre-state idents so we can capture the specific ident related to index
             // creation.
-            origIdents = kvCatalog->getAllIdents(_opCtx);
+            origIdents = durableCatalog->getAllIdents(_opCtx);
         }
 
         DBDirectClient client(_opCtx);
@@ -2184,25 +2191,27 @@ public:
         // The idents are created and persisted with the "ready: false" write. There should be two
         // new index idents visible at this time.
         const std::vector<std::string> indexes =
-            getNewIndexIdentsAtTime(kvCatalog, origIdents, indexCreateInitTs);
+            getNewIndexIdentsAtTime(durableCatalog, origIdents, indexCreateInitTs);
         ASSERT_EQ(static_cast<std::size_t>(2), indexes.size()) << " Num idents: " << indexes.size();
 
         ASSERT_FALSE(
-            getIndexMetaData(getMetaDataAtTime(kvCatalog, nss, indexCreateInitTs), "a_1").ready);
+            getIndexMetaData(getMetaDataAtTime(durableCatalog, nss, indexCreateInitTs), "a_1")
+                .ready);
         ASSERT_FALSE(
-            getIndexMetaData(getMetaDataAtTime(kvCatalog, nss, indexCreateInitTs), "b_1").ready);
+            getIndexMetaData(getMetaDataAtTime(durableCatalog, nss, indexCreateInitTs), "b_1")
+                .ready);
 
         // Assert the `a_1` index becomes ready at the next oplog entry time.
         ASSERT_TRUE(
-            getIndexMetaData(getMetaDataAtTime(kvCatalog, nss, indexAComplete), "a_1").ready);
+            getIndexMetaData(getMetaDataAtTime(durableCatalog, nss, indexAComplete), "a_1").ready);
         ASSERT_FALSE(
-            getIndexMetaData(getMetaDataAtTime(kvCatalog, nss, indexAComplete), "b_1").ready);
+            getIndexMetaData(getMetaDataAtTime(durableCatalog, nss, indexAComplete), "b_1").ready);
 
         // Assert the `b_1` index becomes ready at the last oplog entry time.
         ASSERT_TRUE(
-            getIndexMetaData(getMetaDataAtTime(kvCatalog, nss, indexBComplete), "a_1").ready);
+            getIndexMetaData(getMetaDataAtTime(durableCatalog, nss, indexBComplete), "a_1").ready);
         ASSERT_TRUE(
-            getIndexMetaData(getMetaDataAtTime(kvCatalog, nss, indexBComplete), "b_1").ready);
+            getIndexMetaData(getMetaDataAtTime(durableCatalog, nss, indexBComplete), "b_1").ready);
     }
 };
 
@@ -2210,7 +2219,7 @@ class TimestampMultiIndexBuildsDuringRename : public StorageTimestampTest {
 public:
     void run() {
         auto storageEngine = _opCtx->getServiceContext()->getStorageEngine();
-        auto kvCatalog = storageEngine->getCatalog();
+        auto durableCatalog = storageEngine->getCatalog();
 
         NamespaceString nss("unittests.timestampMultiIndexBuildsDuringRename");
         reset(nss);
@@ -2250,7 +2259,7 @@ public:
 
         // Save the pre-state idents so we can capture the specific ident related to index
         // creation.
-        std::vector<std::string> origIdents = kvCatalog->getAllIdents(_opCtx);
+        std::vector<std::string> origIdents = durableCatalog->getAllIdents(_opCtx);
 
         // Rename collection.
         BSONObj renameResult;
@@ -2290,26 +2299,27 @@ public:
         // We expect one new collection ident and one new index ident (the _id index) during this
         // rename.
         assertRenamedCollectionIdentsAtTimestamp(
-            kvCatalog, origIdents, /*expectedNewIndexIdents*/ 1, indexCreateInitTs);
+            durableCatalog, origIdents, /*expectedNewIndexIdents*/ 1, indexCreateInitTs);
 
         // We expect one new collection ident and three new index idents (including the _id index)
         // after this rename. The a_1 and b_1 index idents are created and persisted with the
         // "ready: true" write.
         assertRenamedCollectionIdentsAtTimestamp(
-            kvCatalog, origIdents, /*expectedNewIndexIdents*/ 3, indexBComplete);
+            durableCatalog, origIdents, /*expectedNewIndexIdents*/ 3, indexBComplete);
 
         // Assert the `a_1` index becomes ready at the next oplog entry time.
         ASSERT_TRUE(
-            getIndexMetaData(getMetaDataAtTime(kvCatalog, renamedNss, indexAComplete), "a_1")
+            getIndexMetaData(getMetaDataAtTime(durableCatalog, renamedNss, indexAComplete), "a_1")
                 .ready);
-        assertIndexMetaDataMissing(getMetaDataAtTime(kvCatalog, renamedNss, indexAComplete), "b_1");
+        assertIndexMetaDataMissing(getMetaDataAtTime(durableCatalog, renamedNss, indexAComplete),
+                                   "b_1");
 
         // Assert the `b_1` index becomes ready at the last oplog entry time.
         ASSERT_TRUE(
-            getIndexMetaData(getMetaDataAtTime(kvCatalog, renamedNss, indexBComplete), "a_1")
+            getIndexMetaData(getMetaDataAtTime(durableCatalog, renamedNss, indexBComplete), "a_1")
                 .ready);
         ASSERT_TRUE(
-            getIndexMetaData(getMetaDataAtTime(kvCatalog, renamedNss, indexBComplete), "b_1")
+            getIndexMetaData(getMetaDataAtTime(durableCatalog, renamedNss, indexBComplete), "b_1")
                 .ready);
     }
 };
@@ -2318,7 +2328,7 @@ class TimestampIndexDrops : public StorageTimestampTest {
 public:
     void run() {
         auto storageEngine = _opCtx->getServiceContext()->getStorageEngine();
-        auto kvCatalog = storageEngine->getCatalog();
+        auto durableCatalog = storageEngine->getCatalog();
 
         NamespaceString nss("unittests.timestampIndexDrops");
         reset(nss);
@@ -2341,7 +2351,7 @@ public:
 
         // Save the pre-state idents so we can capture the specific ident related to index
         // creation.
-        std::vector<std::string> origIdents = kvCatalog->getAllIdents(_opCtx);
+        std::vector<std::string> origIdents = durableCatalog->getAllIdents(_opCtx);
 
         std::vector<Timestamp> afterCreateTimestamps;
         std::vector<std::string> indexIdents;
@@ -2353,15 +2363,17 @@ public:
             afterCreateTimestamps.push_back(_clock->reserveTicks(1).asTimestamp());
 
             // Add the new ident to the vector and reset the current idents.
-            indexIdents.push_back(getNewIndexIdentAtTime(kvCatalog, origIdents, Timestamp::min()));
-            origIdents = kvCatalog->getAllIdents(_opCtx);
+            indexIdents.push_back(
+                getNewIndexIdentAtTime(durableCatalog, origIdents, Timestamp::min()));
+            origIdents = durableCatalog->getAllIdents(_opCtx);
         }
 
         // Ensure each index is visible at the correct timestamp, and not before.
         for (size_t i = 0; i < indexIdents.size(); i++) {
             auto beforeTs = (i == 0) ? beforeIndexBuild : afterCreateTimestamps[i - 1];
-            assertIdentsMissingAtTimestamp(kvCatalog, "", indexIdents[i], beforeTs);
-            assertIdentsExistAtTimestamp(kvCatalog, "", indexIdents[i], afterCreateTimestamps[i]);
+            assertIdentsMissingAtTimestamp(durableCatalog, "", indexIdents[i], beforeTs);
+            assertIdentsExistAtTimestamp(
+                durableCatalog, "", indexIdents[i], afterCreateTimestamps[i]);
         }
 
         const LogicalTime beforeDropTs = _clock->getClusterTime();
@@ -2381,10 +2393,10 @@ public:
         for (size_t i = 0; i < nIdents; i++) {
             OneOffRead oor(_opCtx, beforeDropTs.addTicks(i + 1).asTimestamp());
 
-            auto ident = getDroppedIndexIdent(kvCatalog, origIdents);
+            auto ident = getDroppedIndexIdent(durableCatalog, origIdents);
             indexIdents.erase(std::remove(indexIdents.begin(), indexIdents.end(), ident));
 
-            origIdents = kvCatalog->getAllIdents(_opCtx);
+            origIdents = durableCatalog->getAllIdents(_opCtx);
         }
         ASSERT_EQ(indexIdents.size(), 0ul) << "Dropped idents should match created idents";
     }
@@ -2535,11 +2547,11 @@ public:
 
             // Grab the existing idents to identify the ident created by the index build.
             auto storageEngine = _opCtx->getServiceContext()->getStorageEngine();
-            auto kvCatalog = storageEngine->getCatalog();
+            auto durableCatalog = storageEngine->getCatalog();
             std::vector<std::string> origIdents;
             {
                 AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_IS);
-                origIdents = kvCatalog->getAllIdents(_opCtx);
+                origIdents = durableCatalog->getAllIdents(_opCtx);
             }
 
             auto indexSpec = BSON("createIndexes" << nss.coll() << "ns" << nss.ns() << "v"
@@ -2562,15 +2574,16 @@ public:
 
             AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_IS);
             const std::string indexIdent =
-                getNewIndexIdentAtTime(kvCatalog, origIdents, Timestamp::min());
+                getNewIndexIdentAtTime(durableCatalog, origIdents, Timestamp::min());
             assertIdentsMissingAtTimestamp(
-                kvCatalog, "", indexIdent, beforeBuildTime.asTimestamp());
-            assertIdentsExistAtTimestamp(kvCatalog, "", indexIdent, startBuildTs);
+                durableCatalog, "", indexIdent, beforeBuildTime.asTimestamp());
+            assertIdentsExistAtTimestamp(durableCatalog, "", indexIdent, startBuildTs);
 
             // On a primary, the index build should start and finish at `startBuildTs` because it is
             // built in the foreground.
             ASSERT_TRUE(
-                getIndexMetaData(getMetaDataAtTime(kvCatalog, nss, startBuildTs), "field_1").ready);
+                getIndexMetaData(getMetaDataAtTime(durableCatalog, nss, startBuildTs), "field_1")
+                    .ready);
         }
     }
 };
@@ -2579,7 +2592,7 @@ class ViewCreationSeparateTransaction : public StorageTimestampTest {
 public:
     void run() {
         auto storageEngine = _opCtx->getServiceContext()->getStorageEngine();
-        auto kvCatalog = storageEngine->getCatalog();
+        auto durableCatalog = storageEngine->getCatalog();
 
         const NamespaceString backingCollNss("unittests.backingColl");
         reset(backingCollNss);
@@ -2611,12 +2624,12 @@ public:
         {
             Lock::GlobalRead read(_opCtx);
             auto systemViewsMd = getMetaDataAtTime(
-                kvCatalog, systemViewsNss, Timestamp(systemViewsCreateTs.asULL() - 1));
+                durableCatalog, systemViewsNss, Timestamp(systemViewsCreateTs.asULL() - 1));
             ASSERT_EQ("", systemViewsMd.ns)
                 << systemViewsNss
                 << " incorrectly exists before creation. CreateTs: " << systemViewsCreateTs;
 
-            systemViewsMd = getMetaDataAtTime(kvCatalog, systemViewsNss, systemViewsCreateTs);
+            systemViewsMd = getMetaDataAtTime(durableCatalog, systemViewsNss, systemViewsCreateTs);
             ASSERT_EQ(systemViewsNss.ns(), systemViewsMd.ns);
 
             AutoGetCollection autoColl(_opCtx, systemViewsNss, LockMode::MODE_IS);
@@ -2679,13 +2692,13 @@ public:
         ASSERT_GT(indexOp.getTimestamp(), futureTs) << op.toBSON();
         AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_IS);
         auto storageEngine = _opCtx->getServiceContext()->getStorageEngine();
-        auto kvCatalog = storageEngine->getCatalog();
-        auto indexIdent = kvCatalog->getIndexIdent(_opCtx, nss, "user_1_db_1");
-        assertIdentsMissingAtTimestamp(kvCatalog, "", indexIdent, pastTs);
-        assertIdentsMissingAtTimestamp(kvCatalog, "", indexIdent, presentTs);
-        assertIdentsMissingAtTimestamp(kvCatalog, "", indexIdent, futureTs);
-        assertIdentsExistAtTimestamp(kvCatalog, "", indexIdent, indexCreateTs);
-        assertIdentsExistAtTimestamp(kvCatalog, "", indexIdent, nullTs);
+        auto durableCatalog = storageEngine->getCatalog();
+        auto indexIdent = durableCatalog->getIndexIdent(_opCtx, nss, "user_1_db_1");
+        assertIdentsMissingAtTimestamp(durableCatalog, "", indexIdent, pastTs);
+        assertIdentsMissingAtTimestamp(durableCatalog, "", indexIdent, presentTs);
+        assertIdentsMissingAtTimestamp(durableCatalog, "", indexIdent, futureTs);
+        assertIdentsExistAtTimestamp(durableCatalog, "", indexIdent, indexCreateTs);
+        assertIdentsExistAtTimestamp(durableCatalog, "", indexIdent, nullTs);
     }
 };