SERVER-41813 Allow ViewCatalog lookup without validating system views

5 files changed, 150 insertions, 71 deletions

diff --git a/src/mongo/db/views/durable_view_catalog.cpp b/src/mongo/db/views/durable_view_catalog.cpp
index 308e5d14d50..30206b54d91 100644
--- a/src/mongo/db/views/durable_view_catalog.cpp
+++ b/src/mongo/db/views/durable_view_catalog.cpp
@@ -70,68 +70,85 @@ const std::string& DurableViewCatalogImpl::getName() const {
     return _db->name();
 }
 
-Status DurableViewCatalogImpl::iterate(OperationContext* opCtx, Callback callback) {
+void DurableViewCatalogImpl::iterate(OperationContext* opCtx, Callback callback) {
+    _iterate(opCtx, callback, ViewCatalogLookupBehavior::kValidateDurableViews);
+}
+
+void DurableViewCatalogImpl::iterateIgnoreInvalidEntries(OperationContext* opCtx,
+                                                         Callback callback) {
+    _iterate(opCtx, callback, ViewCatalogLookupBehavior::kAllowInvalidDurableViews);
+}
+
+void DurableViewCatalogImpl::_iterate(OperationContext* opCtx,
+                                      Callback callback,
+                                      ViewCatalogLookupBehavior lookupBehavior) {
     invariant(opCtx->lockState()->isCollectionLockedForMode(_db->getSystemViewsName(), MODE_IS));
 
     Collection* systemViews = _db->getCollection(opCtx, _db->getSystemViewsName());
-    if (!systemViews)
-        return Status::OK();
+    if (!systemViews) {
+        return;
+    }
 
     auto cursor = systemViews->getCursor(opCtx);
     while (auto record = cursor->next()) {
-        RecordData& data = record->data;
-
-        // 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(data.data(), data.size(), BSONVersion::kLatest));
-        BSONObj viewDef = data.toBson();
-
-        // Check read definitions for correct structure, and refuse reading past invalid
-        // definitions. Ignore any further view definitions.
-        bool valid = true;
-        for (const BSONElement& e : viewDef) {
-            std::string name(e.fieldName());
-            valid &= name == "_id" || name == "viewOn" || name == "pipeline" || name == "collation";
+        BSONObj viewDefinition;
+        try {
+            viewDefinition = _validateViewDefinition(opCtx, record->data);
+            uassertStatusOK(callback(viewDefinition));
+        } catch (const ExceptionFor<ErrorCodes::InvalidViewDefinition>& ex) {
+            if (lookupBehavior == ViewCatalogLookupBehavior::kValidateDurableViews) {
+                throw ex;
+            }
         }
+    }
+}
 
-        const auto viewName = viewDef["_id"].str();
-        const auto viewNameIsValid = NamespaceString::validCollectionComponent(viewName) &&
-            NamespaceString::validDBName(nsToDatabaseSubstring(viewName));
-        valid &= viewNameIsValid;
+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(), BSONVersion::kLatest));
+    BSONObj viewDefinition = recordData.toBson();
 
-        // 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.db() == _db->name();
-        }
+    bool valid = true;
 
-        valid &= NamespaceString::validCollectionName(viewDef["viewOn"].str());
+    for (const BSONElement& e : viewDefinition) {
+        std::string name(e.fieldName());
+        valid &= name == "_id" || name == "viewOn" || name == "pipeline" || name == "collation";
+    }
 
-        const bool hasPipeline = viewDef.hasField("pipeline");
-        valid &= hasPipeline;
-        if (hasPipeline) {
-            valid &= viewDef["pipeline"].type() == mongo::Array;
-        }
+    const auto viewName = viewDefinition["_id"].str();
+    const auto viewNameIsValid = NamespaceString::validCollectionComponent(viewName) &&
+        NamespaceString::validDBName(nsToDatabaseSubstring(viewName));
+    valid &= viewNameIsValid;
 
-        valid &=
-            (!viewDef.hasField("collation") || viewDef["collation"].type() == BSONType::Object);
+    // 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.db() == _db->name();
+    }
 
-        if (!valid) {
-            return {ErrorCodes::InvalidViewDefinition,
-                    str::stream() << "found invalid view definition " << viewDef["_id"]
-                                  << " while reading '"
-                                  << _db->getSystemViewsName()
-                                  << "'"};
-        }
+    valid &= NamespaceString::validCollectionName(viewDefinition["viewOn"].str());
 
-        Status callbackStatus = callback(viewDef);
-        if (!callbackStatus.isOK()) {
-            return callbackStatus;
-        }
+    const bool hasPipeline = viewDefinition.hasField("pipeline");
+    valid &= hasPipeline;
+    if (hasPipeline) {
+        valid &= viewDefinition["pipeline"].type() == mongo::Array;
     }
-    return Status::OK();
+
+    valid &= (!viewDefinition.hasField("collation") ||
+              viewDefinition["collation"].type() == BSONType::Object);
+
+    uassert(ErrorCodes::InvalidViewDefinition,
+            str::stream() << "found invalid view definition " << viewDefinition["_id"]
+                          << " while reading '"
+                          << _db->getSystemViewsName()
+                          << "'",
+            valid);
+
+    return viewDefinition;
 }
 
 void DurableViewCatalogImpl::upsert(OperationContext* opCtx,
diff --git a/src/mongo/db/views/durable_view_catalog.h b/src/mongo/db/views/durable_view_catalog.h
index 41c2f5837c3..03bacb2f3fb 100644
--- a/src/mongo/db/views/durable_view_catalog.h
+++ b/src/mongo/db/views/durable_view_catalog.h
@@ -41,6 +41,14 @@ namespace mongo {
 class BSONObj;
 class Database;
 class OperationContext;
+class RecordData;
+
+/**
+ * ViewCatalogLookupBehavior specifies whether a lookup into the view catalog should attempt to
+ * validate the durable entries that currently exist within the catalog. This validation should
+ * rarely be skipped.
+ */
+enum class ViewCatalogLookupBehavior { kValidateDurableViews, kAllowInvalidDurableViews };
 
 /**
  * Interface for system.views collection operations associated with view catalog management.
@@ -60,7 +68,8 @@ public:
     static void onExternalChange(OperationContext* opCtx, const NamespaceString& name);
 
     using Callback = std::function<Status(const BSONObj& view)>;
-    virtual Status iterate(OperationContext* opCtx, Callback callback) = 0;
+    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;
@@ -77,12 +86,21 @@ class DurableViewCatalogImpl final : public DurableViewCatalog {
 public:
     explicit DurableViewCatalogImpl(Database* db) : _db(db) {}
 
-    Status iterate(OperationContext* opCtx, Callback callback);
+    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 std::string& 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/view_catalog.cpp b/src/mongo/db/views/view_catalog.cpp
index c212bc2d720..b952731f82c 100644
--- a/src/mongo/db/views/view_catalog.cpp
+++ b/src/mongo/db/views/view_catalog.cpp
@@ -87,10 +87,12 @@ Status ViewCatalog::reloadIfNeeded(OperationContext* opCtx) {
         NamespaceString(_durable->getName(), NamespaceString::kSystemDotViewsCollectionName),
         MODE_IS);
     stdx::unique_lock<stdx::mutex> lk(_mutex);
-    return _reloadIfNeeded(lk, opCtx);
+    return _reloadIfNeeded(lk, opCtx, ViewCatalogLookupBehavior::kValidateDurableViews);
 }
 
-Status ViewCatalog::_reloadIfNeeded(WithLock lk, OperationContext* opCtx) {
+Status ViewCatalog::_reloadIfNeeded(WithLock lk,
+                                    OperationContext* opCtx,
+                                    ViewCatalogLookupBehavior lookupBehavior) {
     if (_valid.load())
         return Status::OK();
 
@@ -99,7 +101,7 @@ Status ViewCatalog::_reloadIfNeeded(WithLock lk, OperationContext* opCtx) {
     // Need to reload, first clear our cache.
     _viewMap.clear();
 
-    Status status = _durable->iterate(opCtx, [&](const BSONObj& view) -> Status {
+    auto reloadCallback = [&](const BSONObj& view) -> Status {
         BSONObj collationSpec = view.hasField("collation") ? view["collation"].Obj() : BSONObj();
         auto collator = parseCollator(opCtx, collationSpec);
         if (!collator.isOK()) {
@@ -125,15 +127,25 @@ Status ViewCatalog::_reloadIfNeeded(WithLock lk, OperationContext* opCtx) {
                                                                    pipeline,
                                                                    std::move(collator.getValue()));
         return Status::OK();
-    });
-    _valid.store(status.isOK());
+    };
 
-    if (!status.isOK()) {
+    try {
+        if (lookupBehavior == ViewCatalogLookupBehavior::kValidateDurableViews) {
+            _durable->iterate(opCtx, reloadCallback);
+        } else if (lookupBehavior == ViewCatalogLookupBehavior::kAllowInvalidDurableViews) {
+            _durable->iterateIgnoreInvalidEntries(opCtx, reloadCallback);
+        } else {
+            MONGO_UNREACHABLE;
+        }
+    } catch (const DBException& ex) {
+        auto status = ex.toStatus();
         LOG(0) << "could not load view catalog for database " << _durable->getName() << ": "
                << status;
+        return status;
     }
 
-    return status;
+    _valid.store(true);
+    return Status::OK();
 }
 
 void ViewCatalog::iterate(OperationContext* opCtx, ViewIteratorCallback callback) {
@@ -142,7 +154,7 @@ void ViewCatalog::iterate(OperationContext* opCtx, ViewIteratorCallback callback
         NamespaceString(_durable->getName(), NamespaceString::kSystemDotViewsCollectionName),
         MODE_IS);
     stdx::lock_guard<stdx::mutex> lk(_mutex);
-    uassertStatusOK(_reloadIfNeeded(lk, opCtx));
+    uassertStatusOK(_reloadIfNeeded(lk, opCtx, ViewCatalogLookupBehavior::kValidateDurableViews));
     for (auto&& view : _viewMap) {
         callback(*view.second);
     }
@@ -324,7 +336,8 @@ Status ViewCatalog::_validateCollation(WithLock lk,
                                        const ViewDefinition& view,
                                        const std::vector<NamespaceString>& refs) {
     for (auto&& potentialViewNss : refs) {
-        auto otherView = _lookup(lk, opCtx, potentialViewNss.ns());
+        auto otherView = _lookup(
+            lk, opCtx, potentialViewNss.ns(), ViewCatalogLookupBehavior::kValidateDurableViews);
         if (otherView &&
             !CollatorInterface::collatorsMatch(view.defaultCollator(),
                                                otherView->defaultCollator())) {
@@ -354,7 +367,8 @@ Status ViewCatalog::createView(OperationContext* opCtx,
         return Status(ErrorCodes::BadValue,
                       "View must be created on a view or collection in the same database");
 
-    if (_lookup(lk, opCtx, StringData(viewName.ns())))
+    if (_lookup(
+            lk, opCtx, StringData(viewName.ns()), ViewCatalogLookupBehavior::kValidateDurableViews))
         return Status(ErrorCodes::NamespaceExists, "Namespace already exists");
 
     if (!NamespaceString::validCollectionName(viewOn.coll()))
@@ -386,7 +400,8 @@ Status ViewCatalog::modifyView(OperationContext* opCtx,
         return Status(ErrorCodes::BadValue,
                       "View must be created on a view or collection in the same database");
 
-    auto viewPtr = _lookup(lk, opCtx, viewName.ns());
+    auto viewPtr =
+        _lookup(lk, opCtx, viewName.ns(), ViewCatalogLookupBehavior::kValidateDurableViews);
     if (!viewPtr)
         return Status(ErrorCodes::NamespaceNotFound,
                       str::stream() << "cannot modify missing view " << viewName.ns());
@@ -418,7 +433,8 @@ Status ViewCatalog::dropView(OperationContext* opCtx, const NamespaceString& vie
     _requireValidCatalog(lk, opCtx);
 
     // Save a copy of the view definition in case we need to roll back.
-    auto viewPtr = _lookup(lk, opCtx, viewName.ns());
+    auto viewPtr =
+        _lookup(lk, opCtx, viewName.ns(), ViewCatalogLookupBehavior::kValidateDurableViews);
     if (!viewPtr) {
         return {ErrorCodes::NamespaceNotFound,
                 str::stream() << "cannot drop missing view: " << viewName.ns()};
@@ -443,7 +459,8 @@ Status ViewCatalog::dropView(OperationContext* opCtx, const NamespaceString& vie
 
 std::shared_ptr<ViewDefinition> ViewCatalog::_lookup(WithLock lk,
                                                      OperationContext* opCtx,
-                                                     StringData ns) {
+                                                     StringData ns,
+                                                     ViewCatalogLookupBehavior lookupBehavior) {
     // We expect the catalog to be valid, so short-circuit other checks for best performance.
     if (MONGO_unlikely(!_valid.load())) {
         // If the catalog is invalid, we want to avoid references to virtualized or other invalid
@@ -451,7 +468,7 @@ std::shared_ptr<ViewDefinition> ViewCatalog::_lookup(WithLock lk,
         // invalid view definitions.
         if (!NamespaceString::validCollectionName(ns))
             return nullptr;
-        Status status = _reloadIfNeeded(lk, opCtx);
+        Status status = _reloadIfNeeded(lk, opCtx, lookupBehavior);
         // In case of errors we've already logged a message. Only uassert if there actually is
         // a user connection, as otherwise we'd crash the server. The catalog will remain invalid,
         // and any views after the first invalid one are ignored.
@@ -472,7 +489,17 @@ std::shared_ptr<ViewDefinition> ViewCatalog::lookup(OperationContext* opCtx, Str
         NamespaceString(_durable->getName(), NamespaceString::kSystemDotViewsCollectionName),
         MODE_IS);
     stdx::lock_guard<stdx::mutex> lk(_mutex);
-    return _lookup(lk, opCtx, ns);
+    return _lookup(lk, opCtx, ns, ViewCatalogLookupBehavior::kValidateDurableViews);
+}
+
+std::shared_ptr<ViewDefinition> ViewCatalog::lookupWithoutValidatingDurableViews(
+    OperationContext* opCtx, StringData ns) {
+    Lock::CollectionLock systemViewsLock(
+        opCtx,
+        NamespaceString(_durable->getName(), NamespaceString::kSystemDotViewsCollectionName),
+        MODE_IS);
+    stdx::lock_guard<stdx::mutex> lk(_mutex);
+    return _lookup(lk, opCtx, ns, ViewCatalogLookupBehavior::kAllowInvalidDurableViews);
 }
 
 StatusWith<ResolvedView> ViewCatalog::resolveView(OperationContext* opCtx,
@@ -504,11 +531,13 @@ StatusWith<ResolvedView> ViewCatalog::resolveView(OperationContext* opCtx,
         for (; depth < ViewGraph::kMaxViewDepth; depth++) {
             // If the catalog has been invalidated, bail and restart.
             if (!_valid.load()) {
-                uassertStatusOK(_reloadIfNeeded(lock, opCtx));
+                uassertStatusOK(
+                    _reloadIfNeeded(lock, opCtx, ViewCatalogLookupBehavior::kValidateDurableViews));
                 break;
             }
 
-            auto view = _lookup(lock, opCtx, resolvedNss->ns());
+            auto view = _lookup(
+                lock, opCtx, resolvedNss->ns(), ViewCatalogLookupBehavior::kValidateDurableViews);
             if (!view) {
                 // Return error status if pipeline is too large.
                 int pipelineSize = 0;
diff --git a/src/mongo/db/views/view_catalog.h b/src/mongo/db/views/view_catalog.h
index 05509f5e2fb..8aee3c1b730 100644
--- a/src/mongo/db/views/view_catalog.h
+++ b/src/mongo/db/views/view_catalog.h
@@ -119,6 +119,13 @@ public:
     std::shared_ptr<ViewDefinition> lookup(OperationContext* opCtx, StringData nss);
 
     /**
+     * Same functionality as above, except this function skips validating durable views in the view
+     * catalog.
+     */
+    std::shared_ptr<ViewDefinition> lookupWithoutValidatingDurableViews(OperationContext* opCtx,
+                                                                        StringData nss);
+
+    /**
      * Resolve the views on 'nss', transforming the pipeline appropriately. This function returns a
      * fully-resolved view definition containing the backing namespace, the resolved pipeline and
      * the collation to use for the operation.
@@ -172,11 +179,17 @@ private:
                               const ViewDefinition& view,
                               const std::vector<NamespaceString>& refs);
 
-    std::shared_ptr<ViewDefinition> _lookup(WithLock, OperationContext* opCtx, StringData ns);
-    Status _reloadIfNeeded(WithLock, OperationContext* opCtx);
+    std::shared_ptr<ViewDefinition> _lookup(WithLock,
+                                            OperationContext* opCtx,
+                                            StringData ns,
+                                            ViewCatalogLookupBehavior lookupBehavior);
+    Status _reloadIfNeeded(WithLock,
+                           OperationContext* opCtx,
+                           ViewCatalogLookupBehavior lookupBehavior);
 
     void _requireValidCatalog(WithLock lk, OperationContext* opCtx) {
-        uassertStatusOK(_reloadIfNeeded(lk, opCtx));
+        uassertStatusOK(
+            _reloadIfNeeded(lk, opCtx, ViewCatalogLookupBehavior::kValidateDurableViews));
         invariant(_valid.load());
     }
 
diff --git a/src/mongo/db/views/view_catalog_test.cpp b/src/mongo/db/views/view_catalog_test.cpp
index 8ad236b37df..7802f9a42c5 100644
--- a/src/mongo/db/views/view_catalog_test.cpp
+++ b/src/mongo/db/views/view_catalog_test.cpp
@@ -78,9 +78,11 @@ public:
     static const std::string name;
 
     using Callback = std::function<Status(const BSONObj& view)>;
-    virtual Status iterate(OperationContext* opCtx, Callback callback) {
+    virtual void iterate(OperationContext* opCtx, Callback callback) {
+        ++_iterateCount;
+    }
+    virtual void iterateIgnoreInvalidEntries(OperationContext* opCtx, Callback callback) {
         ++_iterateCount;
-        return Status::OK();
     }
     virtual void upsert(OperationContext* opCtx, const NamespaceString& name, const BSONObj& view) {
         ++_upsertCount;