SERVER-57250 Merge ViewCatalog into CollectionCatalog

52 files changed, 1632 insertions, 1658 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
diff --git a/src/mongo/db/catalog_raii.cpp b/src/mongo/db/catalog_raii.cpp
index 13635b0a16c..20d9f14ba47 100644
--- a/src/mongo/db/catalog_raii.cpp
+++ b/src/mongo/db/catalog_raii.cpp
@@ -37,7 +37,6 @@
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/s/collection_sharding_state.h"
 #include "mongo/db/s/database_sharding_state.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/fail_point.h"
 
@@ -296,17 +295,14 @@ AutoGetCollection::AutoGetCollection(
         return;
     }
 
-    if (_autoDb->getDb()) {
-        _view = ViewCatalog::get(opCtx)->lookup(opCtx, _resolvedNss);
-        uassert(ErrorCodes::CommandNotSupportedOnView,
-                str::stream() << "Namespace " << _resolvedNss.ns() << " is a timeseries collection",
-                !_view || viewMode == AutoGetCollectionViewMode::kViewsPermitted ||
-                    !_view->timeseries());
-        uassert(ErrorCodes::CommandNotSupportedOnView,
-                str::stream() << "Namespace " << _resolvedNss.ns()
-                              << " is a view, not a collection",
-                !_view || viewMode == AutoGetCollectionViewMode::kViewsPermitted);
-    }
+    _view = catalog->lookupView(opCtx, _resolvedNss);
+    uassert(ErrorCodes::CommandNotSupportedOnView,
+            str::stream() << "Namespace " << _resolvedNss.ns() << " is a timeseries collection",
+            !_view || viewMode == AutoGetCollectionViewMode::kViewsPermitted ||
+                !_view->timeseries());
+    uassert(ErrorCodes::CommandNotSupportedOnView,
+            str::stream() << "Namespace " << _resolvedNss.ns() << " is a view, not a collection",
+            !_view || viewMode == AutoGetCollectionViewMode::kViewsPermitted);
 }
 
 Collection* AutoGetCollection::getWritableCollection(OperationContext* opCtx,
@@ -392,14 +388,7 @@ AutoGetCollectionLockFree::AutoGetCollectionLockFree(OperationContext* opCtx,
         return;
     }
 
-    // Returns nullptr for 'viewCatalog' if db does not exist.
-    const TenantDatabaseName tenantDbName(boost::none, _resolvedNss.db());
-    auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, tenantDbName);
-    if (!viewCatalog) {
-        return;
-    }
-
-    _view = viewCatalog->lookup(opCtx, _resolvedNss);
+    _view = catalog->lookupView(opCtx, _resolvedNss);
     uassert(ErrorCodes::CommandNotSupportedOnView,
             str::stream() << "Namespace " << _resolvedNss.ns() << " is a timeseries collection",
             !_view || viewMode == AutoGetCollectionViewMode::kViewsPermitted ||
diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript
index 6e17b42669a..98c4b4c3a9e 100644
--- a/src/mongo/db/commands/SConscript
+++ b/src/mongo/db/commands/SConscript
@@ -233,7 +233,9 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/crypto/encrypted_field_config',
+        '$BUILD_DIR/mongo/db/auth/auth',
         '$BUILD_DIR/mongo/db/catalog/catalog_helpers',
+        '$BUILD_DIR/mongo/db/catalog/collection_catalog',
         '$BUILD_DIR/mongo/db/catalog_raii',
         '$BUILD_DIR/mongo/db/commands',
     ]
@@ -407,6 +409,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/timeseries/timeseries_conversion_util',
         '$BUILD_DIR/mongo/db/timeseries/timeseries_stats',
         '$BUILD_DIR/mongo/db/transaction',
+        '$BUILD_DIR/mongo/db/views/view_catalog_helpers',
         '$BUILD_DIR/mongo/db/views/views_mongod',
         '$BUILD_DIR/mongo/executor/async_request_executor',
         '$BUILD_DIR/mongo/idl/feature_flag',
diff --git a/src/mongo/db/commands/compact.cpp b/src/mongo/db/commands/compact.cpp
index ca5e15affa9..6aa009ea216 100644
--- a/src/mongo/db/commands/compact.cpp
+++ b/src/mongo/db/commands/compact.cpp
@@ -41,7 +41,6 @@
 #include "mongo/db/curop.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/repl/replication_coordinator.h"
-#include "mongo/db/views/view_catalog.h"
 
 namespace mongo {
 
diff --git a/src/mongo/db/commands/create_indexes.cpp b/src/mongo/db/commands/create_indexes.cpp
index 92c3efb2bbf..66d463e220f 100644
--- a/src/mongo/db/commands/create_indexes.cpp
+++ b/src/mongo/db/commands/create_indexes.cpp
@@ -39,6 +39,7 @@
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/catalog/clustered_collection_util.h"
 #include "mongo/db/catalog/collection.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/database.h"
@@ -66,7 +67,6 @@
 #include "mongo/db/timeseries/catalog_helper.h"
 #include "mongo/db/timeseries/timeseries_commands_conversion_helper.h"
 #include "mongo/db/timeseries/timeseries_index_schema_conversion_functions.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/idl/command_generic_argument.h"
 #include "mongo/logv2/log.h"
 #include "mongo/platform/compiler.h"
@@ -379,12 +379,9 @@ CreateIndexesReply runCreateIndexesOnNewCollection(
     bool createCollImplicitly) {
     WriteUnitOfWork wunit(opCtx);
 
-    const TenantDatabaseName tenantDbName(boost::none, ns.db());
-    auto databaseHolder = DatabaseHolder::get(opCtx);
-    auto db = databaseHolder->getDb(opCtx, tenantDbName);
     uassert(ErrorCodes::CommandNotSupportedOnView,
             "Cannot create indexes on a view",
-            !db || !ViewCatalog::get(opCtx)->lookup(opCtx, ns));
+            !CollectionCatalog::get(opCtx)->lookupView(opCtx, ns));
 
     if (createCollImplicitly) {
         for (const auto& spec : specs) {
diff --git a/src/mongo/db/commands/drop_indexes.cpp b/src/mongo/db/commands/drop_indexes.cpp
index 9dd71eeefb0..3f60b169975 100644
--- a/src/mongo/db/commands/drop_indexes.cpp
+++ b/src/mongo/db/commands/drop_indexes.cpp
@@ -36,6 +36,7 @@
 
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/catalog/collection.h"
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/catalog/drop_indexes.h"
 #include "mongo/db/catalog/index_catalog.h"
@@ -54,7 +55,6 @@
 #include "mongo/db/timeseries/catalog_helper.h"
 #include "mongo/db/timeseries/timeseries_commands_conversion_helper.h"
 #include "mongo/db/vector_clock.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/exit_code.h"
 #include "mongo/util/quick_exit.h"
@@ -161,7 +161,7 @@ public:
 
         AutoGetCollection autoColl(opCtx, toReIndexNss, MODE_X);
         if (!autoColl) {
-            if (ViewCatalog::get(opCtx)->lookup(opCtx, toReIndexNss))
+            if (CollectionCatalog::get(opCtx)->lookupView(opCtx, toReIndexNss))
                 uasserted(ErrorCodes::CommandNotSupportedOnView, "can't re-index a view");
             else
                 uasserted(ErrorCodes::NamespaceNotFound, "collection does not exist");
diff --git a/src/mongo/db/commands/fle2_compact.cpp b/src/mongo/db/commands/fle2_compact.cpp
index 017b5840407..7588cbc6735 100644
--- a/src/mongo/db/commands/fle2_compact.cpp
+++ b/src/mongo/db/commands/fle2_compact.cpp
@@ -35,10 +35,10 @@
 
 #include "mongo/crypto/encryption_fields_gen.h"
 #include "mongo/db/auth/authorization_session.h"
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/rename_collection.h"
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/commands.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 
 namespace mongo {
@@ -149,7 +149,7 @@ StatusWith<CompactStats> compactEncryptedCompactionCollection(
     // Check the data collection exists and is not a view
     auto edc = catalog->lookupCollectionByNamespace(opCtx, edcNss);
     if (!edc) {
-        if (ViewCatalog::get(opCtx)->lookup(opCtx, edcNss)) {
+        if (catalog->lookupView(opCtx, edcNss)) {
             return Status(ErrorCodes::CommandNotSupportedOnView,
                           "cannot compact structured encryption data on a view");
         }
diff --git a/src/mongo/db/commands/list_collections.cpp b/src/mongo/db/commands/list_collections.cpp
index 741e19ba454..3c5fa674aa9 100644
--- a/src/mongo/db/commands/list_collections.cpp
+++ b/src/mongo/db/commands/list_collections.cpp
@@ -41,6 +41,7 @@
 #include "mongo/bson/util/bson_extract.h"
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/catalog/collection.h"
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/collection_catalog_helper.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/catalog/database_holder.h"
@@ -65,7 +66,6 @@
 #include "mongo/db/storage/storage_engine.h"
 #include "mongo/db/storage/storage_options.h"
 #include "mongo/db/timeseries/timeseries_constants.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 
 namespace mongo {
@@ -332,7 +332,7 @@ public:
                 // Acquire only the global lock and set up a consistent in-memory catalog and
                 // storage snapshot.
                 AutoGetDbForReadMaybeLockFree lockFreeReadBlock(opCtx, dbName);
-                auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, tenantDbName);
+                auto catalog = CollectionCatalog::get(opCtx);
 
                 CurOpFailpointHelpers::waitWhileFailPointEnabled(&hangBeforeListCollections,
                                                                  opCtx,
@@ -343,8 +343,8 @@ public:
                 auto ws = std::make_unique<WorkingSet>();
                 auto root = std::make_unique<QueuedDataStage>(expCtx.get(), ws.get());
 
-                // If the ViewCatalog pointer is valid, then the database exists.
-                if (viewCatalog) {
+                if (DatabaseHolder::get(opCtx)->dbExists(opCtx,
+                                                         TenantDatabaseName(boost::none, dbName))) {
                     if (auto collNames = _getExactNameMatches(matcher.get())) {
                         for (auto&& collName : *collNames) {
                             auto nss = NamespaceString(dbName, collName);
@@ -372,8 +372,7 @@ public:
                                         opCtx, collection, includePendingDrops, nameOnly);
                                 }
 
-                                auto view =
-                                    viewCatalog->lookupWithoutValidatingDurableViews(opCtx, nss);
+                                auto view = catalog->lookupViewWithoutValidatingDurable(opCtx, nss);
                                 if (view && view->timeseries()) {
                                     if (auto bucketsCollection = CollectionCatalog::get(opCtx)
                                                                      ->lookupCollectionByNamespace(
@@ -399,7 +398,7 @@ public:
                         auto perCollectionWork = [&](const CollectionPtr& collection) {
                             if (collection && collection->getTimeseriesOptions() &&
                                 !collection->ns().isDropPendingNamespace() &&
-                                viewCatalog->lookupWithoutValidatingDurableViews(
+                                catalog->lookupViewWithoutValidatingDurable(
                                     opCtx, collection->ns().getTimeseriesViewNamespace()) &&
                                 (!authorizedCollections ||
                                  as->isAuthorizedForAnyActionOnResource(
@@ -454,7 +453,7 @@ public:
                             ListCollectionsFilter::makeTypeCollectionFilter());
 
                     if (!skipViews) {
-                        viewCatalog->iterate(dbName, [&](const ViewDefinition& view) {
+                        catalog->iterateViews(opCtx, dbName, [&](const ViewDefinition& view) {
                             if (authorizedCollections &&
                                 !as->isAuthorizedForAnyActionOnResource(
                                     ResourcePattern::forExactNamespace(view.name()))) {
diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp
index eda074f1924..fb650d38acf 100644
--- a/src/mongo/db/commands/run_aggregate.cpp
+++ b/src/mongo/db/commands/run_aggregate.cpp
@@ -81,7 +81,7 @@
 #include "mongo/db/stats/resource_consumption_metrics.h"
 #include "mongo/db/storage/storage_options.h"
 #include "mongo/db/views/view.h"
-#include "mongo/db/views/view_catalog.h"
+#include "mongo/db/views/view_catalog_helpers.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/scopeguard.h"
 #include "mongo/util/string_map.h"
@@ -291,13 +291,12 @@ StatusWith<StringMap<ExpressionContext::ResolvedNamespace>> resolveInvolvedNames
         return {StringMap<ExpressionContext::ResolvedNamespace>()};
     }
 
-    // Acquire a single const view of the database's ViewCatalog (if it exists) and use it for all
-    // view definition resolutions that follow. This prevents the view definitions cached in
-    // 'resolvedNamespaces' from changing relative to those in the acquired ViewCatalog. The
-    // resolution of the view definitions below might lead into an endless cycle if any are allowed
-    // to change.
-    const TenantDatabaseName tenantDbName(boost::none, request.getNamespace().db());
-    auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, tenantDbName);
+    // Acquire a single const view of the CollectionCatalog and use it for all view and collection
+    // lookups and view definition resolutions that follow. This prevents the view definitions
+    // cached in 'resolvedNamespaces' from changing relative to those in the acquired ViewCatalog.
+    // The resolution of the view definitions below might lead into an endless cycle if any are
+    // allowed to change.
+    auto catalog = CollectionCatalog::get(opCtx);
 
     std::deque<NamespaceString> involvedNamespacesQueue(pipelineInvolvedNamespaces.begin(),
                                                         pipelineInvolvedNamespaces.end());
@@ -312,9 +311,8 @@ StatusWith<StringMap<ExpressionContext::ResolvedNamespace>> resolveInvolvedNames
         }
 
         // If 'ns' refers to a view namespace, then we resolve its definition.
-        auto resolveViewDefinition = [&](const NamespaceString& ns,
-                                         std::shared_ptr<const ViewCatalog> vcp) -> Status {
-            auto resolvedView = vcp->resolveView(opCtx, ns, boost::none);
+        auto resolveViewDefinition = [&](const NamespaceString& ns) -> Status {
+            auto resolvedView = view_catalog_helpers::resolveView(opCtx, catalog, ns, boost::none);
             if (!resolvedView.isOK()) {
                 return resolvedView.getStatus().withContext(
                     str::stream() << "Failed to resolve view '" << involvedNs.ns());
@@ -322,7 +320,7 @@ StatusWith<StringMap<ExpressionContext::ResolvedNamespace>> resolveInvolvedNames
 
             auto&& underlyingNs = resolvedView.getValue().getNamespace();
             // Attempt to acquire UUID of the underlying collection using lock free method.
-            auto uuid = CollectionCatalog::get(opCtx)->lookupUUIDByNSS(opCtx, underlyingNs);
+            auto uuid = catalog->lookupUUIDByNSS(opCtx, underlyingNs);
             resolvedNamespaces[ns.coll()] = {
                 underlyingNs, resolvedView.getValue().getPipeline(), uuid};
 
@@ -348,14 +346,7 @@ StatusWith<StringMap<ExpressionContext::ResolvedNamespace>> resolveInvolvedNames
                 // require a lookup stage involving a view on the 'local' database.
                 // If the involved namespace is 'local.system.tenantMigration.oplogView', resolve
                 // its view definition.
-                const TenantDatabaseName involvedTenantDbName(boost::none, involvedNs.db());
-                auto involvedDbViewCatalog =
-                    DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, involvedTenantDbName);
-
-                // It is safe to assume that the ViewCatalog for the `local` database always
-                // exists because replica sets forbid dropping the oplog and the `local` database.
-                invariant(involvedDbViewCatalog);
-                auto status = resolveViewDefinition(involvedNs, involvedDbViewCatalog);
+                auto status = resolveViewDefinition(involvedNs);
                 if (!status.isOK()) {
                     return status;
                 }
@@ -367,8 +358,8 @@ StatusWith<StringMap<ExpressionContext::ResolvedNamespace>> resolveInvolvedNames
                 // that the inverse scenario (mistaking a view for a collection) is not an issue
                 // because $merge/$out cannot target a view.
                 auto nssToCheck = NamespaceString(request.getNamespace().db(), involvedNs.coll());
-                if (viewCatalog && viewCatalog->lookup(opCtx, nssToCheck)) {
-                    auto status = resolveViewDefinition(nssToCheck, viewCatalog);
+                if (catalog->lookupView(opCtx, nssToCheck)) {
+                    auto status = resolveViewDefinition(nssToCheck);
                     if (!status.isOK()) {
                         return status;
                     }
@@ -376,18 +367,14 @@ StatusWith<StringMap<ExpressionContext::ResolvedNamespace>> resolveInvolvedNames
                     resolvedNamespaces[involvedNs.coll()] = {involvedNs, std::vector<BSONObj>{}};
                 }
             }
-        } else if (!viewCatalog ||
-                   CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, involvedNs)) {
+        } else if (catalog->lookupCollectionByNamespace(opCtx, involvedNs)) {
             // Attempt to acquire UUID of the collection using lock free method.
-            auto uuid = CollectionCatalog::get(opCtx)->lookupUUIDByNSS(opCtx, involvedNs);
-            // If the aggregation database exists and 'involvedNs' refers to a collection namespace,
-            // then we resolve it as an empty pipeline in order to read directly from the underlying
-            // collection. If the database doesn't exist, then we still resolve it as an empty
-            // pipeline because 'involvedNs' doesn't refer to a view namespace in our consistent
-            // snapshot of the view catalog.
+            auto uuid = catalog->lookupUUIDByNSS(opCtx, involvedNs);
+            // If 'involvedNs' refers to a collection namespace, then we resolve it as an empty
+            // pipeline in order to read directly from the underlying collection.
             resolvedNamespaces[involvedNs.coll()] = {involvedNs, std::vector<BSONObj>{}, uuid};
-        } else if (viewCatalog->lookup(opCtx, involvedNs)) {
-            auto status = resolveViewDefinition(involvedNs, viewCatalog);
+        } else if (catalog->lookupView(opCtx, involvedNs)) {
+            auto status = resolveViewDefinition(involvedNs);
             if (!status.isOK()) {
                 return status;
             }
@@ -409,18 +396,13 @@ Status collatorCompatibleWithPipeline(OperationContext* opCtx,
                                       StringData dbName,
                                       const CollatorInterface* collator,
                                       const LiteParsedPipeline& liteParsedPipeline) {
-    const TenantDatabaseName tenantDbName(boost::none, dbName);
-    auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, tenantDbName);
-    if (!viewCatalog) {
-        return Status::OK();
-    }
     auto catalog = CollectionCatalog::get(opCtx);
     for (auto&& potentialViewNs : liteParsedPipeline.getInvolvedNamespaces()) {
         if (catalog->lookupCollectionByNamespace(opCtx, potentialViewNs)) {
             continue;
         }
 
-        auto view = viewCatalog->lookup(opCtx, potentialViewNs);
+        auto view = catalog->lookupView(opCtx, potentialViewNs);
         if (!view) {
             continue;
         }
@@ -686,6 +668,7 @@ Status runAggregate(OperationContext* opCtx,
     std::vector<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> execs;
     boost::intrusive_ptr<ExpressionContext> expCtx;
     auto curOp = CurOp::get(opCtx);
+    auto catalog = CollectionCatalog::get(opCtx);
 
     {
         // If we are in a transaction, check whether the parsed pipeline supports being in
@@ -718,19 +701,15 @@ Status runAggregate(OperationContext* opCtx,
             // a stream on an entire db or across the cluster.
             const TenantDatabaseName origTenantDbName(boost::none, origNss.db());
             if (!origNss.isCollectionlessAggregateNS()) {
-                auto viewCatalog =
-                    DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, origTenantDbName);
-                if (viewCatalog) {
-                    auto view = viewCatalog->lookup(opCtx, origNss);
-                    uassert(ErrorCodes::CommandNotSupportedOnView,
-                            str::stream()
-                                << "Namespace " << origNss.ns() << " is a timeseries collection",
-                            !view || !view->timeseries());
-                    uassert(ErrorCodes::CommandNotSupportedOnView,
-                            str::stream()
-                                << "Namespace " << origNss.ns() << " is a view, not a collection",
-                            !view);
-                }
+                auto view = catalog->lookupView(opCtx, origNss);
+                uassert(ErrorCodes::CommandNotSupportedOnView,
+                        str::stream()
+                            << "Namespace " << origNss.ns() << " is a timeseries collection",
+                        !view || !view->timeseries());
+                uassert(ErrorCodes::CommandNotSupportedOnView,
+                        str::stream()
+                            << "Namespace " << origNss.ns() << " is a view, not a collection",
+                        !view);
             }
 
             // If the user specified an explicit collation, adopt it; otherwise, use the simple
@@ -816,16 +795,15 @@ Status runAggregate(OperationContext* opCtx,
             auto timeSeriesCollator =
                 ctx->getView()->timeseries() ? request.getCollation() : boost::none;
 
-            // Check that the database/view catalog still exist, in case this is a lock-free
+            // Check that the database still exists, in case this is a lock-free
             // operation. It's possible for a view to disappear after we release locks below, so
             // it's safe to quit early if the view disappears while running lock-free.
             const TenantDatabaseName tenantDbName(boost::none, nss.db());
-            auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, tenantDbName);
             uassert(ErrorCodes::NamespaceNotFound,
                     str::stream() << "Namespace '" << nss << "' no longer exists",
-                    viewCatalog);
-            auto resolvedView =
-                uassertStatusOK(viewCatalog->resolveView(opCtx, nss, timeSeriesCollator));
+                    DatabaseHolder::get(opCtx)->dbExists(opCtx, tenantDbName));
+            auto resolvedView = uassertStatusOK(
+                view_catalog_helpers::resolveView(opCtx, catalog, nss, timeSeriesCollator));
 
             // With the view & collation resolved, we can relinquish locks.
             resetContext();
diff --git a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
index 119b9149557..81553af9c89 100644
--- a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
+++ b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
@@ -78,7 +78,6 @@
 #include "mongo/db/session_txn_record_gen.h"
 #include "mongo/db/timeseries/timeseries_index_schema_conversion_functions.h"
 #include "mongo/db/vector_clock.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 #include "mongo/rpc/get_status_from_command_result.h"
 #include "mongo/s/pm2423_feature_flags_gen.h"
diff --git a/src/mongo/db/commands/validate_db_metadata_cmd.cpp b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
index 6af54a47d11..c3a00ff74e6 100644
--- a/src/mongo/db/commands/validate_db_metadata_cmd.cpp
+++ b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
@@ -32,6 +32,7 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/catalog/collection.h"
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/collection_catalog_helper.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/index_catalog.h"
@@ -41,8 +42,9 @@
 #include "mongo/db/commands/validate_db_metadata_gen.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/multitenancy.h"
-#include "mongo/db/views/view_catalog.h"
+#include "mongo/db/views/view_catalog_helpers.h"
 #include "mongo/logv2/log.h"
+
 namespace mongo {
 namespace {
 void overrideAPIParams(OperationContext* opCtx, const APIParamsForCmd& params) {
@@ -142,13 +144,10 @@ public:
 
                 // If there is no collection name present in the input, run validation against all
                 // the collections.
-                if (auto viewCatalog =
-                        DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, tenantDbName)) {
-                    viewCatalog->iterate(tenantDbName.dbName(),
-                                         [this, opCtx](const ViewDefinition& view) {
-                                             return _validateView(opCtx, view);
-                                         });
-                }
+                collectionCatalog->iterateViews(
+                    opCtx, tenantDbName.dbName(), [this, opCtx](const ViewDefinition& view) {
+                        return _validateView(opCtx, view);
+                    });
 
                 for (auto collIt = collectionCatalog->begin(opCtx, tenantDbName);
                      collIt != collectionCatalog->end(opCtx);
@@ -166,7 +165,7 @@ public:
          * Returns false, if the evaluation needs to be aborted.
          */
         bool _validateView(OperationContext* opCtx, const ViewDefinition& view) {
-            auto pipelineStatus = ViewCatalog::validatePipeline(opCtx, view);
+            auto pipelineStatus = view_catalog_helpers::validatePipeline(opCtx, view);
             if (!pipelineStatus.isOK()) {
                 ErrorReplyElement error(view.name().ns(),
                                         ErrorCodes::APIStrictError,
diff --git a/src/mongo/db/commands/write_commands.cpp b/src/mongo/db/commands/write_commands.cpp
index 9b63f16641c..e7584569d7c 100644
--- a/src/mongo/db/commands/write_commands.cpp
+++ b/src/mongo/db/commands/write_commands.cpp
@@ -33,6 +33,7 @@
 #include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/bson/mutable/document.h"
 #include "mongo/bson/mutable/element.h"
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/collection_operation_source.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/document_validation.h"
@@ -71,7 +72,6 @@
 #include "mongo/db/timeseries/timeseries_constants.h"
 #include "mongo/db/timeseries/timeseries_stats.h"
 #include "mongo/db/transaction_participant.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/db/write_concern.h"
 #include "mongo/logv2/log.h"
 #include "mongo/logv2/redaction.h"
diff --git a/src/mongo/db/db_raii.cpp b/src/mongo/db/db_raii.cpp
index b7d4c5b199f..ab1f8220acf 100644
--- a/src/mongo/db/db_raii.cpp
+++ b/src/mongo/db/db_raii.cpp
@@ -33,6 +33,7 @@
 
 #include "mongo/db/db_raii.h"
 
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/concurrency/locker.h"
 #include "mongo/db/curop.h"
@@ -40,7 +41,6 @@
 #include "mongo/db/s/collection_sharding_state.h"
 #include "mongo/db/s/database_sharding_state.h"
 #include "mongo/db/storage/snapshot_helper.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 
 namespace mongo {
@@ -125,7 +125,7 @@ Status checkSecondaryCollection(OperationContext* opCtx,
  * Returns true if 'nss' is a view. False if the view doesn't exist.
  */
 bool isSecondaryNssAView(OperationContext* opCtx, const NamespaceString& nss) {
-    return ViewCatalog::get(opCtx)->lookup(opCtx, nss).get();
+    return CollectionCatalog::get(opCtx)->lookupView(opCtx, nss).get();
 }
 
 /**
diff --git a/src/mongo/db/query/find.cpp b/src/mongo/db/query/find.cpp
index c380e4419e2..d2f36155e13 100644
--- a/src/mongo/db/query/find.cpp
+++ b/src/mongo/db/query/find.cpp
@@ -63,7 +63,6 @@
 #include "mongo/db/stats/resource_consumption_metrics.h"
 #include "mongo/db/stats/top.h"
 #include "mongo/db/storage/storage_options.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 #include "mongo/s/chunk_version.h"
 #include "mongo/s/stale_exception.h"
diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index cf8acb72165..950b18feb56 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -283,6 +283,7 @@ env.Library(
         'replication_consistency_markers_idl',
     ],
     LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/db/catalog/collection_options',
         '$BUILD_DIR/mongo/db/catalog_raii',
         '$BUILD_DIR/mongo/db/storage/journal_flusher',
     ],
diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp
index 06486b86af9..f8bf232917c 100644
--- a/src/mongo/db/repl/oplog.cpp
+++ b/src/mongo/db/repl/oplog.cpp
@@ -99,7 +99,6 @@
 #include "mongo/db/storage/storage_engine.h"
 #include "mongo/db/storage/storage_options.h"
 #include "mongo/db/transaction_participant.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 #include "mongo/platform/random.h"
 #include "mongo/rpc/get_status_from_command_result.h"
@@ -1208,7 +1207,7 @@ Status applyOperation_inlock(OperationContext* opCtx,
     const bool haveWrappingWriteUnitOfWork = opCtx->lockState()->inAWriteUnitOfWork();
     uassert(ErrorCodes::CommandNotSupportedOnView,
             str::stream() << "applyOps not supported on view: " << requestNss.ns(),
-            collection || !ViewCatalog::get(opCtx)->lookup(opCtx, requestNss));
+            collection || !CollectionCatalog::get(opCtx)->lookupView(opCtx, requestNss));
 
     // Decide whether to timestamp the write with the 'ts' field found in the operation. In general,
     // we do this for secondary oplog application, but there are some exceptions.
@@ -1821,14 +1820,8 @@ Status applyCommand_inlock(OperationContext* opCtx,
         return {ErrorCodes::InvalidNamespace, "invalid ns: " + std::string(nss.ns())};
     }
     {
-        // Command application doesn't always acquire the global writer lock for transaction
-        // commands, so we acquire its own locks here.
-        Lock::DBLock lock(opCtx, nss.db(), MODE_IS);
-        const TenantDatabaseName tenantDbName(boost::none, nss.db());
-        auto databaseHolder = DatabaseHolder::get(opCtx);
-        auto db = databaseHolder->getDb(opCtx, tenantDbName);
-        if (db && !CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, nss) &&
-            ViewCatalog::get(opCtx)->lookup(opCtx, nss)) {
+        auto catalog = CollectionCatalog::get(opCtx);
+        if (!catalog->lookupCollectionByNamespace(opCtx, nss) && catalog->lookupView(opCtx, nss)) {
             return {ErrorCodes::CommandNotSupportedOnView,
                     str::stream() << "applyOps not supported on view:" << nss.ns()};
         }
diff --git a/src/mongo/db/repl/tenant_migration_shard_merge_util.cpp b/src/mongo/db/repl/tenant_migration_shard_merge_util.cpp
index 2d9d7fa0025..3052a0ce33f 100644
--- a/src/mongo/db/repl/tenant_migration_shard_merge_util.cpp
+++ b/src/mongo/db/repl/tenant_migration_shard_merge_util.cpp
@@ -37,6 +37,7 @@
 #include <fmt/format.h>
 
 #include "mongo/bson/bsonobjbuilder.h"
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/create_collection.h"
 #include "mongo/db/catalog/uncommitted_collections.h"
 #include "mongo/db/concurrency/d_concurrency.h"
@@ -49,7 +50,6 @@
 #include "mongo/db/repl/tenant_migration_shared_data.h"
 #include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/wiredtiger/wiredtiger_import.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/future_util.h"
 
diff --git a/src/mongo/db/s/rename_collection_coordinator.cpp b/src/mongo/db/s/rename_collection_coordinator.cpp
index 577b55a4ed2..b8c882d4def 100644
--- a/src/mongo/db/s/rename_collection_coordinator.cpp
+++ b/src/mongo/db/s/rename_collection_coordinator.cpp
@@ -44,7 +44,6 @@
 #include "mongo/db/s/sharding_logging.h"
 #include "mongo/db/s/sharding_state.h"
 #include "mongo/db/s/sharding_util.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/idl/idl_parser.h"
 #include "mongo/logv2/log.h"
 #include "mongo/s/catalog/sharding_catalog_client.h"
@@ -198,15 +197,10 @@ ExecutorFuture<void> RenameCollectionCoordinator::_runImpl(
 
                     // Make sure the target namespace is not a view
                     {
-                        Lock::DBLock dbLock(opCtx, toNss.db(), MODE_IS);
-                        const TenantDatabaseName tenantDbName(boost::none, toNss.db());
-                        const auto db = DatabaseHolder::get(opCtx)->getDb(opCtx, tenantDbName);
-                        if (db) {
-                            uassert(ErrorCodes::CommandNotSupportedOnView,
-                                    str::stream() << "Can't rename to target collection `" << toNss
-                                                  << "` because it is a view.",
-                                    !ViewCatalog::get(opCtx)->lookup(opCtx, toNss));
-                        }
+                        uassert(ErrorCodes::CommandNotSupportedOnView,
+                                str::stream() << "Can't rename to target collection `" << toNss
+                                              << "` because it is a view.",
+                                !CollectionCatalog::get(opCtx)->lookupView(opCtx, toNss));
                     }
 
                     const auto optTargetCollType = getShardedCollection(opCtx, toNss);
diff --git a/src/mongo/db/s/set_shard_version_command.cpp b/src/mongo/db/s/set_shard_version_command.cpp
index 847b62da872..efc2d36aa63 100644
--- a/src/mongo/db/s/set_shard_version_command.cpp
+++ b/src/mongo/db/s/set_shard_version_command.cpp
@@ -35,6 +35,7 @@
 #include "mongo/db/auth/action_type.h"
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/auth/privilege.h"
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/client.h"
 #include "mongo/db/commands.h"
@@ -44,7 +45,6 @@
 #include "mongo/db/s/collection_sharding_runtime.h"
 #include "mongo/db/s/shard_filtering_metadata_refresh.h"
 #include "mongo/db/s/sharding_state.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/grid.h"
@@ -150,7 +150,7 @@ public:
             // for this check, only to validate if a view already exists for this namespace.
             if (autoDb->getDb() &&
                 !CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, nss) &&
-                ViewCatalog::get(opCtx)->lookupWithoutValidatingDurableViews(opCtx, nss)) {
+                CollectionCatalog::get(opCtx)->lookupViewWithoutValidatingDurable(opCtx, nss)) {
                 return true;
             }
 
diff --git a/src/mongo/db/stats/storage_stats.cpp b/src/mongo/db/stats/storage_stats.cpp
index cf7d184cab4..c67830fca04 100644
--- a/src/mongo/db/stats/storage_stats.cpp
+++ b/src/mongo/db/stats/storage_stats.cpp
@@ -32,6 +32,7 @@
 #include "mongo/platform/basic.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_catalog.h"
 #include "mongo/db/db_raii.h"
@@ -39,7 +40,6 @@
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/timeseries/bucket_catalog.h"
 #include "mongo/db/timeseries/timeseries_stats.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 
 #include "mongo/db/stats/storage_stats.h"
diff --git a/src/mongo/db/timeseries/SConscript b/src/mongo/db/timeseries/SConscript
index e9572660176..be96b02e209 100644
--- a/src/mongo/db/timeseries/SConscript
+++ b/src/mongo/db/timeseries/SConscript
@@ -29,6 +29,8 @@ env.Library(
         '$BUILD_DIR/mongo/db/catalog/database_holder',
         '$BUILD_DIR/mongo/db/commands/server_status',
         '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
+        '$BUILD_DIR/mongo/db/namespace_string',
+        '$BUILD_DIR/mongo/db/server_options_core',
         '$BUILD_DIR/mongo/db/views/views',
         '$BUILD_DIR/mongo/util/fail_point',
         'timeseries_options',
diff --git a/src/mongo/db/timeseries/bucket_catalog.cpp b/src/mongo/db/timeseries/bucket_catalog.cpp
index fda3bb92bfa..d817f3e1bc5 100644
--- a/src/mongo/db/timeseries/bucket_catalog.cpp
+++ b/src/mongo/db/timeseries/bucket_catalog.cpp
@@ -39,7 +39,6 @@
 #include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/timeseries/timeseries_options.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/platform/compiler.h"
 #include "mongo/stdx/thread.h"
 #include "mongo/util/fail_point.h"
diff --git a/src/mongo/db/timeseries/bucket_catalog_test.cpp b/src/mongo/db/timeseries/bucket_catalog_test.cpp
index e83f242df8f..93352e64bb9 100644
--- a/src/mongo/db/timeseries/bucket_catalog_test.cpp
+++ b/src/mongo/db/timeseries/bucket_catalog_test.cpp
@@ -33,7 +33,6 @@
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/timeseries/bucket_catalog.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/stdx/future.h"
 #include "mongo/unittest/bson_test_util.h"
 #include "mongo/unittest/death_test.h"
diff --git a/src/mongo/db/ttl.cpp b/src/mongo/db/ttl.cpp
index f3f945ae50c..1c91a5fc470 100644
--- a/src/mongo/db/ttl.cpp
+++ b/src/mongo/db/ttl.cpp
@@ -37,6 +37,7 @@
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/auth/user_name.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_catalog.h"
 #include "mongo/db/client.h"
@@ -58,7 +59,6 @@
 #include "mongo/db/timeseries/bucket_catalog.h"
 #include "mongo/db/ttl_collection_cache.h"
 #include "mongo/db/ttl_gen.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 #include "mongo/s/grid.h"
 #include "mongo/util/background.h"
diff --git a/src/mongo/db/views/SConscript b/src/mongo/db/views/SConscript
index 5f4388c0d22..62fb715de39 100644
--- a/src/mongo/db/views/SConscript
+++ b/src/mongo/db/views/SConscript
@@ -14,6 +14,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/views/views',
     ],
     LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/db/audit',
         '$BUILD_DIR/mongo/db/catalog/database_holder',
         '$BUILD_DIR/mongo/db/multitenancy',
     ],
@@ -23,21 +24,23 @@ env.Library(
     target='views',
     source=[
         'view.cpp',
-        'view_catalog.cpp',
         'view_graph.cpp',
     ],
     LIBDEPS=[
-        '$BUILD_DIR/mongo/base',
-        '$BUILD_DIR/mongo/db/catalog/collection',
-        '$BUILD_DIR/mongo/db/pipeline/aggregation',
         '$BUILD_DIR/mongo/db/query/collation/collator_factory_interface',
-        '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
+    ],
+)
+
+env.Library(
+    target='view_catalog_helpers',
+    source=[
+        'view_catalog_helpers.cpp',
+    ],
+    LIBDEPS=[
+        '$BUILD_DIR/mongo/db/pipeline/aggregation',
         'resolved_view',
+        'views',
     ],
-    LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/db/audit',
-        '$BUILD_DIR/mongo/db/multitenancy',
-    ]
 )
 
 env.Library(
@@ -71,6 +74,7 @@ env.CppUnitTest(
         '$BUILD_DIR/mongo/db/repl/replmocks',
         '$BUILD_DIR/mongo/s/is_mongos',
         '$BUILD_DIR/mongo/unittest/unittest',
+        'view_catalog_helpers',
         'views',
         'views_mongod',
     ],
diff --git a/src/mongo/db/views/durable_view_catalog.cpp b/src/mongo/db/views/durable_view_catalog.cpp
index f756c052d0e..bbc2d1f59ef 100644
--- a/src/mongo/db/views/durable_view_catalog.cpp
+++ b/src/mongo/db/views/durable_view_catalog.cpp
@@ -35,7 +35,9 @@
 
 #include <string>
 
+#include "mongo/db/audit.h"
 #include "mongo/db/catalog/collection.h"
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/concurrency/d_concurrency.h"
@@ -45,7 +47,6 @@
 #include "mongo/db/operation_context.h"
 #include "mongo/db/storage/record_data.h"
 #include "mongo/db/tenant_database_name.h"
-#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 #include "mongo/stdx/unordered_set.h"
 #include "mongo/util/assert_util.h"
@@ -61,15 +62,10 @@ void DurableViewCatalog::onExternalChange(OperationContext* opCtx, const Namespa
         NamespaceString(name.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
 
     // On an external change, an invalid view definition can be detected when the view catalog
-    // is reloaded. This will prevent any further usage of the view catalog until the invalid
-    // view definitions are removed. We use kValidateDurableViews here to catch any invalid view
-    // definitions in the view catalog to make it unusable for subsequent callers.
-    if (ViewCatalog::shouldIgnoreExternalChange(opCtx, name)) {
-        return;
-    }
-
-    ViewCatalog::reload(opCtx, name.db(), ViewCatalogLookupBehavior::kValidateDurableViews)
-        .ignore();
+    // is reloaded. This will prevent any further usage of the views for this database until the
+    // invalid view definitions are removed.
+    auto catalog = CollectionCatalog::get(opCtx);
+    catalog->reloadViews(opCtx, name.db()).ignore();
 }
 
 void DurableViewCatalog::onSystemViewsCollectionDrop(OperationContext* opCtx,
@@ -79,14 +75,24 @@ void DurableViewCatalog::onSystemViewsCollectionDrop(OperationContext* opCtx,
         NamespaceString(name.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
     dassert(name.coll() == NamespaceString::kSystemDotViewsCollectionName);
 
-    const TenantDatabaseName tenantDbName(boost::none, name.db());
-    auto databaseHolder = DatabaseHolder::get(opCtx);
-    auto db = databaseHolder->getDb(opCtx, tenantDbName);
-    if (db) {
-        // If the 'system.views' collection is dropped, we need to clear the in-memory state of the
-        // view catalog.
-        ViewCatalog::clear(opCtx, name.db());
-    }
+    auto catalog = CollectionCatalog::get(opCtx);
+
+    // First, iterate through the views on this database and audit them before they are dropped.
+    catalog->iterateViews(opCtx,
+                          name.db(),
+                          [&](const ViewDefinition& view) -> bool {
+                              audit::logDropView(opCtx->getClient(),
+                                                 view.name(),
+                                                 view.viewOn().ns(),
+                                                 view.pipeline(),
+                                                 ErrorCodes::OK);
+                              return true;
+                          },
+                          ViewCatalogLookupBehavior::kAllowInvalidViews);
+
+    // If the 'system.views' collection is dropped, we need to clear the in-memory state of the
+    // view catalog.
+    catalog->clearViews(opCtx, name.db());
 }
 
 // DurableViewCatalogImpl
@@ -95,17 +101,13 @@ const std::string& DurableViewCatalogImpl::getName() const {
     return _db->name().dbName();
 }
 
-bool DurableViewCatalogImpl::belongsTo(const Database* db) const {
-    return _db == db;
-}
-
 void DurableViewCatalogImpl::iterate(OperationContext* opCtx, Callback callback) {
-    _iterate(opCtx, callback, ViewCatalogLookupBehavior::kValidateDurableViews);
+    _iterate(opCtx, callback, ViewCatalogLookupBehavior::kValidateViews);
 }
 
 void DurableViewCatalogImpl::iterateIgnoreInvalidEntries(OperationContext* opCtx,
                                                          Callback callback) {
-    _iterate(opCtx, callback, ViewCatalogLookupBehavior::kAllowInvalidDurableViews);
+    _iterate(opCtx, callback, ViewCatalogLookupBehavior::kAllowInvalidViews);
 }
 
 void DurableViewCatalogImpl::_iterate(OperationContext* opCtx,
@@ -126,7 +128,7 @@ void DurableViewCatalogImpl::_iterate(OperationContext* opCtx,
             viewDefinition = _validateViewDefinition(opCtx, record->data);
             uassertStatusOK(callback(viewDefinition));
         } catch (const ExceptionFor<ErrorCodes::InvalidViewDefinition>& ex) {
-            if (lookupBehavior == ViewCatalogLookupBehavior::kValidateDurableViews) {
+            if (lookupBehavior == ViewCatalogLookupBehavior::kValidateViews) {
                 throw ex;
             }
         }
diff --git a/src/mongo/db/views/durable_view_catalog.h b/src/mongo/db/views/durable_view_catalog.h
index bf27160e3f1..0546636225d 100644
--- a/src/mongo/db/views/durable_view_catalog.h
+++ b/src/mongo/db/views/durable_view_catalog.h
@@ -44,11 +44,11 @@ class OperationContext;
 class RecordData;
 
 /**
- * ViewCatalogLookupBehavior specifies whether a lookup into the view catalog should attempt to
- * validate the durable entries that currently exist within the catalog. This validation should
+ * ViewCatalogLookupBehavior specifies whether a lookup into the view catalog should validate the
+ * entries or allow the operation to proceed with an invalid entry present. This validation should
  * rarely be skipped.
  */
-enum class ViewCatalogLookupBehavior { kValidateDurableViews, kAllowInvalidDurableViews };
+enum class ViewCatalogLookupBehavior { kValidateViews, kAllowInvalidViews };
 
 /**
  * Interface for system.views collection operations associated with view catalog management.
@@ -81,7 +81,6 @@ public:
                         const BSONObj& view) = 0;
     virtual void remove(OperationContext* opCtx, const NamespaceString& name) = 0;
     virtual const std::string& getName() const = 0;
-    virtual bool belongsTo(const Database* db) const = 0;
     virtual ~DurableViewCatalog() = default;
 };
 
@@ -100,7 +99,6 @@ public:
     void upsert(OperationContext* opCtx, const NamespaceString& name, const BSONObj& view);
     void remove(OperationContext* opCtx, const NamespaceString& name);
     const std::string& getName() const;
-    bool belongsTo(const Database* db) const;
 
 private:
     void _iterate(OperationContext* opCtx,
diff --git a/src/mongo/db/views/view_catalog.cpp b/src/mongo/db/views/view_catalog.cpp
deleted file mode 100644
index e33a250d4d6..00000000000
--- a/src/mongo/db/views/view_catalog.cpp
+++ /dev/null
@@ -1,911 +0,0 @@
-/**
- *    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::kDefault
-
-#include "mongo/platform/basic.h"
-
-#include "mongo/db/views/view_catalog.h"
-
-#include <memory>
-#include <string>
-
-#include "mongo/base/status_with.h"
-#include "mongo/base/string_data.h"
-#include "mongo/bson/util/builder.h"
-#include "mongo/db/api_parameters.h"
-#include "mongo/db/audit.h"
-#include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/database.h"
-#include "mongo/db/curop.h"
-#include "mongo/db/namespace_string.h"
-#include "mongo/db/operation_context.h"
-#include "mongo/db/pipeline/aggregate_command_gen.h"
-#include "mongo/db/pipeline/document_source.h"
-#include "mongo/db/pipeline/expression_context.h"
-#include "mongo/db/pipeline/lite_parsed_pipeline.h"
-#include "mongo/db/pipeline/pipeline.h"
-#include "mongo/db/pipeline/process_interface/stub_mongo_process_interface.h"
-#include "mongo/db/query/collation/collator_factory_interface.h"
-#include "mongo/db/storage/recovery_unit.h"
-#include "mongo/db/views/resolved_view.h"
-#include "mongo/db/views/view.h"
-#include "mongo/db/views/view_graph.h"
-#include "mongo/logv2/log.h"
-#include "mongo/util/fail_point.h"
-
-namespace mongo {
-
-namespace {
-/**
- * Helper class to manage copy-on-write for the ViewCatalog.
- */
-class ViewCatalogWriter {
-public:
-    ViewCatalogWriter(Mutex& mutex,
-                      std::shared_ptr<const ViewCatalog> instance,
-                      std::shared_ptr<ViewCatalog>* storage)
-        : _mutex(mutex), _read(std::move(instance)), _storage(storage) {}
-
-    ViewCatalogWriter(ViewCatalogWriter&&) = delete;
-    ViewCatalogWriter& operator=(ViewCatalogWriter&&) = delete;
-
-    const ViewCatalog* operator->() const {
-        if (_write)
-            return _write.get();
-
-        return _read.get();
-    }
-
-    ViewCatalog* writable() {
-        if (!_write) {
-            _lock = stdx::unique_lock<Mutex>(_mutex);
-            // TODO (SERVER-57250): This atomic_load will be deprecated in C++20
-            // We must copy from `_storage` here under the lock so we include any changes that may
-            // have happened since we copied '_read'.
-            _write = std::make_shared<ViewCatalog>(*atomic_load(_storage));
-            _read.reset();
-        }
-        return _write.get();
-    }
-
-    void commit() {
-        if (_write) {
-            atomic_store(_storage, _write);
-            // Set _read and clear _write so we can use this instance in read mode after the commit.
-            _read = std::move(_write);
-            _lock.unlock();
-        }
-    }
-
-private:
-    Mutex& _mutex;
-    stdx::unique_lock<Mutex> _lock;
-    std::shared_ptr<const ViewCatalog> _read;
-    std::shared_ptr<ViewCatalog> _write;
-    std::shared_ptr<ViewCatalog>* _storage;
-};
-
-/**
- * Decoration on the ServiceContext for storing the latest ViewCatalog instance and its associated
- * write mutex.
- */
-class ViewCatalogStorage {
-public:
-    std::shared_ptr<const ViewCatalog> get() const {
-        return atomic_load(&_catalog);
-    }
-
-    void set(std::shared_ptr<ViewCatalog> instance) {
-        atomic_store(&_catalog, std::move(instance));
-    }
-
-    ViewCatalogWriter writer() {
-        return ViewCatalogWriter(_mutex, get(), &_catalog);
-    }
-
-    void setIgnoreExternalChange(StringData dbName, bool value) {
-        stdx::lock_guard lk{_externalChangeMutex};
-        if (value) {
-            _ignoreExternalChange.emplace(dbName);
-        } else {
-            _ignoreExternalChange.erase(dbName);
-        }
-    }
-
-    bool shouldIgnoreExternalChange(StringData dbName) const {
-        stdx::lock_guard lk{_externalChangeMutex};
-        auto it = _ignoreExternalChange.find(dbName);
-        return it != _ignoreExternalChange.end();
-    }
-
-private:
-    std::shared_ptr<ViewCatalog> _catalog = std::make_shared<ViewCatalog>();
-    mutable Mutex _mutex = MONGO_MAKE_LATCH("ViewCatalogStorage::_mutex");  // Serializes writes
-    mutable Mutex _externalChangeMutex = MONGO_MAKE_LATCH(
-        "ViewCatalogStorage::_externalChangeMutex");  // Guards _ignoreExternalChange set
-    StringSet _ignoreExternalChange;
-};  // namespace
-const auto getViewCatalog = ServiceContext::declareDecoration<ViewCatalogStorage>();
-
-StatusWith<std::unique_ptr<CollatorInterface>> 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);
-}
-}  // namespace
-
-std::shared_ptr<const ViewCatalog> ViewCatalog::get(ServiceContext* svcCtx) {
-    return getViewCatalog(svcCtx).get();
-}
-
-std::shared_ptr<const ViewCatalog> ViewCatalog::get(OperationContext* opCtx) {
-    return get(opCtx->getServiceContext());
-}
-
-Status ViewCatalog::registerDatabase(OperationContext* opCtx,
-                                     StringData dbName,
-                                     std::unique_ptr<DurableViewCatalog> durable) {
-    auto catalog = getViewCatalog(opCtx->getServiceContext()).writer();
-    auto it = catalog.writable()->_viewsForDatabase.find(dbName);
-    if (it != catalog.writable()->_viewsForDatabase.end()) {
-        return {ErrorCodes::AlreadyInitialized, "ViewCatalog entry for database already set"};
-    }
-
-    auto& vfdb = catalog.writable()->_viewsForDatabase[dbName];
-    vfdb.durable = std::move(durable);
-    vfdb.valid = false;
-    vfdb.viewGraphNeedsRefresh = true;
-    catalog.commit();
-    return Status::OK();
-}
-
-void ViewCatalog::unregisterDatabase(OperationContext* opCtx, Database* db) {
-    auto catalog = getViewCatalog(opCtx->getServiceContext()).writer();
-    auto it = catalog.writable()->_viewsForDatabase.find(db->name().dbName());
-    if (it != catalog.writable()->_viewsForDatabase.end() && it->second.durable->belongsTo(db)) {
-        catalog.writable()->_viewsForDatabase.erase(it);
-        catalog.commit();
-    }
-}
-
-Status ViewCatalog::reload(OperationContext* opCtx,
-                           StringData dbName,
-                           ViewCatalogLookupBehavior lookupBehavior) {
-    auto catalog = getViewCatalog(opCtx->getServiceContext()).writer();
-    invariant(opCtx->lockState()->isCollectionLockedForMode(
-        NamespaceString(dbName, NamespaceString::kSystemDotViewsCollectionName), MODE_IS));
-    auto result = catalog.writable()->_reload(
-        opCtx, dbName, ViewCatalogLookupBehavior::kValidateDurableViews, true);
-    catalog.commit();
-    return result;
-}
-
-Status ViewCatalog::_reload(OperationContext* opCtx,
-                            StringData dbName,
-                            ViewCatalogLookupBehavior lookupBehavior,
-                            bool reloadForCollectionCatalog) {
-    LOGV2_DEBUG(22546, 1, "Reloading view catalog for database", "db"_attr = dbName);
-
-    auto it = _viewsForDatabase.find(dbName);
-    invariant(it != _viewsForDatabase.end());
-    auto& vfdb = it->second;
-
-    vfdb.viewMap.clear();
-    vfdb.valid = false;
-    vfdb.viewGraphNeedsRefresh = true;
-    vfdb.stats = {};
-
-    absl::flat_hash_set<NamespaceString> viewNamesForDb;
-
-    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()) {
-                vfdb.stats.userTimeseries += 1;
-            } else {
-                vfdb.stats.userViews += 1;
-            }
-        } else {
-            vfdb.stats.internal += 1;
-        }
-
-        vfdb.viewMap[viewName.ns()] = std::move(viewDef);
-        if (reloadForCollectionCatalog) {
-            viewNamesForDb.insert(viewName);
-        }
-        return Status::OK();
-    };
-
-    try {
-        if (lookupBehavior == ViewCatalogLookupBehavior::kValidateDurableViews) {
-            vfdb.durable->iterate(opCtx, reloadCallback);
-        } else if (lookupBehavior == ViewCatalogLookupBehavior::kAllowInvalidDurableViews) {
-            vfdb.durable->iterateIgnoreInvalidEntries(opCtx, reloadCallback);
-        } else {
-            MONGO_UNREACHABLE;
-        }
-        if (reloadForCollectionCatalog) {
-            CollectionCatalog::write(
-                opCtx,
-                [&dbName, viewsForDb = std::move(viewNamesForDb)](CollectionCatalog& catalog) {
-                    // TODO SERVER-63206: Use instead TenantDatabaseName passed in by caller.
-                    catalog.replaceViewsForDatabase(TenantDatabaseName(boost::none, dbName),
-                                                    std::move(viewsForDb));
-                });
-        }
-    } catch (const DBException& ex) {
-        auto status = ex.toStatus();
-        LOGV2(22547,
-              "Could not load view catalog for database",
-              "db"_attr = vfdb.durable->getName(),
-              "error"_attr = status);
-        return status;
-    }
-
-    vfdb.valid = true;
-    return Status::OK();
-}
-
-void ViewCatalog::clear(OperationContext* opCtx, StringData dbName) {
-    auto catalog = getViewCatalog(opCtx->getServiceContext()).writer();
-    auto it = catalog.writable()->_viewsForDatabase.find(dbName);
-    invariant(it != catalog.writable()->_viewsForDatabase.end());
-    auto& vfdb = it->second;
-
-    // First, iterate through the views on this database and audit them before they are dropped.
-    for (auto&& view : vfdb.viewMap) {
-        audit::logDropView(opCtx->getClient(),
-                           (*view.second).name(),
-                           (*view.second).viewOn().ns(),
-                           (*view.second).pipeline(),
-                           ErrorCodes::OK);
-    }
-
-    vfdb.viewMap.clear();
-    vfdb.viewGraph.clear();
-    vfdb.valid = true;
-    vfdb.viewGraphNeedsRefresh = false;
-    vfdb.stats = {};
-    // TODO SERVER-63206: Use instead TenantDatabaseName passed in by caller.
-    CollectionCatalog::write(opCtx, [db = dbName.toString()](CollectionCatalog& catalog) {
-        catalog.replaceViewsForDatabase(TenantDatabaseName(boost::none, db), {});
-    });
-    catalog.commit();
-}
-
-bool ViewCatalog::shouldIgnoreExternalChange(OperationContext* opCtx, const NamespaceString& name) {
-    return getViewCatalog(opCtx->getServiceContext()).shouldIgnoreExternalChange(name.db());
-}
-
-void ViewCatalog::ViewsForDatabase::requireValidCatalog() const {
-    uassert(ErrorCodes::InvalidViewDefinition,
-            "Invalid view definition detected in the view catalog. Remove the invalid view "
-            "manually to prevent disallowing any further usage of the view catalog.",
-            valid);
-}
-
-void ViewCatalog::iterate(StringData dbName, ViewIteratorCallback callback) const {
-    auto it = _viewsForDatabase.find(dbName);
-    if (it == _viewsForDatabase.end()) {
-        return;
-    }
-    auto& vfdb = it->second;
-
-    vfdb.requireValidCatalog();
-    for (auto&& view : vfdb.viewMap) {
-        if (!callback(*view.second)) {
-            break;
-        }
-    }
-}
-
-Status ViewCatalog::_createOrUpdateView(OperationContext* opCtx,
-                                        const NamespaceString& viewName,
-                                        const NamespaceString& viewOn,
-                                        const BSONArray& pipeline,
-                                        std::unique_ptr<CollatorInterface> collator) {
-    invariant(opCtx->lockState()->isDbLockedForMode(viewName.db(), MODE_IX));
-    invariant(opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
-    invariant(opCtx->lockState()->isCollectionLockedForMode(
-        NamespaceString(viewName.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
-
-    auto it = _viewsForDatabase.find(viewName.db());
-    invariant(it != _viewsForDatabase.end());
-    auto& vfdb = it->second;
-    vfdb.requireValidCatalog();
-
-    // Build the BSON definition for this view to be saved in the durable view catalog. If the
-    // collation is empty, omit it from the definition altogether.
-    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 = _upsertIntoGraph(opCtx, *(view.get()));
-    if (!graphStatus.isOK()) {
-        return graphStatus;
-    }
-
-    vfdb.durable->upsert(opCtx, viewName, viewDefBuilder.obj());
-    vfdb.viewMap[viewName.ns()] = view;
-
-    // Reload the view catalog with the changes applied.
-    auto res =
-        _reload(opCtx, viewName.db(), ViewCatalogLookupBehavior::kValidateDurableViews, false);
-    if (res.isOK()) {
-        // Register the view in the CollectionCatalog mapping from ResourceID->namespace
-        auto viewRid = ResourceId(RESOURCE_COLLECTION, viewName.ns());
-
-        CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
-            catalog.registerView(viewName);
-            catalog.addResource(viewRid, viewName.ns());
-        });
-
-        opCtx->recoveryUnit()->onRollback([viewName, opCtx, viewRid]() {
-            CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
-                catalog.removeResource(viewRid, viewName.ns());
-                catalog.deregisterView(viewName);
-            });
-        });
-    }
-    return res;
-}
-
-Status ViewCatalog::_upsertIntoGraph(OperationContext* opCtx, const ViewDefinition& viewDef) {
-    auto it = _viewsForDatabase.find(viewDef.name().db());
-    invariant(it != _viewsForDatabase.end());
-    auto& vfdb = it->second;
-
-    // Performs the insert into the graph.
-    auto doInsert = [this, opCtx, &vfdb](const ViewDefinition& viewDef,
-                                         bool needsValidation) -> Status {
-        // Validate that the pipeline is eligible to serve as a view definition. If it is, this
-        // will also return the set of involved namespaces.
-        auto pipelineStatus = validatePipeline(opCtx, viewDef);
-        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 vfdb.viewGraph.insertAndValidate(viewDef, refs, pipelineSize);
-        } else {
-            vfdb.viewGraph.insertWithoutValidating(viewDef, refs, pipelineSize);
-            return Status::OK();
-        }
-    };
-
-    if (vfdb.viewGraphNeedsRefresh) {
-        vfdb.viewGraph.clear();
-        for (auto&& iter : vfdb.viewMap) {
-            auto status = doInsert(*(iter.second.get()), false);
-            // If we cannot fully refresh the graph, we will keep '_viewGraphNeedsRefresh' true.
-            if (!status.isOK()) {
-                return status;
-            }
-        }
-        // Only if the inserts completed without error will we no longer need a refresh.
-        vfdb.viewGraphNeedsRefresh = false;
-    }
-
-    // Remove the view definition first in case this is an update. If it is not in the graph, it
-    // is simply a no-op.
-    vfdb.viewGraph.remove(viewDef.name());
-
-    return doInsert(viewDef, true);
-}
-
-StatusWith<stdx::unordered_set<NamespaceString>> ViewCatalog::validatePipeline(
-    OperationContext* opCtx, const ViewDefinition& viewDef) {
-    const LiteParsedPipeline liteParsedPipeline(viewDef.viewOn(), viewDef.pipeline());
-    const auto involvedNamespaces = liteParsedPipeline.getInvolvedNamespaces();
-
-    // The API version pipeline validation should be skipped for time-series view because of
-    // following reasons:
-    //     - the view pipeline is not created by (or visible to) the end-user and should be skipped.
-    //     - the view pipeline can have stages that are not allowed in stable API version '1' eg.
-    //       '$_internalUnpackBucket'.
-    bool performApiVersionChecks = !viewDef.timeseries();
-
-    liteParsedPipeline.validate(opCtx, performApiVersionChecks);
-
-    // Verify that this is a legitimate pipeline specification by making sure it parses
-    // correctly. In order to parse a pipeline we need to resolve any namespaces involved to a
-    // collection and a pipeline, but in this case we don't need this map to be accurate since
-    // we will not be evaluating the pipeline.
-    StringMap<ExpressionContext::ResolvedNamespace> resolvedNamespaces;
-    for (auto&& nss : involvedNamespaces) {
-        resolvedNamespaces[nss.coll()] = {nss, {}};
-    }
-    boost::intrusive_ptr<ExpressionContext> expCtx =
-        new ExpressionContext(opCtx,
-                              AggregateCommandRequest(viewDef.viewOn(), viewDef.pipeline()),
-                              CollatorInterface::cloneCollator(viewDef.defaultCollator()),
-                              // We can use a stub MongoProcessInterface because we are only parsing
-                              // the Pipeline for validation here. We won't do anything with the
-                              // pipeline that will require a real implementation.
-                              std::make_shared<StubMongoProcessInterface>(),
-                              std::move(resolvedNamespaces),
-                              boost::none);
-
-    // If the feature compatibility version is not kLatest, and we are validating features as
-    // primary, ban the use of new agg features introduced in kLatest to prevent them from being
-    // persisted in the catalog.
-    // (Generic FCV reference): This FCV check should exist across LTS binary versions.
-    multiversion::FeatureCompatibilityVersion fcv;
-    if (serverGlobalParams.validateFeaturesAsPrimary.load() &&
-        serverGlobalParams.featureCompatibility.isLessThan(multiversion::GenericFCV::kLatest,
-                                                           &fcv)) {
-        expCtx->maxFeatureCompatibilityVersion = fcv;
-    }
-
-    // The pipeline parser needs to know that we're parsing a pipeline for a view definition
-    // to apply some additional checks.
-    expCtx->isParsingViewDefinition = true;
-
-    try {
-        auto pipeline =
-            Pipeline::parse(viewDef.pipeline(), std::move(expCtx), [&](const Pipeline& pipeline) {
-                // Validate that the view pipeline does not contain any ineligible stages.
-                const auto& sources = pipeline.getSources();
-                const auto firstPersistentStage =
-                    std::find_if(sources.begin(), sources.end(), [](const auto& source) {
-                        return source->constraints().writesPersistentData();
-                    });
-
-                uassert(ErrorCodes::OptionNotSupportedOnView,
-                        str::stream()
-                            << "The aggregation stage "
-                            << firstPersistentStage->get()->getSourceName() << " in location "
-                            << std::distance(sources.begin(), firstPersistentStage)
-                            << " of the pipeline cannot be used in the view definition of "
-                            << viewDef.name().ns() << " because it writes to disk",
-                        firstPersistentStage == sources.end());
-
-                uassert(ErrorCodes::OptionNotSupportedOnView,
-                        "$changeStream cannot be used in a view definition",
-                        sources.empty() || !sources.front()->constraints().isChangeStreamStage());
-
-                std::for_each(sources.begin(), sources.end(), [](auto& stage) {
-                    uassert(ErrorCodes::InvalidNamespace,
-                            str::stream() << "'" << stage->getSourceName()
-                                          << "' cannot be used in a view definition",
-                            !stage->constraints().isIndependentOfAnyCollection);
-                });
-            });
-    } catch (const DBException& ex) {
-        return ex.toStatus();
-    }
-
-    return std::move(involvedNamespaces);
-}
-
-Status ViewCatalog::_validateCollation(OperationContext* opCtx,
-                                       const ViewDefinition& view,
-                                       const std::vector<NamespaceString>& refs) const {
-    for (auto&& potentialViewNss : refs) {
-        auto otherView =
-            _lookup(opCtx, potentialViewNss, ViewCatalogLookupBehavior::kValidateDurableViews);
-        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 ViewCatalog::createView(OperationContext* opCtx,
-                               const NamespaceString& viewName,
-                               const NamespaceString& viewOn,
-                               const BSONArray& pipeline,
-                               const BSONObj& collation) {
-    invariant(opCtx->lockState()->isDbLockedForMode(viewName.db(), MODE_IX));
-    invariant(opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
-    invariant(opCtx->lockState()->isCollectionLockedForMode(
-        NamespaceString(viewName.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
-
-    auto& catalogStorage = getViewCatalog(opCtx->getServiceContext());
-    auto catalog = catalogStorage.writer();
-
-    if (viewName.db() != viewOn.db())
-        return Status(ErrorCodes::BadValue,
-                      "View must be created on a view or collection in the same database");
-
-    if (catalog->_lookup(opCtx, viewName, ViewCatalogLookupBehavior::kValidateDurableViews))
-        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 = parseCollator(opCtx, collation);
-    if (!collator.isOK())
-        return collator.getStatus();
-
-    Status result = Status::OK();
-    {
-        ON_BLOCK_EXIT([&catalogStorage, &viewName] {
-            catalogStorage.setIgnoreExternalChange(viewName.db(), false);
-        });
-        catalogStorage.setIgnoreExternalChange(viewName.db(), true);
-
-        result = catalog.writable()->_createOrUpdateView(
-            opCtx, viewName, viewOn, pipeline, std::move(collator.getValue()));
-    }
-    if (result.isOK()) {
-        catalog.commit();
-    }
-    return result;
-}
-
-Status ViewCatalog::modifyView(OperationContext* opCtx,
-                               const NamespaceString& viewName,
-                               const NamespaceString& viewOn,
-                               const BSONArray& pipeline) {
-    invariant(opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_X));
-    invariant(opCtx->lockState()->isCollectionLockedForMode(
-        NamespaceString(viewName.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
-
-    auto& catalogStorage = getViewCatalog(opCtx->getServiceContext());
-    auto catalog = catalogStorage.writer();
-
-    if (viewName.db() != viewOn.db())
-        return Status(ErrorCodes::BadValue,
-                      "View must be created on a view or collection in the same database");
-
-    auto viewPtr =
-        catalog->_lookup(opCtx, viewName, ViewCatalogLookupBehavior::kValidateDurableViews);
-    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());
-
-    opCtx->recoveryUnit()->onRollback([viewName, opCtx]() {
-        auto viewRid = ResourceId(RESOURCE_COLLECTION, viewName.ns());
-
-        CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
-            catalog.addResource(viewRid, viewName.ns());
-        });
-    });
-
-    Status result = Status::OK();
-    {
-        ON_BLOCK_EXIT([&catalogStorage, &viewName] {
-            catalogStorage.setIgnoreExternalChange(viewName.db(), false);
-        });
-        catalogStorage.setIgnoreExternalChange(viewName.db(), true);
-
-        result = catalog.writable()->_createOrUpdateView(
-            opCtx,
-            viewName,
-            viewOn,
-            pipeline,
-            CollatorInterface::cloneCollator(viewPtr->defaultCollator()));
-    }
-
-    if (result.isOK()) {
-        catalog.commit();
-    }
-
-    return result;
-}
-
-Status ViewCatalog::dropView(OperationContext* opCtx, const NamespaceString& viewName) {
-    invariant(opCtx->lockState()->isDbLockedForMode(viewName.db(), MODE_IX));
-    invariant(opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
-    invariant(opCtx->lockState()->isCollectionLockedForMode(
-        NamespaceString(viewName.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
-
-    auto& catalogStorage = getViewCatalog(opCtx->getServiceContext());
-    auto catalog = catalogStorage.writer();
-
-    auto it = catalog.writable()->_viewsForDatabase.find(viewName.db());
-    invariant(it != catalog.writable()->_viewsForDatabase.end());
-    auto& vfdb = it->second;
-    vfdb.requireValidCatalog();
-
-    Status result = Status::OK();
-
-    {
-        ON_BLOCK_EXIT([&catalogStorage, &viewName] {
-            catalogStorage.setIgnoreExternalChange(viewName.db(), false);
-        });
-
-        catalogStorage.setIgnoreExternalChange(viewName.db(), true);
-
-        // Save a copy of the view definition in case we need to roll back.
-        auto viewPtr =
-            catalog->_lookup(opCtx, viewName, ViewCatalogLookupBehavior::kValidateDurableViews);
-        if (!viewPtr) {
-            return {ErrorCodes::NamespaceNotFound,
-                    str::stream() << "cannot drop missing view: " << viewName.ns()};
-        }
-
-        invariant(vfdb.valid);
-        vfdb.durable->remove(opCtx, viewName);
-        vfdb.viewGraph.remove(viewPtr->name());
-        vfdb.viewMap.erase(viewName.ns());
-
-        auto viewRid = ResourceId(RESOURCE_COLLECTION, viewName.ns());
-        CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
-            catalog.removeResource(viewRid, viewName.ns());
-        });
-
-        opCtx->recoveryUnit()->onCommit([viewName, opCtx](auto ts) {
-            CollectionCatalog::write(
-                opCtx, [&](CollectionCatalog& catalog) { catalog.deregisterView(viewName); });
-        });
-
-        opCtx->recoveryUnit()->onRollback([viewName, opCtx, viewRid]() {
-            CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
-                catalog.addResource(viewRid, viewName.ns());
-            });
-        });
-
-        // Reload the view catalog with the changes applied.
-        result = catalog.writable()->_reload(
-            opCtx, viewName.db(), ViewCatalogLookupBehavior::kValidateDurableViews, false);
-    }
-    catalog.commit();
-    return result;
-}
-
-std::shared_ptr<const ViewDefinition> ViewCatalog::_lookup(
-    OperationContext* opCtx,
-    const NamespaceString& ns,
-    ViewCatalogLookupBehavior lookupBehavior) const {
-    auto it = _viewsForDatabase.find(ns.db());
-    if (it == _viewsForDatabase.end()) {
-        return nullptr;
-    }
-    auto& vfdb = it->second;
-
-    ViewMap::const_iterator vmit = vfdb.viewMap.find(ns.ns());
-    if (vmit != vfdb.viewMap.end()) {
-        return vmit->second;
-    }
-    return nullptr;
-}
-
-std::shared_ptr<ViewDefinition> ViewCatalog::_lookup(OperationContext* opCtx,
-                                                     const NamespaceString& ns,
-                                                     ViewCatalogLookupBehavior lookupBehavior) {
-    return std::const_pointer_cast<ViewDefinition>(
-        std::as_const(*this)._lookup(opCtx, ns, lookupBehavior));
-}
-
-std::shared_ptr<const ViewDefinition> ViewCatalog::lookup(OperationContext* opCtx,
-                                                          const NamespaceString& ns) const {
-    auto it = _viewsForDatabase.find(ns.db());
-    if (it == _viewsForDatabase.end()) {
-        return nullptr;
-    }
-    auto& vfdb = it->second;
-
-    if (!vfdb.valid && opCtx->getClient()->isFromUserConnection()) {
-        // We want to avoid lookups on invalid collection names.
-        if (!NamespaceString::validCollectionName(ns.ns())) {
-            return nullptr;
-        }
-
-        // 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.
-        vfdb.requireValidCatalog();
-    }
-
-    return _lookup(opCtx, ns, ViewCatalogLookupBehavior::kValidateDurableViews);
-}
-
-std::shared_ptr<const ViewDefinition> ViewCatalog::lookupWithoutValidatingDurableViews(
-    OperationContext* opCtx, const NamespaceString& ns) const {
-    return _lookup(opCtx, ns, ViewCatalogLookupBehavior::kAllowInvalidDurableViews);
-}
-
-StatusWith<ResolvedView> ViewCatalog::resolveView(
-    OperationContext* opCtx,
-    const NamespaceString& nss,
-    boost::optional<BSONObj> timeSeriesCollator) const {
-    auto it = _viewsForDatabase.find(nss.db());
-    uassert(ErrorCodes::NamespaceNotFound,
-            str::stream() << "View " << nss << " not found",
-            it != _viewsForDatabase.end());
-    auto& vfdb = it->second;
-    vfdb.requireValidCatalog();
-
-    // Points to the name of the most resolved namespace.
-    const NamespaceString* resolvedNss = &nss;
-
-    // Holds the combination of all the resolved views.
-    std::vector<BSONObj> resolvedPipeline;
-
-    // If the catalog has not been tampered with, all views seen during the resolution will have
-    // the same collation. As an optimization, we fill out the collation spec only once.
-    boost::optional<BSONObj> collation;
-
-    // The last seen view definition, which owns the NamespaceString pointed to by
-    // 'resolvedNss'.
-    std::shared_ptr<ViewDefinition> lastViewDefinition;
-
-    std::vector<NamespaceString> dependencyChain{nss};
-
-    int depth = 0;
-    boost::optional<bool> mixedData = boost::none;
-    boost::optional<TimeseriesOptions> tsOptions = boost::none;
-    for (; depth < ViewGraph::kMaxViewDepth; depth++) {
-        auto view = _lookup(opCtx, *resolvedNss, ViewCatalogLookupBehavior::kValidateDurableViews);
-        if (!view) {
-            // Return error status if pipeline is too large.
-            int pipelineSize = 0;
-            for (auto obj : resolvedPipeline) {
-                pipelineSize += obj.objsize();
-            }
-            if (pipelineSize > ViewGraph::kMaxViewPipelineSizeBytes) {
-                return {ErrorCodes::ViewPipelineMaxSizeExceeded,
-                        str::stream() << "View pipeline exceeds maximum size; maximum size is "
-                                      << ViewGraph::kMaxViewPipelineSizeBytes};
-            }
-
-            auto curOp = CurOp::get(opCtx);
-            curOp->debug().addResolvedViews(dependencyChain, resolvedPipeline);
-
-            return StatusWith<ResolvedView>(
-                {*resolvedNss,
-                 std::move(resolvedPipeline),
-                 collation ? std::move(collation.get()) : CollationSpec::kSimpleSpec,
-                 tsOptions,
-                 mixedData});
-        }
-
-        resolvedNss = &view->viewOn();
-
-        if (storageGlobalParams.restore) {
-            // During a selective restore procedure, skip checking options as the collection may no
-            // longer exist.
-            continue;
-        }
-
-        if (view->timeseries()) {
-            // Use the lock-free collection lookup, to ensure compatibility with lock-free read
-            // operations.
-            auto tsCollection = CollectionCatalog::get(opCtx)->lookupCollectionByNamespaceForRead(
-                opCtx, *resolvedNss);
-            uassert(6067201,
-                    str::stream() << "expected time-series buckets collection " << *resolvedNss
-                                  << " to exist",
-                    tsCollection);
-            if (tsCollection) {
-                mixedData = tsCollection->getTimeseriesBucketsMayHaveMixedSchemaData();
-                tsOptions = tsCollection->getTimeseriesOptions();
-            }
-        }
-
-        dependencyChain.push_back(*resolvedNss);
-        if (!collation) {
-            if (timeSeriesCollator) {
-                collation = *timeSeriesCollator;
-            } else {
-                collation = view->defaultCollator() ? view->defaultCollator()->getSpec().toBSON()
-                                                    : CollationSpec::kSimpleSpec;
-            }
-        }
-
-        // Prepend the underlying view's pipeline to the current working pipeline.
-        const std::vector<BSONObj>& toPrepend = view->pipeline();
-        resolvedPipeline.insert(resolvedPipeline.begin(), toPrepend.begin(), toPrepend.end());
-
-        // If the first stage is a $collStats, then we return early with the viewOn namespace.
-        if (toPrepend.size() > 0 && !toPrepend[0]["$collStats"].eoo()) {
-            auto curOp = CurOp::get(opCtx);
-            curOp->debug().addResolvedViews(dependencyChain, resolvedPipeline);
-
-            return StatusWith<ResolvedView>(
-                {*resolvedNss, std::move(resolvedPipeline), std::move(collation.get())});
-        }
-    }
-
-    if (depth >= ViewGraph::kMaxViewDepth) {
-        return {ErrorCodes::ViewDepthLimitExceeded,
-                str::stream() << "View depth too deep or view cycle detected; maximum depth is "
-                              << ViewGraph::kMaxViewDepth};
-    }
-
-    MONGO_UNREACHABLE;
-}
-
-boost::optional<ViewCatalog::Stats> ViewCatalog::getStats(StringData dbName) const {
-    auto it = _viewsForDatabase.find(dbName);
-    if (it == _viewsForDatabase.end()) {
-        return boost::none;
-    }
-    auto& vfdb = it->second;
-    return vfdb.stats;
-}
-}  // namespace mongo
diff --git a/src/mongo/db/views/view_catalog.h b/src/mongo/db/views/view_catalog.h
deleted file mode 100644
index a55ff015c14..00000000000
--- a/src/mongo/db/views/view_catalog.h
+++ /dev/null
@@ -1,271 +0,0 @@
-/**
- *    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 <map>
-#include <memory>
-#include <string>
-#include <tuple>
-#include <vector>
-
-#include "mongo/base/status.h"
-#include "mongo/base/status_with.h"
-#include "mongo/base/string_data.h"
-#include "mongo/db/namespace_string.h"
-#include "mongo/db/views/durable_view_catalog.h"
-#include "mongo/db/views/resolved_view.h"
-#include "mongo/db/views/view.h"
-#include "mongo/db/views/view_graph.h"
-#include "mongo/platform/mutex.h"
-#include "mongo/util/concurrency/with_lock.h"
-#include "mongo/util/string_map.h"
-
-namespace mongo {
-class OperationContext;
-class Database;
-
-/**
- * In-memory data structure for view definitions. Instances returned by get() are immutable,
- * modifications through the static functions copy the existing instance and perform the
- * modification on the copy. A new call to get() is necessary to observe the modification.
- *
- * Writes via the static functions are thread-safe and serialized with a mutex.
- *
- * The static methods refresh the in-memory map with the views catalog collection if necessary,
- * throwing if the refresh fails.
- */
-class ViewCatalog {
-public:
-    using ViewMap = StringMap<std::shared_ptr<ViewDefinition>>;
-    using ViewIteratorCallback = std::function<bool(const ViewDefinition& view)>;
-
-    static std::shared_ptr<const ViewCatalog> get(ServiceContext* svcCtx);
-    static std::shared_ptr<const ViewCatalog> get(OperationContext* opCtx);
-
-    /**
-     * Add an entry to the ViewCatalog for the given database, backed by the durable storage
-     * 'catalog'.
-     */
-    static Status registerDatabase(OperationContext* opCtx,
-                                   StringData dbName,
-                                   std::unique_ptr<DurableViewCatalog> catalog);
-
-    /**
-     * Removes the ViewCatalog entries assocated with 'db' if any. Should be called when when a
-     * `DatabaseImpl` that has previously registered is about to be destructed (e.g. when closing a
-     * database).
-     */
-    static void unregisterDatabase(OperationContext* opCtx, Database* db);
-
-    /**
-     * Iterates through the catalog, applying 'callback' to each view. This callback function
-     * executes under the catalog's mutex, so it must not access other methods of the catalog,
-     * acquire locks or run for a long time. If the 'callback' returns false, the iterator exits
-     * early.
-     *
-     * Caller must ensure corresponding database exists.
-     */
-    void iterate(StringData dbName, ViewIteratorCallback callback) const;
-
-    /**
-     * Create a new view 'viewName' with contents defined by running the specified aggregation
-     * 'pipeline' with collation 'collation' on a collection or view 'viewOn'. This method will
-     * check correctness with respect to the view catalog, but will not check for conflicts with the
-     * database's catalog, so the check for an existing collection with the same name must be done
-     * before calling createView.
-     *
-     * Must be in WriteUnitOfWork. View creation rolls back if the unit of work aborts.
-     *
-     * Caller must ensure corresponding database exists.
-     */
-    static Status createView(OperationContext* opCtx,
-                             const NamespaceString& viewName,
-                             const NamespaceString& viewOn,
-                             const BSONArray& pipeline,
-                             const BSONObj& collation);
-
-    /**
-     * Drop the view named 'viewName'.
-     *
-     * Must be in WriteUnitOfWork. The drop rolls back if the unit of work aborts.
-     *
-     * Caller must ensure corresponding database exists.
-     */
-    static Status dropView(OperationContext* opCtx, const NamespaceString& viewName);
-
-    /**
-     * Modify the view named 'viewName' to have the new 'viewOn' and 'pipeline'.
-     *
-     * Must be in WriteUnitOfWork. The modification rolls back if the unit of work aborts.
-     *
-     * Caller must ensure corresponding database exists.
-     */
-    static Status modifyView(OperationContext* opCtx,
-                             const NamespaceString& viewName,
-                             const NamespaceString& viewOn,
-                             const BSONArray& pipeline);
-
-    /**
-     * Look up the 'nss' in the view catalog, returning a shared pointer to a View definition, or
-     * nullptr if it doesn't exist.
-     *
-     * Caller must ensure corresponding database exists.
-     */
-    std::shared_ptr<const ViewDefinition> lookup(OperationContext* opCtx,
-                                                 const NamespaceString& nss) const;
-
-    /**
-     * Same functionality as above, except this function skips validating durable views in the view
-     * catalog.
-     *
-     * Caller must ensure corresponding database exists.
-     */
-    std::shared_ptr<const ViewDefinition> lookupWithoutValidatingDurableViews(
-        OperationContext* opCtx, const NamespaceString& nss) const;
-
-    /**
-     * Resolve the views on 'nss', transforming the pipeline appropriately. This function returns a
-     * fully-resolved view definition containing the backing namespace, the resolved pipeline and
-     * the collation to use for the operation.
-     *
-     * With SERVER-54597, we allow queries on timeseries collections *only* to specify non-default
-     * collations. So in the case of queries on timeseries collections, we create a ResolvedView
-     * with the request's collation (timeSeriesCollator) rather than the collection's default
-     * collator.
-     *
-     * Caller must ensure corresponding database exists.
-     */
-    StatusWith<ResolvedView> resolveView(OperationContext* opCtx,
-                                         const NamespaceString& nss,
-                                         boost::optional<BSONObj> timeseriesCollator) const;
-
-    /**
-     * Usage statistics about this view catalog.
-     * Total views = internal + userViews + userTimeseries.
-     */
-    struct Stats {
-        int userViews = 0;
-        int userTimeseries = 0;
-        int internal = 0;
-    };
-
-    /**
-     * Returns view statistics for the specified database.
-     */
-    boost::optional<Stats> getStats(StringData dbName) const;
-
-    /**
-     * Returns Status::OK with the set of involved namespaces if the given pipeline is eligible to
-     * act as a view definition. Otherwise, returns ErrorCodes::OptionNotSupportedOnView.
-     */
-    static StatusWith<stdx::unordered_set<NamespaceString>> validatePipeline(
-        OperationContext* opCtx, const ViewDefinition& viewDef);
-
-    /**
-     * Reloads the in-memory state of the view catalog from the 'system.views' collection catalog.
-     * If the 'lookupBehavior' is 'kValidateDurableViews', then 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.
-     */
-    static Status reload(OperationContext* opCtx,
-                         StringData dbName,
-                         ViewCatalogLookupBehavior lookupBehavior);
-
-    /**
-     * Clears the in-memory state of the view catalog.
-     */
-    static void clear(OperationContext* opCtx, StringData dbName);
-
-    /**
-     * The view catalog needs to ignore external changes for its own modifications.
-     */
-    static bool shouldIgnoreExternalChange(OperationContext* opCtx, const NamespaceString& name);
-
-private:
-    Status _createOrUpdateView(OperationContext* opCtx,
-                               const NamespaceString& viewName,
-                               const NamespaceString& viewOn,
-                               const BSONArray& pipeline,
-                               std::unique_ptr<CollatorInterface> collator);
-    /**
-     * 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);
-
-    /**
-     * 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;
-
-    std::shared_ptr<const ViewDefinition> _lookup(OperationContext* opCtx,
-                                                  const NamespaceString& ns,
-                                                  ViewCatalogLookupBehavior lookupBehavior) const;
-    std::shared_ptr<ViewDefinition> _lookup(OperationContext* opCtx,
-                                            const NamespaceString& ns,
-                                            ViewCatalogLookupBehavior lookupBehavior);
-
-    Status _reload(OperationContext* opCtx,
-                   StringData dbName,
-                   ViewCatalogLookupBehavior lookupBehavior,
-                   bool reloadForCollectionCatalog);
-
-    /**
-     * Holds all data for the views associated with a particular database. Prior to 5.3, the
-     * ViewCatalog object was owned by the Database object as a decoration. It has now transitioned
-     * to a global catalog, as a decoration on the ServiceContext. Each database gets its own record
-     * here, comprising the same information that was previously stored as top-level information
-     * prior to 5.3.
-     */
-    struct ViewsForDatabase {
-        ViewMap viewMap;
-        std::shared_ptr<DurableViewCatalog> durable;
-        bool valid = false;
-        ViewGraph viewGraph;
-        bool viewGraphNeedsRefresh = true;
-        Stats stats;
-
-        /**
-         * uasserts with the InvalidViewDefinition error if the current in-memory state of the view
-         * catalog for the given database is invalid. This ensures that calling into the view
-         * catalog while it is invalid renders it inoperable.
-         */
-        void requireValidCatalog() const;
-    };
-    StringMap<ViewsForDatabase> _viewsForDatabase;
-};
-}  // namespace mongo
diff --git a/src/mongo/db/views/view_catalog_helpers.cpp b/src/mongo/db/views/view_catalog_helpers.cpp
new file mode 100644
index 00000000000..a253f16adc3
--- /dev/null
+++ b/src/mongo/db/views/view_catalog_helpers.cpp
@@ -0,0 +1,235 @@
+/**
+ *    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.
+ */
+
+#include "mongo/db/views/view_catalog_helpers.h"
+
+#include "mongo/db/catalog/collection_catalog.h"
+#include "mongo/db/curop.h"
+#include "mongo/db/pipeline/aggregate_command_gen.h"
+#include "mongo/db/pipeline/document_source.h"
+#include "mongo/db/pipeline/expression_context.h"
+#include "mongo/db/pipeline/lite_parsed_pipeline.h"
+#include "mongo/db/pipeline/pipeline.h"
+#include "mongo/db/pipeline/process_interface/stub_mongo_process_interface.h"
+#include "mongo/db/views/view_graph.h"
+
+namespace mongo {
+namespace view_catalog_helpers {
+
+StatusWith<stdx::unordered_set<NamespaceString>> validatePipeline(OperationContext* opCtx,
+                                                                  const ViewDefinition& viewDef) {
+    const LiteParsedPipeline liteParsedPipeline(viewDef.viewOn(), viewDef.pipeline());
+    const auto involvedNamespaces = liteParsedPipeline.getInvolvedNamespaces();
+
+    // The API version pipeline validation should be skipped for time-series view because of
+    // following reasons:
+    //     - the view pipeline is not created by (or visible to) the end-user and should be skipped.
+    //     - the view pipeline can have stages that are not allowed in stable API version '1' eg.
+    //       '$_internalUnpackBucket'.
+    bool performApiVersionChecks = !viewDef.timeseries();
+
+    liteParsedPipeline.validate(opCtx, performApiVersionChecks);
+
+    // Verify that this is a legitimate pipeline specification by making sure it parses
+    // correctly. In order to parse a pipeline we need to resolve any namespaces involved to a
+    // collection and a pipeline, but in this case we don't need this map to be accurate since
+    // we will not be evaluating the pipeline.
+    StringMap<ExpressionContext::ResolvedNamespace> resolvedNamespaces;
+    for (auto&& nss : involvedNamespaces) {
+        resolvedNamespaces[nss.coll()] = {nss, {}};
+    }
+    boost::intrusive_ptr<ExpressionContext> expCtx =
+        new ExpressionContext(opCtx,
+                              AggregateCommandRequest(viewDef.viewOn(), viewDef.pipeline()),
+                              CollatorInterface::cloneCollator(viewDef.defaultCollator()),
+                              // We can use a stub MongoProcessInterface because we are only parsing
+                              // the Pipeline for validation here. We won't do anything with the
+                              // pipeline that will require a real implementation.
+                              std::make_shared<StubMongoProcessInterface>(),
+                              std::move(resolvedNamespaces),
+                              boost::none);
+
+    // If the feature compatibility version is not kLatest, and we are validating features as
+    // primary, ban the use of new agg features introduced in kLatest to prevent them from being
+    // persisted in the catalog.
+    // (Generic FCV reference): This FCV check should exist across LTS binary versions.
+    multiversion::FeatureCompatibilityVersion fcv;
+    if (serverGlobalParams.validateFeaturesAsPrimary.load() &&
+        serverGlobalParams.featureCompatibility.isLessThan(multiversion::GenericFCV::kLatest,
+                                                           &fcv)) {
+        expCtx->maxFeatureCompatibilityVersion = fcv;
+    }
+
+    // The pipeline parser needs to know that we're parsing a pipeline for a view definition
+    // to apply some additional checks.
+    expCtx->isParsingViewDefinition = true;
+
+    try {
+        auto pipeline =
+            Pipeline::parse(viewDef.pipeline(), std::move(expCtx), [&](const Pipeline& pipeline) {
+                // Validate that the view pipeline does not contain any ineligible stages.
+                const auto& sources = pipeline.getSources();
+                const auto firstPersistentStage =
+                    std::find_if(sources.begin(), sources.end(), [](const auto& source) {
+                        return source->constraints().writesPersistentData();
+                    });
+
+                uassert(ErrorCodes::OptionNotSupportedOnView,
+                        str::stream()
+                            << "The aggregation stage "
+                            << firstPersistentStage->get()->getSourceName() << " in location "
+                            << std::distance(sources.begin(), firstPersistentStage)
+                            << " of the pipeline cannot be used in the view definition of "
+                            << viewDef.name().ns() << " because it writes to disk",
+                        firstPersistentStage == sources.end());
+
+                uassert(ErrorCodes::OptionNotSupportedOnView,
+                        "$changeStream cannot be used in a view definition",
+                        sources.empty() || !sources.front()->constraints().isChangeStreamStage());
+
+                std::for_each(sources.begin(), sources.end(), [](auto& stage) {
+                    uassert(ErrorCodes::InvalidNamespace,
+                            str::stream() << "'" << stage->getSourceName()
+                                          << "' cannot be used in a view definition",
+                            !stage->constraints().isIndependentOfAnyCollection);
+                });
+            });
+    } catch (const DBException& ex) {
+        return ex.toStatus();
+    }
+
+    return std::move(involvedNamespaces);
+}
+
+StatusWith<ResolvedView> resolveView(OperationContext* opCtx,
+                                     std::shared_ptr<const CollectionCatalog> catalog,
+                                     const NamespaceString& nss,
+                                     boost::optional<BSONObj> timeSeriesCollator) {
+    // Points to the name of the most resolved namespace.
+    const NamespaceString* resolvedNss = &nss;
+
+    // Holds the combination of all the resolved views.
+    std::vector<BSONObj> resolvedPipeline;
+
+    // If the catalog has not been tampered with, all views seen during the resolution will have
+    // the same collation. As an optimization, we fill out the collation spec only once.
+    boost::optional<BSONObj> collation;
+
+    // The last seen view definition, which owns the NamespaceString pointed to by
+    // 'resolvedNss'.
+    std::shared_ptr<ViewDefinition> lastViewDefinition;
+
+    std::vector<NamespaceString> dependencyChain{nss};
+
+    int depth = 0;
+    boost::optional<bool> mixedData = boost::none;
+    boost::optional<TimeseriesOptions> tsOptions = boost::none;
+
+    for (; depth < ViewGraph::kMaxViewDepth; depth++) {
+        auto view = catalog->lookupView(opCtx, *resolvedNss);
+        if (!view) {
+            // Return error status if pipeline is too large.
+            int pipelineSize = 0;
+            for (auto obj : resolvedPipeline) {
+                pipelineSize += obj.objsize();
+            }
+            if (pipelineSize > ViewGraph::kMaxViewPipelineSizeBytes) {
+                return {ErrorCodes::ViewPipelineMaxSizeExceeded,
+                        str::stream() << "View pipeline exceeds maximum size; maximum size is "
+                                      << ViewGraph::kMaxViewPipelineSizeBytes};
+            }
+
+            auto curOp = CurOp::get(opCtx);
+            curOp->debug().addResolvedViews(dependencyChain, resolvedPipeline);
+
+            return StatusWith<ResolvedView>(
+                {*resolvedNss,
+                 std::move(resolvedPipeline),
+                 collation ? std::move(collation.get()) : CollationSpec::kSimpleSpec,
+                 tsOptions,
+                 mixedData});
+        }
+
+        resolvedNss = &view->viewOn();
+
+        if (storageGlobalParams.restore) {
+            // During a selective restore procedure, skip checking options as the collection may no
+            // longer exist.
+            continue;
+        }
+
+        if (view->timeseries()) {
+            // Use the lock-free collection lookup, to ensure compatibility with lock-free read
+            // operations.
+            auto tsCollection = catalog->lookupCollectionByNamespaceForRead(opCtx, *resolvedNss);
+            uassert(6067201,
+                    str::stream() << "expected time-series buckets collection " << *resolvedNss
+                                  << " to exist",
+                    tsCollection);
+            if (tsCollection) {
+                mixedData = tsCollection->getTimeseriesBucketsMayHaveMixedSchemaData();
+                tsOptions = tsCollection->getTimeseriesOptions();
+            }
+        }
+
+        dependencyChain.push_back(*resolvedNss);
+        if (!collation) {
+            if (timeSeriesCollator) {
+                collation = *timeSeriesCollator;
+            } else {
+                collation = view->defaultCollator() ? view->defaultCollator()->getSpec().toBSON()
+                                                    : CollationSpec::kSimpleSpec;
+            }
+        }
+
+        // Prepend the underlying view's pipeline to the current working pipeline.
+        const std::vector<BSONObj>& toPrepend = view->pipeline();
+        resolvedPipeline.insert(resolvedPipeline.begin(), toPrepend.begin(), toPrepend.end());
+
+        // If the first stage is a $collStats, then we return early with the viewOn namespace.
+        if (toPrepend.size() > 0 && !toPrepend[0]["$collStats"].eoo()) {
+            auto curOp = CurOp::get(opCtx);
+            curOp->debug().addResolvedViews(dependencyChain, resolvedPipeline);
+
+            return StatusWith<ResolvedView>(
+                {*resolvedNss, std::move(resolvedPipeline), std::move(collation.get())});
+        }
+    }
+
+    if (depth >= ViewGraph::kMaxViewDepth) {
+        return {ErrorCodes::ViewDepthLimitExceeded,
+                str::stream() << "View depth too deep or view cycle detected; maximum depth is "
+                              << ViewGraph::kMaxViewDepth};
+    }
+
+    MONGO_UNREACHABLE;
+}
+
+}  // namespace view_catalog_helpers
+}  // namespace mongo
diff --git a/src/mongo/db/views/view_catalog_helpers.h b/src/mongo/db/views/view_catalog_helpers.h
new file mode 100644
index 00000000000..02f87dd2b4e
--- /dev/null
+++ b/src/mongo/db/views/view_catalog_helpers.h
@@ -0,0 +1,69 @@
+/**
+ *    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 <boost/optional.hpp>
+
+#include "mongo/base/status_with.h"
+#include "mongo/db/catalog/collection_catalog.h"
+#include "mongo/db/namespace_string.h"
+#include "mongo/db/operation_context.h"
+#include "mongo/db/views/resolved_view.h"
+#include "mongo/db/views/view.h"
+
+namespace mongo {
+namespace view_catalog_helpers {
+
+/**
+ * Returns Status::OK with the set of involved namespaces if the given pipeline is eligible to
+ * act as a view definition. Otherwise, returns ErrorCodes::OptionNotSupportedOnView.
+ */
+StatusWith<stdx::unordered_set<NamespaceString>> validatePipeline(OperationContext* opCtx,
+                                                                  const ViewDefinition& viewDef);
+
+/**
+ * Resolve the views on 'nss', transforming the pipeline appropriately. This function returns a
+ * fully-resolved view definition containing the backing namespace, the resolved pipeline and
+ * the collation to use for the operation.
+ *
+ * With SERVER-54597, we allow queries on timeseries collections *only* to specify non-default
+ * collations. So in the case of queries on timeseries collections, we create a ResolvedView
+ * with the request's collation (timeSeriesCollator) rather than the collection's default
+ * collator.
+ *
+ * Caller must ensure corresponding database exists.
+ */
+StatusWith<ResolvedView> resolveView(OperationContext* opCtx,
+                                     std::shared_ptr<const CollectionCatalog> catalog,
+                                     const NamespaceString& nss,
+                                     boost::optional<BSONObj> timeseriesCollator);
+
+}  // namespace view_catalog_helpers
+}  // namespace mongo
diff --git a/src/mongo/db/views/view_catalog_test.cpp b/src/mongo/db/views/view_catalog_test.cpp
index 8fb0712477f..119b5b5456a 100644
--- a/src/mongo/db/views/view_catalog_test.cpp
+++ b/src/mongo/db/views/view_catalog_test.cpp
@@ -53,7 +53,7 @@
 #include "mongo/db/service_context.h"
 #include "mongo/db/views/durable_view_catalog.h"
 #include "mongo/db/views/view.h"
-#include "mongo/db/views/view_catalog.h"
+#include "mongo/db/views/view_catalog_helpers.h"
 #include "mongo/db/views/view_graph.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/str.h"
@@ -105,8 +105,8 @@ public:
         CatalogTestFixture::tearDown();
     }
 
-    auto getViewCatalog() {
-        return ViewCatalog::get(operationContext());
+    auto getCatalog() {
+        return CollectionCatalog::get(operationContext());
     }
 
     Status createView(OperationContext* opCtx,
@@ -122,7 +122,8 @@ public:
             MODE_X);
 
         WriteUnitOfWork wuow(opCtx);
-        Status s = ViewCatalog::createView(opCtx, viewName, viewOn, pipeline, collation);
+        Status s = getCatalog()->createView(
+            opCtx, viewName, viewOn, pipeline, collation, view_catalog_helpers::validatePipeline);
         wuow.commit();
 
         return s;
@@ -140,7 +141,8 @@ public:
             MODE_X);
 
         WriteUnitOfWork wuow(opCtx);
-        Status s = ViewCatalog::modifyView(opCtx, viewName, viewOn, pipeline);
+        Status s = getCatalog()->modifyView(
+            opCtx, viewName, viewOn, pipeline, view_catalog_helpers::validatePipeline);
         wuow.commit();
 
         return s;
@@ -155,7 +157,7 @@ public:
             MODE_X);
 
         WriteUnitOfWork wuow(opCtx);
-        Status s = ViewCatalog::dropView(opCtx, viewName);
+        Status s = getCatalog()->dropView(opCtx, viewName);
         wuow.commit();
 
         return s;
@@ -168,7 +170,7 @@ public:
     std::shared_ptr<const ViewDefinition> lookup(OperationContext* opCtx,
                                                  const NamespaceString& ns) {
         Lock::DBLock dbLock(operationContext(), NamespaceString(ns).db(), MODE_IS);
-        return getViewCatalog()->lookup(operationContext(), ns);
+        return getCatalog()->lookupView(operationContext(), ns);
     }
 
 private:
@@ -516,8 +518,7 @@ TEST_F(ViewCatalogFixture, LookupRIDExistingView) {
     ASSERT_OK(createView(operationContext(), viewName, viewOn, emptyPipeline, emptyCollation));
 
     auto resourceID = ResourceId(RESOURCE_COLLECTION, "db.view"_sd);
-    auto collectionCatalog = CollectionCatalog::get(operationContext());
-    ASSERT(collectionCatalog->lookupResourceName(resourceID).get() == "db.view");
+    ASSERT(getCatalog()->lookupResourceName(resourceID).get() == "db.view");
 }
 
 TEST_F(ViewCatalogFixture, LookupRIDExistingViewRollback) {
@@ -532,12 +533,15 @@ TEST_F(ViewCatalogFixture, LookupRIDExistingViewRollback) {
             MODE_X);
 
         WriteUnitOfWork wunit(operationContext());
-        ASSERT_OK(ViewCatalog::createView(
-            operationContext(), viewName, viewOn, emptyPipeline, emptyCollation));
+        ASSERT_OK(getCatalog()->createView(operationContext(),
+                                           viewName,
+                                           viewOn,
+                                           emptyPipeline,
+                                           emptyCollation,
+                                           view_catalog_helpers::validatePipeline));
     }
     auto resourceID = ResourceId(RESOURCE_COLLECTION, "db.view"_sd);
-    auto collectionCatalog = CollectionCatalog::get(operationContext());
-    ASSERT(!collectionCatalog->lookupResourceName(resourceID));
+    ASSERT(!getCatalog()->lookupResourceName(resourceID));
 }
 
 TEST_F(ViewCatalogFixture, LookupRIDAfterDrop) {
@@ -548,8 +552,7 @@ TEST_F(ViewCatalogFixture, LookupRIDAfterDrop) {
     ASSERT_OK(dropView(operationContext(), viewName));
 
     auto resourceID = ResourceId(RESOURCE_COLLECTION, "db.view"_sd);
-    auto collectionCatalog = CollectionCatalog::get(operationContext());
-    ASSERT(!collectionCatalog->lookupResourceName(resourceID));
+    ASSERT(!getCatalog()->lookupResourceName(resourceID));
 }
 
 TEST_F(ViewCatalogFixture, LookupRIDAfterDropRollback) {
@@ -560,9 +563,8 @@ TEST_F(ViewCatalogFixture, LookupRIDAfterDropRollback) {
     {
         WriteUnitOfWork wunit(operationContext());
         ASSERT_OK(createView(operationContext(), viewName, viewOn, emptyPipeline, emptyCollation));
-        ASSERT(CollectionCatalog::get(operationContext())->lookupResourceName(resourceID).get() ==
-               viewName.ns());
         wunit.commit();
+        ASSERT(getCatalog()->lookupResourceName(resourceID).get() == viewName.ns());
     }
 
     {
@@ -574,11 +576,11 @@ TEST_F(ViewCatalogFixture, LookupRIDAfterDropRollback) {
             MODE_X);
 
         WriteUnitOfWork wunit(operationContext());
-        ASSERT_OK(ViewCatalog::dropView(operationContext(), viewName));
+        ASSERT_OK(getCatalog()->dropView(operationContext(), viewName));
+        // Do not commit, rollback.
     }
-
-    ASSERT(CollectionCatalog::get(operationContext())->lookupResourceName(resourceID).get() ==
-           viewName.ns());
+    // Make sure drop was rolled back and view is still in catalog.
+    ASSERT(getCatalog()->lookupResourceName(resourceID).get() == viewName.ns());
 }
 
 TEST_F(ViewCatalogFixture, LookupRIDAfterModify) {
@@ -588,8 +590,7 @@ TEST_F(ViewCatalogFixture, LookupRIDAfterModify) {
     auto resourceID = ResourceId(RESOURCE_COLLECTION, "db.view"_sd);
     ASSERT_OK(createView(operationContext(), viewName, viewOn, emptyPipeline, emptyCollation));
     ASSERT_OK(modifyView(operationContext(), viewName, viewOn, emptyPipeline));
-    ASSERT(CollectionCatalog::get(operationContext())->lookupResourceName(resourceID).get() ==
-           viewName.ns());
+    ASSERT(getCatalog()->lookupResourceName(resourceID).get() == viewName.ns());
 }
 
 TEST_F(ViewCatalogFixture, LookupRIDAfterModifyRollback) {
@@ -600,10 +601,10 @@ TEST_F(ViewCatalogFixture, LookupRIDAfterModifyRollback) {
     {
         WriteUnitOfWork wunit(operationContext());
         ASSERT_OK(createView(operationContext(), viewName, viewOn, emptyPipeline, emptyCollation));
-        ASSERT(CollectionCatalog::get(operationContext())->lookupResourceName(resourceID).get() ==
-               viewName.ns());
         wunit.commit();
+        ASSERT(getCatalog()->lookupResourceName(resourceID).get() == viewName.ns());
     }
+
     {
         Lock::DBLock dbLock(operationContext(), viewName.db(), MODE_IX);
         Lock::CollectionLock collLock(operationContext(), viewName, MODE_X);
@@ -613,12 +614,16 @@ TEST_F(ViewCatalogFixture, LookupRIDAfterModifyRollback) {
             MODE_X);
 
         WriteUnitOfWork wunit(operationContext());
-        ASSERT_OK(ViewCatalog::modifyView(operationContext(), viewName, viewOn, emptyPipeline));
-        ASSERT(CollectionCatalog::get(operationContext())->lookupResourceName(resourceID).get() ==
-               viewName.ns());
+        ASSERT_OK(getCatalog()->modifyView(operationContext(),
+                                           viewName,
+                                           viewOn,
+                                           emptyPipeline,
+                                           view_catalog_helpers::validatePipeline));
+        ASSERT(getCatalog()->lookupResourceName(resourceID).get() == viewName.ns());
+        // Do not commit, rollback.
     }
-    ASSERT(CollectionCatalog::get(operationContext())->lookupResourceName(resourceID).get() ==
-           viewName.ns());
+    // Make sure view resource is still available after rollback.
+    ASSERT(getCatalog()->lookupResourceName(resourceID).get() == viewName.ns());
 }
 
 TEST_F(ViewCatalogFixture, CreateViewThenDropAndLookup) {
@@ -644,7 +649,7 @@ TEST_F(ViewCatalogFixture, Iterate) {
     std::set<std::string> viewNames = {"db.view1", "db.view2", "db.view3"};
 
     Lock::DBLock dbLock(operationContext(), "db", MODE_IX);
-    getViewCatalog()->iterate("db", [&viewNames](const ViewDefinition& view) {
+    getCatalog()->iterateViews(operationContext(), "db", [&viewNames](const ViewDefinition& view) {
         std::string name = view.name().toString();
         ASSERT(viewNames.end() != viewNames.find(name));
         viewNames.erase(name);
@@ -672,7 +677,8 @@ TEST_F(ViewCatalogFixture, ResolveViewCorrectPipeline) {
     ASSERT_OK(createView(operationContext(), view3, view2, pipeline3.arr(), emptyCollation));
 
     Lock::DBLock dbLock(operationContext(), "db", MODE_IX);
-    auto resolvedView = getViewCatalog()->resolveView(operationContext(), view3, boost::none);
+    auto resolvedView =
+        view_catalog_helpers::resolveView(operationContext(), getCatalog(), view3, boost::none);
     ASSERT(resolvedView.isOK());
 
     std::vector<BSONObj> expected = {BSON("$match" << BSON("foo" << 1)),
@@ -692,8 +698,8 @@ TEST_F(ViewCatalogFixture, ResolveViewOnCollectionNamespace) {
     const NamespaceString collectionNamespace("db.coll");
 
     Lock::DBLock dbLock(operationContext(), "db", MODE_IS);
-    auto resolvedView = uassertStatusOK(
-        getViewCatalog()->resolveView(operationContext(), collectionNamespace, boost::none));
+    auto resolvedView = uassertStatusOK(view_catalog_helpers::resolveView(
+        operationContext(), getCatalog(), collectionNamespace, boost::none));
 
     ASSERT_EQ(resolvedView.getNamespace(), collectionNamespace);
     ASSERT_EQ(resolvedView.getPipeline().size(), 0U);
@@ -716,7 +722,8 @@ TEST_F(ViewCatalogFixture, ResolveViewCorrectlyExtractsDefaultCollation) {
     ASSERT_OK(createView(operationContext(), view2, view1, pipeline2.arr(), collation));
 
     Lock::DBLock dbLock(operationContext(), "db", MODE_IS);
-    auto resolvedView = getViewCatalog()->resolveView(operationContext(), view2, boost::none);
+    auto resolvedView =
+        view_catalog_helpers::resolveView(operationContext(), getCatalog(), view2, boost::none);
     ASSERT(resolvedView.isOK());
 
     ASSERT_EQ(resolvedView.getValue().getNamespace(), viewOn);