SERVER-57250 Merge ViewCatalog into CollectionCatalog

8 files changed, 389 insertions, 1256 deletions

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);