1 files changed, 193 insertions, 212 deletions

diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index cdaef1c8e55..4e5df27b592 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -52,13 +52,10 @@ const ServiceContext::Decoration<LatestCollectionCatalog> getCatalog =
     ServiceContext::declareDecoration<LatestCollectionCatalog>();
 
 std::shared_ptr<CollectionCatalog> batchedCatalogWriteInstance;
-absl::flat_hash_set<Collection*> batchedCatalogClonedCollections;
 
 const OperationContext::Decoration<std::shared_ptr<const CollectionCatalog>> stashedCatalog =
     OperationContext::declareDecoration<std::shared_ptr<const CollectionCatalog>>();
 
-const auto maxUuid = UUID::parse("FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF").getValue();
-const auto minUuid = UUID::parse("00000000-0000-0000-0000-000000000000").getValue();
 }  // namespace
 
 class IgnoreExternalViewChangesForDatabase {
@@ -94,12 +91,12 @@ public:
 
     static void setCollectionInCatalog(CollectionCatalog& catalog,
                                        std::shared_ptr<Collection> collection) {
-        catalog._collections = catalog._collections.set(collection->ns(), collection);
-        catalog._catalog = catalog._catalog.set(collection->uuid(), collection);
+        catalog._collections[collection->ns()] = collection;
+        catalog._catalog[collection->uuid()] = collection;
         // TODO SERVER-64608 Use tenantID from ns
         auto dbIdPair = std::make_pair(TenantDatabaseName(boost::none, collection->ns().db()),
                                        collection->uuid());
-        catalog._orderedCollections = catalog._orderedCollections.set(dbIdPair, collection);
+        catalog._orderedCollections[dbIdPair] = collection;
     }
 
     PublishCatalogUpdates(OperationContext* opCtx,
@@ -133,7 +130,7 @@ public:
                 case UncommittedCatalogUpdates::Entry::Action::kRenamedCollection: {
                     writeJobs.push_back(
                         [& from = entry.nss, &to = entry.renameTo](CollectionCatalog& catalog) {
-                            catalog._collections = catalog._collections.erase(from);
+                            catalog._collections.erase(from);
 
                             auto fromStr = from.ns();
                             auto toStr = to.ns();
@@ -154,9 +151,10 @@ public:
                     break;
                 }
                 case UncommittedCatalogUpdates::Entry::Action::kRecreatedCollection: {
-                    writeJobs.push_back([opCtx = _opCtx, collection = entry.collection](
-                                            CollectionCatalog& catalog) {
-                        catalog.registerCollection(opCtx, std::move(collection));
+                    writeJobs.push_back([opCtx = _opCtx,
+                                         collection = entry.collection,
+                                         uuid = *entry.externalUUID](CollectionCatalog& catalog) {
+                        catalog.registerCollection(opCtx, uuid, std::move(collection));
                     });
                     // Fallthrough to the createCollection case to finish committing the collection.
                 }
@@ -223,65 +221,90 @@ private:
     UncommittedCatalogUpdates& _uncommittedCatalogUpdates;
 };
 
-CollectionCatalog::iterator::iterator(const TenantDatabaseName& tenantDbName,
-                                      OrderedCollectionMap::iterator it,
-                                      const OrderedCollectionMap& map)
-    : _map{map}, _mapIter{it}, _end(_map.upper_bound(std::make_pair(tenantDbName, maxUuid))) {
-    _skipUncommitted();
+CollectionCatalog::iterator::iterator(OperationContext* opCtx,
+                                      const TenantDatabaseName& tenantDbName,
+                                      const CollectionCatalog& catalog)
+    : _opCtx(opCtx), _tenantDbName(tenantDbName), _catalog(&catalog) {
+    auto minUuid = UUID::parse("00000000-0000-0000-0000-000000000000").getValue();
+
+    _mapIter = _catalog->_orderedCollections.lower_bound(std::make_pair(_tenantDbName, minUuid));
+
+    // Start with the first collection that is visible outside of its transaction.
+    while (!_exhausted() && !_mapIter->second->isCommitted()) {
+        _mapIter++;
+    }
+
+    if (!_exhausted()) {
+        _uuid = _mapIter->first.second;
+    }
 }
 
+CollectionCatalog::iterator::iterator(OperationContext* opCtx,
+                                      std::map<std::pair<TenantDatabaseName, UUID>,
+                                               std::shared_ptr<Collection>>::const_iterator mapIter,
+                                      const CollectionCatalog& catalog)
+    : _opCtx(opCtx), _mapIter(mapIter), _catalog(&catalog) {}
+
 CollectionCatalog::iterator::value_type CollectionCatalog::iterator::operator*() {
-    if (_mapIter == _map.end()) {
-        return nullptr;
+    if (_exhausted()) {
+        return CollectionPtr();
     }
-    return _mapIter->second.get();
+
+    return {
+        _opCtx, _mapIter->second.get(), LookupCollectionForYieldRestore(_mapIter->second->ns())};
+}
+
+Collection* CollectionCatalog::iterator::getWritableCollection(OperationContext* opCtx) {
+    return CollectionCatalog::get(opCtx)->lookupCollectionByUUIDForMetadataWrite(
+        opCtx, operator*()->uuid());
+}
+
+boost::optional<UUID> CollectionCatalog::iterator::uuid() {
+    return _uuid;
 }
 
 CollectionCatalog::iterator CollectionCatalog::iterator::operator++() {
-    invariant(_mapIter != _map.end());
-    invariant(_mapIter != _end);
     _mapIter++;
-    _skipUncommitted();
-    return *this;
-}
 
-bool CollectionCatalog::iterator::operator==(const iterator& other) const {
-    invariant(_map == other._map);
+    // Skip any collections that are not yet visible outside of their respective transactions.
+    while (!_exhausted() && !_mapIter->second->isCommitted()) {
+        _mapIter++;
+    }
 
-    if (other._mapIter == other._map.end()) {
-        return _mapIter == _map.end();
-    } else if (_mapIter == _map.end()) {
-        return other._mapIter == other._map.end();
+    if (_exhausted()) {
+        // If the iterator is at the end of the map or now points to an entry that does not
+        // correspond to the correct database.
+        _mapIter = _catalog->_orderedCollections.end();
+        _uuid = boost::none;
+        return *this;
     }
 
-    return _mapIter->first.second == other._mapIter->first.second;
+    _uuid = _mapIter->first.second;
+    return *this;
 }
 
-bool CollectionCatalog::iterator::operator!=(const iterator& other) const {
-    return !(*this == other);
+CollectionCatalog::iterator CollectionCatalog::iterator::operator++(int) {
+    auto oldPosition = *this;
+    ++(*this);
+    return oldPosition;
 }
 
-void CollectionCatalog::iterator::_skipUncommitted() {
-    // Advance to the next collection that is visible outside of its transaction.
-    while (_mapIter != _end && !_mapIter->second->isCommitted()) {
-        ++_mapIter;
+bool CollectionCatalog::iterator::operator==(const iterator& other) const {
+    invariant(_catalog == other._catalog);
+    if (other._mapIter == _catalog->_orderedCollections.end()) {
+        return _uuid == boost::none;
     }
-}
 
-CollectionCatalog::Range::Range(const OrderedCollectionMap& map,
-                                const TenantDatabaseName& tenantDbName)
-    : _map{map}, _tenantDbName{tenantDbName} {}
-
-CollectionCatalog::iterator CollectionCatalog::Range::begin() const {
-    return {_tenantDbName, _map.lower_bound(std::make_pair(_tenantDbName, minUuid)), _map};
+    return _uuid == other._uuid;
 }
 
-CollectionCatalog::iterator CollectionCatalog::Range::end() const {
-    return {_tenantDbName, _map.upper_bound(std::make_pair(_tenantDbName, maxUuid)), _map};
+bool CollectionCatalog::iterator::operator!=(const iterator& other) const {
+    return !(*this == other);
 }
 
-bool CollectionCatalog::Range::empty() const {
-    return begin() == end();
+bool CollectionCatalog::iterator::_exhausted() {
+    return _mapIter == _catalog->_orderedCollections.end() ||
+        _mapIter->first.first != _tenantDbName;
 }
 
 std::shared_ptr<const CollectionCatalog> CollectionCatalog::get(ServiceContext* svcCtx) {
@@ -435,13 +458,6 @@ void CollectionCatalog::write(ServiceContext* svcCtx, CatalogWriteFn job) {
 
 void CollectionCatalog::write(OperationContext* opCtx,
                               std::function<void(CollectionCatalog&)> job) {
-    // Calling the writer must be done with the GlobalLock held. Otherwise we risk having the
-    // BatchedCollectionCatalogWriter and this caller concurrently modifying the catalog. This is
-    // because normal operations calling this will all be serialized, but
-    // BatchedCollectionCatalogWriter skips this mechanism as it knows it is the sole user of the
-    // server by holding a Global MODE_X lock.
-    invariant(opCtx->lockState()->isNoop() || opCtx->lockState()->isLocked());
-
     // If global MODE_X lock are held we can re-use a cloned CollectionCatalog instance when
     // 'batchedCatalogWriteInstance' is set. Make sure we are the one holding the write lock.
     if (batchedCatalogWriteInstance) {
@@ -453,33 +469,25 @@ void CollectionCatalog::write(OperationContext* opCtx,
     write(opCtx->getServiceContext(), std::move(job));
 }
 
-Status CollectionCatalog::createView(OperationContext* opCtx,
-                                     const NamespaceString& viewName,
-                                     const NamespaceString& viewOn,
-                                     const BSONArray& pipeline,
-                                     const BSONObj& collation,
-                                     const ViewsForDatabase::PipelineValidatorFn& pipelineValidator,
-                                     const ViewUpsertMode insertViewMode) const {
-    // A view document direct write can occur via the oplog application path, which may only hold a
-    // lock on the collection being updated (the database views collection).
-    invariant(insertViewMode == ViewUpsertMode::kAlreadyDurableView ||
-              opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
+Status CollectionCatalog::createView(
+    OperationContext* opCtx,
+    const NamespaceString& viewName,
+    const NamespaceString& viewOn,
+    const BSONArray& pipeline,
+    const BSONObj& collation,
+    const ViewsForDatabase::PipelineValidatorFn& pipelineValidator) const {
+    invariant(opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
     invariant(opCtx->lockState()->isCollectionLockedForMode(
         NamespaceString(viewName.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
 
-    invariant(_viewsForDatabase.find(viewName.db()));
+    invariant(_viewsForDatabase.contains(viewName.db()));
     const ViewsForDatabase& viewsForDb = *_getViewsForDatabase(opCtx, viewName.db());
 
-    auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
-    if (uncommittedCatalogUpdates.shouldIgnoreExternalViewChanges(viewName.db())) {
-        return Status::OK();
-    }
-
     if (viewName.db() != viewOn.db())
         return Status(ErrorCodes::BadValue,
                       "View must be created on a view or collection in the same database");
 
-    if (viewsForDb.lookup(viewName) || _collections.find(viewName))
+    if (viewsForDb.lookup(viewName) || _collections.contains(viewName))
         return Status(ErrorCodes::NamespaceExists, "Namespace already exists");
 
     if (!NamespaceString::validCollectionName(viewOn.coll()))
@@ -500,8 +508,7 @@ Status CollectionCatalog::createView(OperationContext* opCtx,
                                      pipeline,
                                      pipelineValidator,
                                      std::move(collator.getValue()),
-                                     ViewsForDatabase{viewsForDb},
-                                     insertViewMode);
+                                     ViewsForDatabase{viewsForDb});
     }
 
     return result;
@@ -517,7 +524,7 @@ Status CollectionCatalog::modifyView(
     invariant(opCtx->lockState()->isCollectionLockedForMode(
         NamespaceString(viewName.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
 
-    invariant(_viewsForDatabase.find(viewName.db()));
+    invariant(_viewsForDatabase.contains(viewName.db()));
     const ViewsForDatabase& viewsForDb = *_getViewsForDatabase(opCtx, viewName.db());
 
     if (viewName.db() != viewOn.db())
@@ -543,8 +550,7 @@ Status CollectionCatalog::modifyView(
                                      pipeline,
                                      pipelineValidator,
                                      CollatorInterface::cloneCollator(viewPtr->defaultCollator()),
-                                     ViewsForDatabase{viewsForDb},
-                                     ViewUpsertMode::kUpdateView);
+                                     ViewsForDatabase{viewsForDb});
     }
 
     return result;
@@ -555,12 +561,12 @@ Status CollectionCatalog::dropView(OperationContext* opCtx, const NamespaceStrin
     invariant(opCtx->lockState()->isCollectionLockedForMode(
         NamespaceString(viewName.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
 
-    invariant(_viewsForDatabase.find(viewName.db()));
+    invariant(_viewsForDatabase.contains(viewName.db()));
     const ViewsForDatabase& viewsForDb = *_getViewsForDatabase(opCtx, viewName.db());
     viewsForDb.requireValidCatalog();
 
     // Make sure the view exists before proceeding.
-    if (!viewsForDb.lookup(viewName)) {
+    if (auto viewPtr = viewsForDb.lookup(viewName); !viewPtr) {
         return {ErrorCodes::NamespaceNotFound,
                 str::stream() << "cannot drop missing view: " << viewName.ns()};
     }
@@ -604,10 +610,9 @@ Status CollectionCatalog::reloadViews(OperationContext* opCtx, StringData dbName
 
     // Create a copy of the ViewsForDatabase instance to modify it. Reset the views for this
     // database, but preserve the DurableViewCatalog pointer.
-    const ViewsForDatabase* viewsForDbPtr = _viewsForDatabase.find(dbName);
-    invariant(viewsForDbPtr);
-    ViewsForDatabase viewsForDb = *viewsForDbPtr;
-
+    auto it = _viewsForDatabase.find(dbName);
+    invariant(it != _viewsForDatabase.end());
+    ViewsForDatabase viewsForDb{it->second.durable};
     viewsForDb.valid = false;
     viewsForDb.viewGraphNeedsRefresh = true;
     viewsForDb.viewMap.clear();
@@ -672,29 +677,28 @@ void CollectionCatalog::onOpenDatabase(OperationContext* opCtx,
     invariant(opCtx->lockState()->isDbLockedForMode(dbName, MODE_IS));
     uassert(ErrorCodes::AlreadyInitialized,
             str::stream() << "Database " << dbName << " is already initialized",
-            !_viewsForDatabase.find(dbName));
+            _viewsForDatabase.find(dbName) == _viewsForDatabase.end());
 
-    _viewsForDatabase = _viewsForDatabase.set(dbName.toString(), std::move(viewsForDb));
+    _viewsForDatabase[dbName] = std::move(viewsForDb);
 }
 
 void CollectionCatalog::onCloseDatabase(OperationContext* opCtx, TenantDatabaseName tenantDbName) {
     invariant(opCtx->lockState()->isDbLockedForMode(tenantDbName.dbName(), MODE_X));
     auto rid = ResourceId(RESOURCE_DATABASE, tenantDbName.dbName());
     removeResource(rid, tenantDbName.dbName());
-    _viewsForDatabase = _viewsForDatabase.erase(tenantDbName.dbName());
+    _viewsForDatabase.erase(tenantDbName.dbName());
 }
 
-void CollectionCatalog::onCloseCatalog() {
-    if (_shadowCatalog) {
-        return;
-    }
-
+void CollectionCatalog::onCloseCatalog(OperationContext* opCtx) {
+    invariant(opCtx->lockState()->isW());
+    invariant(!_shadowCatalog);
     _shadowCatalog.emplace();
     for (auto& entry : _catalog)
-        _shadowCatalog = _shadowCatalog->insert({entry.first, entry.second->ns()});
+        _shadowCatalog->insert({entry.first, entry.second->ns()});
 }
 
-void CollectionCatalog::onOpenCatalog() {
+void CollectionCatalog::onOpenCatalog(OperationContext* opCtx) {
+    invariant(opCtx->lockState()->isW());
     invariant(_shadowCatalog);
     _shadowCatalog.reset();
     ++_epoch;
@@ -704,10 +708,6 @@ uint64_t CollectionCatalog::getEpoch() const {
     return _epoch;
 }
 
-CollectionCatalog::Range CollectionCatalog::range(const TenantDatabaseName& tenantDbName) const {
-    return {_orderedCollections, tenantDbName};
-}
-
 std::shared_ptr<const Collection> CollectionCatalog::lookupCollectionByUUIDForRead(
     OperationContext* opCtx, const UUID& uuid) const {
     auto [found, uncommittedColl, newColl] =
@@ -761,7 +761,6 @@ Collection* CollectionCatalog::lookupCollectionByUUIDForMetadataWrite(OperationC
     // on the thread doing the batch write and it would trigger the regular path where we do a
     // copy-on-write on the catalog when committing.
     if (_isCatalogBatchWriter()) {
-        batchedCatalogClonedCollections.emplace(cloned.get());
         PublishCatalogUpdates::setCollectionInCatalog(*batchedCatalogWriteInstance,
                                                       std::move(cloned));
         return ptr;
@@ -795,8 +794,8 @@ bool CollectionCatalog::isCollectionAwaitingVisibility(UUID uuid) const {
 }
 
 std::shared_ptr<Collection> CollectionCatalog::_lookupCollectionByUUID(UUID uuid) const {
-    const std::shared_ptr<Collection>* coll = _catalog.find(uuid);
-    return coll ? *coll : nullptr;
+    auto foundIt = _catalog.find(uuid);
+    return foundIt == _catalog.end() ? nullptr : foundIt->second;
 }
 
 std::shared_ptr<const Collection> CollectionCatalog::lookupCollectionByNamespaceForRead(
@@ -813,8 +812,8 @@ std::shared_ptr<const Collection> CollectionCatalog::lookupCollectionByNamespace
         return nullptr;
     }
 
-    const std::shared_ptr<Collection>* collPtr = _collections.find(nss);
-    auto coll = collPtr ? *collPtr : nullptr;
+    auto it = _collections.find(nss);
+    auto coll = (it == _collections.end() ? nullptr : it->second);
     return (coll && coll->isCommitted()) ? coll : nullptr;
 }
 
@@ -844,8 +843,8 @@ Collection* CollectionCatalog::lookupCollectionByNamespaceForMetadataWrite(
         return nullptr;
     }
 
-    const std::shared_ptr<Collection>* collPtr = _collections.find(nss);
-    auto coll = collPtr ? *collPtr : nullptr;
+    auto it = _collections.find(nss);
+    auto coll = (it == _collections.end() ? nullptr : it->second);
 
     if (!coll || !coll->isCommitted())
         return nullptr;
@@ -865,7 +864,6 @@ Collection* CollectionCatalog::lookupCollectionByNamespaceForMetadataWrite(
     // on the thread doing the batch write and it would trigger the regular path where we do a
     // copy-on-write on the catalog when committing.
     if (_isCatalogBatchWriter()) {
-        batchedCatalogClonedCollections.emplace(cloned.get());
         PublishCatalogUpdates::setCollectionInCatalog(*batchedCatalogWriteInstance,
                                                       std::move(cloned));
         return ptr;
@@ -892,8 +890,8 @@ CollectionPtr CollectionCatalog::lookupCollectionByNamespace(OperationContext* o
         return nullptr;
     }
 
-    const std::shared_ptr<Collection>* collPtr = _collections.find(nss);
-    auto coll = collPtr ? *collPtr : nullptr;
+    auto it = _collections.find(nss);
+    auto coll = (it == _collections.end() ? nullptr : it->second);
     return (coll && coll->isCommitted())
         ? CollectionPtr(opCtx, coll.get(), LookupCollectionForYieldRestore(coll->ns()))
         : nullptr;
@@ -911,21 +909,20 @@ boost::optional<NamespaceString> CollectionCatalog::lookupNSSByUUID(OperationCon
         return boost::none;
     }
 
-    const std::shared_ptr<Collection>* collPtr = _catalog.find(uuid);
-    if (collPtr) {
-        auto coll = *collPtr;
-        boost::optional<NamespaceString> ns = coll->ns();
-        invariant(!ns.value().isEmpty());
-        return coll->isCommitted() ? ns : boost::none;
+    auto foundIt = _catalog.find(uuid);
+    if (foundIt != _catalog.end()) {
+        boost::optional<NamespaceString> ns = foundIt->second->ns();
+        invariant(!ns.get().isEmpty());
+        return _collections.find(ns.get())->second->isCommitted() ? ns : boost::none;
     }
 
     // Only in the case that the catalog is closed and a UUID is currently unknown, resolve it
     // using the pre-close state. This ensures that any tasks reloading the catalog can see their
     // own updates.
     if (_shadowCatalog) {
-        auto* shadowIt = _shadowCatalog->find(uuid);
-        if (shadowIt)
-            return *shadowIt;
+        auto shadowIt = _shadowCatalog->find(uuid);
+        if (shadowIt != _shadowCatalog->end())
+            return shadowIt->second;
     }
     return boost::none;
 }
@@ -941,11 +938,10 @@ boost::optional<UUID> CollectionCatalog::lookupUUIDByNSS(OperationContext* opCtx
         return boost::none;
     }
 
-    const std::shared_ptr<Collection>* collPtr = _collections.find(nss);
-    if (collPtr) {
-        auto coll = *collPtr;
-        const boost::optional<UUID>& uuid = coll->uuid();
-        return coll->isCommitted() ? uuid : boost::none;
+    auto it = _collections.find(nss);
+    if (it != _collections.end()) {
+        const boost::optional<UUID>& uuid = it->second->uuid();
+        return it->second->isCommitted() ? uuid : boost::none;
     }
     return boost::none;
 }
@@ -1090,34 +1086,23 @@ std::vector<TenantDatabaseName> CollectionCatalog::getAllDbNames() const {
     return ret;
 }
 
-void CollectionCatalog::setAllDatabaseProfileFilters(std::shared_ptr<ProfileFilter> filter) {
-    auto dbProfileSettingsWriter = _databaseProfileSettings.transient();
-    for (const auto& [dbName, settings] : _databaseProfileSettings) {
-        ProfileSettings clone = settings;
-        clone.filter = filter;
-        dbProfileSettingsWriter.set(dbName, std::move(clone));
-    }
-    _databaseProfileSettings = dbProfileSettingsWriter.persistent();
-}
-
 void CollectionCatalog::setDatabaseProfileSettings(
     StringData dbName, CollectionCatalog::ProfileSettings newProfileSettings) {
-    _databaseProfileSettings =
-        _databaseProfileSettings.set(dbName.toString(), std::move(newProfileSettings));
+    _databaseProfileSettings[dbName] = newProfileSettings;
 }
 
 CollectionCatalog::ProfileSettings CollectionCatalog::getDatabaseProfileSettings(
     StringData dbName) const {
-    const ProfileSettings* settings = _databaseProfileSettings.find(dbName);
-    if (settings) {
-        return *settings;
+    auto it = _databaseProfileSettings.find(dbName);
+    if (it != _databaseProfileSettings.end()) {
+        return it->second;
     }
 
     return {serverGlobalParams.defaultProfile, ProfileFilter::getDefault()};
 }
 
 void CollectionCatalog::clearDatabaseProfileSettings(StringData dbName) {
-    _databaseProfileSettings = _databaseProfileSettings.erase(dbName.toString());
+    _databaseProfileSettings.erase(dbName);
 }
 
 CollectionCatalog::Stats CollectionCatalog::getStats() const {
@@ -1145,9 +1130,9 @@ CollectionCatalog::ViewCatalogSet CollectionCatalog::getViewCatalogDbNames(
 }
 
 void CollectionCatalog::registerCollection(OperationContext* opCtx,
+                                           const UUID& uuid,
                                            std::shared_ptr<Collection> coll) {
     auto nss = coll->ns();
-    auto uuid = coll->uuid();
     // TODO SERVER-64608 Use tenantId from nss
     auto tenantDbName = TenantDatabaseName(boost::none, nss.db());
     _ensureNamespaceDoesNotExist(opCtx, nss, NamespaceType::kAll);
@@ -1162,12 +1147,12 @@ void CollectionCatalog::registerCollection(OperationContext* opCtx,
     auto dbIdPair = std::make_pair(tenantDbName, uuid);
 
     // Make sure no entry related to this uuid.
-    invariant(!_catalog.find(uuid));
+    invariant(_catalog.find(uuid) == _catalog.end());
     invariant(_orderedCollections.find(dbIdPair) == _orderedCollections.end());
 
-    _catalog = _catalog.set(uuid, coll);
-    _collections = _collections.set(nss, coll);
-    _orderedCollections = _orderedCollections.set(dbIdPair, coll);
+    _catalog[uuid] = coll;
+    _collections[nss] = coll;
+    _orderedCollections[dbIdPair] = coll;
 
     if (!nss.isOnInternalDb() && !nss.isSystem()) {
         _stats.userCollections += 1;
@@ -1193,7 +1178,7 @@ void CollectionCatalog::registerCollection(OperationContext* opCtx,
 
 std::shared_ptr<Collection> CollectionCatalog::deregisterCollection(OperationContext* opCtx,
                                                                     const UUID& uuid) {
-    invariant(_catalog.find(uuid));
+    invariant(_catalog.find(uuid) != _catalog.end());
 
     auto coll = std::move(_catalog[uuid]);
     auto ns = coll->ns();
@@ -1204,12 +1189,12 @@ std::shared_ptr<Collection> CollectionCatalog::deregisterCollection(OperationCon
     LOGV2_DEBUG(20281, 1, "Deregistering collection", logAttrs(ns), "uuid"_attr = uuid);
 
     // Make sure collection object exists.
-    invariant(_collections.find(ns));
+    invariant(_collections.find(ns) != _collections.end());
     invariant(_orderedCollections.find(dbIdPair) != _orderedCollections.end());
 
-    _orderedCollections = _orderedCollections.erase(dbIdPair);
-    _collections = _collections.erase(ns);
-    _catalog = _catalog.erase(uuid);
+    _orderedCollections.erase(dbIdPair);
+    _collections.erase(ns);
+    _catalog.erase(uuid);
 
     if (!ns.isOnInternalDb() && !ns.isSystem()) {
         _stats.userCollections -= 1;
@@ -1242,18 +1227,18 @@ void CollectionCatalog::registerUncommittedView(OperationContext* opCtx,
     // namespaces here.
     _ensureNamespaceDoesNotExist(opCtx, nss, NamespaceType::kCollection);
 
-    _uncommittedViews = _uncommittedViews.insert(nss);
+    _uncommittedViews.emplace(nss);
 }
 
 void CollectionCatalog::deregisterUncommittedView(const NamespaceString& nss) {
-    _uncommittedViews = _uncommittedViews.erase(nss);
+    _uncommittedViews.erase(nss);
 }
 
 void CollectionCatalog::_ensureNamespaceDoesNotExist(OperationContext* opCtx,
                                                      const NamespaceString& nss,
                                                      NamespaceType type) const {
     auto existingCollection = _collections.find(nss);
-    if (existingCollection) {
+    if (existingCollection != _collections.end()) {
         LOGV2(5725001,
               "Conflicted registering namespace, already have a collection with the same namespace",
               "nss"_attr = nss);
@@ -1261,7 +1246,7 @@ void CollectionCatalog::_ensureNamespaceDoesNotExist(OperationContext* opCtx,
     }
 
     if (type == NamespaceType::kAll) {
-        if (_uncommittedViews.find(nss)) {
+        if (_uncommittedViews.contains(nss)) {
             LOGV2(5725002,
                   "Conflicted registering namespace, already have a view with the same namespace",
                   "nss"_attr = nss);
@@ -1289,25 +1274,26 @@ void CollectionCatalog::deregisterAllCollectionsAndViews() {
         auto ns = entry.second->ns();
 
         LOGV2_DEBUG(20283, 1, "Deregistering collection", logAttrs(ns), "uuid"_attr = uuid);
+
+        entry.second.reset();
     }
 
-    _collections = {};
-    _orderedCollections = {};
-    _catalog = {};
-    _viewsForDatabase = {};
+    _collections.clear();
+    _orderedCollections.clear();
+    _catalog.clear();
+    _viewsForDatabase.clear();
     _stats = {};
 
-    _resourceInformation = {};
+    _resourceInformation.clear();
 }
 
 void CollectionCatalog::clearViews(OperationContext* opCtx, StringData dbName) const {
     invariant(opCtx->lockState()->isCollectionLockedForMode(
         NamespaceString(dbName, NamespaceString::kSystemDotViewsCollectionName), MODE_X));
 
-    const ViewsForDatabase* viewsForDbPtr = _viewsForDatabase.find(dbName);
-    invariant(viewsForDbPtr);
-
-    ViewsForDatabase viewsForDb = *viewsForDbPtr;
+    auto it = _viewsForDatabase.find(dbName);
+    invariant(it != _viewsForDatabase.end());
+    ViewsForDatabase viewsForDb = it->second;
 
     viewsForDb.viewMap.clear();
     viewsForDb.viewGraph.clear();
@@ -1318,6 +1304,16 @@ void CollectionCatalog::clearViews(OperationContext* opCtx, StringData dbName) c
         catalog._replaceViewsForDatabase(dbName, std::move(viewsForDb));
     });
 }
+
+CollectionCatalog::iterator CollectionCatalog::begin(OperationContext* opCtx,
+                                                     const TenantDatabaseName& tenantDbName) const {
+    return iterator(opCtx, tenantDbName, *this);
+}
+
+CollectionCatalog::iterator CollectionCatalog::end(OperationContext* opCtx) const {
+    return iterator(opCtx, _orderedCollections.end(), *this);
+}
+
 boost::optional<std::string> CollectionCatalog::lookupResourceName(const ResourceId& rid) const {
     invariant(rid.getType() == RESOURCE_DATABASE || rid.getType() == RESOURCE_COLLECTION);
 
@@ -1325,6 +1321,7 @@ boost::optional<std::string> CollectionCatalog::lookupResourceName(const Resourc
     if (search == _resourceInformation.end()) {
         return boost::none;
     }
+
     const std::set<std::string>& namespaces = search->second;
 
     // When there are multiple namespaces mapped to the same ResourceId, return boost::none as the
@@ -1344,14 +1341,12 @@ void CollectionCatalog::removeResource(const ResourceId& rid, const std::string&
         return;
     }
 
-    std::set<std::string> namespaces = search->second;
+    std::set<std::string>& namespaces = search->second;
     namespaces.erase(entry);
 
     // Remove the map entry if this is the last namespace in the set for the ResourceId.
     if (namespaces.size() == 0) {
-        _resourceInformation = _resourceInformation.erase(search, rid);
-    } else {
-        _resourceInformation = _resourceInformation.set(rid, std::move(namespaces));
+        _resourceInformation.erase(search);
     }
 }
 
@@ -1361,17 +1356,16 @@ void CollectionCatalog::addResource(const ResourceId& rid, const std::string& en
     auto search = _resourceInformation.find(rid);
     if (search == _resourceInformation.end()) {
         std::set<std::string> newSet = {entry};
-        _resourceInformation = _resourceInformation.set(rid, std::move(newSet));
+        _resourceInformation.insert(std::make_pair(rid, newSet));
         return;
     }
 
-    if (const auto& namespaces = search->second; namespaces.count(entry) > 0) {
+    std::set<std::string>& namespaces = search->second;
+    if (namespaces.count(entry) > 0) {
         return;
     }
 
-    std::set<std::string> namespaces = search->second;
     namespaces.insert(entry);
-    _resourceInformation = _resourceInformation.set(rid, std::move(namespaces));
 }
 
 void CollectionCatalog::invariantHasExclusiveAccessToCollection(OperationContext* opCtx,
@@ -1391,15 +1385,15 @@ boost::optional<const ViewsForDatabase&> CollectionCatalog::_getViewsForDatabase
         return uncommittedViews;
     }
 
-    const ViewsForDatabase* viewsForDb = _viewsForDatabase.find(dbName);
-    if (!viewsForDb) {
+    auto it = _viewsForDatabase.find(dbName);
+    if (it == _viewsForDatabase.end()) {
         return boost::none;
     }
-    return *viewsForDb;
+    return it->second;
 }
 
 void CollectionCatalog::_replaceViewsForDatabase(StringData dbName, ViewsForDatabase&& views) {
-    _viewsForDatabase = _viewsForDatabase.set(dbName.toString(), std::move(views));
+    _viewsForDatabase[dbName] = std::move(views);
 }
 
 Status CollectionCatalog::_createOrUpdateView(
@@ -1409,19 +1403,15 @@ Status CollectionCatalog::_createOrUpdateView(
     const BSONArray& pipeline,
     const ViewsForDatabase::PipelineValidatorFn& pipelineValidator,
     std::unique_ptr<CollatorInterface> collator,
-    ViewsForDatabase&& viewsForDb,
-    ViewUpsertMode insertViewMode) const {
-    // A view document direct write can occur via the oplog application path, which may only hold a
-    // lock on the collection being updated (the database views collection).
-    invariant(insertViewMode == ViewUpsertMode::kAlreadyDurableView ||
-              opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
+    ViewsForDatabase&& viewsForDb) const {
+    invariant(opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
     invariant(opCtx->lockState()->isCollectionLockedForMode(
         NamespaceString(viewName.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
 
     viewsForDb.requireValidCatalog();
 
-    // Build the BSON definition for this view to be saved in the durable view catalog and/or to
-    // insert in the viewMap. If the collation is empty, omit it from the definition altogether.
+    // Build the BSON definition for this view to be saved in the durable view catalog. If the
+    // collation is empty, omit it from the definition altogether.
     BSONObjBuilder viewDefBuilder;
     viewDefBuilder.append("_id", viewName.ns());
     viewDefBuilder.append("viewOn", viewOn.coll());
@@ -1430,42 +1420,25 @@ Status CollectionCatalog::_createOrUpdateView(
         viewDefBuilder.append("collation", collator->getSpec().toBSON());
     }
 
-    BSONObj viewDef = viewDefBuilder.obj();
     BSONObj ownedPipeline = pipeline.getOwned();
-    ViewDefinition view(
+    auto view = std::make_shared<ViewDefinition>(
         viewName.db(), viewName.coll(), viewOn.coll(), ownedPipeline, std::move(collator));
 
-    // If the view is already in the durable view catalog, we don't need to validate the graph. If
-    // we need to update the durable view catalog, we need to check that the resulting dependency
-    // graph is acyclic and within the maximum depth.
-    const bool viewGraphNeedsValidation = insertViewMode != ViewUpsertMode::kAlreadyDurableView;
-    Status graphStatus =
-        viewsForDb.upsertIntoGraph(opCtx, view, pipelineValidator, viewGraphNeedsValidation);
+    // Check that the resulting dependency graph is acyclic and within the maximum depth.
+    Status graphStatus = viewsForDb.upsertIntoGraph(opCtx, *(view.get()), pipelineValidator);
     if (!graphStatus.isOK()) {
         return graphStatus;
     }
 
-    if (insertViewMode != ViewUpsertMode::kAlreadyDurableView) {
-        viewsForDb.durable->upsert(opCtx, viewName, viewDef);
-    }
+    viewsForDb.durable->upsert(opCtx, viewName, viewDefBuilder.obj());
 
+    viewsForDb.viewMap.clear();
     viewsForDb.valid = false;
-    auto res = [&] {
-        switch (insertViewMode) {
-            case ViewUpsertMode::kCreateView:
-            case ViewUpsertMode::kAlreadyDurableView:
-                return viewsForDb.insert(opCtx, viewDef);
-            case ViewUpsertMode::kUpdateView:
-                viewsForDb.viewMap.clear();
-                viewsForDb.viewGraphNeedsRefresh = true;
-                viewsForDb.stats = {};
-
-                // Reload the view catalog with the changes applied.
-                return viewsForDb.reload(opCtx);
-        }
-        MONGO_UNREACHABLE;
-    }();
+    viewsForDb.viewGraphNeedsRefresh = true;
+    viewsForDb.stats = {};
 
+    // Reload the view catalog with the changes applied.
+    auto res = viewsForDb.reload(opCtx);
     if (res.isOK()) {
         auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
         uncommittedCatalogUpdates.addView(opCtx, viewName);
@@ -1484,9 +1457,16 @@ bool CollectionCatalog::_isCatalogBatchWriter() const {
 
 bool CollectionCatalog::_alreadyClonedForBatchedWriter(
     const std::shared_ptr<Collection>& collection) const {
-    // We may skip cloning the Collection instance if and only if have already cloned it for write
-    // use in this batch writer.
-    return _isCatalogBatchWriter() && batchedCatalogClonedCollections.contains(collection.get());
+    // We may skip cloning the Collection instance if and only if we are currently in a batched
+    // catalog write and all references to this Collection is owned by the cloned CollectionCatalog
+    // instance owned by the batch writer. i.e. the Collection is uniquely owned by the batch
+    // writer. When the batch writer initially clones the catalog, all collections will have a
+    // 'use_count' of at least kNumCollectionReferencesStored*2 (because there are at least 2
+    // catalog instances). To check for uniquely owned we need to check that the reference count is
+    // exactly kNumCollectionReferencesStored (owned by a single catalog) while also account for the
+    // instance that is extracted from the catalog and provided as a parameter to this function, we
+    // therefore need to add 1.
+    return _isCatalogBatchWriter() && collection.use_count() == kNumCollectionReferencesStored + 1;
 }
 
 CollectionCatalogStasher::CollectionCatalogStasher(OperationContext* opCtx)
@@ -1547,7 +1527,10 @@ const Collection* LookupCollectionForYieldRestore::operator()(OperationContext*
     // state. After a query yields its locks, the replication state may have changed, invalidating
     // our current choice of ReadSource. Using the same preconditions, change our ReadSource if
     // necessary.
-    SnapshotHelper::changeReadSourceIfNeeded(opCtx, collection->ns());
+    auto [newReadSource, _] = SnapshotHelper::shouldChangeReadSource(opCtx, collection->ns());
+    if (newReadSource) {
+        opCtx->recoveryUnit()->setTimestampReadSource(*newReadSource);
+    }
 
     return collection.get();
 }
@@ -1556,7 +1539,6 @@ BatchedCollectionCatalogWriter::BatchedCollectionCatalogWriter(OperationContext*
     : _opCtx(opCtx) {
     invariant(_opCtx->lockState()->isW());
     invariant(!batchedCatalogWriteInstance);
-    invariant(batchedCatalogClonedCollections.empty());
 
     auto& storage = getCatalog(_opCtx->getServiceContext());
     // hold onto base so if we need to delete it we can do it outside of the lock
@@ -1579,7 +1561,6 @@ BatchedCollectionCatalogWriter::~BatchedCollectionCatalogWriter() {
     // Clear out batched pointer so no more attempts of batching are made
     _batchedInstance = nullptr;
     batchedCatalogWriteInstance = nullptr;
-    batchedCatalogClonedCollections.clear();
 }
 
 }  // namespace mongo