SERVER-25680 Don't check for view if collection exists

9 files changed, 50 insertions, 47 deletions

diff --git a/src/mongo/db/catalog/coll_mod.cpp b/src/mongo/db/catalog/coll_mod.cpp
index a10ac780db9..cd287bb27cc 100644
--- a/src/mongo/db/catalog/coll_mod.cpp
+++ b/src/mongo/db/catalog/coll_mod.cpp
@@ -56,10 +56,13 @@ Status collMod(OperationContext* txn,
     Collection* coll = db ? db->getCollection(nss) : nullptr;
 
     // May also modify a view instead of a collection.
-    auto view = db ? db->getViewCatalog()->lookup(txn, nss.ns()) : nullptr;
-    boost::optional<ViewDefinition> newView;
-    if (view)
-        newView = {*view};
+    boost::optional<ViewDefinition> view;
+    if (db && !coll) {
+        auto sharedView = db->getViewCatalog()->lookup(txn, nss.ns());
+        if (sharedView) {
+            view = {*sharedView};
+        }
+    }
 
     // This can kill all cursors so don't allow running it while a background operation is in
     // progress.
@@ -245,7 +248,7 @@ Status collMod(OperationContext* txn,
                     Status(ErrorCodes::InvalidOptions, "not a valid aggregation pipeline");
                 continue;
             }
-            newView->setPipeline(e);
+            view->setPipeline(e);
         } else if (str::equals("viewOn", e.fieldName())) {
             if (!view) {
                 errorStatus =
@@ -257,7 +260,7 @@ Status collMod(OperationContext* txn,
                     Status(ErrorCodes::InvalidOptions, "'viewOn' option must be a string");
                 continue;
             }
-            newView->setViewOn(NamespaceString(dbName, e.str()));
+            view->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.
@@ -310,10 +313,10 @@ Status collMod(OperationContext* txn,
         ViewCatalog* catalog = db->getViewCatalog();
 
         BSONArrayBuilder pipeline;
-        for (auto& item : newView->pipeline()) {
+        for (auto& item : view->pipeline()) {
             pipeline.append(item);
         }
-        errorStatus = catalog->modifyView(txn, nss, newView->viewOn(), BSONArray(pipeline.obj()));
+        errorStatus = catalog->modifyView(txn, nss, view->viewOn(), BSONArray(pipeline.obj()));
         if (!errorStatus.isOK()) {
             return errorStatus;
         }
diff --git a/src/mongo/db/catalog/drop_indexes.cpp b/src/mongo/db/catalog/drop_indexes.cpp
index cdeeae8cede..dfa22b06561 100644
--- a/src/mongo/db/catalog/drop_indexes.cpp
+++ b/src/mongo/db/catalog/drop_indexes.cpp
@@ -61,6 +61,11 @@ Status wrappedRun(OperationContext* txn,
 
     // If db/collection does not exist, short circuit and return.
     if (!db || !collection) {
+        if (db && db->getViewCatalog()->lookup(txn, toDeleteNs)) {
+            return {ErrorCodes::CommandNotSupportedOnView,
+                    str::stream() << "Cannot drop indexes on view " << toDeleteNs};
+        }
+
         return Status(ErrorCodes::NamespaceNotFound, "ns not found");
     }
 
@@ -145,7 +150,6 @@ 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);
@@ -155,11 +159,6 @@ Status dropIndexes(OperationContext* txn,
                     str::stream() << "Not primary while dropping indexes in " << nss.ns()};
         }
 
-        if (db && db->getViewCatalog()->lookup(txn, nss.ns())) {
-            return {ErrorCodes::CommandNotSupportedOnView,
-                    str::stream() << "Cannot drop indexes on view " << nss.ns()};
-        }
-
         WriteUnitOfWork wunit(txn);
         Status status = wrappedRun(txn, dbName, nss.ns(), autoDb.getDb(), idxDescriptor, result);
         if (!status.isOK()) {
diff --git a/src/mongo/db/commands/compact.cpp b/src/mongo/db/commands/compact.cpp
index bf768fe131b..9be54eb39ab 100644
--- a/src/mongo/db/commands/compact.cpp
+++ b/src/mongo/db/commands/compact.cpp
@@ -149,7 +149,8 @@ public:
         Database* const collDB = autoDb.getDb();
 
         Collection* collection = collDB ? collDB->getCollection(nss) : nullptr;
-        auto view = collDB ? collDB->getViewCatalog()->lookup(txn, nss.ns()) : nullptr;
+        auto view =
+            collDB && !collection ? collDB->getViewCatalog()->lookup(txn, nss.ns()) : nullptr;
 
         // If db/collection does not exist, short circuit and return.
         if (!collDB || !collection) {
diff --git a/src/mongo/db/commands/create_indexes.cpp b/src/mongo/db/commands/create_indexes.cpp
index eb50818d63e..9e058b8efce 100644
--- a/src/mongo/db/commands/create_indexes.cpp
+++ b/src/mongo/db/commands/create_indexes.cpp
@@ -187,13 +187,6 @@ public:
         }
 
         Database* db = dbHolder().get(txn, ns.db());
-
-        if (db && db->getViewCatalog()->lookup(txn, ns.ns())) {
-            errmsg = "cannot create indexes on a view";
-            return appendCommandStatus(result,
-                                       Status(ErrorCodes::CommandNotSupportedOnView, errmsg));
-        }
-
         if (!db) {
             db = dbHolder().openDb(txn, ns.db());
         }
@@ -202,6 +195,11 @@ public:
         if (collection) {
             result.appendBool("createdCollectionAutomatically", false);
         } else {
+            if (db->getViewCatalog()->lookup(txn, ns.ns())) {
+                errmsg = "Cannot create indexes on a view";
+                return appendCommandStatus(result, {ErrorCodes::CommandNotSupportedOnView, errmsg});
+            }
+
             MONGO_WRITE_CONFLICT_RETRY_LOOP_BEGIN {
                 WriteUnitOfWork wunit(txn);
                 collection = db->createCollection(txn, ns.ns(), CollectionOptions());
diff --git a/src/mongo/db/commands/drop_indexes.cpp b/src/mongo/db/commands/drop_indexes.cpp
index 17fefcfa1ba..da54c84968b 100644
--- a/src/mongo/db/commands/drop_indexes.cpp
+++ b/src/mongo/db/commands/drop_indexes.cpp
@@ -133,10 +133,8 @@ public:
         OldClientContext ctx(txn, toReIndexNs.ns());
 
         Collection* collection = ctx.db()->getCollection(toReIndexNs.ns());
-        auto view = ctx.db()->getViewCatalog()->lookup(txn, toReIndexNs.ns());
-
         if (!collection) {
-            if (view)
+            if (ctx.db()->getViewCatalog()->lookup(txn, toReIndexNs.ns()))
                 return appendCommandStatus(
                     result, {ErrorCodes::CommandNotSupportedOnView, "can't re-index a view"});
             else
diff --git a/src/mongo/db/commands/find_and_modify.cpp b/src/mongo/db/commands/find_and_modify.cpp
index 5a2a1f45435..5282fa72250 100644
--- a/src/mongo/db/commands/find_and_modify.cpp
+++ b/src/mongo/db/commands/find_and_modify.cpp
@@ -395,11 +395,6 @@ public:
                 }
 
                 AutoGetOrCreateDb autoDb(txn, dbName, MODE_IX);
-                if (autoDb.getDb()->getViewCatalog()->lookup(txn, 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.
@@ -418,6 +413,11 @@ public:
                 }
 
                 Collection* const collection = autoDb.getDb()->getCollection(nsString.ns());
+                if (!collection && autoDb.getDb()->getViewCatalog()->lookup(txn, nsString.ns())) {
+                    return appendCommandStatus(result,
+                                               {ErrorCodes::CommandNotSupportedOnView,
+                                                "findAndModify not supported on a view"});
+                }
                 auto statusWithPlanExecutor =
                     getExecutorDelete(txn, opDebug, collection, &parsedDelete);
                 if (!statusWithPlanExecutor.isOK()) {
@@ -472,12 +472,6 @@ public:
                 }
 
                 AutoGetOrCreateDb autoDb(txn, dbName, MODE_IX);
-                if (autoDb.getDb()->getViewCatalog()->lookup(txn, 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.
@@ -496,6 +490,11 @@ public:
                 }
 
                 Collection* collection = autoDb.getDb()->getCollection(nsString.ns());
+                if (!collection && autoDb.getDb()->getViewCatalog()->lookup(txn, nsString.ns())) {
+                    return appendCommandStatus(result,
+                                               {ErrorCodes::CommandNotSupportedOnView,
+                                                "findAndModify not supported on a view"});
+                }
 
                 // Create the collection if it does not exist when performing an upsert
                 // because the update stage does not create its own collection.
diff --git a/src/mongo/db/commands/pipeline_command.cpp b/src/mongo/db/commands/pipeline_command.cpp
index 4f64276be72..20fa4ce7510 100644
--- a/src/mongo/db/commands/pipeline_command.cpp
+++ b/src/mongo/db/commands/pipeline_command.cpp
@@ -175,7 +175,8 @@ StatusWith<StringMap<ExpressionContext::ResolvedNamespace>> resolveInvolvedNames
     // This is necessary to prevent a cycle from being formed among the view definitions cached in
     // 'resolvedNamespaces' because we won't re-resolve a view namespace we've already encountered.
     AutoGetDb autoDb(txn, expCtx->ns.db(), MODE_IS);
-    ViewCatalog* viewCatalog = autoDb.getDb() ? autoDb.getDb()->getViewCatalog() : nullptr;
+    Database* const db = autoDb.getDb();
+    ViewCatalog* viewCatalog = db ? db->getViewCatalog() : nullptr;
 
     const auto& pipelineInvolvedNamespaces = pipeline->getInvolvedCollections();
     std::deque<NamespaceString> involvedNamespacesQueue(pipelineInvolvedNamespaces.begin(),
@@ -190,9 +191,15 @@ StatusWith<StringMap<ExpressionContext::ResolvedNamespace>> resolveInvolvedNames
             continue;
         }
 
-        if (viewCatalog && viewCatalog->lookup(txn, involvedNs.ns())) {
-            // If the database exists and 'involvedNs' refers to a view namespace, then we resolve
-            // its definition.
+        if (!db || db->getCollection(involvedNs.ns())) {
+            // If the database exists and 'involvedNs' refers to a collection namespace, then we
+            // resolve it as an empty pipeline in order to read directly from the underlying
+            // collection. If the database doesn't exist, then we still resolve it as an empty
+            // pipeline because 'involvedNs' doesn't refer to a view namespace in our consistent
+            // snapshot of the view catalog.
+            resolvedNamespaces[involvedNs.coll()] = {involvedNs, std::vector<BSONObj>{}};
+        } else if (viewCatalog->lookup(txn, involvedNs.ns())) {
+            // If 'involvedNs' refers to a view namespace, then we resolve its definition.
             auto resolvedView = viewCatalog->resolveView(txn, involvedNs);
             if (!resolvedView.isOK()) {
                 return {ErrorCodes::FailedToParse,
@@ -220,11 +227,8 @@ StatusWith<StringMap<ExpressionContext::ResolvedNamespace>> resolveInvolvedNames
                                            resolvedViewInvolvedNamespaces.begin(),
                                            resolvedViewInvolvedNamespaces.end());
         } else {
-            // If the database exists and 'involvedNs' refers to a collection namespace, then we
-            // resolve it as an empty pipeline in order to read directly from the underlying
-            // collection. If the database doesn't exist, then we still resolve it as an empty
-            // pipeline because 'involvedNs' doesn't refer to a view namespace in our consistent
-            // snapshot of the view catalog.
+            // 'involvedNs' is neither a view nor a collection, so resolve it as an empty pipeline
+            // to treat it as reading from a non-existent collection.
             resolvedNamespaces[involvedNs.coll()] = {involvedNs, std::vector<BSONObj>{}};
         }
     }
diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp
index 2e610523d99..92f4137198d 100644
--- a/src/mongo/db/repl/oplog.cpp
+++ b/src/mongo/db/repl/oplog.cpp
@@ -972,7 +972,7 @@ Status applyCommand_inlock(OperationContext* txn, const BSONObj& op) {
     }
     {
         Database* db = dbHolder().get(txn, nss.ns());
-        if (db && db->getViewCatalog()->lookup(txn, nss.ns())) {
+        if (db && !db->getCollection(nss.ns()) && db->getViewCatalog()->lookup(txn, nss.ns())) {
             return {ErrorCodes::CommandNotSupportedOnView,
                     str::stream() << "applyOps not supported on view:" << nss.ns()};
         }
diff --git a/src/mongo/db/s/set_shard_version_command.cpp b/src/mongo/db/s/set_shard_version_command.cpp
index de56695a4cb..ff371a0bce3 100644
--- a/src/mongo/db/s/set_shard_version_command.cpp
+++ b/src/mongo/db/s/set_shard_version_command.cpp
@@ -187,7 +187,8 @@ public:
             autoDb.emplace(txn, nss.db(), MODE_IS);
 
             // Views do not require a shard version check.
-            if (autoDb->getDb() && autoDb->getDb()->getViewCatalog()->lookup(txn, nss.ns())) {
+            if (autoDb->getDb() && !autoDb->getDb()->getCollection(nss.ns()) &&
+                autoDb->getDb()->getViewCatalog()->lookup(txn, nss.ns())) {
                 return true;
             }