SERVER-40717 Remove CollectionCatalogEntry and KVColletionCatalogEntry

69 files changed, 347 insertions, 979 deletions

diff --git a/src/mongo/db/catalog/capped_utils.cpp b/src/mongo/db/catalog/capped_utils.cpp
index 2b77bbe794d..889ca73ddd2 100644
--- a/src/mongo/db/catalog/capped_utils.cpp
+++ b/src/mongo/db/catalog/capped_utils.cpp
@@ -35,7 +35,6 @@
 
 #include "mongo/base/error_codes.h"
 #include "mongo/db/background.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/create_collection.h"
 #include "mongo/db/catalog/document_validation.h"
 #include "mongo/db/catalog/drop_collection.h"
diff --git a/src/mongo/db/catalog/capped_utils_test.cpp b/src/mongo/db/catalog/capped_utils_test.cpp
index ce2afb79b1d..d621beb280d 100644
--- a/src/mongo/db/catalog/capped_utils_test.cpp
+++ b/src/mongo/db/catalog/capped_utils_test.cpp
@@ -32,7 +32,6 @@
 #include <memory>
 
 #include "mongo/db/catalog/capped_utils.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/client.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/repl/replication_coordinator.h"
diff --git a/src/mongo/db/catalog/catalog_control.cpp b/src/mongo/db/catalog/catalog_control.cpp
index 92e9ec691da..95c1381f91e 100644
--- a/src/mongo/db/catalog/catalog_control.cpp
+++ b/src/mongo/db/catalog/catalog_control.cpp
@@ -119,17 +119,8 @@ void openCatalog(OperationContext* opCtx, const MinVisibleTimestampMap& minVisib
     for (auto indexNamespace : indexesToRebuild.getValue()) {
         NamespaceString collNss(indexNamespace.first);
         auto indexName = indexNamespace.second;
-
-        auto collCatalogEntry =
-            CollectionCatalog::get(opCtx).lookupCollectionCatalogEntryByNamespace(collNss);
-        invariant(collCatalogEntry,
-                  str::stream() << "couldn't get collection catalog entry for collection "
-                                << collNss.toString());
-
-        auto indexSpecs =
-            getIndexNameObjs(opCtx, collCatalogEntry, [&indexName](const std::string& name) {
-                return name == indexName;
-            });
+        auto indexSpecs = getIndexNameObjs(
+            opCtx, collNss, [&indexName](const std::string& name) { return name == indexName; });
         if (!indexSpecs.isOK() || indexSpecs.getValue().first.empty()) {
             fassert(40689,
                     {ErrorCodes::InternalError,
@@ -155,11 +146,8 @@ void openCatalog(OperationContext* opCtx, const MinVisibleTimestampMap& minVisib
     for (const auto& entry : nsToIndexNameObjMap) {
         NamespaceString collNss(entry.first);
 
-        auto collCatalogEntry =
-            CollectionCatalog::get(opCtx).lookupCollectionCatalogEntryByNamespace(collNss);
-        invariant(collCatalogEntry,
-                  str::stream() << "couldn't get collection catalog entry for collection "
-                                << collNss.toString());
+        auto collection = CollectionCatalog::get(opCtx).lookupCollectionByNamespace(collNss);
+        invariant(collection, str::stream() << "couldn't get collection " << collNss.toString());
 
         for (const auto& indexName : entry.second.first) {
             log() << "openCatalog: rebuilding index: collection: " << collNss.toString()
@@ -167,7 +155,7 @@ void openCatalog(OperationContext* opCtx, const MinVisibleTimestampMap& minVisib
         }
 
         std::vector<BSONObj> indexSpecs = entry.second.second;
-        fassert(40690, rebuildIndexesOnCollection(opCtx, collCatalogEntry, indexSpecs));
+        fassert(40690, rebuildIndexesOnCollection(opCtx, collection, indexSpecs));
     }
 
     // Open all databases and repopulate the CollectionCatalog.
diff --git a/src/mongo/db/catalog/coll_mod.cpp b/src/mongo/db/catalog/coll_mod.cpp
index f801ee42d29..569dbb1bd0a 100644
--- a/src/mongo/db/catalog/coll_mod.cpp
+++ b/src/mongo/db/catalog/coll_mod.cpp
@@ -38,7 +38,6 @@
 
 #include "mongo/bson/simple_bsonelement_comparator.h"
 #include "mongo/db/background.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/client.h"
diff --git a/src/mongo/db/catalog/collection.h b/src/mongo/db/catalog/collection.h
index dd2af83ba7d..7cbaa9d5ced 100644
--- a/src/mongo/db/catalog/collection.h
+++ b/src/mongo/db/catalog/collection.h
@@ -57,7 +57,6 @@
 
 namespace mongo {
 class CappedCallback;
-class CollectionCatalogEntry;
 class ExtentManager;
 class IndexCatalog;
 class IndexCatalogEntry;
@@ -177,10 +176,10 @@ public:
          * Constructs a Collection object. This does not persist any state to the storage engine,
          * only constructs an in-memory representation of what already exists on disk.
          */
-        virtual std::unique_ptr<Collection> make(
-            OperationContext* opCtx,
-            CollectionUUID uuid,
-            CollectionCatalogEntry* collectionCatalogEntry) const = 0;
+        virtual std::unique_ptr<Collection> make(OperationContext* opCtx,
+                                                 const NamespaceString& nss,
+                                                 CollectionUUID uuid,
+                                                 std::unique_ptr<RecordStore> rs) const = 0;
     };
 
     /**
@@ -193,9 +192,6 @@ public:
 
     virtual bool ok() const = 0;
 
-    virtual CollectionCatalogEntry* getCatalogEntry() = 0;
-    virtual const CollectionCatalogEntry* getCatalogEntry() const = 0;
-
     virtual CollectionInfoCache* infoCache() = 0;
     virtual const CollectionInfoCache* infoCache() const = 0;
 
diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index d34b865f076..546a81cc13f 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -48,26 +48,20 @@ class FinishDropCollectionChange : public RecoveryUnit::Change {
 public:
     FinishDropCollectionChange(CollectionCatalog* catalog,
                                std::unique_ptr<Collection> coll,
-                               std::unique_ptr<CollectionCatalogEntry> catalogEntry,
                                CollectionUUID uuid)
-        : _catalog(catalog),
-          _coll(std::move(coll)),
-          _catalogEntry(std::move(catalogEntry)),
-          _uuid(uuid) {}
+        : _catalog(catalog), _coll(std::move(coll)), _uuid(uuid) {}
 
     void commit(boost::optional<Timestamp>) override {
         _coll.reset();
-        _catalogEntry.reset();
     }
 
     void rollback() override {
-        _catalog->registerCollection(_uuid, std::move(_catalogEntry), std::move(_coll));
+        _catalog->registerCollection(_uuid, std::move(_coll));
     }
 
 private:
     CollectionCatalog* _catalog;
     std::unique_ptr<Collection> _coll;
-    std::unique_ptr<CollectionCatalogEntry> _catalogEntry;
     CollectionUUID _uuid;
 };
 
@@ -82,39 +76,23 @@ CollectionCatalog::iterator::iterator(StringData dbName,
     stdx::lock_guard<stdx::mutex> lock(_catalog->_catalogLock);
     _mapIter = _catalog->_orderedCollections.lower_bound(std::make_pair(_dbName, minUuid));
 
-    // The entry _mapIter points to is valid if it's not at the end of _orderedCollections and
-    // the entry's database is the same as dbName.
-    while (_mapIter != _catalog->_orderedCollections.end() && _mapIter->first.first == _dbName &&
-           _mapIter->second->collectionPtr == nullptr) {
-        _mapIter++;
-    }
     if (_mapIter != _catalog->_orderedCollections.end() && _mapIter->first.first == _dbName) {
         _uuid = _mapIter->first.second;
     }
 }
 
 CollectionCatalog::iterator::iterator(
-    std::map<std::pair<std::string, CollectionUUID>, CollectionInfo*>::const_iterator mapIter)
+    std::map<std::pair<std::string, CollectionUUID>, Collection*>::const_iterator mapIter)
     : _mapIter(mapIter) {}
 
-CollectionCatalog::iterator::pointer CollectionCatalog::iterator::operator->() {
-    stdx::lock_guard<stdx::mutex> lock(_catalog->_catalogLock);
-    _repositionIfNeeded();
-    if (_exhausted()) {
-        return nullptr;
-    }
-
-    return &_mapIter->second->collectionPtr;
-}
-
-CollectionCatalog::iterator::reference CollectionCatalog::iterator::operator*() {
+const CollectionCatalog::iterator::value_type CollectionCatalog::iterator::operator*() {
     stdx::lock_guard<stdx::mutex> lock(_catalog->_catalogLock);
     _repositionIfNeeded();
     if (_exhausted()) {
         return _nullCollection;
     }
 
-    return _mapIter->second->collectionPtr;
+    return _mapIter->second;
 }
 
 boost::optional<CollectionUUID> CollectionCatalog::iterator::uuid() {
@@ -124,20 +102,17 @@ boost::optional<CollectionUUID> CollectionCatalog::iterator::uuid() {
 CollectionCatalog::iterator CollectionCatalog::iterator::operator++() {
     stdx::lock_guard<stdx::mutex> lock(_catalog->_catalogLock);
 
-    // Skip over CollectionInfo that has CatalogEntry but has no Collection object.
-    do {
-        if (!_repositionIfNeeded()) {
-            _mapIter++;  // If the position was not updated, increment iterator to next element.
-        }
+    if (!_repositionIfNeeded()) {
+        _mapIter++;  // If the position was not updated, increment iterator to next element.
+    }
 
-        if (_exhausted()) {
-            // If the iterator is at the end of the map or now points to an entry that does not
-            // correspond to the correct database.
-            _mapIter = _catalog->_orderedCollections.end();
-            _uuid = boost::none;
-            return *this;
-        }
-    } while (_mapIter->second->collectionPtr == nullptr);
+    if (_exhausted()) {
+        // If the iterator is at the end of the map or now points to an entry that does not
+        // correspond to the correct database.
+        _mapIter = _catalog->_orderedCollections.end();
+        _uuid = boost::none;
+        return *this;
+    }
 
     _uuid = _mapIter->first.second;
     return *this;
@@ -172,18 +147,10 @@ bool CollectionCatalog::iterator::_repositionIfNeeded() {
     // If the map has been modified, find the entry the iterator was on, or the one right after it.
     _mapIter = _catalog->_orderedCollections.lower_bound(std::make_pair(_dbName, *_uuid));
 
-    // It is possible that the collection object is gone while catalog entry is
-    // still in the map. Skip that type of entries.
-    while (!_exhausted() && _mapIter->second->collectionPtr == nullptr) {
-        _mapIter++;
-    }
-
     if (_exhausted()) {
         return true;
     }
 
-    invariant(_mapIter->second->collectionPtr);
-
     // If the old pair matches the previous DB name and UUID, the iterator was not repositioned.
     auto dbUuidPair = _mapIter->first;
     bool repositioned = !(dbUuidPair.first == _dbName && dbUuidPair.second == _uuid);
@@ -217,11 +184,8 @@ void CollectionCatalog::setCollectionNamespace(OperationContext* opCtx,
     // Collection's namespace string under '_catalogLock'.
     invariant(coll);
     stdx::lock_guard<stdx::mutex> lock(_catalogLock);
-    CollectionCatalogEntry* catalogEntry =
-        _collections[fromCollection]->collectionCatalogEntry.get();
 
     coll->setNs(toCollection);
-    catalogEntry->setNs(toCollection);
 
     _collections[toCollection] = _collections[fromCollection];
     _collections.erase(fromCollection);
@@ -232,10 +196,9 @@ void CollectionCatalog::setCollectionNamespace(OperationContext* opCtx,
     removeResource(oldRid, fromCollection.ns());
     addResource(newRid, toCollection.ns());
 
-    opCtx->recoveryUnit()->onRollback([this, coll, fromCollection, toCollection, catalogEntry] {
+    opCtx->recoveryUnit()->onRollback([this, coll, fromCollection, toCollection] {
         stdx::lock_guard<stdx::mutex> lock(_catalogLock);
         coll->setNs(std::move(fromCollection));
-        catalogEntry->setNs(fromCollection);
 
         _collections[fromCollection] = _collections[toCollection];
         _collections.erase(toCollection);
@@ -260,7 +223,7 @@ void CollectionCatalog::onCloseCatalog(OperationContext* opCtx) {
     invariant(!_shadowCatalog);
     _shadowCatalog.emplace();
     for (auto& entry : _catalog)
-        _shadowCatalog->insert({entry.first, entry.second.collection->ns()});
+        _shadowCatalog->insert({entry.first, entry.second->ns()});
 }
 
 void CollectionCatalog::onOpenCatalog(OperationContext* opCtx) {
@@ -277,43 +240,20 @@ Collection* CollectionCatalog::lookupCollectionByUUID(CollectionUUID uuid) const
 
 Collection* CollectionCatalog::_lookupCollectionByUUID(WithLock, CollectionUUID uuid) const {
     auto foundIt = _catalog.find(uuid);
-    return foundIt == _catalog.end() || foundIt->second.collectionPtr == nullptr
-        ? nullptr
-        : foundIt->second.collection.get();
+    return foundIt == _catalog.end() ? nullptr : foundIt->second.get();
 }
 
 Collection* CollectionCatalog::lookupCollectionByNamespace(const NamespaceString& nss) const {
     stdx::lock_guard<stdx::mutex> lock(_catalogLock);
     auto it = _collections.find(nss);
-    return it == _collections.end() || it->second->collectionPtr == nullptr
-        ? nullptr
-        : it->second->collection.get();
-}
-
-CollectionCatalogEntry* CollectionCatalog::lookupCollectionCatalogEntryByUUID(
-    CollectionUUID uuid) const {
-    stdx::lock_guard<stdx::mutex> lock(_catalogLock);
-    return _lookupCollectionCatalogEntryByUUID(lock, uuid);
-}
-
-CollectionCatalogEntry* CollectionCatalog::_lookupCollectionCatalogEntryByUUID(
-    WithLock, CollectionUUID uuid) const {
-    auto foundIt = _catalog.find(uuid);
-    return foundIt == _catalog.end() ? nullptr : foundIt->second.collectionCatalogEntry.get();
-}
-
-CollectionCatalogEntry* CollectionCatalog::lookupCollectionCatalogEntryByNamespace(
-    const NamespaceString& nss) const {
-    stdx::lock_guard<stdx::mutex> lock(_catalogLock);
-    auto it = _collections.find(nss);
-    return it == _collections.end() ? nullptr : it->second->collectionCatalogEntry.get();
+    return it == _collections.end() ? nullptr : it->second;
 }
 
 boost::optional<NamespaceString> CollectionCatalog::lookupNSSByUUID(CollectionUUID uuid) const {
     stdx::lock_guard<stdx::mutex> lock(_catalogLock);
     auto foundIt = _catalog.find(uuid);
     if (foundIt != _catalog.end()) {
-        NamespaceString ns = foundIt->second.collectionCatalogEntry->ns();
+        NamespaceString ns = foundIt->second->ns();
         invariant(!ns.isEmpty());
         return ns;
     }
@@ -338,7 +278,7 @@ boost::optional<CollectionUUID> CollectionCatalog::lookupUUIDByNSS(
     // The entry _mapIter points to is valid if it's not at the end of _orderedCollections and
     // the entry's database is the same as dbName.
     while (it != _orderedCollections.end() && it->first.first == nss.db()) {
-        if (it->second->collectionCatalogEntry->ns() == nss) {
+        if (it->second->ns() == nss) {
             return it->first.second;
         }
         ++it;
@@ -352,13 +292,12 @@ bool CollectionCatalog::checkIfCollectionSatisfiable(CollectionUUID uuid,
 
     stdx::lock_guard<stdx::mutex> lock(_catalogLock);
     auto collection = _lookupCollectionByUUID(lock, uuid);
-    auto catalogEntry = _lookupCollectionCatalogEntryByUUID(lock, uuid);
 
-    if (!collection || !catalogEntry) {
+    if (!collection) {
         return false;
     }
 
-    return predicate(collection, catalogEntry);
+    return predicate(collection);
 }
 
 std::vector<CollectionUUID> CollectionCatalog::getAllCollectionUUIDsFromDb(
@@ -386,7 +325,7 @@ std::vector<NamespaceString> CollectionCatalog::getAllCollectionNamesFromDb(
     for (auto it = _orderedCollections.lower_bound(std::make_pair(dbName.toString(), minUuid));
          it != _orderedCollections.end() && it->first.first == dbName;
          ++it) {
-        ret.push_back(it->second->collectionCatalogEntry->ns());
+        ret.push_back(it->second->ns());
     }
     return ret;
 }
@@ -404,16 +343,12 @@ std::vector<std::string> CollectionCatalog::getAllDbNames() const {
     return ret;
 }
 
-void CollectionCatalog::registerCollection(
-    CollectionUUID uuid,
-    std::unique_ptr<CollectionCatalogEntry> collectionCatalogEntry,
-    std::unique_ptr<Collection> coll) {
+void CollectionCatalog::registerCollection(CollectionUUID uuid, std::unique_ptr<Collection> coll) {
     stdx::lock_guard<stdx::mutex> lock(_catalogLock);
 
-    LOG(0) << "Registering catalog entry " << collectionCatalogEntry->ns() << " with UUID " << uuid;
-    LOG(0) << "Registering collection object " << coll->ns() << " with UUID " << uuid;
+    LOG(0) << "Registering collection " << coll->ns() << " with UUID " << uuid;
 
-    auto ns = collectionCatalogEntry->ns();
+    auto ns = coll->ns();
     auto dbName = ns.db().toString();
     auto dbIdPair = std::make_pair(dbName, uuid);
 
@@ -422,12 +357,9 @@ void CollectionCatalog::registerCollection(
     invariant(_collections.find(ns) == _collections.end());
     invariant(_orderedCollections.find(dbIdPair) == _orderedCollections.end());
 
-    auto collPtr = coll.get();
-    CollectionInfo collectionInfo = {std::move(coll), collPtr, std::move(collectionCatalogEntry)};
-
-    _catalog[uuid] = std::move(collectionInfo);
-    _collections[ns] = &_catalog[uuid];
-    _orderedCollections[dbIdPair] = &_catalog[uuid];
+    _catalog[uuid] = std::move(coll);
+    _collections[ns] = _catalog[uuid].get();
+    _orderedCollections[dbIdPair] = _catalog[uuid].get();
 
     auto dbRid = ResourceId(RESOURCE_DATABASE, dbName);
     addResource(dbRid, dbName);
@@ -436,22 +368,17 @@ void CollectionCatalog::registerCollection(
     addResource(collRid, ns.ns());
 }
 
-std::tuple<std::unique_ptr<Collection>, std::unique_ptr<CollectionCatalogEntry>>
-CollectionCatalog::deregisterCollection(CollectionUUID uuid) {
+std::unique_ptr<Collection> CollectionCatalog::deregisterCollection(CollectionUUID uuid) {
     stdx::lock_guard<stdx::mutex> lock(_catalogLock);
 
     invariant(_catalog.find(uuid) != _catalog.end());
-    invariant(_catalog[uuid].collection);
-    invariant(_catalog[uuid].collectionCatalogEntry);
 
-    auto coll = std::move(_catalog[uuid].collection);
-    auto catalogEntry = std::move(_catalog[uuid].collectionCatalogEntry);
+    auto coll = std::move(_catalog[uuid]);
     auto ns = coll->ns();
     auto dbName = ns.db().toString();
     auto dbIdPair = std::make_pair(dbName, uuid);
 
-    LOG(0) << "Deregistering collection object " << ns << " with UUID " << uuid;
-    LOG(0) << "Deregistering catalog entry " << ns << " with UUID " << uuid;
+    LOG(0) << "Deregistering collection " << ns << " with UUID " << uuid;
 
     // Make sure collection object exists.
     invariant(_collections.find(ns) != _collections.end());
@@ -468,30 +395,27 @@ CollectionCatalog::deregisterCollection(CollectionUUID uuid) {
     // references to the erased element.
     _generationNumber++;
 
-    return std::make_tuple(std::move(coll), std::move(catalogEntry));
+    return coll;
 }
 
 RecoveryUnit::Change* CollectionCatalog::makeFinishDropCollectionChange(
-    std::unique_ptr<Collection> coll,
-    std::unique_ptr<CollectionCatalogEntry> catalogEntry,
-    CollectionUUID uuid) {
-    return new FinishDropCollectionChange(this, std::move(coll), std::move(catalogEntry), uuid);
+    std::unique_ptr<Collection> coll, CollectionUUID uuid) {
+    return new FinishDropCollectionChange(this, std::move(coll), uuid);
 }
 
-void CollectionCatalog::deregisterAllCatalogEntriesAndCollectionObjects() {
+void CollectionCatalog::deregisterAllCollections() {
     stdx::lock_guard<stdx::mutex> lock(_catalogLock);
 
-    LOG(0) << "Deregistering all the catalog entries and collection objects";
+    LOG(0) << "Deregistering all the collections";
     for (auto& entry : _catalog) {
         auto uuid = entry.first;
-        auto ns = entry.second.collectionCatalogEntry->ns();
+        auto ns = entry.second->ns();
         auto dbName = ns.db().toString();
         auto dbIdPair = std::make_pair(dbName, uuid);
 
         LOG(0) << "Deregistering collection " << ns << " with UUID " << uuid;
 
-        entry.second.collection.reset();
-        entry.second.collectionCatalogEntry.reset();
+        entry.second.reset();
     }
 
     _collections.clear();
diff --git a/src/mongo/db/catalog/collection_catalog.h b/src/mongo/db/catalog/collection_catalog.h
index fa470daf915..6e508db8f5c 100644
--- a/src/mongo/db/catalog/collection_catalog.h
+++ b/src/mongo/db/catalog/collection_catalog.h
@@ -32,11 +32,10 @@
 #include <functional>
 #include <map>
 #include <set>
-#include <unordered_map>
 
 #include "mongo/db/catalog/collection.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/service_context.h"
+#include "mongo/stdx/unordered_map.h"
 #include "mongo/util/uuid.h"
 
 namespace mongo {
@@ -53,23 +52,18 @@ class CollectionCatalog {
     CollectionCatalog& operator=(const CollectionCatalog&) = delete;
 
     friend class iterator;
-    struct CollectionInfo;
 
 public:
-    using CollectionInfoFn = std::function<bool(const Collection* collection,
-                                                const CollectionCatalogEntry* catalogEntry)>;
+    using CollectionInfoFn = std::function<bool(const Collection* collection)>;
 
     class iterator {
     public:
         using value_type = Collection*;
-        using pointer = const value_type*;
-        using reference = const value_type&;
 
         iterator(StringData dbName, uint64_t genNum, const CollectionCatalog& catalog);
-        iterator(std::map<std::pair<std::string, CollectionUUID>,
-                          CollectionCatalog::CollectionInfo*>::const_iterator mapIter);
-        pointer operator->();
-        reference operator*();
+        iterator(
+            std::map<std::pair<std::string, CollectionUUID>, Collection*>::const_iterator mapIter);
+        const value_type operator*();
         iterator operator++();
         iterator operator++(int);
         boost::optional<CollectionUUID> uuid();
@@ -96,7 +90,7 @@ public:
         std::string _dbName;
         boost::optional<CollectionUUID> _uuid;
         uint64_t _genNum;
-        std::map<std::pair<std::string, CollectionUUID>, CollectionInfo*>::const_iterator _mapIter;
+        std::map<std::pair<std::string, CollectionUUID>, Collection*>::const_iterator _mapIter;
         const CollectionCatalog* _catalog;
         static constexpr Collection* _nullCollection = nullptr;
     };
@@ -120,29 +114,25 @@ public:
     void onCloseDatabase(OperationContext* opCtx, std::string dbName);
 
     /**
-     * Register the collection object and collection catalog entry with `uuid`.
+     * Register the collection with `uuid`.
      */
-    void registerCollection(CollectionUUID uuid,
-                            std::unique_ptr<CollectionCatalogEntry> collectionCatalogEntry,
-                            std::unique_ptr<Collection> collection);
+    void registerCollection(CollectionUUID uuid, std::unique_ptr<Collection> collection);
 
     /**
-     * Deregister the collection object and collection catalog entry.
+     * Deregister the collection.
      */
-    std::tuple<std::unique_ptr<Collection>, std::unique_ptr<CollectionCatalogEntry>>
-    deregisterCollection(CollectionUUID uuid);
+    std::unique_ptr<Collection> deregisterCollection(CollectionUUID uuid);
 
     /**
      * Returns the RecoveryUnit's Change for dropping the collection
      */
     RecoveryUnit::Change* makeFinishDropCollectionChange(std::unique_ptr<Collection>,
-                                                         std::unique_ptr<CollectionCatalogEntry>,
                                                          CollectionUUID uuid);
 
     /**
-     * Deregister all the collection objects and catalog entries.
+     * Deregister all the collection objects.
      */
-    void deregisterAllCatalogEntriesAndCollectionObjects();
+    void deregisterAllCollections();
 
     /**
      * This function gets the Collection pointer that corresponds to the CollectionUUID.
@@ -154,16 +144,6 @@ public:
     Collection* lookupCollectionByUUID(CollectionUUID uuid) const;
 
     /**
-     * This function gets the CollectionCatalogEntry pointer that corresponds to the
-     * CollectionUUID.
-     * The required locks must be obtained prior to calling this function, or else the found
-     * CollectionCatalogEntry pointer might no longer be valid when the call returns.
-     *
-     * Returns nullptr if the 'uuid' is not known.
-     */
-    CollectionCatalogEntry* lookupCollectionCatalogEntryByUUID(CollectionUUID uuid) const;
-
-    /**
      * This function gets the Collection pointer that corresponds to the NamespaceString.
      * The required locks must be obtained prior to calling this function, or else the found
      * Collection pointer may no longer be valid when the call returns.
@@ -173,17 +153,6 @@ public:
     Collection* lookupCollectionByNamespace(const NamespaceString& nss) const;
 
     /**
-     * This function gets the CollectionCatalogEntry pointer that corresponds to the
-     * CollectionUUID.
-     * The required locks must be obtained prior to calling this function, or else the found
-     * CollectionCatalogEntry pointer might no longer be valid when the call returns.
-     *
-     * Returns nullptr if the 'uuid' is not known.
-     */
-    CollectionCatalogEntry* lookupCollectionCatalogEntryByNamespace(
-        const NamespaceString& nss) const;
-
-    /**
      * This function gets the NamespaceString from the collection catalog entry that
      * corresponds to CollectionUUID uuid. If no collection exists with the uuid, return
      * boost::none. See onCloseCatalog/onOpenCatalog for more info.
@@ -270,12 +239,11 @@ private:
     friend class CollectionCatalog::iterator;
 
     Collection* _lookupCollectionByUUID(WithLock, CollectionUUID uuid) const;
-    CollectionCatalogEntry* _lookupCollectionCatalogEntryByUUID(WithLock,
-                                                                CollectionUUID uuid) const;
 
     const std::vector<CollectionUUID>& _getOrdering_inlock(const StringData& db,
                                                            const stdx::lock_guard<stdx::mutex>&);
     mutable mongo::stdx::mutex _catalogLock;
+
     /**
      * When present, indicates that the catalog is in closed state, and contains a map from UUID
      * to pre-close NSS. See also onCloseCatalog.
@@ -284,23 +252,14 @@ private:
         mongo::stdx::unordered_map<CollectionUUID, NamespaceString, CollectionUUID::Hash>>
         _shadowCatalog;
 
-    /**
-     * Unordered map from Collection UUID to the corresponding Collection object and
-     * CollectionCatalogEntry object.
-     */
-    struct CollectionInfo {
-        std::unique_ptr<Collection> collection;
-        Collection* collectionPtr;  // This store the address to the collection object
-        std::unique_ptr<CollectionCatalogEntry> collectionCatalogEntry;
-    };
-    mongo::stdx::unordered_map<CollectionUUID, CollectionInfo, CollectionUUID::Hash> _catalog;
-
-    /**
-     * Ordered map from <database name, collection UUID> to a Collection object.
-     */
-    std::map<std::pair<std::string, CollectionUUID>, CollectionInfo*> _orderedCollections;
-
-    mongo::stdx::unordered_map<NamespaceString, CollectionInfo*> _collections;
+    using CollectionCatalogMap = mongo::stdx::unordered_map<CollectionUUID,
+                                                            std::unique_ptr<Collection>,
+                                                            CollectionUUID::Hash>;
+    using OrderedCollectionMap = std::map<std::pair<std::string, CollectionUUID>, Collection*>;
+    using NamespaceCollectionMap = mongo::stdx::unordered_map<NamespaceString, Collection*>;
+    CollectionCatalogMap _catalog;
+    OrderedCollectionMap _orderedCollections;  // Ordered by <dbName, collUUID> pair
+    NamespaceCollectionMap _collections;
 
     /**
      * Generation number to track changes to the catalog that could invalidate iterators.
diff --git a/src/mongo/db/catalog/collection_catalog_entry.h b/src/mongo/db/catalog/collection_catalog_entry.h
deleted file mode 100644
index cbf53669e47..00000000000
--- a/src/mongo/db/catalog/collection_catalog_entry.h
+++ /dev/null
@@ -1,91 +0,0 @@
-/**
- *    Copyright (C) 2018-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_options.h"
-#include "mongo/db/index/multikey_paths.h"
-#include "mongo/db/namespace_string.h"
-#include "mongo/db/record_id.h"
-#include "mongo/db/server_options.h"
-#include "mongo/db/storage/kv/kv_prefix.h"
-#include "mongo/db/storage/record_store.h"
-
-namespace mongo {
-
-class Collection;
-class IndexDescriptor;
-class OperationContext;
-
-// Indicates which protocol an index build is using.
-enum class IndexBuildProtocol {
-    /**
-     * Refers to the pre-FCV 4.2 index build protocol for building indexes in replica sets.
-     * Index builds must complete on the primary before replicating, and are not resumable in
-     * any scenario.
-     */
-    kSinglePhase,
-    /**
-     * Refers to the FCV 4.2 two-phase index build protocol for building indexes in replica
-     * sets. Indexes are built simultaneously on all nodes and are resumable during the draining
-     * phase.
-     */
-    kTwoPhase
-};
-
-class CollectionCatalogEntry {
-public:
-    /**
-     * Incremented when breaking changes are made to the index build procedure so that other servers
-     * know whether or not to resume or discard unfinished index builds.
-     */
-    static const int kIndexBuildVersion = 1;
-
-    CollectionCatalogEntry(StringData ns) : _ns(ns) {}
-    virtual ~CollectionCatalogEntry() {}
-
-    const NamespaceString& ns() const {
-        return _ns;
-    }
-
-    void setNs(NamespaceString ns) {
-        _ns = std::move(ns);
-    }
-
-    // ------- indexes ----------
-
-
-    virtual RecordStore* getRecordStore() = 0;
-    virtual const RecordStore* getRecordStore() const = 0;
-
-private:
-    NamespaceString _ns;
-};
-}
diff --git a/src/mongo/db/catalog/collection_catalog_entry_mock.h b/src/mongo/db/catalog/collection_catalog_entry_mock.h
deleted file mode 100644
index d3951a43f57..00000000000
--- a/src/mongo/db/catalog/collection_catalog_entry_mock.h
+++ /dev/null
@@ -1,51 +0,0 @@
-/**
- *    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/db/catalog/collection_catalog_entry.h"
-
-namespace mongo {
-class CollectionCatalogEntryMock : public CollectionCatalogEntry {
-public:
-    CollectionCatalogEntryMock(StringData ns) : CollectionCatalogEntry(ns) {}
-
-    CollectionOptions getCollectionOptions(OperationContext* opCtx) const {
-        return CollectionOptions();
-    }
-
-
-    RecordStore* getRecordStore() {
-        return nullptr;
-    }
-    const RecordStore* getRecordStore() const {
-        return nullptr;
-    }
-};
-}
diff --git a/src/mongo/db/catalog/collection_catalog_helper.cpp b/src/mongo/db/catalog/collection_catalog_helper.cpp
index 2a10fef2066..2cb11844d71 100644
--- a/src/mongo/db/catalog/collection_catalog_helper.cpp
+++ b/src/mongo/db/catalog/collection_catalog_helper.cpp
@@ -30,7 +30,6 @@
 #include "mongo/db/catalog/collection_catalog_helper.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/concurrency/d_concurrency.h"
 
 namespace mongo {
@@ -63,11 +62,10 @@ void forEachCollectionFromDb(OperationContext* opCtx,
         opCtx->recoveryUnit()->abandonSnapshot();
 
         auto collection = catalog.lookupCollectionByUUID(uuid);
-        auto catalogEntry = catalog.lookupCollectionCatalogEntryByUUID(uuid);
-        if (!collection || !catalogEntry || catalogEntry->ns() != *nss)
+        if (!collection || collection->ns() != *nss)
             continue;
 
-        if (!callback(collection, catalogEntry))
+        if (!callback(collection))
             break;
 
         MONGO_FAIL_POINT_PAUSE_WHILE_SET(hangBeforeGettingNextCollection);
diff --git a/src/mongo/db/catalog/collection_catalog_test.cpp b/src/mongo/db/catalog/collection_catalog_test.cpp
index 1c335aa4d71..32bf6ab8047 100644
--- a/src/mongo/db/catalog/collection_catalog_test.cpp
+++ b/src/mongo/db/catalog/collection_catalog_test.cpp
@@ -32,7 +32,6 @@
 #include <boost/optional/optional_io.hpp>
 
 #include "mongo/db/catalog/catalog_test_fixture.h"
-#include "mongo/db/catalog/collection_catalog_entry_mock.h"
 #include "mongo/db/catalog/collection_catalog_helper.h"
 #include "mongo/db/catalog/collection_mock.h"
 #include "mongo/db/concurrency/lock_manager_defs.h"
@@ -64,10 +63,9 @@ public:
         ASSERT_GT(nextUUID, colUUID);
 
         auto collection = std::make_unique<CollectionMock>(nss);
-        auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
         col = collection.get();
         // Register dummy collection in catalog.
-        catalog.registerCollection(colUUID, std::move(catalogEntry), std::move(collection));
+        catalog.registerCollection(colUUID, std::move(collection));
     }
 
 protected:
@@ -89,16 +87,14 @@ public:
 
             auto fooUuid = CollectionUUID::gen();
             auto fooColl = std::make_unique<CollectionMock>(fooNss);
-            auto fooCatalogEntry = std::make_unique<CollectionCatalogEntryMock>(fooNss.ns());
 
             auto barUuid = CollectionUUID::gen();
             auto barColl = std::make_unique<CollectionMock>(barNss);
-            auto barCatalogEntry = std::make_unique<CollectionCatalogEntryMock>(barNss.ns());
 
             dbMap["foo"].insert(std::make_pair(fooUuid, fooColl.get()));
             dbMap["bar"].insert(std::make_pair(barUuid, barColl.get()));
-            catalog.registerCollection(fooUuid, std::move(fooCatalogEntry), std::move(fooColl));
-            catalog.registerCollection(barUuid, std::move(barCatalogEntry), std::move(barColl));
+            catalog.registerCollection(fooUuid, std::move(fooColl));
+            catalog.registerCollection(barUuid, std::move(barColl));
         }
     }
 
@@ -273,10 +269,9 @@ public:
         for (int i = 0; i < 5; i++) {
             NamespaceString nss("resourceDb", "coll" + std::to_string(i));
             auto coll = std::make_unique<CollectionMock>(nss);
-            auto newCatalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
             auto uuid = coll->uuid();
 
-            catalog.registerCollection(uuid.get(), std::move(newCatalogEntry), std::move(coll));
+            catalog.registerCollection(uuid.get(), std::move(coll));
         }
 
         int numEntries = 0;
@@ -317,7 +312,7 @@ protected:
 namespace {
 
 TEST_F(CollectionCatalogResourceTest, RemoveAllResources) {
-    catalog.deregisterAllCatalogEntriesAndCollectionObjects();
+    catalog.deregisterAllCollections();
 
     const std::string dbName = "resourceDb";
     auto rid = ResourceId(RESOURCE_DATABASE, dbName);
@@ -498,13 +493,12 @@ TEST_F(CollectionCatalogTest, InsertAfterLookup) {
     auto newUUID = CollectionUUID::gen();
     NamespaceString newNss(nss.db(), "newcol");
     auto newCollUnique = std::make_unique<CollectionMock>(newNss);
-    auto newCatalogEntry = std::make_unique<CollectionCatalogEntryMock>(newNss.ns());
     auto newCol = newCollUnique.get();
 
     // Ensure that looking up non-existing UUIDs doesn't affect later registration of those UUIDs.
     ASSERT(catalog.lookupCollectionByUUID(newUUID) == nullptr);
     ASSERT_EQUALS(catalog.lookupNSSByUUID(newUUID), boost::none);
-    catalog.registerCollection(newUUID, std::move(newCatalogEntry), std::move(newCollUnique));
+    catalog.registerCollection(newUUID, std::move(newCollUnique));
     ASSERT_EQUALS(catalog.lookupCollectionByUUID(newUUID), newCol);
     ASSERT_EQUALS(*catalog.lookupNSSByUUID(colUUID), nss);
 }
@@ -519,9 +513,8 @@ TEST_F(CollectionCatalogTest, RenameCollection) {
     auto uuid = CollectionUUID::gen();
     NamespaceString oldNss(nss.db(), "oldcol");
     auto collUnique = std::make_unique<CollectionMock>(oldNss);
-    auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(oldNss.ns());
     auto collection = collUnique.get();
-    catalog.registerCollection(uuid, std::move(catalogEntry), std::move(collUnique));
+    catalog.registerCollection(uuid, std::move(collUnique));
     ASSERT_EQUALS(catalog.lookupCollectionByUUID(uuid), collection);
 
     NamespaceString newNss(nss.db(), "newcol");
@@ -543,14 +536,13 @@ TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsNewlyCreated
     auto newUUID = CollectionUUID::gen();
     NamespaceString newNss(nss.db(), "newcol");
     auto newCollUnique = std::make_unique<CollectionMock>(newNss);
-    auto newCatalogEntry = std::make_unique<CollectionCatalogEntryMock>(newNss.ns());
     auto newCol = newCollUnique.get();
 
     // Ensure that looking up non-existing UUIDs doesn't affect later registration of those UUIDs.
     catalog.onCloseCatalog(&opCtx);
     ASSERT(catalog.lookupCollectionByUUID(newUUID) == nullptr);
     ASSERT_EQUALS(catalog.lookupNSSByUUID(newUUID), boost::none);
-    catalog.registerCollection(newUUID, std::move(newCatalogEntry), std::move(newCollUnique));
+    catalog.registerCollection(newUUID, std::move(newCollUnique));
     ASSERT_EQUALS(catalog.lookupCollectionByUUID(newUUID), newCol);
     ASSERT_EQUALS(*catalog.lookupNSSByUUID(colUUID), nss);
 
@@ -563,14 +555,13 @@ TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsNewlyCreated
 TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsFreshestNSS) {
     NamespaceString newNss(nss.db(), "newcol");
     auto newCollUnique = std::make_unique<CollectionMock>(newNss);
-    auto newCatalogEntry = std::make_unique<CollectionCatalogEntryMock>(newNss.ns());
     auto newCol = newCollUnique.get();
 
     catalog.onCloseCatalog(&opCtx);
     catalog.deregisterCollection(colUUID);
     ASSERT(catalog.lookupCollectionByUUID(colUUID) == nullptr);
     ASSERT_EQUALS(*catalog.lookupNSSByUUID(colUUID), nss);
-    catalog.registerCollection(colUUID, std::move(newCatalogEntry), std::move(newCollUnique));
+    catalog.registerCollection(colUUID, std::move(newCollUnique));
     ASSERT_EQUALS(catalog.lookupCollectionByUUID(colUUID), newCol);
     ASSERT_EQUALS(*catalog.lookupNSSByUUID(colUUID), newNss);
 
@@ -597,9 +588,8 @@ TEST_F(CollectionCatalogTest, GetAllCollectionNamesAndGetAllDbNames) {
     std::vector<NamespaceString> nsss = {aColl, b1Coll, b2Coll, cColl, d1Coll, d2Coll, d3Coll};
     for (auto& nss : nsss) {
         auto newColl = std::make_unique<CollectionMock>(nss);
-        auto newCatalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
         auto uuid = CollectionUUID::gen();
-        catalog.registerCollection(uuid, std::move(newCatalogEntry), std::move(newColl));
+        catalog.registerCollection(uuid, std::move(newColl));
     }
 
     std::vector<NamespaceString> dCollList = {d1Coll, d2Coll, d3Coll};
@@ -610,7 +600,7 @@ TEST_F(CollectionCatalogTest, GetAllCollectionNamesAndGetAllDbNames) {
     std::vector<std::string> dbNames = {"dbA", "dbB", "dbC", "dbD", "testdb"};
     ASSERT(catalog.getAllDbNames() == dbNames);
 
-    catalog.deregisterAllCatalogEntriesAndCollectionObjects();
+    catalog.deregisterAllCollections();
 }
 
 class ForEachCollectionFromDbTest : public CatalogTestFixture {
@@ -639,16 +629,11 @@ TEST_F(ForEachCollectionFromDbTest, ForEachCollectionFromDb) {
     {
         auto dbLock = std::make_unique<Lock::DBLock>(opCtx, "db", MODE_IX);
         int numCollectionsTraversed = 0;
-        catalog::forEachCollectionFromDb(
-            opCtx,
-            "db",
-            MODE_X,
-            [&](const Collection* collection, const CollectionCatalogEntry* catalogEntry) {
-                ASSERT_TRUE(
-                    opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_X));
-                numCollectionsTraversed++;
-                return true;
-            });
+        catalog::forEachCollectionFromDb(opCtx, "db", MODE_X, [&](const Collection* collection) {
+            ASSERT_TRUE(opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_X));
+            numCollectionsTraversed++;
+            return true;
+        });
 
         ASSERT_EQUALS(numCollectionsTraversed, 3);
     }
@@ -656,16 +641,11 @@ TEST_F(ForEachCollectionFromDbTest, ForEachCollectionFromDb) {
     {
         auto dbLock = std::make_unique<Lock::DBLock>(opCtx, "db2", MODE_IX);
         int numCollectionsTraversed = 0;
-        catalog::forEachCollectionFromDb(
-            opCtx,
-            "db2",
-            MODE_IS,
-            [&](const Collection* collection, const CollectionCatalogEntry* catalogEntry) {
-                ASSERT_TRUE(
-                    opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_IS));
-                numCollectionsTraversed++;
-                return true;
-            });
+        catalog::forEachCollectionFromDb(opCtx, "db2", MODE_IS, [&](const Collection* collection) {
+            ASSERT_TRUE(opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_IS));
+            numCollectionsTraversed++;
+            return true;
+        });
 
         ASSERT_EQUALS(numCollectionsTraversed, 1);
     }
@@ -673,14 +653,10 @@ TEST_F(ForEachCollectionFromDbTest, ForEachCollectionFromDb) {
     {
         auto dbLock = std::make_unique<Lock::DBLock>(opCtx, "db3", MODE_IX);
         int numCollectionsTraversed = 0;
-        catalog::forEachCollectionFromDb(
-            opCtx,
-            "db3",
-            MODE_S,
-            [&](const Collection* collection, const CollectionCatalogEntry* catalogEntry) {
-                numCollectionsTraversed++;
-                return true;
-            });
+        catalog::forEachCollectionFromDb(opCtx, "db3", MODE_S, [&](const Collection* collection) {
+            numCollectionsTraversed++;
+            return true;
+        });
 
         ASSERT_EQUALS(numCollectionsTraversed, 0);
     }
@@ -697,13 +673,13 @@ TEST_F(ForEachCollectionFromDbTest, ForEachCollectionFromDbWithPredicate) {
             opCtx,
             "db",
             MODE_X,
-            [&](const Collection* collection, const CollectionCatalogEntry* catalogEntry) {
+            [&](const Collection* collection) {
                 ASSERT_TRUE(
                     opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_X));
                 numCollectionsTraversed++;
                 return true;
             },
-            [&](const Collection* collection, const CollectionCatalogEntry* catalogEntry) {
+            [&](const Collection* collection) {
                 ASSERT_TRUE(
                     opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_NONE));
                 return DurableCatalog::get(opCtx)
@@ -721,13 +697,13 @@ TEST_F(ForEachCollectionFromDbTest, ForEachCollectionFromDbWithPredicate) {
             opCtx,
             "db",
             MODE_IX,
-            [&](const Collection* collection, const CollectionCatalogEntry* catalogEntry) {
+            [&](const Collection* collection) {
                 ASSERT_TRUE(
                     opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_IX));
                 numCollectionsTraversed++;
                 return true;
             },
-            [&](const Collection* collection, const CollectionCatalogEntry* catalogEntry) {
+            [&](const Collection* collection) {
                 ASSERT_TRUE(
                     opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_NONE));
                 return !DurableCatalog::get(opCtx)
diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp
index 59b64121247..884f1ca87f3 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -43,7 +43,6 @@
 #include "mongo/bson/simple_bsonobj_comparator.h"
 #include "mongo/db/background.h"
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/collection_info_cache_impl.h"
 #include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/catalog/document_validation.h"
@@ -196,15 +195,13 @@ using std::vector;
 using logger::LogComponent;
 
 CollectionImpl::CollectionImpl(OperationContext* opCtx,
-                               StringData fullNS,
+                               const NamespaceString& nss,
                                OptionalCollectionUUID uuid,
-                               CollectionCatalogEntry* details,
-                               RecordStore* recordStore)
+                               std::unique_ptr<RecordStore> recordStore)
     : _magic(kMagicNumber),
-      _ns(fullNS),
+      _ns(nss),
       _uuid(uuid),
-      _details(details),
-      _recordStore(recordStore),
+      _recordStore(std::move(recordStore)),
       _needCappedLock(supportsDocLocking() && _recordStore && _recordStore->isCapped() &&
                       _ns.db() != "local"),
       _infoCache(std::make_unique<CollectionInfoCacheImpl>(this, _ns)),
@@ -232,11 +229,10 @@ CollectionImpl::~CollectionImpl() {
 
 std::unique_ptr<Collection> CollectionImpl::FactoryImpl::make(
     OperationContext* opCtx,
+    const NamespaceString& nss,
     CollectionUUID uuid,
-    CollectionCatalogEntry* collectionCatalogEntry) const {
-    auto rs = collectionCatalogEntry->getRecordStore();
-    const auto nss = collectionCatalogEntry->ns();
-    return std::make_unique<CollectionImpl>(opCtx, nss.ns(), uuid, collectionCatalogEntry, rs);
+    std::unique_ptr<RecordStore> rs) const {
+    return std::make_unique<CollectionImpl>(opCtx, nss, uuid, std::move(rs));
 }
 
 void CollectionImpl::init(OperationContext* opCtx) {
@@ -1335,7 +1331,7 @@ Status CollectionImpl::validate(OperationContext* opCtx,
     try {
         ValidateResultsMap indexNsResultsMap;
         BSONObjBuilder keysPerIndex;  // not using subObjStart to be exception safe
-        IndexConsistency indexConsistency(opCtx, this, ns(), _recordStore, background);
+        IndexConsistency indexConsistency(opCtx, this, ns(), _recordStore.get(), background);
         RecordStoreValidateAdaptor indexValidator = RecordStoreValidateAdaptor(
             opCtx, &indexConsistency, level, _indexCatalog.get(), &indexNsResultsMap);
 
@@ -1344,7 +1340,7 @@ Status CollectionImpl::validate(OperationContext* opCtx,
             << " (UUID: " << (uuid() ? uuid()->toString() : "none") << ")";
         log(LogComponent::kIndex) << "validating collection " << ns() << uuidString;
         _validateRecordStore(
-            opCtx, _recordStore, level, background, &indexValidator, results, output);
+            opCtx, _recordStore.get(), level, background, &indexValidator, results, output);
 
         // Validate in-memory catalog information with the persisted info.
         _validateCatalogEntry(opCtx, this, _validatorDoc, results);
@@ -1364,7 +1360,7 @@ Status CollectionImpl::validate(OperationContext* opCtx,
                     << "Index inconsistencies were detected on collection " << ns()
                     << ". Starting the second phase of index validation to gather concise errors.";
                 _gatherIndexEntryErrors(opCtx,
-                                        _recordStore,
+                                        _recordStore.get(),
                                         _indexCatalog.get(),
                                         &indexConsistency,
                                         &indexValidator,
@@ -1375,8 +1371,11 @@ Status CollectionImpl::validate(OperationContext* opCtx,
 
         // Validate index key count.
         if (results->valid) {
-            _validateIndexKeyCount(
-                opCtx, _indexCatalog.get(), _recordStore, &indexValidator, &indexNsResultsMap);
+            _validateIndexKeyCount(opCtx,
+                                   _indexCatalog.get(),
+                                   _recordStore.get(),
+                                   &indexValidator,
+                                   &indexNsResultsMap);
         }
 
         // Report the validation results for the user to see
@@ -1407,7 +1406,7 @@ Status CollectionImpl::touch(OperationContext* opCtx,
                              BSONObjBuilder* output) const {
     if (touchData) {
         BSONObjBuilder b;
-        Status status = _recordStore->touch(opCtx, &b);
+        Status status = _recordStore.get()->touch(opCtx, &b);
         if (!status.isOK())
             return status;
         output->append("data", b.obj());
@@ -1444,7 +1443,7 @@ void CollectionImpl::setNs(NamespaceString nss) {
     _ns = std::move(nss);
     _indexCatalog->setNs(_ns);
     _infoCache->setNs(_ns);
-    _recordStore->setNs(_ns);
+    _recordStore.get()->setNs(_ns);
 }
 
 void CollectionImpl::indexBuildSuccess(OperationContext* opCtx, IndexCatalogEntry* index) {
diff --git a/src/mongo/db/catalog/collection_impl.h b/src/mongo/db/catalog/collection_impl.h
index b91028a847f..180becf62d8 100644
--- a/src/mongo/db/catalog/collection_impl.h
+++ b/src/mongo/db/catalog/collection_impl.h
@@ -31,7 +31,6 @@
 
 #include "mongo/bson/timestamp.h"
 #include "mongo/db/catalog/collection.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/concurrency/d_concurrency.h"
 
@@ -47,33 +46,24 @@ public:
     enum ValidationLevel { OFF, MODERATE, STRICT_V };
 
     explicit CollectionImpl(OperationContext* opCtx,
-                            StringData fullNS,
+                            const NamespaceString& nss,
                             OptionalCollectionUUID uuid,
-                            CollectionCatalogEntry* details,
-                            RecordStore* recordStore);
+                            std::unique_ptr<RecordStore> recordStore);
 
     ~CollectionImpl();
 
     class FactoryImpl : public Factory {
     public:
-        std::unique_ptr<Collection> make(
-            OperationContext* opCtx,
-            CollectionUUID uuid,
-            CollectionCatalogEntry* collectionCatalogEntry) const final;
+        std::unique_ptr<Collection> make(OperationContext* opCtx,
+                                         const NamespaceString& nss,
+                                         CollectionUUID uuid,
+                                         std::unique_ptr<RecordStore> rs) const final;
     };
 
     bool ok() const final {
         return _magic == kMagicNumber;
     }
 
-    CollectionCatalogEntry* getCatalogEntry() final {
-        return _details;
-    }
-
-    const CollectionCatalogEntry* getCatalogEntry() const final {
-        return _details;
-    }
-
     CollectionInfoCache* infoCache() final {
         return _infoCache.get();
     }
@@ -101,11 +91,11 @@ public:
     }
 
     const RecordStore* getRecordStore() const final {
-        return _recordStore;
+        return _recordStore.get();
     }
 
     RecordStore* getRecordStore() final {
-        return _recordStore;
+        return _recordStore.get();
     }
 
     bool requiresIdIndex() const final;
@@ -401,10 +391,9 @@ private:
 
     NamespaceString _ns;
     OptionalCollectionUUID _uuid;
-    CollectionCatalogEntry* const _details;
 
     // The RecordStore may be null during a repair operation.
-    RecordStore* const _recordStore;
+    std::unique_ptr<RecordStore> _recordStore;  // owned
     const bool _needCappedLock;
     std::unique_ptr<CollectionInfoCache> _infoCache;
     std::unique_ptr<IndexCatalog> _indexCatalog;
diff --git a/src/mongo/db/catalog/collection_mock.h b/src/mongo/db/catalog/collection_mock.h
index 645205a1828..a3efa28633c 100644
--- a/src/mongo/db/catalog/collection_mock.h
+++ b/src/mongo/db/catalog/collection_mock.h
@@ -60,13 +60,6 @@ public:
         std::abort();
     }
 
-    CollectionCatalogEntry* getCatalogEntry() {
-        std::abort();
-    }
-    const CollectionCatalogEntry* getCatalogEntry() const {
-        std::abort();
-    }
-
     CollectionInfoCache* infoCache() {
         std::abort();
     }
diff --git a/src/mongo/db/catalog/create_collection_test.cpp b/src/mongo/db/catalog/create_collection_test.cpp
index c9ebc485480..af99db09759 100644
--- a/src/mongo/db/catalog/create_collection_test.cpp
+++ b/src/mongo/db/catalog/create_collection_test.cpp
@@ -32,7 +32,6 @@
 #include <memory>
 
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/create_collection.h"
 #include "mongo/db/client.h"
 #include "mongo/db/db_raii.h"
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index 12853dd15bc..0b537c99e2e 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -43,7 +43,6 @@
 #include "mongo/db/audit.h"
 #include "mongo/db/background.h"
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/collection_catalog_helper.h"
 #include "mongo/db/catalog/collection_impl.h"
 #include "mongo/db/catalog/collection_options.h"
@@ -165,8 +164,7 @@ void DatabaseImpl::init(OperationContext* const opCtx) const {
 void DatabaseImpl::clearTmpCollections(OperationContext* opCtx) const {
     invariant(opCtx->lockState()->isDbLockedForMode(name(), MODE_IX));
 
-    CollectionCatalog::CollectionInfoFn callback = [&](const Collection* collection,
-                                                       const CollectionCatalogEntry* catalogEntry) {
+    CollectionCatalog::CollectionInfoFn callback = [&](const Collection* collection) {
         try {
             WriteUnitOfWork wuow(opCtx);
             Status status = dropCollection(opCtx, collection->ns(), {});
@@ -183,10 +181,9 @@ void DatabaseImpl::clearTmpCollections(OperationContext* opCtx) const {
         return true;
     };
 
-    CollectionCatalog::CollectionInfoFn predicate =
-        [&](const Collection* collection, const CollectionCatalogEntry* catalogEntry) {
-            return DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, collection->ns()).temp;
-        };
+    CollectionCatalog::CollectionInfoFn predicate = [&](const Collection* collection) {
+        return DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, collection->ns()).temp;
+    };
 
     catalog::forEachCollectionFromDb(opCtx, name(), MODE_X, callback, predicate);
 }
@@ -248,10 +245,7 @@ void DatabaseImpl::getStats(OperationContext* opCtx, BSONObjBuilder* output, dou
     invariant(opCtx->lockState()->isDbLockedForMode(name(), MODE_IS));
 
     catalog::forEachCollectionFromDb(
-        opCtx,
-        name(),
-        MODE_IS,
-        [&](const Collection* collection, const CollectionCatalogEntry* catalogEntry) -> bool {
+        opCtx, name(), MODE_IS, [&](const Collection* collection) -> bool {
             nCollections += 1;
             objects += collection->numRecords(opCtx);
             size += collection->dataSize(opCtx);
@@ -468,11 +462,9 @@ Status DatabaseImpl::_finishDropCollection(OperationContext* opCtx,
     if (!status.isOK())
         return status;
 
-    auto[removedColl, removedCatalogEntry] =
-        CollectionCatalog::get(opCtx).deregisterCollection(uuid);
+    auto removedColl = CollectionCatalog::get(opCtx).deregisterCollection(uuid);
     opCtx->recoveryUnit()->registerChange(
-        CollectionCatalog::get(opCtx).makeFinishDropCollectionChange(
-            std::move(removedColl), std::move(removedCatalogEntry), uuid));
+        CollectionCatalog::get(opCtx).makeFinishDropCollectionChange(std::move(removedColl), uuid));
 
     return Status::OK();
 }
@@ -642,16 +634,14 @@ Collection* DatabaseImpl::createCollection(OperationContext* opCtx,
     log() << "createCollection: " << nss << " with " << (generatedUUID ? "generated" : "provided")
           << " UUID: " << optionsWithUUID.uuid.get() << " and options: " << options.toBSON();
 
-    // Create CollectionCatalogEntry
-    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
+    auto storageEngine = opCtx->getServiceContext()->getStorageEngine();
+    auto statusWithRecordStore = storageEngine->getCatalog()->createCollection(
+        opCtx, nss, optionsWithUUID, true /*allocateDefaultSpace*/);
+    massertStatusOK(statusWithRecordStore.getStatus());
+    std::unique_ptr<RecordStore> rs = std::move(statusWithRecordStore.getValue());
     std::unique_ptr<Collection> ownedCollection = Collection::Factory::get(opCtx)->make(
-        opCtx, optionsWithUUID.uuid.get(), ownedCatalogEntry.get());
+        opCtx, nss, optionsWithUUID.uuid.get(), std::move(rs));
     auto collection = ownedCollection.get();
     ownedCollection->init(opCtx);
 
@@ -663,12 +653,8 @@ Collection* DatabaseImpl::createCollection(OperationContext* opCtx,
 
     auto& catalog = CollectionCatalog::get(opCtx);
     auto uuid = ownedCollection->uuid().get();
-    catalog.registerCollection(uuid, std::move(ownedCatalogEntry), std::move(ownedCollection));
-    opCtx->recoveryUnit()->onRollback([uuid, &catalog] {
-        auto[removedColl, removedCatalogEntry] = catalog.deregisterCollection(uuid);
-        removedColl.reset();
-        removedCatalogEntry.reset();
-    });
+    catalog.registerCollection(uuid, std::move(ownedCollection));
+    opCtx->recoveryUnit()->onRollback([uuid, &catalog] { catalog.deregisterCollection(uuid); });
 
     BSONObj fullIdIndexSpec;
 
diff --git a/src/mongo/db/catalog/database_test.cpp b/src/mongo/db/catalog/database_test.cpp
index 53f58eb1cb5..32d779b093d 100644
--- a/src/mongo/db/catalog/database_test.cpp
+++ b/src/mongo/db/catalog/database_test.cpp
@@ -36,7 +36,6 @@
 
 #include "mongo/bson/util/builder.h"
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/client.h"
 #include "mongo/db/concurrency/d_concurrency.h"
diff --git a/src/mongo/db/catalog/index_build_block.cpp b/src/mongo/db/catalog/index_build_block.cpp
index ab732f89fab..2857e1f0e7e 100644
--- a/src/mongo/db/catalog/index_build_block.cpp
+++ b/src/mongo/db/catalog/index_build_block.cpp
@@ -36,7 +36,6 @@
 #include <vector>
 
 #include "mongo/db/catalog/collection.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/index/index_descriptor.h"
diff --git a/src/mongo/db/catalog/index_builds_manager.cpp b/src/mongo/db/catalog/index_builds_manager.cpp
index 18b19bbd85e..cbc9217ee12 100644
--- a/src/mongo/db/catalog/index_builds_manager.cpp
+++ b/src/mongo/db/catalog/index_builds_manager.cpp
@@ -35,7 +35,6 @@
 
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/catalog/index_timestamp_helper.h"
 #include "mongo/db/concurrency/write_conflict_exception.h"
@@ -139,8 +138,8 @@ StatusWith<std::pair<long long, long long>> IndexBuildsManager::startBuildingInd
     OperationContext* opCtx, NamespaceString ns, const UUID& buildUUID) {
     auto builder = _getBuilder(buildUUID);
 
-    auto cce = CollectionCatalog::get(opCtx).lookupCollectionCatalogEntryByNamespace(ns);
-    auto rs = cce ? cce->getRecordStore() : nullptr;
+    auto coll = CollectionCatalog::get(opCtx).lookupCollectionByNamespace(ns);
+    auto rs = coll ? coll->getRecordStore() : nullptr;
 
     // Iterate all records in the collection. Delete them if they aren't valid BSON. Index them
     // if they are.
diff --git a/src/mongo/db/catalog/index_catalog_entry_impl.cpp b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
index 6736d545453..80a6215a139 100644
--- a/src/mongo/db/catalog/index_catalog_entry_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
@@ -37,7 +37,6 @@
 #include <memory>
 
 #include "mongo/base/init.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/collection_info_cache_impl.h"
 #include "mongo/db/concurrency/d_concurrency.h"
 #include "mongo/db/concurrency/write_conflict_exception.h"
@@ -51,7 +50,6 @@
 #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"
@@ -61,18 +59,15 @@ namespace mongo {
 using std::string;
 
 IndexCatalogEntryImpl::IndexCatalogEntryImpl(OperationContext* const opCtx,
-                                             const StringData ns,
-                                             CollectionCatalogEntry* const collection,
+                                             const NamespaceString& nss,
                                              std::unique_ptr<IndexDescriptor> descriptor,
                                              CollectionInfoCache* const infoCache)
-    : _ns(ns.toString()),
-      _collection(collection),
+    : _ns(nss),
       _descriptor(std::move(descriptor)),
       _infoCache(infoCache),
       _ordering(Ordering::make(_descriptor->keyPattern())),
       _isReady(false),
-      _prefix(DurableCatalog::get(opCtx)->getIndexPrefix(
-          opCtx, collection->ns(), _descriptor->indexName())) {
+      _prefix(DurableCatalog::get(opCtx)->getIndexPrefix(opCtx, nss, _descriptor->indexName())) {
     _descriptor->_cachedEntry = this;
 
     _isReady = _catalogIsReady(opCtx);
@@ -256,7 +251,7 @@ void IndexCatalogEntryImpl::setMultikey(OperationContext* opCtx,
     // OperationContext and we can safely defer that write to the end of the batch.
     if (MultikeyPathTracker::get(opCtx).isTrackingMultikeyPathInfo()) {
         MultikeyPathInfo info;
-        info.nss = _collection->ns();
+        info.nss = _ns;
         info.indexName = _descriptor->indexName();
         info.multikeyPaths = paths;
         MultikeyPathTracker::get(opCtx).addMultikeyPathInfo(info);
@@ -299,7 +294,7 @@ void IndexCatalogEntryImpl::setMultikey(OperationContext* opCtx,
     auto txnParticipant = TransactionParticipant::get(opCtx);
     if (txnParticipant && txnParticipant.inMultiDocumentTransaction()) {
         TransactionParticipant::SideTransactionBlock sideTxn(opCtx);
-        writeConflictRetry(opCtx, "set index multikey", _collection->ns().ns(), [&] {
+        writeConflictRetry(opCtx, "set index multikey", _ns.ns(), [&] {
             WriteUnitOfWork wuow(opCtx);
             auto writeTs = LogicalClock::get(opCtx)->getClusterTime().asTimestamp();
             auto status = opCtx->recoveryUnit()->setTimestamp(writeTs);
@@ -310,14 +305,14 @@ void IndexCatalogEntryImpl::setMultikey(OperationContext* opCtx,
             }
             fassert(31164, status);
             indexMetadataHasChanged = DurableCatalog::get(opCtx)->setIndexIsMultikey(
-                opCtx, _collection->ns(), _descriptor->indexName(), paths);
+                opCtx, _ns, _descriptor->indexName(), paths);
             opCtx->recoveryUnit()->onCommit([onMultikeyCommitFn, indexMetadataHasChanged](
                 boost::optional<Timestamp>) { onMultikeyCommitFn(indexMetadataHasChanged); });
             wuow.commit();
         });
     } else {
         indexMetadataHasChanged = DurableCatalog::get(opCtx)->setIndexIsMultikey(
-            opCtx, _collection->ns(), _descriptor->indexName(), paths);
+            opCtx, _ns, _descriptor->indexName(), paths);
     }
 
     opCtx->recoveryUnit()->onCommit([onMultikeyCommitFn, indexMetadataHasChanged](
@@ -332,7 +327,7 @@ void IndexCatalogEntryImpl::setMultikey(OperationContext* opCtx,
     // previous write in the transaction.
     if (txnParticipant && txnParticipant.inMultiDocumentTransaction()) {
         txnParticipant.addUncommittedMultikeyPathInfo(
-            MultikeyPathInfo{_collection->ns(), _descriptor->indexName(), std::move(paths)});
+            MultikeyPathInfo{_ns, _descriptor->indexName(), std::move(paths)});
     }
 }
 
@@ -344,24 +339,21 @@ void IndexCatalogEntryImpl::setNs(NamespaceString ns) {
 // ----
 
 bool IndexCatalogEntryImpl::_catalogIsReady(OperationContext* opCtx) const {
-    return DurableCatalog::get(opCtx)->isIndexReady(
-        opCtx, _collection->ns(), _descriptor->indexName());
+    return DurableCatalog::get(opCtx)->isIndexReady(opCtx, _ns, _descriptor->indexName());
 }
 
 bool IndexCatalogEntryImpl::_catalogIsPresent(OperationContext* opCtx) const {
-    return DurableCatalog::get(opCtx)->isIndexPresent(
-        opCtx, _collection->ns(), _descriptor->indexName());
+    return DurableCatalog::get(opCtx)->isIndexPresent(opCtx, _ns, _descriptor->indexName());
 }
 
 bool IndexCatalogEntryImpl::_catalogIsMultikey(OperationContext* opCtx,
                                                MultikeyPaths* multikeyPaths) const {
     return DurableCatalog::get(opCtx)->isIndexMultikey(
-        opCtx, _collection->ns(), _descriptor->indexName(), multikeyPaths);
+        opCtx, _ns, _descriptor->indexName(), multikeyPaths);
 }
 
 KVPrefix IndexCatalogEntryImpl::_catalogGetPrefix(OperationContext* opCtx) const {
-    return DurableCatalog::get(opCtx)->getIndexPrefix(
-        opCtx, _collection->ns(), _descriptor->indexName());
+    return DurableCatalog::get(opCtx)->getIndexPrefix(opCtx, _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 f7da6b660a5..cb697a3d177 100644
--- a/src/mongo/db/catalog/index_catalog_entry_impl.h
+++ b/src/mongo/db/catalog/index_catalog_entry_impl.h
@@ -46,7 +46,6 @@
 namespace mongo {
 
 class CollatorInterface;
-class CollectionCatalogEntry;
 class CollectionInfoCache;
 class IndexAccessMethod;
 class IndexDescriptor;
@@ -60,8 +59,7 @@ class IndexCatalogEntryImpl : public IndexCatalogEntry {
 public:
     explicit IndexCatalogEntryImpl(
         OperationContext* opCtx,
-        StringData ns,
-        CollectionCatalogEntry* collection,           // not owned
+        const NamespaceString& nss,
         std::unique_ptr<IndexDescriptor> descriptor,  // ownership passes to me
         CollectionInfoCache* infoCache);              // not owned, optional
 
@@ -196,8 +194,6 @@ private:
 
     NamespaceString _ns;
 
-    CollectionCatalogEntry* _collection;  // not owned here
-
     std::unique_ptr<IndexDescriptor> _descriptor;  // owned here
 
     CollectionInfoCache* _infoCache;  // not owned here
diff --git a/src/mongo/db/catalog/index_catalog_impl.cpp b/src/mongo/db/catalog/index_catalog_impl.cpp
index f74dfbdb3b9..1ad82134514 100644
--- a/src/mongo/db/catalog/index_catalog_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_impl.cpp
@@ -41,7 +41,6 @@
 #include "mongo/db/audit.h"
 #include "mongo/db/background.h"
 #include "mongo/db/catalog/collection.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/disable_index_spec_namespace_generation_gen.h"
 #include "mongo/db/catalog/index_catalog_entry_impl.h"
 #include "mongo/db/catalog/index_key_validate.h"
@@ -138,11 +137,8 @@ IndexCatalogEntry* IndexCatalogImpl::_setupInMemoryStructures(
     }
 
     auto* const descriptorPtr = descriptor.get();
-    auto entry = std::make_shared<IndexCatalogEntryImpl>(opCtx,
-                                                         _collection->ns().ns(),
-                                                         _collection->getCatalogEntry(),
-                                                         std::move(descriptor),
-                                                         _collection->infoCache());
+    auto entry = std::make_shared<IndexCatalogEntryImpl>(
+        opCtx, _collection->ns(), std::move(descriptor), _collection->infoCache());
 
     IndexDescriptor* desc = entry->descriptor();
 
diff --git a/src/mongo/db/catalog/index_consistency.cpp b/src/mongo/db/catalog/index_consistency.cpp
index ce328ef417b..ef2585bd0c1 100644
--- a/src/mongo/db/catalog/index_consistency.cpp
+++ b/src/mongo/db/catalog/index_consistency.cpp
@@ -32,7 +32,6 @@
 
 #include "mongo/platform/basic.h"
 
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/catalog/index_consistency.h"
diff --git a/src/mongo/db/catalog/rename_collection.cpp b/src/mongo/db/catalog/rename_collection.cpp
index b575f1f5d88..bb3d6f97f32 100644
--- a/src/mongo/db/catalog/rename_collection.cpp
+++ b/src/mongo/db/catalog/rename_collection.cpp
@@ -35,7 +35,6 @@
 
 #include "mongo/db/background.h"
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/document_validation.h"
 #include "mongo/db/catalog/drop_collection.h"
diff --git a/src/mongo/db/catalog/rename_collection_test.cpp b/src/mongo/db/catalog/rename_collection_test.cpp
index 249d79928f6..0b9c2d1639d 100644
--- a/src/mongo/db/catalog/rename_collection_test.cpp
+++ b/src/mongo/db/catalog/rename_collection_test.cpp
@@ -35,7 +35,6 @@
 #include <vector>
 
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/index_catalog.h"
@@ -1141,45 +1140,6 @@ TEST_F(RenameCollectionTest, CollectionPointerRemainsValidThroughRename) {
     ASSERT_EQ(targetColl->ns(), _targetNss);
 }
 
-TEST_F(RenameCollectionTest, CollectionCatalogEntryPointerRemainsValidThroughRename) {
-    _createCollection(_opCtx.get(), _sourceNss);
-    Lock::GlobalWrite globalWrite(_opCtx.get());
-
-    // Get a pointer to the source collection, and ensure that it reports the expected namespace
-    // string.
-    Collection* sourceColl = _getCollection_inlock(_opCtx.get(), _sourceNss);
-    ASSERT(sourceColl);
-    auto* sourceCatalogEntry = sourceColl->getCatalogEntry();
-    ASSERT(sourceCatalogEntry);
-    ASSERT_EQ(sourceCatalogEntry->ns(), _sourceNss);
-
-    ASSERT_OK(renameCollection(_opCtx.get(), _sourceNss, _targetNss, {}));
-
-    // Verify that the CollectionCatalogEntry reports that its namespace is now the target
-    // namespace.
-    ASSERT_EQ(sourceCatalogEntry->ns(), _targetNss);
-}
-
-TEST_F(RenameCollectionTest, CatalogPointersRenameValidThroughRenameAfterDroppingTarget) {
-    _createCollection(_opCtx.get(), _sourceNss);
-    _createCollection(_opCtx.get(), _targetNss);
-    Lock::GlobalWrite globalWrite(_opCtx.get());
-
-    Collection* sourceColl = _getCollection_inlock(_opCtx.get(), _sourceNss);
-    ASSERT(sourceColl);
-    auto* sourceCatalogEntry = sourceColl->getCatalogEntry();
-    ASSERT(sourceCatalogEntry);
-
-    RenameCollectionOptions options;
-    options.dropTarget = true;
-    ASSERT_OK(renameCollection(_opCtx.get(), _sourceNss, _targetNss, options));
-
-    // The same catalog pointers should now report that they are associated with the target
-    // namespace.
-    ASSERT_EQ(sourceColl->ns(), _targetNss);
-    ASSERT_EQ(sourceCatalogEntry->ns(), _targetNss);
-}
-
 TEST_F(RenameCollectionTest, CatalogPointersRenameValidThroughRenameForApplyOps) {
     _createCollection(_opCtx.get(), _sourceNss);
     Collection* sourceColl = AutoGetCollectionForRead(_opCtx.get(), _sourceNss).getCollection();
@@ -1197,29 +1157,6 @@ TEST_F(RenameCollectionTest, CatalogPointersRenameValidThroughRenameForApplyOps)
     ASSERT_EQ(targetColl->ns(), _targetNss);
 }
 
-TEST_F(RenameCollectionTest, RenameAcrossDatabasesDoesNotPreserveCatalogPointers) {
-    _createCollection(_opCtx.get(), _sourceNss);
-    Lock::GlobalWrite globalWrite(_opCtx.get());
-
-    // Get a pointer to the source collection, and ensure that it reports the expected namespace
-    // string.
-    Collection* sourceColl = _getCollection_inlock(_opCtx.get(), _sourceNss);
-    ASSERT(sourceColl);
-    auto* sourceCatalogEntry = sourceColl->getCatalogEntry();
-    ASSERT(sourceCatalogEntry);
-
-    ASSERT_OK(renameCollection(_opCtx.get(), _sourceNss, _targetNssDifferentDb, {}));
-
-    // Verify that the CollectionCatalogEntry reports that its namespace is now the target
-    // namespace.
-    Collection* targetColl = _getCollection_inlock(_opCtx.get(), _targetNssDifferentDb);
-    ASSERT(targetColl);
-    ASSERT_NE(targetColl, sourceColl);
-    auto* targetCatalogEntry = targetColl->getCatalogEntry();
-    ASSERT(targetCatalogEntry);
-    ASSERT_NE(targetCatalogEntry, sourceCatalogEntry);
-}
-
 TEST_F(RenameCollectionTest, CollectionCatalogMappingRemainsIntactThroughRename) {
     _createCollection(_opCtx.get(), _sourceNss);
     Lock::GlobalWrite globalWrite(_opCtx.get());
diff --git a/src/mongo/db/cloner.cpp b/src/mongo/db/cloner.cpp
index f2d86cd8f64..60baaaa1e59 100644
--- a/src/mongo/db/cloner.cpp
+++ b/src/mongo/db/cloner.cpp
@@ -40,7 +40,6 @@
 #include "mongo/bson/util/builder.h"
 #include "mongo/client/authenticate.h"
 #include "mongo/db/catalog/collection.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/catalog/database_holder.h"
diff --git a/src/mongo/db/commands/dbhash.cpp b/src/mongo/db/commands/dbhash.cpp
index b24fcc3fd45..473b323f6b5 100644
--- a/src/mongo/db/commands/dbhash.cpp
+++ b/src/mongo/db/commands/dbhash.cpp
@@ -228,55 +228,50 @@ public:
         std::set<std::string> cappedCollectionSet;
 
         bool noError = true;
-        catalog::forEachCollectionFromDb(
-            opCtx,
-            dbname,
-            MODE_IS,
-            [&](const Collection* collection, const CollectionCatalogEntry* catalogEntry) {
-                auto collNss = collection->ns();
-
-                if (collNss.size() - 1 <= dbname.size()) {
-                    errmsg = str::stream() << "weird fullCollectionName [" << collNss.toString()
-                                           << "]";
-                    noError = false;
-                    return false;
-                }
+        catalog::forEachCollectionFromDb(opCtx, dbname, MODE_IS, [&](const Collection* collection) {
+            auto collNss = collection->ns();
 
-                // Only include 'system' collections that are replicated.
-                bool isReplicatedSystemColl =
-                    (replicatedSystemCollections.count(collNss.coll().toString()) > 0);
-                if (collNss.isSystem() && !isReplicatedSystemColl)
-                    return true;
+            if (collNss.size() - 1 <= dbname.size()) {
+                errmsg = str::stream() << "weird fullCollectionName [" << collNss.toString() << "]";
+                noError = false;
+                return false;
+            }
 
-                if (collNss.coll().startsWith("tmp.mr.")) {
-                    // We skip any incremental map reduce collections as they also aren't
-                    // replicated.
-                    return true;
-                }
+            // Only include 'system' collections that are replicated.
+            bool isReplicatedSystemColl =
+                (replicatedSystemCollections.count(collNss.coll().toString()) > 0);
+            if (collNss.isSystem() && !isReplicatedSystemColl)
+                return true;
 
-                if (desiredCollections.size() > 0 &&
-                    desiredCollections.count(collNss.coll().toString()) == 0)
-                    return true;
+            if (collNss.coll().startsWith("tmp.mr.")) {
+                // We skip any incremental map reduce collections as they also aren't
+                // replicated.
+                return true;
+            }
 
-                // Don't include 'drop pending' collections.
-                if (collNss.isDropPendingNamespace())
-                    return true;
+            if (desiredCollections.size() > 0 &&
+                desiredCollections.count(collNss.coll().toString()) == 0)
+                return true;
 
-                if (collection->isCapped()) {
-                    cappedCollectionSet.insert(collNss.coll().toString());
-                }
+            // Don't include 'drop pending' collections.
+            if (collNss.isDropPendingNamespace())
+                return true;
 
-                if (OptionalCollectionUUID uuid = collection->uuid()) {
-                    collectionToUUIDMap[collNss.coll().toString()] = uuid;
-                }
+            if (collection->isCapped()) {
+                cappedCollectionSet.insert(collNss.coll().toString());
+            }
 
-                // Compute the hash for this collection.
-                std::string hash = _hashCollection(opCtx, db, collNss);
+            if (OptionalCollectionUUID uuid = collection->uuid()) {
+                collectionToUUIDMap[collNss.coll().toString()] = uuid;
+            }
 
-                collectionToHashMap[collNss.coll().toString()] = hash;
+            // Compute the hash for this collection.
+            std::string hash = _hashCollection(opCtx, db, collNss);
 
-                return true;
-            });
+            collectionToHashMap[collNss.coll().toString()] = hash;
+
+            return true;
+        });
         if (!noError)
             return false;
 
diff --git a/src/mongo/db/commands/drop_indexes.cpp b/src/mongo/db/commands/drop_indexes.cpp
index 4e93324f7d4..1b4b9491d7f 100644
--- a/src/mongo/db/commands/drop_indexes.cpp
+++ b/src/mongo/db/commands/drop_indexes.cpp
@@ -36,7 +36,6 @@
 
 #include "mongo/db/background.h"
 #include "mongo/db/catalog/collection.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/catalog/drop_indexes.h"
 #include "mongo/db/catalog/index_catalog.h"
diff --git a/src/mongo/db/commands/list_collections.cpp b/src/mongo/db/commands/list_collections.cpp
index cb0a65f607f..fc05460ac46 100644
--- a/src/mongo/db/commands/list_collections.cpp
+++ b/src/mongo/db/commands/list_collections.cpp
@@ -39,7 +39,6 @@
 #include "mongo/bson/util/bson_extract.h"
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/catalog/collection.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/collection_catalog_helper.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/catalog/index_catalog.h"
@@ -321,11 +320,7 @@ public:
                     }
                 } else {
                     mongo::catalog::forEachCollectionFromDb(
-                        opCtx,
-                        dbname,
-                        MODE_IS,
-                        [&](const Collection* collection,
-                            const CollectionCatalogEntry* catalogEntry) {
+                        opCtx, dbname, MODE_IS, [&](const Collection* collection) {
                             if (authorizedCollections &&
                                 (!as->isAuthorizedForAnyActionOnResource(
                                     ResourcePattern::forExactNamespace(collection->ns())))) {
diff --git a/src/mongo/db/commands/list_indexes.cpp b/src/mongo/db/commands/list_indexes.cpp
index bc9d164b65f..aa64d182e01 100644
--- a/src/mongo/db/commands/list_indexes.cpp
+++ b/src/mongo/db/commands/list_indexes.cpp
@@ -33,7 +33,6 @@
 
 #include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/db/auth/authorization_session.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/clientcursor.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/concurrency/write_conflict_exception.h"
diff --git a/src/mongo/db/commands/mr.cpp b/src/mongo/db/commands/mr.cpp
index e4bc431be5e..f7e8cebcf41 100644
--- a/src/mongo/db/commands/mr.cpp
+++ b/src/mongo/db/commands/mr.cpp
@@ -39,7 +39,6 @@
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/bson/dotted_path_support.h"
 #include "mongo/db/catalog/collection.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/document_validation.h"
 #include "mongo/db/catalog/index_catalog.h"
diff --git a/src/mongo/db/commands/rename_collection_cmd.cpp b/src/mongo/db/commands/rename_collection_cmd.cpp
index 7107a370ce0..e047376f281 100644
--- a/src/mongo/db/commands/rename_collection_cmd.cpp
+++ b/src/mongo/db/commands/rename_collection_cmd.cpp
@@ -31,7 +31,6 @@
 
 #include "mongo/client/dbclient_cursor.h"
 #include "mongo/db/catalog/collection.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/catalog/rename_collection.h"
diff --git a/src/mongo/db/commands/resize_oplog.cpp b/src/mongo/db/commands/resize_oplog.cpp
index 45331d2609e..7c2a954d1af 100644
--- a/src/mongo/db/commands/resize_oplog.cpp
+++ b/src/mongo/db/commands/resize_oplog.cpp
@@ -35,7 +35,6 @@
 
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/catalog/collection.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/db_raii.h"
diff --git a/src/mongo/db/commands/validate.cpp b/src/mongo/db/commands/validate.cpp
index f45e87b8148..5ac507f630d 100644
--- a/src/mongo/db/commands/validate.cpp
+++ b/src/mongo/db/commands/validate.cpp
@@ -32,7 +32,7 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/catalog/collection.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
+
 #include "mongo/db/client.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/db_raii.h"
diff --git a/src/mongo/db/index_builds_coordinator.cpp b/src/mongo/db/index_builds_coordinator.cpp
index b25bba1f5fb..292f119de7c 100644
--- a/src/mongo/db/index_builds_coordinator.cpp
+++ b/src/mongo/db/index_builds_coordinator.cpp
@@ -153,7 +153,7 @@ IndexBuildsCoordinator::~IndexBuildsCoordinator() {
 
 StatusWith<std::pair<long long, long long>> IndexBuildsCoordinator::startIndexRebuildForRecovery(
     OperationContext* opCtx,
-    CollectionCatalogEntry* cce,
+    const NamespaceString& nss,
     const std::vector<BSONObj>& specs,
     const UUID& buildUUID) {
     // Index builds in recovery mode have the global write lock.
@@ -171,8 +171,6 @@ StatusWith<std::pair<long long, long long>> IndexBuildsCoordinator::startIndexRe
         indexNames.push_back(name);
     }
 
-    const NamespaceString nss(cce->ns());
-
     ReplIndexBuildState::IndexCatalogStats indexCatalogStats;
 
     auto& collectionCatalog = CollectionCatalog::get(getGlobalServiceContext());
diff --git a/src/mongo/db/index_builds_coordinator.h b/src/mongo/db/index_builds_coordinator.h
index ff4933cdbbc..39e6b690cda 100644
--- a/src/mongo/db/index_builds_coordinator.h
+++ b/src/mongo/db/index_builds_coordinator.h
@@ -42,6 +42,7 @@
 #include "mongo/db/database_index_builds_tracker.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/repl_index_build_state.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/stdx/condition_variable.h"
 #include "mongo/stdx/mutex.h"
 #include "mongo/util/concurrency/with_lock.h"
@@ -124,7 +125,7 @@ public:
      */
     StatusWith<std::pair<long long, long long>> startIndexRebuildForRecovery(
         OperationContext* opCtx,
-        CollectionCatalogEntry* cce,
+        const NamespaceString& nss,
         const std::vector<BSONObj>& specs,
         const UUID& buildUUID);
 
diff --git a/src/mongo/db/op_observer_impl.cpp b/src/mongo/db/op_observer_impl.cpp
index a8f67c8af4b..b541cd20946 100644
--- a/src/mongo/db/op_observer_impl.cpp
+++ b/src/mongo/db/op_observer_impl.cpp
@@ -34,7 +34,6 @@
 #include <limits>
 
 #include "mongo/bson/bsonobjbuilder.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/catalog/database_holder.h"
diff --git a/src/mongo/db/pipeline/document_source_change_stream_test.cpp b/src/mongo/db/pipeline/document_source_change_stream_test.cpp
index f2f728e9128..f50808b97e9 100644
--- a/src/mongo/db/pipeline/document_source_change_stream_test.cpp
+++ b/src/mongo/db/pipeline/document_source_change_stream_test.cpp
@@ -36,7 +36,6 @@
 #include "mongo/bson/bsonobj.h"
 #include "mongo/bson/json.h"
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_catalog_entry_mock.h"
 #include "mongo/db/catalog/collection_mock.h"
 #include "mongo/db/pipeline/aggregation_context_fixture.h"
 #include "mongo/db/pipeline/document.h"
@@ -416,9 +415,8 @@ TEST_F(ChangeStreamStageTest, ShouldRejectBothStartAtOperationTimeAndResumeAfter
 
     // Need to put the collection in the collection catalog so the resume token is valid.
     auto collection = std::make_unique<CollectionMock>(nss);
-    auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
     CollectionCatalog::get(getExpCtx()->opCtx)
-        .registerCollection(testUuid(), std::move(catalogEntry), std::move(collection));
+        .registerCollection(testUuid(), std::move(collection));
 
     ASSERT_THROWS_CODE(
         DSChangeStream::createFromBson(
@@ -439,9 +437,8 @@ TEST_F(ChangeStreamStageTest, ShouldRejectBothStartAfterAndResumeAfterOptions) {
 
     // Need to put the collection in the collection catalog so the resume token is validcollection
     auto collection = std::make_unique<CollectionMock>(nss);
-    auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
     auto& catalog = CollectionCatalog::get(opCtx);
-    catalog.registerCollection(testUuid(), std::move(catalogEntry), std::move(collection));
+    catalog.registerCollection(testUuid(), std::move(collection));
 
     ASSERT_THROWS_CODE(
         DSChangeStream::createFromBson(
@@ -462,9 +459,8 @@ TEST_F(ChangeStreamStageTest, ShouldRejectBothStartAtOperationTimeAndStartAfterO
 
     // Need to put the collection in the collection catalog so the resume token is valid.
     auto collection = std::make_unique<CollectionMock>(nss);
-    auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
     auto& catalog = CollectionCatalog::get(opCtx);
-    catalog.registerCollection(testUuid(), std::move(catalogEntry), std::move(collection));
+    catalog.registerCollection(testUuid(), std::move(collection));
 
     ASSERT_THROWS_CODE(
         DSChangeStream::createFromBson(
@@ -485,9 +481,8 @@ TEST_F(ChangeStreamStageTest, ShouldRejectResumeAfterWithResumeTokenMissingUUID)
 
     // Need to put the collection in the collection catalog so the resume token is valid.
     auto collection = std::make_unique<CollectionMock>(nss);
-    auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
     auto& catalog = CollectionCatalog::get(opCtx);
-    catalog.registerCollection(testUuid(), std::move(catalogEntry), std::move(collection));
+    catalog.registerCollection(testUuid(), std::move(collection));
 
     ASSERT_THROWS_CODE(
         DSChangeStream::createFromBson(
@@ -1514,9 +1509,7 @@ TEST_F(ChangeStreamStageTest, DocumentKeyShouldIncludeShardKeyFromResumeToken) {
     const auto uuid = testUuid();
 
     auto collection = std::make_unique<CollectionMock>(nss);
-    auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
-    CollectionCatalog::get(getExpCtx()->opCtx)
-        .registerCollection(uuid, std::move(catalogEntry), std::move(collection));
+    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, std::move(collection));
 
     BSONObj o2 = BSON("_id" << 1 << "shardKey" << 2);
     auto resumeToken = makeResumeToken(ts, uuid, o2);
@@ -1561,9 +1554,7 @@ TEST_F(ChangeStreamStageTest, DocumentKeyShouldNotIncludeShardKeyFieldsIfNotPres
     const auto uuid = testUuid();
 
     auto collection = std::make_unique<CollectionMock>(nss);
-    auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
-    CollectionCatalog::get(getExpCtx()->opCtx)
-        .registerCollection(uuid, std::move(catalogEntry), std::move(collection));
+    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, std::move(collection));
 
     BSONObj o2 = BSON("_id" << 1 << "shardKey" << 2);
     auto resumeToken = makeResumeToken(ts, uuid, o2);
@@ -1605,9 +1596,7 @@ TEST_F(ChangeStreamStageTest, ResumeAfterFailsIfResumeTokenDoesNotContainUUID) {
     const auto uuid = testUuid();
 
     auto collection = std::make_unique<CollectionMock>(nss);
-    auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
-    CollectionCatalog::get(getExpCtx()->opCtx)
-        .registerCollection(uuid, std::move(catalogEntry), std::move(collection));
+    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, std::move(collection));
 
     // Create a resume token from only the timestamp.
     auto resumeToken = makeResumeToken(ts);
@@ -1660,9 +1649,8 @@ TEST_F(ChangeStreamStageTest, ResumeAfterWithTokenFromInvalidateShouldFail) {
 
     // Need to put the collection in the collection catalog so the resume token is valid.
     auto collection = std::make_unique<CollectionMock>(nss);
-    auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
     CollectionCatalog::get(getExpCtx()->opCtx)
-        .registerCollection(testUuid(), std::move(catalogEntry), std::move(collection));
+        .registerCollection(testUuid(), std::move(collection));
 
     const auto resumeTokenInvalidate =
         makeResumeToken(kDefaultTs,
@@ -2046,9 +2034,7 @@ TEST_F(ChangeStreamStageDBTest, DocumentKeyShouldIncludeShardKeyFromResumeToken)
     const auto uuid = testUuid();
 
     auto collection = std::make_unique<CollectionMock>(nss);
-    auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
-    CollectionCatalog::get(getExpCtx()->opCtx)
-        .registerCollection(uuid, std::move(catalogEntry), std::move(collection));
+    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, std::move(collection));
 
     BSONObj o2 = BSON("_id" << 1 << "shardKey" << 2);
     auto resumeToken = makeResumeToken(ts, uuid, o2);
@@ -2084,9 +2070,7 @@ TEST_F(ChangeStreamStageDBTest, DocumentKeyShouldNotIncludeShardKeyFieldsIfNotPr
     const auto uuid = testUuid();
 
     auto collection = std::make_unique<CollectionMock>(nss);
-    auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
-    CollectionCatalog::get(getExpCtx()->opCtx)
-        .registerCollection(uuid, std::move(catalogEntry), std::move(collection));
+    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, std::move(collection));
 
     BSONObj o2 = BSON("_id" << 1 << "shardKey" << 2);
     auto resumeToken = makeResumeToken(ts, uuid, o2);
@@ -2123,9 +2107,7 @@ TEST_F(ChangeStreamStageDBTest, DocumentKeyShouldNotIncludeShardKeyIfResumeToken
     const auto uuid = testUuid();
 
     auto collection = std::make_unique<CollectionMock>(nss);
-    auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
-    CollectionCatalog::get(getExpCtx()->opCtx)
-        .registerCollection(uuid, std::move(catalogEntry), std::move(collection));
+    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, std::move(collection));
 
     // Create a resume token from only the timestamp.
     auto resumeToken = makeResumeToken(ts);
@@ -2161,9 +2143,8 @@ TEST_F(ChangeStreamStageDBTest, ResumeAfterWithTokenFromInvalidateShouldFail) {
 
     // Need to put the collection in the collection catalog so the resume token is valid.
     auto collection = std::make_unique<CollectionMock>(nss);
-    auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
     CollectionCatalog::get(getExpCtx()->opCtx)
-        .registerCollection(testUuid(), std::move(catalogEntry), std::move(collection));
+        .registerCollection(testUuid(), std::move(collection));
 
     const auto resumeTokenInvalidate =
         makeResumeToken(kDefaultTs,
@@ -2184,9 +2165,7 @@ TEST_F(ChangeStreamStageDBTest, ResumeAfterWithTokenFromDropDatabase) {
     const auto uuid = testUuid();
 
     auto collection = std::make_unique<CollectionMock>(nss);
-    auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
-    CollectionCatalog::get(getExpCtx()->opCtx)
-        .registerCollection(uuid, std::move(catalogEntry), std::move(collection));
+    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, std::move(collection));
 
     // Create a resume token from only the timestamp, similar to a 'dropDatabase' entry.
     auto resumeToken = makeResumeToken(
@@ -2215,9 +2194,7 @@ TEST_F(ChangeStreamStageDBTest, StartAfterSucceedsEvenIfResumeTokenDoesNotContai
     const auto uuid = testUuid();
 
     auto collection = std::make_unique<CollectionMock>(nss);
-    auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
-    CollectionCatalog::get(getExpCtx()->opCtx)
-        .registerCollection(uuid, std::move(catalogEntry), std::move(collection));
+    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, std::move(collection));
 
     // Create a resume token from only the timestamp, similar to a 'dropDatabase' entry.
     auto resumeToken = makeResumeToken(kDefaultTs);
diff --git a/src/mongo/db/query/query_request_test.cpp b/src/mongo/db/query/query_request_test.cpp
index 1ad7544a371..ed4d369602b 100644
--- a/src/mongo/db/query/query_request_test.cpp
+++ b/src/mongo/db/query/query_request_test.cpp
@@ -34,7 +34,6 @@
 #include <boost/optional/optional_io.hpp>
 
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_catalog_entry_mock.h"
 #include "mongo/db/catalog/collection_mock.h"
 #include "mongo/db/dbmessage.h"
 #include "mongo/db/json.h"
@@ -1478,9 +1477,8 @@ TEST_F(QueryRequestTest, ParseFromUUID) {
     const CollectionUUID uuid = UUID::gen();
     const NamespaceString nss("test.testns");
     auto coll = std::make_unique<CollectionMock>(nss);
-    auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
     CollectionCatalog& catalog = CollectionCatalog::get(opCtx.get());
-    catalog.registerCollection(uuid, std::move(catalogEntry), std::move(coll));
+    catalog.registerCollection(uuid, std::move(coll));
     QueryRequest qr(NamespaceStringOrUUID("test", uuid));
     // Ensure a call to refreshNSS succeeds.
     qr.refreshNSS(opCtx.get());
diff --git a/src/mongo/db/repair_database.cpp b/src/mongo/db/repair_database.cpp
index 95cdae1bf1a..3ac2f9c6a06 100644
--- a/src/mongo/db/repair_database.cpp
+++ b/src/mongo/db/repair_database.cpp
@@ -42,7 +42,6 @@
 #include "mongo/db/background.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/document_validation.h"
@@ -52,6 +51,7 @@
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/index_builds_coordinator.h"
 #include "mongo/db/logical_clock.h"
+#include "mongo/db/namespace_string.h"
 #include "mongo/db/query/query_knobs_gen.h"
 #include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/storage_engine.h"
@@ -61,7 +61,7 @@
 namespace mongo {
 
 StatusWith<IndexNameObjs> getIndexNameObjs(OperationContext* opCtx,
-                                           CollectionCatalogEntry* cce,
+                                           const NamespaceString& nss,
                                            std::function<bool(const std::string&)> filter) {
     IndexNameObjs ret;
     std::vector<std::string>& indexNames = ret.first;
@@ -69,7 +69,7 @@ StatusWith<IndexNameObjs> getIndexNameObjs(OperationContext* opCtx,
     auto durableCatalog = DurableCatalog::get(opCtx);
     {
         // Fetch all indexes
-        durableCatalog->getAllIndexes(opCtx, cce->ns(), &indexNames);
+        durableCatalog->getAllIndexes(opCtx, nss, &indexNames);
         auto newEnd =
             std::remove_if(indexNames.begin(),
                            indexNames.end(),
@@ -80,7 +80,7 @@ StatusWith<IndexNameObjs> getIndexNameObjs(OperationContext* opCtx,
 
 
         for (const auto& name : indexNames) {
-            BSONObj spec = durableCatalog->getIndexSpec(opCtx, cce->ns(), name);
+            BSONObj spec = durableCatalog->getIndexSpec(opCtx, nss, name);
             using IndexVersion = IndexDescriptor::IndexVersion;
             IndexVersion indexVersion = IndexVersion::kV1;
             if (auto indexVersionElem = spec[IndexDescriptor::kIndexVersionFieldName]) {
@@ -111,7 +111,7 @@ StatusWith<IndexNameObjs> getIndexNameObjs(OperationContext* opCtx,
 }
 
 Status rebuildIndexesOnCollection(OperationContext* opCtx,
-                                  CollectionCatalogEntry* cce,
+                                  Collection* collection,
                                   const std::vector<BSONObj>& indexSpecs) {
     // Skip the rest if there are no indexes to rebuild.
     if (indexSpecs.empty())
@@ -120,15 +120,15 @@ Status rebuildIndexesOnCollection(OperationContext* opCtx,
     // Rebuild the indexes provided by 'indexSpecs'.
     IndexBuildsCoordinator* indexBuildsCoord = IndexBuildsCoordinator::get(opCtx);
     UUID buildUUID = UUID::gen();
-    auto swRebuild =
-        indexBuildsCoord->startIndexRebuildForRecovery(opCtx, cce, indexSpecs, buildUUID);
+    auto swRebuild = indexBuildsCoord->startIndexRebuildForRecovery(
+        opCtx, collection->ns(), indexSpecs, buildUUID);
     if (!swRebuild.isOK()) {
         return swRebuild.getStatus();
     }
 
     auto[numRecords, dataSize] = swRebuild.getValue();
 
-    auto rs = cce->getRecordStore();
+    auto rs = collection->getRecordStore();
 
     // Update the record store stats after finishing and committing the index builds.
     WriteUnitOfWork wuow(opCtx);
@@ -157,15 +157,13 @@ Status repairCollections(OperationContext* opCtx,
 
     for (const auto& nss : colls) {
         opCtx->checkForInterrupt();
-
-        CollectionCatalogEntry* cce =
-            CollectionCatalog::get(opCtx).lookupCollectionCatalogEntryByNamespace(nss);
-        auto swIndexNameObjs = getIndexNameObjs(opCtx, cce);
+        auto collection = CollectionCatalog::get(opCtx).lookupCollectionByNamespace(nss);
+        auto swIndexNameObjs = getIndexNameObjs(opCtx, nss);
         if (!swIndexNameObjs.isOK())
             return swIndexNameObjs.getStatus();
 
         std::vector<BSONObj> indexSpecs = swIndexNameObjs.getValue().second;
-        Status status = rebuildIndexesOnCollection(opCtx, cce, indexSpecs);
+        Status status = rebuildIndexesOnCollection(opCtx, collection, indexSpecs);
         if (!status.isOK())
             return status;
 
diff --git a/src/mongo/db/repair_database.h b/src/mongo/db/repair_database.h
index b8543e0c973..50fed16a86c 100644
--- a/src/mongo/db/repair_database.h
+++ b/src/mongo/db/repair_database.h
@@ -35,10 +35,11 @@
 #include "mongo/bson/bsonobj.h"
 
 namespace mongo {
-class CollectionCatalogEntry;
+class Collection;
+class StorageEngine;
+class NamespaceString;
 class OperationContext;
 class Status;
-class StorageEngine;
 class StringData;
 
 typedef std::pair<std::vector<std::string>, std::vector<BSONObj>> IndexNameObjs;
@@ -51,7 +52,7 @@ typedef std::pair<std::vector<std::string>, std::vector<BSONObj>> IndexNameObjs;
  *               should be included in the result.
  */
 StatusWith<IndexNameObjs> getIndexNameObjs(OperationContext* opCtx,
-                                           CollectionCatalogEntry* cce,
+                                           const NamespaceString& nss,
                                            std::function<bool(const std::string&)> filter =
                                                [](const std::string& indexName) { return true; });
 
@@ -60,7 +61,7 @@ StatusWith<IndexNameObjs> getIndexNameObjs(OperationContext* opCtx,
  * One example usage is when a 'dropIndex' command is rolled back. The dropped index must be remade.
  */
 Status rebuildIndexesOnCollection(OperationContext* opCtx,
-                                  CollectionCatalogEntry* cce,
+                                  Collection* collection,
                                   const std::vector<BSONObj>& indexSpecs);
 
 /**
diff --git a/src/mongo/db/repair_database_and_check_version.cpp b/src/mongo/db/repair_database_and_check_version.cpp
index 3e3e395c0ba..f9f2bdb64bb 100644
--- a/src/mongo/db/repair_database_and_check_version.cpp
+++ b/src/mongo/db/repair_database_and_check_version.cpp
@@ -36,7 +36,6 @@
 #include <functional>
 
 #include "mongo/db/catalog/collection.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/create_collection.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/catalog/database_holder.h"
@@ -139,11 +138,11 @@ Status restoreMissingFeatureCompatibilityVersionDocument(OperationContext* opCtx
  * Returns true if the collection associated with the given CollectionCatalogEntry has an index on
  * the _id field
  */
-bool checkIdIndexExists(OperationContext* opCtx, const CollectionCatalogEntry* catalogEntry) {
+bool checkIdIndexExists(OperationContext* opCtx, const NamespaceString& nss) {
     auto durableCatalog = DurableCatalog::get(opCtx);
-    auto indexCount = durableCatalog->getTotalIndexCount(opCtx, catalogEntry->ns());
+    auto indexCount = durableCatalog->getTotalIndexCount(opCtx, nss);
     auto indexNames = std::vector<std::string>(indexCount);
-    durableCatalog->getAllIndexes(opCtx, catalogEntry->ns(), &indexNames);
+    durableCatalog->getAllIndexes(opCtx, nss, &indexNames);
 
     for (auto name : indexNames) {
         if (name == "_id_") {
@@ -217,14 +216,13 @@ 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 = 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
             // that was created manually by the user. Check the list of indexes to confirm index
             // does not exist before attempting to build it or returning an error.
-            if (requiresIndex && !hasAutoIndexIdField && !checkIdIndexExists(opCtx, catalogEntry)) {
+            if (requiresIndex && !hasAutoIndexIdField && !checkIdIndexExists(opCtx, coll->ns())) {
                 log() << "collection " << coll->ns() << " is missing an _id index; building it now";
                 auto status = buildMissingIdIndex(opCtx, coll);
                 if (!status.isOK()) {
@@ -285,15 +283,8 @@ void rebuildIndexes(OperationContext* opCtx, StorageEngine* storageEngine) {
     for (auto&& indexNamespace : indexesToRebuild) {
         NamespaceString collNss(indexNamespace.first);
         const std::string& indexName = indexNamespace.second;
-
-        CollectionCatalogEntry* cce =
-            CollectionCatalog::get(opCtx).lookupCollectionCatalogEntryByNamespace(collNss);
-        invariant(cce,
-                  str::stream() << "couldn't get collection catalog entry for collection "
-                                << collNss.toString());
-
         auto swIndexSpecs = getIndexNameObjs(
-            opCtx, cce, [&indexName](const std::string& name) { return name == indexName; });
+            opCtx, collNss, [&indexName](const std::string& name) { return name == indexName; });
         if (!swIndexSpecs.isOK() || swIndexSpecs.getValue().first.empty()) {
             fassert(40590,
                     {ErrorCodes::InternalError,
@@ -315,14 +306,13 @@ void rebuildIndexes(OperationContext* opCtx, StorageEngine* storageEngine) {
     for (const auto& entry : nsToIndexNameObjMap) {
         NamespaceString collNss(entry.first);
 
-        auto collCatalogEntry =
-            CollectionCatalog::get(opCtx).lookupCollectionCatalogEntryByNamespace(collNss);
+        auto collection = CollectionCatalog::get(opCtx).lookupCollectionByNamespace(collNss);
         for (const auto& indexName : entry.second.first) {
             log() << "Rebuilding index. Collection: " << collNss << " Index: " << indexName;
         }
 
         std::vector<BSONObj> indexSpecs = entry.second.second;
-        fassert(40592, rebuildIndexesOnCollection(opCtx, collCatalogEntry, indexSpecs));
+        fassert(40592, rebuildIndexesOnCollection(opCtx, collection, indexSpecs));
     }
 }
 
diff --git a/src/mongo/db/repl/dbcheck.cpp b/src/mongo/db/repl/dbcheck.cpp
index 31071a1af71..8a3827db897 100644
--- a/src/mongo/db/repl/dbcheck.cpp
+++ b/src/mongo/db/repl/dbcheck.cpp
@@ -31,7 +31,6 @@
 
 #include "mongo/bson/simple_bsonelement_comparator.h"
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/health_log.h"
diff --git a/src/mongo/db/repl/idempotency_test_fixture.cpp b/src/mongo/db/repl/idempotency_test_fixture.cpp
index cf367c53a30..06b4394509f 100644
--- a/src/mongo/db/repl/idempotency_test_fixture.cpp
+++ b/src/mongo/db/repl/idempotency_test_fixture.cpp
@@ -37,7 +37,6 @@
 
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/index_catalog.h"
diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp
index 327f6fbf5ef..19176f7478b 100644
--- a/src/mongo/db/repl/oplog.cpp
+++ b/src/mongo/db/repl/oplog.cpp
@@ -48,7 +48,6 @@
 #include "mongo/db/catalog/coll_mod.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/create_collection.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/drop_collection.h"
diff --git a/src/mongo/db/repl/rollback_test_fixture.cpp b/src/mongo/db/repl/rollback_test_fixture.cpp
index 70d500b440d..a157dcf76df 100644
--- a/src/mongo/db/repl/rollback_test_fixture.cpp
+++ b/src/mongo/db/repl/rollback_test_fixture.cpp
@@ -34,7 +34,6 @@
 #include <memory>
 #include <string>
 
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/client.h"
 #include "mongo/db/db_raii.h"
diff --git a/src/mongo/db/repl/rs_rollback.cpp b/src/mongo/db/repl/rs_rollback.cpp
index 6f7bd28bedf..a1b76201cdf 100644
--- a/src/mongo/db/repl/rs_rollback.cpp
+++ b/src/mongo/db/repl/rs_rollback.cpp
@@ -40,7 +40,6 @@
 #include "mongo/bson/util/bson_extract.h"
 #include "mongo/db/auth/authorization_manager.h"
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/document_validation.h"
 #include "mongo/db/catalog/index_catalog.h"
diff --git a/src/mongo/db/repl/rs_rollback_test.cpp b/src/mongo/db/repl/rs_rollback_test.cpp
index 26262101061..84a3a8d803b 100644
--- a/src/mongo/db/repl/rs_rollback_test.cpp
+++ b/src/mongo/db/repl/rs_rollback_test.cpp
@@ -36,7 +36,6 @@
 #include <utility>
 
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/drop_indexes.h"
 #include "mongo/db/catalog/index_catalog.h"
diff --git a/src/mongo/db/repl/storage_interface_impl.cpp b/src/mongo/db/repl/storage_interface_impl.cpp
index 6eca7179b8b..9e4b67fb2e3 100644
--- a/src/mongo/db/repl/storage_interface_impl.cpp
+++ b/src/mongo/db/repl/storage_interface_impl.cpp
@@ -47,7 +47,6 @@
 #include "mongo/db/catalog/coll_mod.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/document_validation.h"
 #include "mongo/db/catalog/index_catalog.h"
diff --git a/src/mongo/db/repl/storage_interface_impl_test.cpp b/src/mongo/db/repl/storage_interface_impl_test.cpp
index 56b2b97c2ea..912599d9e9c 100644
--- a/src/mongo/db/repl/storage_interface_impl_test.cpp
+++ b/src/mongo/db/repl/storage_interface_impl_test.cpp
@@ -34,7 +34,6 @@
 #include <memory>
 
 #include "mongo/bson/bsonmisc.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/catalog/document_validation.h"
diff --git a/src/mongo/db/repl/sync_tail_test.cpp b/src/mongo/db/repl/sync_tail_test.cpp
index 9ec52ed34e3..35b07187677 100644
--- a/src/mongo/db/repl/sync_tail_test.cpp
+++ b/src/mongo/db/repl/sync_tail_test.cpp
@@ -36,7 +36,6 @@
 
 #include "mongo/bson/util/bson_extract.h"
 #include "mongo/db/catalog/collection.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/catalog/database_holder.h"
diff --git a/src/mongo/db/repl_index_build_state.h b/src/mongo/db/repl_index_build_state.h
index 9b44ee9e75a..cd50f2c0289 100644
--- a/src/mongo/db/repl_index_build_state.h
+++ b/src/mongo/db/repl_index_build_state.h
@@ -35,7 +35,6 @@
 #include <vector>
 
 #include "mongo/bson/bsonobj.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/commit_quorum_options.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/namespace_string.h"
@@ -46,6 +45,22 @@
 
 namespace mongo {
 
+// Indicates which protocol an index build is using.
+enum class IndexBuildProtocol {
+    /**
+     * Refers to the pre-FCV 4.2 index build protocol for building indexes in replica sets.
+     * Index builds must complete on the primary before replicating, and are not resumable in
+     * any scenario.
+     */
+    kSinglePhase,
+    /**
+     * Refers to the FCV 4.2 two-phase index build protocol for building indexes in replica
+     * sets. Indexes are built simultaneously on all nodes and are resumable during the draining
+     * phase.
+     */
+    kTwoPhase
+};
+
 /**
  * Tracks the cross replica set progress of a particular index build identified by a build UUID.
  *
diff --git a/src/mongo/db/s/migration_destination_manager.cpp b/src/mongo/db/s/migration_destination_manager.cpp
index b5c1a8f6cb5..01d36506bcf 100644
--- a/src/mongo/db/s/migration_destination_manager.cpp
+++ b/src/mongo/db/s/migration_destination_manager.cpp
@@ -37,7 +37,6 @@
 #include <vector>
 
 #include "mongo/db/auth/authorization_session.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/document_validation.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/dbhelpers.h"
diff --git a/src/mongo/db/storage/SConscript b/src/mongo/db/storage/SConscript
index edcb3989753..4f4b8f2e43e 100644
--- a/src/mongo/db/storage/SConscript
+++ b/src/mongo/db/storage/SConscript
@@ -394,7 +394,6 @@ env.Library(
     target='durable_catalog_impl',
     source=[
         'durable_catalog_impl.cpp',
-        'kv/kv_collection_catalog_entry.cpp', # TODO: SERVER-40717 Avoid source under kv sub-directory
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/bson/util/bson_extract',
diff --git a/src/mongo/db/storage/bson_collection_catalog_entry.cpp b/src/mongo/db/storage/bson_collection_catalog_entry.cpp
index d691779de5e..b6485febf42 100644
--- a/src/mongo/db/storage/bson_collection_catalog_entry.cpp
+++ b/src/mongo/db/storage/bson_collection_catalog_entry.cpp
@@ -103,10 +103,6 @@ void parseMultikeyPathsFromBytes(BSONObj multikeyPathsObj, MultikeyPaths* multik
 const StringData BSONCollectionCatalogEntry::kIndexBuildScanning = "scanning"_sd;
 const StringData BSONCollectionCatalogEntry::kIndexBuildDraining = "draining"_sd;
 
-BSONCollectionCatalogEntry::BSONCollectionCatalogEntry(StringData ns)
-    : CollectionCatalogEntry(ns) {}
-
-
 // --------------------------
 
 void BSONCollectionCatalogEntry::IndexMetaData::updateTTLSetting(long long newExpireSeconds) {
diff --git a/src/mongo/db/storage/bson_collection_catalog_entry.h b/src/mongo/db/storage/bson_collection_catalog_entry.h
index d33b895d237..15405b1942d 100644
--- a/src/mongo/db/storage/bson_collection_catalog_entry.h
+++ b/src/mongo/db/storage/bson_collection_catalog_entry.h
@@ -32,7 +32,7 @@
 #include <string>
 #include <vector>
 
-#include "mongo/db/catalog/collection_catalog_entry.h"
+#include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/index/multikey_paths.h"
 #include "mongo/db/storage/kv/kv_prefix.h"
 
@@ -42,12 +42,18 @@ namespace mongo {
  * This is a helper class for any storage engine that wants to store catalog information
  * as BSON. It is totally optional to use this.
  */
-class BSONCollectionCatalogEntry : public CollectionCatalogEntry {
+class BSONCollectionCatalogEntry {
 public:
     static const StringData kIndexBuildScanning;
     static const StringData kIndexBuildDraining;
 
-    BSONCollectionCatalogEntry(StringData ns);
+    /**
+     * Incremented when breaking changes are made to the index build procedure so that other servers
+     * know whether or not to resume or discard unfinished index builds.
+     */
+    static constexpr int kIndexBuildVersion = 1;
+
+    BSONCollectionCatalogEntry() = default;
 
     virtual ~BSONCollectionCatalogEntry() {}
 
diff --git a/src/mongo/db/storage/durable_catalog.h b/src/mongo/db/storage/durable_catalog.h
index a7c85f38e3b..65970cde75a 100644
--- a/src/mongo/db/storage/durable_catalog.h
+++ b/src/mongo/db/storage/durable_catalog.h
@@ -30,10 +30,11 @@
 #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/index/index_descriptor.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
+#include "mongo/db/repl_index_build_state.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"
@@ -102,10 +103,7 @@ public:
      */
     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(
+    virtual StatusWith<std::unique_ptr<RecordStore>> createCollection(
         OperationContext* opCtx,
         const NamespaceString& nss,
         const CollectionOptions& options,
diff --git a/src/mongo/db/storage/durable_catalog_impl.cpp b/src/mongo/db/storage/durable_catalog_impl.cpp
index 148b1b195b3..4d509eb8ff5 100644
--- a/src/mongo/db/storage/durable_catalog_impl.cpp
+++ b/src/mongo/db/storage/durable_catalog_impl.cpp
@@ -42,7 +42,6 @@
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/storage/durable_catalog_feature_tracker.h"
-#include "mongo/db/storage/kv/kv_collection_catalog_entry.h"
 #include "mongo/db/storage/kv/kv_engine.h"
 #include "mongo/db/storage/record_store.h"
 #include "mongo/db/storage/recovery_unit.h"
@@ -770,31 +769,7 @@ StatusWith<std::string> DurableCatalogImpl::newOrphanedIdent(OperationContext* o
     return StatusWith<std::string>(std::move(ns));
 }
 
-std::unique_ptr<CollectionCatalogEntry> DurableCatalogImpl::makeCollectionCatalogEntry(
-    OperationContext* opCtx, const NamespaceString& nss, bool forRepair) {
-    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, nss);
-    uassert(ErrorCodes::MustDowngrade,
-            str::stream() << "Collection does not have UUID in DurableCatalogImpl. Collection: "
-                          << nss,
-            md.options.uuid);
-
-    auto ident = getCollectionIdent(nss);
-
-    std::unique_ptr<RecordStore> rs;
-    if (forRepair) {
-        // Using a NULL rs since we don't want to open this record store before it has been
-        // repaired. This also ensures that if we try to use it, it will blow up.
-        rs = nullptr;
-    } else {
-        rs = _engine->getEngine()->getGroupedRecordStore(
-            opCtx, nss.ns(), ident, md.options, md.prefix);
-        invariant(rs);
-    }
-
-    return std::make_unique<KVCollectionCatalogEntry>(_engine, nss.ns(), ident, std::move(rs));
-}
-
-StatusWith<std::unique_ptr<CollectionCatalogEntry>> DurableCatalogImpl::createCollection(
+StatusWith<std::unique_ptr<RecordStore>> DurableCatalogImpl::createCollection(
     OperationContext* opCtx,
     const NamespaceString& nss,
     const CollectionOptions& options,
@@ -802,14 +777,13 @@ StatusWith<std::unique_ptr<CollectionCatalogEntry>> DurableCatalogImpl::createCo
     invariant(opCtx->lockState()->isDbLockedForMode(nss.db(), MODE_IX));
     invariant(nss.coll().size() > 0);
 
-    if (CollectionCatalog::get(opCtx).lookupCollectionCatalogEntryByNamespace(nss)) {
+    if (CollectionCatalog::get(opCtx).lookupCollectionByNamespace(nss)) {
         return Status(ErrorCodes::NamespaceExists,
                       str::stream() << "collection already exists " << nss);
     }
 
     KVPrefix prefix = KVPrefix::getNextPrefix(nss);
 
-    // need to create it
     Status status = _addEntry(opCtx, nss, options, prefix);
     if (!status.isOK())
         return status;
@@ -838,7 +812,7 @@ StatusWith<std::unique_ptr<CollectionCatalogEntry>> DurableCatalogImpl::createCo
     auto rs = _engine->getEngine()->getGroupedRecordStore(opCtx, nss.ns(), ident, options, prefix);
     invariant(rs);
 
-    return std::make_unique<KVCollectionCatalogEntry>(_engine, nss.ns(), ident, std::move(rs));
+    return std::move(rs);
 }
 
 Status DurableCatalogImpl::renameCollection(OperationContext* opCtx,
@@ -868,9 +842,7 @@ Status DurableCatalogImpl::renameCollection(OperationContext* opCtx,
 Status DurableCatalogImpl::dropCollection(OperationContext* opCtx, const NamespaceString& nss) {
     invariant(opCtx->lockState()->isCollectionLockedForMode(nss, MODE_X));
 
-    CollectionCatalogEntry* const entry =
-        CollectionCatalog::get(opCtx).lookupCollectionCatalogEntryByNamespace(nss);
-    if (!entry) {
+    if (!CollectionCatalog::get(opCtx).lookupCollectionByNamespace(nss)) {
         return Status(ErrorCodes::NamespaceNotFound, "cannnot find collection to drop");
     }
 
@@ -897,8 +869,7 @@ Status DurableCatalogImpl::dropCollection(OperationContext* opCtx, const Namespa
         return status;
     }
 
-    // This will lazily delete the KVCollectionCatalogEntry and notify the storageEngine to
-    // drop the collection only on WUOW::commit().
+    // This will notify the storageEngine to drop the collection only on WUOW::commit().
     opCtx->recoveryUnit()->onCommit(
         [ opCtx, catalog = this, nss, uuid, ident ](boost::optional<Timestamp> commitTimestamp) {
             StorageEngineInterface* engine = catalog->_engine;
diff --git a/src/mongo/db/storage/durable_catalog_impl.h b/src/mongo/db/storage/durable_catalog_impl.h
index 364f67bd320..a00680456bc 100644
--- a/src/mongo/db/storage/durable_catalog_impl.h
+++ b/src/mongo/db/storage/durable_catalog_impl.h
@@ -103,15 +103,10 @@ public:
 
     std::string newInternalIdent();
 
-    std::unique_ptr<CollectionCatalogEntry> makeCollectionCatalogEntry(OperationContext* opCtx,
-                                                                       const NamespaceString& nss,
-                                                                       bool forRepair);
-
-    StatusWith<std::unique_ptr<CollectionCatalogEntry>> createCollection(
-        OperationContext* opCtx,
-        const NamespaceString& nss,
-        const CollectionOptions& options,
-        bool allocateDefaultSpace);
+    StatusWith<std::unique_ptr<RecordStore>> createCollection(OperationContext* opCtx,
+                                                              const NamespaceString& nss,
+                                                              const CollectionOptions& options,
+                                                              bool allocateDefaultSpace);
 
     Status renameCollection(OperationContext* opCtx,
                             const NamespaceString& fromNss,
diff --git a/src/mongo/db/storage/kv/SConscript b/src/mongo/db/storage/kv/SConscript
index a3d28216314..79e0c703b51 100644
--- a/src/mongo/db/storage/kv/SConscript
+++ b/src/mongo/db/storage/kv/SConscript
@@ -45,7 +45,7 @@ env.Library(
 env.CppUnitTest(
     target='db_storage_kv_engine_test',
     source=[
-        'kv_collection_catalog_entry_test.cpp',
+        'durable_catalog_test.cpp',
         'kv_drop_pending_ident_reaper_test.cpp',
         'storage_engine_test.cpp',
     ],
diff --git a/src/mongo/db/storage/kv/kv_collection_catalog_entry_test.cpp b/src/mongo/db/storage/kv/durable_catalog_test.cpp
index 6045139591d..c125d8a8d22 100644
--- a/src/mongo/db/storage/kv/kv_collection_catalog_entry_test.cpp
+++ b/src/mongo/db/storage/kv/durable_catalog_test.cpp
@@ -34,7 +34,6 @@
 #include <string>
 
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/collection_mock.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/index/multikey_paths.h"
@@ -57,15 +56,15 @@ static std::string kConstraintViolationsTableIdent("constraintViolations");
 // Update version as breaking changes are introduced into the index build procedure.
 static const long kExpectedVersion = 1;
 
-class KVCollectionCatalogEntryTest : public ServiceContextTest {
+class DurableCatalogTest : public ServiceContextTest {
 public:
-    KVCollectionCatalogEntryTest()
-        : _nss("unittests.kv_collection_catalog_entry"),
+    DurableCatalogTest()
+        : _nss("unittests.durable_catalog"),
           _storageEngine(new DevNullKVEngine(), StorageEngineOptions()) {
         _storageEngine.finishInit();
     }
 
-    ~KVCollectionCatalogEntryTest() {
+    ~DurableCatalogTest() {
         _storageEngine.cleanShutdown();
     }
 
@@ -84,14 +83,12 @@ public:
             const bool allocateDefaultSpace = true;
             CollectionOptions options;
             options.uuid = UUID::gen();
-            auto statusWithCatalogEntry = _storageEngine.getCatalog()->createCollection(
+            auto statusWithRecordStore = _storageEngine.getCatalog()->createCollection(
                 opCtx.get(), _nss, options, allocateDefaultSpace);
-            ASSERT_OK(statusWithCatalogEntry.getStatus());
+            ASSERT_OK(statusWithRecordStore.getStatus());
             auto collection = std::make_unique<CollectionMock>(_nss);
             CollectionCatalog::get(opCtx.get())
-                .registerCollection(options.uuid.get(),
-                                    std::move(statusWithCatalogEntry.getValue()),
-                                    std::move(collection));
+                .registerCollection(options.uuid.get(), std::move(collection));
             wuow.commit();
         }
     }
@@ -160,7 +157,7 @@ private:
     size_t numIndexesCreated = 0;
 };
 
-TEST_F(KVCollectionCatalogEntryTest, MultikeyPathsForBtreeIndexInitializedToVectorOfEmptySets) {
+TEST_F(DurableCatalogTest, MultikeyPathsForBtreeIndexInitializedToVectorOfEmptySets) {
     std::string indexName = createIndex(BSON("a" << 1 << "b" << 1));
     auto opCtx = newOperationContext();
     DurableCatalog* catalog = getCatalog();
@@ -171,7 +168,7 @@ TEST_F(KVCollectionCatalogEntryTest, MultikeyPathsForBtreeIndexInitializedToVect
     }
 }
 
-TEST_F(KVCollectionCatalogEntryTest, CanSetIndividualPathComponentOfBtreeIndexAsMultikey) {
+TEST_F(DurableCatalogTest, CanSetIndividualPathComponentOfBtreeIndexAsMultikey) {
     std::string indexName = createIndex(BSON("a" << 1 << "b" << 1));
     auto opCtx = newOperationContext();
     DurableCatalog* catalog = getCatalog();
@@ -184,7 +181,7 @@ TEST_F(KVCollectionCatalogEntryTest, CanSetIndividualPathComponentOfBtreeIndexAs
     }
 }
 
-TEST_F(KVCollectionCatalogEntryTest, MultikeyPathsAccumulateOnDifferentFields) {
+TEST_F(DurableCatalogTest, MultikeyPathsAccumulateOnDifferentFields) {
     std::string indexName = createIndex(BSON("a" << 1 << "b" << 1));
     auto opCtx = newOperationContext();
     DurableCatalog* catalog = getCatalog();
@@ -205,7 +202,7 @@ TEST_F(KVCollectionCatalogEntryTest, MultikeyPathsAccumulateOnDifferentFields) {
     }
 }
 
-TEST_F(KVCollectionCatalogEntryTest, MultikeyPathsAccumulateOnDifferentComponentsOfTheSameField) {
+TEST_F(DurableCatalogTest, MultikeyPathsAccumulateOnDifferentComponentsOfTheSameField) {
     std::string indexName = createIndex(BSON("a.b" << 1));
     auto opCtx = newOperationContext();
     DurableCatalog* catalog = getCatalog();
@@ -226,7 +223,7 @@ TEST_F(KVCollectionCatalogEntryTest, MultikeyPathsAccumulateOnDifferentComponent
     }
 }
 
-TEST_F(KVCollectionCatalogEntryTest, NoOpWhenSpecifiedPathComponentsAlreadySetAsMultikey) {
+TEST_F(DurableCatalogTest, NoOpWhenSpecifiedPathComponentsAlreadySetAsMultikey) {
     std::string indexName = createIndex(BSON("a" << 1));
     auto opCtx = newOperationContext();
     DurableCatalog* catalog = getCatalog();
@@ -247,7 +244,7 @@ TEST_F(KVCollectionCatalogEntryTest, NoOpWhenSpecifiedPathComponentsAlreadySetAs
     }
 }
 
-TEST_F(KVCollectionCatalogEntryTest, CanSetMultipleFieldsAndComponentsAsMultikey) {
+TEST_F(DurableCatalogTest, CanSetMultipleFieldsAndComponentsAsMultikey) {
     std::string indexName = createIndex(BSON("a.b.c" << 1 << "a.b.d" << 1));
     auto opCtx = newOperationContext();
     DurableCatalog* catalog = getCatalog();
@@ -260,7 +257,7 @@ TEST_F(KVCollectionCatalogEntryTest, CanSetMultipleFieldsAndComponentsAsMultikey
     }
 }
 
-DEATH_TEST_F(KVCollectionCatalogEntryTest,
+DEATH_TEST_F(DurableCatalogTest,
              CannotOmitPathLevelMultikeyInfoWithBtreeIndex,
              "Invariant failure !multikeyPaths.empty()") {
     std::string indexName = createIndex(BSON("a" << 1 << "b" << 1));
@@ -269,7 +266,7 @@ DEATH_TEST_F(KVCollectionCatalogEntryTest,
     catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, MultikeyPaths{});
 }
 
-DEATH_TEST_F(KVCollectionCatalogEntryTest,
+DEATH_TEST_F(DurableCatalogTest,
              AtLeastOnePathComponentMustCauseIndexToBeMultikey,
              "Invariant failure somePathIsMultikey") {
     std::string indexName = createIndex(BSON("a" << 1 << "b" << 1));
@@ -279,7 +276,7 @@ DEATH_TEST_F(KVCollectionCatalogEntryTest,
         opCtx.get(), ns(), indexName, {std::set<size_t>{}, std::set<size_t>{}});
 }
 
-TEST_F(KVCollectionCatalogEntryTest, PathLevelMultikeyTrackingIsSupportedBy2dsphereIndexes) {
+TEST_F(DurableCatalogTest, PathLevelMultikeyTrackingIsSupportedBy2dsphereIndexes) {
     std::string indexType = IndexNames::GEO_2DSPHERE;
     std::string indexName = createIndex(BSON("a" << indexType << "b" << 1), indexType);
     auto opCtx = newOperationContext();
@@ -291,7 +288,7 @@ TEST_F(KVCollectionCatalogEntryTest, PathLevelMultikeyTrackingIsSupportedBy2dsph
     }
 }
 
-TEST_F(KVCollectionCatalogEntryTest, PathLevelMultikeyTrackingIsNotSupportedByAllIndexTypes) {
+TEST_F(DurableCatalogTest, PathLevelMultikeyTrackingIsNotSupportedByAllIndexTypes) {
     std::string indexTypes[] = {
         IndexNames::GEO_2D, IndexNames::GEO_HAYSTACK, IndexNames::TEXT, IndexNames::HASHED};
 
@@ -307,7 +304,7 @@ TEST_F(KVCollectionCatalogEntryTest, PathLevelMultikeyTrackingIsNotSupportedByAl
     }
 }
 
-TEST_F(KVCollectionCatalogEntryTest, CanSetEntireTextIndexAsMultikey) {
+TEST_F(DurableCatalogTest, CanSetEntireTextIndexAsMultikey) {
     std::string indexType = IndexNames::TEXT;
     std::string indexName = createIndex(BSON("a" << indexType << "b" << 1), indexType);
     auto opCtx = newOperationContext();
@@ -321,7 +318,7 @@ TEST_F(KVCollectionCatalogEntryTest, CanSetEntireTextIndexAsMultikey) {
     }
 }
 
-TEST_F(KVCollectionCatalogEntryTest, NoOpWhenEntireIndexAlreadySetAsMultikey) {
+TEST_F(DurableCatalogTest, NoOpWhenEntireIndexAlreadySetAsMultikey) {
     std::string indexType = IndexNames::TEXT;
     std::string indexName = createIndex(BSON("a" << indexType << "b" << 1), indexType);
     auto opCtx = newOperationContext();
@@ -343,7 +340,7 @@ TEST_F(KVCollectionCatalogEntryTest, NoOpWhenEntireIndexAlreadySetAsMultikey) {
     }
 }
 
-TEST_F(KVCollectionCatalogEntryTest, SinglePhaseIndexBuild) {
+TEST_F(DurableCatalogTest, SinglePhaseIndexBuild) {
     std::string indexName = createIndex(BSON("a" << 1));
     auto opCtx = newOperationContext();
     DurableCatalog* catalog = getCatalog();
@@ -367,7 +364,7 @@ TEST_F(KVCollectionCatalogEntryTest, SinglePhaseIndexBuild) {
     ASSERT_FALSE(catalog->getConstraintViolationsIdent(opCtx.get(), ns(), indexName));
 }
 
-TEST_F(KVCollectionCatalogEntryTest, TwoPhaseIndexBuild) {
+TEST_F(DurableCatalogTest, TwoPhaseIndexBuild) {
     std::string indexName =
         createIndex(BSON("a" << 1), IndexNames::BTREE, IndexBuildProtocol::kTwoPhase);
     auto opCtx = newOperationContext();
@@ -415,7 +412,7 @@ TEST_F(KVCollectionCatalogEntryTest, TwoPhaseIndexBuild) {
     ASSERT_FALSE(catalog->getConstraintViolationsIdent(opCtx.get(), ns(), indexName));
 }
 
-DEATH_TEST_F(KVCollectionCatalogEntryTest,
+DEATH_TEST_F(DurableCatalogTest,
              SinglePhaseIllegalScanPhase,
              "Invariant failure md.indexes[offset].runTwoPhaseBuild") {
     std::string indexName = createIndex(BSON("a" << 1));
@@ -426,7 +423,7 @@ DEATH_TEST_F(KVCollectionCatalogEntryTest,
         opCtx.get(), ns(), indexName, kSideWritesTableIdent, kConstraintViolationsTableIdent);
 }
 
-DEATH_TEST_F(KVCollectionCatalogEntryTest,
+DEATH_TEST_F(DurableCatalogTest,
              SinglePhaseIllegalDrainPhase,
              "Invariant failure md.indexes[offset].runTwoPhaseBuild") {
     std::string indexName = createIndex(BSON("a" << 1));
@@ -435,7 +432,7 @@ DEATH_TEST_F(KVCollectionCatalogEntryTest,
     catalog->setIndexBuildDraining(opCtx.get(), ns(), indexName);
 }
 
-DEATH_TEST_F(KVCollectionCatalogEntryTest,
+DEATH_TEST_F(DurableCatalogTest,
              CannotSetIndividualPathComponentsOfTextIndexAsMultikey,
              "Invariant failure multikeyPaths.empty()") {
     std::string indexType = IndexNames::TEXT;
diff --git a/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp b/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp
deleted file mode 100644
index f0fbd5904de..00000000000
--- a/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp
+++ /dev/null
@@ -1,59 +0,0 @@
-/**
- *    Copyright (C) 2018-present MongoDB, Inc.
- *
- *    This program is free software: you can redistribute it and/or modify
- *    it under the terms of the Server Side Public License, version 1,
- *    as published by MongoDB, Inc.
- *
- *    This program is distributed in the hope that it will be useful,
- *    but WITHOUT ANY WARRANTY; without even the implied warranty of
- *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *    Server Side Public License for more details.
- *
- *    You should have received a copy of the Server Side Public License
- *    along with this program. If not, see
- *    <http://www.mongodb.com/licensing/server-side-public-license>.
- *
- *    As a special exception, the copyright holders give permission to link the
- *    code of portions of this program with the OpenSSL library under certain
- *    conditions as described in each individual source file and distribute
- *    linked combinations including the program with the OpenSSL library. You
- *    must comply with the Server Side Public License in all respects for
- *    all of the code used other than as permitted herein. If you modify file(s)
- *    with this exception, you may extend this exception to your version of the
- *    file(s), but you are not obligated to do so. If you do not wish to do so,
- *    delete this exception statement from your version. If you delete this
- *    exception statement from all source files in the program, then also delete
- *    it in the license file.
- */
-
-#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kStorage
-
-#include "mongo/platform/basic.h"
-
-#include "mongo/db/storage/kv/kv_collection_catalog_entry.h"
-
-#include <memory>
-
-#include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/index/index_descriptor.h"
-#include "mongo/db/storage/durable_catalog.h"
-#include "mongo/db/storage/durable_catalog_feature_tracker.h"
-#include "mongo/db/storage/kv/kv_engine.h"
-#include "mongo/db/storage/storage_engine_impl.h"
-#include "mongo/util/log.h"
-
-namespace mongo {
-
-using std::string;
-
-KVCollectionCatalogEntry::KVCollectionCatalogEntry(StorageEngineInterface* engine,
-                                                   StringData ns,
-                                                   StringData ident,
-                                                   std::unique_ptr<RecordStore> rs)
-    : BSONCollectionCatalogEntry(ns),
-      _engine(engine),
-      _ident(ident.toString()),
-      _recordStore(std::move(rs)) {}
-
-}  // namespace mongo
diff --git a/src/mongo/db/storage/kv/kv_collection_catalog_entry.h b/src/mongo/db/storage/kv/kv_collection_catalog_entry.h
deleted file mode 100644
index e9133a607f5..00000000000
--- a/src/mongo/db/storage/kv/kv_collection_catalog_entry.h
+++ /dev/null
@@ -1,64 +0,0 @@
-/**
- *    Copyright (C) 2018-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 <memory>
-
-#include "mongo/db/catalog/collection_catalog_entry.h"
-#include "mongo/db/server_options.h"
-#include "mongo/db/storage/bson_collection_catalog_entry.h"
-#include "mongo/db/storage/record_store.h"
-
-namespace mongo {
-
-class StorageEngineInterface;
-
-class KVCollectionCatalogEntry final : public BSONCollectionCatalogEntry {
-public:
-    KVCollectionCatalogEntry(StorageEngineInterface* engine,
-                             StringData ns,
-                             StringData ident,
-                             std::unique_ptr<RecordStore> rs);
-
-    ~KVCollectionCatalogEntry() final {}
-
-    RecordStore* getRecordStore() {
-        return _recordStore.get();
-    }
-    const RecordStore* getRecordStore() const {
-        return _recordStore.get();
-    }
-
-private:
-    StorageEngineInterface* const _engine;  // not owned
-    std::string _ident;
-    std::unique_ptr<RecordStore> _recordStore;  // owned
-};
-}
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 7928f52e83b..e39bae73f0c 100644
--- a/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
+++ b/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
@@ -29,7 +29,6 @@
 
 #include "mongo/db/storage/kv/kv_engine_test_harness.h"
 
-#include "mongo/db/catalog/collection_catalog_entry_mock.h"
 #include "mongo/db/catalog/collection_impl.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/operation_context_noop.h"
@@ -146,7 +145,7 @@ TEST(KVEngineTestHarness, SimpleSorted1) {
     ASSERT(engine);
 
     std::string ident = "abc";
-    std::string ns = "mydb.mycoll";
+    auto ns = NamespaceString("mydb.mycoll");
 
     std::unique_ptr<RecordStore> rs;
     {
@@ -158,22 +157,20 @@ TEST(KVEngineTestHarness, SimpleSorted1) {
     }
 
 
-    std::unique_ptr<CollectionCatalogEntryMock> catalogEntry =
-        std::make_unique<CollectionCatalogEntryMock>(ns);
     std::unique_ptr<CollectionImpl> collection;
     {
         MyOperationContext opCtx(engine);
         WriteUnitOfWork uow(&opCtx);
-        collection =
-            std::make_unique<CollectionImpl>(&opCtx, ns, UUID::gen(), catalogEntry.get(), rs.get());
+        collection = std::make_unique<CollectionImpl>(&opCtx, ns, UUID::gen(), std::move(rs));
         uow.commit();
     }
 
-    IndexDescriptor desc(
-        collection.get(),
-        "",
-        BSON("v" << static_cast<int>(IndexDescriptor::kLatestIndexVersion) << "ns" << ns << "key"
-                 << BSON("a" << 1)));
+    IndexDescriptor desc(collection.get(),
+                         "",
+                         BSON("v" << static_cast<int>(IndexDescriptor::kLatestIndexVersion) << "ns"
+                                  << ns.ns()
+                                  << "key"
+                                  << BSON("a" << 1)));
     std::unique_ptr<SortedDataInterface> sorted;
     {
         MyOperationContext opCtx(engine);
@@ -686,7 +683,7 @@ DEATH_TEST_F(DurableCatalogImplTest, TerminateOnNonNumericIndexVersion, "Fatal A
     ASSERT(engine);
 
     std::string ident = "abc";
-    std::string ns = "mydb.mycoll";
+    auto ns = NamespaceString("mydb.mycoll");
 
     std::unique_ptr<RecordStore> rs;
     {
@@ -697,14 +694,11 @@ DEATH_TEST_F(DurableCatalogImplTest, TerminateOnNonNumericIndexVersion, "Fatal A
         uow.commit();
     }
 
-    std::unique_ptr<CollectionCatalogEntryMock> catalogEntry =
-        std::make_unique<CollectionCatalogEntryMock>(ns);
     std::unique_ptr<CollectionImpl> collection;
     {
         MyOperationContext opCtx(engine);
         WriteUnitOfWork uow(&opCtx);
-        collection =
-            std::make_unique<CollectionImpl>(&opCtx, ns, UUID::gen(), catalogEntry.get(), rs.get());
+        collection = std::make_unique<CollectionImpl>(&opCtx, ns, UUID::gen(), std::move(rs));
         uow.commit();
     }
 
@@ -713,7 +707,7 @@ DEATH_TEST_F(DurableCatalogImplTest, TerminateOnNonNumericIndexVersion, "Fatal A
                          BSON("v"
                               << "1"
                               << "ns"
-                              << ns
+                              << ns.ns()
                               << "key"
                               << BSON("a" << 1)));
     std::unique_ptr<SortedDataInterface> sorted;
diff --git a/src/mongo/db/storage/storage_engine_impl.cpp b/src/mongo/db/storage/storage_engine_impl.cpp
index 92e91b1cec6..f7677175fa7 100644
--- a/src/mongo/db/storage/storage_engine_impl.cpp
+++ b/src/mongo/db/storage/storage_engine_impl.cpp
@@ -238,14 +238,30 @@ void StorageEngineImpl::loadCatalog(OperationContext* opCtx) {
 void StorageEngineImpl::_initCollection(OperationContext* opCtx,
                                         const NamespaceString& nss,
                                         bool forRepair) {
-    auto catalogEntry = _catalog->makeCollectionCatalogEntry(opCtx, nss, forRepair);
+    BSONCollectionCatalogEntry::MetaData md = _catalog->getMetaData(opCtx, nss);
+    uassert(ErrorCodes::MustDowngrade,
+            str::stream() << "Collection does not have UUID in KVCatalog. Collection: " << nss,
+            md.options.uuid);
+
+    auto ident = _catalog->getCollectionIdent(nss);
+
+    std::unique_ptr<RecordStore> rs;
+    if (forRepair) {
+        // Using a NULL rs since we don't want to open this record store before it has been
+        // repaired. This also ensures that if we try to use it, it will blow up.
+        rs = nullptr;
+    } else {
+        rs = _engine->getGroupedRecordStore(opCtx, nss.ns(), ident, md.options, md.prefix);
+        invariant(rs);
+    }
+
     auto uuid = _catalog->getCollectionOptions(opCtx, nss).uuid.get();
 
     auto collectionFactory = Collection::Factory::get(getGlobalServiceContext());
-    auto collection = collectionFactory->make(opCtx, uuid, catalogEntry.get());
+    auto collection = collectionFactory->make(opCtx, nss, uuid, std::move(rs));
 
     auto& collectionCatalog = CollectionCatalog::get(getGlobalServiceContext());
-    collectionCatalog.registerCollection(uuid, std::move(catalogEntry), std::move(collection));
+    collectionCatalog.registerCollection(uuid, std::move(collection));
 }
 
 void StorageEngineImpl::closeCatalog(OperationContext* opCtx) {
@@ -254,7 +270,7 @@ void StorageEngineImpl::closeCatalog(OperationContext* opCtx) {
         LOG_FOR_RECOVERY(kCatalogLogLevel) << "loadCatalog:";
         _dumpCatalog(opCtx);
     }
-    CollectionCatalog::get(opCtx).deregisterAllCatalogEntriesAndCollectionObjects();
+    CollectionCatalog::get(opCtx).deregisterAllCollections();
 
     _catalog.reset();
     _catalogRecordStore.reset();
@@ -507,8 +523,7 @@ void StorageEngineImpl::cleanShutdown() {
         _timestampMonitor->removeListener(&_minOfCheckpointAndOldestTimestampListener);
     }
 
-    CollectionCatalog::get(getGlobalServiceContext())
-        .deregisterAllCatalogEntriesAndCollectionObjects();
+    CollectionCatalog::get(getGlobalServiceContext()).deregisterAllCollections();
 
     _catalog.reset();
     _catalogRecordStore.reset();
@@ -600,11 +615,10 @@ Status StorageEngineImpl::_dropCollectionsNoTimestamp(OperationContext* opCtx,
             firstError = result;
         }
 
-        auto[removedColl, removedCatalogEntry] =
-            CollectionCatalog::get(opCtx).deregisterCollection(uuid);
+        auto removedColl = CollectionCatalog::get(opCtx).deregisterCollection(uuid);
         opCtx->recoveryUnit()->registerChange(
-            CollectionCatalog::get(opCtx).makeFinishDropCollectionChange(
-                std::move(removedColl), std::move(removedCatalogEntry), uuid));
+            CollectionCatalog::get(opCtx).makeFinishDropCollectionChange(std::move(removedColl),
+                                                                         uuid));
     }
 
     untimestampedDropWuow.commit();
@@ -960,24 +974,19 @@ void StorageEngineImpl::TimestampMonitor::removeListener(TimestampListener* list
 int64_t StorageEngineImpl::sizeOnDiskForDb(OperationContext* opCtx, StringData dbName) {
     int64_t size = 0;
 
-    catalog::forEachCollectionFromDb(
-        opCtx,
-        dbName,
-        MODE_IS,
-        [&](const Collection* collection, const CollectionCatalogEntry* catalogEntry) {
-            size += catalogEntry->getRecordStore()->storageSize(opCtx);
+    catalog::forEachCollectionFromDb(opCtx, dbName, MODE_IS, [&](const Collection* collection) {
+        size += collection->getRecordStore()->storageSize(opCtx);
 
-            std::vector<std::string> indexNames;
-            _catalog->getAllIndexes(opCtx, collection->ns(), &indexNames);
+        std::vector<std::string> indexNames;
+        _catalog->getAllIndexes(opCtx, collection->ns(), &indexNames);
 
-            for (size_t i = 0; i < indexNames.size(); i++) {
-                std::string ident =
-                    _catalog->getIndexIdent(opCtx, catalogEntry->ns(), indexNames[i]);
-                size += _engine->getIdentSize(opCtx, ident);
-            }
+        for (size_t i = 0; i < indexNames.size(); i++) {
+            std::string ident = _catalog->getIndexIdent(opCtx, collection->ns(), indexNames[i]);
+            size += _engine->getIdentSize(opCtx, ident);
+        }
 
-            return true;
-        });
+        return true;
+    });
 
     return size;
 }
diff --git a/src/mongo/db/storage/storage_engine_test_fixture.h b/src/mongo/db/storage/storage_engine_test_fixture.h
index 2cea5dce158..ffbb42ee70c 100644
--- a/src/mongo/db/storage/storage_engine_test_fixture.h
+++ b/src/mongo/db/storage/storage_engine_test_fixture.h
@@ -55,10 +55,10 @@ public:
         AutoGetDb db(opCtx, ns.db(), LockMode::MODE_X);
         CollectionOptions options;
         options.uuid = UUID::gen();
-        auto catalogEntry = unittest::assertGet(
+        auto rs = unittest::assertGet(
             _storageEngine->getCatalog()->createCollection(opCtx, ns, options, true));
-        CollectionCatalog::get(opCtx).registerCollection(
-            options.uuid.get(), std::move(catalogEntry), std::make_unique<CollectionMock>(ns));
+        CollectionCatalog::get(opCtx).registerCollection(options.uuid.get(),
+                                                         std::make_unique<CollectionMock>(ns));
 
         return _storageEngine->getCatalog()->getCollectionIdent(ns);
     }
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
index 4f78bcaf68a..ac2d365846c 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
@@ -54,7 +54,6 @@
 #include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/db/bson/dotted_path_support.h"
 #include "mongo/db/catalog/collection.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/client.h"
 #include "mongo/db/commands/server_status_metric.h"
 #include "mongo/db/concurrency/locker.h"
diff --git a/src/mongo/db/ttl.cpp b/src/mongo/db/ttl.cpp
index e3d14867b05..f9d90aecc4c 100644
--- a/src/mongo/db/ttl.cpp
+++ b/src/mongo/db/ttl.cpp
@@ -37,7 +37,6 @@
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/auth/user_name.h"
 #include "mongo/db/catalog/collection.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/client.h"
diff --git a/src/mongo/dbtests/indexcatalogtests.cpp b/src/mongo/dbtests/indexcatalogtests.cpp
index 2c2356d51c5..512a3e4b087 100644
--- a/src/mongo/dbtests/indexcatalogtests.cpp
+++ b/src/mongo/dbtests/indexcatalogtests.cpp
@@ -30,7 +30,6 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/catalog/collection.h"
-#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/client.h"
 #include "mongo/db/db_raii.h"