SERVER-75497 Convert ordered containers in CollectionCatalog to immutable

23 files changed, 285 insertions, 252 deletions

diff --git a/src/mongo/db/catalog/catalog_control.cpp b/src/mongo/db/catalog/catalog_control.cpp
index 3e94b7db845..f4c8cd93c53 100644
--- a/src/mongo/db/catalog/catalog_control.cpp
+++ b/src/mongo/db/catalog/catalog_control.cpp
@@ -147,8 +147,7 @@ PreviousCatalogState closeCatalog(OperationContext* opCtx) {
     auto databaseHolder = DatabaseHolder::get(opCtx);
     auto catalog = CollectionCatalog::get(opCtx);
     for (auto&& dbName : allDbs) {
-        for (auto collIt = catalog->begin(opCtx, dbName); collIt != catalog->end(opCtx); ++collIt) {
-            auto coll = *collIt;
+        for (auto&& coll : catalog->range(dbName)) {
             if (!coll) {
                 break;
             }
diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index cba44c2ac04..d8ba2b6b52f 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -64,6 +64,7 @@ const ServiceContext::Decoration<LatestCollectionCatalog> getCatalog =
 // batched write is ongoing without having to take locks.
 std::shared_ptr<CollectionCatalog> batchedCatalogWriteInstance;
 AtomicWord<bool> ongoingBatchedWrite{false};
+absl::flat_hash_set<Collection*> batchedCatalogClonedCollections;
 
 const RecoveryUnit::Snapshot::Decoration<std::shared_ptr<const CollectionCatalog>> stashedCatalog =
     RecoveryUnit::Snapshot::declareDecoration<std::shared_ptr<const CollectionCatalog>>();
@@ -158,7 +159,7 @@ public:
         catalog._collections = catalog._collections.set(collection->ns(), collection);
         catalog._catalog = catalog._catalog.set(collection->uuid(), collection);
         auto dbIdPair = std::make_pair(collection->ns().dbName(), collection->uuid());
-        catalog._orderedCollections[dbIdPair] = collection;
+        catalog._orderedCollections = catalog._orderedCollections.set(dbIdPair, collection);
 
         catalog._pendingCommitNamespaces = catalog._pendingCommitNamespaces.erase(collection->ns());
         catalog._pendingCommitUUIDs = catalog._pendingCommitUUIDs.erase(collection->uuid());
@@ -260,7 +261,6 @@ public:
                                          commitTime](CollectionCatalog& catalog) {
                         // Override existing Collection on this namespace
                         catalog._registerCollection(opCtx,
-                                                    uuid,
                                                     std::move(collection),
                                                     /*twoPhase=*/false,
                                                     /*ts=*/commitTime);
@@ -370,83 +370,64 @@ private:
     UncommittedCatalogUpdates& _uncommittedCatalogUpdates;
 };
 
-CollectionCatalog::iterator::iterator(OperationContext* opCtx,
-                                      const DatabaseName& dbName,
-                                      const CollectionCatalog& catalog)
-    : _opCtx(opCtx), _dbName(dbName), _catalog(&catalog) {
-
-    _mapIter = _catalog->_orderedCollections.lower_bound(std::make_pair(_dbName, minUuid));
-
-    // Start with the first collection that is visible outside of its transaction.
-    while (!_exhausted() && !_mapIter->second->isCommitted()) {
-        _mapIter++;
-    }
-
-    if (!_exhausted()) {
-        _uuid = _mapIter->first.second;
-    }
+CollectionCatalog::iterator::iterator(const DatabaseName& dbName,
+                                      OrderedCollectionMap::iterator it,
+                                      const OrderedCollectionMap& map)
+    : _map{map}, _mapIter{it}, _end(_map.upper_bound(std::make_pair(dbName, maxUuid))) {
+    _skipUncommitted();
 }
 
-CollectionCatalog::iterator::iterator(
-    OperationContext* opCtx,
-    std::map<std::pair<DatabaseName, UUID>, std::shared_ptr<Collection>>::const_iterator mapIter,
-    const CollectionCatalog& catalog)
-    : _opCtx(opCtx), _mapIter(mapIter), _catalog(&catalog) {}
-
 CollectionCatalog::iterator::value_type CollectionCatalog::iterator::operator*() {
-    if (_exhausted()) {
+    if (_mapIter == _map.end()) {
         return nullptr;
     }
-
     return _mapIter->second.get();
 }
 
-UUID CollectionCatalog::iterator::uuid() const {
-    invariant(_uuid);
-    return *_uuid;
-}
-
 CollectionCatalog::iterator CollectionCatalog::iterator::operator++() {
+    invariant(_mapIter != _map.end());
+    invariant(_mapIter != _end);
     _mapIter++;
+    _skipUncommitted();
+    return *this;
+}
 
-    // Skip any collections that are not yet visible outside of their respective transactions.
-    while (!_exhausted() && !_mapIter->second->isCommitted()) {
-        _mapIter++;
-    }
+bool CollectionCatalog::iterator::operator==(const iterator& other) const {
+    invariant(_map == other._map);
 
-    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;
+    if (other._mapIter == other._map.end()) {
+        return _mapIter == _map.end();
+    } else if (_mapIter == _map.end()) {
+        return other._mapIter == other._map.end();
     }
 
-    _uuid = _mapIter->first.second;
-    return *this;
+    return _mapIter->first.second == other._mapIter->first.second;
 }
 
-CollectionCatalog::iterator CollectionCatalog::iterator::operator++(int) {
-    auto oldPosition = *this;
-    ++(*this);
-    return oldPosition;
+bool CollectionCatalog::iterator::operator!=(const iterator& other) const {
+    return !(*this == other);
 }
 
-bool CollectionCatalog::iterator::operator==(const iterator& other) const {
-    invariant(_catalog == other._catalog);
-    if (other._mapIter == _catalog->_orderedCollections.end()) {
-        return _uuid == boost::none;
+void CollectionCatalog::iterator::_skipUncommitted() {
+    // Advance to the next collection that is visible outside of its transaction.
+    while (_mapIter != _end && !_mapIter->second->isCommitted()) {
+        ++_mapIter;
     }
+}
+
+CollectionCatalog::Range::Range(const OrderedCollectionMap& map, const DatabaseName& dbName)
+    : _map{map}, _dbName{dbName} {}
 
-    return _uuid == other._uuid;
+CollectionCatalog::iterator CollectionCatalog::Range::begin() const {
+    return {_dbName, _map.lower_bound(std::make_pair(_dbName, minUuid)), _map};
 }
 
-bool CollectionCatalog::iterator::operator!=(const iterator& other) const {
-    return !(*this == other);
+CollectionCatalog::iterator CollectionCatalog::Range::end() const {
+    return {_dbName, _map.upper_bound(std::make_pair(_dbName, maxUuid)), _map};
 }
 
-bool CollectionCatalog::iterator::_exhausted() {
-    return _mapIter == _catalog->_orderedCollections.end() || _mapIter->first.first != _dbName;
+bool CollectionCatalog::Range::empty() const {
+    return begin() == end();
 }
 
 std::shared_ptr<const CollectionCatalog> CollectionCatalog::latest(ServiceContext* svcCtx) {
@@ -1340,6 +1321,10 @@ uint64_t CollectionCatalog::getEpoch() const {
     return _epoch;
 }
 
+CollectionCatalog::Range CollectionCatalog::range(const DatabaseName& dbName) const {
+    return {_orderedCollections, dbName};
+}
+
 std::shared_ptr<const Collection> CollectionCatalog::_getCollectionByUUID(OperationContext* opCtx,
                                                                           const UUID& uuid) const {
     // It's important to look in UncommittedCatalogUpdates before OpenedCollections because in a
@@ -1402,6 +1387,7 @@ Collection* CollectionCatalog::lookupCollectionByUUIDForMetadataWrite(OperationC
     // on the thread doing the batch write and it would trigger the regular path where we do a
     // copy-on-write on the catalog when committing.
     if (_isCatalogBatchWriter()) {
+        batchedCatalogClonedCollections.emplace(cloned.get());
         // Do not update min valid timestamp in batched write as the write is not corresponding to
         // an oplog entry. If the write require an update to this timestamp it is the responsibility
         // of the user.
@@ -1530,6 +1516,7 @@ Collection* CollectionCatalog::lookupCollectionByNamespaceForMetadataWrite(
     // on the thread doing the batch write and it would trigger the regular path where we do a
     // copy-on-write on the catalog when committing.
     if (_isCatalogBatchWriter()) {
+        batchedCatalogClonedCollections.emplace(cloned.get());
         // Do not update min valid timestamp in batched write as the write is not corresponding to
         // an oplog entry. If the write require an update to this timestamp it is the responsibility
         // of the user.
@@ -1898,26 +1885,24 @@ CollectionCatalog::ViewCatalogSet CollectionCatalog::getViewCatalogDbNames(
 }
 
 void CollectionCatalog::registerCollection(OperationContext* opCtx,
-                                           const UUID& uuid,
                                            std::shared_ptr<Collection> coll,
                                            boost::optional<Timestamp> commitTime) {
     invariant(opCtx->lockState()->isW());
-    _registerCollection(opCtx, uuid, std::move(coll), /*twoPhase=*/false, commitTime);
+    _registerCollection(opCtx, std::move(coll), /*twoPhase=*/false, commitTime);
 }
 
 void CollectionCatalog::registerCollectionTwoPhase(OperationContext* opCtx,
-                                                   const UUID& uuid,
                                                    std::shared_ptr<Collection> coll,
                                                    boost::optional<Timestamp> commitTime) {
-    _registerCollection(opCtx, uuid, std::move(coll), /*twoPhase=*/true, commitTime);
+    _registerCollection(opCtx, std::move(coll), /*twoPhase=*/true, commitTime);
 }
 
 void CollectionCatalog::_registerCollection(OperationContext* opCtx,
-                                            const UUID& uuid,
                                             std::shared_ptr<Collection> coll,
                                             bool twoPhase,
                                             boost::optional<Timestamp> commitTime) {
     auto nss = coll->ns();
+    auto uuid = coll->uuid();
     _ensureNamespaceDoesNotExist(opCtx, nss, NamespaceType::kAll);
 
     LOGV2_DEBUG(20280,
@@ -1935,7 +1920,7 @@ void CollectionCatalog::_registerCollection(OperationContext* opCtx,
 
     _catalog = _catalog.set(uuid, coll);
     _collections = _collections.set(nss, coll);
-    _orderedCollections[dbIdPair] = coll;
+    _orderedCollections = _orderedCollections.set(dbIdPair, coll);
     if (twoPhase) {
         _pendingCommitNamespaces = _pendingCommitNamespaces.set(nss, coll);
         _pendingCommitUUIDs = _pendingCommitUUIDs.set(uuid, coll);
@@ -2015,7 +2000,7 @@ std::shared_ptr<Collection> CollectionCatalog::deregisterCollection(
         }
     }
 
-    _orderedCollections.erase(dbIdPair);
+    _orderedCollections = _orderedCollections.erase(dbIdPair);
     _collections = _collections.erase(ns);
     _catalog = _catalog.erase(uuid);
     _pendingCommitNamespaces = _pendingCommitNamespaces.erase(ns);
@@ -2114,7 +2099,7 @@ void CollectionCatalog::deregisterAllCollectionsAndViews(ServiceContext* svcCtx)
     }
 
     _collections = {};
-    _orderedCollections.clear();
+    _orderedCollections = {};
     _catalog = {};
     _viewsForDatabase = {};
     _dropPendingCollection = {};
@@ -2166,15 +2151,6 @@ void CollectionCatalog::notifyIdentDropped(const std::string& ident) {
     _dropPendingIndex = _dropPendingIndex.erase(ident);
 }
 
-CollectionCatalog::iterator CollectionCatalog::begin(OperationContext* opCtx,
-                                                     const DatabaseName& dbName) const {
-    return iterator(opCtx, dbName, *this);
-}
-
-CollectionCatalog::iterator CollectionCatalog::end(OperationContext* opCtx) const {
-    return iterator(opCtx, _orderedCollections.end(), *this);
-}
-
 void CollectionCatalog::invariantHasExclusiveAccessToCollection(OperationContext* opCtx,
                                                                 const NamespaceString& nss) {
     invariant(hasExclusiveAccessToCollection(opCtx, nss), nss.toStringForErrorMsg());
@@ -2226,22 +2202,16 @@ bool CollectionCatalog::_isCatalogBatchWriter() const {
 
 bool CollectionCatalog::_alreadyClonedForBatchedWriter(
     const std::shared_ptr<Collection>& collection) const {
-    // We may skip cloning the Collection instance if and only if we are currently in a batched
-    // catalog write and all references to this Collection is owned by the cloned CollectionCatalog
-    // instance owned by the batch writer. i.e. the Collection is uniquely owned by the batch
-    // writer. When the batch writer initially clones the catalog, all collections will have a
-    // 'use_count' of at least kNumCollectionReferencesStored*2 (because there are at least 2
-    // catalog instances). To check for uniquely owned we need to check that the reference count is
-    // exactly kNumCollectionReferencesStored (owned by a single catalog) while also account for the
-    // instance that is extracted from the catalog and provided as a parameter to this function, we
-    // therefore need to add 1.
-    return _isCatalogBatchWriter() && collection.use_count() == kNumCollectionReferencesStored + 1;
+    // We may skip cloning the Collection instance if and only if have already cloned it for write
+    // use in this batch writer.
+    return _isCatalogBatchWriter() && batchedCatalogClonedCollections.contains(collection.get());
 }
 
 BatchedCollectionCatalogWriter::BatchedCollectionCatalogWriter(OperationContext* opCtx)
     : _opCtx(opCtx) {
     invariant(_opCtx->lockState()->isW());
     invariant(!batchedCatalogWriteInstance);
+    invariant(batchedCatalogClonedCollections.empty());
 
     auto& storage = getCatalog(_opCtx->getServiceContext());
     // hold onto base so if we need to delete it we can do it outside of the lock
@@ -2266,6 +2236,7 @@ BatchedCollectionCatalogWriter::~BatchedCollectionCatalogWriter() {
     ongoingBatchedWrite.store(false);
     _batchedInstance = nullptr;
     batchedCatalogWriteInstance = nullptr;
+    batchedCatalogClonedCollections.clear();
 }
 
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/collection_catalog.h b/src/mongo/db/catalog/collection_catalog.h
index a5086bba80d..da306512fec 100644
--- a/src/mongo/db/catalog/collection_catalog.h
+++ b/src/mongo/db/catalog/collection_catalog.h
@@ -43,6 +43,7 @@
 #include "mongo/db/views/view.h"
 #include "mongo/stdx/unordered_map.h"
 #include "mongo/util/functional.h"
+#include "mongo/util/immutable/map.h"
 #include "mongo/util/immutable/unordered_map.h"
 #include "mongo/util/immutable/unordered_set.h"
 #include "mongo/util/uuid.h"
@@ -51,47 +52,44 @@ namespace mongo {
 
 class CollectionCatalog {
     friend class iterator;
+    using OrderedCollectionMap =
+        immutable::map<std::pair<DatabaseName, UUID>, std::shared_ptr<Collection>>;
 
 public:
     using CollectionInfoFn = std::function<bool(const Collection* collection)>;
 
-    // Number of how many Collection references for a single Collection that is stored in the
-    // catalog. Used to determine whether there are external references (uniquely owned). Needs to
-    // be kept in sync with the data structures below.
-    static constexpr size_t kNumCollectionReferencesStored = 3;
 
     class iterator {
     public:
         using value_type = const Collection*;
 
-        iterator(OperationContext* opCtx,
-                 const DatabaseName& dbName,
-                 const CollectionCatalog& catalog);
-        iterator(OperationContext* opCtx,
-                 std::map<std::pair<DatabaseName, UUID>,
-                          std::shared_ptr<Collection>>::const_iterator mapIter,
-                 const CollectionCatalog& catalog);
+        iterator(const DatabaseName& dbName,
+                 OrderedCollectionMap::iterator it,
+                 const OrderedCollectionMap& catalog);
         value_type operator*();
         iterator operator++();
-        iterator operator++(int);
-        UUID uuid() const;
-
-        /*
-         * Equality operators == and != do not attempt to reposition the iterators being compared.
-         * The behavior for comparing invalid iterators is undefined.
-         */
         bool operator==(const iterator& other) const;
         bool operator!=(const iterator& other) const;
 
     private:
-        bool _exhausted();
+        void _skipUncommitted();
 
-        OperationContext* _opCtx;
-        DatabaseName _dbName;
-        boost::optional<UUID> _uuid;
-        std::map<std::pair<DatabaseName, UUID>, std::shared_ptr<Collection>>::const_iterator
+        const OrderedCollectionMap& _map;
+        immutable::map<std::pair<DatabaseName, UUID>, std::shared_ptr<Collection>>::iterator
             _mapIter;
-        const CollectionCatalog* _catalog;
+        immutable::map<std::pair<DatabaseName, UUID>, std::shared_ptr<Collection>>::iterator _end;
+    };
+
+    class Range {
+    public:
+        Range(const OrderedCollectionMap&, const DatabaseName& dbName);
+        iterator begin() const;
+        iterator end() const;
+        bool empty() const;
+
+    private:
+        OrderedCollectionMap _map;
+        DatabaseName _dbName;
     };
 
     struct ProfileSettings {
@@ -308,7 +306,6 @@ public:
      * The global lock must be held in exclusive mode.
      */
     void registerCollection(OperationContext* opCtx,
-                            const UUID& uuid,
                             std::shared_ptr<Collection> collection,
                             boost::optional<Timestamp> commitTime);
 
@@ -319,7 +316,6 @@ public:
      * 'onCreateCollection' which sets up the necessary state for finishing the two-phase commit.
      */
     void registerCollectionTwoPhase(OperationContext* opCtx,
-                                    const UUID& uuid,
                                     std::shared_ptr<Collection> collection,
                                     boost::optional<Timestamp> commitTime);
 
@@ -632,8 +628,13 @@ public:
      */
     uint64_t getEpoch() const;
 
-    iterator begin(OperationContext* opCtx, const DatabaseName& dbName) const;
-    iterator end(OperationContext* opCtx) const;
+    /**
+     * Provides an iterable range for the collections belonging to the specified database.
+     *
+     * Will not observe any updates made to the catalog after the creation of the 'Range'. The
+     * 'Range' object just remain in scope for the duration of the iteration.
+     */
+    Range range(const DatabaseName& dbName) const;
 
     /**
      * Ensures we have a MODE_X lock on a collection or MODE_IX lock for newly created collections.
@@ -663,13 +664,12 @@ private:
                                                            const UUID& uuid) const;
 
     /**
-     * Register the collection with `uuid`.
+     * Register the collection.
      *
      * If 'twoPhase' is true, this call must be followed by 'onCreateCollection' which continues the
      * two-phase commit process.
      */
     void _registerCollection(OperationContext* opCtx,
-                             const UUID& uuid,
                              std::shared_ptr<Collection> collection,
                              bool twoPhase,
                              boost::optional<Timestamp> commitTime);
@@ -798,8 +798,6 @@ private:
 
     using CollectionCatalogMap =
         immutable::unordered_map<UUID, std::shared_ptr<Collection>, UUID::Hash>;
-    using OrderedCollectionMap =
-        std::map<std::pair<DatabaseName, UUID>, std::shared_ptr<Collection>>;
     using NamespaceCollectionMap =
         immutable::unordered_map<NamespaceString, std::shared_ptr<Collection>>;
     using UncommittedViewsSet = immutable::unordered_set<NamespaceString>;
diff --git a/src/mongo/db/catalog/collection_catalog_bm.cpp b/src/mongo/db/catalog/collection_catalog_bm.cpp
index a78b9d1e3d6..57898636095 100644
--- a/src/mongo/db/catalog/collection_catalog_bm.cpp
+++ b/src/mongo/db/catalog/collection_catalog_bm.cpp
@@ -80,7 +80,6 @@ void createCollections(OperationContext* opCtx, int numCollections) {
         const NamespaceString nss("collection_catalog_bm", std::to_string(i));
         CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
             catalog.registerCollection(opCtx,
-                                       UUID::gen(),
                                        std::make_shared<CollectionMock>(nss),
                                        /*ts=*/boost::none);
         });
@@ -124,7 +123,127 @@ void BM_CollectionCatalogWriteBatchedWithGlobalExclusiveLock(benchmark::State& s
     }
 }
 
+void BM_CollectionCatalogCreateDropCollection(benchmark::State& state) {
+    auto serviceContext = setupServiceContext();
+    ThreadClient threadClient(serviceContext);
+    ServiceContext::UniqueOperationContext opCtx = threadClient->makeOperationContext();
+    Lock::GlobalLock globalLk(opCtx.get(), MODE_X);
+
+    createCollections(opCtx.get(), state.range(0));
+
+    for (auto _ : state) {
+        benchmark::ClobberMemory();
+        CollectionCatalog::write(opCtx.get(), [&](CollectionCatalog& catalog) {
+            const NamespaceString nss("collection_catalog_bm", std::to_string(state.range(0)));
+            const UUID uuid = UUID::gen();
+            catalog.registerCollection(
+                opCtx.get(), std::make_shared<CollectionMock>(uuid, nss), boost::none);
+            catalog.deregisterCollection(opCtx.get(), uuid, false, boost::none);
+        });
+    }
+}
+
+void BM_CollectionCatalogCreateNCollectionsBatched(benchmark::State& state) {
+    for (auto _ : state) {
+        benchmark::ClobberMemory();
+
+        auto serviceContext = setupServiceContext();
+        ThreadClient threadClient(serviceContext);
+        ServiceContext::UniqueOperationContext opCtx = threadClient->makeOperationContext();
+
+        Lock::GlobalLock globalLk(opCtx.get(), MODE_X);
+        BatchedCollectionCatalogWriter batched(opCtx.get());
+
+        auto numCollections = state.range(0);
+        for (auto i = 0; i < numCollections; i++) {
+            const NamespaceString nss("collection_catalog_bm", std::to_string(i));
+            CollectionCatalog::write(opCtx.get(), [&](CollectionCatalog& catalog) {
+                catalog.registerCollection(
+                    opCtx.get(), std::make_shared<CollectionMock>(nss), boost::none);
+            });
+        }
+    }
+}
+
+void BM_CollectionCatalogCreateNCollections(benchmark::State& state) {
+    for (auto _ : state) {
+        benchmark::ClobberMemory();
+
+        auto serviceContext = setupServiceContext();
+        ThreadClient threadClient(serviceContext);
+        ServiceContext::UniqueOperationContext opCtx = threadClient->makeOperationContext();
+        Lock::GlobalLock globalLk(opCtx.get(), MODE_X);
+
+        auto numCollections = state.range(0);
+        for (auto i = 0; i < numCollections; i++) {
+            const NamespaceString nss("collection_catalog_bm", std::to_string(i));
+            CollectionCatalog::write(opCtx.get(), [&](CollectionCatalog& catalog) {
+                catalog.registerCollection(
+                    opCtx.get(), std::make_shared<CollectionMock>(nss), boost::none);
+            });
+        }
+    }
+}
+
+void BM_CollectionCatalogLookupCollectionByNamespace(benchmark::State& state) {
+    auto serviceContext = setupServiceContext();
+    ThreadClient threadClient(serviceContext);
+    ServiceContext::UniqueOperationContext opCtx = threadClient->makeOperationContext();
+
+    createCollections(opCtx.get(), state.range(0));
+    const NamespaceString nss("collection_catalog_bm", std::to_string(state.range(0) / 2));
+
+    for (auto _ : state) {
+        benchmark::ClobberMemory();
+        auto coll =
+            CollectionCatalog::get(opCtx.get())->lookupCollectionByNamespace(opCtx.get(), nss);
+        invariant(coll);
+    }
+}
+
+void BM_CollectionCatalogLookupCollectionByUUID(benchmark::State& state) {
+    auto serviceContext = setupServiceContext();
+    ThreadClient threadClient(serviceContext);
+    ServiceContext::UniqueOperationContext opCtx = threadClient->makeOperationContext();
+
+    createCollections(opCtx.get(), state.range(0));
+    const NamespaceString nss("collection_catalog_bm", std::to_string(state.range(0) / 2));
+    auto coll = CollectionCatalog::get(opCtx.get())->lookupCollectionByNamespace(opCtx.get(), nss);
+    invariant(coll->ns() == nss);
+    const UUID uuid = coll->uuid();
+
+    for (auto _ : state) {
+        benchmark::ClobberMemory();
+        auto res = CollectionCatalog::get(opCtx.get())->lookupCollectionByUUID(opCtx.get(), uuid);
+        invariant(res == coll);
+    }
+}
+
+void BM_CollectionCatalogIterateCollections(benchmark::State& state) {
+    auto serviceContext = setupServiceContext();
+    ThreadClient threadClient(serviceContext);
+    ServiceContext::UniqueOperationContext opCtx = threadClient->makeOperationContext();
+
+    createCollections(opCtx.get(), state.range(0));
+
+    for (auto _ : state) {
+        benchmark::ClobberMemory();
+        auto catalog = CollectionCatalog::get(opCtx.get());
+        auto count = 0;
+        for ([[maybe_unused]] auto&& coll : catalog->range(
+                 DatabaseName::createDatabaseName_forTest(boost::none, "collection_catalog_bm"))) {
+            benchmark::DoNotOptimize(count++);
+        }
+    }
+}
+
 BENCHMARK(BM_CollectionCatalogWrite)->Ranges({{{1}, {100'000}}});
 BENCHMARK(BM_CollectionCatalogWriteBatchedWithGlobalExclusiveLock)->Ranges({{{1}, {100'000}}});
+BENCHMARK(BM_CollectionCatalogCreateDropCollection)->Ranges({{{1}, {100'000}}});
+BENCHMARK(BM_CollectionCatalogCreateNCollectionsBatched)->Ranges({{{1}, {100'000}}});
+BENCHMARK(BM_CollectionCatalogCreateNCollections)->Ranges({{{1}, {32'768}}});
+BENCHMARK(BM_CollectionCatalogLookupCollectionByNamespace)->Ranges({{{1}, {100'000}}});
+BENCHMARK(BM_CollectionCatalogLookupCollectionByUUID)->Ranges({{{1}, {100'000}}});
+BENCHMARK(BM_CollectionCatalogIterateCollections)->Ranges({{{1}, {100'000}}});
 
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/collection_catalog_helper.cpp b/src/mongo/db/catalog/collection_catalog_helper.cpp
index e74c8cd05b9..c9dadb96b57 100644
--- a/src/mongo/db/catalog/collection_catalog_helper.cpp
+++ b/src/mongo/db/catalog/collection_catalog_helper.cpp
@@ -69,11 +69,9 @@ void forEachCollectionFromDb(OperationContext* opCtx,
                              CollectionCatalog::CollectionInfoFn predicate) {
 
     auto catalogForIteration = CollectionCatalog::get(opCtx);
-    for (auto collectionIt = catalogForIteration->begin(opCtx, dbName);
-         collectionIt != catalogForIteration->end(opCtx);) {
-        auto uuid = collectionIt.uuid();
+    for (auto&& coll : catalogForIteration->range(dbName)) {
+        auto uuid = coll->uuid();
         if (predicate && !catalogForIteration->checkIfCollectionSatisfiable(uuid, predicate)) {
-            ++collectionIt;
             continue;
         }
 
@@ -97,10 +95,6 @@ void forEachCollectionFromDb(OperationContext* opCtx,
             clk.reset();
         }
 
-        // Increment iterator before calling callback. This allows for collection deletion inside
-        // this callback even if we are in batched inplace mode.
-        ++collectionIt;
-
         // The NamespaceString couldn't be resolved from the uuid, so the collection was dropped.
         if (!collection)
             continue;
diff --git a/src/mongo/db/catalog/collection_catalog_test.cpp b/src/mongo/db/catalog/collection_catalog_test.cpp
index 8678535c174..5e23436c58b 100644
--- a/src/mongo/db/catalog/collection_catalog_test.cpp
+++ b/src/mongo/db/catalog/collection_catalog_test.cpp
@@ -78,12 +78,7 @@ public:
         std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(colUUID, nss);
         col = CollectionPtr(collection.get());
         // Register dummy collection in catalog.
-        catalog.registerCollection(opCtx.get(), colUUID, collection, boost::none);
-
-        // Validate that kNumCollectionReferencesStored is correct, add one reference for the one we
-        // hold in this function.
-        ASSERT_EQUALS(collection.use_count(),
-                      CollectionCatalog::kNumCollectionReferencesStored + 1);
+        catalog.registerCollection(opCtx.get(), collection, boost::none);
     }
 
     void tearDown() {
@@ -115,17 +110,14 @@ public:
             NamespaceString barNss = NamespaceString::createNamespaceString_forTest(
                 "bar", "coll" + std::to_string(counter));
 
-            auto fooUuid = UUID::gen();
             std::shared_ptr<Collection> fooColl = std::make_shared<CollectionMock>(fooNss);
-
-            auto barUuid = UUID::gen();
             std::shared_ptr<Collection> barColl = std::make_shared<CollectionMock>(barNss);
 
-            dbMap["foo"].insert(std::make_pair(fooUuid, fooColl.get()));
-            dbMap["bar"].insert(std::make_pair(barUuid, barColl.get()));
+            dbMap["foo"].insert(std::make_pair(fooColl->uuid(), fooColl.get()));
+            dbMap["bar"].insert(std::make_pair(barColl->uuid(), barColl.get()));
 
-            catalog.registerCollection(opCtx.get(), fooUuid, fooColl, boost::none);
-            catalog.registerCollection(opCtx.get(), barUuid, barColl, boost::none);
+            catalog.registerCollection(opCtx.get(), fooColl, boost::none);
+            catalog.registerCollection(opCtx.get(), barColl, boost::none);
         }
     }
 
@@ -154,9 +146,11 @@ public:
     void checkCollections(const DatabaseName& dbName) {
         unsigned long counter = 0;
         const auto dbNameStr = dbName.toString_forTest();
-        for (auto [orderedIt, catalogIt] =
-                 std::tuple{collsIterator(dbNameStr), catalog.begin(opCtx.get(), dbName)};
-             catalogIt != catalog.end(opCtx.get()) && orderedIt != collsIteratorEnd(dbNameStr);
+
+        auto orderedIt = collsIterator(dbNameStr);
+        auto catalogRange = catalog.range(dbName);
+        auto catalogIt = catalogRange.begin();
+        for (; catalogIt != catalogRange.end() && orderedIt != collsIteratorEnd(dbNameStr);
              ++catalogIt, ++orderedIt) {
 
             auto catalogColl = *catalogIt;
@@ -191,17 +185,13 @@ public:
             NamespaceString nss = NamespaceString::createNamespaceString_forTest(
                 "resourceDb", "coll" + std::to_string(i));
             std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
-            auto uuid = collection->uuid();
 
-            catalog.registerCollection(opCtx.get(), uuid, std::move(collection), boost::none);
+            catalog.registerCollection(opCtx.get(), std::move(collection), boost::none);
         }
 
         int numEntries = 0;
-        for (auto it = catalog.begin(
-                 opCtx.get(), DatabaseName::createDatabaseName_forTest(boost::none, "resourceDb"));
-             it != catalog.end(opCtx.get());
-             it++) {
-            auto coll = *it;
+        for (auto&& coll :
+             catalog.range(DatabaseName::createDatabaseName_forTest(boost::none, "resourceDb"))) {
             auto collName = coll->ns();
             ResourceId rid(RESOURCE_COLLECTION, collName);
 
@@ -213,11 +203,8 @@ public:
 
     void tearDown() {
         std::vector<UUID> collectionsToDeregister;
-        for (auto it = catalog.begin(
-                 opCtx.get(), DatabaseName::createDatabaseName_forTest(boost::none, "resourceDb"));
-             it != catalog.end(opCtx.get());
-             ++it) {
-            auto coll = *it;
+        for (auto&& coll :
+             catalog.range(DatabaseName::createDatabaseName_forTest(boost::none, "resourceDb"))) {
             auto uuid = coll->uuid();
             if (!coll) {
                 break;
@@ -231,10 +218,8 @@ public:
         }
 
         int numEntries = 0;
-        for (auto it = catalog.begin(
-                 opCtx.get(), DatabaseName::createDatabaseName_forTest(boost::none, "resourceDb"));
-             it != catalog.end(opCtx.get());
-             it++) {
+        for ([[maybe_unused]] auto&& coll :
+             catalog.range(DatabaseName::createDatabaseName_forTest(boost::none, "resourceDb"))) {
             numEntries++;
         }
         ASSERT_EQ(0, numEntries);
@@ -317,14 +302,14 @@ TEST_F(CollectionCatalogIterationTest, EndAtEndOfSection) {
 }
 
 TEST_F(CollectionCatalogIterationTest, GetUUIDWontRepositionEvenIfEntryIsDropped) {
-    auto it =
-        catalog.begin(opCtx.get(), DatabaseName::createDatabaseName_forTest(boost::none, "bar"));
+    auto range = catalog.range(DatabaseName::createDatabaseName_forTest(boost::none, "bar"));
+    auto it = range.begin();
     auto collsIt = collsIterator("bar");
     auto uuid = collsIt->first;
     catalog.deregisterCollection(opCtx.get(), uuid, /*isDropPending=*/false, boost::none);
     dropColl("bar", uuid);
 
-    ASSERT_EQUALS(uuid, it.uuid());
+    ASSERT_EQUALS(uuid, (*it)->uuid());
 }
 
 TEST_F(CollectionCatalogTest, OnCreateCollection) {
@@ -349,13 +334,13 @@ TEST_F(CollectionCatalogTest, LookupNSSByUUID) {
 TEST_F(CollectionCatalogTest, InsertAfterLookup) {
     auto newUUID = UUID::gen();
     NamespaceString newNss = NamespaceString::createNamespaceString_forTest(nss.dbName(), "newcol");
-    std::shared_ptr<Collection> newCollShared = std::make_shared<CollectionMock>(newNss);
+    std::shared_ptr<Collection> newCollShared = std::make_shared<CollectionMock>(newUUID, newNss);
     auto newCol = newCollShared.get();
 
     // Ensure that looking up non-existing UUIDs doesn't affect later registration of those UUIDs.
     ASSERT(catalog.lookupCollectionByUUID(opCtx.get(), newUUID) == nullptr);
     ASSERT_EQUALS(catalog.lookupNSSByUUID(opCtx.get(), newUUID), boost::none);
-    catalog.registerCollection(opCtx.get(), newUUID, std::move(newCollShared), boost::none);
+    catalog.registerCollection(opCtx.get(), std::move(newCollShared), boost::none);
     ASSERT_EQUALS(catalog.lookupCollectionByUUID(opCtx.get(), newUUID), newCol);
     ASSERT_EQUALS(*catalog.lookupNSSByUUID(opCtx.get(), colUUID), nss);
 }
@@ -402,7 +387,7 @@ TEST_F(CollectionCatalogTest, RenameCollection) {
     NamespaceString oldNss = NamespaceString::createNamespaceString_forTest(nss.dbName(), "oldcol");
     std::shared_ptr<Collection> collShared = std::make_shared<CollectionMock>(uuid, oldNss);
     auto collection = collShared.get();
-    catalog.registerCollection(opCtx.get(), uuid, std::move(collShared), boost::none);
+    catalog.registerCollection(opCtx.get(), std::move(collShared), boost::none);
     CollectionPtr yieldableColl(catalog.lookupCollectionByUUID(opCtx.get(), uuid));
     ASSERT(yieldableColl);
     ASSERT_EQUALS(yieldableColl, CollectionPtr(collection));
@@ -461,7 +446,7 @@ TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsOldNSSIfDrop
 TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsNewlyCreatedNSS) {
     auto newUUID = UUID::gen();
     NamespaceString newNss = NamespaceString::createNamespaceString_forTest(nss.dbName(), "newcol");
-    std::shared_ptr<Collection> newCollShared = std::make_shared<CollectionMock>(newNss);
+    std::shared_ptr<Collection> newCollShared = std::make_shared<CollectionMock>(newUUID, newNss);
     auto newCol = newCollShared.get();
 
     // Ensure that looking up non-existing UUIDs doesn't affect later registration of those UUIDs.
@@ -472,7 +457,7 @@ TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsNewlyCreated
 
     ASSERT(catalog.lookupCollectionByUUID(opCtx.get(), newUUID) == nullptr);
     ASSERT_EQUALS(catalog.lookupNSSByUUID(opCtx.get(), newUUID), boost::none);
-    catalog.registerCollection(opCtx.get(), newUUID, std::move(newCollShared), boost::none);
+    catalog.registerCollection(opCtx.get(), std::move(newCollShared), boost::none);
     ASSERT_EQUALS(catalog.lookupCollectionByUUID(opCtx.get(), newUUID), newCol);
     ASSERT_EQUALS(*catalog.lookupNSSByUUID(opCtx.get(), colUUID), nss);
 
@@ -488,7 +473,7 @@ TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsNewlyCreated
 
 TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsFreshestNSS) {
     NamespaceString newNss = NamespaceString::createNamespaceString_forTest(nss.dbName(), "newcol");
-    std::shared_ptr<Collection> newCollShared = std::make_shared<CollectionMock>(newNss);
+    std::shared_ptr<Collection> newCollShared = std::make_shared<CollectionMock>(colUUID, newNss);
     auto newCol = newCollShared.get();
 
     {
@@ -501,7 +486,7 @@ TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsFreshestNSS)
     ASSERT_EQUALS(*catalog.lookupNSSByUUID(opCtx.get(), colUUID), nss);
     {
         Lock::GlobalWrite lk(opCtx.get());
-        catalog.registerCollection(opCtx.get(), colUUID, std::move(newCollShared), boost::none);
+        catalog.registerCollection(opCtx.get(), std::move(newCollShared), boost::none);
     }
 
     ASSERT_EQUALS(catalog.lookupCollectionByUUID(opCtx.get(), colUUID), newCol);
@@ -543,8 +528,7 @@ TEST_F(CollectionCatalogTest, GetAllCollectionNamesAndGetAllDbNames) {
     std::vector<NamespaceString> nsss = {aColl, b1Coll, b2Coll, cColl, d1Coll, d2Coll, d3Coll};
     for (auto& nss : nsss) {
         std::shared_ptr<Collection> newColl = std::make_shared<CollectionMock>(nss);
-        auto uuid = UUID::gen();
-        catalog.registerCollection(opCtx.get(), uuid, std::move(newColl), boost::none);
+        catalog.registerCollection(opCtx.get(), std::move(newColl), boost::none);
     }
 
     std::vector<NamespaceString> dCollList = {d1Coll, d2Coll, d3Coll};
@@ -576,8 +560,7 @@ TEST_F(CollectionCatalogTest, GetAllDbNamesForTenant) {
     std::vector<NamespaceString> nsss = {dbA, dbB, dbC, dbD};
     for (auto& nss : nsss) {
         std::shared_ptr<Collection> newColl = std::make_shared<CollectionMock>(nss);
-        auto uuid = UUID::gen();
-        catalog.registerCollection(opCtx.get(), uuid, std::move(newColl), boost::none);
+        catalog.registerCollection(opCtx.get(), std::move(newColl), boost::none);
     }
 
     std::vector<DatabaseName> dbNamesForTid1 = {
@@ -606,8 +589,7 @@ TEST_F(CollectionCatalogTest, GetAllTenants) {
 
     for (auto& nss : nsss) {
         std::shared_ptr<Collection> newColl = std::make_shared<CollectionMock>(nss);
-        auto uuid = UUID::gen();
-        catalog.registerCollection(opCtx.get(), uuid, std::move(newColl), boost::none);
+        catalog.registerCollection(opCtx.get(), std::move(newColl), boost::none);
     }
 
     std::set<TenantId> expectedTenants = {tid1, tid2};
@@ -652,8 +634,7 @@ TEST_F(CollectionCatalogTest, GetAllCollectionNamesAndGetAllDbNamesWithUncommitt
     std::vector<NamespaceString> nsss = {aColl, b1Coll, b2Coll, cColl, d1Coll, d2Coll, d3Coll};
     for (auto& nss : nsss) {
         std::shared_ptr<Collection> newColl = std::make_shared<CollectionMock>(nss);
-        auto uuid = UUID::gen();
-        catalog.registerCollection(opCtx.get(), uuid, std::move(newColl), boost::none);
+        catalog.registerCollection(opCtx.get(), std::move(newColl), boost::none);
     }
 
     // One dbName with only an invisible collection does not appear in dbNames. Use const_cast to
diff --git a/src/mongo/db/catalog/collection_writer_test.cpp b/src/mongo/db/catalog/collection_writer_test.cpp
index 15368437ff4..f509ffa8bb9 100644
--- a/src/mongo/db/catalog/collection_writer_test.cpp
+++ b/src/mongo/db/catalog/collection_writer_test.cpp
@@ -59,9 +59,8 @@ protected:
 
         std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(kNss);
         CollectionCatalog::write(getServiceContext(), [&](CollectionCatalog& catalog) {
-            auto uuid = collection->uuid();
             catalog.registerCollection(
-                operationContext(), uuid, std::move(collection), /*ts=*/boost::none);
+                operationContext(), std::move(collection), /*ts=*/boost::none);
         });
     }
 
@@ -255,7 +254,6 @@ public:
             for (size_t i = 0; i < NumCollections; ++i) {
                 catalog.registerCollection(
                     operationContext(),
-                    UUID::gen(),
                     std::make_shared<CollectionMock>(NamespaceString::createNamespaceString_forTest(
                         "many", fmt::format("coll{}", i))),
                     /*ts=*/boost::none);
diff --git a/src/mongo/db/catalog/database_holder_impl.cpp b/src/mongo/db/catalog/database_holder_impl.cpp
index 964dc468284..49ef7b48e1a 100644
--- a/src/mongo/db/catalog/database_holder_impl.cpp
+++ b/src/mongo/db/catalog/database_holder_impl.cpp
@@ -179,8 +179,7 @@ void DatabaseHolderImpl::dropDb(OperationContext* opCtx, Database* db) {
     invariant(opCtx->lockState()->isDbLockedForMode(name, MODE_X));
 
     auto catalog = CollectionCatalog::get(opCtx);
-    for (auto collIt = catalog->begin(opCtx, name); collIt != catalog->end(opCtx); ++collIt) {
-        auto coll = *collIt;
+    for (auto&& coll : catalog->range(name)) {
         if (!coll) {
             break;
         }
@@ -196,8 +195,7 @@ void DatabaseHolderImpl::dropDb(OperationContext* opCtx, Database* db) {
 
     auto const serviceContext = opCtx->getServiceContext();
 
-    for (auto collIt = catalog->begin(opCtx, name); collIt != catalog->end(opCtx); ++collIt) {
-        auto coll = *collIt;
+    for (auto&& coll : catalog->range(name)) {
         if (!coll) {
             break;
         }
diff --git a/src/mongo/db/catalog/drop_database.cpp b/src/mongo/db/catalog/drop_database.cpp
index 9e157623bf8..15dee02d730 100644
--- a/src/mongo/db/catalog/drop_database.cpp
+++ b/src/mongo/db/catalog/drop_database.cpp
@@ -273,9 +273,7 @@ Status _dropDatabase(OperationContext* opCtx, const DatabaseName& dbName, bool a
         catalog = CollectionCatalog::get(opCtx);
 
         std::vector<NamespaceString> collectionsToDrop;
-        for (auto collIt = catalog->begin(opCtx, db->name()); collIt != catalog->end(opCtx);
-             ++collIt) {
-            auto collection = *collIt;
+        for (auto&& collection : catalog->range(db->name())) {
             if (!collection) {
                 break;
             }
diff --git a/src/mongo/db/catalog/uncommitted_catalog_updates.cpp b/src/mongo/db/catalog/uncommitted_catalog_updates.cpp
index 751c0b25f6f..87109f28314 100644
--- a/src/mongo/db/catalog/uncommitted_catalog_updates.cpp
+++ b/src/mongo/db/catalog/uncommitted_catalog_updates.cpp
@@ -122,7 +122,7 @@ void UncommittedCatalogUpdates::_createCollection(OperationContext* opCtx,
 
             // This will throw when registering a namespace which is already in use.
             CollectionCatalog::write(opCtx, [&, coll = createdColl](CollectionCatalog& catalog) {
-                catalog.registerCollectionTwoPhase(opCtx, uuid, coll, /*ts=*/boost::none);
+                catalog.registerCollectionTwoPhase(opCtx, coll, /*ts=*/boost::none);
             });
 
             opCtx->recoveryUnit()->onRollback([uuid](OperationContext* opCtx) {
diff --git a/src/mongo/db/commands/dbhash.cpp b/src/mongo/db/commands/dbhash.cpp
index dc727fbdb4d..b6e7297f766 100644
--- a/src/mongo/db/commands/dbhash.cpp
+++ b/src/mongo/db/commands/dbhash.cpp
@@ -311,8 +311,8 @@ public:
             return true;
         };
 
-        for (auto it = catalog->begin(opCtx, dbName); it != catalog->end(opCtx); ++it) {
-            UUID uuid = it.uuid();
+        for (auto&& coll : catalog->range(dbName)) {
+            UUID uuid = coll->uuid();
 
             // The namespace must be found as the UUID is fetched from the same
             // CollectionCatalog instance.
@@ -322,14 +322,14 @@ public:
             // TODO:SERVER-75848 Make this lock-free
             Lock::CollectionLock clk(opCtx, *nss, MODE_IS);
 
-            const Collection* coll = nullptr;
+            const Collection* collection = nullptr;
             if (nss->isGlobalIndex()) {
                 // TODO SERVER-74209: Reading earlier than the minimum valid snapshot is not
                 // supported for global indexes. It appears that the primary and secondaries apply
                 // operations differently resulting in hash mismatches. This requires further
                 // investigation. In the meantime, global indexes use the behaviour prior to
                 // point-in-time lookups.
-                coll = *it;
+                collection = coll;
 
                 if (auto readTimestamp =
                         opCtx->recoveryUnit()->getPointInTimeReadTimestamp(opCtx)) {
@@ -344,18 +344,18 @@ public:
                             !minSnapshot || *readTimestamp >= *minSnapshot);
                 }
             } else {
-                coll = catalog->establishConsistentCollection(
+                collection = catalog->establishConsistentCollection(
                     opCtx,
                     {dbName, uuid},
                     opCtx->recoveryUnit()->getPointInTimeReadTimestamp(opCtx));
 
-                if (!coll) {
+                if (!collection) {
                     // The collection did not exist at the read timestamp with the given UUID.
                     continue;
                 }
             }
 
-            (void)checkAndHashCollection(coll);
+            (void)checkAndHashCollection(collection);
         }
 
         BSONObjBuilder bb(result.subobjStart("collections"));
diff --git a/src/mongo/db/commands/list_collections.cpp b/src/mongo/db/commands/list_collections.cpp
index 4d192815e27..571c3156786 100644
--- a/src/mongo/db/commands/list_collections.cpp
+++ b/src/mongo/db/commands/list_collections.cpp
@@ -440,10 +440,8 @@ public:
                         // needing to yield as we don't take any locks.
                         if (opCtx->isLockFreeReadsOp()) {
                             auto collectionCatalog = CollectionCatalog::get(opCtx);
-                            for (auto it = collectionCatalog->begin(opCtx, dbName);
-                                 it != collectionCatalog->end(opCtx);
-                                 ++it) {
-                                perCollectionWork(*it);
+                            for (auto&& coll : collectionCatalog->range(dbName)) {
+                                perCollectionWork(coll);
                             }
                         } else {
                             mongo::catalog::forEachCollectionFromDb(
diff --git a/src/mongo/db/commands/validate_db_metadata_cmd.cpp b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
index d4347d72884..43ef8d739f6 100644
--- a/src/mongo/db/commands/validate_db_metadata_cmd.cpp
+++ b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
@@ -149,12 +149,10 @@ public:
                         return _validateView(opCtx, view);
                     });
 
-                for (auto collIt = collectionCatalog->begin(opCtx, dbName);
-                     collIt != collectionCatalog->end(opCtx);
-                     ++collIt) {
+                for (auto&& coll : collectionCatalog->range(dbName)) {
                     if (!_validateNamespace(
                             opCtx,
-                            collectionCatalog->lookupNSSByUUID(opCtx, collIt.uuid()).value())) {
+                            collectionCatalog->lookupNSSByUUID(opCtx, coll->uuid()).value())) {
                         return;
                     }
                 }
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 d84412e5284..ebf1557d215 100644
--- a/src/mongo/db/pipeline/document_source_change_stream_test.cpp
+++ b/src/mongo/db/pipeline/document_source_change_stream_test.cpp
@@ -496,8 +496,7 @@ TEST_F(ChangeStreamStageTest, ShouldRejectBothStartAtOperationTimeAndResumeAfter
         Lock::GlobalWrite lk(expCtx->opCtx);
         std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
         CollectionCatalog::write(expCtx->opCtx, [&](CollectionCatalog& catalog) {
-            catalog.registerCollection(
-                expCtx->opCtx, testUuid(), std::move(collection), /*ts=*/boost::none);
+            catalog.registerCollection(expCtx->opCtx, std::move(collection), /*ts=*/boost::none);
         });
     }
 
@@ -523,8 +522,7 @@ TEST_F(ChangeStreamStageTest, ShouldRejectBothStartAfterAndResumeAfterOptions) {
         Lock::GlobalWrite lk(opCtx);
         std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
         CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
-            catalog.registerCollection(
-                opCtx, testUuid(), std::move(collection), /*ts=*/boost::none);
+            catalog.registerCollection(opCtx, std::move(collection), /*ts=*/boost::none);
         });
     }
 
@@ -555,8 +553,7 @@ TEST_F(ChangeStreamStageTest, ShouldRejectBothStartAtOperationTimeAndStartAfterO
         Lock::GlobalWrite lk(opCtx);
         std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
         CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
-            catalog.registerCollection(
-                opCtx, testUuid(), std::move(collection), /*ts=*/boost::none);
+            catalog.registerCollection(opCtx, std::move(collection), /*ts=*/boost::none);
         });
     }
 
@@ -582,8 +579,7 @@ TEST_F(ChangeStreamStageTest, ShouldRejectResumeAfterWithResumeTokenMissingUUID)
         Lock::GlobalWrite lk(opCtx);
         std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
         CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
-            catalog.registerCollection(
-                opCtx, testUuid(), std::move(collection), /*ts=*/boost::none);
+            catalog.registerCollection(opCtx, std::move(collection), /*ts=*/boost::none);
         });
     }
 
@@ -2767,10 +2763,10 @@ TEST_F(ChangeStreamStageTest, DocumentKeyShouldNotIncludeShardKeyWhenNoO2FieldIn
 
     {
         Lock::GlobalWrite lk(getExpCtx()->opCtx);
-        std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
+        std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(uuid, nss);
         CollectionCatalog::write(getExpCtx()->opCtx, [&](CollectionCatalog& catalog) {
             catalog.registerCollection(
-                getExpCtx()->opCtx, uuid, std::move(collection), /*ts=*/boost::none);
+                getExpCtx()->opCtx, std::move(collection), /*ts=*/boost::none);
         });
     }
 
@@ -2814,10 +2810,10 @@ TEST_F(ChangeStreamStageTest, DocumentKeyShouldUseO2FieldInOplog) {
 
     {
         Lock::GlobalWrite lk(getExpCtx()->opCtx);
-        std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
+        std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(uuid, nss);
         CollectionCatalog::write(getExpCtx()->opCtx, [&](CollectionCatalog& catalog) {
             catalog.registerCollection(
-                getExpCtx()->opCtx, uuid, std::move(collection), /*ts=*/boost::none);
+                getExpCtx()->opCtx, std::move(collection), /*ts=*/boost::none);
         });
     }
 
@@ -2857,14 +2853,13 @@ TEST_F(ChangeStreamStageTest, DocumentKeyShouldUseO2FieldInOplog) {
 
 TEST_F(ChangeStreamStageTest, ResumeAfterFailsIfResumeTokenDoesNotContainUUID) {
     const Timestamp ts(3, 45);
-    const auto uuid = testUuid();
 
     {
         Lock::GlobalWrite lk(getExpCtx()->opCtx);
         std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
         CollectionCatalog::write(getExpCtx()->opCtx, [&](CollectionCatalog& catalog) {
             catalog.registerCollection(
-                getExpCtx()->opCtx, uuid, std::move(collection), /*ts=*/boost::none);
+                getExpCtx()->opCtx, std::move(collection), /*ts=*/boost::none);
         });
     }
 
@@ -2931,7 +2926,7 @@ TEST_F(ChangeStreamStageTest, ResumeAfterWithTokenFromInvalidateShouldFail) {
         std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
         CollectionCatalog::write(expCtx->opCtx, [&](CollectionCatalog& catalog) {
             catalog.registerCollection(
-                getExpCtx()->opCtx, testUuid(), std::move(collection), /*ts=*/boost::none);
+                getExpCtx()->opCtx, std::move(collection), /*ts=*/boost::none);
         });
     }
 
@@ -3423,10 +3418,10 @@ TEST_F(ChangeStreamStageDBTest, DocumentKeyShouldNotIncludeShardKeyWhenNoO2Field
 
     {
         Lock::GlobalWrite lk(getExpCtx()->opCtx);
-        std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
+        std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(uuid, nss);
         CollectionCatalog::write(getExpCtx()->opCtx, [&](CollectionCatalog& catalog) {
             catalog.registerCollection(
-                getExpCtx()->opCtx, uuid, std::move(collection), /*ts=*/boost::none);
+                getExpCtx()->opCtx, std::move(collection), /*ts=*/boost::none);
         });
     }
 
@@ -3465,10 +3460,10 @@ TEST_F(ChangeStreamStageDBTest, DocumentKeyShouldUseO2FieldInOplog) {
 
     {
         Lock::GlobalWrite lk(getExpCtx()->opCtx);
-        std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
+        std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(uuid, nss);
         CollectionCatalog::write(getExpCtx()->opCtx, [&](CollectionCatalog& catalog) {
             catalog.registerCollection(
-                getExpCtx()->opCtx, uuid, std::move(collection), /*ts=*/boost::none);
+                getExpCtx()->opCtx, std::move(collection), /*ts=*/boost::none);
         });
     }
 
@@ -3510,7 +3505,7 @@ TEST_F(ChangeStreamStageDBTest, ResumeAfterWithTokenFromInvalidateShouldFail) {
         std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
         CollectionCatalog::write(expCtx->opCtx, [&](CollectionCatalog& catalog) {
             catalog.registerCollection(
-                getExpCtx()->opCtx, testUuid(), std::move(collection), /*ts=*/boost::none);
+                getExpCtx()->opCtx, std::move(collection), /*ts=*/boost::none);
         });
     }
 
@@ -3535,10 +3530,10 @@ TEST_F(ChangeStreamStageDBTest, ResumeAfterWithTokenFromDropDatabase) {
 
     {
         Lock::GlobalWrite lk(getExpCtx()->opCtx);
-        std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
+        std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(uuid, nss);
         CollectionCatalog::write(getExpCtx()->opCtx, [&](CollectionCatalog& catalog) {
             catalog.registerCollection(
-                getExpCtx()->opCtx, uuid, std::move(collection), /*ts=*/boost::none);
+                getExpCtx()->opCtx, std::move(collection), /*ts=*/boost::none);
         });
     }
 
@@ -3573,10 +3568,10 @@ TEST_F(ChangeStreamStageDBTest, StartAfterSucceedsEvenIfResumeTokenDoesNotContai
 
     {
         Lock::GlobalWrite lk(getExpCtx()->opCtx);
-        std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
+        std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(uuid, nss);
         CollectionCatalog::write(getExpCtx()->opCtx, [&](CollectionCatalog& catalog) {
             catalog.registerCollection(
-                getExpCtx()->opCtx, uuid, std::move(collection), /*ts=*/boost::none);
+                getExpCtx()->opCtx, std::move(collection), /*ts=*/boost::none);
         });
     }
 
diff --git a/src/mongo/db/repl/shard_merge_recipient_op_observer.cpp b/src/mongo/db/repl/shard_merge_recipient_op_observer.cpp
index 754b1a50c52..1ad5080a42c 100644
--- a/src/mongo/db/repl/shard_merge_recipient_op_observer.cpp
+++ b/src/mongo/db/repl/shard_merge_recipient_op_observer.cpp
@@ -97,9 +97,7 @@ void deleteTenantDataWhenMergeAborts(OperationContext* opCtx,
             IndexBuildsCoordinator::get(opCtx)->assertNoBgOpInProgForDb(db->name());
 
             auto catalog = CollectionCatalog::get(opCtx);
-            for (auto collIt = catalog->begin(opCtx, db->name()); collIt != catalog->end(opCtx);
-                 ++collIt) {
-                auto collection = *collIt;
+            for (auto&& collection : catalog->range(db->name())) {
                 if (!collection) {
                     break;
                 }
diff --git a/src/mongo/db/s/config/sharding_catalog_manager_shard_collection_test.cpp b/src/mongo/db/s/config/sharding_catalog_manager_shard_collection_test.cpp
index 5424ecebd46..e3ec02257ed 100644
--- a/src/mongo/db/s/config/sharding_catalog_manager_shard_collection_test.cpp
+++ b/src/mongo/db/s/config/sharding_catalog_manager_shard_collection_test.cpp
@@ -91,8 +91,7 @@ TEST_F(CreateFirstChunksTest, NonEmptyCollection_NoZones_OneChunkToPrimary) {
         Lock::GlobalWrite lk(operationContext());
         CollectionCatalog::write(getServiceContext(), [&](CollectionCatalog& catalog) {
             catalog.registerCollection(operationContext(),
-                                       uuid,
-                                       std::make_shared<CollectionMock>(kNamespace),
+                                       std::make_shared<CollectionMock>(uuid, kNamespace),
                                        /*ts=*/boost::none);
         });
     }
diff --git a/src/mongo/db/s/move_primary/move_primary_database_cloner.cpp b/src/mongo/db/s/move_primary/move_primary_database_cloner.cpp
index 624d7c4c775..9f69616a90a 100644
--- a/src/mongo/db/s/move_primary/move_primary_database_cloner.cpp
+++ b/src/mongo/db/s/move_primary/move_primary_database_cloner.cpp
@@ -151,10 +151,8 @@ void MovePrimaryDatabaseCloner::calculateListCatalogEntriesForRecipient() {
 
     if (opCtx->isLockFreeReadsOp()) {
         auto collectionCatalog = CollectionCatalog::get(opCtx);
-        for (auto it = collectionCatalog->begin(opCtx, _dbName);
-             it != collectionCatalog->end(opCtx);
-             ++it) {
-            parseCollection(*it);
+        for (auto&& coll : collectionCatalog->range(_dbName)) {
+            parseCollection(coll);
         }
     } else {
         mongo::catalog::forEachCollectionFromDb(opCtx, _dbName, MODE_IS, parseCollection);
diff --git a/src/mongo/db/s/shardsvr_check_metadata_consistency_participant_command.cpp b/src/mongo/db/s/shardsvr_check_metadata_consistency_participant_command.cpp
index 0d57f8494ed..72a61486b5a 100644
--- a/src/mongo/db/s/shardsvr_check_metadata_consistency_participant_command.cpp
+++ b/src/mongo/db/s/shardsvr_check_metadata_consistency_participant_command.cpp
@@ -266,10 +266,7 @@ public:
                 std::vector<CollectionPtr> localCollections;
                 switch (metadata_consistency_util::getCommandLevel(nss)) {
                     case MetadataConsistencyCommandLevelEnum::kDatabaseLevel: {
-                        for (auto it = collCatalogSnapshot->begin(opCtx, nss.dbName());
-                             it != collCatalogSnapshot->end(opCtx);
-                             ++it) {
-                            const auto coll = *it;
+                        for (auto&& coll : collCatalogSnapshot->range(nss.dbName())) {
                             if (!coll) {
                                 continue;
                             }
diff --git a/src/mongo/db/startup_recovery.cpp b/src/mongo/db/startup_recovery.cpp
index ea6d937e2c0..77c57e7e1a6 100644
--- a/src/mongo/db/startup_recovery.cpp
+++ b/src/mongo/db/startup_recovery.cpp
@@ -212,8 +212,7 @@ Status ensureCollectionProperties(OperationContext* opCtx,
                                   const DatabaseName& dbName,
                                   EnsureIndexPolicy ensureIndexPolicy) {
     auto catalog = CollectionCatalog::get(opCtx);
-    for (auto collIt = catalog->begin(opCtx, dbName); collIt != catalog->end(opCtx); ++collIt) {
-        auto coll = *collIt;
+    for (auto&& coll : catalog->range(dbName)) {
         if (!coll) {
             break;
         }
@@ -233,7 +232,7 @@ Status ensureCollectionProperties(OperationContext* opCtx,
                   logAttrs(*coll));
             if (EnsureIndexPolicy::kBuildMissing == ensureIndexPolicy) {
                 auto writableCollection =
-                    catalog->lookupCollectionByUUIDForMetadataWrite(opCtx, collIt.uuid());
+                    catalog->lookupCollectionByUUIDForMetadataWrite(opCtx, coll->uuid());
                 auto status = buildMissingIdIndex(opCtx, writableCollection);
                 if (!status.isOK()) {
                     LOGV2_ERROR(21021,
diff --git a/src/mongo/db/storage/kv/durable_catalog_test.cpp b/src/mongo/db/storage/kv/durable_catalog_test.cpp
index 5a29311d40b..bd49f858a33 100644
--- a/src/mongo/db/storage/kv/durable_catalog_test.cpp
+++ b/src/mongo/db/storage/kv/durable_catalog_test.cpp
@@ -119,7 +119,6 @@ public:
 
         CollectionCatalog::write(operationContext(), [&](CollectionCatalog& catalog) {
             catalog.registerCollection(operationContext(),
-                                       options.uuid.value(),
                                        std::move(collection),
                                        /*ts=*/boost::none);
         });
diff --git a/src/mongo/db/storage/storage_engine_impl.cpp b/src/mongo/db/storage/storage_engine_impl.cpp
index b6f3ad8533e..7b11a3f6232 100644
--- a/src/mongo/db/storage/storage_engine_impl.cpp
+++ b/src/mongo/db/storage/storage_engine_impl.cpp
@@ -428,8 +428,7 @@ void StorageEngineImpl::_initCollection(OperationContext* opCtx,
     auto collection = collectionFactory->make(opCtx, nss, catalogId, md, std::move(rs));
 
     CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
-        catalog.registerCollection(
-            opCtx, md->options.uuid.value(), std::move(collection), /*commitTime*/ minValidTs);
+        catalog.registerCollection(opCtx, std::move(collection), /*commitTime*/ minValidTs);
     });
 }
 
@@ -1399,9 +1398,8 @@ int64_t StorageEngineImpl::sizeOnDiskForDb(OperationContext* opCtx, const Databa
 
     if (opCtx->isLockFreeReadsOp()) {
         auto collectionCatalog = CollectionCatalog::get(opCtx);
-        for (auto it = collectionCatalog->begin(opCtx, dbName); it != collectionCatalog->end(opCtx);
-             ++it) {
-            perCollectionWork(*it);
+        for (auto&& coll : collectionCatalog->range(dbName)) {
+            perCollectionWork(coll);
         }
     } else {
         catalog::forEachCollectionFromDb(opCtx, dbName, MODE_IS, perCollectionWork);
diff --git a/src/mongo/db/storage/storage_engine_test_fixture.h b/src/mongo/db/storage/storage_engine_test_fixture.h
index d0a79882e9f..6e9c3ef8cce 100644
--- a/src/mongo/db/storage/storage_engine_test_fixture.h
+++ b/src/mongo/db/storage/storage_engine_test_fixture.h
@@ -82,8 +82,7 @@ public:
             std::move(rs));
 
         CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
-            catalog.registerCollection(
-                opCtx, options.uuid.get(), std::move(coll), /*ts=*/boost::none);
+            catalog.registerCollection(opCtx, std::move(coll), /*ts=*/boost::none);
         });
 
         return {{_storageEngine->getCatalog()->getEntry(catalogId)}};
diff --git a/src/mongo/db/storage/storage_util.cpp b/src/mongo/db/storage/storage_util.cpp
index 7339dde91f2..e7634b858c9 100644
--- a/src/mongo/db/storage/storage_util.cpp
+++ b/src/mongo/db/storage/storage_util.cpp
@@ -59,8 +59,7 @@ auto removeEmptyDirectory =
         auto collectionCatalog = CollectionCatalog::latest(svcCtx);
         const DatabaseName& dbName = ns.dbName();
         if (!storageEngine->isUsingDirectoryPerDb() ||
-            (storageEngine->supportsPendingDrops() &&
-             collectionCatalog->begin(nullptr, dbName) != collectionCatalog->end(nullptr))) {
+            (storageEngine->supportsPendingDrops() && !collectionCatalog->range(dbName).empty())) {
             return;
         }
 
@@ -69,7 +68,7 @@ auto removeEmptyDirectory =
 
         if (!ec) {
             LOGV2(4888200, "Removed empty database directory", logAttrs(dbName));
-        } else if (collectionCatalog->begin(nullptr, dbName) == collectionCatalog->end(nullptr)) {
+        } else if (collectionCatalog->range(dbName).empty()) {
             // It is possible for a new collection to be created in the database between when we
             // check whether the database is empty and actually attempting to remove the directory.
             // In this case, don't log that the removal failed because it is expected. However,