SERVER-24823 Add admin commands, durable system.views catalog

Consolidate disallowed commands in views_all_commands.js testing.

17 files changed, 424 insertions, 47 deletions

diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index 653b91f902c..f561bc79d53 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -711,7 +711,7 @@ serveronlyLibdeps = [
     "storage/storage_engine_metadata",
     "storage/storage_options",
     "update_index_data",
-    "views/views",
+    "views/views_mongod",
 ]
 
 if wiredtiger:
diff --git a/src/mongo/db/catalog/coll_mod.cpp b/src/mongo/db/catalog/coll_mod.cpp
index dde29faa503..75769b5c752 100644
--- a/src/mongo/db/catalog/coll_mod.cpp
+++ b/src/mongo/db/catalog/coll_mod.cpp
@@ -30,6 +30,8 @@
 
 #include "mongo/db/catalog/coll_mod.h"
 
+#include <boost/optional.hpp>
+
 #include "mongo/db/background.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_catalog_entry.h"
@@ -39,6 +41,8 @@
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/repl/replication_coordinator_global.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/storage/recovery_unit.h"
+#include "mongo/db/views/view_catalog.h"
 
 namespace mongo {
 Status collMod(OperationContext* txn,
@@ -49,14 +53,20 @@ Status collMod(OperationContext* txn,
     ScopedTransaction transaction(txn, MODE_IX);
     AutoGetDb autoDb(txn, dbName, MODE_X);
     Database* const db = autoDb.getDb();
-    Collection* coll = db ? db->getCollection(nss) : NULL;
+    Collection* coll = db ? db->getCollection(nss) : nullptr;
+
+    // May also modify a view instead of a collection.
+    const ViewDefinition* view = db ? db->getViewCatalog()->lookup(nss.ns()) : nullptr;
+    boost::optional<ViewDefinition> newView;
+    if (view)
+        newView = {*view};
 
     // This can kill all cursors so don't allow running it while a background operation is in
     // progress.
     BackgroundOperation::assertNoBgOpInProgForNs(nss);
 
-    // If db/collection does not exist, short circuit and return.
-    if (!db || !coll) {
+    // If db/collection/view does not exist, short circuit and return.
+    if (!db || (!coll && !view)) {
         return Status(ErrorCodes::NamespaceNotFound, "ns does not exist");
     }
 
@@ -75,6 +85,7 @@ Status collMod(OperationContext* txn,
 
     Status errorStatus = Status::OK();
 
+    // TODO(SERVER-25004): Separate parsing and catalog modification
     BSONForEach(e, cmdObj) {
         if (str::equals("collMod", e.fieldName())) {
             // no-op
@@ -83,6 +94,11 @@ Status collMod(OperationContext* txn,
         } else if (QueryRequest::cmdOptionMaxTimeMS == e.fieldNameStringData()) {
             // no-op
         } else if (str::equals("index", e.fieldName())) {
+            if (view) {
+                errorStatus = Status(ErrorCodes::InvalidOptions, "cannot modify indexes on a view");
+                continue;
+            }
+
             BSONObj indexObj = e.Obj();
             BSONObj keyPattern = indexObj.getObjectField("keyPattern");
 
@@ -132,17 +148,60 @@ Status collMod(OperationContext* txn,
                 result->appendAs(newExpireSecs, "expireAfterSeconds_new");
             }
         } else if (str::equals("validator", e.fieldName())) {
+            if (view) {
+                errorStatus = Status(ErrorCodes::InvalidOptions,
+                                     "cannot modify validation options on a view");
+                continue;
+            }
+
             auto status = coll->setValidator(txn, e.Obj());
             if (!status.isOK())
                 errorStatus = std::move(status);
         } else if (str::equals("validationLevel", e.fieldName())) {
+            if (view) {
+                errorStatus = Status(ErrorCodes::InvalidOptions,
+                                     "cannot modify validation options on a view");
+                continue;
+            }
+
             auto status = coll->setValidationLevel(txn, e.String());
             if (!status.isOK())
                 errorStatus = std::move(status);
         } else if (str::equals("validationAction", e.fieldName())) {
+            if (view) {
+                errorStatus = Status(ErrorCodes::InvalidOptions,
+                                     "cannot modify validation options on a view");
+                continue;
+            }
+
             auto status = coll->setValidationAction(txn, e.String());
             if (!status.isOK())
                 errorStatus = std::move(status);
+        } else if (str::equals("pipeline", e.fieldName())) {
+            if (!view) {
+                errorStatus = Status(ErrorCodes::InvalidOptions,
+                                     "'pipeline' option only supported on a view");
+                continue;
+            }
+            if (!e.isABSONObj()) {
+                errorStatus =
+                    Status(ErrorCodes::InvalidOptions, "not a valid aggregation pipeline");
+                continue;
+            }
+            newView->setPipeline(e);
+        } else if (str::equals("viewOn", e.fieldName())) {
+            if (!view) {
+                errorStatus =
+                    Status(ErrorCodes::InvalidOptions, "'viewOn' option only supported on a view");
+                continue;
+            }
+            if (e.type() != mongo::String) {
+                errorStatus =
+                    Status(ErrorCodes::InvalidOptions, "'viewOn' option must be a string");
+                continue;
+            }
+            NamespaceString nss(dbName, e.str());
+            newView->setViewOn(NamespaceString(dbName, e.str()));
         } else {
             // As of SERVER-17312 we only support these two options. When SERVER-17320 is
             // resolved this will need to be enhanced to handle other options.
@@ -157,6 +216,12 @@ Status collMod(OperationContext* txn,
                 continue;
             }
 
+            if (view) {
+                errorStatus = Status(ErrorCodes::InvalidOptions,
+                                     str::stream() << "option not supported on a view: " << name);
+                continue;
+            }
+
             CollectionCatalogEntry* cce = coll->getCatalogEntry();
 
             const int oldFlags = cce->getCollectionOptions(txn).flags;
@@ -179,6 +244,18 @@ Status collMod(OperationContext* txn,
         }
     }
 
+    // Actually update the view if it was parsed successfully.
+    if (view && errorStatus.isOK()) {
+        ViewCatalog* catalog = db->getViewCatalog();
+        catalog->dropView(txn, nss);
+
+        BSONArrayBuilder pipeline;
+        for (auto& item : newView->pipeline()) {
+            pipeline.append(item);
+        }
+        errorStatus = catalog->createView(txn, nss, newView->viewOn(), pipeline.obj());
+    }
+
     if (!errorStatus.isOK()) {
         return errorStatus;
     }
diff --git a/src/mongo/db/catalog/database.cpp b/src/mongo/db/catalog/database.cpp
index ab3ac9c7a6d..75fa8db90bf 100644
--- a/src/mongo/db/catalog/database.cpp
+++ b/src/mongo/db/catalog/database.cpp
@@ -204,7 +204,8 @@ Database::Database(OperationContext* txn, StringData name, DatabaseCatalogEntry*
       _profileName(_name + ".system.profile"),
       _indexesName(_name + ".system.indexes"),
       _viewsName(_name + ".system.views"),
-      _views(txn, this) {
+      _durableViews(DurableViewCatalogImpl(this)),
+      _views(txn, &_durableViews) {
     Status status = validateDBName(_name);
     if (!status.isOK()) {
         warning() << "tried to open invalid db: " << _name << endl;
@@ -349,6 +350,10 @@ void Database::getStats(OperationContext* opCtx, BSONObjBuilder* output, double
     _dbEntry->appendExtraStats(opCtx, output, scale);
 }
 
+void Database::dropView(OperationContext* txn, StringData fullns) {
+    _views.dropView(txn, NamespaceString(fullns));
+}
+
 Status Database::dropCollection(OperationContext* txn, StringData fullns) {
     invariant(txn->lockState()->isDbLockedForMode(name(), MODE_X));
 
diff --git a/src/mongo/db/catalog/database.h b/src/mongo/db/catalog/database.h
index 5f27d0f82dc..b04584b7361 100644
--- a/src/mongo/db/catalog/database.h
+++ b/src/mongo/db/catalog/database.h
@@ -151,6 +151,8 @@ public:
 
     Status dropCollection(OperationContext* txn, StringData fullns);
 
+    void dropView(OperationContext* txn, StringData fullns);
+
     Collection* createCollection(OperationContext* txn,
                                  StringData ns,
                                  const CollectionOptions& options = CollectionOptions(),
@@ -245,7 +247,9 @@ private:
     int _profile;  // 0=off.
 
     CollectionMap _collections;
-    ViewCatalog _views;
+
+    DurableViewCatalogImpl _durableViews;  // interface for system.views operations
+    ViewCatalog _views;                    // in-memory representation of _durableViews
 
     friend class Collection;
     friend class NamespaceDetails;
diff --git a/src/mongo/db/catalog/drop_collection.cpp b/src/mongo/db/catalog/drop_collection.cpp
index 32bc00edf90..c436f758b76 100644
--- a/src/mongo/db/catalog/drop_collection.cpp
+++ b/src/mongo/db/catalog/drop_collection.cpp
@@ -44,6 +44,7 @@
 #include "mongo/db/repl/replication_coordinator_global.h"
 #include "mongo/db/server_options.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/views/view_catalog.h"
 #include "mongo/util/log.h"
 
 namespace mongo {
@@ -63,9 +64,9 @@ Status dropCollection(OperationContext* txn,
         AutoGetDb autoDb(txn, dbname, MODE_X);
         Database* const db = autoDb.getDb();
         Collection* coll = db ? db->getCollection(collectionName) : nullptr;
+        ViewDefinition* view = db ? db->getViewCatalog()->lookup(collectionName.ns()) : nullptr;
 
-        // If db/collection does not exist, short circuit and return.
-        if (!db || !coll) {
+        if (!db || (!coll && !view)) {
             return Status(ErrorCodes::NamespaceNotFound, "ns not found");
         }
 
@@ -81,21 +82,26 @@ Status dropCollection(OperationContext* txn,
                                         << collectionName.ns());
         }
 
-        int numIndexes = coll->getIndexCatalog()->numIndexesTotal(txn);
-
-        BackgroundOperation::assertNoBgOpInProgForNs(collectionName.ns());
-
         WriteUnitOfWork wunit(txn);
-        Status s = db->dropCollection(txn, collectionName.ns());
-
         result.append("ns", collectionName.ns());
 
-        if (!s.isOK()) {
-            return s;
-        }
+        if (coll) {
+            invariant(!view);
+            int numIndexes = coll->getIndexCatalog()->numIndexesTotal(txn);
 
-        result.append("nIndexesWas", numIndexes);
+            BackgroundOperation::assertNoBgOpInProgForNs(collectionName.ns());
 
+            Status s = db->dropCollection(txn, collectionName.ns());
+
+            if (!s.isOK()) {
+                return s;
+            }
+
+            result.append("nIndexesWas", numIndexes);
+        } else {
+            invariant(view);
+            db->dropView(txn, collectionName.ns());
+        }
         wunit.commit();
     }
     MONGO_WRITE_CONFLICT_RETRY_LOOP_END(txn, "drop", collectionName.ns());
diff --git a/src/mongo/db/catalog/drop_indexes.cpp b/src/mongo/db/catalog/drop_indexes.cpp
index 7d6387e4ae8..cab438315a9 100644
--- a/src/mongo/db/catalog/drop_indexes.cpp
+++ b/src/mongo/db/catalog/drop_indexes.cpp
@@ -135,13 +135,19 @@ Status dropIndexes(OperationContext* txn,
     MONGO_WRITE_CONFLICT_RETRY_LOOP_BEGIN {
         ScopedTransaction transaction(txn, MODE_IX);
         AutoGetDb autoDb(txn, dbName, MODE_X);
+        Database* db = autoDb.getDb();
 
         bool userInitiatedWritesAndNotPrimary = txn->writesAreReplicated() &&
             !repl::getGlobalReplicationCoordinator()->canAcceptWritesFor(nss);
 
         if (userInitiatedWritesAndNotPrimary) {
-            return Status(ErrorCodes::NotMaster,
-                          str::stream() << "Not primary while dropping indexes in " << nss.ns());
+            return {ErrorCodes::NotMaster,
+                    str::stream() << "Not primary while dropping indexes in " << nss.ns()};
+        }
+
+        if (db && db->getViewCatalog()->lookup(nss.ns())) {
+            return {ErrorCodes::CommandNotSupportedOnView,
+                    str::stream() << "Cannot drop indexes on view " << nss.ns()};
         }
 
         WriteUnitOfWork wunit(txn);
diff --git a/src/mongo/db/commands/compact.cpp b/src/mongo/db/commands/compact.cpp
index 8746aeab13c..4c68ff88fe9 100644
--- a/src/mongo/db/commands/compact.cpp
+++ b/src/mongo/db/commands/compact.cpp
@@ -144,16 +144,21 @@ public:
         if (cmdObj.hasElement("validate"))
             compactOptions.validateDocuments = cmdObj["validate"].trueValue();
 
-
         ScopedTransaction transaction(txn, MODE_IX);
         AutoGetDb autoDb(txn, db, MODE_X);
         Database* const collDB = autoDb.getDb();
-        Collection* collection = collDB ? collDB->getCollection(nss) : NULL;
+
+        Collection* collection = collDB ? collDB->getCollection(nss) : nullptr;
+        auto view = collDB ? collDB->getViewCatalog()->lookup(nss.ns()) : nullptr;
 
         // If db/collection does not exist, short circuit and return.
         if (!collDB || !collection) {
-            errmsg = "namespace does not exist";
-            return false;
+            if (view)
+                return appendCommandStatus(
+                    result, {ErrorCodes::CommandNotSupportedOnView, "can't compact a view"});
+            else
+                return appendCommandStatus(
+                    result, {ErrorCodes::NamespaceNotFound, "collection does not exist"});
         }
 
         OldClientContext ctx(txn, nss.ns());
diff --git a/src/mongo/db/commands/create_indexes.cpp b/src/mongo/db/commands/create_indexes.cpp
index 5d6b951dcbd..e37eb45233d 100644
--- a/src/mongo/db/commands/create_indexes.cpp
+++ b/src/mongo/db/commands/create_indexes.cpp
@@ -48,6 +48,7 @@
 #include "mongo/db/s/collection_metadata.h"
 #include "mongo/db/s/collection_sharding_state.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/views/view_catalog.h"
 #include "mongo/s/shard_key_pattern.h"
 #include "mongo/util/scopeguard.h"
 
@@ -186,6 +187,13 @@ public:
         }
 
         Database* db = dbHolder().get(txn, ns.db());
+
+        if (db && db->getViewCatalog()->lookup(ns.ns())) {
+            errmsg = "cannot create indexes on a view";
+            return appendCommandStatus(result,
+                                       Status(ErrorCodes::CommandNotSupportedOnView, errmsg));
+        }
+
         if (!db) {
             db = dbHolder().openDb(txn, ns.db());
         }
diff --git a/src/mongo/db/commands/drop_indexes.cpp b/src/mongo/db/commands/drop_indexes.cpp
index 2b334a51e87..fb98444202e 100644
--- a/src/mongo/db/commands/drop_indexes.cpp
+++ b/src/mongo/db/commands/drop_indexes.cpp
@@ -124,22 +124,27 @@ public:
              BSONObjBuilder& result) {
         DBDirectClient db(txn);
 
-        const NamespaceString toDeleteNs = parseNsCollectionRequired(dbname, jsobj);
+        const NamespaceString toReIndexNs = parseNsCollectionRequired(dbname, jsobj);
 
-        LOG(0) << "CMD: reIndex " << toDeleteNs << endl;
+        LOG(0) << "CMD: reIndex " << toReIndexNs << endl;
 
         ScopedTransaction transaction(txn, MODE_IX);
         Lock::DBLock dbXLock(txn->lockState(), dbname, MODE_X);
-        OldClientContext ctx(txn, toDeleteNs.ns());
+        OldClientContext ctx(txn, toReIndexNs.ns());
 
-        Collection* collection = ctx.db()->getCollection(toDeleteNs.ns());
+        Collection* collection = ctx.db()->getCollection(toReIndexNs.ns());
+        auto view = ctx.db()->getViewCatalog()->lookup(toReIndexNs.ns());
 
         if (!collection) {
-            errmsg = "ns not found";
-            return false;
+            if (view)
+                return appendCommandStatus(
+                    result, {ErrorCodes::CommandNotSupportedOnView, "can't re-index a view"});
+            else
+                return appendCommandStatus(
+                    result, {ErrorCodes::NamespaceNotFound, "collection does not exist"});
         }
 
-        BackgroundOperation::assertNoBgOpInProgForNs(toDeleteNs.ns());
+        BackgroundOperation::assertNoBgOpInProgForNs(toReIndexNs.ns());
 
         vector<BSONObj> all;
         {
diff --git a/src/mongo/db/commands/find_and_modify.cpp b/src/mongo/db/commands/find_and_modify.cpp
index 0619c8e91c3..07655e32572 100644
--- a/src/mongo/db/commands/find_and_modify.cpp
+++ b/src/mongo/db/commands/find_and_modify.cpp
@@ -395,6 +395,11 @@ public:
                 }
 
                 AutoGetOrCreateDb autoDb(txn, dbName, MODE_IX);
+                if (autoDb.getDb()->getViewCatalog()->lookup(nsString.ns())) {
+                    return appendCommandStatus(result,
+                                               {ErrorCodes::CommandNotSupportedOnView,
+                                                "findAndModify not supported on views"});
+                }
                 Lock::CollectionLock collLock(txn->lockState(), nsString.ns(), MODE_IX);
 
                 // Attach the namespace and database profiling level to the current op.
@@ -467,6 +472,12 @@ public:
                 }
 
                 AutoGetOrCreateDb autoDb(txn, dbName, MODE_IX);
+                if (autoDb.getDb()->getViewCatalog()->lookup(nsString.ns())) {
+                    return appendCommandStatus(result,
+                                               {ErrorCodes::CommandNotSupportedOnView,
+                                                "findAndModify not supported on views"});
+                }
+
                 Lock::CollectionLock collLock(txn->lockState(), nsString.ns(), MODE_IX);
 
                 // Attach the namespace and database profiling level to the current op.
diff --git a/src/mongo/db/views/SConscript b/src/mongo/db/views/SConscript
index 3c08e13b4b3..0a63322240c 100644
--- a/src/mongo/db/views/SConscript
+++ b/src/mongo/db/views/SConscript
@@ -3,6 +3,18 @@
 Import("env")
 
 env.Library(
+    target='views_mongod',
+    source=[
+        'durable_view_catalog.cpp',
+    ],
+    LIBDEPS=[
+        '$BUILD_DIR/mongo/db/dbhelpers',
+        '$BUILD_DIR/mongo/db/views/views',
+        '$BUILD_DIR/mongo/db/catalog/catalog',
+    ]
+)
+
+env.Library(
     target='views',
     source=[
         'view.cpp',
diff --git a/src/mongo/db/views/durable_view_catalog.cpp b/src/mongo/db/views/durable_view_catalog.cpp
new file mode 100644
index 00000000000..2fa3de3364d
--- /dev/null
+++ b/src/mongo/db/views/durable_view_catalog.cpp
@@ -0,0 +1,99 @@
+/**
+ *    Copyright (C) 2016 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kStorage
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/views/durable_view_catalog.h"
+
+#include <string>
+
+#include "mongo/db/catalog/collection.h"
+#include "mongo/db/catalog/database.h"
+#include "mongo/db/dbhelpers.h"
+#include "mongo/db/namespace_string.h"
+#include "mongo/db/operation_context.h"
+#include "mongo/db/storage/record_data.h"
+#include "mongo/util/assert_util.h"
+#include "mongo/util/log.h"
+
+namespace mongo {
+
+void DurableViewCatalogImpl::iterate(OperationContext* txn, Callback callback) {
+    dassert(txn->lockState()->isDbLockedForMode(_db->name(), MODE_X));
+    Collection* systemViews = _db->getCollection(_db->getSystemViewsName());
+    if (!systemViews)
+        return;
+
+    auto cursor = systemViews->getCursor(txn);
+    while (auto record = cursor->next()) {
+        RecordData& data = record->data;
+
+        // Check the document is valid BSON, with only the expected fields.
+        fassertStatusOK(40224, validateBSON(data.data(), data.size()));
+        BSONObj viewDef = data.toBson();
+
+        // Make sure we fail when new fields get added to the definition, so we fail safe in case
+        // of future format upgrades.
+        for (const BSONElement& e : viewDef) {
+            std::string name(e.fieldName());
+            fassert(40225, name == "_id" || name == "viewOn" || name == "pipeline");
+        }
+        NamespaceString viewName(viewDef["_id"].str());
+        fassert(40226, viewName.db() == _db->name());
+
+        callback(viewDef);
+    }
+}
+
+void DurableViewCatalogImpl::insert(OperationContext* txn, const BSONObj& view) {
+    dassert(txn->lockState()->isDbLockedForMode(_db->name(), MODE_X));
+    Collection* systemViews = _db->getOrCreateCollection(txn, _db->getSystemViewsName());
+
+    OpDebug* const opDebug = nullptr;
+    const bool enforceQuota = false;
+    LOG(2) << "insert view " << view << " in " << _db->getSystemViewsName();
+    uassertStatusOK(systemViews->insertDocument(txn, view, opDebug, enforceQuota));
+}
+
+void DurableViewCatalogImpl::remove(OperationContext* txn, const NamespaceString& name) {
+    dassert(txn->lockState()->isDbLockedForMode(_db->name(), MODE_X));
+    Collection* systemViews = _db->getCollection(_db->getSystemViewsName());
+    if (!systemViews)
+        return;
+    const bool requireIndex = false;
+    RecordId id = Helpers::findOne(txn, systemViews, BSON("_id" << name.ns()), requireIndex);
+    if (!id.isNormal())
+        return;
+
+    LOG(2) << "remove view " << name << " from " << _db->getSystemViewsName();
+    OpDebug* const opDebug = nullptr;
+    systemViews->deleteDocument(txn, id, opDebug);
+}
+}  // namespace mongo
diff --git a/src/mongo/db/views/durable_view_catalog.h b/src/mongo/db/views/durable_view_catalog.h
new file mode 100644
index 00000000000..68dec43255f
--- /dev/null
+++ b/src/mongo/db/views/durable_view_catalog.h
@@ -0,0 +1,68 @@
+/**
+ *    Copyright (C) 2016 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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/stdx/functional.h"
+
+namespace mongo {
+
+class BSONObj;
+class Database;
+class NamespaceString;
+class OperationContext;
+
+/**
+ * Interface for system.views collection operations associated with view catalog management.
+ * Methods must be called from within a WriteUnitOfWork, and with the DBLock held.
+ */
+class DurableViewCatalog {
+public:
+    using Callback = stdx::function<void(const BSONObj& view)>;
+
+    virtual void iterate(OperationContext* txn, Callback callback) = 0;
+    virtual void insert(OperationContext* txn, const BSONObj& view) = 0;
+    virtual void remove(OperationContext* txn, const NamespaceString& name) = 0;
+};
+
+/**
+ * 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* txn, Callback callback);
+    void insert(OperationContext* txn, const BSONObj& view);
+    void remove(OperationContext* txn, const NamespaceString& name);
+
+private:
+    Database* const _db;
+};
+}  // namespace mongo
diff --git a/src/mongo/db/views/view.cpp b/src/mongo/db/views/view.cpp
index 380f5eeaed5..5a584c1177b 100644
--- a/src/mongo/db/views/view.cpp
+++ b/src/mongo/db/views/view.cpp
@@ -43,4 +43,18 @@ ViewDefinition::ViewDefinition(StringData dbName,
         _pipeline.push_back(e.Obj().getOwned());
     }
 }
+
+void ViewDefinition::setViewOn(const NamespaceString& viewOnNss) {
+    invariant(_viewNss.db() == viewOnNss.db());
+    _viewOnNss = viewOnNss;
+}
+
+void ViewDefinition::setPipeline(const BSONElement& pipeline) {
+    invariant(pipeline.type() == Array);
+    _pipeline.clear();
+    for (BSONElement e : pipeline.Obj()) {
+        BSONObj value = e.Obj();
+        _pipeline.push_back(value.copy());
+    }
+}
 }  // namespace mongo
diff --git a/src/mongo/db/views/view.h b/src/mongo/db/views/view.h
index 5238a8638fd..c06f5d26a52 100644
--- a/src/mongo/db/views/view.h
+++ b/src/mongo/db/views/view.h
@@ -79,6 +79,13 @@ public:
         return _pipeline;
     }
 
+    void setViewOn(const NamespaceString& viewOnNss);
+
+    /**
+     * Pipeline must be of type array.
+     */
+    void setPipeline(const BSONElement& pipeline);
+
 private:
     NamespaceString _viewNss;
     NamespaceString _viewOnNss;
diff --git a/src/mongo/db/views/view_catalog.cpp b/src/mongo/db/views/view_catalog.cpp
index 42457340d1e..2f1ea88a1d1 100644
--- a/src/mongo/db/views/view_catalog.cpp
+++ b/src/mongo/db/views/view_catalog.cpp
@@ -26,6 +26,8 @@
 *    it in the license file.
 */
 
+#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kDefault
+
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/views/view_catalog.h"
@@ -34,10 +36,15 @@
 
 #include "mongo/base/status_with.h"
 #include "mongo/base/string_data.h"
+#include "mongo/db/catalog/collection.h"
+#include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/pipeline/aggregation_request.h"
+#include "mongo/db/pipeline/document_source.h"
+#include "mongo/db/pipeline/pipeline.h"
 #include "mongo/db/server_parameters.h"
+#include "mongo/db/storage/recovery_unit.h"
 #include "mongo/db/views/view.h"
 
 namespace {
@@ -83,7 +90,14 @@ BSONObj ResolvedViewDefinition::asExpandedViewAggregation(const AggregationReque
     return aggregationBuilder.obj();
 }
 
-ViewCatalog::ViewCatalog(OperationContext* txn, Database* database) {}
+ViewCatalog::ViewCatalog(OperationContext* txn, DurableViewCatalog* durable) : _durable(durable) {
+    durable->iterate(txn, [&](const BSONObj& view) {
+        NamespaceString viewName(view["_id"].str());
+        ViewDefinition def(
+            viewName.db(), viewName.coll(), view["viewOn"].str(), view["pipeline"].Obj());
+        _viewMap[viewName.ns()] = std::make_shared<ViewDefinition>(def);
+    });
+}
 
 Status ViewCatalog::createView(OperationContext* txn,
                                const NamespaceString& viewName,
@@ -92,21 +106,41 @@ Status ViewCatalog::createView(OperationContext* txn,
     if (!enableViews)
         return Status(ErrorCodes::CommandNotSupported, "View support not enabled");
 
-    if (lookup(StringData(viewName.ns())))
-        return Status(ErrorCodes::NamespaceExists, "Namespace already exists");
 
     if (viewName.db() != viewOn.db())
         return Status(ErrorCodes::BadValue,
                       "View must be created on a view or collection in the same database");
 
+    if (lookup(StringData(viewName.ns())))
+        return Status(ErrorCodes::NamespaceExists, "Namespace already exists");
+
+    if (!NamespaceString::validCollectionName(viewOn.coll()))
+        return Status(ErrorCodes::InvalidNamespace,
+                      str::stream() << "invalid name for 'viewOn': " << viewOn.coll());
+
+    // TODO(SERVER-24768): Need to ensure view is correct and doesn't introduce a cycle.
+
+    BSONObj viewDef =
+        BSON("_id" << viewName.ns() << "viewOn" << viewOn.coll() << "pipeline" << pipeline);
+    _durable->insert(txn, viewDef);
+
     BSONObj ownedPipeline = pipeline.getOwned();
-    txn->recoveryUnit()->onCommit([this, viewName, viewOn, ownedPipeline]() {
-        _viewMap[viewName.ns()] = std::make_shared<ViewDefinition>(
-            viewName.db(), viewName.coll(), viewOn.coll(), ownedPipeline);
-    });
+    _viewMap[viewName.ns()] = std::make_shared<ViewDefinition>(
+        viewName.db(), viewName.coll(), viewOn.coll(), ownedPipeline);
+    txn->recoveryUnit()->onRollback([this, viewName]() { this->_viewMap.erase(viewName.ns()); });
     return Status::OK();
 }
 
+void ViewCatalog::dropView(OperationContext* txn, const NamespaceString& viewName) {
+    _durable->remove(txn, viewName);
+    // Save a copy of the view definition in case we need to roll back.
+    ViewDefinition savedDefinition = *lookup(viewName.ns());
+    _viewMap.erase(viewName.ns());
+    txn->recoveryUnit()->onRollback([this, viewName, savedDefinition]() {
+        this->_viewMap[viewName.ns()] = std::make_shared<ViewDefinition>(savedDefinition);
+    });
+}
+
 ViewDefinition* ViewCatalog::lookup(StringData ns) {
     ViewMap::const_iterator it = _viewMap.find(ns);
     if (it != _viewMap.end()) {
diff --git a/src/mongo/db/views/view_catalog.h b/src/mongo/db/views/view_catalog.h
index ebdf7e0602c..91c9f694d48 100644
--- a/src/mongo/db/views/view_catalog.h
+++ b/src/mongo/db/views/view_catalog.h
@@ -28,17 +28,24 @@
 
 #pragma once
 
+#include <map>
+#include <memory>
+#include <string>
+#include <tuple>
+#include <vector>
+
+#include "mongo/base/status.h"
 #include "mongo/base/status_with.h"
+#include "mongo/base/string_data.h"
+#include "mongo/db/namespace_string.h"
+#include "mongo/db/views/durable_view_catalog.h"
 #include "mongo/db/views/view.h"
 #include "mongo/util/string_map.h"
 
 namespace mongo {
 class AggregationRequest;
 class Database;
-class NamespaceString;
 class OperationContext;
-class Status;
-class StringData;
 
 /**
  * Represents a fully-resolved view: a non-view namespace with a corresponding aggregation pipeline.
@@ -67,7 +74,7 @@ public:
     using ViewMap = StringMap<std::shared_ptr<ViewDefinition>>;
     static const std::uint32_t kMaxViewDepth;
 
-    ViewCatalog(OperationContext* txn, Database* database);
+    ViewCatalog(OperationContext* txn, DurableViewCatalog* durable);
 
     ViewMap::const_iterator begin() const {
         return _viewMap.begin();
@@ -78,12 +85,13 @@ public:
     }
 
     /**
-     * Create a new view.
+     * Create a new view 'viewName' with contents defined by running the specified aggregation
+     * 'pipeline' on a collection or view 'viewOn'. This method will check correctness with
+     * respect to the view catalog, but will not check for conflicts with the database's catalog,
+     * so the check for an existing collection with the same name must be done before calling
+     * createView.
      *
-     * @param viewName The name of the view being created.
-     * @param viewOn The name of the view or collection upon which this view is defined.
-     * @param pipeline The aggregation pipeline that defines the aggregation on the backing
-     * namespace.
+     * Must be in WriteUnitOfWork. View creation rolls back if the unit of work aborts.
      */
     Status createView(OperationContext* txn,
                       const NamespaceString& viewName,
@@ -91,6 +99,13 @@ public:
                       const BSONObj& pipeline);
 
     /**
+     * Drop the view named 'viewName'.
+     *
+     * Must be in WriteUnitOfWork. The drop rolls back if the unit of work aborts.
+     */
+    void dropView(OperationContext* txn, const NamespaceString& viewName);
+
+    /**
      * Look up the namespace in the view catalog, returning a pointer to a View definition, or
      * nullptr if it doesn't exist. Note that the caller does not own the pointer.
      *
@@ -111,5 +126,6 @@ public:
 
 private:
     ViewMap _viewMap;
+    DurableViewCatalog* _durable;
 };
 }  // namespace mongo