SERVER-63731 Initialize views in `CollectionCatalog`

30 files changed, 696 insertions, 941 deletions

diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index f9642dbd38e..236ec7eb15c 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -1294,6 +1294,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/catalog/collection',
+        '$BUILD_DIR/mongo/db/catalog/collection_catalog',
         '$BUILD_DIR/mongo/db/catalog/collection_validation',
         '$BUILD_DIR/mongo/db/catalog/database_holder',
         '$BUILD_DIR/mongo/db/catalog/document_validation',
@@ -2283,7 +2284,6 @@ env.Library(
         'ttl_d',
         'update/update_driver',
         'update_index_data',
-        'views/views_mongod',
         'windows_options' if env.TargetOSIs('windows') else [],
     ],
     LIBDEPS=[
diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript
index 824f94a4a8e..c99ee4074eb 100644
--- a/src/mongo/db/catalog/SConscript
+++ b/src/mongo/db/catalog/SConscript
@@ -293,7 +293,11 @@ env.Library(
         'views_for_database.cpp',
     ],
     LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/db/audit',
+        '$BUILD_DIR/mongo/db/catalog/collection_crud',
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
+        '$BUILD_DIR/mongo/db/curop',
+        '$BUILD_DIR/mongo/db/index/index_access_method',
         '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/profile_filter',
         '$BUILD_DIR/mongo/db/query/collation/collator_factory_interface',
@@ -304,6 +308,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/storage/bson_collection_catalog_entry',
         '$BUILD_DIR/mongo/db/storage/snapshot_helper',
         '$BUILD_DIR/mongo/db/storage/storage_options',
+        '$BUILD_DIR/mongo/db/views/util',
         '$BUILD_DIR/mongo/db/views/views',
         '$BUILD_DIR/mongo/util/namespace_string_util',
         'collection',
@@ -386,6 +391,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/index/index_access_methods',
         '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/op_observer/op_observer',
+        '$BUILD_DIR/mongo/db/query_exec',
         '$BUILD_DIR/mongo/db/record_id_helpers',
         '$BUILD_DIR/mongo/db/repl/drop_pending_collection_reaper',
         '$BUILD_DIR/mongo/db/repl/oplog',
@@ -408,7 +414,6 @@ env.Library(
         '$BUILD_DIR/mongo/db/ttl_collection_cache',
         '$BUILD_DIR/mongo/db/vector_clock',
         '$BUILD_DIR/mongo/db/views/view_catalog_helpers',
-        '$BUILD_DIR/mongo/db/views/views_mongod',
         'catalog_helpers',
         'catalog_stats',
         'clustered_collection_options',
diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index 79ec4f243e7..00992cd519e 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -79,6 +79,13 @@ bool isCollectionCompatible(std::shared_ptr<Collection> coll, Timestamp readTime
     return readTimestamp >= *minValidSnapshot;
 }
 
+void assertViewCatalogValid(const ViewsForDatabase& viewsForDb) {
+    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.",
+            viewsForDb.valid());
+}
+
 }  // namespace
 
 class IgnoreExternalViewChangesForDatabase {
@@ -587,12 +594,10 @@ Status CollectionCatalog::createView(OperationContext* opCtx,
                                      const NamespaceString& viewName,
                                      const NamespaceString& viewOn,
                                      const BSONArray& pipeline,
+                                     const ViewsForDatabase::PipelineValidatorFn& validatePipeline,
                                      const BSONObj& collation,
-                                     const ViewsForDatabase::PipelineValidatorFn& pipelineValidator,
-                                     const ViewUpsertMode insertViewMode) const {
-    // A view document direct write can occur via the oplog application path, which may only hold a
-    // lock on the collection being updated (the database views collection).
-    invariant(insertViewMode == ViewUpsertMode::kAlreadyDurableView ||
+                                     ViewsForDatabase::Durability durability) const {
+    invariant(durability == ViewsForDatabase::Durability::kAlreadyDurable ||
               opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
     invariant(opCtx->lockState()->isCollectionLockedForMode(
         NamespaceString(viewName.dbName(), NamespaceString::kSystemDotViewsCollectionName),
@@ -617,25 +622,24 @@ Status CollectionCatalog::createView(OperationContext* opCtx,
         return Status(ErrorCodes::InvalidNamespace,
                       str::stream() << "invalid name for 'viewOn': " << viewOn.coll());
 
-    auto collator = ViewsForDatabase::parseCollator(opCtx, collation);
-    if (!collator.isOK())
-        return collator.getStatus();
+    IgnoreExternalViewChangesForDatabase ignore(opCtx, viewName.dbName());
 
-    Status result = Status::OK();
-    {
-        IgnoreExternalViewChangesForDatabase ignore(opCtx, viewName.dbName());
+    assertViewCatalogValid(viewsForDb);
+    auto systemViews = _lookupSystemViews(opCtx, viewName.dbName());
+
+    ViewsForDatabase writable{viewsForDb};
+    auto status = writable.insert(
+        opCtx, systemViews, viewName, viewOn, pipeline, validatePipeline, collation, durability);
+
+    if (status.isOK()) {
+        auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
+        uncommittedCatalogUpdates.addView(opCtx, viewName);
+        uncommittedCatalogUpdates.replaceViewsForDatabase(viewName.dbName(), std::move(writable));
 
-        result = _createOrUpdateView(opCtx,
-                                     viewName,
-                                     viewOn,
-                                     pipeline,
-                                     pipelineValidator,
-                                     std::move(collator.getValue()),
-                                     ViewsForDatabase{viewsForDb},
-                                     insertViewMode);
+        PublishCatalogUpdates::ensureRegisteredWithRecoveryUnit(opCtx, uncommittedCatalogUpdates);
     }
 
-    return result;
+    return status;
 }
 
 Status CollectionCatalog::modifyView(
@@ -643,7 +647,7 @@ Status CollectionCatalog::modifyView(
     const NamespaceString& viewName,
     const NamespaceString& viewOn,
     const BSONArray& pipeline,
-    const ViewsForDatabase::PipelineValidatorFn& pipelineValidator) const {
+    const ViewsForDatabase::PipelineValidatorFn& validatePipeline) const {
     invariant(opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_X));
     invariant(opCtx->lockState()->isCollectionLockedForMode(
         NamespaceString(viewName.dbName(), NamespaceString::kSystemDotViewsCollectionName),
@@ -664,21 +668,29 @@ Status CollectionCatalog::modifyView(
         return Status(ErrorCodes::InvalidNamespace,
                       str::stream() << "invalid name for 'viewOn': " << viewOn.coll());
 
-    Status result = Status::OK();
-    {
-        IgnoreExternalViewChangesForDatabase ignore(opCtx, viewName.dbName());
+    IgnoreExternalViewChangesForDatabase ignore(opCtx, viewName.dbName());
+
+    assertViewCatalogValid(viewsForDb);
+    auto systemViews = _lookupSystemViews(opCtx, viewName.dbName());
 
-        result = _createOrUpdateView(opCtx,
-                                     viewName,
-                                     viewOn,
-                                     pipeline,
-                                     pipelineValidator,
-                                     CollatorInterface::cloneCollator(viewPtr->defaultCollator()),
-                                     ViewsForDatabase{viewsForDb},
-                                     ViewUpsertMode::kUpdateView);
+    ViewsForDatabase writable{viewsForDb};
+    auto status = writable.update(opCtx,
+                                  systemViews,
+                                  viewName,
+                                  viewOn,
+                                  pipeline,
+                                  validatePipeline,
+                                  CollatorInterface::cloneCollator(viewPtr->defaultCollator()));
+
+    if (status.isOK()) {
+        auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
+        uncommittedCatalogUpdates.addView(opCtx, viewName);
+        uncommittedCatalogUpdates.replaceViewsForDatabase(viewName.dbName(), std::move(writable));
+
+        PublishCatalogUpdates::ensureRegisteredWithRecoveryUnit(opCtx, uncommittedCatalogUpdates);
     }
 
-    return result;
+    return status;
 }
 
 Status CollectionCatalog::dropView(OperationContext* opCtx, const NamespaceString& viewName) const {
@@ -688,7 +700,7 @@ Status CollectionCatalog::dropView(OperationContext* opCtx, const NamespaceStrin
         MODE_X));
     invariant(_viewsForDatabase.contains(viewName.dbName()));
     const ViewsForDatabase& viewsForDb = *_getViewsForDatabase(opCtx, viewName.dbName());
-    viewsForDb.requireValidCatalog();
+    assertViewCatalogValid(viewsForDb);
 
     // Make sure the view exists before proceeding.
     if (auto viewPtr = viewsForDb.lookup(viewName); !viewPtr) {
@@ -700,15 +712,13 @@ Status CollectionCatalog::dropView(OperationContext* opCtx, const NamespaceStrin
     {
         IgnoreExternalViewChangesForDatabase ignore(opCtx, viewName.dbName());
 
-        ViewsForDatabase writable{viewsForDb};
+        auto systemViews = _lookupSystemViews(opCtx, viewName.dbName());
 
-        writable.durable->remove(opCtx, viewName);
-        writable.viewGraph.remove(viewName);
-        writable.viewMap.erase(viewName);
-        writable.stats = {};
+        ViewsForDatabase writable{viewsForDb};
+        writable.remove(opCtx, systemViews, viewName);
 
         // Reload the view catalog with the changes applied.
-        result = writable.reload(opCtx);
+        result = writable.reload(opCtx, systemViews);
         if (result.isOK()) {
             auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
             uncommittedCatalogUpdates.removeView(viewName);
@@ -734,17 +744,8 @@ Status CollectionCatalog::reloadViews(OperationContext* opCtx, const DatabaseNam
 
     LOGV2_DEBUG(22546, 1, "Reloading view catalog for database", "db"_attr = dbName.toString());
 
-    // 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);
+    ViewsForDatabase viewsForDb;
+    auto status = viewsForDb.reload(opCtx, _lookupSystemViews(opCtx, dbName));
     CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
         catalog._replaceViewsForDatabase(dbName, std::move(viewsForDb));
     });
@@ -939,17 +940,6 @@ void CollectionCatalog::dropCollection(OperationContext* opCtx,
     PublishCatalogUpdates::ensureRegisteredWithRecoveryUnit(opCtx, uncommittedCatalogUpdates);
 }
 
-void CollectionCatalog::onOpenDatabase(OperationContext* opCtx,
-                                       const DatabaseName& 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, DatabaseName dbName) {
     invariant(opCtx->lockState()->isDbLockedForMode(dbName, MODE_X));
     ResourceCatalog::get(opCtx->getServiceContext()).remove({RESOURCE_DATABASE, dbName}, dbName);
@@ -1239,24 +1229,17 @@ boost::optional<RecordId> CollectionCatalog::lookupCatalogIdByNSS(
     return boost::none;
 }
 
-void CollectionCatalog::iterateViews(OperationContext* opCtx,
-                                     const DatabaseName& dbName,
-                                     ViewIteratorCallback callback,
-                                     ViewCatalogLookupBehavior lookupBehavior) const {
+void CollectionCatalog::iterateViews(
+    OperationContext* opCtx,
+    const DatabaseName& dbName,
+    const std::function<bool(const ViewDefinition& view)>& callback) 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;
-        }
-    }
+    assertViewCatalogValid(*viewsForDb);
+    viewsForDb->iterate(callback);
 }
 
 std::shared_ptr<const ViewDefinition> CollectionCatalog::lookupView(
@@ -1266,7 +1249,7 @@ std::shared_ptr<const ViewDefinition> CollectionCatalog::lookupView(
         return nullptr;
     }
 
-    if (!viewsForDb->valid && opCtx->getClient()->isFromUserConnection()) {
+    if (!viewsForDb->valid() && opCtx->getClient()->isFromUserConnection()) {
         // We want to avoid lookups on invalid collection names.
         if (!NamespaceString::validCollectionName(ns.ns())) {
             return nullptr;
@@ -1275,7 +1258,7 @@ std::shared_ptr<const ViewDefinition> CollectionCatalog::lookupView(
         // 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();
+        assertViewCatalogValid(*viewsForDb);
     }
 
     return viewsForDb->lookup(ns);
@@ -1418,10 +1401,7 @@ CollectionCatalog::Stats CollectionCatalog::getStats() const {
 boost::optional<ViewsForDatabase::Stats> CollectionCatalog::getViewStatsForDatabase(
     OperationContext* opCtx, const DatabaseName& dbName) const {
     auto viewsForDb = _getViewsForDatabase(opCtx, dbName);
-    if (!viewsForDb) {
-        return boost::none;
-    }
-    return viewsForDb->stats;
+    return viewsForDb ? boost::make_optional(viewsForDb->stats()) : boost::none;
 }
 
 CollectionCatalog::ViewCatalogSet CollectionCatalog::getViewCatalogDbNames(
@@ -1504,6 +1484,19 @@ void CollectionCatalog::_registerCollection(OperationContext* opCtx,
     auto& resourceCatalog = ResourceCatalog::get(opCtx->getServiceContext());
     resourceCatalog.add({RESOURCE_DATABASE, nss.dbName()}, nss.dbName());
     resourceCatalog.add({RESOURCE_COLLECTION, nss}, nss);
+
+    if (!storageGlobalParams.repair && coll->ns().isSystemDotViews()) {
+        auto [it, emplaced] = _viewsForDatabase.try_emplace(coll->ns().dbName());
+        if (auto status = it->second.reload(opCtx, _lookupSystemViews(opCtx, coll->ns().dbName()));
+            !status.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(status),
+                                  logAttrs(coll->ns()));
+        }
+    }
 }
 
 std::shared_ptr<Collection> CollectionCatalog::deregisterCollection(
@@ -1560,6 +1553,10 @@ std::shared_ptr<Collection> CollectionCatalog::deregisterCollection(
 
     ResourceCatalog::get(opCtx->getServiceContext()).remove({RESOURCE_COLLECTION, ns}, ns);
 
+    if (!storageGlobalParams.repair && coll->ns().isSystemDotViews()) {
+        _viewsForDatabase.erase(coll->ns().dbName());
+    }
+
     return coll;
 }
 
@@ -1746,13 +1743,10 @@ void CollectionCatalog::clearViews(OperationContext* opCtx, const DatabaseName&
 
     auto it = _viewsForDatabase.find(dbName);
     invariant(it != _viewsForDatabase.end());
+
     ViewsForDatabase viewsForDb = it->second;
+    viewsForDb.clear(opCtx);
 
-    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));
     });
@@ -1912,6 +1906,12 @@ void CollectionCatalog::invariantHasExclusiveAccessToCollection(OperationContext
               nss.toString());
 }
 
+CollectionPtr CollectionCatalog::_lookupSystemViews(OperationContext* opCtx,
+                                                    const DatabaseName& dbName) const {
+    return lookupCollectionByNamespace(opCtx,
+                                       {dbName, NamespaceString::kSystemDotViewsCollectionName});
+}
+
 boost::optional<const ViewsForDatabase&> CollectionCatalog::_getViewsForDatabase(
     OperationContext* opCtx, const DatabaseName& dbName) const {
     auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
@@ -1932,83 +1932,6 @@ void CollectionCatalog::_replaceViewsForDatabase(const DatabaseName& dbName,
     _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,
-    ViewUpsertMode insertViewMode) const {
-    // A view document direct write can occur via the oplog application path, which may only hold a
-    // lock on the collection being updated (the database views collection).
-    invariant(insertViewMode == ViewUpsertMode::kAlreadyDurableView ||
-              opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
-    invariant(opCtx->lockState()->isCollectionLockedForMode(
-        NamespaceString(viewName.dbName(), NamespaceString::kSystemDotViewsCollectionName),
-        MODE_X));
-
-    viewsForDb.requireValidCatalog();
-
-    // Build the BSON definition for this view to be saved in the durable view catalog and/or to
-    // insert in the viewMap. If the collation is empty, omit it from the definition altogether.
-    BSONObjBuilder viewDefBuilder;
-    viewDefBuilder.append("_id", NamespaceStringUtil::serialize(viewName));
-    viewDefBuilder.append("viewOn", viewOn.coll());
-    viewDefBuilder.append("pipeline", pipeline);
-    if (collator) {
-        viewDefBuilder.append("collation", collator->getSpec().toBSON());
-    }
-
-    BSONObj viewDef = viewDefBuilder.obj();
-    BSONObj ownedPipeline = pipeline.getOwned();
-    ViewDefinition view(
-        viewName.dbName(), viewName.coll(), viewOn.coll(), ownedPipeline, std::move(collator));
-
-    // If the view is already in the durable view catalog, we don't need to validate the graph. If
-    // we need to update the durable view catalog, we need to check that the resulting dependency
-    // graph is acyclic and within the maximum depth.
-    const bool viewGraphNeedsValidation = insertViewMode != ViewUpsertMode::kAlreadyDurableView;
-    Status graphStatus =
-        viewsForDb.upsertIntoGraph(opCtx, view, pipelineValidator, viewGraphNeedsValidation);
-    if (!graphStatus.isOK()) {
-        return graphStatus;
-    }
-
-    if (insertViewMode != ViewUpsertMode::kAlreadyDurableView) {
-        viewsForDb.durable->upsert(opCtx, viewName, viewDef);
-    }
-
-    viewsForDb.valid = false;
-    auto res = [&] {
-        switch (insertViewMode) {
-            case ViewUpsertMode::kCreateView:
-            case ViewUpsertMode::kAlreadyDurableView:
-                return viewsForDb.insert(opCtx, viewDef, viewName.tenantId());
-            case ViewUpsertMode::kUpdateView:
-                viewsForDb.viewMap.clear();
-                viewsForDb.viewGraphNeedsRefresh = true;
-                viewsForDb.stats = {};
-
-                // Reload the view catalog with the changes applied.
-                return viewsForDb.reload(opCtx);
-        }
-        MONGO_UNREACHABLE;
-    }();
-
-    if (res.isOK()) {
-        auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
-        uncommittedCatalogUpdates.addView(opCtx, viewName);
-        uncommittedCatalogUpdates.replaceViewsForDatabase(viewName.dbName(), std::move(viewsForDb));
-
-        PublishCatalogUpdates::ensureRegisteredWithRecoveryUnit(opCtx, uncommittedCatalogUpdates);
-    }
-
-    return res;
-}
-
-
 bool CollectionCatalog::_isCatalogBatchWriter() const {
     return batchedCatalogWriteInstance.get() == this;
 }
diff --git a/src/mongo/db/catalog/collection_catalog.h b/src/mongo/db/catalog/collection_catalog.h
index a33c5efee8f..e66a275e08d 100644
--- a/src/mongo/db/catalog/collection_catalog.h
+++ b/src/mongo/db/catalog/collection_catalog.h
@@ -49,7 +49,6 @@ class CollectionCatalog {
 
 public:
     using CollectionInfoFn = std::function<bool(const CollectionPtr& collection)>;
-    using ViewIteratorCallback = std::function<bool(const ViewDefinition& view)>;
 
     // Number of how many Collection references for a single Collection that is stored in the
     // catalog. Used to determine whether there are external references (uniquely owned). Needs to
@@ -112,19 +111,6 @@ public:
         }
     };
 
-    enum class ViewUpsertMode {
-        // Insert all data for that view into the view map, view graph, and durable view catalog.
-        kCreateView,
-
-        // Insert into the view map and view graph without reinserting the view into the durable
-        // view catalog. Skip view graph validation.
-        kAlreadyDurableView,
-
-        // Reload the view map, insert into the view graph (flagging it as needing refresh), and
-        // update the durable view catalog.
-        kUpdateView,
-    };
-
     static std::shared_ptr<const CollectionCatalog> get(ServiceContext* svcCtx);
     static std::shared_ptr<const CollectionCatalog> get(OperationContext* opCtx);
 
@@ -153,20 +139,22 @@ public:
 
     /**
      * Create a new view 'viewName' with contents defined by running the specified aggregation
-     * 'pipeline' with collation 'collation' on a collection or view 'viewOn'.
+     * 'pipeline' with collation 'collation' on a collection or view 'viewOn'. May insert this view
+     * into the system.views collection depending on 'durability'.
      *
      * Must be in WriteUnitOfWork. View creation rolls back if the unit of work aborts.
      *
      * Caller must ensure corresponding database exists. Expects db.system.views MODE_X lock and
-     * view namespace MODE_IX lock (unless 'insertViewMode' is set to kAlreadyDurableView).
+     * view namespace MODE_IX lock (unless 'durability' is set to kAlreadyDurable).
      */
     Status createView(OperationContext* opCtx,
                       const NamespaceString& viewName,
                       const NamespaceString& viewOn,
                       const BSONArray& pipeline,
+                      const ViewsForDatabase::PipelineValidatorFn& validatePipeline,
                       const BSONObj& collation,
-                      const ViewsForDatabase::PipelineValidatorFn& pipelineValidator,
-                      ViewUpsertMode insertViewMode = ViewUpsertMode::kCreateView) const;
+                      ViewsForDatabase::Durability durability =
+                          ViewsForDatabase::Durability::kNotYetDurable) const;
 
     /**
      * Drop the view named 'viewName'.
@@ -188,7 +176,7 @@ public:
                       const NamespaceString& viewName,
                       const NamespaceString& viewOn,
                       const BSONArray& pipeline,
-                      const ViewsForDatabase::PipelineValidatorFn& pipelineValidator) const;
+                      const ViewsForDatabase::PipelineValidatorFn& validatePipeline) const;
 
     /**
      * Reloads the in-memory state of the view catalog from the 'system.views' collection. The
@@ -263,14 +251,6 @@ public:
     void dropCollection(OperationContext* opCtx, Collection* coll, bool isDropPending) const;
 
     /**
-     * Initializes view records for database 'dbName'. Can throw a 'WriteConflictException' if this
-     * database has already been initialized.
-     */
-    void onOpenDatabase(OperationContext* opCtx,
-                        const DatabaseName& dbName,
-                        ViewsForDatabase&& viewsForDb);
-
-    /**
      * Removes the view records associated with 'dbName', if any, from the in-memory
      * representation of the catalog. Should be called when Database instance is closed. Requires X
      * lock on database namespace.
@@ -426,11 +406,9 @@ public:
      *
      * Caller must ensure corresponding database exists.
      */
-    void iterateViews(
-        OperationContext* opCtx,
-        const DatabaseName& dbName,
-        ViewIteratorCallback callback,
-        ViewCatalogLookupBehavior lookupBehavior = ViewCatalogLookupBehavior::kValidateViews) const;
+    void iterateViews(OperationContext* opCtx,
+                      const DatabaseName& dbName,
+                      const std::function<bool(const ViewDefinition& view)>& callback) const;
 
     /**
      * Look up the 'nss' in the view catalog, returning a shared pointer to a View definition,
@@ -634,6 +612,8 @@ private:
 
     std::shared_ptr<Collection> _lookupCollectionByUUID(UUID uuid) const;
 
+    CollectionPtr _lookupSystemViews(OperationContext* opCtx, const DatabaseName& dbName) const;
+
     /**
      * Retrieves the views for a given database, including any uncommitted changes for this
      * operation.
@@ -653,18 +633,6 @@ private:
     void _replaceViewsForDatabase(const DatabaseName& 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,
-                               ViewUpsertMode insertViewMode) const;
-
-    /**
      * Returns true if this CollectionCatalog instance is part of an ongoing batched catalog write.
      */
     bool _isCatalogBatchWriter() const;
diff --git a/src/mongo/db/catalog/database.h b/src/mongo/db/catalog/database.h
index bd8ea3dbed1..98731e53421 100644
--- a/src/mongo/db/catalog/database.h
+++ b/src/mongo/db/catalog/database.h
@@ -84,7 +84,7 @@ public:
     /**
      * Sets up internal memory structures.
      */
-    virtual Status init(OperationContext* opCtx) = 0;
+    virtual void init(OperationContext* opCtx) = 0;
 
     virtual const DatabaseName& name() const = 0;
 
diff --git a/src/mongo/db/catalog/database_holder_impl.cpp b/src/mongo/db/catalog/database_holder_impl.cpp
index b14fe192576..39449948c54 100644
--- a/src/mongo/db/catalog/database_holder_impl.cpp
+++ b/src/mongo/db/catalog/database_holder_impl.cpp
@@ -128,10 +128,6 @@ Database* DatabaseHolderImpl::openDb(OperationContext* opCtx,
         if (it != _dbs.end() && !it->second) {
             _dbs.erase(it);
         }
-
-        // 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.
-        _c.notify_all();
     });
 
     // Check casing in lock to avoid transient duplicates.
@@ -153,54 +149,22 @@ Database* DatabaseHolderImpl::openDb(OperationContext* opCtx,
     }
 
     std::unique_ptr<DatabaseImpl> newDb = std::make_unique<DatabaseImpl>(dbName);
-    Status status = newDb->init(opCtx);
-    while (!status.isOK()) {
-        // If we get here, then initializing the database failed because another concurrent writer
-        // already registered their own Database instance with the ViewCatalog. We need to wait for
-        // them to finish.
-        lk.lock();
-
-        auto it = _dbs.find(dbName);
-        if (it != _dbs.end() && it->second) {
-            // Creating databases only requires a DB lock in MODE_IX. Thus databases can be created
-            // concurrently. If this thread "lost the race", return the database object that was
-            // persisted in the `_dbs` map.
-            removeDbGuard.dismiss();
-            return it->second;
-        }
-
-        // Consider using OperationContext::waitForConditionOrInterrupt if the logic here changes
-        // in such a way that we can easily express it as a predicate for that function.
-        _c.wait_for(lk, stdx::chrono::milliseconds(1));
-
-        it = _dbs.find(dbName);
-        if (it != _dbs.end() && it->second) {
-            // As above, another writer finished successfully, return the persisted object.
-            removeDbGuard.dismiss();
-            return it->second;
-        }
-
-        lk.unlock();
-
-        // Before we continue make sure we haven't been killed
-        opCtx->checkForInterrupt();
-
-        // At this point it's possible that the other writer just hasn't finished yet, or that they
-        // failed. In either case, we should check and see if we can initialize the database now.
-        status = newDb->init(opCtx);
-    }
+    newDb->init(opCtx);
 
     // Finally replace our nullptr entry with the new Database pointer.
     removeDbGuard.dismiss();
     lk.lock();
+    auto it = _dbs.find(dbName);
+    invariant(it != _dbs.end());
+    if (it->second) {
+        // Creating databases only requires a DB lock in MODE_IX, thus databases can be concurrently
+        // created. If this thread lost the race, return the database object that was already
+        // created.
+        return it->second;
+    }
+    it->second = newDb.release();
 
-    invariant(!_dbs[dbName]);
-    auto* db = newDb.release();
-    _dbs[dbName] = db;
-    invariant(_getNamesWithConflictingCasing_inlock(dbName).empty());
-    _c.notify_all();
-
-    return db;
+    return it->second;
 }
 
 void DatabaseHolderImpl::dropDb(OperationContext* opCtx, Database* db) {
diff --git a/src/mongo/db/catalog/database_holder_impl.h b/src/mongo/db/catalog/database_holder_impl.h
index 96500a6f4a5..1c96bfdf42b 100644
--- a/src/mongo/db/catalog/database_holder_impl.h
+++ b/src/mongo/db/catalog/database_holder_impl.h
@@ -32,7 +32,6 @@
 #include "mongo/db/catalog/database_holder.h"
 
 #include "mongo/db/database_name.h"
-#include "mongo/stdx/condition_variable.h"
 #include "mongo/util/concurrency/mutex.h"
 #include "mongo/util/string_map.h"
 
@@ -77,7 +76,6 @@ private:
 
     typedef stdx::unordered_map<DatabaseName, Database*> DBs;
     mutable SimpleMutex _m;
-    mutable stdx::condition_variable _c;
     DBs _dbs;
 };
 
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index 84d11b97f6f..fd637657b8e 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -154,9 +154,9 @@ Status DatabaseImpl::validateDBName(StringData dbname) {
 }
 
 DatabaseImpl::DatabaseImpl(const DatabaseName& dbName)
-    : _name(dbName), _viewsName(_name, DurableViewCatalog::viewsCollectionName().toString()) {}
+    : _name(dbName), _viewsName(_name, NamespaceString::kSystemDotViewsCollectionName) {}
 
-Status DatabaseImpl::init(OperationContext* const opCtx) {
+void DatabaseImpl::init(OperationContext* const opCtx) {
     Status status = validateDBName(_name.db());
 
     if (!status.isOK()) {
@@ -179,48 +179,6 @@ 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 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.
-        //
-        // 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, 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, 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;
-        }
-    }
-
     // When in restore mode, views created on collections that weren't restored will be removed. We
     // only do this during startup when the global lock is held.
     if (storageGlobalParams.restore && opCtx->lockState()->isW()) {
@@ -289,8 +247,6 @@ Status DatabaseImpl::init(OperationContext* const opCtx) {
                           "reason"_attr = e.reason());
         }
     }
-
-    return status;
 }
 
 void DatabaseImpl::setDropPending(OperationContext* opCtx, bool dropPending) {
@@ -758,8 +714,8 @@ Status DatabaseImpl::createView(OperationContext* opCtx,
                                                            viewName,
                                                            viewOnNss,
                                                            pipeline,
-                                                           options.collation,
-                                                           view_catalog_helpers::validatePipeline);
+                                                           view_catalog_helpers::validatePipeline,
+                                                           options.collation);
     }
 
     audit::logCreateView(
diff --git a/src/mongo/db/catalog/database_impl.h b/src/mongo/db/catalog/database_impl.h
index 489df90ab64..42d18f8a8d7 100644
--- a/src/mongo/db/catalog/database_impl.h
+++ b/src/mongo/db/catalog/database_impl.h
@@ -39,7 +39,7 @@ class DatabaseImpl final : public Database {
 public:
     explicit DatabaseImpl(const DatabaseName& dbName);
 
-    Status init(OperationContext*) final;
+    void init(OperationContext*) final;
 
     const DatabaseName& name() const final {
         return _name;
diff --git a/src/mongo/db/catalog/views_for_database.cpp b/src/mongo/db/catalog/views_for_database.cpp
index 927d362f95d..cc375e975a2 100644
--- a/src/mongo/db/catalog/views_for_database.cpp
+++ b/src/mongo/db/catalog/views_for_database.cpp
@@ -1,5 +1,5 @@
 /**
- *    Copyright (C) 2018-present MongoDB, Inc.
+ *    Copyright (C) 2022-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,
@@ -30,104 +30,200 @@
 
 #include "views_for_database.h"
 
+#include "mongo/db/audit.h"
+#include "mongo/db/catalog/collection_write_path.h"
+#include "mongo/db/curop.h"
+#include "mongo/db/index/index_access_method.h"
 #include "mongo/db/multitenancy_gen.h"
 #include "mongo/db/server_feature_flags_gen.h"
+#include "mongo/db/views/util.h"
 #include "mongo/logv2/log.h"
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kStorage
 
 
 namespace mongo {
+namespace {
+RecordId find(OperationContext* opCtx,
+              const CollectionPtr& systemViews,
+              const NamespaceString& ns) {
+    return systemViews->getIndexCatalog()
+        ->findIdIndex(opCtx)
+        ->getEntry()
+        ->accessMethod()
+        ->asSortedData()
+        ->findSingle(
+            opCtx,
+            systemViews,
+            BSON("_id" <<
+                 // TODO SERVER-67155 Move this check into a function on NamespaceString.
+                 (!gMultitenancySupport ||
+                          (serverGlobalParams.featureCompatibility.isVersionInitialized() &&
+                           gFeatureFlagRequireTenantID.isEnabled(
+                               serverGlobalParams.featureCompatibility))
+                      ? ns.toString()
+                      : ns.toStringWithTenantId())));
+}
 
-StatusWith<std::unique_ptr<CollatorInterface>> ViewsForDatabase::parseCollator(
-    OperationContext* opCtx, BSONObj collationSpec) {
+StatusWith<std::unique_ptr<CollatorInterface>> parseCollator(OperationContext* opCtx,
+                                                             const BSONObj& collator) {
     // 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);
+    return !collator.isEmpty()
+        ? CollatorFactoryInterface::get(opCtx->getServiceContext())->makeFromBSON(collator)
+        : nullptr;
 }
+}  // namespace
 
-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 {
+    auto it = _viewMap.find(ns.coll());
+    return it != _viewMap.end() ? it->second : nullptr;
 }
 
-std::shared_ptr<const ViewDefinition> ViewsForDatabase::lookup(const NamespaceString& ns) const {
-    ViewMap::const_iterator it = viewMap.find(ns);
-    if (it != viewMap.end()) {
-        return it->second;
+void ViewsForDatabase::iterate(
+    const std::function<bool(const ViewDefinition& view)>& callback) const {
+    for (auto&& view : _viewMap) {
+        if (!callback(*view.second)) {
+            return;
+        }
     }
-    return nullptr;
 }
 
-Status ViewsForDatabase::reload(OperationContext* opCtx) {
-    try {
-        durable->iterate(opCtx, [&](const BSONObj& view) {
-            return _insert(opCtx, view, durable->getName().tenantId());
-        });
-    } 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;
+Status ViewsForDatabase::reload(OperationContext* opCtx, const CollectionPtr& systemViews) {
+    _viewMap.clear();
+    _valid = false;
+    _viewGraphNeedsRefresh = true;
+    _stats = {};
+
+    if (!systemViews) {
+        _valid = true;
+        return Status::OK();
     }
-    valid = true;
+
+    invariant(opCtx->lockState()->isCollectionLockedForMode(systemViews->ns(), MODE_IS));
+
+    auto cursor = systemViews->getCursor(opCtx);
+    while (auto record = cursor->next()) {
+        // Check the document is valid BSON, with only the expected fields.
+        // Use the latest BSON validation version. Existing view definitions are allowed to contain
+        // decimal data even if decimal is disabled.
+        fassert(40224, validateBSON(record->data.data(), record->data.size()));
+
+        auto view = record->data.toBson();
+        try {
+            view_util::validateViewDefinitionBSON(opCtx, view, systemViews->ns().dbName());
+        } catch (const DBException& ex) {
+            return ex.toStatus();
+        }
+
+        auto collatorElem = view["collation"];
+        auto collator = parseCollator(opCtx, collatorElem ? collatorElem.Obj() : BSONObj{});
+        if (!collator.isOK()) {
+            return collator.getStatus();
+        }
+
+        auto viewName = NamespaceStringUtil::deserialize(systemViews->ns().tenantId(),
+                                                         view.getStringField("_id"));
+
+        if (auto status = _upsertIntoMap(
+                opCtx,
+                std::make_shared<ViewDefinition>(viewName.dbName(),
+                                                 viewName.coll(),
+                                                 view.getStringField("viewOn"),
+                                                 BSONArray{view.getObjectField("pipeline")},
+                                                 std::move(collator.getValue())));
+            !status.isOK()) {
+            LOGV2(22547,
+                  "Could not load view catalog for database",
+                  "db"_attr = systemViews->ns().dbName(),
+                  "error"_attr = status);
+
+            return status;
+        }
+    }
+
+    _valid = true;
     return Status::OK();
 }
 
-
 Status ViewsForDatabase::insert(OperationContext* opCtx,
-                                const BSONObj& view,
-                                const boost::optional<TenantId>& tenantId) {
-    auto status = _insert(opCtx, view, tenantId);
-    if (!status.isOK()) {
-        LOGV2(5387000,
-              "Could not insert view",
-              "db"_attr = durable->getName(),
-              "error"_attr = status);
+                                const CollectionPtr& systemViews,
+                                const NamespaceString& viewName,
+                                const NamespaceString& viewOn,
+                                const BSONArray& pipeline,
+                                const PipelineValidatorFn& validatePipeline,
+                                const BSONObj& collator,
+                                Durability durability) {
+    _valid = false;
+
+    auto parsedCollator = parseCollator(opCtx, collator);
+    if (!parsedCollator.isOK()) {
+        return parsedCollator.getStatus();
+    }
+
+    auto view = std::make_shared<ViewDefinition>(viewName.dbName(),
+                                                 viewName.coll(),
+                                                 viewOn.coll(),
+                                                 pipeline,
+                                                 std::move(parsedCollator.getValue()));
+
+    // Skip validating the view graph if the view is already durable.
+    if (auto status = _upsertIntoGraph(
+            opCtx, *view, validatePipeline, durability == Durability::kNotYetDurable);
+        !status.isOK()) {
+        return status;
+    }
+
+    if (durability == Durability::kNotYetDurable) {
+        if (auto status = _upsertIntoCatalog(opCtx, systemViews, *view); !status.isOK()) {
+            return status;
+        }
+    }
+
+    if (auto status = _upsertIntoMap(opCtx, std::move(view)); !status.isOK()) {
+        LOGV2(
+            5387000, "Could not insert view", "db"_attr = viewName.dbName(), "error"_attr = status);
         return status;
     }
-    valid = true;
+
+    _valid = true;
     return Status::OK();
 };
 
-Status ViewsForDatabase::_insert(OperationContext* opCtx,
-                                 const BSONObj& view,
-                                 const boost::optional<TenantId>& tenantId) {
-    BSONObj collationSpec = view.hasField("collation") ? view["collation"].Obj() : BSONObj();
-    auto collator = parseCollator(opCtx, collationSpec);
-    if (!collator.isOK()) {
-        return collator.getStatus();
+Status ViewsForDatabase::update(OperationContext* opCtx,
+                                const CollectionPtr& systemViews,
+                                const NamespaceString& viewName,
+                                const NamespaceString& viewOn,
+                                const BSONArray& pipeline,
+                                const PipelineValidatorFn& validatePipeline,
+                                std::unique_ptr<CollatorInterface> collator) {
+    _valid = false;
+
+    auto view = std::make_shared<ViewDefinition>(
+        viewName.dbName(), viewName.coll(), viewOn.coll(), pipeline, std::move(collator));
+
+    if (auto status = _upsertIntoGraph(opCtx, *view, validatePipeline, true); !status.isOK()) {
+        return status;
     }
 
-    NamespaceString viewName = NamespaceStringUtil::deserialize(tenantId, view["_id"].str());
+    if (auto status = _upsertIntoCatalog(opCtx, systemViews, *view); !status.isOK()) {
+        return status;
+    }
 
-    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());
-        }
+    if (auto status = reload(opCtx, systemViews); !status.isOK()) {
+        return status;
     }
 
-    auto viewDef = std::make_shared<ViewDefinition>(viewName.dbName(),
-                                                    viewName.coll(),
-                                                    view["viewOn"].str(),
-                                                    pipeline,
-                                                    std::move(collator.getValue()));
+    _valid = true;
+    return Status::OK();
+}
 
+Status ViewsForDatabase::_upsertIntoMap(OperationContext* opCtx,
+                                        std::shared_ptr<ViewDefinition> view) {
     // Cannot have a secondary view on a system.buckets collection, only the time-series
     // collection view.
-    if (viewDef->viewOn().isTimeseriesBucketsCollection() &&
-        viewDef->name() != viewDef->viewOn().getTimeseriesViewNamespace()) {
+    if (view->viewOn().isTimeseriesBucketsCollection() &&
+        view->name() != view->viewOn().getTimeseriesViewNamespace()) {
         return {
             ErrorCodes::InvalidNamespace,
             "Invalid view: cannot define a view over a system.buckets namespace except by "
@@ -135,41 +231,24 @@ Status ViewsForDatabase::_insert(OperationContext* opCtx,
         };
     }
 
-    if (!viewName.isOnInternalDb() && !viewName.isSystem()) {
-        if (viewDef->timeseries()) {
-            stats.userTimeseries += 1;
+    if (!view->name().isOnInternalDb() && !view->name().isSystem()) {
+        if (view->timeseries()) {
+            _stats.userTimeseries += 1;
         } else {
-            stats.userViews += 1;
+            _stats.userViews += 1;
         }
     } else {
-        stats.internal += 1;
+        _stats.internal += 1;
     }
 
-    viewMap[viewName] = std::move(viewDef);
+    _viewMap[view->name().coll()] = view;
     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,
-                                         const bool needsValidation) {
+Status ViewsForDatabase::_upsertIntoGraph(OperationContext* opCtx,
+                                          const ViewDefinition& viewDef,
+                                          const PipelineValidatorFn& validatePipeline,
+                                          bool needsValidation) {
     // Performs the insert into the graph.
     auto doInsert = [this, opCtx, &validatePipeline](const ViewDefinition& viewDef,
                                                      bool needsValidation) -> Status {
@@ -196,20 +275,20 @@ Status ViewsForDatabase::upsertIntoGraph(OperationContext* opCtx,
 
         if (needsValidation) {
             // Check the collation of all the dependent namespaces before updating the graph.
-            auto collationStatus = validateCollation(opCtx, viewDef, refs);
+            auto collationStatus = _validateCollation(opCtx, viewDef, refs);
             if (!collationStatus.isOK()) {
                 return collationStatus;
             }
-            return viewGraph.insertAndValidate(viewDef, refs, pipelineSize);
+            return _viewGraph.insertAndValidate(viewDef, refs, pipelineSize);
         } else {
-            viewGraph.insertWithoutValidating(viewDef, refs, pipelineSize);
+            _viewGraph.insertWithoutValidating(viewDef, refs, pipelineSize);
             return Status::OK();
         }
     };
 
-    if (viewGraphNeedsRefresh) {
-        viewGraph.clear();
-        for (auto&& iter : viewMap) {
+    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()) {
@@ -217,14 +296,118 @@ Status ViewsForDatabase::upsertIntoGraph(OperationContext* opCtx,
             }
         }
         // Only if the inserts completed without error will we no longer need a refresh.
-        viewGraphNeedsRefresh = false;
+        _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());
+    _viewGraph.remove(viewDef.name());
 
     return doInsert(viewDef, needsValidation);
 }
 
+Status ViewsForDatabase::_upsertIntoCatalog(OperationContext* opCtx,
+                                            const CollectionPtr& systemViews,
+                                            const ViewDefinition& view) {
+    // Build the BSON definition for this view to be saved in the durable view catalog and/or to
+    // insert in the viewMap. If the collation is empty, omit it from the definition altogether.
+    BSONObjBuilder viewDefBuilder;
+    viewDefBuilder.append("_id", NamespaceStringUtil::serialize(view.name()));
+    viewDefBuilder.append("viewOn", view.viewOn().coll());
+    viewDefBuilder.append("pipeline", view.pipeline());
+    if (auto collator = view.defaultCollator()) {
+        viewDefBuilder.append("collation", collator->getSpec().toBSON());
+    }
+    auto viewObj = viewDefBuilder.obj();
+
+    auto id = find(opCtx, systemViews, view.name());
+    Snapshotted<BSONObj> oldView;
+    if (!id.isValid() || !systemViews->findDoc(opCtx, id, &oldView)) {
+        LOGV2_DEBUG(22544,
+                    2,
+                    "Insert view to system views catalog",
+                    "view"_attr = view.name(),
+                    "viewCatalog"_attr = systemViews->ns());
+
+        if (auto status = collection_internal::insertDocument(
+                opCtx, systemViews, InsertStatement{viewObj}, &CurOp::get(opCtx)->debug());
+            !status.isOK()) {
+            return status;
+        }
+    } else {
+        CollectionUpdateArgs args;
+        args.update = viewObj;
+        args.criteria = BSON("_id" << NamespaceStringUtil::serialize(view.name()));
+
+        systemViews->updateDocument(opCtx,
+                                    id,
+                                    oldView,
+                                    viewObj,
+                                    true /* indexesAffected */,
+                                    &CurOp::get(opCtx)->debug(),
+                                    &args);
+    }
+
+    return Status::OK();
+}
+
+void ViewsForDatabase::remove(OperationContext* opCtx,
+                              const CollectionPtr& systemViews,
+                              const NamespaceString& ns) {
+    dassert(opCtx->lockState()->isDbLockedForMode(systemViews->ns().dbName(), MODE_IX));
+    dassert(opCtx->lockState()->isCollectionLockedForMode(ns, MODE_IX));
+    dassert(opCtx->lockState()->isCollectionLockedForMode(systemViews->ns(), MODE_X));
+
+    _viewGraph.remove(ns);
+    _viewMap.erase(ns.coll());
+    _stats = {};
+
+    auto id = find(opCtx, systemViews, ns);
+    if (!id.isValid()) {
+        return;
+    }
+
+    LOGV2_DEBUG(22545,
+                2,
+                "Remove view from system views catalog",
+                "view"_attr = ns,
+                "viewCatalog"_attr = systemViews->ns());
+
+    systemViews->deleteDocument(opCtx, kUninitializedStmtId, id, &CurOp::get(opCtx)->debug());
+}
+
+void ViewsForDatabase::clear(OperationContext* opCtx) {
+    for (auto&& [name, view] : _viewMap) {
+        audit::logDropView(opCtx->getClient(),
+                           view->name(),
+                           view->viewOn().ns(),
+                           view->pipeline(),
+                           ErrorCodes::OK);
+    }
+
+    _viewMap.clear();
+    _viewGraph.clear();
+    _valid = true;
+    _viewGraphNeedsRefresh = false;
+    _stats = {};
+}
+
+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();
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/views_for_database.h b/src/mongo/db/catalog/views_for_database.h
index ed5acb98c17..de9b40978da 100644
--- a/src/mongo/db/catalog/views_for_database.h
+++ b/src/mongo/db/catalog/views_for_database.h
@@ -1,5 +1,5 @@
 /**
- *    Copyright (C) 2018-present MongoDB, Inc.
+ *    Copyright (C) 2022-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,
@@ -33,7 +33,6 @@
 
 #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"
@@ -46,7 +45,6 @@ namespace mongo {
  */
 class ViewsForDatabase {
 public:
-    using ViewMap = stdx::unordered_map<NamespaceString, std::shared_ptr<ViewDefinition>>;
     using PipelineValidatorFn = std::function<StatusWith<stdx::unordered_set<NamespaceString>>(
         OperationContext*, const ViewDefinition&)>;
 
@@ -60,69 +58,93 @@ public:
         int internal = 0;
     };
 
-    /**
-     * Helper method to build a collator from its spec.
-     */
-    static StatusWith<std::unique_ptr<CollatorInterface>> parseCollator(OperationContext* opCtx,
-                                                                        BSONObj collationSpec);
+    enum class Durability {
+        // The view is not yet inserted into the system.views collection.
+        kNotYetDurable,
 
-    std::shared_ptr<DurableViewCatalog> durable;
-    ViewMap viewMap;
-    bool valid = false;
-    ViewGraph viewGraph;
-    bool viewGraphNeedsRefresh = true;
-    Stats stats;
-    bool ignoreExternalChange = false;
+        // The view is already present in the system.views collection.
+        kAlreadyDurable,
+    };
 
-    /**
-     * 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;
+    bool valid() const {
+        return _valid;
+    }
+
+    Stats stats() const {
+        return _stats;
+    }
+
+    std::shared_ptr<const ViewDefinition> lookup(const NamespaceString& ns) const;
+
+    void iterate(const std::function<bool(const ViewDefinition& view)>& callback) const;
 
     /**
-     * Returns the 'ViewDefiniton' assocated with namespace 'ns' if one exists, nullptr otherwise.
+     * Reloads views from the system.views collection.
      */
-    std::shared_ptr<const ViewDefinition> lookup(const NamespaceString& ns) const;
+    Status reload(OperationContext* opCtx, const CollectionPtr& systemViews);
 
+    Status insert(OperationContext* opCtx,
+                  const CollectionPtr& systemViews,
+                  const NamespaceString& viewName,
+                  const NamespaceString& viewOn,
+                  const BSONArray& pipeline,
+                  const PipelineValidatorFn& validatePipeline,
+                  const BSONObj& collator,
+                  Durability durability);
+
+    Status update(OperationContext* opCtx,
+                  const CollectionPtr& systemViews,
+                  const NamespaceString& viewName,
+                  const NamespaceString& viewOn,
+                  const BSONArray& pipeline,
+                  const PipelineValidatorFn& validatePipeline,
+                  std::unique_ptr<CollatorInterface> collator);
+
+    void remove(OperationContext* opCtx,
+                const CollectionPtr& systemViews,
+                const NamespaceString& ns);
+
+    void clear(OperationContext* opCtx);
+
+private:
     /**
-     * Reloads the views for this database by iterating the DurableViewCatalog.
+     * Inserts or updates the given view into the view map.
      */
-    Status reload(OperationContext* opCtx);
+    Status _upsertIntoMap(OperationContext* opCtx, std::shared_ptr<ViewDefinition> view);
 
     /**
-     * Inserts the view into the view map.
+     * 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.
+     * needsValidation controls whether we check that the resulting dependency graph is acyclic and
+     * within the maximum depth.
      */
-    Status insert(OperationContext* opCtx,
-                  const BSONObj& view,
-                  const boost::optional<TenantId>& tenantId);
+    Status _upsertIntoGraph(OperationContext* opCtx,
+                            const ViewDefinition& viewDef,
+                            const PipelineValidatorFn& validatePipeline,
+                            bool needsValidation);
 
     /**
-     * Returns Status::OK if each view namespace in 'refs' has the same default collation as
-     * 'view'. Otherwise, returns ErrorCodes::OptionNotSupportedOnView.
+     * Inserts or updates the given view into the system.views collection.
      */
-    Status validateCollation(OperationContext* opCtx,
-                             const ViewDefinition& view,
-                             const std::vector<NamespaceString>& refs) const;
+    Status _upsertIntoCatalog(OperationContext* opCtx,
+                              const CollectionPtr& systemViews,
+                              const ViewDefinition& view);
 
     /**
-     * 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. needsValidation can be set to false if the view already exists in the
-     * durable view catalog and skips checking that the resulting dependency graph is acyclic and
-     * within the maximum depth.
+     * Returns OK if each view namespace in 'refs' has the same default collation as the given view.
+     * Otherwise, returns ErrorCodes::OptionNotSupportedOnView.
      */
-    Status upsertIntoGraph(OperationContext* opCtx,
-                           const ViewDefinition& viewDef,
-                           const PipelineValidatorFn&,
-                           bool needsValidation);
+    Status _validateCollation(OperationContext* opCtx,
+                              const ViewDefinition& view,
+                              const std::vector<NamespaceString>& refs) const;
 
-private:
-    Status _insert(OperationContext* opCtx,
-                   const BSONObj& view,
-                   const boost::optional<TenantId>& tenantId);
+    StringMap<std::shared_ptr<ViewDefinition>> _viewMap;
+    ViewGraph _viewGraph;
+
+    bool _valid = false;
+    bool _viewGraphNeedsRefresh = true;
+
+    Stats _stats;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript
index efc59435bfd..c9c82af7422 100644
--- a/src/mongo/db/commands/SConscript
+++ b/src/mongo/db/commands/SConscript
@@ -379,7 +379,6 @@ env.Library(
         '$BUILD_DIR/mongo/db/timeseries/timeseries_stats',
         '$BUILD_DIR/mongo/db/transaction/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/rpc/rewrite_state_change_errors',
         '$BUILD_DIR/mongo/s/grid',
@@ -649,6 +648,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/auth/auth',
         '$BUILD_DIR/mongo/db/auth/authprivilege',
+        '$BUILD_DIR/mongo/db/catalog/collection_catalog',
         '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/query_expressions',
         '$BUILD_DIR/mongo/db/server_options_core',
diff --git a/src/mongo/db/curop.cpp b/src/mongo/db/curop.cpp
index a01bb33437f..0121a310aa4 100644
--- a/src/mongo/db/curop.cpp
+++ b/src/mongo/db/curop.cpp
@@ -393,6 +393,7 @@ static constexpr size_t appendMaxElementSize = 50 * 1024;
 
 bool CurOp::completeAndLogOperation(OperationContext* opCtx,
                                     logv2::LogComponent component,
+                                    std::shared_ptr<ProfileFilter> filter,
                                     boost::optional<size_t> responseLength,
                                     boost::optional<long long> slowMsOverride,
                                     bool forceLog) {
@@ -415,8 +416,7 @@ bool CurOp::completeAndLogOperation(OperationContext* opCtx,
 
     bool shouldLogSlowOp, shouldProfileAtLevel1;
 
-    if (auto filter =
-            CollectionCatalog::get(opCtx)->getDatabaseProfileSettings(getNSS().db()).filter) {
+    if (filter) {
         bool passesFilter = filter->matches(opCtx, _debug, *this);
 
         shouldLogSlowOp = passesFilter;
diff --git a/src/mongo/db/curop.h b/src/mongo/db/curop.h
index 29a63981b68..482145cbb88 100644
--- a/src/mongo/db/curop.h
+++ b/src/mongo/db/curop.h
@@ -424,6 +424,7 @@ public:
      */
     bool completeAndLogOperation(OperationContext* opCtx,
                                  logv2::LogComponent logComponent,
+                                 std::shared_ptr<ProfileFilter> filter,
                                  boost::optional<size_t> responseLength = boost::none,
                                  boost::optional<long long> slowMsOverride = boost::none,
                                  bool forceLog = false);
diff --git a/src/mongo/db/index_builds_coordinator_mongod.cpp b/src/mongo/db/index_builds_coordinator_mongod.cpp
index 183c13a1c32..2f2c3f2d108 100644
--- a/src/mongo/db/index_builds_coordinator_mongod.cpp
+++ b/src/mongo/db/index_builds_coordinator_mongod.cpp
@@ -425,7 +425,11 @@ IndexBuildsCoordinatorMongod::_startIndexBuild(OperationContext* opCtx,
             // Logs the index build statistics if it took longer than the server parameter `slowMs`
             // to complete.
             CurOp::get(opCtx.get())
-                ->completeAndLogOperation(opCtx.get(), MONGO_LOGV2_DEFAULT_COMPONENT);
+                ->completeAndLogOperation(opCtx.get(),
+                                          MONGO_LOGV2_DEFAULT_COMPONENT,
+                                          CollectionCatalog::get(opCtx.get())
+                                              ->getDatabaseProfileSettings(nss.dbName())
+                                              .filter);
         } catch (const DBException& e) {
             LOGV2(4656002, "unable to log operation", "error"_attr = e);
         }
diff --git a/src/mongo/db/op_observer/SConscript b/src/mongo/db/op_observer/SConscript
index 804864cf510..9fd80ff53a1 100644
--- a/src/mongo/db/op_observer/SConscript
+++ b/src/mongo/db/op_observer/SConscript
@@ -84,7 +84,8 @@ env.Library(
         '$BUILD_DIR/mongo/db/timeseries/bucket_catalog',
         '$BUILD_DIR/mongo/db/timeseries/timeseries_extended_range',
         '$BUILD_DIR/mongo/db/transaction/transaction',
-        '$BUILD_DIR/mongo/db/views/views_mongod',
+        '$BUILD_DIR/mongo/db/views/util',
+        '$BUILD_DIR/mongo/db/views/view_catalog_helpers',
         '$BUILD_DIR/mongo/s/coreshard',
         '$BUILD_DIR/mongo/s/grid',
         'op_observer',
diff --git a/src/mongo/db/op_observer/op_observer_impl.cpp b/src/mongo/db/op_observer/op_observer_impl.cpp
index 2e646f5a0da..672ef46bac6 100644
--- a/src/mongo/db/op_observer/op_observer_impl.cpp
+++ b/src/mongo/db/op_observer/op_observer_impl.cpp
@@ -73,7 +73,8 @@
 #include "mongo/db/timeseries/timeseries_extended_range.h"
 #include "mongo/db/transaction/transaction_participant.h"
 #include "mongo/db/transaction/transaction_participant_gen.h"
-#include "mongo/db/views/durable_view_catalog.h"
+#include "mongo/db/views/util.h"
+#include "mongo/db/views/view_catalog_helpers.h"
 #include "mongo/logv2/log.h"
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/grid.h"
@@ -667,15 +668,25 @@ void OpObserverImpl::onInserts(OperationContext* opCtx,
 
     if (nss.coll() == "system.js") {
         Scope::storedFuncMod(opCtx);
-    } else if (nss.coll() == DurableViewCatalog::viewsCollectionName()) {
+    } else if (nss.isSystemDotViews()) {
         try {
             for (auto it = first; it != last; it++) {
-                uassertStatusOK(DurableViewCatalog::onExternalInsert(opCtx, it->doc, nss));
+                view_util::validateViewDefinitionBSON(opCtx, it->doc, nss.dbName());
+
+                uassertStatusOK(CollectionCatalog::get(opCtx)->createView(
+                    opCtx,
+                    NamespaceStringUtil::deserialize(nss.dbName().tenantId(),
+                                                     it->doc.getStringField("_id")),
+                    {nss.dbName(), it->doc.getStringField("viewOn")},
+                    BSONArray{it->doc.getObjectField("pipeline")},
+                    view_catalog_helpers::validatePipeline,
+                    it->doc.getObjectField("collation"),
+                    ViewsForDatabase::Durability::kAlreadyDurable));
             }
         } catch (const DBException&) {
             // If a previous operation left the view catalog in an invalid state, our inserts can
             // fail even if all the definitions are valid. Reloading may help us reset the state.
-            DurableViewCatalog::onExternalChange(opCtx, nss);
+            CollectionCatalog::get(opCtx)->reloadViews(opCtx, nss.dbName()).ignore();
         }
     } else if (nss == NamespaceString::kSessionTransactionsTableNamespace && !lastOpTime.isNull()) {
         for (auto it = first; it != last; it++) {
@@ -919,8 +930,8 @@ void OpObserverImpl::onUpdate(OperationContext* opCtx, const OplogUpdateEntryArg
 
     if (args.nss.coll() == "system.js") {
         Scope::storedFuncMod(opCtx);
-    } else if (args.nss.coll() == DurableViewCatalog::viewsCollectionName()) {
-        DurableViewCatalog::onExternalChange(opCtx, args.nss);
+    } else if (args.nss.isSystemDotViews()) {
+        CollectionCatalog::get(opCtx)->reloadViews(opCtx, args.nss.dbName()).ignore();
     } else if (args.nss == NamespaceString::kSessionTransactionsTableNamespace &&
                !opTime.writeOpTime.isNull()) {
         auto mongoDSessionCatalog = MongoDSessionCatalog::get(opCtx);
@@ -1094,8 +1105,8 @@ void OpObserverImpl::onDelete(OperationContext* opCtx,
 
     if (nss.coll() == "system.js") {
         Scope::storedFuncMod(opCtx);
-    } else if (nss.coll() == DurableViewCatalog::viewsCollectionName()) {
-        DurableViewCatalog::onExternalChange(opCtx, nss);
+    } else if (nss.isSystemDotViews()) {
+        CollectionCatalog::get(opCtx)->reloadViews(opCtx, nss.dbName()).ignore();
     } else if (nss == NamespaceString::kSessionTransactionsTableNamespace &&
                !opTime.writeOpTime.isNull()) {
         auto mongoDSessionCatalog = MongoDSessionCatalog::get(opCtx);
@@ -1284,8 +1295,8 @@ repl::OpTime OpObserverImpl::onDropCollection(OperationContext* opCtx,
             "dropping the server configuration collection (admin.system.version) is not allowed.",
             collectionName != NamespaceString::kServerConfigurationNamespace);
 
-    if (collectionName.coll() == DurableViewCatalog::viewsCollectionName()) {
-        DurableViewCatalog::onSystemViewsCollectionDrop(opCtx, collectionName);
+    if (collectionName.isSystemDotViews()) {
+        CollectionCatalog::get(opCtx)->clearViews(opCtx, collectionName.dbName());
     } else if (collectionName == NamespaceString::kSessionTransactionsTableNamespace) {
         // Disallow this drop if there are currently prepared transactions.
         const auto sessionCatalog = SessionCatalog::get(opCtx);
@@ -1389,9 +1400,9 @@ void OpObserverImpl::postRenameCollection(OperationContext* const opCtx,
                                           const boost::optional<UUID>& dropTargetUUID,
                                           bool stayTemp) {
     if (fromCollection.isSystemDotViews())
-        DurableViewCatalog::onExternalChange(opCtx, fromCollection);
+        CollectionCatalog::get(opCtx)->reloadViews(opCtx, fromCollection.dbName()).ignore();
     if (toCollection.isSystemDotViews())
-        DurableViewCatalog::onExternalChange(opCtx, toCollection);
+        CollectionCatalog::get(opCtx)->reloadViews(opCtx, toCollection.dbName()).ignore();
 }
 
 void OpObserverImpl::onRenameCollection(OperationContext* const opCtx,
diff --git a/src/mongo/db/ops/insert.cpp b/src/mongo/db/ops/insert.cpp
index e3fbdaf617b..41a228f72c0 100644
--- a/src/mongo/db/ops/insert.cpp
+++ b/src/mongo/db/ops/insert.cpp
@@ -38,7 +38,6 @@
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/update/storage_validation.h"
 #include "mongo/db/vector_clock_mutable.h"
-#include "mongo/db/views/durable_view_catalog.h"
 #include "mongo/util/fail_point.h"
 #include "mongo/util/str.h"
 
diff --git a/src/mongo/db/ops/write_ops_exec.cpp b/src/mongo/db/ops/write_ops_exec.cpp
index 0a38119164d..1fbd221c707 100644
--- a/src/mongo/db/ops/write_ops_exec.cpp
+++ b/src/mongo/db/ops/write_ops_exec.cpp
@@ -144,8 +144,12 @@ void finishCurOp(OperationContext* opCtx, CurOp* curOp) {
 
         // Mark the op as complete, and log it if appropriate. Returns a boolean indicating whether
         // this op should be sampled for profiling.
-        const bool shouldProfile =
-            curOp->completeAndLogOperation(opCtx, MONGO_LOGV2_DEFAULT_COMPONENT);
+        const bool shouldProfile = curOp->completeAndLogOperation(
+            opCtx,
+            MONGO_LOGV2_DEFAULT_COMPONENT,
+            CollectionCatalog::get(opCtx)
+                ->getDatabaseProfileSettings(curOp->getNSS().dbName())
+                .filter);
 
         if (shouldProfile) {
             // Stash the current transaction so that writes to the profile collection are not
diff --git a/src/mongo/db/query/SConscript b/src/mongo/db/query/SConscript
index bf2849d5e4e..5b8b1b4093c 100644
--- a/src/mongo/db/query/SConscript
+++ b/src/mongo/db/query/SConscript
@@ -238,7 +238,6 @@ env.Library(
         '$BUILD_DIR/mongo/crypto/fle_fields',
         '$BUILD_DIR/mongo/db/api_parameters',
         '$BUILD_DIR/mongo/db/auth/authprivilege',
-        '$BUILD_DIR/mongo/db/catalog/collection_catalog',
         '$BUILD_DIR/mongo/db/commands/test_commands_enabled',
         '$BUILD_DIR/mongo/db/pipeline/runtime_constants_idl',
         '$BUILD_DIR/mongo/db/repl/read_concern_args',
diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index 0c8bb4df12e..55616a1c826 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -1084,6 +1084,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/catalog/collection',
+        '$BUILD_DIR/mongo/db/catalog/collection_catalog',
         '$BUILD_DIR/mongo/db/catalog/document_validation',
         '$BUILD_DIR/mongo/db/commands/list_collections_filter',
         '$BUILD_DIR/mongo/db/ops/write_ops_exec',
diff --git a/src/mongo/db/service_entry_point_common.cpp b/src/mongo/db/service_entry_point_common.cpp
index 689d3f067fe..24958c3d2f8 100644
--- a/src/mongo/db/service_entry_point_common.cpp
+++ b/src/mongo/db/service_entry_point_common.cpp
@@ -2287,11 +2287,15 @@ void HandleRequest::completeOperation(DbResponse& response) {
 
     // Mark the op as complete, and log it if appropriate. Returns a boolean indicating whether
     // this op should be written to the profiler.
-    const bool shouldProfile = currentOp.completeAndLogOperation(opCtx,
-                                                                 MONGO_LOGV2_DEFAULT_COMPONENT,
-                                                                 response.response.size(),
-                                                                 executionContext->slowMsOverride,
-                                                                 executionContext->forceLog);
+    const bool shouldProfile = currentOp.completeAndLogOperation(
+        opCtx,
+        MONGO_LOGV2_DEFAULT_COMPONENT,
+        CollectionCatalog::get(opCtx)
+            ->getDatabaseProfileSettings(currentOp.getNSS().dbName())
+            .filter,
+        response.response.size(),
+        executionContext->slowMsOverride,
+        executionContext->forceLog);
 
     Top::get(opCtx->getServiceContext())
         .incrementGlobalLatencyStats(
diff --git a/src/mongo/db/views/SConscript b/src/mongo/db/views/SConscript
index a3f5227f5fc..bee9cc79c65 100644
--- a/src/mongo/db/views/SConscript
+++ b/src/mongo/db/views/SConscript
@@ -5,25 +5,6 @@ Import("env")
 env = env.Clone()
 
 env.Library(
-    target='views_mongod',
-    source=[
-        'durable_view_catalog.cpp',
-    ],
-    LIBDEPS=[
-        '$BUILD_DIR/mongo/db/dbhelpers',
-        '$BUILD_DIR/mongo/db/views/views',
-    ],
-    LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/db/audit',
-        '$BUILD_DIR/mongo/db/catalog/collection_crud',
-        '$BUILD_DIR/mongo/db/catalog/database_holder',
-        '$BUILD_DIR/mongo/db/multitenancy',
-        '$BUILD_DIR/mongo/db/server_feature_flags',
-        '$BUILD_DIR/mongo/db/views/view_catalog_helpers',
-    ],
-)
-
-env.Library(
     target='views',
     source=[
         'view.cpp',
@@ -41,6 +22,7 @@ env.Library(
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/catalog/collection',
+        '$BUILD_DIR/mongo/db/catalog/collection_catalog',
         '$BUILD_DIR/mongo/db/pipeline/aggregation',
         'resolved_view',
         'views',
@@ -62,6 +44,17 @@ env.Library(
     ],
 )
 
+env.Library(
+    target='util',
+    source=[
+        'util.cpp',
+    ],
+    LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/util/namespace_string_util',
+        'views',
+    ],
+)
+
 env.CppUnitTest(
     target='db_views_test',
     source=[
@@ -77,10 +70,10 @@ env.CppUnitTest(
         '$BUILD_DIR/mongo/db/query/collation/collator_interface_mock',
         '$BUILD_DIR/mongo/db/query/query_test_service_context',
         '$BUILD_DIR/mongo/db/repl/replmocks',
+        '$BUILD_DIR/mongo/db/shard_role',
         '$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
deleted file mode 100644
index d18d35f894f..00000000000
--- a/src/mongo/db/views/durable_view_catalog.cpp
+++ /dev/null
@@ -1,293 +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.
- */
-
-#include "mongo/db/views/durable_view_catalog.h"
-
-#include <string>
-
-#include "mongo/db/audit.h"
-#include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/collection_write_path.h"
-#include "mongo/db/catalog/database.h"
-#include "mongo/db/catalog/database_holder.h"
-#include "mongo/db/concurrency/d_concurrency.h"
-#include "mongo/db/curop.h"
-#include "mongo/db/database_name.h"
-#include "mongo/db/dbhelpers.h"
-#include "mongo/db/multitenancy_gen.h"
-#include "mongo/db/namespace_string.h"
-#include "mongo/db/operation_context.h"
-#include "mongo/db/server_feature_flags_gen.h"
-#include "mongo/db/storage/record_data.h"
-#include "mongo/db/views/view_catalog_helpers.h"
-#include "mongo/logv2/log.h"
-#include "mongo/stdx/unordered_set.h"
-#include "mongo/util/assert_util.h"
-#include "mongo/util/string_map.h"
-
-#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kStorage
-
-namespace mongo {
-
-namespace {
-void validateViewDefinitionBSON(OperationContext* opCtx,
-                                const BSONObj& viewDefinition,
-                                const DatabaseName& dbName) {
-    // Internal callers should always pass in a valid 'dbName' against which to compare the
-    // 'viewDefinition'.
-    invariant(NamespaceString::validDBName(dbName));
-
-    bool valid = true;
-
-    for (const BSONElement& e : viewDefinition) {
-        std::string name(e.fieldName());
-        valid &= name == "_id" || name == "viewOn" || name == "pipeline" || name == "collation" ||
-            name == "timeseries";
-    }
-
-    NamespaceString viewName =
-        NamespaceStringUtil::deserialize(dbName.tenantId(), viewDefinition["_id"].str());
-
-    const auto viewNameIsValid = NamespaceString::validCollectionComponent(viewName.ns()) &&
-        NamespaceString::validDBName(viewName.dbName());
-    valid &= viewNameIsValid;
-
-    // Only perform validation via NamespaceString if the collection name has been determined to
-    // be valid. If not valid then the NamespaceString constructor will uassert.
-    if (viewNameIsValid) {
-        NamespaceString viewNss(viewName);
-        valid &= viewNss.isValid() && viewNss.dbName() == dbName;
-    }
-
-    valid &= NamespaceString::validCollectionName(viewDefinition["viewOn"].str());
-
-    const bool hasPipeline = viewDefinition.hasField("pipeline");
-    valid &= hasPipeline;
-    if (hasPipeline) {
-        valid &= viewDefinition["pipeline"].type() == mongo::Array;
-    }
-
-    valid &= (!viewDefinition.hasField("collation") ||
-              viewDefinition["collation"].type() == BSONType::Object);
-
-    valid &= !viewDefinition.hasField("timeseries") ||
-        viewDefinition["timeseries"].type() == BSONType::Object;
-
-    uassert(ErrorCodes::InvalidViewDefinition,
-            str::stream() << "found invalid view definition " << viewDefinition["_id"]
-                          << " while reading '"
-                          << NamespaceString(dbName, NamespaceString::kSystemDotViewsCollectionName)
-                          << "'",
-            valid);
-}
-}  // namespace
-
-// DurableViewCatalog
-
-void DurableViewCatalog::onExternalChange(OperationContext* opCtx, const NamespaceString& name) {
-    dassert(opCtx->lockState()->isDbLockedForMode(name.dbName(), MODE_IX));
-    dassert(opCtx->lockState()->isCollectionLockedForMode(
-        NamespaceString(name.dbName(), 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 views for this database until the
-    // invalid view definitions are removed.
-    auto catalog = CollectionCatalog::get(opCtx);
-    catalog->reloadViews(opCtx, name.dbName()).ignore();
-}
-
-Status DurableViewCatalog::onExternalInsert(OperationContext* opCtx,
-                                            const BSONObj& doc,
-                                            const NamespaceString& name) {
-    try {
-        validateViewDefinitionBSON(opCtx, doc, name.dbName());
-    } catch (const DBException& e) {
-        return e.toStatus();
-    }
-
-    auto catalog = CollectionCatalog::get(opCtx);
-
-    NamespaceString viewName =
-        NamespaceStringUtil::deserialize(name.tenantId(), doc.getStringField("_id"));
-    NamespaceString viewOn(name.dbName(), doc.getStringField("viewOn"));
-    BSONArray pipeline(doc.getObjectField("pipeline"));
-    BSONObj collation(doc.getObjectField("collation"));
-
-    return catalog->createView(opCtx,
-                               viewName,
-                               viewOn,
-                               pipeline,
-                               collation,
-                               view_catalog_helpers::validatePipeline,
-                               CollectionCatalog::ViewUpsertMode::kAlreadyDurableView);
-}
-
-void DurableViewCatalog::onSystemViewsCollectionDrop(OperationContext* opCtx,
-                                                     const NamespaceString& name) {
-    dassert(opCtx->lockState()->isDbLockedForMode(name.dbName(), MODE_IX));
-    dassert(opCtx->lockState()->isCollectionLockedForMode(
-        NamespaceString(name.dbName(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
-    dassert(name.coll() == NamespaceString::kSystemDotViewsCollectionName);
-
-    auto catalog = CollectionCatalog::get(opCtx);
-    const DatabaseName& dbName = name.dbName();
-
-    // First, iterate through the views on this database and audit them before they are dropped.
-    catalog->iterateViews(opCtx,
-                          dbName,
-                          [&](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, dbName);
-}
-
-// DurableViewCatalogImpl
-
-const DatabaseName& DurableViewCatalogImpl::getName() const {
-    return _db->name();
-}
-
-void DurableViewCatalogImpl::iterate(OperationContext* opCtx, Callback callback) {
-    _iterate(opCtx, callback, ViewCatalogLookupBehavior::kValidateViews);
-}
-
-void DurableViewCatalogImpl::iterateIgnoreInvalidEntries(OperationContext* opCtx,
-                                                         Callback callback) {
-    _iterate(opCtx, callback, ViewCatalogLookupBehavior::kAllowInvalidViews);
-}
-
-void DurableViewCatalogImpl::_iterate(OperationContext* opCtx,
-                                      Callback callback,
-                                      ViewCatalogLookupBehavior lookupBehavior) {
-    invariant(opCtx->lockState()->isCollectionLockedForMode(_db->getSystemViewsName(), MODE_IS));
-
-    CollectionPtr systemViews = CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(
-        opCtx, _db->getSystemViewsName());
-    if (!systemViews) {
-        return;
-    }
-
-    auto cursor = systemViews->getCursor(opCtx);
-    while (auto record = cursor->next()) {
-        BSONObj viewDefinition;
-        try {
-            viewDefinition = _validateViewDefinition(opCtx, record->data);
-            uassertStatusOK(callback(viewDefinition));
-        } catch (const ExceptionFor<ErrorCodes::InvalidViewDefinition>&) {
-            if (lookupBehavior == ViewCatalogLookupBehavior::kValidateViews) {
-                throw;
-            }
-        }
-    }
-}
-
-BSONObj DurableViewCatalogImpl::_validateViewDefinition(OperationContext* opCtx,
-                                                        const RecordData& recordData) {
-    // Check the document is valid BSON, with only the expected fields.
-    // Use the latest BSON validation version. Existing view definitions are allowed to contain
-    // decimal data even if decimal is disabled.
-    fassert(40224, validateBSON(recordData.data(), recordData.size()));
-    BSONObj viewDefinition = recordData.toBson();
-
-    validateViewDefinitionBSON(opCtx, viewDefinition, _db->name());
-    return viewDefinition;
-}
-
-void DurableViewCatalogImpl::upsert(OperationContext* opCtx,
-                                    const NamespaceString& name,
-                                    const BSONObj& view) {
-    dassert(opCtx->lockState()->isDbLockedForMode(_db->name(), MODE_IX));
-    dassert(opCtx->lockState()->isCollectionLockedForMode(name, MODE_IX));
-
-    NamespaceString systemViewsNs(_db->getSystemViewsName());
-    dassert(opCtx->lockState()->isCollectionLockedForMode(systemViewsNs, MODE_X));
-
-    const CollectionPtr& systemViews =
-        CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, systemViewsNs);
-    invariant(systemViews);
-
-    std::string nssOnDisk = NamespaceStringUtil::serialize(name);
-
-    RecordId id = Helpers::findOne(opCtx, systemViews, BSON("_id" << nssOnDisk));
-
-    Snapshotted<BSONObj> oldView;
-    if (!id.isValid() || !systemViews->findDoc(opCtx, id, &oldView)) {
-        LOGV2_DEBUG(22544,
-                    2,
-                    "Insert view to system views catalog",
-                    "view"_attr = view,
-                    "viewCatalog"_attr = _db->getSystemViewsName());
-        uassertStatusOK(collection_internal::insertDocument(
-            opCtx, systemViews, InsertStatement(view), &CurOp::get(opCtx)->debug()));
-    } else {
-        CollectionUpdateArgs args;
-        args.update = view;
-        args.criteria = BSON("_id" << nssOnDisk);
-
-        const bool assumeIndexesAreAffected = true;
-        systemViews->updateDocument(
-            opCtx, id, oldView, view, assumeIndexesAreAffected, &CurOp::get(opCtx)->debug(), &args);
-    }
-}
-
-void DurableViewCatalogImpl::remove(OperationContext* opCtx, const NamespaceString& name) {
-    dassert(opCtx->lockState()->isDbLockedForMode(_db->name(), MODE_IX));
-    dassert(opCtx->lockState()->isCollectionLockedForMode(name, MODE_IX));
-
-    CollectionPtr systemViews = CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(
-        opCtx, _db->getSystemViewsName());
-    dassert(opCtx->lockState()->isCollectionLockedForMode(systemViews->ns(), MODE_X));
-
-    if (!systemViews)
-        return;
-
-
-    std::string nssOnDisk = NamespaceStringUtil::serialize(name);
-
-    RecordId id = Helpers::findOne(opCtx, systemViews, BSON("_id" << nssOnDisk));
-    if (!id.isValid())
-        return;
-
-    LOGV2_DEBUG(22545,
-                2,
-                "Remove view from system views catalog",
-                "view"_attr = name,
-                "viewCatalog"_attr = _db->getSystemViewsName());
-    systemViews->deleteDocument(opCtx, kUninitializedStmtId, id, &CurOp::get(opCtx)->debug());
-}
-}  // namespace mongo
diff --git a/src/mongo/db/views/durable_view_catalog.h b/src/mongo/db/views/durable_view_catalog.h
deleted file mode 100644
index 596d3e18a32..00000000000
--- a/src/mongo/db/views/durable_view_catalog.h
+++ /dev/null
@@ -1,120 +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 <string>
-
-#include "mongo/base/status.h"
-#include "mongo/base/string_data.h"
-#include "mongo/db/namespace_string.h"
-
-namespace mongo {
-
-class BSONObj;
-class Database;
-class OperationContext;
-class RecordData;
-
-/**
- * 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 { kValidateViews, kAllowInvalidViews };
-
-/**
- * Interface for system.views collection operations associated with view catalog management.
- * All methods must be called from within a WriteUnitOfWork, and with the DBLock held in at
- * least intent mode.
- */
-class DurableViewCatalog {
-public:
-    static constexpr StringData viewsCollectionName() {
-        return NamespaceString::kSystemDotViewsCollectionName;
-    }
-
-    /**
-     * Thread-safe method to mark a catalog name was changed. This will cause the in-memory
-     * view catalog to be reloaded immediately.
-     */
-    static void onExternalChange(OperationContext* opCtx, const NamespaceString& name);
-
-    /**
-     * Thread-safe method to insert into the in-memory view catalog when there is an insert to
-     * 'system.views'.
-     */
-    static Status onExternalInsert(OperationContext* opCtx,
-                                   const BSONObj& doc,
-                                   const NamespaceString& name);
-
-    /**
-     * Thread-safe method to clear the in-memory state of the view catalog when the 'system.views'
-     * collection is dropped.
-     */
-    static void onSystemViewsCollectionDrop(OperationContext* opCtx, const NamespaceString& name);
-
-    using Callback = std::function<Status(const BSONObj& view)>;
-    virtual void iterate(OperationContext* opCtx, Callback callback) = 0;
-    virtual void iterateIgnoreInvalidEntries(OperationContext* opCtx, Callback callback) = 0;
-    virtual void upsert(OperationContext* opCtx,
-                        const NamespaceString& name,
-                        const BSONObj& view) = 0;
-    virtual void remove(OperationContext* opCtx, const NamespaceString& name) = 0;
-    virtual const DatabaseName& getName() const = 0;
-    virtual ~DurableViewCatalog() = default;
-};
-
-/**
- * Actual implementation of DurableViewCatalog for use by the Database class.
- * Implements durability through database operations on the system.views collection.
- */
-class DurableViewCatalogImpl final : public DurableViewCatalog {
-public:
-    explicit DurableViewCatalogImpl(Database* db) : _db(db) {}
-
-    void iterate(OperationContext* opCtx, Callback callback);
-
-    void iterateIgnoreInvalidEntries(OperationContext* opCtx, Callback callback);
-
-    void upsert(OperationContext* opCtx, const NamespaceString& name, const BSONObj& view);
-    void remove(OperationContext* opCtx, const NamespaceString& name);
-    const DatabaseName& getName() const;
-
-private:
-    void _iterate(OperationContext* opCtx,
-                  Callback callback,
-                  ViewCatalogLookupBehavior lookupBehavior);
-
-    BSONObj _validateViewDefinition(OperationContext* opCtx, const RecordData& recordData);
-
-    Database* const _db;
-};
-}  // namespace mongo
diff --git a/src/mongo/db/views/util.cpp b/src/mongo/db/views/util.cpp
new file mode 100644
index 00000000000..828659df2d7
--- /dev/null
+++ b/src/mongo/db/views/util.cpp
@@ -0,0 +1,90 @@
+/**
+ *    Copyright (C) 2022-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/util.h"
+
+#include "mongo/db/multitenancy_gen.h"
+#include "mongo/db/server_feature_flags_gen.h"
+
+namespace mongo::view_util {
+void validateViewDefinitionBSON(OperationContext* opCtx,
+                                const BSONObj& viewDefinition,
+                                const DatabaseName& dbName) {
+    // Internal callers should always pass in a valid 'dbName' against which to compare the
+    // 'viewDefinition'.
+    invariant(NamespaceString::validDBName(dbName));
+
+    bool valid = true;
+
+    for (auto&& [name, value] : viewDefinition) {
+        valid &= name == "_id" || name == "viewOn" || name == "pipeline" || name == "collation" ||
+            name == "timeseries";
+    }
+
+    auto viewNameElem = viewDefinition["_id"];
+    valid &= viewNameElem && viewNameElem.type() == BSONType::String;
+
+    auto viewName = NamespaceStringUtil::deserialize(dbName.tenantId(), viewNameElem.str());
+
+    bool viewNameIsValid = NamespaceString::validCollectionComponent(viewName.ns()) &&
+        NamespaceString::validDBName(viewName.dbName());
+    valid &= viewNameIsValid;
+
+    // Only perform validation via NamespaceString if the collection name has been determined to
+    // be valid. If not valid then the NamespaceString constructor will uassert.
+    if (viewNameIsValid) {
+        valid &= viewName.isValid() && viewName.dbName() == dbName;
+    }
+
+    valid &= NamespaceString::validCollectionName(viewNameElem.str());
+
+    auto viewOn = viewDefinition["viewOn"];
+    valid &= viewOn && viewOn.type() == BSONType::String;
+
+    if (auto pipeline = viewDefinition["pipeline"]) {
+        valid &= pipeline.type() == BSONType::Array;
+        for (auto&& stage : pipeline.Obj()) {
+            valid &= stage.type() == BSONType::Object;
+        }
+    }
+
+    auto collation = viewDefinition["collation"];
+    valid &= !collation || collation.type() == BSONType::Object;
+
+    auto timeseries = viewDefinition["timeseries"];
+    valid &= !timeseries || timeseries.type() == BSONType::Object;
+
+    uassert(ErrorCodes::InvalidViewDefinition,
+            str::stream() << "found invalid view definition " << viewDefinition["_id"]
+                          << " while reading '"
+                          << NamespaceString(dbName, NamespaceString::kSystemDotViewsCollectionName)
+                          << "'",
+            valid);
+}
+}  // namespace mongo::view_util
diff --git a/src/mongo/db/views/util.h b/src/mongo/db/views/util.h
new file mode 100644
index 00000000000..9c3cfcc1889
--- /dev/null
+++ b/src/mongo/db/views/util.h
@@ -0,0 +1,42 @@
+/**
+ *    Copyright (C) 2022-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 "mongo/db/operation_context.h"
+#include "mongo/db/views/view.h"
+
+namespace mongo::view_util {
+/**
+ * Throws if the specified document is not a valid view definition.
+ */
+void validateViewDefinitionBSON(OperationContext* opCtx,
+                                const BSONObj& viewDefinition,
+                                const DatabaseName& dbName);
+}  // namespace mongo::view_util
diff --git a/src/mongo/db/views/view_catalog_test.cpp b/src/mongo/db/views/view_catalog_test.cpp
index 837828ce218..ca44c8ff6cb 100644
--- a/src/mongo/db/views/view_catalog_test.cpp
+++ b/src/mongo/db/views/view_catalog_test.cpp
@@ -51,7 +51,6 @@
 #include "mongo/db/repl/storage_interface_mock.h"
 #include "mongo/db/server_options.h"
 #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_helpers.h"
 #include "mongo/db/views/view_graph.h"
@@ -86,26 +85,9 @@ public:
     void setUp() override {
         CatalogTestFixture::setUp();
 
-        WriteUnitOfWork wuow(operationContext());
-        AutoGetDb autoDb(operationContext(), _dbName, MODE_X);
-        _db = autoDb.ensureDbExists(operationContext());
-        invariant(_db);
-
-        // Create any additional databases used throughout the test.
-        ASSERT(AutoGetDb(operationContext(), DatabaseName(_dbName.tenantId(), "db1"), MODE_X)
-                   .ensureDbExists(operationContext()));
-        ASSERT(AutoGetDb(operationContext(), DatabaseName(_dbName.tenantId(), "db2"), MODE_X)
-                   .ensureDbExists(operationContext()));
-
-        auto durableViewCatalogUnique = std::make_unique<DurableViewCatalogImpl>(_db);
-        durableViewCatalog = durableViewCatalogUnique.get();
-
-        // Create the system views collection for the database.
-        ASSERT(_db->createCollection(
-            operationContext(),
-            NamespaceString(_dbName, NamespaceString::kSystemDotViewsCollectionName)));
-
-        wuow.commit();
+        _db = _createDatabase(_dbName);
+        _createDatabase({_dbName.tenantId(), "db1"});
+        _createDatabase({_dbName.tenantId(), "db2"});
     }
 
     void tearDown() override {
@@ -129,8 +111,8 @@ public:
             MODE_X);
 
         WriteUnitOfWork wuow(opCtx);
-        Status s = getCatalog()->createView(
-            opCtx, viewName, viewOn, pipeline, collation, view_catalog_helpers::validatePipeline);
+        auto s = getCatalog()->createView(
+            opCtx, viewName, viewOn, pipeline, view_catalog_helpers::validatePipeline, collation);
         wuow.commit();
 
         return s;
@@ -148,7 +130,7 @@ public:
             MODE_X);
 
         WriteUnitOfWork wuow(opCtx);
-        Status s = getCatalog()->modifyView(
+        auto s = getCatalog()->modifyView(
             opCtx, viewName, viewOn, pipeline, view_catalog_helpers::validatePipeline);
         wuow.commit();
 
@@ -184,8 +166,19 @@ private:
     DatabaseName _dbName;
     Database* _db;
 
+    Database* _createDatabase(const DatabaseName& dbName) {
+        WriteUnitOfWork wuow{operationContext()};
+
+        NamespaceString ns{dbName, NamespaceString::kSystemDotViewsCollectionName};
+        AutoGetCollection systemViews{operationContext(), ns, MODE_X};
+        auto db = systemViews.ensureDbExists(operationContext());
+        ASSERT(db->createCollection(operationContext(), ns));
+
+        wuow.commit();
+        return db;
+    }
+
 protected:
-    DurableViewCatalogImpl* durableViewCatalog;
     const BSONArray emptyPipeline;
     const BSONObj emptyCollation;
 };
@@ -545,8 +538,8 @@ TEST_F(ViewCatalogFixture, LookupRIDExistingViewRollback) {
                                            viewName,
                                            viewOn,
                                            emptyPipeline,
-                                           emptyCollation,
-                                           view_catalog_helpers::validatePipeline));
+                                           view_catalog_helpers::validatePipeline,
+                                           emptyCollation));
     }
     auto resourceID = ResourceId(RESOURCE_COLLECTION, NamespaceString(boost::none, "db.view"));
     ASSERT(!ResourceCatalog::get(getServiceContext()).name(resourceID));
diff --git a/src/mongo/s/SConscript b/src/mongo/s/SConscript
index af4f719fb53..0a53d677ade 100644
--- a/src/mongo/s/SConscript
+++ b/src/mongo/s/SConscript
@@ -421,6 +421,7 @@ env.Library(
         'sharding_router_api',
     ],
     LIBDEPS=[
+        '$BUILD_DIR/mongo/db/catalog/collection_catalog',
         '$BUILD_DIR/mongo/transport/service_entry_point',
     ],
 )
diff --git a/src/mongo/s/service_entry_point_mongos.cpp b/src/mongo/s/service_entry_point_mongos.cpp
index 3d4d66a9ea2..47d6054c270 100644
--- a/src/mongo/s/service_entry_point_mongos.cpp
+++ b/src/mongo/s/service_entry_point_mongos.cpp
@@ -169,7 +169,13 @@ void HandleRequest::onSuccess(const DbResponse& dbResponse) {
 
     // Mark the op as complete, populate the response length, and log it if appropriate.
     currentOp->completeAndLogOperation(
-        opCtx, logv2::LogComponent::kCommand, dbResponse.response.size(), slowMsOverride);
+        opCtx,
+        logv2::LogComponent::kCommand,
+        CollectionCatalog::get(opCtx)
+            ->getDatabaseProfileSettings(currentOp->getNSS().dbName())
+            .filter,
+        dbResponse.response.size(),
+        slowMsOverride);
 
     // Update the source of stats shown in the db.serverStatus().opLatencies section.
     Top::get(opCtx->getServiceContext())