SERVER-51886 $lookup + $merge pipeline may fail to resolve views correctly when collection names collide

(cherry picked from commit 045829020fb4a3972b316b1ad60e9ed2f285b092)

2 files changed, 99 insertions, 18 deletions

diff --git a/jstests/aggregation/view_resolution_namespace_collision.js b/jstests/aggregation/view_resolution_namespace_collision.js
new file mode 100644
index 00000000000..0f961cbd0a7
--- /dev/null
+++ b/jstests/aggregation/view_resolution_namespace_collision.js
@@ -0,0 +1,60 @@
+/**
+ * Tests that view resolution works correctly when a cross-database aggregation targets a
+ * collection and a view with the same name on different databases.
+ *
+ * @tags: [assumes_no_implicit_collection_creation_after_drop, sbe_incompatible]
+ */
+(function() {
+"use strict";
+
+const dbName = "test";
+const testDB = db.getSiblingDB(dbName);
+const siblingDBName = "otherDB";
+const siblingDB = testDB.getSiblingDB(siblingDBName);
+
+// Since cross-db $out only works against existing databases in a sharded environment, we create a
+// dummy collection on 'siblingDB' to allow this test to run in a sharded environment.
+assert.commandWorked(siblingDB.foo.insert({}));
+
+const sourceName = jsTestName();
+const otherCollName = jsTestName() + "_other_coll";
+const collidingName = "collision_collection";
+const sourceColl = testDB[sourceName];
+const otherColl = testDB[otherCollName];
+
+sourceColl.drop();
+otherColl.drop();
+testDB[collidingName].drop();
+
+assert.commandWorked(sourceColl.insert({_id: 0}));
+assert.commandWorked(otherColl.insert({_id: 0, notsecret: 1, secret: "really secret"}));
+
+// Create a view on 'testDB' that will have the same name as the collection that $merge/$out
+// will create.
+assert.commandWorked(testDB.runCommand(
+    {create: collidingName, viewOn: otherCollName, pipeline: [{$project: {secret: 0}}]}));
+
+// Verify that the view gets resolved correctly when performing a cross database aggregation where
+// the $lookup references a view on the source database and the $merge/$out references a collection
+// on the target database with the same name as the view.
+const lookupStage = {
+    $lookup: {from: collidingName, localField: "_id", foreignField: "_id", as: "matches"}
+};
+const withoutWrite = sourceColl.aggregate([lookupStage]).toArray();
+const mergeStage = {
+    $merge: {into: {db: siblingDBName, coll: collidingName}}
+};
+const outStage = {
+    $out: {db: siblingDBName, coll: collidingName}
+};
+
+// The aggregate should always use the view on 'testDB', not an empty collection on 'siblingDB'.
+for (const writeStage of [mergeStage, outStage]) {
+    sourceColl.aggregate([lookupStage, mergeStage]).toArray();
+    const withWrite = siblingDB[collidingName].find().toArray();
+    assert.eq(withoutWrite, withWrite);
+    siblingDB[collidingName].drop();
+}
+
+siblingDB.dropDatabase();
+})();
\ No newline at end of file
diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp
index 6bc40dcd2ee..10b0754b9a2 100644
--- a/src/mongo/db/commands/run_aggregate.cpp
+++ b/src/mongo/db/commands/run_aggregate.cpp
@@ -293,29 +293,16 @@ StatusWith<StringMap<ExpressionContext::ResolvedNamespace>> resolveInvolvedNames
             continue;
         }
 
-        if (involvedNs.db() != request.getNamespaceString().db()) {
-            // If the involved namespace is not in the same database as the aggregation, it must be
-            // from a $merge to a collection in a different database. Since we cannot write to
-            // views, simply assume that the namespace is a collection.
-            resolvedNamespaces[involvedNs.coll()] = {involvedNs, std::vector<BSONObj>{}};
-        } else if (!db ||
-                   CollectionCatalog::get(opCtx).lookupCollectionByNamespace(opCtx, involvedNs)) {
-            // If the aggregation 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(opCtx, involvedNs.ns())) {
-            // If 'involvedNs' refers to a view namespace, then we resolve its definition.
-            auto resolvedView = viewCatalog->resolveView(opCtx, involvedNs);
+        // If 'ns' refers to a view namespace, then we resolve its definition.
+        auto resolveViewDefinition = [&](const NamespaceString& ns) -> Status {
+            auto resolvedView = viewCatalog->resolveView(opCtx, ns);
             if (!resolvedView.isOK()) {
                 return resolvedView.getStatus().withContext(
                     str::stream() << "Failed to resolve view '" << involvedNs.ns());
             }
 
-            resolvedNamespaces[involvedNs.coll()] = {resolvedView.getValue().getNamespace(),
-                                                     resolvedView.getValue().getPipeline()};
+            resolvedNamespaces[ns.coll()] = {resolvedView.getValue().getNamespace(),
+                                             resolvedView.getValue().getPipeline()};
 
             // We parse the pipeline corresponding to the resolved view in case we must resolve
             // other view namespaces that are also involved.
@@ -327,6 +314,40 @@ StatusWith<StringMap<ExpressionContext::ResolvedNamespace>> resolveInvolvedNames
             involvedNamespacesQueue.insert(involvedNamespacesQueue.end(),
                                            resolvedViewInvolvedNamespaces.begin(),
                                            resolvedViewInvolvedNamespaces.end());
+            return Status::OK();
+        };
+
+        // If the involved namespace is not in the same database as the aggregation, it must be
+        // from an $out or a $merge to a collection in a different database.
+        if (involvedNs.db() != request.getNamespaceString().db()) {
+            // SERVER-51886: It is not correct to assume that we are reading from a collection
+            // because the collection targeted by $out/$merge on a given database can have the same
+            // name as a view on the source database. As such, we determine whether the collection
+            // name references a view on the aggregation request's database. Note that the inverse
+            // scenario (mistaking a view for a collection) is not an issue because $merge/$out
+            // cannot target a view.
+            auto nssToCheck = NamespaceString(request.getNamespaceString().db(), involvedNs.coll());
+            if (viewCatalog && viewCatalog->lookup(opCtx, nssToCheck.ns())) {
+                auto status = resolveViewDefinition(nssToCheck);
+                if (!status.isOK()) {
+                    return status;
+                }
+            } else {
+                resolvedNamespaces[involvedNs.coll()] = {involvedNs, std::vector<BSONObj>{}};
+            }
+        } else if (!db ||
+                   CollectionCatalog::get(opCtx).lookupCollectionByNamespace(opCtx, involvedNs)) {
+            // If the aggregation 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(opCtx, involvedNs.ns())) {
+            auto status = resolveViewDefinition(involvedNs);
+            if (!status.isOK()) {
+                return status;
+            }
         } else {
             // '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.