SERVER-57250 Merge ViewCatalog into CollectionCatalog

21 files changed, 1158 insertions, 267 deletions

diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript
index 6f0345dce1c..e33238fd2d1 100644
--- a/src/mongo/db/catalog/SConscript
+++ b/src/mongo/db/catalog/SConscript
@@ -121,7 +121,9 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/catalog_raii',
         '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
+        '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/index/index_access_method',
+        '$BUILD_DIR/mongo/db/storage/key_string',
         'validate_state',
     ]
 )
@@ -135,6 +137,8 @@ env.Library(
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/db/catalog_raii',
         '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
+        '$BUILD_DIR/mongo/db/curop',
+        '$BUILD_DIR/mongo/db/query/query_knobs',
         '$BUILD_DIR/mongo/db/storage/storage_repair_observer',
         'index_repair',
         'multi_index_block',
@@ -274,17 +278,19 @@ env.Library(
         'collection_catalog.cpp',
         'uncommitted_collections.cpp',
         'uncommitted_multikey.cpp',
+        'views_for_database.cpp',
     ],
     LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
         '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/namespace_string',
         '$BUILD_DIR/mongo/db/profile_filter',
+        '$BUILD_DIR/mongo/db/query/collation/collator_factory_interface',
         '$BUILD_DIR/mongo/db/server_options_core',
         '$BUILD_DIR/mongo/db/service_context',
         '$BUILD_DIR/mongo/db/storage/bson_collection_catalog_entry',
         '$BUILD_DIR/mongo/db/storage/snapshot_helper',
+        '$BUILD_DIR/mongo/db/views/views',
         '$BUILD_DIR/mongo/idl/server_parameter',
         'collection',
     ]
@@ -327,6 +333,7 @@ env.Library(
         "$BUILD_DIR/mongo/base",
         "$BUILD_DIR/mongo/db/catalog_raii",
         "$BUILD_DIR/mongo/db/multitenancy",
+        '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
         "$BUILD_DIR/mongo/db/views/views",
         "$BUILD_DIR/mongo/util/fail_point",
         "collection_catalog",
@@ -377,7 +384,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/transaction',
         '$BUILD_DIR/mongo/db/ttl_collection_cache',
         '$BUILD_DIR/mongo/db/vector_clock',
-        '$BUILD_DIR/mongo/db/views/views',
+        '$BUILD_DIR/mongo/db/views/view_catalog_helpers',
         '$BUILD_DIR/mongo/db/views/views_mongod',
         'catalog_helpers',
         'catalog_stats',
@@ -501,6 +508,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/storage/key_string',
         '$BUILD_DIR/mongo/db/timeseries/timeseries_options',
         '$BUILD_DIR/mongo/db/ttl_collection_cache',
+        '$BUILD_DIR/mongo/db/views/view_catalog_helpers',
         '$BUILD_DIR/mongo/db/views/views',
         '$BUILD_DIR/mongo/db/write_ops',
         'cannot_convert_index_to_unique_info',
diff --git a/src/mongo/db/catalog/capped_utils.cpp b/src/mongo/db/catalog/capped_utils.cpp
index ca43c0a20e2..86ee7a8fb10 100644
--- a/src/mongo/db/catalog/capped_utils.cpp
+++ b/src/mongo/db/catalog/capped_utils.cpp
@@ -34,6 +34,7 @@
 #include "mongo/db/catalog/capped_utils.h"
 
 #include "mongo/base/error_codes.h"
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/create_collection.h"
 #include "mongo/db/catalog/document_validation.h"
 #include "mongo/db/catalog/drop_collection.h"
@@ -52,7 +53,6 @@
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/s/collection_sharding_state.h"
 #include "mongo/db/service_context.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/util/scopeguard.h"
 
 namespace mongo {
@@ -75,7 +75,7 @@ Status emptyCapped(OperationContext* opCtx, const NamespaceString& collectionNam
     CollectionWriter collection(opCtx, collectionName);
     uassert(ErrorCodes::CommandNotSupportedOnView,
             str::stream() << "emptycapped not supported on view: " << collectionName.ns(),
-            collection || !ViewCatalog::get(opCtx)->lookup(opCtx, collectionName));
+            collection || !CollectionCatalog::get(opCtx)->lookupView(opCtx, collectionName));
     uassert(ErrorCodes::NamespaceNotFound, "no such collection", collection);
 
     if (collectionName.isSystem() && !collectionName.isSystemDotProfile()) {
@@ -127,7 +127,7 @@ void cloneCollectionAsCapped(OperationContext* opCtx,
     if (!fromCollection) {
         uassert(ErrorCodes::CommandNotSupportedOnView,
                 str::stream() << "cloneCollectionAsCapped not supported for views: " << fromNss,
-                !ViewCatalog::get(opCtx)->lookup(opCtx, fromNss));
+                !CollectionCatalog::get(opCtx)->lookupView(opCtx, fromNss));
 
         uasserted(ErrorCodes::NamespaceNotFound,
                   str::stream() << "source collection " << fromNss << " does not exist");
diff --git a/src/mongo/db/catalog/catalog_stats.cpp b/src/mongo/db/catalog/catalog_stats.cpp
index f79f354a773..ea7b4ee2276 100644
--- a/src/mongo/db/catalog/catalog_stats.cpp
+++ b/src/mongo/db/catalog/catalog_stats.cpp
@@ -35,7 +35,6 @@
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/commands/server_status.h"
 #include "mongo/db/db_raii.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 
 namespace mongo {
@@ -82,26 +81,21 @@ public:
         stats.clustered = catalogStats.userClustered;
         stats.internalCollections = catalogStats.internal;
 
-        const auto viewCatalogDbNames = catalog->getViewCatalogDbNames();
-        if (const auto viewCatalog = ViewCatalog::get(opCtx)) {
-            for (const auto& tenantDbName : viewCatalogDbNames) {
-                try {
-                    const auto viewStats = viewCatalog->getStats(tenantDbName.dbName());
-                    if (!viewStats) {
-                        // The database may have been dropped between listing the database names and
-                        // looking up the view catalog.
-                        continue;
-                    }
+        const auto viewCatalogDbNames = catalog->getViewCatalogDbNames(opCtx);
+        for (const auto& tenantDbName : viewCatalogDbNames) {
+            try {
+                const auto viewStats =
+                    catalog->getViewStatsForDatabase(opCtx, tenantDbName.dbName());
+                invariant(viewStats);
 
-                    stats.timeseries += viewStats->userTimeseries;
-                    stats.views += viewStats->userViews;
-                    stats.internalViews += viewStats->internal;
-                } catch (ExceptionForCat<ErrorCategory::Interruption>&) {
-                    LOGV2_DEBUG(5578400,
-                                2,
-                                "Failed to collect view catalog statistics",
-                                "db"_attr = tenantDbName);
-                }
+                stats.timeseries += viewStats->userTimeseries;
+                stats.views += viewStats->userViews;
+                stats.internalViews += viewStats->internal;
+            } catch (ExceptionForCat<ErrorCategory::Interruption>&) {
+                LOGV2_DEBUG(5578400,
+                            2,
+                            "Failed to collect view catalog statistics",
+                            "db"_attr = tenantDbName);
             }
         }
 
diff --git a/src/mongo/db/catalog/coll_mod.cpp b/src/mongo/db/catalog/coll_mod.cpp
index 490ab5be065..12a7715d33d 100644
--- a/src/mongo/db/catalog/coll_mod.cpp
+++ b/src/mongo/db/catalog/coll_mod.cpp
@@ -38,7 +38,7 @@
 
 #include "mongo/db/catalog/clustered_collection_util.h"
 #include "mongo/db/catalog/coll_mod_index.h"
-#include "mongo/db/catalog/collection_options.h"
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/collection_uuid_mismatch.h"
 #include "mongo/db/catalog/create_collection.h"
 #include "mongo/db/catalog/index_catalog.h"
@@ -59,7 +59,7 @@
 #include "mongo/db/storage/storage_parameters_gen.h"
 #include "mongo/db/timeseries/timeseries_options.h"
 #include "mongo/db/ttl_collection_cache.h"
-#include "mongo/db/views/view_catalog.h"
+#include "mongo/db/views/view_catalog_helpers.h"
 #include "mongo/idl/command_generic_argument.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/fail_point.h"
@@ -652,7 +652,7 @@ Status _collModInternal(OperationContext* opCtx,
     // May also modify a view instead of a collection.
     boost::optional<ViewDefinition> view;
     if (!coll) {
-        const auto sharedView = ViewCatalog::get(opCtx)->lookup(opCtx, nss);
+        const auto sharedView = CollectionCatalog::get(opCtx)->lookupView(opCtx, nss);
         if (sharedView) {
             // We copy the ViewDefinition as it is modified below to represent the requested state.
             view = {*sharedView};
@@ -720,8 +720,8 @@ Status _collModInternal(OperationContext* opCtx,
     return writeConflictRetry(opCtx, "collMod", nss.ns(), [&] {
         WriteUnitOfWork wunit(opCtx);
 
-        // Handle collMod on a view and return early. The View Catalog handles the creation of oplog
-        // entries for modifications on a view.
+        // Handle collMod on a view and return early. The CollectionCatalog handles the creation of
+        // oplog entries for modifications on a view.
         if (view) {
             if (cmd.getPipeline())
                 view->setPipeline(*cmd.getPipeline());
@@ -734,7 +734,11 @@ Status _collModInternal(OperationContext* opCtx,
                 pipeline.append(item);
             }
             auto errorStatus =
-                ViewCatalog::modifyView(opCtx, nss, view->viewOn(), BSONArray(pipeline.obj()));
+                CollectionCatalog::get(opCtx)->modifyView(opCtx,
+                                                          nss,
+                                                          view->viewOn(),
+                                                          BSONArray(pipeline.obj()),
+                                                          view_catalog_helpers::validatePipeline);
             if (!errorStatus.isOK()) {
                 return errorStatus;
             }
diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index 7901809f2ca..5b3447c0332 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -53,6 +53,11 @@ const ServiceContext::Decoration<LatestCollectionCatalog> getCatalog =
 
 std::shared_ptr<CollectionCatalog> batchedCatalogWriteInstance;
 
+const OperationContext::Decoration<std::shared_ptr<const CollectionCatalog>> stashedCatalog =
+    OperationContext::declareDecoration<std::shared_ptr<const CollectionCatalog>>();
+
+}  // namespace
+
 /**
  * Decoration on RecoveryUnit to store cloned Collections until they are committed or rolled
  * back TODO SERVER-51236: This should be merged with UncommittedCollections
@@ -62,19 +67,25 @@ public:
     struct Entry {
         enum class Action {
             // Writable clone
-            kWritable,
+            kWritableCollection,
             // Marker to indicate that the namespace has been renamed
-            kRenamed,
+            kRenamedCollection,
             // Dropped collection instance
-            kDropped,
+            kDroppedCollection,
             // Recreated collection after drop
-            kRecreated
+            kRecreatedCollection,
+            // Replaced views for a particular database
+            kReplacedViewsForDatabase,
+            // Add a view resource
+            kAddViewResource,
+            // Remove a view resource
+            kRemoveViewResource,
         };
 
-        UUID uuid() const {
-            if (action == Action::kDropped || action == Entry::Action::kRecreated)
-                return *externalUUID;
-            return collection->uuid();
+        boost::optional<UUID> uuid() const {
+            if (action == Action::kWritableCollection || action == Action::kRenamedCollection)
+                return collection->uuid();
+            return externalUUID;
         }
 
         // Type of action this entry has stored. Members below may or may not be set depending on
@@ -82,7 +93,7 @@ public:
         Action action;
 
         // Storage for the actual collection.
-        // Set for actions kWritable, kRecreated. nullptr otherwise.
+        // Set for actions kWritableCollection, kRecreatedCollection. nullptr otherwise.
         std::shared_ptr<Collection> collection;
 
         // Store namespace separately to handle rename and drop without making writable first
@@ -90,24 +101,38 @@ public:
         NamespaceString nss;
 
         // External uuid when not accessible via collection
-        // Set for actions kDropped, kRecreated. boost::none otherwise.
+        // Set for actions kDroppedCollection, kRecreatedCollection. boost::none otherwise.
         boost::optional<UUID> externalUUID;
 
         // New namespace this collection has been renamed to
-        // Set for action kRenamed. Default constructed otherwise.
+        // Set for action kRenamedCollection. Default constructed otherwise.
         NamespaceString renameTo;
+
+        // New set of view information for a database.
+        // Set for action kReplacedViewsForDatabase, boost::none otherwise.
+        boost::optional<ViewsForDatabase> viewsForDb;
     };
 
     /**
+     * Determine if an entry is associated with a collection action (as opposed to a view action).
+     */
+    static bool isCollectionEntry(const Entry& entry) {
+        return (entry.action == Entry::Action::kWritableCollection ||
+                entry.action == Entry::Action::kRenamedCollection ||
+                entry.action == Entry::Action::kDroppedCollection ||
+                entry.action == Entry::Action::kRecreatedCollection);
+    }
+
+    /**
      * Lookup of Collection by UUID. The boolean indicates if this namespace is managed.
      * A managed Collection pointer may be returned as nullptr, which indicates a drop.
      * If the returned boolean is false then the Collection will always be nullptr.
      */
-    std::pair<bool, Collection*> lookup(UUID uuid) const {
+    std::pair<bool, Collection*> lookupCollection(UUID uuid) const {
         // Doing reverse search so we find most recent entry affecting this uuid
         auto it = std::find_if(_entries.rbegin(), _entries.rend(), [uuid](auto&& entry) {
             // Rename actions don't have UUID
-            if (entry.action == Entry::Action::kRenamed)
+            if (entry.action == Entry::Action::kRenamedCollection)
                 return false;
 
             return entry.uuid() == uuid;
@@ -122,40 +147,52 @@ public:
      * A managed Collection pointer may be returned as nullptr, which indicates drop or rename.
      * If the returned boolean is false then the Collection will always be nullptr.
      */
-    std::pair<bool, Collection*> lookup(const NamespaceString& nss) const {
+    std::pair<bool, Collection*> lookupCollection(const NamespaceString& nss) const {
         // Doing reverse search so we find most recent entry affecting this namespace
-        auto it = std::find_if(
-            _entries.rbegin(), _entries.rend(), [&nss](auto&& entry) { return entry.nss == nss; });
+        auto it = std::find_if(_entries.rbegin(), _entries.rend(), [&nss](auto&& entry) {
+            return entry.nss == nss && isCollectionEntry(entry);
+        });
         if (it == _entries.rend())
             return {false, nullptr};
         return {true, it->collection.get()};
     }
 
+    boost::optional<const ViewsForDatabase&> getViewsForDatabase(StringData dbName) const {
+        // Doing reverse search so we find most recent entry affecting this namespace
+        auto it = std::find_if(_entries.rbegin(), _entries.rend(), [&](auto&& entry) {
+            return entry.nss.db() == dbName && entry.viewsForDb;
+        });
+        if (it == _entries.rend())
+            return boost::none;
+        return {*it->viewsForDb};
+    }
+
     /**
      * Manage the lifetime of uncommitted writable collection
      */
-    void writable(std::shared_ptr<Collection> collection) {
+    void writableCollection(std::shared_ptr<Collection> collection) {
         const auto& ns = collection->ns();
-        _entries.push_back({Entry::Action::kWritable, std::move(collection), ns});
+        _entries.push_back({Entry::Action::kWritableCollection, std::move(collection), ns});
     }
 
     /**
      * Manage an uncommitted rename, pointer must have made writable first and should exist in entry
      * list
      */
-    void rename(const Collection* collection, const NamespaceString& from) {
+    void renameCollection(const Collection* collection, const NamespaceString& from) {
         auto it = std::find_if(_entries.rbegin(), _entries.rend(), [collection](auto&& entry) {
             return entry.collection.get() == collection;
         });
         invariant(it != _entries.rend());
         it->nss = collection->ns();
-        _entries.push_back({Entry::Action::kRenamed, nullptr, from, boost::none, it->nss});
+        _entries.push_back(
+            {Entry::Action::kRenamedCollection, nullptr, from, boost::none, it->nss});
     }
 
     /**
      * Manage an uncommitted collection drop
      */
-    void drop(const Collection* collection) {
+    void dropCollection(const Collection* collection) {
         auto it = std::find_if(
             _entries.rbegin(), _entries.rend(), [uuid = collection->uuid()](auto&& entry) {
                 return entry.uuid() == uuid;
@@ -163,13 +200,13 @@ public:
         if (it == _entries.rend()) {
             // Entry with this uuid was not found, add new
             _entries.push_back(
-                {Entry::Action::kDropped, nullptr, collection->ns(), collection->uuid()});
+                {Entry::Action::kDroppedCollection, nullptr, collection->ns(), collection->uuid()});
             return;
         }
 
         // If we have been recreated after drop we can simply just erase this entry so lookup will
         // then find previous drop
-        if (it->action == Entry::Action::kRecreated) {
+        if (it->action == Entry::Action::kRecreatedCollection) {
             _entries.erase(it.base());
             return;
         }
@@ -180,7 +217,7 @@ public:
 
         // Transform found entry into dropped.
         invariant(it->collection.get() == collection);
-        it->action = Entry::Action::kDropped;
+        it->action = Entry::Action::kDroppedCollection;
         it->externalUUID = it->collection->uuid();
         it->collection = nullptr;
     }
@@ -188,9 +225,47 @@ public:
     /**
      * Re-creates a collection that has previously been dropped
      */
-    void createAfterDrop(UUID uuid, std::shared_ptr<Collection> collection) {
+    void createCollectionAfterDrop(UUID uuid, std::shared_ptr<Collection> collection) {
         const auto& ns = collection->ns();
-        _entries.push_back({Entry::Action::kRecreated, std::move(collection), ns, uuid});
+        _entries.push_back({Entry::Action::kRecreatedCollection, std::move(collection), ns, uuid});
+    }
+
+    /**
+     * Replace the ViewsForDatabase instance assocated with database `dbName` with `vfdb`. This is
+     * the primary low-level write method to alter any information about the views associated with a
+     * given database.
+     */
+    void replaceViewsForDatabase(StringData dbName, ViewsForDatabase&& vfdb) {
+        _entries.push_back({Entry::Action::kReplacedViewsForDatabase,
+                            nullptr,
+                            NamespaceString{dbName},
+                            boost::none,
+                            {},
+                            std::move(vfdb)});
+    }
+
+    /**
+     * Adds a ResourceID associated with a view namespace, and registers a preCommitHook to do
+     * conflict-checking on the view namespace.
+     */
+    void addView(OperationContext* opCtx, const NamespaceString nss) {
+        opCtx->recoveryUnit()->registerPreCommitHook([nss](OperationContext* opCtx) {
+            CollectionCatalog::write(opCtx, [opCtx, nss](CollectionCatalog& catalog) {
+                catalog.registerUncommittedView(opCtx, nss);
+            });
+        });
+        opCtx->recoveryUnit()->onRollback([opCtx, nss]() {
+            CollectionCatalog::write(
+                opCtx, [&](CollectionCatalog& catalog) { catalog.deregisterUncommittedView(nss); });
+        });
+        _entries.push_back({Entry::Action::kAddViewResource, nullptr, nss});
+    }
+
+    /**
+     * Removes the ResourceID associated with a view namespace.
+     */
+    void removeView(const NamespaceString nss) {
+        _entries.push_back({Entry::Action::kRemoveViewResource, nullptr, nss});
     }
 
     /**
@@ -202,12 +277,37 @@ public:
         return ret;
     }
 
+    /**
+     * The catalog needs to ignore external view changes for its own modifications. This method
+     * should be used by DDL operations to prevent op observers from triggering additional catalog
+     * operations.
+     */
+    void setIgnoreExternalViewChanges(StringData dbName, bool value) {
+        if (value) {
+            _ignoreExternalViewChanges.emplace(dbName);
+        } else {
+            _ignoreExternalViewChanges.erase(dbName);
+        }
+    }
+
+    /**
+     * The catalog needs to ignore external view changes for its own modifications. This method can
+     * be used by methods called by op observers (e.g. 'CollectionCatalog::reload()') to distinguish
+     * between an external write to 'system.views' and one initiated through the proper view DDL
+     * operations.
+     */
+    bool shouldIgnoreExternalViewChanges(StringData dbName) const {
+        return _ignoreExternalViewChanges.contains(dbName);
+    }
+
     static UncommittedCatalogUpdates& get(OperationContext* opCtx);
 
 private:
     // Store entries in vector, we will do linear search to find what we're looking for but it will
     // be very few entries so it should be fine.
     std::vector<Entry> _entries;
+
+    StringSet _ignoreExternalViewChanges;
 };
 
 const RecoveryUnit::Decoration<UncommittedCatalogUpdates> getUncommittedCatalogUpdates =
@@ -217,10 +317,23 @@ UncommittedCatalogUpdates& UncommittedCatalogUpdates::get(OperationContext* opCt
     return getUncommittedCatalogUpdates(opCtx->recoveryUnit());
 }
 
-const OperationContext::Decoration<std::shared_ptr<const CollectionCatalog>> stashedCatalog =
-    OperationContext::declareDecoration<std::shared_ptr<const CollectionCatalog>>();
+class IgnoreExternalViewChangesForDatabase {
+public:
+    IgnoreExternalViewChangesForDatabase(OperationContext* opCtx, StringData dbName)
+        : _opCtx(opCtx), _dbName(dbName) {
+        auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(_opCtx);
+        uncommittedCatalogUpdates.setIgnoreExternalViewChanges(_dbName, true);
+    }
 
-}  // namespace
+    ~IgnoreExternalViewChangesForDatabase() {
+        auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(_opCtx);
+        uncommittedCatalogUpdates.setIgnoreExternalViewChanges(_dbName, false);
+    }
+
+private:
+    OperationContext* _opCtx;
+    std::string _dbName;
+};
 
 /**
  * Publishes all uncommitted Collection actions registered on UncommittedCatalogUpdates to the
@@ -256,7 +369,7 @@ public:
         auto entries = _uncommittedCatalogUpdates.releaseEntries();
         for (auto&& entry : entries) {
             switch (entry.action) {
-                case UncommittedCatalogUpdates::Entry::Action::kWritable:
+                case UncommittedCatalogUpdates::Entry::Action::kWritableCollection:
                     writeJobs.push_back(
                         [collection = std::move(entry.collection)](CollectionCatalog& catalog) {
                             catalog._collections[collection->ns()] = collection;
@@ -267,7 +380,7 @@ public:
                             catalog._orderedCollections[dbIdPair] = collection;
                         });
                     break;
-                case UncommittedCatalogUpdates::Entry::Action::kRenamed:
+                case UncommittedCatalogUpdates::Entry::Action::kRenamedCollection:
                     writeJobs.push_back(
                         [& from = entry.nss, &to = entry.renameTo](CollectionCatalog& catalog) {
                             catalog._collections.erase(from);
@@ -282,19 +395,39 @@ public:
                             catalog.addResource(newRid, toStr);
                         });
                     break;
-                case UncommittedCatalogUpdates::Entry::Action::kDropped:
+                case UncommittedCatalogUpdates::Entry::Action::kDroppedCollection:
                     writeJobs.push_back(
-                        [opCtx = _opCtx, uuid = entry.uuid()](CollectionCatalog& catalog) {
+                        [opCtx = _opCtx, uuid = *entry.uuid()](CollectionCatalog& catalog) {
                             catalog.deregisterCollection(opCtx, uuid);
                         });
                     break;
-                case UncommittedCatalogUpdates::Entry::Action::kRecreated:
+                case UncommittedCatalogUpdates::Entry::Action::kRecreatedCollection:
                     writeJobs.push_back([opCtx = _opCtx,
                                          collection = std::move(entry.collection),
                                          uuid = *entry.externalUUID](CollectionCatalog& catalog) {
                         catalog.registerCollection(opCtx, uuid, std::move(collection));
                     });
                     break;
+                case UncommittedCatalogUpdates::Entry::Action::kReplacedViewsForDatabase:
+                    writeJobs.push_back(
+                        [dbName = entry.nss.db(),
+                         &viewsForDb = entry.viewsForDb.get()](CollectionCatalog& catalog) {
+                            catalog._replaceViewsForDatabase(dbName, std::move(viewsForDb));
+                        });
+                    break;
+                case UncommittedCatalogUpdates::Entry::Action::kAddViewResource:
+                    writeJobs.push_back([& viewName = entry.nss](CollectionCatalog& catalog) {
+                        auto viewRid = ResourceId(RESOURCE_COLLECTION, viewName.ns());
+                        catalog.addResource(viewRid, viewName.ns());
+                        catalog.deregisterUncommittedView(viewName);
+                    });
+                    break;
+                case UncommittedCatalogUpdates::Entry::Action::kRemoveViewResource:
+                    writeJobs.push_back([& viewName = entry.nss](CollectionCatalog& catalog) {
+                        auto viewRid = ResourceId(RESOURCE_COLLECTION, viewName.ns());
+                        catalog.removeResource(viewRid, viewName.ns());
+                    });
+                    break;
             };
         }
 
@@ -565,6 +698,162 @@ 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 {
+    invariant(opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
+    invariant(opCtx->lockState()->isCollectionLockedForMode(
+        NamespaceString(viewName.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
+
+    invariant(_viewsForDatabase.contains(viewName.db()));
+    const ViewsForDatabase& viewsForDb = *_getViewsForDatabase(opCtx, viewName.db());
+
+    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.contains(viewName))
+        return Status(ErrorCodes::NamespaceExists, "Namespace already exists");
+
+    if (!NamespaceString::validCollectionName(viewOn.coll()))
+        return Status(ErrorCodes::InvalidNamespace,
+                      str::stream() << "invalid name for 'viewOn': " << viewOn.coll());
+
+    auto collator = ViewsForDatabase::parseCollator(opCtx, collation);
+    if (!collator.isOK())
+        return collator.getStatus();
+
+    Status result = Status::OK();
+    {
+        IgnoreExternalViewChangesForDatabase ignore(opCtx, viewName.db());
+
+        result = _createOrUpdateView(opCtx,
+                                     viewName,
+                                     viewOn,
+                                     pipeline,
+                                     pipelineValidator,
+                                     std::move(collator.getValue()),
+                                     ViewsForDatabase{viewsForDb});
+    }
+
+    return result;
+}
+
+Status CollectionCatalog::modifyView(
+    OperationContext* opCtx,
+    const NamespaceString& viewName,
+    const NamespaceString& viewOn,
+    const BSONArray& pipeline,
+    const ViewsForDatabase::PipelineValidatorFn& pipelineValidator) const {
+    invariant(opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_X));
+    invariant(opCtx->lockState()->isCollectionLockedForMode(
+        NamespaceString(viewName.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
+
+    invariant(_viewsForDatabase.contains(viewName.db()));
+    const ViewsForDatabase& viewsForDb = *_getViewsForDatabase(opCtx, viewName.db());
+
+    if (viewName.db() != viewOn.db())
+        return Status(ErrorCodes::BadValue,
+                      "View must be created on a view or collection in the same database");
+
+    auto viewPtr = viewsForDb.lookup(viewName);
+    if (!viewPtr)
+        return Status(ErrorCodes::NamespaceNotFound,
+                      str::stream() << "cannot modify missing view " << viewName.ns());
+
+    if (!NamespaceString::validCollectionName(viewOn.coll()))
+        return Status(ErrorCodes::InvalidNamespace,
+                      str::stream() << "invalid name for 'viewOn': " << viewOn.coll());
+
+    Status result = Status::OK();
+    {
+        IgnoreExternalViewChangesForDatabase ignore(opCtx, viewName.db());
+
+        result = _createOrUpdateView(opCtx,
+                                     viewName,
+                                     viewOn,
+                                     pipeline,
+                                     pipelineValidator,
+                                     CollatorInterface::cloneCollator(viewPtr->defaultCollator()),
+                                     ViewsForDatabase{viewsForDb});
+    }
+
+    return result;
+}
+
+Status CollectionCatalog::dropView(OperationContext* opCtx, const NamespaceString& viewName) const {
+    invariant(opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
+    invariant(opCtx->lockState()->isCollectionLockedForMode(
+        NamespaceString(viewName.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
+
+    invariant(_viewsForDatabase.contains(viewName.db()));
+    const ViewsForDatabase& viewsForDb = *_getViewsForDatabase(opCtx, viewName.db());
+    viewsForDb.requireValidCatalog();
+
+    // Make sure the view exists before proceeding.
+    if (auto viewPtr = viewsForDb.lookup(viewName); !viewPtr) {
+        return {ErrorCodes::NamespaceNotFound,
+                str::stream() << "cannot drop missing view: " << viewName.ns()};
+    }
+
+    Status result = Status::OK();
+    {
+        IgnoreExternalViewChangesForDatabase ignore(opCtx, viewName.db());
+
+        ViewsForDatabase writable{viewsForDb};
+
+        writable.durable->remove(opCtx, viewName);
+        writable.viewGraph.remove(viewName);
+        writable.viewMap.erase(viewName.ns());
+        writable.stats = {};
+
+        // Reload the view catalog with the changes applied.
+        result = writable.reload(opCtx);
+        if (result.isOK()) {
+            auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
+            uncommittedCatalogUpdates.removeView(viewName);
+            uncommittedCatalogUpdates.replaceViewsForDatabase(viewName.db(), std::move(writable));
+
+            PublishCatalogUpdates::ensureRegisteredWithRecoveryUnit(opCtx,
+                                                                    uncommittedCatalogUpdates);
+        }
+    }
+
+    return result;
+}
+
+Status CollectionCatalog::reloadViews(OperationContext* opCtx, StringData dbName) const {
+    invariant(opCtx->lockState()->isCollectionLockedForMode(
+        NamespaceString(dbName, NamespaceString::kSystemDotViewsCollectionName), MODE_IS));
+
+    auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
+    if (uncommittedCatalogUpdates.shouldIgnoreExternalViewChanges(dbName)) {
+        return Status::OK();
+    }
+
+    LOGV2_DEBUG(22546, 1, "Reloading view catalog for database", "db"_attr = dbName);
+
+    // Create a copy of the ViewsForDatabase instance to modify it. Reset the views for this
+    // database, but preserve the DurableViewCatalog pointer.
+    auto it = _viewsForDatabase.find(dbName);
+    invariant(it != _viewsForDatabase.end());
+    ViewsForDatabase viewsForDb{it->second.durable};
+    viewsForDb.valid = false;
+    viewsForDb.viewGraphNeedsRefresh = true;
+    viewsForDb.viewMap.clear();
+    viewsForDb.stats = {};
+
+    auto status = viewsForDb.reload(opCtx);
+    CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
+        catalog._replaceViewsForDatabase(dbName, std::move(viewsForDb));
+    });
+
+    return status;
+}
 
 void CollectionCatalog::onCollectionRename(OperationContext* opCtx,
                                            Collection* coll,
@@ -572,14 +861,14 @@ void CollectionCatalog::onCollectionRename(OperationContext* opCtx,
     invariant(coll);
 
     auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
-    uncommittedCatalogUpdates.rename(coll, fromCollection);
+    uncommittedCatalogUpdates.renameCollection(coll, fromCollection);
 }
 
 void CollectionCatalog::dropCollection(OperationContext* opCtx, Collection* coll) const {
     invariant(coll);
 
     auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
-    uncommittedCatalogUpdates.drop(coll);
+    uncommittedCatalogUpdates.dropCollection(coll);
 
     // Requesting a writable collection normally ensures we have registered PublishCatalogUpdates
     // with the recovery unit. However, when the writable Collection was requested in Inplace mode
@@ -587,10 +876,22 @@ void CollectionCatalog::dropCollection(OperationContext* opCtx, Collection* coll
     PublishCatalogUpdates::ensureRegisteredWithRecoveryUnit(opCtx, uncommittedCatalogUpdates);
 }
 
+void CollectionCatalog::onOpenDatabase(OperationContext* opCtx,
+                                       StringData dbName,
+                                       ViewsForDatabase&& viewsForDb) {
+    invariant(opCtx->lockState()->isDbLockedForMode(dbName, MODE_IS));
+    uassert(ErrorCodes::AlreadyInitialized,
+            str::stream() << "Database " << dbName << " is already initialized",
+            _viewsForDatabase.find(dbName) == _viewsForDatabase.end());
+
+    _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.erase(tenantDbName.dbName());
 }
 
 void CollectionCatalog::onCloseCatalog(OperationContext* opCtx) {
@@ -630,7 +931,7 @@ Collection* CollectionCatalog::lookupCollectionByUUIDForMetadataWrite(OperationC
     }
 
     auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
-    auto [found, uncommittedPtr] = uncommittedCatalogUpdates.lookup(uuid);
+    auto [found, uncommittedPtr] = uncommittedCatalogUpdates.lookupCollection(uuid);
     // If UUID is managed by uncommittedCatalogUpdates return the pointer which will be nullptr in
     // case of a drop. We don't need to check UncommittedCollections as we will never share UUID for
     // a new Collection.
@@ -654,7 +955,7 @@ Collection* CollectionCatalog::lookupCollectionByUUIDForMetadataWrite(OperationC
     invariant(opCtx->lockState()->isCollectionLockedForMode(coll->ns(), MODE_X));
     auto cloned = coll->clone();
     auto ptr = cloned.get();
-    uncommittedCatalogUpdates.writable(std::move(cloned));
+    uncommittedCatalogUpdates.writableCollection(std::move(cloned));
 
     PublishCatalogUpdates::ensureRegisteredWithRecoveryUnit(opCtx, uncommittedCatalogUpdates);
 
@@ -663,7 +964,7 @@ Collection* CollectionCatalog::lookupCollectionByUUIDForMetadataWrite(OperationC
 
 CollectionPtr CollectionCatalog::lookupCollectionByUUID(OperationContext* opCtx, UUID uuid) const {
     auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
-    auto [found, uncommittedPtr] = uncommittedCatalogUpdates.lookup(uuid);
+    auto [found, uncommittedPtr] = uncommittedCatalogUpdates.lookupCollection(uuid);
     // If UUID is managed by uncommittedCatalogUpdates return the pointer which will be nullptr in
     // case of a drop. We don't need to check UncommittedCollections as we will never share UUID for
     // a new Collection.
@@ -709,7 +1010,7 @@ Collection* CollectionCatalog::lookupCollectionByNamespaceForMetadataWrite(
     }
 
     auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
-    auto [found, uncommittedPtr] = uncommittedCatalogUpdates.lookup(nss);
+    auto [found, uncommittedPtr] = uncommittedCatalogUpdates.lookupCollection(nss);
     // If uncommittedPtr is valid, found is always true. Return the pointer as the collection still
     // exists.
     if (uncommittedPtr) {
@@ -737,7 +1038,7 @@ Collection* CollectionCatalog::lookupCollectionByNamespaceForMetadataWrite(
     invariant(opCtx->lockState()->isCollectionLockedForMode(nss, MODE_X));
     auto cloned = coll->clone();
     auto ptr = cloned.get();
-    uncommittedCatalogUpdates.writable(std::move(cloned));
+    uncommittedCatalogUpdates.writableCollection(std::move(cloned));
 
     PublishCatalogUpdates::ensureRegisteredWithRecoveryUnit(opCtx, uncommittedCatalogUpdates);
 
@@ -747,7 +1048,7 @@ Collection* CollectionCatalog::lookupCollectionByNamespaceForMetadataWrite(
 CollectionPtr CollectionCatalog::lookupCollectionByNamespace(OperationContext* opCtx,
                                                              const NamespaceString& nss) const {
     auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
-    auto [found, uncommittedPtr] = uncommittedCatalogUpdates.lookup(nss);
+    auto [found, uncommittedPtr] = uncommittedCatalogUpdates.lookupCollection(nss);
     // If uncommittedPtr is valid, found is always true. Return the pointer as the collection still
     // exists.
     if (uncommittedPtr) {
@@ -775,7 +1076,7 @@ CollectionPtr CollectionCatalog::lookupCollectionByNamespace(OperationContext* o
 boost::optional<NamespaceString> CollectionCatalog::lookupNSSByUUID(OperationContext* opCtx,
                                                                     const UUID& uuid) const {
     auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
-    auto [found, uncommittedPtr] = uncommittedCatalogUpdates.lookup(uuid);
+    auto [found, uncommittedPtr] = uncommittedCatalogUpdates.lookupCollection(uuid);
     // If UUID is managed by uncommittedCatalogUpdates return its corresponding namespace if the
     // Collection exists, boost::none otherwise.
     if (found) {
@@ -810,7 +1111,7 @@ boost::optional<NamespaceString> CollectionCatalog::lookupNSSByUUID(OperationCon
 boost::optional<UUID> CollectionCatalog::lookupUUIDByNSS(OperationContext* opCtx,
                                                          const NamespaceString& nss) const {
     auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
-    auto [found, uncommittedPtr] = uncommittedCatalogUpdates.lookup(nss);
+    auto [found, uncommittedPtr] = uncommittedCatalogUpdates.lookupCollection(nss);
     if (uncommittedPtr) {
         return uncommittedPtr->uuid();
     }
@@ -831,6 +1132,58 @@ boost::optional<UUID> CollectionCatalog::lookupUUIDByNSS(OperationContext* opCtx
     return boost::none;
 }
 
+void CollectionCatalog::iterateViews(OperationContext* opCtx,
+                                     StringData dbName,
+                                     ViewIteratorCallback callback,
+                                     ViewCatalogLookupBehavior lookupBehavior) const {
+    auto viewsForDb = _getViewsForDatabase(opCtx, dbName);
+    if (!viewsForDb) {
+        return;
+    }
+
+    if (lookupBehavior != ViewCatalogLookupBehavior::kAllowInvalidViews) {
+        viewsForDb->requireValidCatalog();
+    }
+
+    for (auto&& view : viewsForDb->viewMap) {
+        if (!callback(*view.second)) {
+            break;
+        }
+    }
+}
+
+std::shared_ptr<const ViewDefinition> CollectionCatalog::lookupView(
+    OperationContext* opCtx, const NamespaceString& ns) const {
+    auto viewsForDb = _getViewsForDatabase(opCtx, ns.db());
+    if (!viewsForDb) {
+        return nullptr;
+    }
+
+    if (!viewsForDb->valid && opCtx->getClient()->isFromUserConnection()) {
+        // We want to avoid lookups on invalid collection names.
+        if (!NamespaceString::validCollectionName(ns.ns())) {
+            return nullptr;
+        }
+
+        // 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.
+        viewsForDb->requireValidCatalog();
+    }
+
+    return viewsForDb->lookup(ns);
+}
+
+std::shared_ptr<const ViewDefinition> CollectionCatalog::lookupViewWithoutValidatingDurable(
+    OperationContext* opCtx, const NamespaceString& ns) const {
+    auto viewsForDb = _getViewsForDatabase(opCtx, ns.db());
+    if (!viewsForDb) {
+        return nullptr;
+    }
+
+    return viewsForDb->lookup(ns);
+}
+
 NamespaceString CollectionCatalog::resolveNamespaceStringOrUUID(
     OperationContext* opCtx, NamespaceStringOrUUID nsOrUUID) const {
     if (auto& nss = nsOrUUID.nss()) {
@@ -942,11 +1295,23 @@ CollectionCatalog::Stats CollectionCatalog::getStats() const {
     return _stats;
 }
 
-CollectionCatalog::ViewCatalogSet CollectionCatalog::getViewCatalogDbNames() const {
+boost::optional<ViewsForDatabase::Stats> CollectionCatalog::getViewStatsForDatabase(
+    OperationContext* opCtx, StringData dbName) const {
+    auto viewsForDb = _getViewsForDatabase(opCtx, dbName);
+    if (!viewsForDb) {
+        return boost::none;
+    }
+    return viewsForDb->stats;
+}
+
+CollectionCatalog::ViewCatalogSet CollectionCatalog::getViewCatalogDbNames(
+    OperationContext* opCtx) const {
     ViewCatalogSet results;
-    for (const auto& dbNameViewSetPair : _views) {
-        results.insert(dbNameViewSetPair.first);
+    for (const auto& dbNameViewSetPair : _viewsForDatabase) {
+        // TODO (SERVER-63206): Return stored TenantDatabaseName
+        results.insert(TenantDatabaseName{boost::none, dbNameViewSetPair.first});
     }
+
     return results;
 }
 
@@ -955,26 +1320,13 @@ void CollectionCatalog::registerCollection(OperationContext* opCtx,
                                            std::shared_ptr<Collection> coll) {
     auto tenantNs = coll->tenantNs();
     auto tenantDbName = tenantNs.createTenantDatabaseName();
-    if (auto it = _views.find(tenantDbName); it != _views.end()) {
-        uassert(ErrorCodes::NamespaceExists,
-                str::stream() << "View already exists. NS: " << tenantNs,
-                !it->second.contains(tenantNs.getNss()));
-    }
-    if (_collections.find(tenantNs.getNss()) != _collections.end()) {
-        auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
-        auto [found, uncommittedPtr] = uncommittedCatalogUpdates.lookup(tenantNs.getNss());
+    if (NonExistenceType::kDropPending ==
+        _ensureNamespaceDoesNotExist(opCtx, tenantNs.getNss(), NamespaceType::kAll)) {
         // If we have an uncommitted drop of this collection we can defer the creation, the register
         // will happen in the same catalog write as the drop.
-        if (found && !uncommittedPtr) {
-            uncommittedCatalogUpdates.createAfterDrop(uuid, std::move(coll));
-            return;
-        }
-
-        LOGV2(20279,
-              "Conflicted creating a collection. ns: {coll_ns} ({coll_uuid}).",
-              "Conflicted creating a collection",
-              logAttrs(*coll));
-        throw WriteConflictException();
+        auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
+        uncommittedCatalogUpdates.createCollectionAfterDrop(uuid, std::move(coll));
+        return;
     }
 
     LOGV2_DEBUG(20280,
@@ -1057,6 +1409,62 @@ std::shared_ptr<Collection> CollectionCatalog::deregisterCollection(OperationCon
     return coll;
 }
 
+void CollectionCatalog::registerUncommittedView(OperationContext* opCtx,
+                                                const NamespaceString& nss) {
+    invariant(opCtx->lockState()->isCollectionLockedForMode(
+        NamespaceString(nss.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
+
+    // Since writing to system.views requires an X lock, we only need to cross-check collection
+    // namespaces here.
+    if (NonExistenceType::kDropPending ==
+        _ensureNamespaceDoesNotExist(opCtx, nss, NamespaceType::kCollection)) {
+        throw WriteConflictException();
+    }
+
+    _uncommittedViews.emplace(nss);
+}
+
+void CollectionCatalog::deregisterUncommittedView(const NamespaceString& nss) {
+    _uncommittedViews.erase(nss);
+}
+
+CollectionCatalog::NonExistenceType CollectionCatalog::_ensureNamespaceDoesNotExist(
+    OperationContext* opCtx, const NamespaceString& nss, NamespaceType type) const {
+    if (_collections.find(nss) != _collections.end()) {
+        auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
+        auto [found, uncommittedPtr] = uncommittedCatalogUpdates.lookupCollection(nss);
+        if (found && !uncommittedPtr) {
+            return NonExistenceType::kDropPending;
+        }
+
+        LOGV2(5725001,
+              "Conflicted registering namespace, already have a collection with the same namespace",
+              "nss"_attr = nss);
+        throw WriteConflictException();
+    }
+
+    if (type == NamespaceType::kAll) {
+        if (_uncommittedViews.contains(nss)) {
+            LOGV2(5725002,
+                  "Conflicted registering namespace, already have a view with the same namespace",
+                  "nss"_attr = nss);
+            throw WriteConflictException();
+        }
+
+        if (auto viewsForDb = _getViewsForDatabase(opCtx, nss.db())) {
+            if (viewsForDb->lookup(nss) != nullptr) {
+                LOGV2(
+                    5725003,
+                    "Conflicted registering namespace, already have a view with the same namespace",
+                    "nss"_attr = nss);
+                throw WriteConflictException();
+            }
+        }
+    }
+
+    return NonExistenceType::kNormal;
+}
+
 void CollectionCatalog::deregisterAllCollectionsAndViews() {
     LOGV2(20282, "Deregistering all the collections");
     for (auto& entry : _catalog) {
@@ -1071,42 +1479,28 @@ void CollectionCatalog::deregisterAllCollectionsAndViews() {
     _collections.clear();
     _orderedCollections.clear();
     _catalog.clear();
-    _views.clear();
+    _viewsForDatabase.clear();
     _stats = {};
 
     _resourceInformation.clear();
 }
 
-void CollectionCatalog::registerView(const NamespaceString& ns) {
-    if (_collections.contains(ns)) {
-        LOGV2(5706100, "Conflicted creating a view", logAttrs(ns));
-        throw WriteConflictException();
-    }
-
-    const TenantDatabaseName tenantDbName(boost::none, ns.db());
-    _views[tenantDbName].insert(ns);
-}
-void CollectionCatalog::deregisterView(const NamespaceString& ns) {
-    const TenantDatabaseName tenantDbName(boost::none, ns.db());
-    auto it = _views.find(tenantDbName);
-    if (it == _views.end()) {
-        return;
-    }
-
-    auto& viewsForDb = it->second;
-    viewsForDb.erase(ns);
-    if (viewsForDb.empty()) {
-        _views.erase(it);
-    }
-}
-
-void CollectionCatalog::replaceViewsForDatabase(const TenantDatabaseName& tenantDbName,
-                                                absl::flat_hash_set<NamespaceString> views) {
-    if (views.empty())
-        _views.erase(tenantDbName);
-    else {
-        _views[tenantDbName] = std::move(views);
-    }
+void CollectionCatalog::clearViews(OperationContext* opCtx, StringData dbName) const {
+    invariant(opCtx->lockState()->isCollectionLockedForMode(
+        NamespaceString(dbName, NamespaceString::kSystemDotViewsCollectionName), MODE_X));
+
+    auto it = _viewsForDatabase.find(dbName);
+    invariant(it != _viewsForDatabase.end());
+    ViewsForDatabase viewsForDb = it->second;
+
+    viewsForDb.viewMap.clear();
+    viewsForDb.viewGraph.clear();
+    viewsForDb.valid = true;
+    viewsForDb.viewGraphNeedsRefresh = false;
+    viewsForDb.stats = {};
+    CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
+        catalog._replaceViewsForDatabase(dbName, std::move(viewsForDb));
+    });
 }
 
 CollectionCatalog::iterator CollectionCatalog::begin(OperationContext* opCtx,
@@ -1172,6 +1566,79 @@ void CollectionCatalog::addResource(const ResourceId& rid, const std::string& en
     namespaces.insert(entry);
 }
 
+boost::optional<const ViewsForDatabase&> CollectionCatalog::_getViewsForDatabase(
+    OperationContext* opCtx, StringData dbName) const {
+    auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
+    auto uncommittedViews = uncommittedCatalogUpdates.getViewsForDatabase(dbName);
+    if (uncommittedViews) {
+        return uncommittedViews;
+    }
+
+    auto it = _viewsForDatabase.find(dbName);
+    if (it == _viewsForDatabase.end()) {
+        return boost::none;
+    }
+    return it->second;
+}
+
+void CollectionCatalog::_replaceViewsForDatabase(StringData dbName, ViewsForDatabase&& views) {
+    _viewsForDatabase[dbName] = std::move(views);
+}
+
+Status CollectionCatalog::_createOrUpdateView(
+    OperationContext* opCtx,
+    const NamespaceString& viewName,
+    const NamespaceString& viewOn,
+    const BSONArray& pipeline,
+    const ViewsForDatabase::PipelineValidatorFn& pipelineValidator,
+    std::unique_ptr<CollatorInterface> collator,
+    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. If the
+    // collation is empty, omit it from the definition altogether.
+    BSONObjBuilder viewDefBuilder;
+    viewDefBuilder.append("_id", viewName.ns());
+    viewDefBuilder.append("viewOn", viewOn.coll());
+    viewDefBuilder.append("pipeline", pipeline);
+    if (collator) {
+        viewDefBuilder.append("collation", collator->getSpec().toBSON());
+    }
+
+    BSONObj ownedPipeline = pipeline.getOwned();
+    auto view = std::make_shared<ViewDefinition>(
+        viewName.db(), viewName.coll(), viewOn.coll(), ownedPipeline, std::move(collator));
+
+    // 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;
+    }
+
+    viewsForDb.durable->upsert(opCtx, viewName, viewDefBuilder.obj());
+
+    viewsForDb.viewMap.clear();
+    viewsForDb.valid = false;
+    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);
+        uncommittedCatalogUpdates.replaceViewsForDatabase(viewName.db(), std::move(viewsForDb));
+
+        PublishCatalogUpdates::ensureRegisteredWithRecoveryUnit(opCtx, uncommittedCatalogUpdates);
+    }
+
+    return res;
+}
+
 CollectionCatalogStasher::CollectionCatalogStasher(OperationContext* opCtx)
     : _opCtx(opCtx), _stashed(false) {}
 
diff --git a/src/mongo/db/catalog/collection_catalog.h b/src/mongo/db/catalog/collection_catalog.h
index f8ec0db2533..9bb23cfa5a8 100644
--- a/src/mongo/db/catalog/collection_catalog.h
+++ b/src/mongo/db/catalog/collection_catalog.h
@@ -34,9 +34,11 @@
 #include <set>
 
 #include "mongo/db/catalog/collection.h"
+#include "mongo/db/catalog/views_for_database.h"
 #include "mongo/db/profile_filter.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/tenant_database_name.h"
+#include "mongo/db/views/view.h"
 #include "mongo/stdx/unordered_map.h"
 #include "mongo/util/uuid.h"
 
@@ -44,12 +46,14 @@ namespace mongo {
 
 class CollectionCatalog;
 class Database;
+class UncommittedCatalogUpdates;
 
 class CollectionCatalog {
     friend class iterator;
 
 public:
     using CollectionInfoFn = std::function<bool(const CollectionPtr& collection)>;
+    using ViewIteratorCallback = std::function<bool(const ViewDefinition& view)>;
 
     /**
      * Defines lifetime and behavior of writable Collections.
@@ -148,6 +152,58 @@ public:
     static void write(OperationContext* opCtx, CatalogWriteFn job);
 
     /**
+     * Create a new view 'viewName' with contents defined by running the specified aggregation
+     * 'pipeline' with collation 'collation' on a collection or view 'viewOn'.
+     *
+     * Must be in WriteUnitOfWork. View creation rolls back if the unit of work aborts.
+     *
+     * Caller must ensure corresponding database exists.
+     */
+    Status createView(OperationContext* opCtx,
+                      const NamespaceString& viewName,
+                      const NamespaceString& viewOn,
+                      const BSONArray& pipeline,
+                      const BSONObj& collation,
+                      const ViewsForDatabase::PipelineValidatorFn& pipelineValidator) const;
+
+    /**
+     * 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.
+     */
+    Status dropView(OperationContext* opCtx, const NamespaceString& viewName) const;
+
+    /**
+     * 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.
+     */
+    Status modifyView(OperationContext* opCtx,
+                      const NamespaceString& viewName,
+                      const NamespaceString& viewOn,
+                      const BSONArray& pipeline,
+                      const ViewsForDatabase::PipelineValidatorFn& pipelineValidator) const;
+
+    /**
+     * Reloads the in-memory state of the view catalog from the 'system.views' collection. The
+     * durable view definitions will be validated. Reading stops on the first invalid entry with
+     * errors logged and returned. Performs no cycle detection, etc.
+     *
+     * This is implicitly called by other methods when write operations are performed on the
+     * view catalog, on external changes to the 'system.views' collection and on the first
+     * opening of a database.
+     *
+     * Callers must re-fetch the catalog to observe changes.
+     *
+     * Requires an IS lock on the 'system.views' collection'.
+     */
+    Status reloadViews(OperationContext* opCtx, StringData dbName) const;
+
+    /**
      * This function is responsible for safely tracking a Collection rename within a
      * WriteUnitOfWork.
      *
@@ -166,6 +222,17 @@ public:
      */
     void dropCollection(OperationContext* opCtx, Collection* coll) const;
 
+    /**
+     * Initializes view records for database 'dbName'. Can throw a 'WriteConflictException' if this
+     * database has already been initialized.
+     */
+    void onOpenDatabase(OperationContext* opCtx, StringData dbName, ViewsForDatabase&& viewsForDb);
+
+    /**
+     * Removes the view records associated with 'tenantDbName', if any, from the in-memory
+     * representation of the catalog. Should be called when Database instance is closed. Requires X
+     * lock on database namespace.
+     */
     void onCloseDatabase(OperationContext* opCtx, TenantDatabaseName tenantDbName);
 
     /**
@@ -181,29 +248,27 @@ public:
     std::shared_ptr<Collection> deregisterCollection(OperationContext* opCtx, const UUID& uuid);
 
     /**
-     * Deregister all the collection objects and view namespaces.
+     * Create a temporary record of an uncommitted view namespace to aid in detecting a simultaneous
+     * attempt to create a collection with the same namespace.
      */
-    void deregisterAllCollectionsAndViews();
+    void registerUncommittedView(OperationContext* opCtx, const NamespaceString& nss);
 
     /**
-     * Register the namespace to be used as a view.
-     *
-     * Throws WriteConflictException if namespace is used by a Collection
+     * Remove the temporary record for an uncommitted view namespace, either on commit or rollback.
      */
-    void registerView(const NamespaceString& ns);
+    void deregisterUncommittedView(const NamespaceString& nss);
 
     /**
-     * Deregister the namespace from being used as a view.
+     * Deregister all the collection objects and view namespaces.
      */
-    void deregisterView(const NamespaceString& ns);
+    void deregisterAllCollectionsAndViews();
 
     /**
-     * Sets all namespaces used by views for a database. Does not validate if they are used by
-     * Collections. When creating new view its namespace should be registered with registerView()
-     * above.
+     * Clears the in-memory state for the views associated with a particular database.
+     *
+     * Callers must re-fetch the catalog to observe changes.
      */
-    void replaceViewsForDatabase(const TenantDatabaseName& tenantDbName,
-                                 absl::flat_hash_set<NamespaceString> views);
+    void clearViews(OperationContext* opCtx, StringData dbName) const;
 
     /**
      * This function gets the Collection pointer that corresponds to the UUID.
@@ -273,6 +338,36 @@ public:
                                           const NamespaceString& nss) const;
 
     /**
+     * Iterates through the views in the catalog associated with database `dbName`, applying
+     * 'callback' to each view.  If the 'callback' returns false, the iterator exits early.
+     *
+     * Caller must ensure corresponding database exists.
+     */
+    void iterateViews(
+        OperationContext* opCtx,
+        StringData dbName,
+        ViewIteratorCallback callback,
+        ViewCatalogLookupBehavior lookupBehavior = ViewCatalogLookupBehavior::kValidateViews) const;
+
+    /**
+     * 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> lookupView(OperationContext* opCtx,
+                                                     const NamespaceString& nss) const;
+
+    /**
+     * 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> lookupViewWithoutValidatingDurable(
+        OperationContext* opCtx, const NamespaceString& nss) const;
+
+    /**
      * Without acquiring any locks resolves the given NamespaceStringOrUUID to an actual namespace.
      * Throws NamespaceNotFound if the collection UUID cannot be resolved to a name, or if the UUID
      * can be resolved, but the resulting collection is in the wrong database.
@@ -365,14 +460,20 @@ public:
     Stats getStats() const;
 
     /**
+     * Returns view statistics for the specified database.
+     */
+    boost::optional<ViewsForDatabase::Stats> getViewStatsForDatabase(OperationContext* opCtx,
+                                                                     StringData dbName) const;
+
+    /**
      * Returns a set of databases, by name, that have view catalogs.
      */
     using ViewCatalogSet = absl::flat_hash_set<TenantDatabaseName>;
-    ViewCatalogSet getViewCatalogDbNames() const;
+    ViewCatalogSet getViewCatalogDbNames(OperationContext* opCtx) const;
 
     /**
-     * Puts the catalog in closed state. In this state, the lookupNSSByUUID method will fall back
-     * to the pre-close state to resolve queries for currently unknown UUIDs. This allows processes,
+     * Puts the catalog in closed state. In this state, the lookupNSSByUUID method will fall back to
+     * the pre-close state to resolve queries for currently unknown UUIDs. This allows processes,
      * like authorization and replication, which need to do lookups outside of database locks, to
      * proceed.
      *
@@ -381,7 +482,7 @@ public:
     void onCloseCatalog(OperationContext* opCtx);
 
     /**
-     * Puts the catatlog back in open state, removing the pre-close state. See onCloseCatalog.
+     * Puts the catalog back in open state, removing the pre-close state. See onCloseCatalog.
      *
      * Must be called with the global lock acquired in exclusive mode.
      */
@@ -403,8 +504,8 @@ public:
 
     /**
      * Lookup the name of a resource by its ResourceId. If there are multiple namespaces mapped to
-     * the same ResourceId entry, we return the boost::none for those namespaces until there is
-     * only one namespace in the set. If the ResourceId is not found, boost::none is returned.
+     * the same ResourceId entry, we return the boost::none for those namespaces until there is only
+     * one namespace in the set. If the ResourceId is not found, boost::none is returned.
      */
     boost::optional<std::string> lookupResourceName(const ResourceId& rid) const;
 
@@ -425,6 +526,48 @@ private:
     std::shared_ptr<Collection> _lookupCollectionByUUID(UUID uuid) const;
 
     /**
+     * Retrieves the views for a given database, including any uncommitted changes for this
+     * operation.
+     */
+    boost::optional<const ViewsForDatabase&> _getViewsForDatabase(OperationContext* opCtx,
+                                                                  StringData dbName) const;
+
+    /**
+     * Sets all namespaces used by views for a database. Will uassert if there is a conflicting
+     * collection name in the catalog.
+     */
+    void _replaceViewsForDatabase(StringData dbName, ViewsForDatabase&& views);
+
+    /**
+     * Helper to take care of shared functionality for 'createView(...)' and 'modifyView(...)'.
+     */
+    Status _createOrUpdateView(OperationContext* opCtx,
+                               const NamespaceString& viewName,
+                               const NamespaceString& viewOn,
+                               const BSONArray& pipeline,
+                               const ViewsForDatabase::PipelineValidatorFn& pipelineValidator,
+                               std::unique_ptr<CollatorInterface> collator,
+                               ViewsForDatabase&& viewsForDb) const;
+
+    /**
+     * Throws 'WriteConflictException' if given namespace is already registered with the catalog, as
+     * either a view or collection. In the case of an collection drop (by the calling thread) that
+     * has not been committed yet, it will not throw, but it will return
+     * 'NonExistenceType::kDropPending' to distinguish from the case that the namespace is simply
+     * not registered with the catalog at all. The results will include namespaces which have been
+     * registered by preCommitHooks on other threads, but which have not truly been committed yet.
+     *
+     * If 'type' is set to 'NamespaceType::kCollection', we will only check for collisions with
+     * collections. If set to 'NamespaceType::kAll', we will check against both collections and
+     * views.
+     */
+    enum class NonExistenceType { kDropPending, kNormal };
+    enum class NamespaceType { kAll, kCollection };
+    NonExistenceType _ensureNamespaceDoesNotExist(OperationContext* opCtx,
+                                                  const NamespaceString& nss,
+                                                  NamespaceType type) const;
+
+    /**
      * When present, indicates that the catalog is in closed state, and contains a map from UUID
      * to pre-close NSS. See also onCloseCatalog.
      */
@@ -435,14 +578,16 @@ private:
         std::map<std::pair<TenantDatabaseName, UUID>, std::shared_ptr<Collection>>;
     using NamespaceCollectionMap =
         stdx::unordered_map<NamespaceString, std::shared_ptr<Collection>>;
+    using UncommittedViewsSet = stdx::unordered_set<NamespaceString>;
     using DatabaseProfileSettingsMap = StringMap<ProfileSettings>;
 
     CollectionCatalogMap _catalog;
     OrderedCollectionMap _orderedCollections;  // Ordered by <tenantDbName, collUUID> pair
     NamespaceCollectionMap _collections;
+    UncommittedViewsSet _uncommittedViews;
 
-    // Map of database names to a set of their views. Only databases with views are present.
-    absl::flat_hash_map<TenantDatabaseName, absl::flat_hash_set<NamespaceString>> _views;
+    // Map of database names to their corresponding views and other associated state.
+    StringMap<ViewsForDatabase> _viewsForDatabase;
 
     // Incremented whenever the CollectionCatalog gets closed and reopened (onCloseCatalog and
     // onOpenCatalog).
diff --git a/src/mongo/db/catalog/collection_catalog_helper.cpp b/src/mongo/db/catalog/collection_catalog_helper.cpp
index 6c7dd9e19bc..c094444b9e3 100644
--- a/src/mongo/db/catalog/collection_catalog_helper.cpp
+++ b/src/mongo/db/catalog/collection_catalog_helper.cpp
@@ -31,7 +31,6 @@
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/concurrency/d_concurrency.h"
-#include "mongo/db/views/view_catalog.h"
 
 namespace mongo {
 
@@ -40,12 +39,13 @@ MONGO_FAIL_POINT_DEFINE(hangBeforeGettingNextCollection);
 namespace catalog {
 
 Status checkIfNamespaceExists(OperationContext* opCtx, const NamespaceString& nss) {
-    if (CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, nss)) {
+    auto catalog = CollectionCatalog::get(opCtx);
+    if (catalog->lookupCollectionByNamespace(opCtx, nss)) {
         return Status(ErrorCodes::NamespaceExists,
                       str::stream() << "Collection " << nss.ns() << " already exists.");
     }
 
-    auto view = ViewCatalog::get(opCtx)->lookup(opCtx, nss);
+    auto view = catalog->lookupView(opCtx, nss);
     if (!view)
         return Status::OK();
 
diff --git a/src/mongo/db/catalog/collection_compact.cpp b/src/mongo/db/catalog/collection_compact.cpp
index 9a5cb25d52a..4fed779eaae 100644
--- a/src/mongo/db/catalog/collection_compact.cpp
+++ b/src/mongo/db/catalog/collection_compact.cpp
@@ -40,7 +40,6 @@
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/index_builds_coordinator.h"
 #include "mongo/db/operation_context.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/assert_util.h"
 
@@ -59,7 +58,7 @@ CollectionPtr getCollectionForCompact(OperationContext* opCtx,
 
     if (!collection) {
         std::shared_ptr<const ViewDefinition> view =
-            ViewCatalog::get(opCtx)->lookup(opCtx, collectionNss);
+            collectionCatalog->lookupView(opCtx, collectionNss);
         uassert(ErrorCodes::CommandNotSupportedOnView, "can't compact a view", !view);
         uasserted(ErrorCodes::NamespaceNotFound, "collection does not exist");
     }
diff --git a/src/mongo/db/catalog/collection_validation.cpp b/src/mongo/db/catalog/collection_validation.cpp
index 51ebcc3bc37..15d7a7f0e72 100644
--- a/src/mongo/db/catalog/collection_validation.cpp
+++ b/src/mongo/db/catalog/collection_validation.cpp
@@ -48,7 +48,6 @@
 #include "mongo/db/record_id_helpers.h"
 #include "mongo/db/storage/key_string.h"
 #include "mongo/db/storage/storage_parameters_gen.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/fail_point.h"
 #include "mongo/util/scopeguard.h"
diff --git a/src/mongo/db/catalog/create_collection.cpp b/src/mongo/db/catalog/create_collection.cpp
index ae9d3d4fa04..c21002851d8 100644
--- a/src/mongo/db/catalog/create_collection.cpp
+++ b/src/mongo/db/catalog/create_collection.cpp
@@ -59,7 +59,6 @@
 #include "mongo/db/tenant_database_name.h"
 #include "mongo/db/tenant_namespace.h"
 #include "mongo/db/timeseries/timeseries_options.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/idl/command_generic_argument.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/fail_point.h"
diff --git a/src/mongo/db/catalog/database_holder.h b/src/mongo/db/catalog/database_holder.h
index 484d8dc6cc2..5b06a1765a7 100644
--- a/src/mongo/db/catalog/database_holder.h
+++ b/src/mongo/db/catalog/database_holder.h
@@ -75,19 +75,6 @@ public:
                           const TenantDatabaseName& tenantDbName) const = 0;
 
     /**
-     * Fetches the ViewCatalog decorating the Database matching 'tenantDbName', or returns nullptr
-     * if the database does not exist. The returned ViewCatalog is safe to access without a lock
-     * because it is held as a shared_ptr.
-     *
-     * The ViewCatalog must be fetched through this interface if the caller holds no database lock
-     * to ensure the Database object is safe to access. This class' internal mutex provides
-     * concurrency protection around looking up and accessing the Database object matching
-     * 'tenantDbName'.
-     */
-    virtual std::shared_ptr<const ViewCatalog> getViewCatalog(
-        OperationContext* opCtx, const TenantDatabaseName& tenantDbName) const = 0;
-
-    /**
      * Retrieves a database reference if it is already opened, or opens it if it hasn't been
      * opened/created yet. Must be called with the database locked in X-mode.
      *
diff --git a/src/mongo/db/catalog/database_holder_impl.cpp b/src/mongo/db/catalog/database_holder_impl.cpp
index a8c45cd732d..28e44082c2a 100644
--- a/src/mongo/db/catalog/database_holder_impl.cpp
+++ b/src/mongo/db/catalog/database_holder_impl.cpp
@@ -44,7 +44,6 @@
 #include "mongo/db/service_context.h"
 #include "mongo/db/stats/top.h"
 #include "mongo/db/storage/storage_engine.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 
 namespace mongo {
@@ -75,21 +74,8 @@ bool DatabaseHolderImpl::dbExists(OperationContext* opCtx,
             NamespaceString::validDBName(tenantDbName.dbName(),
                                          NamespaceString::DollarInDbNameBehavior::Allow));
     stdx::lock_guard<SimpleMutex> lk(_m);
-    return _dbs.find(tenantDbName) != _dbs.end();
-}
-
-std::shared_ptr<const ViewCatalog> DatabaseHolderImpl::getViewCatalog(
-    OperationContext* opCtx, const TenantDatabaseName& tenantDbName) const {
-    stdx::lock_guard<SimpleMutex> lk(_m);
-    DBs::const_iterator it = _dbs.find(tenantDbName);
-    if (it != _dbs.end()) {
-        const Database* db = it->second;
-        if (db) {
-            return ViewCatalog::get(opCtx);
-        }
-    }
-
-    return nullptr;
+    auto it = _dbs.find(tenantDbName);
+    return it != _dbs.end() && it->second != nullptr;
 }
 
 std::set<TenantDatabaseName> DatabaseHolderImpl::_getNamesWithConflictingCasing_inlock(
@@ -139,9 +125,8 @@ Database* DatabaseHolderImpl::openDb(OperationContext* opCtx,
     if (auto db = _dbs[tenantDbName])
         return db;
 
-    std::unique_ptr<DatabaseImpl> newDb;
     // We've inserted a nullptr entry for dbname: make sure to remove it on unsuccessful exit.
-    ScopeGuard removeDbGuard([this, &lk, &newDb, opCtx, tenantDbName] {
+    ScopeGuard removeDbGuard([this, &lk, opCtx, tenantDbName] {
         if (!lk.owns_lock())
             lk.lock();
         auto it = _dbs.find(tenantDbName);
@@ -149,9 +134,6 @@ Database* DatabaseHolderImpl::openDb(OperationContext* opCtx,
         if (it != _dbs.end() && !it->second) {
             _dbs.erase(it);
         }
-        if (newDb) {
-            ViewCatalog::unregisterDatabase(opCtx, newDb.get());
-        }
 
         // In case anyone else is trying to open the same DB simultaneously and waiting on our
         // result, we should notify them we failed and let them try in our place.
@@ -176,7 +158,7 @@ Database* DatabaseHolderImpl::openDb(OperationContext* opCtx,
             *justCreated = true;
     }
 
-    newDb = std::make_unique<DatabaseImpl>(tenantDbName);
+    std::unique_ptr<DatabaseImpl> newDb = std::make_unique<DatabaseImpl>(tenantDbName);
     Status status = newDb->init(opCtx);
     while (!status.isOK()) {
         // If we get here, then initializing the database failed because another concurrent writer
@@ -296,25 +278,13 @@ void DatabaseHolderImpl::close(OperationContext* opCtx, const TenantDatabaseName
                                          NamespaceString::DollarInDbNameBehavior::Allow));
     invariant(opCtx->lockState()->isDbLockedForMode(tenantDbName.dbName(), MODE_X));
 
-    stdx::unique_lock<SimpleMutex> lk(_m);
+    stdx::lock_guard<SimpleMutex> lk(_m);
 
     DBs::const_iterator it = _dbs.find(tenantDbName);
     if (it == _dbs.end()) {
         return;
     }
-
-    // Unlock to unregister this database from the ViewCatalog, then reacquire the lock.
     auto db = it->second;
-    lk.unlock();
-    ViewCatalog::unregisterDatabase(opCtx, db);
-    lk.lock();
-
-    // It's possible another thread altered the record before we reacquired the lock, so make sure
-    // we still have the same database.
-    it = _dbs.find(tenantDbName);
-    if (it == _dbs.end() || db != it->second) {
-        return;
-    }
 
     LOGV2_DEBUG(20311, 2, "DatabaseHolder::close", "db"_attr = tenantDbName);
 
diff --git a/src/mongo/db/catalog/database_holder_impl.h b/src/mongo/db/catalog/database_holder_impl.h
index 36aad9b731e..849ba757e19 100644
--- a/src/mongo/db/catalog/database_holder_impl.h
+++ b/src/mongo/db/catalog/database_holder_impl.h
@@ -46,9 +46,6 @@ public:
 
     bool dbExists(OperationContext* opCtx, const TenantDatabaseName& tenantDbName) const override;
 
-    std::shared_ptr<const ViewCatalog> getViewCatalog(
-        OperationContext* opCtx, const TenantDatabaseName& tenantDbName) const override;
-
     Database* openDb(OperationContext* opCtx,
                      const TenantDatabaseName& tenantDbName,
                      bool* justCreated = nullptr) override;
diff --git a/src/mongo/db/catalog/database_holder_mock.h b/src/mongo/db/catalog/database_holder_mock.h
index 8086e0d9164..9ef8c2229f7 100644
--- a/src/mongo/db/catalog/database_holder_mock.h
+++ b/src/mongo/db/catalog/database_holder_mock.h
@@ -46,11 +46,6 @@ public:
         return false;
     }
 
-    std::shared_ptr<const ViewCatalog> getViewCatalog(
-        OperationContext* const opCtx, const TenantDatabaseName& tenantDbName) const override {
-        return nullptr;
-    }
-
     Database* openDb(OperationContext* opCtx,
                      const TenantDatabaseName& tenantDbName,
                      bool* justCreated = nullptr) override {
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index 79ff87bb762..16d77cf9f14 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -76,7 +76,7 @@
 #include "mongo/db/storage/storage_options.h"
 #include "mongo/db/storage/storage_util.h"
 #include "mongo/db/system_index.h"
-#include "mongo/db/views/view_catalog.h"
+#include "mongo/db/views/view_catalog_helpers.h"
 #include "mongo/logv2/log.h"
 #include "mongo/platform/random.h"
 #include "mongo/util/assert_util.h"
@@ -167,12 +167,6 @@ Status DatabaseImpl::init(OperationContext* const opCtx) {
         uasserted(10028, status.toString());
     }
 
-    auto durableViewCatalog = std::make_unique<DurableViewCatalogImpl>(this);
-    status = ViewCatalog::registerDatabase(opCtx, _name.dbName(), std::move(durableViewCatalog));
-    if (!status.isOK()) {
-        return status;
-    }
-
     auto catalog = CollectionCatalog::get(opCtx);
     for (const auto& uuid : catalog->getAllCollectionUUIDsFromDb(_name)) {
         CollectionWriter collection(
@@ -189,28 +183,43 @@ Status DatabaseImpl::init(OperationContext* const opCtx) {
 
     // When in repair mode, record stores are not loaded. Thus the ViewsCatalog cannot be reloaded.
     if (!storageGlobalParams.repair) {
-        // At construction time of the viewCatalog, the CollectionCatalog map wasn't initialized
-        // yet, so no system.views collection would be found. Now that we're sufficiently
-        // initialized, reload the viewCatalog to populate its in-memory state. If there are
-        // problems with the catalog contents as might be caused by incorrect mongod versions or
-        // similar, they are found right away.
+        // At construction time of this DatabaseImpl, the CollectionCatalog map wasn't populated
+        // with collections for this database yet, so no system.views collection would be found to
+        // populate the views. Now that we've loaded the collections, reload the view definitions
+        // from system.views to populate the views portion of the CollectionCatalog. If there are
+        // problems with the durable catalog contents, as might be caused by incorrect mongod
+        // versions or similar, they are found right away.
         //
-        // We take an IS lock here because the ViewCatalog::reload API requires it for other uses.
-        // Realistically no one else can be accessing the collection, and there's no chance of this
-        // blocking.
-        Lock::CollectionLock systemViewsLock(
-            opCtx,
-            NamespaceString(_name.dbName(), NamespaceString::kSystemDotViewsCollectionName),
-            MODE_IS);
-        Status reloadStatus = ViewCatalog::reload(
-            opCtx, _name.dbName(), ViewCatalogLookupBehavior::kValidateDurableViews);
-        if (!reloadStatus.isOK()) {
-            LOGV2_WARNING_OPTIONS(20326,
-                                  {logv2::LogTag::kStartupWarnings},
-                                  "Unable to parse views; remove any invalid views "
-                                  "from the collection to restore server functionality",
-                                  "error"_attr = redact(reloadStatus),
-                                  "namespace"_attr = _viewsName);
+        // Even though no one can be writing to system.views at this point, we must take an IS lock
+        // because the ViewsForDatabase::reload API requires it for other uses.
+        try {
+            Lock::CollectionLock systemViewsLock(
+                opCtx,
+                NamespaceString(_name.dbName(), NamespaceString::kSystemDotViewsCollectionName),
+                MODE_IS);
+            ViewsForDatabase viewsForDb{std::make_unique<DurableViewCatalogImpl>(this)};
+            Status reloadStatus = viewsForDb.reload(opCtx);
+            if (!reloadStatus.isOK()) {
+                LOGV2_WARNING_OPTIONS(20326,
+                                      {logv2::LogTag::kStartupWarnings},
+                                      "Unable to parse views; remove any invalid views "
+                                      "from the collection to restore server functionality",
+                                      "error"_attr = redact(reloadStatus),
+                                      "namespace"_attr = _viewsName);
+            }
+
+            CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
+                catalog.onOpenDatabase(opCtx, _name.dbName(), std::move(viewsForDb));
+            });
+        } catch (DBException& ex) {
+            // Another operation may have tried to simultaneously open the database and register it
+            // with the CollectionCatalog. If that's the case, error out here and handle the
+            // conflict one level up.
+            if (ex.code() == ErrorCodes::AlreadyInitialized) {
+                return ex.toStatus();
+            }
+
+            throw;
         }
     }
 
@@ -218,10 +227,12 @@ Status DatabaseImpl::init(OperationContext* const opCtx) {
     if (storageGlobalParams.restore) {
         invariant(opCtx->lockState()->isW());
 
+        // Refresh our copy of the catalog, since we may have modified it above.
+        catalog = CollectionCatalog::get(opCtx);
         try {
-            auto viewCatalog = ViewCatalog::get(opCtx);
-            viewCatalog->iterate(_name.dbName(), [&](const ViewDefinition& view) {
-                auto swResolvedView = viewCatalog->resolveView(opCtx, view.name(), boost::none);
+            catalog->iterateViews(opCtx, _name.dbName(), [&](const ViewDefinition& view) {
+                auto swResolvedView =
+                    view_catalog_helpers::resolveView(opCtx, catalog, view.name(), boost::none);
                 if (!swResolvedView.isOK()) {
                     LOGV2_WARNING(6260802,
                                   "Could not resolve view during restore",
@@ -246,7 +257,7 @@ Status DatabaseImpl::init(OperationContext* const opCtx) {
                       "resolvedNs"_attr = resolvedNs);
 
                 WriteUnitOfWork wuow(opCtx);
-                Status status = viewCatalog->dropView(opCtx, view.name());
+                Status status = catalog->dropView(opCtx, view.name());
                 if (!status.isOK()) {
                     LOGV2_WARNING(6260804,
                                   "Failed to remove view on unrestored collection",
@@ -353,10 +364,11 @@ void DatabaseImpl::getStats(OperationContext* opCtx,
         });
 
 
-    ViewCatalog::get(opCtx)->iterate(name().dbName(), [&](const ViewDefinition& view) {
-        nViews += 1;
-        return true;
-    });
+    CollectionCatalog::get(opCtx)->iterateViews(
+        opCtx, name().dbName(), [&](const ViewDefinition& view) {
+            nViews += 1;
+            return true;
+        });
 
     output->appendNumber("collections", nCollections);
     output->appendNumber("views", nViews);
@@ -406,7 +418,7 @@ Status DatabaseImpl::dropView(OperationContext* opCtx, NamespaceString viewName)
     dassert(opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
     dassert(opCtx->lockState()->isCollectionLockedForMode(NamespaceString(_viewsName), MODE_X));
 
-    Status status = ViewCatalog::dropView(opCtx, viewName);
+    Status status = CollectionCatalog::get(opCtx)->dropView(opCtx, viewName);
     Top::get(opCtx->getServiceContext()).collectionDropped(viewName);
     return status;
 }
@@ -417,7 +429,9 @@ Status DatabaseImpl::dropCollection(OperationContext* opCtx,
     // Cannot drop uncommitted collections.
     invariant(!UncommittedCollections::getForTxn(opCtx, nss));
 
-    if (!CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, nss)) {
+    auto catalog = CollectionCatalog::get(opCtx);
+
+    if (!catalog->lookupCollectionByNamespace(opCtx, nss)) {
         // Collection doesn't exist so don't bother validating if it can be dropped.
         return Status::OK();
     }
@@ -426,13 +440,12 @@ Status DatabaseImpl::dropCollection(OperationContext* opCtx,
 
     if (nss.isSystem()) {
         if (nss.isSystemDotProfile()) {
-            if (CollectionCatalog::get(opCtx)->getDatabaseProfileLevel(_name.dbName()) != 0)
+            if (catalog->getDatabaseProfileLevel(_name.dbName()) != 0)
                 return Status(ErrorCodes::IllegalOperation,
                               "turn off profiling before dropping system.profile collection");
         } else if (nss.isSystemDotViews()) {
             if (!MONGO_unlikely(allowSystemViewsDrop.shouldFail())) {
-                const auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, _name);
-                const auto viewStats = viewCatalog->getStats(_name.dbName());
+                const auto viewStats = catalog->getViewStatsForDatabase(opCtx, _name.dbName());
                 uassert(ErrorCodes::CommandFailed,
                         str::stream() << "cannot drop collection " << nss
                                       << " when time-series collections are present.",
@@ -758,7 +771,12 @@ Status DatabaseImpl::createView(OperationContext* opCtx,
         status = {ErrorCodes::InvalidNamespace,
                   str::stream() << "invalid namespace name for a view: " + viewName.toString()};
     } else {
-        status = ViewCatalog::createView(opCtx, viewName, viewOnNss, pipeline, options.collation);
+        status = CollectionCatalog::get(opCtx)->createView(opCtx,
+                                                           viewName,
+                                                           viewOnNss,
+                                                           pipeline,
+                                                           options.collation,
+                                                           view_catalog_helpers::validatePipeline);
     }
 
     audit::logCreateView(
diff --git a/src/mongo/db/catalog/drop_collection.cpp b/src/mongo/db/catalog/drop_collection.cpp
index d50fab0de91..58b546bcbf5 100644
--- a/src/mongo/db/catalog/drop_collection.cpp
+++ b/src/mongo/db/catalog/drop_collection.cpp
@@ -34,6 +34,7 @@
 #include "mongo/db/catalog/drop_collection.h"
 
 #include "mongo/db/audit.h"
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/collection_uuid_mismatch.h"
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/catalog/uncommitted_collections.h"
@@ -46,7 +47,6 @@
 #include "mongo/db/s/collection_sharding_state.h"
 #include "mongo/db/server_options.h"
 #include "mongo/db/service_context.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/fail_point.h"
 
@@ -87,7 +87,8 @@ Status _dropView(OperationContext* opCtx,
         return ex.toStatus();
     }
 
-    auto view = ViewCatalog::get(opCtx)->lookupWithoutValidatingDurableViews(opCtx, collectionName);
+    auto view =
+        CollectionCatalog::get(opCtx)->lookupViewWithoutValidatingDurable(opCtx, collectionName);
     if (!view) {
         Status status = Status(ErrorCodes::NamespaceNotFound, "ns not found");
         audit::logDropView(opCtx->getClient(), collectionName, "", {}, status.code());
@@ -95,7 +96,7 @@ Status _dropView(OperationContext* opCtx,
     }
 
     // Validates the view or throws an "invalid view" error.
-    ViewCatalog::get(opCtx)->lookup(opCtx, collectionName);
+    CollectionCatalog::get(opCtx)->lookupView(opCtx, collectionName);
 
     // Operations all lock system.views in the end to prevent deadlock.
     Lock::CollectionLock systemViewsLock(opCtx, db->getSystemViewsName(), MODE_X);
@@ -384,8 +385,8 @@ Status _dropCollection(OperationContext* opCtx,
                     false /* appendNs */);
             };
 
-            auto view =
-                ViewCatalog::get(opCtx)->lookupWithoutValidatingDurableViews(opCtx, collectionName);
+            auto view = CollectionCatalog::get(opCtx)->lookupViewWithoutValidatingDurable(
+                opCtx, collectionName);
             if (!view) {
                 // Timeseries bucket collection may exist even without the view. If that is the case
                 // delete it.
diff --git a/src/mongo/db/catalog/drop_indexes.cpp b/src/mongo/db/catalog/drop_indexes.cpp
index 34c5b6c6285..596615bbead 100644
--- a/src/mongo/db/catalog/drop_indexes.cpp
+++ b/src/mongo/db/catalog/drop_indexes.cpp
@@ -35,6 +35,7 @@
 
 #include <boost/algorithm/string/join.hpp>
 
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/collection_uuid_mismatch.h"
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/client.h"
@@ -49,7 +50,6 @@
 #include "mongo/db/s/collection_sharding_state.h"
 #include "mongo/db/s/database_sharding_state.h"
 #include "mongo/db/service_context.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/visit_helper.h"
 
@@ -65,7 +65,7 @@ Status checkView(OperationContext* opCtx,
                  const NamespaceString& nss,
                  const CollectionPtr& collection) {
     if (!collection) {
-        if (ViewCatalog::get(opCtx)->lookup(opCtx, nss)) {
+        if (CollectionCatalog::get(opCtx)->lookupView(opCtx, nss)) {
             return Status(ErrorCodes::CommandNotSupportedOnView,
                           str::stream() << "Cannot drop indexes on view " << nss);
         }
diff --git a/src/mongo/db/catalog/rename_collection.cpp b/src/mongo/db/catalog/rename_collection.cpp
index 279c3d4e53a..7ae108ff44a 100644
--- a/src/mongo/db/catalog/rename_collection.cpp
+++ b/src/mongo/db/catalog/rename_collection.cpp
@@ -59,7 +59,6 @@
 #include "mongo/db/s/operation_sharding_state.h"
 #include "mongo/db/server_options.h"
 #include "mongo/db/service_context.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/fail_point.h"
 #include "mongo/util/scopeguard.h"
@@ -111,7 +110,7 @@ Status checkSourceAndTargetNamespaces(OperationContext* opCtx,
     auto catalog = CollectionCatalog::get(opCtx);
     const auto sourceColl = catalog->lookupCollectionByNamespace(opCtx, source);
     if (!sourceColl) {
-        if (ViewCatalog::get(opCtx)->lookup(opCtx, source))
+        if (CollectionCatalog::get(opCtx)->lookupView(opCtx, source))
             return Status(ErrorCodes::CommandNotSupportedOnView,
                           str::stream() << "cannot rename view: " << source);
         return Status(ErrorCodes::NamespaceNotFound,
@@ -123,7 +122,7 @@ Status checkSourceAndTargetNamespaces(OperationContext* opCtx,
     const auto targetColl = catalog->lookupCollectionByNamespace(opCtx, target);
 
     if (!targetColl) {
-        if (ViewCatalog::get(opCtx)->lookup(opCtx, target))
+        if (CollectionCatalog::get(opCtx)->lookupView(opCtx, target))
             return Status(ErrorCodes::NamespaceExists,
                           str::stream() << "a view already exists with that name: " << target);
     } else {
@@ -511,7 +510,7 @@ Status renameBetweenDBs(OperationContext* opCtx,
     auto catalog = CollectionCatalog::get(opCtx);
     const auto sourceColl = catalog->lookupCollectionByNamespace(opCtx, source);
     if (!sourceColl) {
-        if (ViewCatalog::get(opCtx)->lookup(opCtx, source))
+        if (CollectionCatalog::get(opCtx)->lookupView(opCtx, source))
             return Status(ErrorCodes::CommandNotSupportedOnView,
                           str::stream() << "cannot rename view: " << source);
         return Status(ErrorCodes::NamespaceNotFound, "source namespace does not exist");
@@ -541,7 +540,7 @@ Status renameBetweenDBs(OperationContext* opCtx,
             return Status(ErrorCodes::NamespaceExists, "target namespace exists");
         }
 
-    } else if (ViewCatalog::get(opCtx)->lookup(opCtx, target)) {
+    } else if (CollectionCatalog::get(opCtx)->lookupView(opCtx, target)) {
         return Status(ErrorCodes::NamespaceExists,
                       str::stream() << "a view already exists with that name: " << target);
     }
diff --git a/src/mongo/db/catalog/validate_state.cpp b/src/mongo/db/catalog/validate_state.cpp
index 120325051ef..d5d5a322503 100644
--- a/src/mongo/db/catalog/validate_state.cpp
+++ b/src/mongo/db/catalog/validate_state.cpp
@@ -34,13 +34,13 @@
 #include "mongo/db/catalog/validate_state.h"
 
 #include "mongo/db/catalog/collection.h"
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/index_consistency.h"
 #include "mongo/db/catalog/validate_adaptor.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/index/index_access_method.h"
 #include "mongo/db/operation_context.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/fail_point.h"
 
@@ -80,7 +80,7 @@ ValidateState::ValidateState(OperationContext* opCtx,
         _collection = CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, _nss);
 
     if (!_collection) {
-        if (ViewCatalog::get(opCtx)->lookup(opCtx, _nss)) {
+        if (CollectionCatalog::get(opCtx)->lookupView(opCtx, _nss)) {
             uasserted(ErrorCodes::CommandNotSupportedOnView, "Cannot validate a view");
         }
 
diff --git a/src/mongo/db/catalog/views_for_database.cpp b/src/mongo/db/catalog/views_for_database.cpp
new file mode 100644
index 00000000000..776cf5e3266
--- /dev/null
+++ b/src/mongo/db/catalog/views_for_database.cpp
@@ -0,0 +1,196 @@
+/**
+ *    Copyright (C) 2018-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kStorage
+
+#include "views_for_database.h"
+
+#include "mongo/logv2/log.h"
+
+namespace mongo {
+
+StatusWith<std::unique_ptr<CollatorInterface>> ViewsForDatabase::parseCollator(
+    OperationContext* opCtx, BSONObj collationSpec) {
+    // If 'collationSpec' is empty, return the null collator, which represents the "simple"
+    // collation.
+    if (collationSpec.isEmpty()) {
+        return {nullptr};
+    }
+    return CollatorFactoryInterface::get(opCtx->getServiceContext())->makeFromBSON(collationSpec);
+}
+
+void 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);
+}
+
+std::shared_ptr<const ViewDefinition> ViewsForDatabase::lookup(const NamespaceString& ns) const {
+    ViewMap::const_iterator it = viewMap.find(ns.ns());
+    if (it != viewMap.end()) {
+        return it->second;
+    }
+    return nullptr;
+}
+
+Status ViewsForDatabase::reload(OperationContext* opCtx) {
+    auto reloadCallback = [&](const BSONObj& view) -> Status {
+        BSONObj collationSpec = view.hasField("collation") ? view["collation"].Obj() : BSONObj();
+        auto collator = parseCollator(opCtx, collationSpec);
+        if (!collator.isOK()) {
+            return collator.getStatus();
+        }
+
+        NamespaceString viewName(view["_id"].str());
+
+        auto pipeline = view["pipeline"].Obj();
+        for (auto&& stage : pipeline) {
+            if (BSONType::Object != stage.type()) {
+                return Status(ErrorCodes::InvalidViewDefinition,
+                              str::stream() << "View 'pipeline' entries must be objects, but "
+                                            << viewName.toString()
+                                            << " has a pipeline element of type " << stage.type());
+            }
+        }
+
+        auto viewDef = std::make_shared<ViewDefinition>(viewName.db(),
+                                                        viewName.coll(),
+                                                        view["viewOn"].str(),
+                                                        pipeline,
+                                                        std::move(collator.getValue()));
+
+        if (!viewName.isOnInternalDb() && !viewName.isSystem()) {
+            if (viewDef->timeseries()) {
+                stats.userTimeseries += 1;
+            } else {
+                stats.userViews += 1;
+            }
+        } else {
+            stats.internal += 1;
+        }
+
+        viewMap[viewName.ns()] = std::move(viewDef);
+        return Status::OK();
+    };
+
+    try {
+        durable->iterate(opCtx, reloadCallback);
+    } catch (const DBException& ex) {
+        auto status = ex.toStatus();
+        LOGV2(22547,
+              "Could not load view catalog for database",
+              "db"_attr = durable->getName(),
+              "error"_attr = status);
+        return status;
+    }
+
+    valid = true;
+
+    return Status::OK();
+}
+
+Status ViewsForDatabase::validateCollation(OperationContext* opCtx,
+                                           const ViewDefinition& view,
+                                           const std::vector<NamespaceString>& refs) const {
+    for (auto&& potentialViewNss : refs) {
+        auto otherView = lookup(potentialViewNss);
+        if (otherView &&
+            !CollatorInterface::collatorsMatch(view.defaultCollator(),
+                                               otherView->defaultCollator())) {
+            return {ErrorCodes::OptionNotSupportedOnView,
+                    str::stream() << "View " << view.name().toString()
+                                  << " has conflicting collation with view "
+                                  << otherView->name().toString()};
+        }
+    }
+    return Status::OK();
+}
+
+Status ViewsForDatabase::upsertIntoGraph(OperationContext* opCtx,
+                                         const ViewDefinition& viewDef,
+                                         const PipelineValidatorFn& validatePipeline) {
+    // Performs the insert into the graph.
+    auto doInsert = [this, opCtx, &validatePipeline](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);
+        if (!pipelineStatus.isOK()) {
+            if (needsValidation) {
+                uassertStatusOKWithContext(pipelineStatus.getStatus(),
+                                           str::stream() << "Invalid pipeline for view "
+                                                         << viewDef.name().ns());
+            }
+            return pipelineStatus.getStatus();
+        }
+
+        auto involvedNamespaces = pipelineStatus.getValue();
+        std::vector<NamespaceString> refs(involvedNamespaces.begin(), involvedNamespaces.end());
+        refs.push_back(viewDef.viewOn());
+
+        int pipelineSize = 0;
+        for (auto obj : viewDef.pipeline()) {
+            pipelineSize += obj.objsize();
+        }
+
+        if (needsValidation) {
+            // Check the collation of all the dependent namespaces before updating the graph.
+            auto collationStatus = validateCollation(opCtx, viewDef, refs);
+            if (!collationStatus.isOK()) {
+                return collationStatus;
+            }
+            return viewGraph.insertAndValidate(viewDef, refs, pipelineSize);
+        } else {
+            viewGraph.insertWithoutValidating(viewDef, refs, pipelineSize);
+            return Status::OK();
+        }
+    };
+
+    if (viewGraphNeedsRefresh) {
+        viewGraph.clear();
+        for (auto&& iter : viewMap) {
+            auto status = doInsert(*(iter.second.get()), false);
+            // If we cannot fully refresh the graph, we will keep '_viewGraphNeedsRefresh' true.
+            if (!status.isOK()) {
+                return status;
+            }
+        }
+        // Only if the inserts completed without error will we no longer need a refresh.
+        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());
+
+    return doInsert(viewDef, true);
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/catalog/views_for_database.h b/src/mongo/db/catalog/views_for_database.h
new file mode 100644
index 00000000000..914adf60df7
--- /dev/null
+++ b/src/mongo/db/catalog/views_for_database.h
@@ -0,0 +1,113 @@
+/**
+ *    Copyright (C) 2018-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include <functional>
+
+#include "mongo/db/operation_context.h"
+#include "mongo/db/query/collation/collator_factory_interface.h"
+#include "mongo/db/views/durable_view_catalog.h"
+#include "mongo/db/views/view.h"
+#include "mongo/db/views/view_graph.h"
+#include "mongo/stdx/unordered_map.h"
+#include "mongo/util/string_map.h"
+
+namespace mongo {
+
+/**
+ * Holds all data for the views associated with a particular database.
+ */
+class ViewsForDatabase {
+public:
+    using ViewMap = StringMap<std::shared_ptr<ViewDefinition>>;
+    using PipelineValidatorFn = std::function<StatusWith<stdx::unordered_set<NamespaceString>>(
+        OperationContext*, const ViewDefinition&)>;
+
+    /**
+     * Usage statistics about the views associated with a single database.
+     * Total views = internal + userViews + userTimeseries.
+     */
+    struct Stats {
+        int userViews = 0;
+        int userTimeseries = 0;
+        int internal = 0;
+    };
+
+    /**
+     * Helper method to build a collator from its spec.
+     */
+    static StatusWith<std::unique_ptr<CollatorInterface>> parseCollator(OperationContext* opCtx,
+                                                                        BSONObj collationSpec);
+
+    std::shared_ptr<DurableViewCatalog> durable;
+    ViewMap viewMap;
+    bool valid = false;
+    ViewGraph viewGraph;
+    bool viewGraphNeedsRefresh = true;
+    Stats stats;
+    bool ignoreExternalChange = false;
+
+    /**
+     * uasserts with the InvalidViewDefinition error if the current in-memory state of the views for
+     * this database is invalid which can happen as a result of direct writes to the 'system.views'
+     * collection or data corruption. This prevents further use of views on this database until the
+     * issue is resolved.
+     */
+    void requireValidCatalog() const;
+
+    /**
+     * Returns the 'ViewDefiniton' assocated with namespace 'ns' if one exists, nullptr otherwise.
+     */
+    std::shared_ptr<const ViewDefinition> lookup(const NamespaceString& ns) const;
+
+    /**
+     * Reloads the views for this database by iterating the DurableViewCatalog.
+     */
+    Status reload(OperationContext* opCtx);
+
+    /**
+     * Returns Status::OK if each view namespace in 'refs' has the same default collation as
+     * 'view'. Otherwise, returns ErrorCodes::OptionNotSupportedOnView.
+     */
+    Status validateCollation(OperationContext* opCtx,
+                             const ViewDefinition& view,
+                             const std::vector<NamespaceString>& refs) const;
+
+    /**
+     * Parses the view definition pipeline, attempts to upsert into the view graph, and
+     * refreshes the graph if necessary. Returns an error status if the resulting graph
+     * would be invalid.
+     */
+    Status upsertIntoGraph(OperationContext* opCtx,
+                           const ViewDefinition& viewDef,
+                           const PipelineValidatorFn&);
+};
+
+}  // namespace mongo