SERVER-71222 CollectionCatalog supports point-in-time catalog lookups by UUID

9 files changed, 781 insertions, 563 deletions

diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index 1a9dec3dccd..383196fed4c 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -285,18 +285,19 @@ public:
                     // we must update the minVisibleTimestamp with the appropriate value. This is
                     // fine because the collection should not be visible in the catalog until we
                     // call setCommitted(true).
-                    writeJobs.push_back([coll = entry.collection.get(),
-                                         commitTime](CollectionCatalog& catalog) {
-                        if (commitTime) {
-                            coll->setMinimumVisibleSnapshot(commitTime.value());
-                            coll->setMinimumValidSnapshot(commitTime.value());
-                        }
-                        catalog._pushCatalogIdForNSS(coll->ns(), coll->getCatalogId(), commitTime);
-
-                        catalog._pendingCommitNamespaces.erase(coll->ns());
-                        catalog._pendingCommitUUIDs.erase(coll->uuid());
-                        coll->setCommitted(true);
-                    });
+                    writeJobs.push_back(
+                        [coll = entry.collection.get(), commitTime](CollectionCatalog& catalog) {
+                            if (commitTime) {
+                                coll->setMinimumVisibleSnapshot(commitTime.value());
+                                coll->setMinimumValidSnapshot(commitTime.value());
+                            }
+                            catalog._pushCatalogIdForNSSAndUUID(
+                                coll->ns(), coll->uuid(), coll->getCatalogId(), commitTime);
+
+                            catalog._pendingCommitNamespaces.erase(coll->ns());
+                            catalog._pendingCommitUUIDs.erase(coll->uuid());
+                            coll->setCommitted(true);
+                        });
                     break;
                 }
                 case UncommittedCatalogUpdates::Entry::Action::kReplacedViewsForDatabase: {
@@ -834,14 +835,7 @@ const Collection* CollectionCatalog::_openCollection(
         return _openCollectionAtLatestByUUID(opCtx, *nssOrUUID.uuid());
     }
 
-    // TODO SERVER-71222: Handle lookup at point-in-time by UUID.
-    const auto& nss = nssOrUUID.nss();
-    if (nss) {
-        return _openCollectionAtPointInTimeByNamespace(opCtx, *nss, *readTimestamp);
-    } else {
-        return _openCollectionAtPointInTimeByNamespace(
-            opCtx, resolveNamespaceStringOrUUID(opCtx, nssOrUUID), *readTimestamp);
-    }
+    return _openCollectionAtPointInTimeByNamespaceOrUUID(opCtx, nssOrUUID, *readTimestamp);
 }
 
 const Collection* CollectionCatalog::_openCollectionAtLatestByNamespace(
@@ -860,7 +854,7 @@ const Collection* CollectionCatalog::_openCollectionAtLatestByNamespace(
         catalogId = pendingCollection->getCatalogId();
     } else {
         auto lookupResult = lookupCatalogIdByNSS(nss, boost::none);
-        invariant(lookupResult.result == CatalogIdLookup::NamespaceExistence::kExists);
+        invariant(lookupResult.result == CatalogIdLookup::Existence::kExists);
         catalogId = lookupResult.id;
     }
 
@@ -1006,14 +1000,16 @@ const Collection* CollectionCatalog::_openCollectionAtLatestByUUID(OperationCont
     openedCollections.store(latestCollection, nss, uuid);
     return latestCollection.get();
 }
-const Collection* CollectionCatalog::_openCollectionAtPointInTimeByNamespace(
-    OperationContext* opCtx, const NamespaceString& nss, Timestamp readTimestamp) const {
+const Collection* CollectionCatalog::_openCollectionAtPointInTimeByNamespaceOrUUID(
+    OperationContext* opCtx,
+    const NamespaceStringOrUUID& nssOrUUID,
+    Timestamp readTimestamp) const {
     auto& openedCollections = OpenedCollections::get(opCtx);
 
     // Try to find a catalog entry matching 'readTimestamp'.
-    auto catalogEntry = _fetchPITCatalogEntry(opCtx, nss, readTimestamp);
+    auto catalogEntry = _fetchPITCatalogEntry(opCtx, nssOrUUID, readTimestamp);
     if (!catalogEntry) {
-        openedCollections.store(nullptr, nss, boost::none);
+        openedCollections.store(nullptr, nssOrUUID.nss(), nssOrUUID.uuid());
         return nullptr;
     }
 
@@ -1021,7 +1017,7 @@ const Collection* CollectionCatalog::_openCollectionAtPointInTimeByNamespace(
 
     // Return the in-memory Collection instance if it is compatible with the read timestamp.
     if (isExistingCollectionCompatible(latestCollection, readTimestamp)) {
-        openedCollections.store(latestCollection, nss, latestCollection->uuid());
+        openedCollections.store(latestCollection, latestCollection->ns(), latestCollection->uuid());
         return latestCollection.get();
     }
 
@@ -1030,7 +1026,8 @@ const Collection* CollectionCatalog::_openCollectionAtPointInTimeByNamespace(
     auto compatibleCollection =
         _createCompatibleCollection(opCtx, latestCollection, readTimestamp, catalogEntry.get());
     if (compatibleCollection) {
-        openedCollections.store(compatibleCollection, nss, compatibleCollection->uuid());
+        openedCollections.store(
+            compatibleCollection, compatibleCollection->ns(), compatibleCollection->uuid());
         return compatibleCollection.get();
     }
 
@@ -1038,50 +1035,99 @@ const Collection* CollectionCatalog::_openCollectionAtPointInTimeByNamespace(
     // Collection instance from scratch.
     auto newCollection = _createNewPITCollection(opCtx, readTimestamp, catalogEntry.get());
     if (newCollection) {
-        openedCollections.store(newCollection, nss, newCollection->uuid());
+        openedCollections.store(newCollection, newCollection->ns(), newCollection->uuid());
         return newCollection.get();
     }
 
-    openedCollections.store(nullptr, nss, boost::none);
+    openedCollections.store(nullptr, nssOrUUID.nss(), nssOrUUID.uuid());
     return nullptr;
 }
 
+CollectionCatalog::CatalogIdLookup CollectionCatalog::_checkWithOldestCatalogIdTimestampMaintained(
+    boost::optional<Timestamp> ts) const {
+    // If the request was with a time prior to the oldest maintained time it is unknown, otherwise
+    // we know it is not existing.
+    return {RecordId{},
+            ts && *ts < _oldestCatalogIdTimestampMaintained
+                ? CollectionCatalog::CatalogIdLookup::Existence::kUnknown
+                : CollectionCatalog::CatalogIdLookup::Existence::kNotExists};
+}
+
+CollectionCatalog::CatalogIdLookup CollectionCatalog::_findCatalogIdInRange(
+    boost::optional<Timestamp> ts, const std::vector<TimestampedCatalogId>& range) const {
+    if (!ts) {
+        auto catalogId = range.back().id;
+        if (catalogId) {
+            return {*catalogId, CatalogIdLookup::Existence::kExists};
+        }
+        return {RecordId{}, CatalogIdLookup::Existence::kNotExists};
+    }
+
+    auto rangeIt =
+        std::upper_bound(range.begin(), range.end(), *ts, [](const auto& ts, const auto& entry) {
+            return ts < entry.ts;
+        });
+    if (rangeIt == range.begin()) {
+        return _checkWithOldestCatalogIdTimestampMaintained(ts);
+    }
+    // Upper bound returns an iterator to the first entry with a larger timestamp. Decrement the
+    // iterator to get the last entry where the time is less or equal.
+    auto catalogId = (--rangeIt)->id;
+    if (catalogId) {
+        if (*catalogId != kUnknownRangeMarkerId) {
+            return {*catalogId, CatalogIdLookup::Existence::kExists};
+        } else {
+            return {RecordId{}, CatalogIdLookup::Existence::kUnknown};
+        }
+    }
+    return {RecordId{}, CatalogIdLookup::Existence::kNotExists};
+}
+
 boost::optional<DurableCatalogEntry> CollectionCatalog::_fetchPITCatalogEntry(
     OperationContext* opCtx,
-    const NamespaceString& nss,
+    const NamespaceStringOrUUID& nssOrUUID,
     boost::optional<Timestamp> readTimestamp) const {
-    auto [catalogId, result] = lookupCatalogIdByNSS(nss, readTimestamp);
-    if (result == CatalogIdLookup::NamespaceExistence::kNotExists) {
+    auto [catalogId, result] = nssOrUUID.nss()
+        ? lookupCatalogIdByNSS(*nssOrUUID.nss(), readTimestamp)
+        : lookupCatalogIdByUUID(*nssOrUUID.uuid(), readTimestamp);
+    if (result == CatalogIdLookup::Existence::kNotExists) {
         return boost::none;
     }
 
     auto writeCatalogIdAfterScan = [&](const boost::optional<DurableCatalogEntry>& catalogEntry) {
         CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
-            catalog._insertCatalogIdForNSSAfterScan(
-                nss,
+            // Insert catalogId for both the namespace and UUID if the catalog entry is found.
+            catalog._insertCatalogIdForNSSAndUUIDAfterScan(
+                catalogEntry ? catalogEntry->metadata->nss : nssOrUUID.nss(),
+                catalogEntry ? catalogEntry->metadata->options.uuid : nssOrUUID.uuid(),
                 catalogEntry ? boost::make_optional(catalogEntry->catalogId) : boost::none,
                 *readTimestamp);
         });
     };
 
-    if (result == CatalogIdLookup::NamespaceExistence::kUnknown) {
+    if (result == CatalogIdLookup::Existence::kUnknown) {
         // We shouldn't receive kUnknown when we don't have a timestamp since no timestamp means
         // we're operating on the latest.
         invariant(readTimestamp);
 
         // Scan durable catalog when we don't have accurate catalogId mapping for this timestamp.
         gCollectionCatalogSection.numScansDueToMissingMapping.fetchAndAdd(1);
-        auto catalogEntry = DurableCatalog::get(opCtx)->scanForCatalogEntryByNss(opCtx, nss);
+        auto catalogEntry = nssOrUUID.nss()
+            ? DurableCatalog::get(opCtx)->scanForCatalogEntryByNss(opCtx, *nssOrUUID.nss())
+            : DurableCatalog::get(opCtx)->scanForCatalogEntryByUUID(opCtx, *nssOrUUID.uuid());
         writeCatalogIdAfterScan(catalogEntry);
         return catalogEntry;
     }
 
     auto catalogEntry = DurableCatalog::get(opCtx)->getParsedCatalogEntry(opCtx, catalogId);
-    if (!catalogEntry || nss != catalogEntry->metadata->nss) {
+    if (const auto& nss = nssOrUUID.nss();
+        !catalogEntry || (nss && nss != catalogEntry->metadata->nss)) {
         invariant(readTimestamp);
         // If no entry is found or the entry contains a different namespace, the mapping might be
         // incorrect since it is incomplete after startup; scans durable catalog to confirm.
-        auto catalogEntry = DurableCatalog::get(opCtx)->scanForCatalogEntryByNss(opCtx, nss);
+        auto catalogEntry = nss
+            ? DurableCatalog::get(opCtx)->scanForCatalogEntryByNss(opCtx, *nss)
+            : DurableCatalog::get(opCtx)->scanForCatalogEntryByUUID(opCtx, *nssOrUUID.uuid());
         writeCatalogIdAfterScan(catalogEntry);
         return catalogEntry;
     }
@@ -1593,45 +1639,18 @@ bool CollectionCatalog::containsCollection(OperationContext* opCtx,
 
 CollectionCatalog::CatalogIdLookup CollectionCatalog::lookupCatalogIdByNSS(
     const NamespaceString& nss, boost::optional<Timestamp> ts) const {
-    if (auto it = _catalogIds.find(nss); it != _catalogIds.end()) {
-        const auto& range = it->second;
-        if (!ts) {
-            auto catalogId = range.back().id;
-            if (catalogId) {
-                return {*catalogId, CatalogIdLookup::NamespaceExistence::kExists};
-            }
-
-            return {RecordId{}, CatalogIdLookup::NamespaceExistence::kNotExists};
-        }
+    if (auto it = _nssCatalogIds.find(nss); it != _nssCatalogIds.end()) {
+        return _findCatalogIdInRange(ts, it->second);
+    }
+    return _checkWithOldestCatalogIdTimestampMaintained(ts);
+}
 
-        auto rangeIt = std::upper_bound(
-            range.begin(), range.end(), *ts, [](const auto& ts, const auto& entry) {
-                return ts < entry.ts;
-            });
-        if (rangeIt == range.begin()) {
-            return {RecordId{},
-                    *ts < _oldestCatalogIdTimestampMaintained
-                        ? CatalogIdLookup::NamespaceExistence::kUnknown
-                        : CatalogIdLookup::NamespaceExistence::kNotExists};
-        }
-        // Upper bound returns an iterator to the first entry with a larger timestamp. Decrement the
-        // iterator to get the last entry where the time is less or equal.
-        auto catalogId = (--rangeIt)->id;
-        if (catalogId) {
-            if (*catalogId != kUnknownRangeMarkerId) {
-                return {*catalogId, CatalogIdLookup::NamespaceExistence::kExists};
-            } else {
-                return {RecordId{}, CatalogIdLookup::NamespaceExistence::kUnknown};
-            }
-        }
-        return {RecordId{}, CatalogIdLookup::NamespaceExistence::kNotExists};
+CollectionCatalog::CatalogIdLookup CollectionCatalog::lookupCatalogIdByUUID(
+    const UUID& uuid, boost::optional<Timestamp> ts) const {
+    if (auto it = _uuidCatalogIds.find(uuid); it != _uuidCatalogIds.end()) {
+        return _findCatalogIdInRange(ts, it->second);
     }
-    // If the namespace was requested with a time prior to the oldest maintained time it is unknown,
-    // otherwise we know it is not existing
-    auto existence = ts && *ts < _oldestCatalogIdTimestampMaintained
-        ? CatalogIdLookup::NamespaceExistence::kUnknown
-        : CatalogIdLookup::NamespaceExistence::kNotExists;
-    return {RecordId{}, existence};
+    return _checkWithOldestCatalogIdTimestampMaintained(ts);
 }
 
 void CollectionCatalog::iterateViews(
@@ -1902,7 +1921,7 @@ void CollectionCatalog::_registerCollection(OperationContext* opCtx,
 
     if (commitTime && !commitTime->isNull()) {
         coll->setMinimumValidSnapshot(commitTime.value());
-        _pushCatalogIdForNSS(nss, coll->getCatalogId(), commitTime);
+        _pushCatalogIdForNSSAndUUID(nss, uuid, coll->getCatalogId(), commitTime);
     }
 
 
@@ -1977,7 +1996,7 @@ std::shared_ptr<Collection> CollectionCatalog::deregisterCollection(
 
     // Push drop unless this is a rollback of a create
     if (coll->isCommitted()) {
-        _pushCatalogIdForNSS(ns, boost::none, commitTime);
+        _pushCatalogIdForNSSAndUUID(ns, uuid, boost::none, commitTime);
     }
 
     if (!ns.isOnInternalDb() && !ns.isSystem()) {
@@ -2056,57 +2075,67 @@ void CollectionCatalog::_ensureNamespaceDoesNotExist(OperationContext* opCtx,
     }
 }
 
-void CollectionCatalog::_pushCatalogIdForNSS(const NamespaceString& nss,
-                                             boost::optional<RecordId> catalogId,
-                                             boost::optional<Timestamp> ts) {
+void CollectionCatalog::_pushCatalogIdForNSSAndUUID(const NamespaceString& nss,
+                                                    const UUID& uuid,
+                                                    boost::optional<RecordId> catalogId,
+                                                    boost::optional<Timestamp> ts) {
     // TODO SERVER-68674: Remove feature flag check.
     if (!feature_flags::gPointInTimeCatalogLookups.isEnabledAndIgnoreFCV()) {
         // No-op.
         return;
     }
 
-    auto& ids = _catalogIds[nss];
+    auto& nssIds = _nssCatalogIds[nss];
+    auto& uuidIds = _uuidCatalogIds[uuid];
 
-    if (!ts) {
-        // Make sure untimestamped writes have a single entry in mapping. If we're mixing
-        // timestamped with untimestamped (such as repair). Ignore the untimestamped writes as an
-        // untimestamped deregister will correspond with an untimestamped register. We should leave
-        // the mapping as-is in this case.
-        if (ids.empty() && catalogId) {
-            // This namespace was added due to an untimestamped write, add an entry with min
-            // timestamp
-            ids.push_back(TimestampedCatalogId{catalogId, Timestamp::min()});
-        } else if (ids.size() == 1 && !catalogId) {
-            // This namespace was removed due to an untimestamped write, clear entries.
-            ids.clear();
-        }
+    auto doPushCatalogId = [this, &ts, &catalogId](std::vector<TimestampedCatalogId>& ids,
+                                                   auto& catalogIdsContainer,
+                                                   auto& catalogIdChangesContainer,
+                                                   const auto& key) {
+        if (!ts) {
+            // Make sure untimestamped writes have a single entry in mapping. If we're mixing
+            // timestamped with untimestamped (such as repair). Ignore the untimestamped writes as
+            // an untimestamped deregister will correspond with an untimestamped register. We should
+            // leave the mapping as-is in this case.
+            if (ids.empty() && catalogId) {
+                // This namespace or UUID was added due to an untimestamped write, add an entry
+                // with min timestamp
+                ids.push_back(TimestampedCatalogId{catalogId, Timestamp::min()});
+            } else if (ids.size() == 1 && !catalogId) {
+                // This namespace or UUID was removed due to an untimestamped write, clear entries.
+                ids.clear();
+            }
 
-        // Make sure we erase mapping for namespace if the list is left empty as lookups expect at
-        // least one entry for existing namespaces.
-        if (ids.empty()) {
-            _catalogIds.erase(nss);
+            // Make sure we erase mapping for namespace or UUID if the list is left empty as
+            // lookups expect at least one entry for existing namespaces or UUIDs.
+            if (ids.empty()) {
+                catalogIdsContainer.erase(key);
+            }
+
+            return;
         }
 
-        return;
-    }
+        // An entry could exist already if concurrent writes are performed, keep the latest change
+        // in that case.
+        if (!ids.empty() && ids.back().ts == *ts) {
+            ids.back().id = catalogId;
+            return;
+        }
 
-    // An entry could exist already if concurrent writes are performed, keep the latest change in
-    // that case.
-    if (!ids.empty() && ids.back().ts == *ts) {
-        ids.back().id = catalogId;
-        return;
-    }
+        // Otherwise, push new entry at the end. Timestamp is always increasing
+        invariant(ids.empty() || ids.back().ts < *ts);
+        // If the catalogId is the same as last entry, there's nothing we need to do. This can
+        // happen when the catalog is reopened.
+        if (!ids.empty() && ids.back().id == catalogId) {
+            return;
+        }
 
-    // Otherwise, push new entry at the end. Timestamp is always increasing
-    invariant(ids.empty() || ids.back().ts < *ts);
-    // If the catalogId is the same as last entry, there's nothing we need to do. This can happen
-    // when the catalog is reopened.
-    if (!ids.empty() && ids.back().id == catalogId) {
-        return;
-    }
+        ids.push_back(TimestampedCatalogId{catalogId, *ts});
+        _markForCatalogIdCleanupIfNeeded(key, catalogIdChangesContainer, ids);
+    };
 
-    ids.push_back(TimestampedCatalogId{catalogId, *ts});
-    _markNamespaceForCatalogIdCleanupIfNeeded(nss, ids);
+    doPushCatalogId(nssIds, _nssCatalogIds, _nssCatalogIdChanges, nss);
+    doPushCatalogId(uuidIds, _uuidCatalogIds, _uuidCatalogIdChanges, uuid);
 }
 
 void CollectionCatalog::_pushCatalogIdForRename(const NamespaceString& from,
@@ -2120,20 +2149,20 @@ void CollectionCatalog::_pushCatalogIdForRename(const NamespaceString& from,
 
     // Get 'toIds' first, it may need to instantiate in the container which invalidates all
     // references.
-    auto& toIds = _catalogIds[to];
-    auto& fromIds = _catalogIds.at(from);
+    auto& toIds = _nssCatalogIds[to];
+    auto& fromIds = _nssCatalogIds.at(from);
     invariant(!fromIds.empty());
 
     // Make sure untimestamped writes have a single entry in mapping. We move the single entry from
     // 'from' to 'to'. We do not have to worry about mixing timestamped and untimestamped like
-    // _pushCatalogIdForNSS.
+    // _pushCatalogId.
     if (!ts) {
         // We should never perform rename in a mixed-mode environment. 'from' should contain a
         // single entry and there should be nothing in 'to' .
         invariant(fromIds.size() == 1);
         invariant(toIds.empty());
         toIds.push_back(TimestampedCatalogId{fromIds.back().id, Timestamp::min()});
-        _catalogIds.erase(from);
+        _nssCatalogIds.erase(from);
         return;
     }
 
@@ -2144,7 +2173,7 @@ void CollectionCatalog::_pushCatalogIdForRename(const NamespaceString& from,
     } else {
         invariant(toIds.empty() || toIds.back().ts < *ts);
         toIds.push_back(TimestampedCatalogId{fromIds.back().id, *ts});
-        _markNamespaceForCatalogIdCleanupIfNeeded(to, toIds);
+        _markForCatalogIdCleanupIfNeeded(to, _nssCatalogIdChanges, toIds);
     }
 
     // Re-write latest entry if timestamp match (multiple changes occured in this transaction),
@@ -2154,92 +2183,108 @@ void CollectionCatalog::_pushCatalogIdForRename(const NamespaceString& from,
     } else {
         invariant(fromIds.empty() || fromIds.back().ts < *ts);
         fromIds.push_back(TimestampedCatalogId{boost::none, *ts});
-        _markNamespaceForCatalogIdCleanupIfNeeded(from, fromIds);
+        _markForCatalogIdCleanupIfNeeded(from, _nssCatalogIdChanges, fromIds);
     }
 }
 
-void CollectionCatalog::_insertCatalogIdForNSSAfterScan(const NamespaceString& nss,
-                                                        boost::optional<RecordId> catalogId,
-                                                        Timestamp ts) {
+void CollectionCatalog::_insertCatalogIdForNSSAndUUIDAfterScan(boost::optional<NamespaceString> nss,
+                                                               boost::optional<UUID> uuid,
+                                                               boost::optional<RecordId> catalogId,
+                                                               Timestamp ts) {
     // TODO SERVER-68674: Remove feature flag check.
     if (!feature_flags::gPointInTimeCatalogLookups.isEnabledAndIgnoreFCV()) {
         // No-op.
         return;
     }
 
-    auto& ids = _catalogIds[nss];
+    auto doInsert = [this, &catalogId, &ts](std::vector<TimestampedCatalogId>& ids,
+                                            auto& catalogIdChangesContainer,
+                                            const auto& key) {
+        // Binary search for to the entry with same or larger timestamp
+        auto it = std::lower_bound(
+            ids.begin(), ids.end(), ts, [](const auto& entry, const Timestamp& ts) {
+                return entry.ts < ts;
+            });
 
-    // Binary search for to the entry with same or larger timestamp
-    auto it =
-        std::lower_bound(ids.begin(), ids.end(), ts, [](const auto& entry, const Timestamp& ts) {
-            return entry.ts < ts;
-        });
+        // The logic of what we need to do differs whether we are inserting a valid catalogId or
+        // not.
+        if (catalogId) {
+            if (it != ids.end()) {
+                // An entry could exist already if concurrent writes are performed, keep the latest
+                // change in that case.
+                if (it->ts == ts) {
+                    it->id = catalogId;
+                    return;
+                }
 
-    // The logic of what we need to do differs whether we are inserting a valid catalogId or not.
-    if (catalogId) {
-        if (it != ids.end()) {
-            // An entry could exist already if concurrent writes are performed, keep the latest
-            // change in that case.
-            if (it->ts == ts) {
-                it->id = catalogId;
-                return;
+                // If next element has same catalogId, we can adjust its timestamp to cover a longer
+                // range
+                if (it->id == catalogId) {
+                    it->ts = ts;
+                    _markForCatalogIdCleanupIfNeeded(key, catalogIdChangesContainer, ids);
+                    return;
+                }
             }
 
-            // If next element has same catalogId, we can adjust its timestamp to cover a longer
-            // range
-            if (it->id == catalogId) {
-                it->ts = ts;
-                _markNamespaceForCatalogIdCleanupIfNeeded(nss, ids);
-                return;
-            }
+            // Otherwise insert new entry at timestamp
+            ids.insert(it, TimestampedCatalogId{catalogId, ts});
+            _markForCatalogIdCleanupIfNeeded(key, catalogIdChangesContainer, ids);
+            return;
         }
 
-        // Otherwise insert new entry at timestamp
-        ids.insert(it, TimestampedCatalogId{catalogId, ts});
-        _markNamespaceForCatalogIdCleanupIfNeeded(nss, ids);
-        return;
-    }
+        // Avoid inserting missing mapping when the list has grown past the threshold. Will cause
+        // the system to fall back to scanning the durable catalog.
+        if (ids.size() >= kMaxCatalogIdMappingLengthForMissingInsert) {
+            return;
+        }
 
-    // Avoid inserting missing mapping when the list has grown past the threshold. Will cause the
-    // system to fall back to scanning the durable catalog.
-    if (ids.size() >= kMaxCatalogIdMappingLengthForMissingInsert) {
-        return;
-    }
+        if (it != ids.end() && it->ts == ts) {
+            // An entry could exist already if concurrent writes are performed, keep the latest
+            // change in that case.
+            it->id = boost::none;
+        } else {
+            // Otherwise insert new entry
+            it = ids.insert(it, TimestampedCatalogId{boost::none, ts});
+        }
 
-    if (it != ids.end() && it->ts == ts) {
-        // An entry could exist already if concurrent writes are performed, keep the latest change
-        // in that case.
-        it->id = boost::none;
-    } else {
-        // Otherwise insert new entry
-        it = ids.insert(it, TimestampedCatalogId{boost::none, ts});
-    }
+        // The iterator is positioned on the added/modified element above, reposition it to the next
+        // entry
+        ++it;
 
-    // The iterator is positioned on the added/modified element above, reposition it to the next
-    // entry
-    ++it;
+        // We don't want to assume that the namespace or UUID remains not existing until the next
+        // entry, as there can be times where the namespace or UUID actually does exist. To make
+        // sure we trigger the scanning of the durable catalog in this range we will insert a bogus
+        // entry using an invalid RecordId at the next timestamp. This will treat the range forward
+        // as unknown.
+        auto nextTs = ts + 1;
+
+        // If the next entry is on the next timestamp already, we can skip adding the bogus entry.
+        // If this function is called for a previously unknown namespace or UUID, we may not have
+        // any future valid entries and the iterator would be positioned at and at this point.
+        if (it == ids.end() || it->ts != nextTs) {
+            ids.insert(it, TimestampedCatalogId{kUnknownRangeMarkerId, nextTs});
+        }
 
-    // We don't want to assume that the namespace remains not existing until the next entry, as
-    // there can be times where the namespace actually does exist. To make sure we trigger the
-    // scanning of the durable catalog in this range we will insert a bogus entry using an invalid
-    // RecordId at the next timestamp. This will treat the range forward as unknown.
-    auto nextTs = ts + 1;
+        _markForCatalogIdCleanupIfNeeded(key, catalogIdChangesContainer, ids);
+    };
 
-    // If the next entry is on the next timestamp already, we can skip adding the bogus entry. If
-    // this function is called for a previously unknown namespace, we may not have any future valid
-    // entries and the iterator would be positioned at and at this point.
-    if (it == ids.end() || it->ts != nextTs) {
-        ids.insert(it, TimestampedCatalogId{kUnknownRangeMarkerId, nextTs});
+    if (nss) {
+        doInsert(_nssCatalogIds[*nss], _nssCatalogIdChanges, *nss);
     }
 
-    _markNamespaceForCatalogIdCleanupIfNeeded(nss, ids);
+    if (uuid) {
+        doInsert(_uuidCatalogIds[*uuid], _uuidCatalogIdChanges, *uuid);
+    }
 }
 
-void CollectionCatalog::_markNamespaceForCatalogIdCleanupIfNeeded(
-    const NamespaceString& nss, const std::vector<TimestampedCatalogId>& ids) {
+template <class Key, class CatalogIdChangesContainer>
+void CollectionCatalog::_markForCatalogIdCleanupIfNeeded(
+    const Key& key,
+    CatalogIdChangesContainer& catalogIdChangesContainer,
+    const std::vector<TimestampedCatalogId>& ids) {
 
-    auto markForCleanup = [this, &nss](Timestamp ts) {
-        _catalogIdChanges.insert(nss);
+    auto markForCleanup = [this, &key, &catalogIdChangesContainer](Timestamp ts) {
+        catalogIdChangesContainer.insert(key);
         if (ts < _lowestCatalogIdTimestampForCleanup) {
             _lowestCatalogIdTimestampForCleanup = ts;
         }
@@ -2367,88 +2412,103 @@ void CollectionCatalog::cleanupForOldestTimestampAdvanced(Timestamp oldest) {
         nextLowestCleanupTimestamp = std::min(nextLowestCleanupTimestamp, it->ts);
     };
 
-    // Iterate over all namespaces that is marked that they need cleanup
-    for (auto it = _catalogIdChanges.begin(), end = _catalogIdChanges.end(); it != end;) {
-        auto& range = _catalogIds[*it];
-
-        // Binary search for next larger timestamp
-        auto rangeIt = std::upper_bound(
-            range.begin(), range.end(), oldest, [](const auto& ts, const auto& entry) {
-                return ts < entry.ts;
-            });
-
-        // Continue if there is nothing to cleanup for this timestamp yet
-        if (rangeIt == range.begin()) {
-            // There should always be at least two entries in the range when we hit this branch. For
-            // the namespace to be put in '_catalogIdChanges' we normally need at least two entries.
-            // The namespace could require cleanup with just a single entry if
-            // 'cleanupForCatalogReopen' leaves a single drop entry in the range. But because we
-            // cannot initialize the namespace with a single drop there must have been a non-drop
-            // entry earlier that got cleaned up in a previous call to
-            // 'cleanupForOldestTimestampAdvanced', which happens when the oldest timestamp advances
-            // past the drop timestamp. This guarantees that the oldest timestamp is larger than the
-            // timestamp in the single drop entry resulting in this branch cannot be taken when we
-            // only have a drop in the range.
-            invariant(range.size() > 1);
-            assignLowestCleanupTimestamp(range);
-            ++it;
-            continue;
-        }
+    auto doCleanup = [this, &oldest, &assignLowestCleanupTimestamp](
+                         auto& catalogIdsContainer, auto& catalogIdChangesContainer) {
+        for (auto it = catalogIdChangesContainer.begin(), end = catalogIdChangesContainer.end();
+             it != end;) {
+            auto& range = catalogIdsContainer[*it];
+
+            // Binary search for next larger timestamp
+            auto rangeIt = std::upper_bound(
+                range.begin(), range.end(), oldest, [](const auto& ts, const auto& entry) {
+                    return ts < entry.ts;
+                });
+
+            // Continue if there is nothing to cleanup for this timestamp yet
+            if (rangeIt == range.begin()) {
+                // There should always be at least two entries in the range when we hit this
+                // branch. For the namespace to be put in '_nssCatalogIdChanges' we normally
+                // need at least two entries. The namespace could require cleanup with just a
+                // single entry if 'cleanupForCatalogReopen' leaves a single drop entry in the
+                // range. But because we cannot initialize the namespace with a single drop
+                // there must have been a non-drop entry earlier that got cleaned up in a
+                // previous call to 'cleanupForOldestTimestampAdvanced', which happens when the
+                // oldest timestamp advances past the drop timestamp. This guarantees that the
+                // oldest timestamp is larger than the timestamp in the single drop entry
+                // resulting in this branch cannot be taken when we only have a drop in the
+                // range.
+                invariant(range.size() > 1);
+                assignLowestCleanupTimestamp(range);
+                ++it;
+                continue;
+            }
 
-        // The iterator is positioned to the closest entry that has a larger timestamp, decrement to
-        // get a lower or equal timestamp
-        --rangeIt;
+            // The iterator is positioned to the closest entry that has a larger timestamp,
+            // decrement to get a lower or equal timestamp
+            --rangeIt;
 
-        // Erase range, we will leave at least one element due to the decrement above
-        range.erase(range.begin(), rangeIt);
+            // Erase range, we will leave at least one element due to the decrement above
+            range.erase(range.begin(), rangeIt);
 
-        // If more changes are needed for this namespace, keep it in the set and keep track of
-        // lowest timestamp.
-        if (range.size() > 1) {
-            assignLowestCleanupTimestamp(range);
-            ++it;
-            continue;
-        }
+            // If more changes are needed for this namespace, keep it in the set and keep track
+            // of lowest timestamp.
+            if (range.size() > 1) {
+                assignLowestCleanupTimestamp(range);
+                ++it;
+                continue;
+            }
+            // If the last remaining element is a drop earlier than the oldest timestamp, we can
+            // remove tracking this namespace
+            if (range.back().id == boost::none) {
+                catalogIdsContainer.erase(*it);
+            }
 
-        // If the last remaining element is a drop earlier than the oldest timestamp, we can remove
-        // tracking this namespace
-        if (range.back().id == boost::none) {
-            _catalogIds.erase(*it);
+            // Unmark this namespace or UUID for needing changes.
+            catalogIdChangesContainer.erase(it++);
         }
+    };
 
-        // Unmark this namespace for needing changes.
-        _catalogIdChanges.erase(it++);
-    }
+    // Iterate over all namespaces and UUIDs that is marked that they need cleanup
+    doCleanup(_nssCatalogIds, _nssCatalogIdChanges);
+    doCleanup(_uuidCatalogIds, _uuidCatalogIdChanges);
 
     _lowestCatalogIdTimestampForCleanup = nextLowestCleanupTimestamp;
     _oldestCatalogIdTimestampMaintained = std::max(_oldestCatalogIdTimestampMaintained, oldest);
 }
 
 void CollectionCatalog::cleanupForCatalogReopen(Timestamp stable) {
-    _catalogIdChanges.clear();
+    _nssCatalogIdChanges.clear();
+    _uuidCatalogIdChanges.clear();
     _lowestCatalogIdTimestampForCleanup = Timestamp::max();
     _oldestCatalogIdTimestampMaintained = std::min(_oldestCatalogIdTimestampMaintained, stable);
 
-    for (auto it = _catalogIds.begin(); it != _catalogIds.end();) {
-        auto& ids = it->second;
-
-        // Remove all larger timestamps in this range
-        ids.erase(std::upper_bound(ids.begin(),
-                                   ids.end(),
-                                   stable,
-                                   [](Timestamp ts, const auto& entry) { return ts < entry.ts; }),
-                  ids.end());
+    auto removeLargerTimestamps = [this, &stable](auto& catalogIdsContainer,
+                                                  auto& catalogIdChangesContainer) {
+        for (auto it = catalogIdsContainer.begin(); it != catalogIdsContainer.end();) {
+            auto& ids = it->second;
+
+            // Remove all larger timestamps in this range
+            ids.erase(
+                std::upper_bound(ids.begin(),
+                                 ids.end(),
+                                 stable,
+                                 [](Timestamp ts, const auto& entry) { return ts < entry.ts; }),
+                ids.end());
+
+            // Remove namespace or UUID if there are no entries left
+            if (ids.empty()) {
+                catalogIdsContainer.erase(it++);
+                continue;
+            }
 
-        // Remove namespace if there are no entries left
-        if (ids.empty()) {
-            _catalogIds.erase(it++);
-            continue;
+            // Calculate when this namespace needs to be cleaned up next
+            _markForCatalogIdCleanupIfNeeded(it->first, catalogIdChangesContainer, ids);
+            ++it;
         }
+    };
 
-        // Calculate when this namespace needs to be cleaned up next
-        _markNamespaceForCatalogIdCleanupIfNeeded(it->first, ids);
-        ++it;
-    }
+    removeLargerTimestamps(_nssCatalogIds, _nssCatalogIdChanges);
+    removeLargerTimestamps(_uuidCatalogIds, _uuidCatalogIdChanges);
 }
 
 void CollectionCatalog::invariantHasExclusiveAccessToCollection(OperationContext* opCtx,
diff --git a/src/mongo/db/catalog/collection_catalog.h b/src/mongo/db/catalog/collection_catalog.h
index f10430dbe5a..ed5eda10c1b 100644
--- a/src/mongo/db/catalog/collection_catalog.h
+++ b/src/mongo/db/catalog/collection_catalog.h
@@ -450,23 +450,25 @@ public:
     bool containsCollection(OperationContext* opCtx, const Collection* collection) const;
 
     /**
-     * Returns the CatalogId for a given 'nss' at timestamp 'ts'.
+     * Returns the CatalogId for a given 'nss' or 'uuid' at timestamp 'ts'.
      */
     struct CatalogIdLookup {
-        enum class NamespaceExistence {
-            // Namespace exists at time 'ts' and catalogId set in 'id'.
+        enum class Existence {
+            // Namespace or UUID exists at time 'ts' and catalogId set in 'id'.
             kExists,
-            // Namespace does not exist at time 'ts'.
+            // Namespace or UUID does not exist at time 'ts'.
             kNotExists,
-            // Namespace existence at time 'ts' is unknown. The durable catalog must be scanned to
-            // determine.
+            // Namespace or UUID existence at time 'ts' is unknown. The durable catalog must be
+            // scanned to determine.
             kUnknown
         };
         RecordId id;
-        NamespaceExistence result;
+        Existence result;
     };
     CatalogIdLookup lookupCatalogIdByNSS(const NamespaceString& nss,
                                          boost::optional<Timestamp> ts = boost::none) const;
+    CatalogIdLookup lookupCatalogIdByUUID(const UUID& uuid,
+                                          boost::optional<Timestamp> ts = boost::none) const;
 
     /**
      * Iterates through the views in the catalog associated with database `dbName`, applying
@@ -706,7 +708,7 @@ private:
      */
     boost::optional<DurableCatalogEntry> _fetchPITCatalogEntry(
         OperationContext* opCtx,
-        const NamespaceString& nss,
+        const NamespaceStringOrUUID& nssOrUUID,
         boost::optional<Timestamp> readTimestamp) const;
 
     /**
@@ -786,12 +788,13 @@ private:
         Timestamp ts;
     };
 
-    // Push a catalogId for namespace at given Timestamp. Timestamp needs to be larger than other
-    // entries for this namespace. boost::none for catalogId represent drop, boost::none for
-    // timestamp turns this operation into a no-op.
-    void _pushCatalogIdForNSS(const NamespaceString& nss,
-                              boost::optional<RecordId> catalogId,
-                              boost::optional<Timestamp> ts);
+    // Push a catalogId for namespace and UUID at given Timestamp. Timestamp needs to be larger than
+    // other entries for this namespace and UUID. boost::none for catalogId represent drop,
+    // boost::none for timestamp turns this operation into a no-op.
+    void _pushCatalogIdForNSSAndUUID(const NamespaceString& nss,
+                                     const UUID& uuid,
+                                     boost::optional<RecordId> catalogId,
+                                     boost::optional<Timestamp> ts);
 
     // Push a catalogId for 'from' and 'to' for a rename operation at given Timestamp. Timestamp
     // needs to be larger than other entries for these namespaces. boost::none for timestamp turns
@@ -800,16 +803,19 @@ private:
                                  const NamespaceString& to,
                                  boost::optional<Timestamp> ts);
 
-    // Inserts a catalogId for namespace at given Timestamp. Used after scanning the durable catalog
-    // for a correct mapping at the given timestamp.
-    void _insertCatalogIdForNSSAfterScan(const NamespaceString& nss,
-                                         boost::optional<RecordId> catalogId,
-                                         Timestamp ts);
+    // Inserts a catalogId for namespace and UUID at given Timestamp, if not boost::none. Used after
+    // scanning the durable catalog for a correct mapping at the given timestamp.
+    void _insertCatalogIdForNSSAndUUIDAfterScan(boost::optional<NamespaceString> nss,
+                                                boost::optional<UUID> uuid,
+                                                boost::optional<RecordId> catalogId,
+                                                Timestamp ts);
 
-    // Helper to calculate if a namespace needs to be marked for cleanup for a set of timestamped
-    // catalogIds
-    void _markNamespaceForCatalogIdCleanupIfNeeded(const NamespaceString& nss,
-                                                   const std::vector<TimestampedCatalogId>& ids);
+    // Helper to calculate if a namespace or UUID needs to be marked for cleanup for a set of
+    // timestamped catalogIds
+    template <class Key, class CatalogIdChangesContainer>
+    void _markForCatalogIdCleanupIfNeeded(const Key& key,
+                                          CatalogIdChangesContainer& catalogIdChangesContainer,
+                                          const std::vector<TimestampedCatalogId>& ids);
 
     /**
      * Returns true if catalog information about this namespace or UUID should be looked up from the
@@ -843,9 +849,16 @@ private:
                                                          const NamespaceString& nss) const;
     const Collection* _openCollectionAtLatestByUUID(OperationContext* opCtx,
                                                     const UUID& uuid) const;
-    const Collection* _openCollectionAtPointInTimeByNamespace(OperationContext* opCtx,
-                                                              const NamespaceString& nss,
-                                                              Timestamp readTimestamp) const;
+    const Collection* _openCollectionAtPointInTimeByNamespaceOrUUID(
+        OperationContext* opCtx,
+        const NamespaceStringOrUUID& nssOrUUID,
+        Timestamp readTimestamp) const;
+
+    // Helpers for 'lookupCatalogIdByNSS' and 'lookupCatalogIdByUUID'.
+    CatalogIdLookup _checkWithOldestCatalogIdTimestampMaintained(
+        boost::optional<Timestamp> ts) const;
+    CatalogIdLookup _findCatalogIdInRange(boost::optional<Timestamp> ts,
+                                          const std::vector<TimestampedCatalogId>& range) const;
 
     /**
      * When present, indicates that the catalog is in closed state, and contains a map from UUID
@@ -873,11 +886,15 @@ private:
     absl::flat_hash_map<NamespaceString, std::shared_ptr<Collection>> _pendingCommitNamespaces;
     absl::flat_hash_map<UUID, std::shared_ptr<Collection>, UUID::Hash> _pendingCommitUUIDs;
 
-    // CatalogId mappings for all known namespaces for the CollectionCatalog. The vector is sorted
-    // on timestamp.
-    absl::flat_hash_map<NamespaceString, std::vector<TimestampedCatalogId>> _catalogIds;
-    // Set of namespaces that need cleanup when the oldest timestamp advances sufficiently.
-    absl::flat_hash_set<NamespaceString> _catalogIdChanges;
+    // CatalogId mappings for all known namespaces and UUIDs for the CollectionCatalog. The vector
+    // is sorted on timestamp. UUIDs will have at most two entries. One for the create and another
+    // for the drop. UUIDs stay the same across collection renames.
+    absl::flat_hash_map<NamespaceString, std::vector<TimestampedCatalogId>> _nssCatalogIds;
+    absl::flat_hash_map<UUID, std::vector<TimestampedCatalogId>, UUID::Hash> _uuidCatalogIds;
+    // Set of namespaces and UUIDs that need cleanup when the oldest timestamp advances
+    // sufficiently.
+    absl::flat_hash_set<NamespaceString> _nssCatalogIdChanges;
+    absl::flat_hash_set<UUID, UUID::Hash> _uuidCatalogIdChanges;
     // Point at which the oldest timestamp need to advance for there to be any catalogId namespace
     // that can be cleaned up
     Timestamp _lowestCatalogIdTimestampForCleanup = Timestamp::max();
diff --git a/src/mongo/db/catalog/collection_catalog_test.cpp b/src/mongo/db/catalog/collection_catalog_test.cpp
index 6f7d643ffc6..2239cd69725 100644
--- a/src/mongo/db/catalog/collection_catalog_test.cpp
+++ b/src/mongo/db/catalog/collection_catalog_test.cpp
@@ -843,13 +843,27 @@ public:
         return CollectionCatalog::get(opCtx.get());
     }
 
-    void createCollection(OperationContext* opCtx,
+    UUID createCollection(OperationContext* opCtx,
                           const NamespaceString& nss,
                           Timestamp timestamp) {
         _setupDDLOperation(opCtx, timestamp);
         WriteUnitOfWork wuow(opCtx);
-        _createCollection(opCtx, nss);
+        UUID uuid = _createCollection(opCtx, nss);
         wuow.commit();
+        return uuid;
+    }
+
+    CollectionCatalog::CatalogIdLookup lookupCatalogId(const NamespaceString& nss,
+                                                       const UUID& uuid,
+                                                       boost::optional<Timestamp> ts) {
+        // Verify that lookups and NSS and UUID yield the same result.
+        CollectionCatalog::CatalogIdLookup nssLookup = catalog()->lookupCatalogIdByNSS(nss, ts);
+        CollectionCatalog::CatalogIdLookup uuidLookup = catalog()->lookupCatalogIdByUUID(uuid, ts);
+
+        ASSERT_EQ(nssLookup.result, uuidLookup.result);
+        ASSERT_EQ(nssLookup.id, uuidLookup.id);
+
+        return nssLookup;
     }
 
     void dropCollection(OperationContext* opCtx, const NamespaceString& nss, Timestamp timestamp) {
@@ -1063,7 +1077,7 @@ private:
         opCtx->recoveryUnit()->setCommitTimestamp(timestamp);
     }
 
-    void _createCollection(OperationContext* opCtx,
+    UUID _createCollection(OperationContext* opCtx,
                            const NamespaceString& nss,
                            boost::optional<UUID> uuid = boost::none) {
         AutoGetDb databaseWriteGuard(opCtx, nss.dbName(), MODE_IX);
@@ -1092,6 +1106,7 @@ private:
 
         // Adds the collection to the in-memory catalog.
         CollectionCatalog::get(opCtx)->onCreateCollection(opCtx, std::move(ownedCollection));
+        return *options.uuid;
     }
 
     void _dropCollection(OperationContext* opCtx, const NamespaceString& nss, Timestamp timestamp) {
@@ -1302,7 +1317,9 @@ private:
                               .get(),
                           coll);
             } else if (auto uuid = nssOrUUID.uuid()) {
-                // TODO SERVER-71222: Check UUID->catalogId mapping here.
+                auto catalogEntry =
+                    DurableCatalog::get(opCtx)->scanForCatalogEntryByUUID(opCtx, *uuid);
+                ASSERT(!catalogEntry);
 
                 // Lookups from the catalog should return the newly opened collection (in this case
                 // nullptr).
@@ -2146,21 +2163,21 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCreate) {
     });
 
     // Create collection and extract the catalogId
-    createCollection(opCtx.get(), nss, Timestamp(1, 2));
+    UUID uuid = createCollection(opCtx.get(), nss, Timestamp(1, 2));
     RecordId rid = catalog()->lookupCollectionByNamespace(opCtx.get(), nss)->getCatalogId();
 
     // Lookup without timestamp returns latest catalogId
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, boost::none).id, rid);
+    ASSERT_EQ(lookupCatalogId(nss, uuid, boost::none).id, rid);
     // Lookup before create returns unknown if looking before oldest
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 0)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, uuid, Timestamp(1, 0)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
     // Lookup before create returns not exists if looking after oldest
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 1)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(nss, uuid, Timestamp(1, 1)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
     // Lookup at create returns catalogId
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 2)).id, rid);
+    ASSERT_EQ(lookupCatalogId(nss, uuid, Timestamp(1, 2)).id, rid);
     // Lookup after create returns catalogId
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 3)).id, rid);
+    ASSERT_EQ(lookupCatalogId(nss, uuid, Timestamp(1, 3)).id, rid);
 }
 
 TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingDrop) {
@@ -2175,29 +2192,29 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingDrop) {
     });
 
     // Create and drop collection. We have a time window where the namespace exists
-    createCollection(opCtx.get(), nss, Timestamp(1, 5));
+    UUID uuid = createCollection(opCtx.get(), nss, Timestamp(1, 5));
     RecordId rid = catalog()->lookupCollectionByNamespace(opCtx.get(), nss)->getCatalogId();
     dropCollection(opCtx.get(), nss, Timestamp(1, 10));
 
     // Lookup without timestamp returns none
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, boost::none).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(nss, uuid, boost::none).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
     // Lookup before create and oldest returns unknown
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 0)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, uuid, Timestamp(1, 0)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
     // Lookup before create returns not exists
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 4)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(nss, uuid, Timestamp(1, 4)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
     // Lookup at create returns catalogId
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 5)).id, rid);
+    ASSERT_EQ(lookupCatalogId(nss, uuid, Timestamp(1, 5)).id, rid);
     // Lookup after create returns catalogId
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 6)).id, rid);
+    ASSERT_EQ(lookupCatalogId(nss, uuid, Timestamp(1, 6)).id, rid);
     // Lookup at drop returns none
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 10)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(nss, uuid, Timestamp(1, 10)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
     // Lookup after drop returns none
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 20)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(nss, uuid, Timestamp(1, 20)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
 }
 
 TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingRename) {
@@ -2214,42 +2231,46 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingRename) {
 
     // Create and rename collection. We have two windows where the collection exists but for
     // different namespaces
-    createCollection(opCtx.get(), from, Timestamp(1, 5));
+    UUID uuid = createCollection(opCtx.get(), from, Timestamp(1, 5));
     RecordId rid = catalog()->lookupCollectionByNamespace(opCtx.get(), from)->getCatalogId();
     renameCollection(opCtx.get(), from, to, Timestamp(1, 10));
 
-    // Lookup without timestamp on 'from' returns none
+    // Lookup without timestamp on 'from' returns none. By 'uuid' returns catalogId
     ASSERT_EQ(catalog()->lookupCatalogIdByNSS(from, boost::none).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByUUID(uuid, boost::none).id, rid);
     // Lookup before create and oldest returns unknown
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(from, Timestamp(1, 0)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(from, uuid, Timestamp(1, 0)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
     // Lookup before create returns not exists
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(from, Timestamp(1, 4)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(from, uuid, Timestamp(1, 4)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
     // Lookup at create returns catalogId
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(from, Timestamp(1, 5)).id, rid);
+    ASSERT_EQ(lookupCatalogId(from, uuid, Timestamp(1, 5)).id, rid);
     // Lookup after create returns catalogId
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(from, Timestamp(1, 6)).id, rid);
-    // Lookup at rename on 'from' returns none
+    ASSERT_EQ(lookupCatalogId(from, uuid, Timestamp(1, 6)).id, rid);
+    // Lookup at rename on 'from' returns none. By 'uuid' returns catalogId
     ASSERT_EQ(catalog()->lookupCatalogIdByNSS(from, Timestamp(1, 10)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
-    // Lookup after rename on 'from' returns none
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByUUID(uuid, Timestamp(1, 10)).id, rid);
+    // Lookup after rename on 'from' returns none. By 'uuid' returns catalogId
     ASSERT_EQ(catalog()->lookupCatalogIdByNSS(from, Timestamp(1, 20)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByUUID(uuid, Timestamp(1, 20)).id, rid);
 
     // Lookup without timestamp on 'to' returns catalogId
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(to, boost::none).id, rid);
+    ASSERT_EQ(lookupCatalogId(to, uuid, boost::none).id, rid);
     // Lookup before rename and oldest on 'to' returns unknown
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(to, Timestamp(1, 0)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
-    // Lookup before rename on 'to' returns not exists
+    ASSERT_EQ(lookupCatalogId(to, uuid, Timestamp(1, 0)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    // Lookup before rename on 'to' returns not exists. By 'uuid' returns catalogId
     ASSERT_EQ(catalog()->lookupCatalogIdByNSS(to, Timestamp(1, 9)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByUUID(uuid, Timestamp(1, 9)).id, rid);
     // Lookup at rename on 'to' returns catalogId
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(to, Timestamp(1, 10)).id, rid);
+    ASSERT_EQ(lookupCatalogId(to, uuid, Timestamp(1, 10)).id, rid);
     // Lookup after rename on 'to' returns catalogId
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(to, Timestamp(1, 20)).id, rid);
+    ASSERT_EQ(lookupCatalogId(to, uuid, Timestamp(1, 20)).id, rid);
 }
 
 TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingRenameDropTarget) {
@@ -2267,24 +2288,39 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingRenameDropTarget) {
     // Create collections. The 'to' namespace will exist for one collection from Timestamp(1, 6)
     // until it is dropped by the rename at Timestamp(1, 10), after which the 'to' namespace will
     // correspond to the renamed collection.
-    createCollection(opCtx.get(), from, Timestamp(1, 5));
-    createCollection(opCtx.get(), to, Timestamp(1, 6));
+    UUID uuid = createCollection(opCtx.get(), from, Timestamp(1, 5));
+    UUID originalUUID = createCollection(opCtx.get(), to, Timestamp(1, 6));
     RecordId rid = catalog()->lookupCollectionByNamespace(opCtx.get(), from)->getCatalogId();
     RecordId originalToRid =
         catalog()->lookupCollectionByNamespace(opCtx.get(), to)->getCatalogId();
     renameCollection(opCtx.get(), from, to, Timestamp(1, 10));
 
-    // Lookup without timestamp on 'to' returns latest catalog id
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(to, boost::none).id, rid);
+    // Lookup without timestamp on 'to' and 'uuid' returns latest catalog id. By 'originalUUID'
+    // returns not exists as the target was dropped.
+    ASSERT_EQ(lookupCatalogId(to, uuid, boost::none).id, rid);
+    ASSERT_EQ(catalog()->lookupCatalogIdByUUID(originalUUID, boost::none).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
     // Lookup before rename and oldest on 'to' returns unknown
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(to, Timestamp(1, 0)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(to, uuid, Timestamp(1, 0)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(to, originalUUID, Timestamp(1, 0)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
     // Lookup before rename on 'to' returns the original rid
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(to, Timestamp(1, 9)).id, originalToRid);
-    // Lookup at rename timestamp on 'to' returns catalogId
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(to, Timestamp(1, 10)).id, rid);
-    // Lookup after rename on 'to' returns catalogId
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(to, Timestamp(1, 20)).id, rid);
+    ASSERT_EQ(lookupCatalogId(to, originalUUID, Timestamp(1, 9)).id, originalToRid);
+    // Lookup before rename on 'from' returns the rid
+    ASSERT_EQ(lookupCatalogId(from, uuid, Timestamp(1, 9)).id, rid);
+    // Lookup at rename timestamp on 'to' and 'uuid' returns catalogId
+    ASSERT_EQ(lookupCatalogId(to, uuid, Timestamp(1, 10)).id, rid);
+    // Lookup at rename timestamp on 'originalUUID' returns not exists as it was dropped during the
+    // rename.
+    ASSERT_EQ(catalog()->lookupCatalogIdByUUID(originalUUID, Timestamp(1, 10)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
+    // Lookup after rename on 'to' and 'uuid' returns catalogId
+    ASSERT_EQ(lookupCatalogId(to, uuid, Timestamp(1, 20)).id, rid);
+    // Lookup after rename timestamp on 'originalUUID' returns not exists as it was dropped during
+    // the rename.
+    ASSERT_EQ(catalog()->lookupCatalogIdByUUID(originalUUID, Timestamp(1, 20)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
 }
 
 TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingDropCreate) {
@@ -2294,31 +2330,47 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingDropCreate) {
     NamespaceString nss = NamespaceString::createNamespaceString_forTest("a.b");
 
     // Create, drop and recreate collection on the same namespace. We have different catalogId.
-    createCollection(opCtx.get(), nss, Timestamp(1, 5));
+    UUID firstUUID = createCollection(opCtx.get(), nss, Timestamp(1, 5));
     RecordId rid1 = catalog()->lookupCollectionByNamespace(opCtx.get(), nss)->getCatalogId();
     dropCollection(opCtx.get(), nss, Timestamp(1, 10));
-    createCollection(opCtx.get(), nss, Timestamp(1, 15));
+    UUID secondUUID = createCollection(opCtx.get(), nss, Timestamp(1, 15));
     RecordId rid2 = catalog()->lookupCollectionByNamespace(opCtx.get(), nss)->getCatalogId();
 
     // Lookup without timestamp returns latest catalogId
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, boost::none).id, rid2);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, boost::none).id, rid2);
+    ASSERT_EQ(catalog()->lookupCatalogIdByUUID(firstUUID, boost::none).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
     // Lookup before first create returns not exists
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 4)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 4)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 4)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
     // Lookup at first create returns first catalogId
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 5)).id, rid1);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 5)).id, rid1);
+    ASSERT_EQ(catalog()->lookupCatalogIdByUUID(secondUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
     // Lookup after first create returns first catalogId
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 6)).id, rid1);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 6)).id, rid1);
+    ASSERT_EQ(catalog()->lookupCatalogIdByUUID(secondUUID, Timestamp(1, 6)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
     // Lookup at drop returns none
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 10)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 10)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 10)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
     // Lookup after drop returns none
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 13)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 13)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 13)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
     // Lookup at second create returns second catalogId
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).id, rid2);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 15)).id, rid2);
+    ASSERT_EQ(catalog()->lookupCatalogIdByUUID(firstUUID, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
     // Lookup after second create returns second catalogId
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 20)).id, rid2);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 20)).id, rid2);
+    ASSERT_EQ(catalog()->lookupCatalogIdByUUID(firstUUID, Timestamp(1, 20)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
 }
 
 TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupEqDrop) {
@@ -2328,7 +2380,7 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupEqDrop) {
     NamespaceString nss = NamespaceString::createNamespaceString_forTest("a.b");
 
     // Create collection and verify we have nothing to cleanup
-    createCollection(opCtx.get(), nss, Timestamp(1, 5));
+    UUID firstUUID = createCollection(opCtx.get(), nss, Timestamp(1, 5));
     ASSERT_FALSE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 1)));
 
     // Drop collection and verify we have nothing to cleanup as long as the oldest timestamp is
@@ -2339,15 +2391,17 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupEqDrop) {
     ASSERT_TRUE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 10)));
 
     // Create new collection and nothing changed with answers to needsCleanupForOldestTimestamp.
-    createCollection(opCtx.get(), nss, Timestamp(1, 15));
+    UUID secondUUID = createCollection(opCtx.get(), nss, Timestamp(1, 15));
     ASSERT_FALSE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 1)));
     ASSERT_FALSE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 5)));
     ASSERT_FALSE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 7)));
     ASSERT_TRUE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 10)));
 
     // We can lookup the old catalogId before we advance the oldest timestamp and cleanup
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 5)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByUUID(secondUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
 
     // Cleanup at drop timestamp
     CollectionCatalog::write(opCtx.get(), [&](CollectionCatalog& c) {
@@ -2356,10 +2410,14 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupEqDrop) {
     // After cleanup, we cannot find the old catalogId anymore. Also verify that we don't need
     // anymore cleanup
     ASSERT_FALSE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 10)));
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 5)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByUUID(firstUUID, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
 }
 
 TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupGtDrop) {
@@ -2369,7 +2427,7 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupGtDrop) {
     NamespaceString nss = NamespaceString::createNamespaceString_forTest("a.b");
 
     // Create collection and verify we have nothing to cleanup
-    createCollection(opCtx.get(), nss, Timestamp(1, 5));
+    UUID firstUUID = createCollection(opCtx.get(), nss, Timestamp(1, 5));
     ASSERT_FALSE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 1)));
 
     // Drop collection and verify we have nothing to cleanup as long as the oldest timestamp is
@@ -2380,15 +2438,17 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupGtDrop) {
     ASSERT_TRUE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 10)));
 
     // Create new collection and nothing changed with answers to needsCleanupForOldestTimestamp.
-    createCollection(opCtx.get(), nss, Timestamp(1, 15));
+    UUID secondUUID = createCollection(opCtx.get(), nss, Timestamp(1, 15));
     ASSERT_FALSE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 1)));
     ASSERT_FALSE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 5)));
     ASSERT_FALSE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 7)));
     ASSERT_TRUE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 12)));
 
     // We can lookup the old catalogId before we advance the oldest timestamp and cleanup
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 5)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByUUID(secondUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
 
     // Cleanup after the drop timestamp
     CollectionCatalog::write(opCtx.get(), [&](CollectionCatalog& c) {
@@ -2398,10 +2458,14 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupGtDrop) {
     // After cleanup, we cannot find the old catalogId anymore. Also verify that we don't need
     // anymore cleanup
     ASSERT_FALSE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 12)));
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 5)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByUUID(firstUUID, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
 }
 
 TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupGtRecreate) {
@@ -2411,7 +2475,7 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupGtRecreate) {
     NamespaceString nss = NamespaceString::createNamespaceString_forTest("a.b");
 
     // Create collection and verify we have nothing to cleanup
-    createCollection(opCtx.get(), nss, Timestamp(1, 5));
+    UUID firstUUID = createCollection(opCtx.get(), nss, Timestamp(1, 5));
     ASSERT_FALSE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 1)));
 
     // Drop collection and verify we have nothing to cleanup as long as the oldest timestamp is
@@ -2422,15 +2486,17 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupGtRecreate) {
     ASSERT_TRUE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 10)));
 
     // Create new collection and nothing changed with answers to needsCleanupForOldestTimestamp.
-    createCollection(opCtx.get(), nss, Timestamp(1, 15));
+    UUID secondUUID = createCollection(opCtx.get(), nss, Timestamp(1, 15));
     ASSERT_FALSE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 1)));
     ASSERT_FALSE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 5)));
     ASSERT_FALSE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 7)));
     ASSERT_TRUE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 20)));
 
     // We can lookup the old catalogId before we advance the oldest timestamp and cleanup
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 5)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByUUID(secondUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
 
     // Cleanup after the recreate timestamp
     CollectionCatalog::write(opCtx.get(), [&](CollectionCatalog& c) {
@@ -2440,10 +2506,14 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupGtRecreate) {
     // After cleanup, we cannot find the old catalogId anymore. Also verify that we don't need
     // anymore cleanup
     ASSERT_FALSE(catalog()->needsCleanupForOldestTimestamp(Timestamp(1, 20)));
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 5)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByUUID(firstUUID, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
 }
 
 TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupMultiple) {
@@ -2453,24 +2523,24 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupMultiple) {
     NamespaceString nss = NamespaceString::createNamespaceString_forTest("a.b");
 
     // Create and drop multiple namespace on the same namespace
-    createCollection(opCtx.get(), nss, Timestamp(1, 5));
+    UUID firstUUID = createCollection(opCtx.get(), nss, Timestamp(1, 5));
     dropCollection(opCtx.get(), nss, Timestamp(1, 10));
-    createCollection(opCtx.get(), nss, Timestamp(1, 15));
+    UUID secondUUID = createCollection(opCtx.get(), nss, Timestamp(1, 15));
     dropCollection(opCtx.get(), nss, Timestamp(1, 20));
-    createCollection(opCtx.get(), nss, Timestamp(1, 25));
+    UUID thirdUUID = createCollection(opCtx.get(), nss, Timestamp(1, 25));
     dropCollection(opCtx.get(), nss, Timestamp(1, 30));
-    createCollection(opCtx.get(), nss, Timestamp(1, 35));
+    UUID fourthUUID = createCollection(opCtx.get(), nss, Timestamp(1, 35));
     dropCollection(opCtx.get(), nss, Timestamp(1, 40));
 
     // Lookup can find all four collections
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 5)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 35)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, thirdUUID, Timestamp(1, 25)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, fourthUUID, Timestamp(1, 35)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
 
     // Cleanup oldest
     CollectionCatalog::write(opCtx.get(), [&](CollectionCatalog& c) {
@@ -2478,14 +2548,14 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupMultiple) {
     });
 
     // Lookup can find the three remaining collections
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 5)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 35)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, thirdUUID, Timestamp(1, 25)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, fourthUUID, Timestamp(1, 35)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
 
     // Cleanup
     CollectionCatalog::write(opCtx.get(), [&](CollectionCatalog& c) {
@@ -2493,14 +2563,14 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupMultiple) {
     });
 
     // Lookup can find the two remaining collections
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 5)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 35)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, thirdUUID, Timestamp(1, 25)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, fourthUUID, Timestamp(1, 35)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
 
     // Cleanup
     CollectionCatalog::write(opCtx.get(), [&](CollectionCatalog& c) {
@@ -2508,14 +2578,14 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupMultiple) {
     });
 
     // Lookup can find the last remaining collections
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 5)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 35)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, thirdUUID, Timestamp(1, 25)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, fourthUUID, Timestamp(1, 35)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
 
     // Cleanup
     CollectionCatalog::write(opCtx.get(), [&](CollectionCatalog& c) {
@@ -2524,14 +2594,14 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupMultiple) {
 
     // Lookup now result in unknown as the oldest timestamp has advanced where mapping has been
     // removed
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 5)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 35)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, thirdUUID, Timestamp(1, 25)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, fourthUUID, Timestamp(1, 35)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
 }
 
 TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupMultipleSingleCall) {
@@ -2541,24 +2611,24 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupMultipleSingleCall
     NamespaceString nss = NamespaceString::createNamespaceString_forTest("a.b");
 
     // Create and drop multiple namespace on the same namespace
-    createCollection(opCtx.get(), nss, Timestamp(1, 5));
+    UUID firstUUID = createCollection(opCtx.get(), nss, Timestamp(1, 5));
     dropCollection(opCtx.get(), nss, Timestamp(1, 10));
-    createCollection(opCtx.get(), nss, Timestamp(1, 15));
+    UUID secondUUID = createCollection(opCtx.get(), nss, Timestamp(1, 15));
     dropCollection(opCtx.get(), nss, Timestamp(1, 20));
-    createCollection(opCtx.get(), nss, Timestamp(1, 25));
+    UUID thirdUUID = createCollection(opCtx.get(), nss, Timestamp(1, 25));
     dropCollection(opCtx.get(), nss, Timestamp(1, 30));
-    createCollection(opCtx.get(), nss, Timestamp(1, 35));
+    UUID fourthUUID = createCollection(opCtx.get(), nss, Timestamp(1, 35));
     dropCollection(opCtx.get(), nss, Timestamp(1, 40));
 
     // Lookup can find all four collections
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 5)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 35)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, thirdUUID, Timestamp(1, 25)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(lookupCatalogId(nss, fourthUUID, Timestamp(1, 35)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
 
     // Cleanup all
     CollectionCatalog::write(opCtx.get(), [&](CollectionCatalog& c) {
@@ -2567,14 +2637,14 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingCleanupMultipleSingleCall
 
     // Lookup now result in unknown as the oldest timestamp has advanced where mapping has been
     // removed
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 5)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 35)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 5)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, thirdUUID, Timestamp(1, 25)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, fourthUUID, Timestamp(1, 35)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
 }
 
 TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingRollback) {
@@ -2588,29 +2658,31 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingRollback) {
     NamespaceString e = NamespaceString::createNamespaceString_forTest("b.e");
 
     // Create and drop multiple namespace on the same namespace
-    createCollection(opCtx.get(), a, Timestamp(1, 1));
+    UUID firstUUID = createCollection(opCtx.get(), a, Timestamp(1, 1));
     dropCollection(opCtx.get(), a, Timestamp(1, 2));
-    createCollection(opCtx.get(), a, Timestamp(1, 3));
-    createCollection(opCtx.get(), b, Timestamp(1, 5));
-    createCollection(opCtx.get(), c, Timestamp(1, 7));
-    createCollection(opCtx.get(), d, Timestamp(1, 8));
-    createCollection(opCtx.get(), e, Timestamp(1, 9));
+    UUID secondUUID = createCollection(opCtx.get(), a, Timestamp(1, 3));
+    UUID thirdUUID = createCollection(opCtx.get(), b, Timestamp(1, 5));
+    UUID fourthUUID = createCollection(opCtx.get(), c, Timestamp(1, 7));
+    UUID fifthUUID = createCollection(opCtx.get(), d, Timestamp(1, 8));
+    UUID sixthUUID = createCollection(opCtx.get(), e, Timestamp(1, 9));
     dropCollection(opCtx.get(), b, Timestamp(1, 10));
 
     // Rollback to Timestamp(1, 8)
     CollectionCatalog::write(
         opCtx.get(), [&](CollectionCatalog& c) { c.cleanupForCatalogReopen(Timestamp(1, 8)); });
 
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(a, boost::none).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(b, boost::none).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(c, boost::none).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(d, boost::none).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(e, boost::none).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(e, firstUUID, boost::none).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(a, secondUUID, boost::none).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(lookupCatalogId(b, thirdUUID, boost::none).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(lookupCatalogId(c, fourthUUID, boost::none).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(lookupCatalogId(d, fifthUUID, boost::none).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(lookupCatalogId(e, sixthUUID, boost::none).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
 }
 
 TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingInsert) {
@@ -2620,12 +2692,12 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingInsert) {
     NamespaceString nss = NamespaceString::createNamespaceString_forTest("a.b");
 
     // Create a collection on the namespace
-    createCollection(opCtx.get(), nss, Timestamp(1, 10));
+    UUID firstUUID = createCollection(opCtx.get(), nss, Timestamp(1, 10));
     dropCollection(opCtx.get(), nss, Timestamp(1, 20));
-    createCollection(opCtx.get(), nss, Timestamp(1, 30));
+    UUID secondUUID = createCollection(opCtx.get(), nss, Timestamp(1, 30));
 
-    auto rid1 = catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 10)).id;
-    auto rid2 = catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 30)).id;
+    auto rid1 = lookupCatalogId(nss, firstUUID, Timestamp(1, 10)).id;
+    auto rid2 = lookupCatalogId(nss, secondUUID, Timestamp(1, 30)).id;
 
     // Simulate startup where we have a range [oldest, stable] by creating and dropping collections
     // and then advancing the oldest timestamp and then reading behind it.
@@ -2634,33 +2706,41 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingInsert) {
     });
 
     // Confirm that the mappings have been cleaned up
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
 
     {
         OneOffRead oor(opCtx.get(), Timestamp(1, 17));
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
         CollectionCatalog::get(opCtx.get())
             ->establishConsistentCollection(opCtx.get(), nss, Timestamp(1, 17));
+        CollectionCatalog::get(opCtx.get())
+            ->establishConsistentCollection(opCtx.get(), {nss.db(), firstUUID}, Timestamp(1, 17));
+        CollectionCatalog::get(opCtx.get())
+            ->establishConsistentCollection(opCtx.get(), {nss.db(), secondUUID}, Timestamp(1, 17));
 
         // Lookups before the inserted timestamp is still unknown
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 11)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 11)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+        ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 11)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
 
         // Lookups at or after the inserted timestamp is found, even if they don't match with WT
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 17)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 17)).id, rid1);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 19)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 19)).id, rid1);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).id, rid1);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 17)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kExists);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 17)).id, rid1);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 19)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kExists);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 19)).id, rid1);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 25)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kExists);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 25)).id, rid1);
         // The entry at Timestamp(1, 30) is unaffected
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 30)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 30)).id, rid2);
+        ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 30)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kExists);
+        ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 30)).id, rid2);
     }
 
     {
@@ -2669,27 +2749,31 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingInsert) {
 
         CollectionCatalog::get(opCtx.get())
             ->establishConsistentCollection(opCtx.get(), nss, Timestamp(1, 12));
+        CollectionCatalog::get(opCtx.get())
+            ->establishConsistentCollection(opCtx.get(), {nss.db(), firstUUID}, Timestamp(1, 12));
+        CollectionCatalog::get(opCtx.get())
+            ->establishConsistentCollection(opCtx.get(), {nss.db(), secondUUID}, Timestamp(1, 12));
 
         // We should now have extended the range from Timestamp(1, 17) to Timestamp(1, 12)
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 12)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 12)).id, rid1);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 16)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 16)).id, rid1);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 17)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 17)).id, rid1);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 19)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 19)).id, rid1);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).id, rid1);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 12)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kExists);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 12)).id, rid1);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 16)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kExists);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 16)).id, rid1);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 17)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kExists);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 17)).id, rid1);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 19)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kExists);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 19)).id, rid1);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 25)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kExists);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 25)).id, rid1);
         // The entry at Timestamp(1, 30) is unaffected
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 30)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 30)).id, rid2);
+        ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 30)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kExists);
+        ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 30)).id, rid2);
     }
 
     {
@@ -2697,27 +2781,35 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingInsert) {
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
         CollectionCatalog::get(opCtx.get())
             ->establishConsistentCollection(opCtx.get(), nss, Timestamp(1, 25));
+        CollectionCatalog::get(opCtx.get())
+            ->establishConsistentCollection(opCtx.get(), {nss.db(), firstUUID}, Timestamp(1, 25));
+        CollectionCatalog::get(opCtx.get())
+            ->establishConsistentCollection(opCtx.get(), {nss.db(), secondUUID}, Timestamp(1, 25));
 
         // Check the entries, most didn't change
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 17)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 17)).id, rid1);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 19)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 19)).id, rid1);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 22)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 22)).id, rid1);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 17)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kExists);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 17)).id, rid1);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 19)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kExists);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 19)).id, rid1);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 22)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kExists);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 22)).id, rid1);
         // At Timestamp(1, 25) we now return kNotExists
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 25)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
+        ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 25)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
         // But next timestamp returns unknown
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 26)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 26)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+        ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 26)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
         // The entry at Timestamp(1, 30) is unaffected
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 30)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 30)).id, rid2);
+        ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 30)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kExists);
+        ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 30)).id, rid2);
     }
 
     {
@@ -2725,38 +2817,51 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingInsert) {
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
         CollectionCatalog::get(opCtx.get())
             ->establishConsistentCollection(opCtx.get(), nss, Timestamp(1, 26));
+        CollectionCatalog::get(opCtx.get())
+            ->establishConsistentCollection(opCtx.get(), {nss.db(), firstUUID}, Timestamp(1, 26));
+        CollectionCatalog::get(opCtx.get())
+            ->establishConsistentCollection(opCtx.get(), {nss.db(), secondUUID}, Timestamp(1, 26));
 
         // We should not have re-written the existing entry at Timestamp(1, 26)
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 17)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 17)).id, rid1);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 19)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 19)).id, rid1);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 22)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 22)).id, rid1);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 17)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kExists);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 17)).id, rid1);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 19)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kExists);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 19)).id, rid1);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 22)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kExists);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 22)).id, rid1);
         // At Timestamp(1, 25) we now return kNotExists
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 25)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
+        ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 25)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
         // But next timestamp returns unknown
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 26)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 27)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 26)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
+        ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 26)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
+        ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 27)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+        ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 27)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
         // The entry at Timestamp(1, 30) is unaffected
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 30)).result,
-                  CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-        ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 30)).id, rid2);
+        ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 30)).result,
+                  CollectionCatalog::CatalogIdLookup::Existence::kExists);
+        ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 30)).id, rid2);
     }
 
+
     // Clean up, check so we are back to the original state
     CollectionCatalog::write(opCtx.get(), [](CollectionCatalog& catalog) {
         catalog.cleanupForOldestTimestampAdvanced(Timestamp(1, 41));
     });
 
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, firstUUID, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
+    ASSERT_EQ(lookupCatalogId(nss, secondUUID, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
 }
 
 TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingInsertUnknown) {
@@ -2773,7 +2878,7 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingInsertUnknown) {
 
     // Reading before the oldest is unknown
     ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
+              CollectionCatalog::CatalogIdLookup::Existence::kUnknown);
 
     // Try to instantiate a non existing collection at this timestamp.
     CollectionCatalog::get(opCtx.get())
@@ -2781,7 +2886,7 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingInsertUnknown) {
 
     // Lookup should now be not existing
     ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
 }
 
 TEST_F(CollectionCatalogTimestampTest, CollectionLifetimeTiedToStorageTransactionLifetime) {
@@ -2878,14 +2983,14 @@ TEST_F(CollectionCatalogNoTimestampTest, CatalogIdMappingNoTimestamp) {
     NamespaceString nss = NamespaceString::createNamespaceString_forTest("a.b");
 
     // Create a collection on the namespace and confirm that we can lookup
-    createCollection(opCtx.get(), nss, Timestamp());
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    UUID uuid = createCollection(opCtx.get(), nss, Timestamp());
+    ASSERT_EQ(lookupCatalogId(nss, uuid, boost::none).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
 
     // Drop the collection and confirm it is also removed from mapping
     dropCollection(opCtx.get(), nss, Timestamp());
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(nss, uuid, boost::none).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
 }
 
 TEST_F(CollectionCatalogNoTimestampTest, CatalogIdMappingNoTimestampRename) {
@@ -2896,25 +3001,25 @@ TEST_F(CollectionCatalogNoTimestampTest, CatalogIdMappingNoTimestampRename) {
     NamespaceString b = NamespaceString::createNamespaceString_forTest("a.b");
 
     // Create a collection on the namespace and confirm that we can lookup
-    createCollection(opCtx.get(), a, Timestamp());
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(a).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    UUID uuid = createCollection(opCtx.get(), a, Timestamp());
+    ASSERT_EQ(lookupCatalogId(a, uuid, boost::none).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
     ASSERT_EQ(catalog()->lookupCatalogIdByNSS(b).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
 
     // Rename the collection and check lookup behavior
     renameCollection(opCtx.get(), a, b, Timestamp());
     ASSERT_EQ(catalog()->lookupCatalogIdByNSS(a).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(b).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(b, uuid, boost::none).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kExists);
 
     // Drop the collection and confirm it is also removed from mapping
     dropCollection(opCtx.get(), b, Timestamp());
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(a).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(b).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(a, uuid, boost::none).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(b, uuid, boost::none).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
 }
 
 TEST_F(CollectionCatalogNoTimestampTest, CatalogIdMappingNoTimestampRenameDropTarget) {
@@ -2925,29 +3030,29 @@ TEST_F(CollectionCatalogNoTimestampTest, CatalogIdMappingNoTimestampRenameDropTa
     NamespaceString b = NamespaceString::createNamespaceString_forTest("a.b");
 
     // Create collections on the namespaces and confirm that we can lookup
-    createCollection(opCtx.get(), a, Timestamp());
-    createCollection(opCtx.get(), b, Timestamp());
-    auto [aId, aResult] = catalog()->lookupCatalogIdByNSS(a);
-    auto [bId, bResult] = catalog()->lookupCatalogIdByNSS(b);
-    ASSERT_EQ(aResult, CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
-    ASSERT_EQ(bResult, CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    UUID uuidA = createCollection(opCtx.get(), a, Timestamp());
+    UUID uuidB = createCollection(opCtx.get(), b, Timestamp());
+    auto [aId, aResult] = lookupCatalogId(a, uuidA, boost::none);
+    auto [bId, bResult] = lookupCatalogId(b, uuidB, boost::none);
+    ASSERT_EQ(aResult, CollectionCatalog::CatalogIdLookup::Existence::kExists);
+    ASSERT_EQ(bResult, CollectionCatalog::CatalogIdLookup::Existence::kExists);
 
     // Rename the collection and check lookup behavior
     renameCollection(opCtx.get(), a, b, Timestamp());
-    auto [aIdAfter, aResultAfter] = catalog()->lookupCatalogIdByNSS(a);
-    auto [bIdAfter, bResultAfter] = catalog()->lookupCatalogIdByNSS(b);
-    ASSERT_EQ(aResultAfter, CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
-    ASSERT_EQ(bResultAfter, CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    auto [aIdAfter, aResultAfter] = catalog()->lookupCatalogIdByNSS(a, boost::none);
+    auto [bIdAfter, bResultAfter] = lookupCatalogId(b, uuidA, boost::none);
+    ASSERT_EQ(aResultAfter, CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
+    ASSERT_EQ(bResultAfter, CollectionCatalog::CatalogIdLookup::Existence::kExists);
     // Verify that the the recordId on b is now what was on a. We performed a rename with
     // dropTarget=true.
     ASSERT_EQ(aId, bIdAfter);
 
     // Drop the collection and confirm it is also removed from mapping
     dropCollection(opCtx.get(), b, Timestamp());
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(a).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
-    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(b).result,
-              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(a, uuidA, boost::none).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
+    ASSERT_EQ(lookupCatalogId(b, uuidB, boost::none).result,
+              CollectionCatalog::CatalogIdLookup::Existence::kNotExists);
 }
 
 DEATH_TEST_F(CollectionCatalogTimestampTest, OpenCollectionInWriteUnitOfWork, "invariant") {
diff --git a/src/mongo/db/catalog/uncommitted_catalog_updates.cpp b/src/mongo/db/catalog/uncommitted_catalog_updates.cpp
index bd0416a700c..321cd42819c 100644
--- a/src/mongo/db/catalog/uncommitted_catalog_updates.cpp
+++ b/src/mongo/db/catalog/uncommitted_catalog_updates.cpp
@@ -275,7 +275,10 @@ OpenedCollections& OpenedCollections::get(OperationContext* opCtx) {
 boost::optional<std::shared_ptr<const Collection>> OpenedCollections::lookupByNamespace(
     const NamespaceString& ns) const {
     auto it = std::find_if(_collections.begin(), _collections.end(), [&ns](const auto& entry) {
-        return entry.nss == ns;
+        if (!entry.nss)
+            return false;
+
+        return entry.nss.value() == ns;
     });
     if (it != _collections.end()) {
         return it->collection;
@@ -298,13 +301,13 @@ boost::optional<std::shared_ptr<const Collection>> OpenedCollections::lookupByUU
 }
 
 void OpenedCollections::store(std::shared_ptr<const Collection> coll,
-                              NamespaceString nss,
+                              boost::optional<NamespaceString> nss,
                               boost::optional<UUID> uuid) {
     if (coll) {
         invariant(nss == coll->ns());
         invariant(uuid == coll->uuid());
     }
-    _collections.push_back({std::move(coll), std::move(nss), uuid});
+    _collections.push_back({std::move(coll), nss, uuid});
 }
 
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/uncommitted_catalog_updates.h b/src/mongo/db/catalog/uncommitted_catalog_updates.h
index 808f3591ae1..9c2a7057229 100644
--- a/src/mongo/db/catalog/uncommitted_catalog_updates.h
+++ b/src/mongo/db/catalog/uncommitted_catalog_updates.h
@@ -306,13 +306,13 @@ public:
      * in the snapshot.
      */
     void store(std::shared_ptr<const Collection> coll,
-               NamespaceString nss,
+               boost::optional<NamespaceString> nss,
                boost::optional<UUID> uuid);
 
 private:
     struct Entry {
         std::shared_ptr<const Collection> collection;
-        NamespaceString nss;
+        boost::optional<NamespaceString> nss;
         boost::optional<UUID> uuid;
     };
 
diff --git a/src/mongo/db/db_raii.cpp b/src/mongo/db/db_raii.cpp
index 83caea3e7ff..aa60b64bed9 100644
--- a/src/mongo/db/db_raii.cpp
+++ b/src/mongo/db/db_raii.cpp
@@ -1080,7 +1080,9 @@ ConsistentCatalogAndSnapshot getConsistentCatalogAndSnapshot(
 
         const auto catalogBeforeSnapshot = CollectionCatalog::get(opCtx);
 
-        // TODO (SERVER-71222): This is broken if the UUID doesn't exist in the latest catalog.
+        // It is incorrect to resolve the UUID here as we haven't established a consistent view of
+        // this UUID yet. During a concurrent rename it can be wrong. This namespace is only used to
+        // determine if it is an internal namespace.
         const auto nss = catalogBeforeSnapshot->resolveNamespaceStringOrUUID(opCtx, nsOrUUID);
 
         // This may modify the read source on the recovery unit for opCtx if the current read source
@@ -1178,8 +1180,6 @@ getCollectionForLockFreeRead(OperationContext* opCtx,
     // places a compatible PIT collection reference in the 'catalog' if needed and the collection
     // exists at that PIT.
     const Collection* coll = catalog->establishConsistentCollection(opCtx, nsOrUUID, readTimestamp);
-    // TODO (SERVER-71222): This is broken if the UUID doesn't exist in the latest catalog.
-    //
     // Note: This call to resolveNamespaceStringOrUUID must happen after getCollectionFromCatalog
     // above, since getCollectionFromCatalog may call openCollection, which could change the result
     // of namespace resolution.
diff --git a/src/mongo/db/storage/durable_catalog.h b/src/mongo/db/storage/durable_catalog.h
index 6f40ef4b501..d44d8f3a645 100644
--- a/src/mongo/db/storage/durable_catalog.h
+++ b/src/mongo/db/storage/durable_catalog.h
@@ -109,6 +109,12 @@ public:
     virtual boost::optional<DurableCatalogEntry> scanForCatalogEntryByNss(
         OperationContext* opCtx, const NamespaceString& nss) const = 0;
 
+    /**
+     * Scans the persisted catalog until an entry is found matching 'uuid'.
+     */
+    virtual boost::optional<DurableCatalogEntry> scanForCatalogEntryByUUID(
+        OperationContext* opCtx, const UUID& uuid) const = 0;
+
     virtual EntryIdentifier getEntry(const RecordId& catalogId) const = 0;
 
     virtual std::string getIndexIdent(OperationContext* opCtx,
diff --git a/src/mongo/db/storage/durable_catalog_impl.cpp b/src/mongo/db/storage/durable_catalog_impl.cpp
index ca7798a578b..8971208e6b9 100644
--- a/src/mongo/db/storage/durable_catalog_impl.cpp
+++ b/src/mongo/db/storage/durable_catalog_impl.cpp
@@ -294,6 +294,30 @@ boost::optional<DurableCatalogEntry> DurableCatalogImpl::scanForCatalogEntryByNs
     return boost::none;
 }
 
+boost::optional<DurableCatalogEntry> DurableCatalogImpl::scanForCatalogEntryByUUID(
+    OperationContext* opCtx, const UUID& uuid) const {
+    auto cursor = _rs->getCursor(opCtx);
+    while (auto record = cursor->next()) {
+        BSONObj obj = record->data.releaseToBson();
+
+        if (isFeatureDocument(obj)) {
+            // Skip over the version document because it doesn't correspond to a collection.
+            continue;
+        }
+
+        std::shared_ptr<BSONCollectionCatalogEntry::MetaData> md = _parseMetaData(obj["md"]);
+        if (md->options.uuid == uuid) {
+            BSONElement idxIdent = obj["idxIdent"];
+            return DurableCatalogEntry{record->id,
+                                       obj["ident"].String(),
+                                       idxIdent.eoo() ? BSONObj() : idxIdent.Obj().getOwned(),
+                                       md};
+        }
+    }
+
+    return boost::none;
+}
+
 DurableCatalog::EntryIdentifier DurableCatalogImpl::getEntry(const RecordId& catalogId) const {
     stdx::lock_guard<Latch> lk(_catalogIdToEntryMapLock);
     auto it = _catalogIdToEntryMap.find(catalogId);
diff --git a/src/mongo/db/storage/durable_catalog_impl.h b/src/mongo/db/storage/durable_catalog_impl.h
index b4295e0a213..8cd9d32d07e 100644
--- a/src/mongo/db/storage/durable_catalog_impl.h
+++ b/src/mongo/db/storage/durable_catalog_impl.h
@@ -67,6 +67,9 @@ public:
     boost::optional<DurableCatalogEntry> scanForCatalogEntryByNss(OperationContext* opCtx,
                                                                   const NamespaceString& nss) const;
 
+    boost::optional<DurableCatalogEntry> scanForCatalogEntryByUUID(OperationContext* opCtx,
+                                                                   const UUID& uuid) const;
+
     EntryIdentifier getEntry(const RecordId& catalogId) const;
 
     std::string getCollectionIdent(const RecordId& catalogId) const;