SERVER-53307 Make ViewCatalog global

5 files changed, 269 insertions, 158 deletions

diff --git a/src/mongo/db/views/durable_view_catalog.cpp b/src/mongo/db/views/durable_view_catalog.cpp
index 49e4fd23fea..0aff1e47984 100644
--- a/src/mongo/db/views/durable_view_catalog.cpp
+++ b/src/mongo/db/views/durable_view_catalog.cpp
@@ -58,19 +58,17 @@ void DurableViewCatalog::onExternalChange(OperationContext* opCtx, const Namespa
     dassert(opCtx->lockState()->isDbLockedForMode(name.db(), MODE_IX));
     dassert(opCtx->lockState()->isCollectionLockedForMode(
         NamespaceString(name.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
-    auto databaseHolder = DatabaseHolder::get(opCtx);
-    auto db = databaseHolder->getDb(opCtx, name.db());
-    if (db) {
-        // On an external change, an invalid view definition can be detected when the view catalog
-        // is reloaded. This will prevent any further usage of the view catalog until the invalid
-        // view definitions are removed. We use kValidateDurableViews here to catch any invalid view
-        // definitions in the view catalog to make it unusable for subsequent callers.
-        if (ViewCatalog::shouldIgnoreExternalChange(opCtx, db, name)) {
-            return;
-        }
 
-        ViewCatalog::reload(opCtx, db, ViewCatalogLookupBehavior::kValidateDurableViews).ignore();
+    // On an external change, an invalid view definition can be detected when the view catalog
+    // is reloaded. This will prevent any further usage of the view catalog until the invalid
+    // view definitions are removed. We use kValidateDurableViews here to catch any invalid view
+    // definitions in the view catalog to make it unusable for subsequent callers.
+    if (ViewCatalog::shouldIgnoreExternalChange(opCtx, name)) {
+        return;
     }
+
+    ViewCatalog::reload(opCtx, name.db(), ViewCatalogLookupBehavior::kValidateDurableViews)
+        .ignore();
 }
 
 void DurableViewCatalog::onSystemViewsCollectionDrop(OperationContext* opCtx,
@@ -85,7 +83,7 @@ void DurableViewCatalog::onSystemViewsCollectionDrop(OperationContext* opCtx,
     if (db) {
         // If the 'system.views' collection is dropped, we need to clear the in-memory state of the
         // view catalog.
-        ViewCatalog::clear(opCtx, db);
+        ViewCatalog::clear(opCtx, name.db());
     }
 }
 
@@ -95,6 +93,10 @@ const std::string& DurableViewCatalogImpl::getName() const {
     return _db->name();
 }
 
+const bool DurableViewCatalogImpl::belongsTo(const Database* db) const {
+    return _db == db;
+}
+
 void DurableViewCatalogImpl::iterate(OperationContext* opCtx, Callback callback) {
     _iterate(opCtx, callback, ViewCatalogLookupBehavior::kValidateDurableViews);
 }
diff --git a/src/mongo/db/views/durable_view_catalog.h b/src/mongo/db/views/durable_view_catalog.h
index 62fd5db2c4d..b51146e013f 100644
--- a/src/mongo/db/views/durable_view_catalog.h
+++ b/src/mongo/db/views/durable_view_catalog.h
@@ -81,6 +81,7 @@ public:
                         const BSONObj& view) = 0;
     virtual void remove(OperationContext* opCtx, const NamespaceString& name) = 0;
     virtual const std::string& getName() const = 0;
+    virtual const bool belongsTo(const Database* db) const = 0;
     virtual ~DurableViewCatalog() = default;
 };
 
@@ -99,6 +100,7 @@ public:
     void upsert(OperationContext* opCtx, const NamespaceString& name, const BSONObj& view);
     void remove(OperationContext* opCtx, const NamespaceString& name);
     const std::string& getName() const;
+    const bool belongsTo(const Database* db) const;
 
 private:
     void _iterate(OperationContext* opCtx,
diff --git a/src/mongo/db/views/view_catalog.cpp b/src/mongo/db/views/view_catalog.cpp
index 500f062dd4e..9a93d7f9f25 100644
--- a/src/mongo/db/views/view_catalog.cpp
+++ b/src/mongo/db/views/view_catalog.cpp
@@ -68,10 +68,8 @@ namespace {
  */
 class ViewCatalogWriter {
 public:
-    ViewCatalogWriter(Mutex& mutex,
-                      std::shared_ptr<const ViewCatalog> instance,
-                      std::shared_ptr<ViewCatalog>* storage)
-        : _mutex(mutex), _read(std::move(instance)), _storage(storage) {}
+    ViewCatalogWriter(Mutex& mutex, std::shared_ptr<ViewCatalog>* storage)
+        : _mutex(mutex), _storage(storage) {}
 
     ViewCatalogWriter(ViewCatalogWriter&&) = delete;
     ViewCatalogWriter& operator=(ViewCatalogWriter&&) = delete;
@@ -80,13 +78,15 @@ public:
         if (_write)
             return _write.get();
 
-        return _read.get();
+        // TODO (SERVER-57250): This atomic_load will be deprecated in C++20
+        return atomic_load(_storage).get();
     }
 
     ViewCatalog* writable() {
         if (!_write) {
             _lock = stdx::unique_lock<Mutex>(_mutex);
-            _write = std::make_shared<ViewCatalog>(*_read);
+            // TODO (SERVER-57250): This atomic_load will be deprecated in C++20
+            _write = std::make_shared<ViewCatalog>(*atomic_load(_storage));
         }
         return _write.get();
     }
@@ -101,7 +101,6 @@ public:
 private:
     Mutex& _mutex;
     stdx::unique_lock<Mutex> _lock;
-    std::shared_ptr<const ViewCatalog> _read;
     std::shared_ptr<ViewCatalog> _write;
     std::shared_ptr<ViewCatalog>* _storage;
 };
@@ -124,27 +123,32 @@ public:
     }
 
     ViewCatalogWriter writer() const {
-        return ViewCatalogWriter(_mutex, get(), &_catalog);
+        return ViewCatalogWriter(_mutex, &_catalog);
     }
 
-    void setIgnoreExternalChange(bool value) const {
-        _ignoreExternalChange = value;
+    void setIgnoreExternalChange(StringData dbName, bool value) const {
+        stdx::lock_guard lk{_externalChangeMutex};
+        if (value) {
+            _ignoreExternalChange.emplace(dbName);
+        } else {
+            _ignoreExternalChange.erase(dbName);
+        }
     }
 
-    bool shouldIgnoreExternalChange() const {
-        return _ignoreExternalChange;
+    bool shouldIgnoreExternalChange(StringData dbName) const {
+        stdx::lock_guard lk{_externalChangeMutex};
+        auto it = _ignoreExternalChange.find(dbName);
+        return it != _ignoreExternalChange.end();
     }
 
 private:
-    mutable std::shared_ptr<ViewCatalog> _catalog;
-    mutable Mutex _mutex = MONGO_MAKE_LATCH("ViewCatalogStorage::mutex");  // Serializes writes
-
-    // This is safe to not be atomic because it is only accessed on the write path for the
-    // 'system.views' collection for this db. Modifications to this collection happens only through
-    // this class and we have it locked with MODE_X.
-    mutable bool _ignoreExternalChange = false;
-};
-auto getViewCatalog = Database::declareDecoration<ViewCatalogStorage>();
+    mutable std::shared_ptr<ViewCatalog> _catalog = std::make_shared<ViewCatalog>();
+    mutable Mutex _mutex = MONGO_MAKE_LATCH("ViewCatalogStorage::_mutex");  // Serializes writes
+    mutable Mutex _externalChangeMutex = MONGO_MAKE_LATCH(
+        "ViewCatalogStorage::_externalChangeMutex");  // Guards _ignoreExternalChange set
+    mutable StringSet _ignoreExternalChange;
+};  // namespace
+const auto getViewCatalog = ServiceContext::declareDecoration<ViewCatalogStorage>();
 
 StatusWith<std::unique_ptr<CollatorInterface>> parseCollator(OperationContext* opCtx,
                                                              BSONObj collationSpec) {
@@ -157,38 +161,68 @@ StatusWith<std::unique_ptr<CollatorInterface>> parseCollator(OperationContext* o
 }
 }  // namespace
 
-std::shared_ptr<const ViewCatalog> ViewCatalog::get(const Database* db) {
-    return getViewCatalog(db).get();
+std::shared_ptr<const ViewCatalog> ViewCatalog::get(ServiceContext* svcCtx) {
+    return getViewCatalog(svcCtx).get();
 }
 
-void ViewCatalog::set(Database* db, std::unique_ptr<ViewCatalog> catalog) {
-    getViewCatalog(db).set(std::move(catalog));
+std::shared_ptr<const ViewCatalog> ViewCatalog::get(OperationContext* opCtx) {
+    return get(opCtx->getServiceContext());
+}
+
+Status ViewCatalog::registerDatabase(OperationContext* opCtx,
+                                     StringData dbName,
+                                     std::unique_ptr<DurableViewCatalog> durable) {
+    auto catalog = getViewCatalog(opCtx->getServiceContext()).writer();
+    auto it = catalog.writable()->_viewsForDatabase.find(dbName);
+    if (it != catalog.writable()->_viewsForDatabase.end()) {
+        return {ErrorCodes::AlreadyInitialized, "ViewCatalog entry for database already set"};
+    }
+
+    auto& vfdb = catalog.writable()->_viewsForDatabase[dbName];
+    vfdb.durable = std::move(durable);
+    vfdb.valid = false;
+    vfdb.viewGraphNeedsRefresh = true;
+    catalog.commit();
+    return Status::OK();
+}
+
+void ViewCatalog::unregisterDatabase(OperationContext* opCtx, Database* db) {
+    auto catalog = getViewCatalog(opCtx->getServiceContext()).writer();
+    auto it = catalog.writable()->_viewsForDatabase.find(db->name());
+    if (it != catalog.writable()->_viewsForDatabase.end() && it->second.durable->belongsTo(db)) {
+        catalog.writable()->_viewsForDatabase.erase(it);
+        catalog.commit();
+    }
 }
 
 Status ViewCatalog::reload(OperationContext* opCtx,
-                           const Database* db,
+                           StringData dbName,
                            ViewCatalogLookupBehavior lookupBehavior) {
-    auto catalog = getViewCatalog(db).writer();
+    auto catalog = getViewCatalog(opCtx->getServiceContext()).writer();
     invariant(opCtx->lockState()->isCollectionLockedForMode(
-        NamespaceString(db->name(), NamespaceString::kSystemDotViewsCollectionName), MODE_IS));
-    auto result =
-        catalog.writable()->_reload(opCtx, ViewCatalogLookupBehavior::kValidateDurableViews, true);
+        NamespaceString(dbName, NamespaceString::kSystemDotViewsCollectionName), MODE_IS));
+    auto result = catalog.writable()->_reload(
+        opCtx, dbName, ViewCatalogLookupBehavior::kValidateDurableViews, true);
     catalog.commit();
     return result;
 }
 
 Status ViewCatalog::_reload(OperationContext* opCtx,
+                            StringData dbName,
                             ViewCatalogLookupBehavior lookupBehavior,
                             bool reloadForCollectionCatalog) {
-    const auto& dbName = _durable->getName();
     LOGV2_DEBUG(22546, 1, "Reloading view catalog for database", "db"_attr = dbName);
 
-    _viewMap.clear();
-    _valid = false;
-    _viewGraphNeedsRefresh = true;
-    _stats = {};
+    auto it = _viewsForDatabase.find(dbName);
+    invariant(it != _viewsForDatabase.end());
+    auto& vfdb = it->second;
+
+    vfdb.viewMap.clear();
+    vfdb.valid = false;
+    vfdb.viewGraphNeedsRefresh = true;
+    vfdb.stats = {};
 
-    absl::flat_hash_set<NamespaceString> viewsForDb;
+    absl::flat_hash_set<NamespaceString> viewNamesForDb;
 
     auto reloadCallback = [&](const BSONObj& view) -> Status {
         BSONObj collationSpec = view.hasField("collation") ? view["collation"].Obj() : BSONObj();
@@ -217,32 +251,33 @@ Status ViewCatalog::_reload(OperationContext* opCtx,
 
         if (!viewName.isOnInternalDb() && !viewName.isSystem()) {
             if (viewDef->timeseries()) {
-                _stats.userTimeseries += 1;
+                vfdb.stats.userTimeseries += 1;
             } else {
-                _stats.userViews += 1;
+                vfdb.stats.userViews += 1;
             }
         } else {
-            _stats.internal += 1;
+            vfdb.stats.internal += 1;
         }
 
-        _viewMap[viewName.ns()] = std::move(viewDef);
+        vfdb.viewMap[viewName.ns()] = std::move(viewDef);
         if (reloadForCollectionCatalog) {
-            viewsForDb.insert(viewName);
+            viewNamesForDb.insert(viewName);
         }
         return Status::OK();
     };
 
     try {
         if (lookupBehavior == ViewCatalogLookupBehavior::kValidateDurableViews) {
-            _durable->iterate(opCtx, reloadCallback);
+            vfdb.durable->iterate(opCtx, reloadCallback);
         } else if (lookupBehavior == ViewCatalogLookupBehavior::kAllowInvalidDurableViews) {
-            _durable->iterateIgnoreInvalidEntries(opCtx, reloadCallback);
+            vfdb.durable->iterateIgnoreInvalidEntries(opCtx, reloadCallback);
         } else {
             MONGO_UNREACHABLE;
         }
         if (reloadForCollectionCatalog) {
             CollectionCatalog::write(
-                opCtx, [&dbName, viewsForDb = std::move(viewsForDb)](CollectionCatalog& catalog) {
+                opCtx,
+                [&dbName, viewsForDb = std::move(viewNamesForDb)](CollectionCatalog& catalog) {
                     catalog.replaceViewsForDatabase(dbName, std::move(viewsForDb));
                 });
         }
@@ -250,20 +285,23 @@ Status ViewCatalog::_reload(OperationContext* opCtx,
         auto status = ex.toStatus();
         LOGV2(22547,
               "Could not load view catalog for database",
-              "db"_attr = _durable->getName(),
+              "db"_attr = vfdb.durable->getName(),
               "error"_attr = status);
         return status;
     }
 
-    _valid = true;
+    vfdb.valid = true;
     return Status::OK();
 }
 
-void ViewCatalog::clear(OperationContext* opCtx, const Database* db) {
-    auto catalog = getViewCatalog(db).writer();
+void ViewCatalog::clear(OperationContext* opCtx, StringData dbName) {
+    auto catalog = getViewCatalog(opCtx->getServiceContext()).writer();
+    auto it = catalog.writable()->_viewsForDatabase.find(dbName);
+    invariant(it != catalog.writable()->_viewsForDatabase.end());
+    auto& vfdb = it->second;
 
     // First, iterate through the views on this database and audit them before they are dropped.
-    for (auto&& view : catalog->_viewMap) {
+    for (auto&& view : vfdb.viewMap) {
         audit::logDropView(opCtx->getClient(),
                            (*view.second).name(),
                            (*view.second).viewOn().ns(),
@@ -271,33 +309,37 @@ void ViewCatalog::clear(OperationContext* opCtx, const Database* db) {
                            ErrorCodes::OK);
     }
 
-    catalog.writable()->_viewMap.clear();
-    catalog.writable()->_viewGraph.clear();
-    catalog.writable()->_valid = true;
-    catalog.writable()->_viewGraphNeedsRefresh = false;
-    catalog.writable()->_stats = {};
-    CollectionCatalog::write(opCtx, [db](CollectionCatalog& catalog) {
-        catalog.replaceViewsForDatabase(db->name(), {});
+    vfdb.viewMap.clear();
+    vfdb.viewGraph.clear();
+    vfdb.valid = true;
+    vfdb.viewGraphNeedsRefresh = false;
+    vfdb.stats = {};
+    CollectionCatalog::write(opCtx, [db = dbName.toString()](CollectionCatalog& catalog) {
+        catalog.replaceViewsForDatabase(db, {});
     });
     catalog.commit();
 }
 
-bool ViewCatalog::shouldIgnoreExternalChange(OperationContext* opCtx,
-                                             const Database* db,
-                                             const NamespaceString& name) {
-    return getViewCatalog(db).shouldIgnoreExternalChange();
+bool ViewCatalog::shouldIgnoreExternalChange(OperationContext* opCtx, const NamespaceString& name) {
+    return getViewCatalog(opCtx->getServiceContext()).shouldIgnoreExternalChange(name.db());
 }
 
-void ViewCatalog::_requireValidCatalog() const {
+void ViewCatalog::ViewsForDatabase::requireValidCatalog() const {
     uassert(ErrorCodes::InvalidViewDefinition,
             "Invalid view definition detected in the view catalog. Remove the invalid view "
             "manually to prevent disallowing any further usage of the view catalog.",
-            _valid);
+            valid);
 }
 
-void ViewCatalog::iterate(ViewIteratorCallback callback) const {
-    _requireValidCatalog();
-    for (auto&& view : _viewMap) {
+void ViewCatalog::iterate(StringData dbName, ViewIteratorCallback callback) const {
+    auto it = _viewsForDatabase.find(dbName);
+    if (it == _viewsForDatabase.end()) {
+        return;
+    }
+    auto& vfdb = it->second;
+
+    vfdb.requireValidCatalog();
+    for (auto&& view : vfdb.viewMap) {
         if (!callback(*view.second)) {
             break;
         }
@@ -314,7 +356,10 @@ Status ViewCatalog::_createOrUpdateView(OperationContext* opCtx,
     invariant(opCtx->lockState()->isCollectionLockedForMode(
         NamespaceString(viewName.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
 
-    _requireValidCatalog();
+    auto it = _viewsForDatabase.find(viewName.db());
+    invariant(it != _viewsForDatabase.end());
+    auto& vfdb = it->second;
+    vfdb.requireValidCatalog();
 
     // 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.
@@ -336,11 +381,12 @@ Status ViewCatalog::_createOrUpdateView(OperationContext* opCtx,
         return graphStatus;
     }
 
-    _durable->upsert(opCtx, viewName, viewDefBuilder.obj());
-    _viewMap[viewName.ns()] = view;
+    vfdb.durable->upsert(opCtx, viewName, viewDefBuilder.obj());
+    vfdb.viewMap[viewName.ns()] = view;
 
     // Reload the view catalog with the changes applied.
-    auto res = _reload(opCtx, ViewCatalogLookupBehavior::kValidateDurableViews, false);
+    auto res =
+        _reload(opCtx, viewName.db(), ViewCatalogLookupBehavior::kValidateDurableViews, false);
     if (res.isOK()) {
         // Register the view in the CollectionCatalog mapping from ResourceID->namespace
         auto viewRid = ResourceId(RESOURCE_COLLECTION, viewName.ns());
@@ -361,9 +407,13 @@ Status ViewCatalog::_createOrUpdateView(OperationContext* opCtx,
 }
 
 Status ViewCatalog::_upsertIntoGraph(OperationContext* opCtx, const ViewDefinition& viewDef) {
+    auto it = _viewsForDatabase.find(viewDef.name().db());
+    invariant(it != _viewsForDatabase.end());
+    auto& vfdb = it->second;
 
     // Performs the insert into the graph.
-    auto doInsert = [this, &opCtx](const ViewDefinition& viewDef, bool needsValidation) -> Status {
+    auto doInsert = [this, opCtx, &vfdb](const ViewDefinition& viewDef,
+                                         bool needsValidation) -> Status {
         // Validate that the pipeline is eligible to serve as a view definition. If it is, this
         // will also return the set of involved namespaces.
         auto pipelineStatus = validatePipeline(opCtx, viewDef);
@@ -391,16 +441,16 @@ Status ViewCatalog::_upsertIntoGraph(OperationContext* opCtx, const ViewDefiniti
             if (!collationStatus.isOK()) {
                 return collationStatus;
             }
-            return _viewGraph.insertAndValidate(viewDef, refs, pipelineSize);
+            return vfdb.viewGraph.insertAndValidate(viewDef, refs, pipelineSize);
         } else {
-            _viewGraph.insertWithoutValidating(viewDef, refs, pipelineSize);
+            vfdb.viewGraph.insertWithoutValidating(viewDef, refs, pipelineSize);
             return Status::OK();
         }
     };
 
-    if (_viewGraphNeedsRefresh) {
-        _viewGraph.clear();
-        for (auto&& iter : _viewMap) {
+    if (vfdb.viewGraphNeedsRefresh) {
+        vfdb.viewGraph.clear();
+        for (auto&& iter : vfdb.viewMap) {
             auto status = doInsert(*(iter.second.get()), false);
             // If we cannot fully refresh the graph, we will keep '_viewGraphNeedsRefresh' true.
             if (!status.isOK()) {
@@ -408,12 +458,12 @@ Status ViewCatalog::_upsertIntoGraph(OperationContext* opCtx, const ViewDefiniti
             }
         }
         // Only if the inserts completed without error will we no longer need a refresh.
-        _viewGraphNeedsRefresh = false;
+        vfdb.viewGraphNeedsRefresh = false;
     }
 
     // Remove the view definition first in case this is an update. If it is not in the graph, it
     // is simply a no-op.
-    _viewGraph.remove(viewDef.name());
+    vfdb.viewGraph.remove(viewDef.name());
 
     return doInsert(viewDef, true);
 }
@@ -522,7 +572,6 @@ Status ViewCatalog::_validateCollation(OperationContext* opCtx,
 }
 
 Status ViewCatalog::createView(OperationContext* opCtx,
-                               const Database* db,
                                const NamespaceString& viewName,
                                const NamespaceString& viewOn,
                                const BSONArray& pipeline,
@@ -532,7 +581,7 @@ Status ViewCatalog::createView(OperationContext* opCtx,
     invariant(opCtx->lockState()->isCollectionLockedForMode(
         NamespaceString(viewName.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
 
-    const auto& catalogStorage = getViewCatalog(db);
+    const auto& catalogStorage = getViewCatalog(opCtx->getServiceContext());
     auto catalog = catalogStorage.writer();
 
     if (viewName.db() != viewOn.db())
@@ -552,8 +601,10 @@ Status ViewCatalog::createView(OperationContext* opCtx,
 
     Status result = Status::OK();
     {
-        ON_BLOCK_EXIT([&catalogStorage] { catalogStorage.setIgnoreExternalChange(false); });
-        catalogStorage.setIgnoreExternalChange(true);
+        ON_BLOCK_EXIT([&catalogStorage, &viewName] {
+            catalogStorage.setIgnoreExternalChange(viewName.db(), false);
+        });
+        catalogStorage.setIgnoreExternalChange(viewName.db(), true);
 
         result = catalog.writable()->_createOrUpdateView(
             opCtx, viewName, viewOn, pipeline, std::move(collator.getValue()));
@@ -565,7 +616,6 @@ Status ViewCatalog::createView(OperationContext* opCtx,
 }
 
 Status ViewCatalog::modifyView(OperationContext* opCtx,
-                               const Database* db,
                                const NamespaceString& viewName,
                                const NamespaceString& viewOn,
                                const BSONArray& pipeline) {
@@ -573,7 +623,7 @@ Status ViewCatalog::modifyView(OperationContext* opCtx,
     invariant(opCtx->lockState()->isCollectionLockedForMode(
         NamespaceString(viewName.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
 
-    const auto& catalogStorage = getViewCatalog(db);
+    const auto& catalogStorage = getViewCatalog(opCtx->getServiceContext());
     auto catalog = catalogStorage.writer();
 
     if (viewName.db() != viewOn.db())
@@ -600,8 +650,10 @@ Status ViewCatalog::modifyView(OperationContext* opCtx,
 
     Status result = Status::OK();
     {
-        ON_BLOCK_EXIT([&catalogStorage] { catalogStorage.setIgnoreExternalChange(false); });
-        catalogStorage.setIgnoreExternalChange(true);
+        ON_BLOCK_EXIT([&catalogStorage, &viewName] {
+            catalogStorage.setIgnoreExternalChange(viewName.db(), false);
+        });
+        catalogStorage.setIgnoreExternalChange(viewName.db(), true);
 
         result = catalog.writable()->_createOrUpdateView(
             opCtx,
@@ -618,24 +670,28 @@ Status ViewCatalog::modifyView(OperationContext* opCtx,
     return result;
 }
 
-Status ViewCatalog::dropView(OperationContext* opCtx,
-                             const Database* db,
-                             const NamespaceString& viewName) {
+Status ViewCatalog::dropView(OperationContext* opCtx, const NamespaceString& viewName) {
     invariant(opCtx->lockState()->isDbLockedForMode(viewName.db(), MODE_IX));
     invariant(opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
     invariant(opCtx->lockState()->isCollectionLockedForMode(
         NamespaceString(viewName.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
 
-    const auto& catalogStorage = getViewCatalog(db);
+    const auto& catalogStorage = getViewCatalog(opCtx->getServiceContext());
     auto catalog = catalogStorage.writer();
-    catalog->_requireValidCatalog();
+
+    auto it = catalog.writable()->_viewsForDatabase.find(viewName.db());
+    invariant(it != catalog.writable()->_viewsForDatabase.end());
+    auto& vfdb = it->second;
+    vfdb.requireValidCatalog();
 
     Status result = Status::OK();
 
     {
-        ON_BLOCK_EXIT([&catalogStorage] { catalogStorage.setIgnoreExternalChange(false); });
+        ON_BLOCK_EXIT([&catalogStorage, &viewName] {
+            catalogStorage.setIgnoreExternalChange(viewName.db(), false);
+        });
 
-        catalogStorage.setIgnoreExternalChange(true);
+        catalogStorage.setIgnoreExternalChange(viewName.db(), true);
 
         // Save a copy of the view definition in case we need to roll back.
         auto viewPtr =
@@ -645,10 +701,10 @@ Status ViewCatalog::dropView(OperationContext* opCtx,
                     str::stream() << "cannot drop missing view: " << viewName.ns()};
         }
 
-        invariant(catalog->_valid);
-        catalog.writable()->_durable->remove(opCtx, viewName);
-        catalog.writable()->_viewGraph.remove(viewPtr->name());
-        catalog.writable()->_viewMap.erase(viewName.ns());
+        invariant(vfdb.valid);
+        vfdb.durable->remove(opCtx, viewName);
+        vfdb.viewGraph.remove(viewPtr->name());
+        vfdb.viewMap.erase(viewName.ns());
 
         auto viewRid = ResourceId(RESOURCE_COLLECTION, viewName.ns());
         CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
@@ -668,7 +724,7 @@ Status ViewCatalog::dropView(OperationContext* opCtx,
 
         // Reload the view catalog with the changes applied.
         result = catalog.writable()->_reload(
-            opCtx, ViewCatalogLookupBehavior::kValidateDurableViews, false);
+            opCtx, viewName.db(), ViewCatalogLookupBehavior::kValidateDurableViews, false);
     }
     catalog.commit();
     return result;
@@ -678,9 +734,15 @@ std::shared_ptr<const ViewDefinition> ViewCatalog::_lookup(
     OperationContext* opCtx,
     const NamespaceString& ns,
     ViewCatalogLookupBehavior lookupBehavior) const {
-    ViewMap::const_iterator it = _viewMap.find(ns.ns());
-    if (it != _viewMap.end()) {
-        return it->second;
+    auto it = _viewsForDatabase.find(ns.db());
+    if (it == _viewsForDatabase.end()) {
+        return nullptr;
+    }
+    auto& vfdb = it->second;
+
+    ViewMap::const_iterator vmit = vfdb.viewMap.find(ns.ns());
+    if (vmit != vfdb.viewMap.end()) {
+        return vmit->second;
     }
     return nullptr;
 }
@@ -694,7 +756,13 @@ std::shared_ptr<ViewDefinition> ViewCatalog::_lookup(OperationContext* opCtx,
 
 std::shared_ptr<const ViewDefinition> ViewCatalog::lookup(OperationContext* opCtx,
                                                           const NamespaceString& ns) const {
-    if (!_valid && opCtx->getClient()->isFromUserConnection()) {
+    auto it = _viewsForDatabase.find(ns.db());
+    if (it == _viewsForDatabase.end()) {
+        return nullptr;
+    }
+    auto& vfdb = it->second;
+
+    if (!vfdb.valid && opCtx->getClient()->isFromUserConnection()) {
         // We want to avoid lookups on invalid collection names.
         if (!NamespaceString::validCollectionName(ns.ns())) {
             return nullptr;
@@ -703,7 +771,7 @@ std::shared_ptr<const ViewDefinition> ViewCatalog::lookup(OperationContext* opCt
         // ApplyOps should work on a valid existing collection, despite the presence of bad views
         // otherwise the server would crash. The view catalog will remain invalid until the bad view
         // definitions are removed.
-        _requireValidCatalog();
+        vfdb.requireValidCatalog();
     }
 
     return _lookup(opCtx, ns, ViewCatalogLookupBehavior::kValidateDurableViews);
@@ -718,7 +786,12 @@ StatusWith<ResolvedView> ViewCatalog::resolveView(
     OperationContext* opCtx,
     const NamespaceString& nss,
     boost::optional<BSONObj> timeSeriesCollator) const {
-    _requireValidCatalog();
+    auto it = _viewsForDatabase.find(nss.db());
+    uassert(ErrorCodes::NamespaceNotFound,
+            str::stream() << "View " << nss << " not found",
+            it != _viewsForDatabase.end());
+    auto& vfdb = it->second;
+    vfdb.requireValidCatalog();
 
     // Points to the name of the most resolved namespace.
     const NamespaceString* resolvedNss = &nss;
@@ -814,7 +887,12 @@ StatusWith<ResolvedView> ViewCatalog::resolveView(
     MONGO_UNREACHABLE;
 }
 
-ViewCatalog::Stats ViewCatalog::getStats() const {
-    return _stats;
+boost::optional<ViewCatalog::Stats> ViewCatalog::getStats(StringData dbName) const {
+    auto it = _viewsForDatabase.find(dbName);
+    if (it == _viewsForDatabase.end()) {
+        return boost::none;
+    }
+    auto& vfdb = it->second;
+    return vfdb.stats;
 }
 }  // namespace mongo
diff --git a/src/mongo/db/views/view_catalog.h b/src/mongo/db/views/view_catalog.h
index b118c532c4c..a55ff015c14 100644
--- a/src/mongo/db/views/view_catalog.h
+++ b/src/mongo/db/views/view_catalog.h
@@ -57,8 +57,7 @@ class Database;
  * modifications through the static functions copy the existing instance and perform the
  * modification on the copy. A new call to get() is necessary to observe the modification.
  *
- * Writes via the static functions are thread-safe and serialized with a mutex per Database -- this
- * is needed as concurrent updates may happen through direct writes to the views catalog collection.
+ * Writes via the static functions are thread-safe and serialized with a mutex.
  *
  * The static methods refresh the in-memory map with the views catalog collection if necessary,
  * throwing if the refresh fails.
@@ -68,19 +67,33 @@ public:
     using ViewMap = StringMap<std::shared_ptr<ViewDefinition>>;
     using ViewIteratorCallback = std::function<bool(const ViewDefinition& view)>;
 
-    static std::shared_ptr<const ViewCatalog> get(const Database* db);
-    static void set(Database* db, std::unique_ptr<ViewCatalog> catalog);
+    static std::shared_ptr<const ViewCatalog> get(ServiceContext* svcCtx);
+    static std::shared_ptr<const ViewCatalog> get(OperationContext* opCtx);
 
-    explicit ViewCatalog(std::unique_ptr<DurableViewCatalog> durable)
-        : _durable(std::move(durable)), _valid(false), _viewGraphNeedsRefresh(true) {}
+    /**
+     * Add an entry to the ViewCatalog for the given database, backed by the durable storage
+     * 'catalog'.
+     */
+    static Status registerDatabase(OperationContext* opCtx,
+                                   StringData dbName,
+                                   std::unique_ptr<DurableViewCatalog> catalog);
+
+    /**
+     * Removes the ViewCatalog entries assocated with 'db' if any. Should be called when when a
+     * `DatabaseImpl` that has previously registered is about to be destructed (e.g. when closing a
+     * database).
+     */
+    static void unregisterDatabase(OperationContext* opCtx, Database* db);
 
     /**
      * Iterates through the catalog, applying 'callback' to each view. This callback function
      * executes under the catalog's mutex, so it must not access other methods of the catalog,
-     * acquire locks or run for a long time. If the 'callback' returns false, the iterator exists
+     * acquire locks or run for a long time. If the 'callback' returns false, the iterator exits
      * early.
+     *
+     * Caller must ensure corresponding database exists.
      */
-    void iterate(ViewIteratorCallback callback) const;
+    void iterate(StringData dbName, ViewIteratorCallback callback) const;
 
     /**
      * Create a new view 'viewName' with contents defined by running the specified aggregation
@@ -90,9 +103,10 @@ public:
      * before calling createView.
      *
      * Must be in WriteUnitOfWork. View creation rolls back if the unit of work aborts.
+     *
+     * Caller must ensure corresponding database exists.
      */
     static Status createView(OperationContext* opCtx,
-                             const Database* db,
                              const NamespaceString& viewName,
                              const NamespaceString& viewOn,
                              const BSONArray& pipeline,
@@ -102,18 +116,19 @@ public:
      * Drop the view named 'viewName'.
      *
      * Must be in WriteUnitOfWork. The drop rolls back if the unit of work aborts.
+     *
+     * Caller must ensure corresponding database exists.
      */
-    static Status dropView(OperationContext* opCtx,
-                           const Database* db,
-                           const NamespaceString& viewName);
+    static Status dropView(OperationContext* opCtx, const NamespaceString& viewName);
 
     /**
      * Modify the view named 'viewName' to have the new 'viewOn' and 'pipeline'.
      *
      * Must be in WriteUnitOfWork. The modification rolls back if the unit of work aborts.
+     *
+     * Caller must ensure corresponding database exists.
      */
     static Status modifyView(OperationContext* opCtx,
-                             const Database* db,
                              const NamespaceString& viewName,
                              const NamespaceString& viewOn,
                              const BSONArray& pipeline);
@@ -121,6 +136,8 @@ public:
     /**
      * Look up the 'nss' in the view catalog, returning a shared pointer to a View definition, or
      * nullptr if it doesn't exist.
+     *
+     * Caller must ensure corresponding database exists.
      */
     std::shared_ptr<const ViewDefinition> lookup(OperationContext* opCtx,
                                                  const NamespaceString& nss) const;
@@ -128,6 +145,8 @@ public:
     /**
      * Same functionality as above, except this function skips validating durable views in the view
      * catalog.
+     *
+     * Caller must ensure corresponding database exists.
      */
     std::shared_ptr<const ViewDefinition> lookupWithoutValidatingDurableViews(
         OperationContext* opCtx, const NamespaceString& nss) const;
@@ -141,6 +160,8 @@ public:
      * collations. So in the case of queries on timeseries collections, we create a ResolvedView
      * with the request's collation (timeSeriesCollator) rather than the collection's default
      * collator.
+     *
+     * Caller must ensure corresponding database exists.
      */
     StatusWith<ResolvedView> resolveView(OperationContext* opCtx,
                                          const NamespaceString& nss,
@@ -157,9 +178,9 @@ public:
     };
 
     /**
-     * Returns statistics for this view catalog.
+     * Returns view statistics for the specified database.
      */
-    Stats getStats() const;
+    boost::optional<Stats> getStats(StringData dbName) const;
 
     /**
      * Returns Status::OK with the set of involved namespaces if the given pipeline is eligible to
@@ -178,20 +199,18 @@ public:
      * database.
      */
     static Status reload(OperationContext* opCtx,
-                         const Database* db,
+                         StringData dbName,
                          ViewCatalogLookupBehavior lookupBehavior);
 
     /**
      * Clears the in-memory state of the view catalog.
      */
-    static void clear(OperationContext* opCtx, const Database* db);
+    static void clear(OperationContext* opCtx, StringData dbName);
 
     /**
      * The view catalog needs to ignore external changes for its own modifications.
      */
-    static bool shouldIgnoreExternalChange(OperationContext* opCtx,
-                                           const Database* db,
-                                           const NamespaceString& name);
+    static bool shouldIgnoreExternalChange(OperationContext* opCtx, const NamespaceString& name);
 
 private:
     Status _createOrUpdateView(OperationContext* opCtx,
@@ -221,21 +240,32 @@ private:
                                             ViewCatalogLookupBehavior lookupBehavior);
 
     Status _reload(OperationContext* opCtx,
+                   StringData dbName,
                    ViewCatalogLookupBehavior lookupBehavior,
                    bool reloadForCollectionCatalog);
 
     /**
-     * uasserts with the InvalidViewDefinition error if the current in-memory state of the view
-     * catalog is invalid. This ensures that calling into the view catalog while it is invalid
-     * renders it inoperable.
+     * Holds all data for the views associated with a particular database. Prior to 5.3, the
+     * ViewCatalog object was owned by the Database object as a decoration. It has now transitioned
+     * to a global catalog, as a decoration on the ServiceContext. Each database gets its own record
+     * here, comprising the same information that was previously stored as top-level information
+     * prior to 5.3.
      */
-    void _requireValidCatalog() const;
+    struct ViewsForDatabase {
+        ViewMap viewMap;
+        std::shared_ptr<DurableViewCatalog> durable;
+        bool valid = false;
+        ViewGraph viewGraph;
+        bool viewGraphNeedsRefresh = true;
+        Stats stats;
 
-    ViewMap _viewMap;
-    std::shared_ptr<DurableViewCatalog> _durable;
-    bool _valid;
-    ViewGraph _viewGraph;
-    bool _viewGraphNeedsRefresh;
-    Stats _stats;
+        /**
+         * uasserts with the InvalidViewDefinition error if the current in-memory state of the view
+         * catalog for the given database is invalid. This ensures that calling into the view
+         * catalog while it is invalid renders it inoperable.
+         */
+        void requireValidCatalog() const;
+    };
+    StringMap<ViewsForDatabase> _viewsForDatabase;
 };
 }  // namespace mongo
diff --git a/src/mongo/db/views/view_catalog_test.cpp b/src/mongo/db/views/view_catalog_test.cpp
index 1836f3feb21..8fb0712477f 100644
--- a/src/mongo/db/views/view_catalog_test.cpp
+++ b/src/mongo/db/views/view_catalog_test.cpp
@@ -101,12 +101,12 @@ public:
         wuow.commit();
     }
 
-    void tearDown() {
+    void tearDown() override {
         CatalogTestFixture::tearDown();
     }
 
-    auto getViewCatalog() const {
-        return ViewCatalog::get(_db);
+    auto getViewCatalog() {
+        return ViewCatalog::get(operationContext());
     }
 
     Status createView(OperationContext* opCtx,
@@ -122,7 +122,7 @@ public:
             MODE_X);
 
         WriteUnitOfWork wuow(opCtx);
-        Status s = ViewCatalog::createView(opCtx, _db, viewName, viewOn, pipeline, collation);
+        Status s = ViewCatalog::createView(opCtx, viewName, viewOn, pipeline, collation);
         wuow.commit();
 
         return s;
@@ -140,7 +140,7 @@ public:
             MODE_X);
 
         WriteUnitOfWork wuow(opCtx);
-        Status s = ViewCatalog::modifyView(opCtx, _db, viewName, viewOn, pipeline);
+        Status s = ViewCatalog::modifyView(opCtx, viewName, viewOn, pipeline);
         wuow.commit();
 
         return s;
@@ -155,7 +155,7 @@ public:
             MODE_X);
 
         WriteUnitOfWork wuow(opCtx);
-        Status s = ViewCatalog::dropView(opCtx, _db, viewName);
+        Status s = ViewCatalog::dropView(opCtx, viewName);
         wuow.commit();
 
         return s;
@@ -533,7 +533,7 @@ TEST_F(ViewCatalogFixture, LookupRIDExistingViewRollback) {
 
         WriteUnitOfWork wunit(operationContext());
         ASSERT_OK(ViewCatalog::createView(
-            operationContext(), db(), viewName, viewOn, emptyPipeline, emptyCollation));
+            operationContext(), viewName, viewOn, emptyPipeline, emptyCollation));
     }
     auto resourceID = ResourceId(RESOURCE_COLLECTION, "db.view"_sd);
     auto collectionCatalog = CollectionCatalog::get(operationContext());
@@ -574,7 +574,7 @@ TEST_F(ViewCatalogFixture, LookupRIDAfterDropRollback) {
             MODE_X);
 
         WriteUnitOfWork wunit(operationContext());
-        ASSERT_OK(ViewCatalog::dropView(operationContext(), db(), viewName));
+        ASSERT_OK(ViewCatalog::dropView(operationContext(), viewName));
     }
 
     ASSERT(CollectionCatalog::get(operationContext())->lookupResourceName(resourceID).get() ==
@@ -613,8 +613,7 @@ TEST_F(ViewCatalogFixture, LookupRIDAfterModifyRollback) {
             MODE_X);
 
         WriteUnitOfWork wunit(operationContext());
-        ASSERT_OK(
-            ViewCatalog::modifyView(operationContext(), db(), viewName, viewOn, emptyPipeline));
+        ASSERT_OK(ViewCatalog::modifyView(operationContext(), viewName, viewOn, emptyPipeline));
         ASSERT(CollectionCatalog::get(operationContext())->lookupResourceName(resourceID).get() ==
                viewName.ns());
     }
@@ -645,7 +644,7 @@ TEST_F(ViewCatalogFixture, Iterate) {
     std::set<std::string> viewNames = {"db.view1", "db.view2", "db.view3"};
 
     Lock::DBLock dbLock(operationContext(), "db", MODE_IX);
-    getViewCatalog()->iterate([&viewNames](const ViewDefinition& view) {
+    getViewCatalog()->iterate("db", [&viewNames](const ViewDefinition& view) {
         std::string name = view.name().toString();
         ASSERT(viewNames.end() != viewNames.find(name));
         viewNames.erase(name);