SERVER-45452 Support for reading from a view (unsharded)

3 files changed, 374 insertions, 256 deletions

diff --git a/jstests/core/views/views_aggregation.js b/jstests/core/views/views_aggregation.js
index 4feac4f7c57..4c0396ad536 100644
--- a/jstests/core/views/views_aggregation.js
+++ b/jstests/core/views/views_aggregation.js
@@ -18,6 +18,7 @@
 load("jstests/aggregation/extras/merge_helpers.js");
 load("jstests/aggregation/extras/utils.js");  // For arrayEq, assertErrorCode, and
                                               // orderedArrayEq.
+load("jstests/libs/fixture_helpers.js");      // For FixtureHelpers.
 
 let viewsDB = db.getSiblingDB("views_aggregation");
 assert.commandWorked(viewsDB.dropDatabase());
@@ -31,28 +32,20 @@ let assertAggResultEq = function(collection, pipeline, expected, ordered) {
     assert(success, tojson({got: arr, expected: expected}));
 };
 let byPopulation = function(a, b) {
-    if (a.pop < b.pop)
-        return -1;
-    else if (a.pop > b.pop)
-        return 1;
-    else
-        return 0;
+    return a.pop - b.pop;
 };
 
 // Populate a collection with some test data.
-let allDocuments = [];
-allDocuments.push({_id: "New York", state: "NY", pop: 7});
-allDocuments.push({_id: "Newark", state: "NJ", pop: 3});
-allDocuments.push({_id: "Palo Alto", state: "CA", pop: 10});
-allDocuments.push({_id: "San Francisco", state: "CA", pop: 4});
-allDocuments.push({_id: "Trenton", state: "NJ", pop: 5});
+const allDocuments = [
+    {_id: "New York", state: "NY", pop: 7},
+    {_id: "Newark", state: "NJ", pop: 3},
+    {_id: "Palo Alto", state: "CA", pop: 10},
+    {_id: "San Francisco", state: "CA", pop: 4},
+    {_id: "Trenton", state: "NJ", pop: 5},
+];
 
 let coll = viewsDB.coll;
-let bulk = coll.initializeUnorderedBulkOp();
-allDocuments.forEach(function(doc) {
-    bulk.insert(doc);
-});
-assert.commandWorked(bulk.execute());
+assert.commandWorked(coll.insert(allDocuments));
 
 // Create views on the data.
 assert.commandWorked(viewsDB.runCommand({create: "emptyPipelineView", viewOn: "coll"}));
@@ -66,160 +59,174 @@ assert.commandWorked(viewsDB.runCommand({
     pipeline: [{$match: {pop: {$gte: 0}}}, {$sort: {pop: 1}}]
 }));
 
-// Find all documents with empty aggregations.
-assertAggResultEq("emptyPipelineView", [], allDocuments);
-assertAggResultEq("identityView", [], allDocuments);
-assertAggResultEq("identityView", [{$match: {}}], allDocuments);
-
-// Filter documents on a view with $match.
-assertAggResultEq(
-    "popSortedView", [{$match: {state: "NY"}}], [{_id: "New York", state: "NY", pop: 7}]);
-
-// An aggregation still works on a view that strips _id.
-assertAggResultEq("noIdView", [{$match: {state: "NY"}}], [{state: "NY", pop: 7}]);
-
-// Aggregations work on views that sort.
-const doOrderedSort = true;
-assertAggResultEq("popSortedView", [], allDocuments.sort(byPopulation), doOrderedSort);
-assertAggResultEq("popSortedView", [{$limit: 1}, {$project: {_id: 1}}], [{_id: "Palo Alto"}]);
-
-// Test that the $out stage errors when writing to a view namespace.
-assertErrorCode(coll, [{$out: "emptyPipelineView"}], ErrorCodes.CommandNotSupportedOnView);
-
-// Test that the $merge stage errors when writing to a view namespace.
-assertMergeFailsForAllModesWithCode({
-    source: viewsDB.coll,
-    target: viewsDB.emptyPipelineView,
-    errorCodes: [ErrorCodes.CommandNotSupportedOnView]
-});
-
-// Test that the $merge stage errors when writing to a view namespace in a foreign database.
-let foreignDB = db.getSiblingDB("views_aggregation_foreign");
-foreignDB.view.drop();
-assert.commandWorked(foreignDB.createView("view", "coll", []));
-
-assertMergeFailsForAllModesWithCode({
-    source: viewsDB.coll,
-    target: foreignDB.view,
-    errorCodes: [ErrorCodes.CommandNotSupportedOnView]
-});
-
-// Test that an aggregate on a view propagates the 'bypassDocumentValidation' option.
-const validatedCollName = "collectionWithValidator";
-viewsDB[validatedCollName].drop();
-assert.commandWorked(
-    viewsDB.createCollection(validatedCollName, {validator: {illegalField: {$exists: false}}}));
-
-viewsDB.invalidDocs.drop();
-viewsDB.invalidDocsView.drop();
-assert.commandWorked(viewsDB.invalidDocs.insert({illegalField: "present"}));
-assert.commandWorked(viewsDB.createView("invalidDocsView", "invalidDocs", []));
+(function testBasicAggregations() {
+    // Find all documents with empty aggregations.
+    assertAggResultEq("emptyPipelineView", [], allDocuments);
+    assertAggResultEq("identityView", [], allDocuments);
+    assertAggResultEq("identityView", [{$match: {}}], allDocuments);
+
+    // Filter documents on a view with $match.
+    assertAggResultEq(
+        "popSortedView", [{$match: {state: "NY"}}], [{_id: "New York", state: "NY", pop: 7}]);
+
+    // An aggregation still works on a view that strips _id.
+    assertAggResultEq("noIdView", [{$match: {state: "NY"}}], [{state: "NY", pop: 7}]);
+
+    // Aggregations work on views that sort.
+    const doOrderedSort = true;
+    assertAggResultEq("popSortedView", [], allDocuments.sort(byPopulation), doOrderedSort);
+    assertAggResultEq("popSortedView", [{$limit: 1}, {$project: {_id: 1}}], [{_id: "Palo Alto"}]);
+})();
+
+(function testAggStagesWritingToViews() {
+    // Test that the $out stage errors when writing to a view namespace.
+    assertErrorCode(coll, [{$out: "emptyPipelineView"}], ErrorCodes.CommandNotSupportedOnView);
+
+    // Test that the $merge stage errors when writing to a view namespace.
+    assertMergeFailsForAllModesWithCode({
+        source: viewsDB.coll,
+        target: viewsDB.emptyPipelineView,
+        errorCodes: [ErrorCodes.CommandNotSupportedOnView]
+    });
 
-assert.commandWorked(
-    viewsDB.runCommand({
-        aggregate: "invalidDocsView",
-        pipeline: [{$out: validatedCollName}],
-        cursor: {},
-        bypassDocumentValidation: true
-    }),
-    "Expected $out insertions to succeed since 'bypassDocumentValidation' was specified");
-
-// Test that an aggregate on a view propagates the 'allowDiskUse' option.
-const extSortLimit = 100 * 1024 * 1024;
-const largeStrSize = 10 * 1024 * 1024;
-const largeStr = new Array(largeStrSize).join('x');
-viewsDB.largeColl.drop();
-for (let i = 0; i <= extSortLimit / largeStrSize; ++i) {
-    assert.commandWorked(viewsDB.largeColl.insert({x: i, largeStr: largeStr}));
-}
-assertErrorCode(viewsDB.largeColl,
-                [{$sort: {x: -1}}],
-                16819,
-                "Expected in-memory sort to fail due to excessive memory usage");
-viewsDB.largeView.drop();
-assert.commandWorked(viewsDB.createView("largeView", "largeColl", []));
-assertErrorCode(viewsDB.largeView,
-                [{$sort: {x: -1}}],
-                16819,
-                "Expected in-memory sort to fail due to excessive memory usage");
+    // Test that the $merge stage errors when writing to a view namespace in a foreign database.
+    let foreignDB = db.getSiblingDB("views_aggregation_foreign");
+    foreignDB.view.drop();
+    assert.commandWorked(foreignDB.createView("view", "coll", []));
 
-assert.commandWorked(
-    viewsDB.runCommand(
-        {aggregate: "largeView", pipeline: [{$sort: {x: -1}}], cursor: {}, allowDiskUse: true}),
-    "Expected aggregate to succeed since 'allowDiskUse' was specified");
+    assertMergeFailsForAllModesWithCode({
+        source: viewsDB.coll,
+        target: foreignDB.view,
+        errorCodes: [ErrorCodes.CommandNotSupportedOnView]
+    });
+})();
+
+(function testOptionsForwarding() {
+    // Test that an aggregate on a view propagates the 'bypassDocumentValidation' option.
+    const validatedCollName = "collectionWithValidator";
+    viewsDB[validatedCollName].drop();
+    assert.commandWorked(
+        viewsDB.createCollection(validatedCollName, {validator: {illegalField: {$exists: false}}}));
+
+    viewsDB.invalidDocs.drop();
+    viewsDB.invalidDocsView.drop();
+    assert.commandWorked(viewsDB.invalidDocs.insert({illegalField: "present"}));
+    assert.commandWorked(viewsDB.createView("invalidDocsView", "invalidDocs", []));
+
+    assert.commandWorked(
+        viewsDB.runCommand({
+            aggregate: "invalidDocsView",
+            pipeline: [{$out: validatedCollName}],
+            cursor: {},
+            bypassDocumentValidation: true
+        }),
+        "Expected $out insertions to succeed since 'bypassDocumentValidation' was specified");
+
+    // Test that an aggregate on a view propagates the 'allowDiskUse' option.
+    const extSortLimit = 100 * 1024 * 1024;
+    const largeStrSize = 10 * 1024 * 1024;
+    const largeStr = new Array(largeStrSize).join('x');
+    viewsDB.largeColl.drop();
+    for (let i = 0; i <= extSortLimit / largeStrSize; ++i) {
+        assert.commandWorked(viewsDB.largeColl.insert({x: i, largeStr: largeStr}));
+    }
+    assertErrorCode(viewsDB.largeColl,
+                    [{$sort: {x: -1}}],
+                    16819,
+                    "Expected in-memory sort to fail due to excessive memory usage");
+    viewsDB.largeView.drop();
+    assert.commandWorked(viewsDB.createView("largeView", "largeColl", []));
+    assertErrorCode(viewsDB.largeView,
+                    [{$sort: {x: -1}}],
+                    16819,
+                    "Expected in-memory sort to fail due to excessive memory usage");
+
+    assert.commandWorked(
+        viewsDB.runCommand(
+            {aggregate: "largeView", pipeline: [{$sort: {x: -1}}], cursor: {}, allowDiskUse: true}),
+        "Expected aggregate to succeed since 'allowDiskUse' was specified");
+})();
 
 // Test explain modes on a view.
-let explainPlan = assert.commandWorked(
-    viewsDB.popSortedView.explain("queryPlanner").aggregate([{$limit: 1}, {$match: {pop: 3}}]));
-assert.eq(
-    explainPlan.stages[0].$cursor.queryPlanner.namespace, "views_aggregation.coll", explainPlan);
-assert(!explainPlan.stages[0].$cursor.hasOwnProperty("executionStats"), explainPlan);
-
-explainPlan = assert.commandWorked(
-    viewsDB.popSortedView.explain("executionStats").aggregate([{$limit: 1}, {$match: {pop: 3}}]));
-assert.eq(
-    explainPlan.stages[0].$cursor.queryPlanner.namespace, "views_aggregation.coll", explainPlan);
-assert(explainPlan.stages[0].$cursor.hasOwnProperty("executionStats"), explainPlan);
-assert.eq(explainPlan.stages[0].$cursor.executionStats.nReturned, 1, explainPlan);
-assert(!explainPlan.stages[0].$cursor.executionStats.hasOwnProperty("allPlansExecution"),
-       explainPlan);
-
-explainPlan = assert.commandWorked(viewsDB.popSortedView.explain("allPlansExecution")
-                                       .aggregate([{$limit: 1}, {$match: {pop: 3}}]));
-assert.eq(
-    explainPlan.stages[0].$cursor.queryPlanner.namespace, "views_aggregation.coll", explainPlan);
-assert(explainPlan.stages[0].$cursor.hasOwnProperty("executionStats"), explainPlan);
-assert.eq(explainPlan.stages[0].$cursor.executionStats.nReturned, 1, explainPlan);
-assert(explainPlan.stages[0].$cursor.executionStats.hasOwnProperty("allPlansExecution"),
-       explainPlan);
-
-// Passing a value of true for the explain option to the aggregation command, without using the
-// shell explain helper, should continue to work.
-explainPlan = assert.commandWorked(
-    viewsDB.popSortedView.aggregate([{$limit: 1}, {$match: {pop: 3}}], {explain: true}));
-assert.eq(
-    explainPlan.stages[0].$cursor.queryPlanner.namespace, "views_aggregation.coll", explainPlan);
-assert(!explainPlan.stages[0].$cursor.hasOwnProperty("executionStats"), explainPlan);
-
-// Test allPlansExecution explain mode on the base collection.
-explainPlan = assert.commandWorked(
-    viewsDB.coll.explain("allPlansExecution").aggregate([{$limit: 1}, {$match: {pop: 3}}]));
-assert.eq(
-    explainPlan.stages[0].$cursor.queryPlanner.namespace, "views_aggregation.coll", explainPlan);
-assert(explainPlan.stages[0].$cursor.hasOwnProperty("executionStats"), explainPlan);
-assert.eq(explainPlan.stages[0].$cursor.executionStats.nReturned, 1, explainPlan);
-assert(explainPlan.stages[0].$cursor.executionStats.hasOwnProperty("allPlansExecution"),
-       explainPlan);
-
-// The explain:true option should not work when paired with the explain shell helper.
-assert.throws(function() {
-    viewsDB.popSortedView.explain("executionStats").aggregate([{$limit: 1}, {$match: {pop: 3}}], {
-        explain: true
+(function testExplainOnView() {
+    let explainPlan = assert.commandWorked(
+        viewsDB.popSortedView.explain("queryPlanner").aggregate([{$limit: 1}, {$match: {pop: 3}}]));
+    assert.eq(explainPlan.stages[0].$cursor.queryPlanner.namespace,
+              "views_aggregation.coll",
+              explainPlan);
+    assert(!explainPlan.stages[0].$cursor.hasOwnProperty("executionStats"), explainPlan);
+
+    explainPlan = assert.commandWorked(viewsDB.popSortedView.explain("executionStats")
+                                           .aggregate([{$limit: 1}, {$match: {pop: 3}}]));
+    assert.eq(explainPlan.stages[0].$cursor.queryPlanner.namespace,
+              "views_aggregation.coll",
+              explainPlan);
+    assert(explainPlan.stages[0].$cursor.hasOwnProperty("executionStats"), explainPlan);
+    assert.eq(explainPlan.stages[0].$cursor.executionStats.nReturned, 1, explainPlan);
+    assert(!explainPlan.stages[0].$cursor.executionStats.hasOwnProperty("allPlansExecution"),
+           explainPlan);
+
+    explainPlan = assert.commandWorked(viewsDB.popSortedView.explain("allPlansExecution")
+                                           .aggregate([{$limit: 1}, {$match: {pop: 3}}]));
+    assert.eq(explainPlan.stages[0].$cursor.queryPlanner.namespace,
+              "views_aggregation.coll",
+              explainPlan);
+    assert(explainPlan.stages[0].$cursor.hasOwnProperty("executionStats"), explainPlan);
+    assert.eq(explainPlan.stages[0].$cursor.executionStats.nReturned, 1, explainPlan);
+    assert(explainPlan.stages[0].$cursor.executionStats.hasOwnProperty("allPlansExecution"),
+           explainPlan);
+
+    // Passing a value of true for the explain option to the aggregation command, without using the
+    // shell explain helper, should continue to work.
+    explainPlan = assert.commandWorked(
+        viewsDB.popSortedView.aggregate([{$limit: 1}, {$match: {pop: 3}}], {explain: true}));
+    assert.eq(explainPlan.stages[0].$cursor.queryPlanner.namespace,
+              "views_aggregation.coll",
+              explainPlan);
+    assert(!explainPlan.stages[0].$cursor.hasOwnProperty("executionStats"), explainPlan);
+
+    // Test allPlansExecution explain mode on the base collection.
+    explainPlan = assert.commandWorked(
+        viewsDB.coll.explain("allPlansExecution").aggregate([{$limit: 1}, {$match: {pop: 3}}]));
+    assert.eq(explainPlan.stages[0].$cursor.queryPlanner.namespace,
+              "views_aggregation.coll",
+              explainPlan);
+    assert(explainPlan.stages[0].$cursor.hasOwnProperty("executionStats"), explainPlan);
+    assert.eq(explainPlan.stages[0].$cursor.executionStats.nReturned, 1, explainPlan);
+    assert(explainPlan.stages[0].$cursor.executionStats.hasOwnProperty("allPlansExecution"),
+           explainPlan);
+
+    // The explain:true option should not work when paired with the explain shell helper.
+    assert.throws(function() {
+        viewsDB.popSortedView.explain("executionStats")
+            .aggregate([{$limit: 1}, {$match: {pop: 3}}], {explain: true});
     });
-});
-
-// The remaining tests involve $lookup and $graphLookup. We cannot lookup into sharded
-// collections, so skip these tests if running in a sharded configuration.
-let isMasterResponse = assert.commandWorked(viewsDB.runCommand("isMaster"));
-const isMongos = (isMasterResponse.msg === "isdbgrid");
-if (isMongos) {
-    jsTest.log("Tests are being run on a mongos; skipping all $lookup and $graphLookup tests.");
-    return;
-}
-
-// Test that the $lookup stage resolves the view namespace referenced in the 'from' field.
-assertAggResultEq(
-    coll.getName(),
-    [
-        {$match: {_id: "New York"}},
-        {$lookup: {from: "identityView", localField: "_id", foreignField: "_id", as: "matched"}},
-        {$unwind: "$matched"},
-        {$project: {_id: 1, matchedId: "$matched._id"}}
-    ],
-    [{_id: "New York", matchedId: "New York"}]);
+})();
+
+(
+    function testLookupAndGraphLookup() {
+        // We cannot lookup into sharded collections, so skip these tests if running in a sharded
+        // configuration.
+        if (FixtureHelpers.isMongos(db)) {
+            jsTest.log(
+                "Tests are being run on a mongos; skipping all $lookup and $graphLookup tests.");
+            return;
+        }
+
+        // Test that the $lookup stage resolves the view namespace referenced in the 'from' field.
+        assertAggResultEq(
+        coll.getName(),
+        [
+            {$match: {_id: "New York"}},
+            {$lookup: {from: "identityView", localField: "_id", foreignField: "_id", as: "matched"}},
+            {$unwind: "$matched"},
+            {$project: {_id: 1, matchedId: "$matched._id"}}
+        ],
+        [{_id: "New York", matchedId: "New York"}]);
 
-// Test that the $graphLookup stage resolves the view namespace referenced in the 'from' field.
-assertAggResultEq(coll.getName(),
+        // Test that the $graphLookup stage resolves the view namespace referenced in the 'from'
+        // field.
+        assertAggResultEq(coll.getName(),
                       [
                         {$match: {_id: "New York"}},
                         {
@@ -236,9 +243,9 @@ assertAggResultEq(coll.getName(),
                       ],
                       [{_id: "New York", matchedId: "New York"}]);
 
-// Test that the $lookup stage resolves the view namespace referenced in the 'from' field of
-// another $lookup stage nested inside of it.
-assert.commandWorked(viewsDB.runCommand({
+        // Test that the $lookup stage resolves the view namespace referenced in the 'from' field of
+        // another $lookup stage nested inside of it.
+        assert.commandWorked(viewsDB.runCommand({
     create: "viewWithLookupInside",
     viewOn: coll.getName(),
     pipeline: [
@@ -248,7 +255,7 @@ assert.commandWorked(viewsDB.runCommand({
     ]
 }));
 
-assertAggResultEq(
+        assertAggResultEq(
         coll.getName(),
         [
           {$match: {_id: "New York"}},
@@ -265,9 +272,9 @@ assertAggResultEq(
         ],
         [{_id: "New York", matchedId1: "New York", matchedId2: "New York"}]);
 
-// Test that the $graphLookup stage resolves the view namespace referenced in the 'from' field
-// of a $lookup stage nested inside of it.
-let graphLookupPipeline = [
+        // Test that the $graphLookup stage resolves the view namespace referenced in the 'from'
+        // field of a $lookup stage nested inside of it.
+        let graphLookupPipeline = [
         {$match: {_id: "New York"}},
         {
           $graphLookup: {
@@ -282,13 +289,13 @@ let graphLookupPipeline = [
         {$project: {_id: 1, matchedId1: "$matched._id", matchedId2: "$matched.matchedId"}}
     ];
 
-assertAggResultEq(coll.getName(),
-                  graphLookupPipeline,
-                  [{_id: "New York", matchedId1: "New York", matchedId2: "New York"}]);
+        assertAggResultEq(coll.getName(),
+                          graphLookupPipeline,
+                          [{_id: "New York", matchedId1: "New York", matchedId2: "New York"}]);
 
-// Test that the $lookup stage on a view with a nested $lookup on a different view resolves the
-// view namespaces referenced in their respective 'from' fields.
-assertAggResultEq(
+        // Test that the $lookup stage on a view with a nested $lookup on a different view resolves
+        // the view namespaces referenced in their respective 'from' fields.
+        assertAggResultEq(
         coll.getName(),
         [
           {$match: {_id: "Trenton"}},
@@ -321,9 +328,33 @@ assertAggResultEq(
            }]
         }]);
 
-// Test that the $facet stage resolves the view namespace referenced in the 'from' field of a
-// $lookup stage nested inside of a $graphLookup stage.
-assertAggResultEq(coll.getName(),
-                  [{$facet: {nested: graphLookupPipeline}}],
-                  [{nested: [{_id: "New York", matchedId1: "New York", matchedId2: "New York"}]}]);
-}());
+        // Test that the $facet stage resolves the view namespace referenced in the 'from' field of
+        // a $lookup stage nested inside of a $graphLookup stage.
+        assertAggResultEq(
+            coll.getName(),
+            [{$facet: {nested: graphLookupPipeline}}],
+            [{nested: [{_id: "New York", matchedId1: "New York", matchedId2: "New York"}]}]);
+    })();
+
+(function testUnionReadFromView() {
+    if (FixtureHelpers.isMongos(db)) {
+        // TODO SERVER-45563 enable these tests in sharded environments.
+        jsTest.log("Tests are being run on a mongos; skipping all $unionWith view tests.");
+        return;
+    }
+    assert.eq(allDocuments.length, coll.aggregate([]).itcount());
+    assert.eq(2 * allDocuments.length,
+              coll.aggregate([{$unionWith: "emptyPipelineView"}]).itcount());
+    assert.eq(2 * allDocuments.length, coll.aggregate([{$unionWith: "identityView"}]).itcount());
+    assert.eq(
+        2 * allDocuments.length,
+        coll.aggregate(
+                [{$unionWith: {coll: "noIdView", pipeline: [{$match: {_id: {$exists: false}}}]}}])
+            .itcount());
+    assert.eq(
+        allDocuments.length + 1,
+        coll.aggregate(
+                [{$unionWith: {coll: "identityView", pipeline: [{$match: {_id: "New York"}}]}}])
+            .itcount());
+})();
+})();
diff --git a/src/mongo/db/pipeline/document_source_union_with.cpp b/src/mongo/db/pipeline/document_source_union_with.cpp
index 67c9fe032e3..a6a31657dbf 100644
--- a/src/mongo/db/pipeline/document_source_union_with.cpp
+++ b/src/mongo/db/pipeline/document_source_union_with.cpp
@@ -30,6 +30,8 @@
 
 #include "mongo/platform/basic.h"
 
+#include <iterator>
+
 #include "mongo/db/pipeline/document_source_union_with.h"
 #include "mongo/db/pipeline/document_source_union_with_gen.h"
 #include "mongo/util/log.h"
@@ -40,6 +42,26 @@ REGISTER_TEST_DOCUMENT_SOURCE(unionWith,
                               DocumentSourceUnionWith::LiteParsed::parse,
                               DocumentSourceUnionWith::createFromBson);
 
+namespace {
+std::unique_ptr<Pipeline, PipelineDeleter> buildPipelineFromViewDefinition(
+    const boost::intrusive_ptr<ExpressionContext>& expCtx,
+    ExpressionContext::ResolvedNamespace resolvedNs,
+    std::vector<BSONObj> currentPipeline) {
+    if (resolvedNs.pipeline.empty()) {
+        return uassertStatusOK(Pipeline::parse(currentPipeline, expCtx->copyWith(resolvedNs.ns)));
+    }
+    auto resolvedPipeline = std::move(resolvedNs.pipeline);
+    resolvedPipeline.reserve(currentPipeline.size() + resolvedPipeline.size());
+    resolvedPipeline.insert(resolvedPipeline.end(),
+                            std::make_move_iterator(currentPipeline.begin()),
+                            std::make_move_iterator(currentPipeline.end()));
+
+    return uassertStatusOK(
+        Pipeline::parse(std::move(resolvedPipeline), expCtx->copyWith(resolvedNs.ns)));
+}
+
+}  // namespace
+
 std::unique_ptr<DocumentSourceUnionWith::LiteParsed> DocumentSourceUnionWith::LiteParsed::parse(
     const NamespaceString& nss, const BSONElement& spec) {
     uassert(ErrorCodes::FailedToParse,
@@ -112,8 +134,8 @@ boost::intrusive_ptr<DocumentSource> DocumentSourceUnionWith::createFromBson(
     }
     return make_intrusive<DocumentSourceUnionWith>(
         expCtx,
-        uassertStatusOK(
-            Pipeline::parse(std::move(pipeline), expCtx->copyWith(std::move(unionNss)))));
+        buildPipelineFromViewDefinition(
+            expCtx, expCtx->getResolvedNamespace(std::move(unionNss)), std::move(pipeline)));
 }
 
 DocumentSource::GetNextResult DocumentSourceUnionWith::doGetNext() {
diff --git a/src/mongo/db/pipeline/document_source_union_with_test.cpp b/src/mongo/db/pipeline/document_source_union_with_test.cpp
index 4562cb7eb6a..a8fd3784573 100644
--- a/src/mongo/db/pipeline/document_source_union_with_test.cpp
+++ b/src/mongo/db/pipeline/document_source_union_with_test.cpp
@@ -169,47 +169,53 @@ TEST_F(DocumentSourceUnionWithTest, UnionsWithNonEmptySubPipelines) {
 }
 
 TEST_F(DocumentSourceUnionWithTest, SerializeAndParseWithPipeline) {
+    auto expCtx = getExpCtx();
+    NamespaceString nsToUnionWith(expCtx->ns.db(), "coll");
+    expCtx->setResolvedNamespaces(StringMap<ExpressionContext::ResolvedNamespace>{
+        {nsToUnionWith.coll().toString(), {nsToUnionWith, std::vector<BSONObj>()}}});
     auto bson =
-        BSON("$unionWith" << BSON("coll"
-                                  << "foo"
-                                  << "pipeline"
-                                  << BSON_ARRAY(
-                                         BSON("$addFields" << BSON("a" << BSON("$const" << 3))))));
-    auto unionWith = DocumentSourceUnionWith::createFromBson(bson.firstElement(), getExpCtx());
+        BSON("$unionWith" << BSON(
+                 "coll" << nsToUnionWith.coll() << "pipeline"
+                        << BSON_ARRAY(BSON("$addFields" << BSON("a" << BSON("$const" << 3))))));
+    auto unionWith = DocumentSourceUnionWith::createFromBson(bson.firstElement(), expCtx);
     ASSERT(unionWith->getSourceName() == DocumentSourceUnionWith::kStageName);
     std::vector<Value> serializedArray;
     unionWith->serializeToArray(serializedArray);
     auto serializedBson = serializedArray[0].getDocument().toBson();
     ASSERT_BSONOBJ_EQ(serializedBson, bson);
-    unionWith = DocumentSourceUnionWith::createFromBson(serializedBson.firstElement(), getExpCtx());
+    unionWith = DocumentSourceUnionWith::createFromBson(serializedBson.firstElement(), expCtx);
     ASSERT(unionWith != nullptr);
     ASSERT(unionWith->getSourceName() == DocumentSourceUnionWith::kStageName);
 }
 
 TEST_F(DocumentSourceUnionWithTest, SerializeAndParseWithoutPipeline) {
-    auto bson = BSON("$unionWith"
-                     << "foo");
-    auto desugaredBson = BSON("$unionWith" << BSON("coll"
-                                                   << "foo"
-                                                   << "pipeline" << BSONArray()));
-    auto unionWith = DocumentSourceUnionWith::createFromBson(bson.firstElement(), getExpCtx());
+    auto expCtx = getExpCtx();
+    NamespaceString nsToUnionWith(expCtx->ns.db(), "coll");
+    expCtx->setResolvedNamespaces(StringMap<ExpressionContext::ResolvedNamespace>{
+        {nsToUnionWith.coll().toString(), {nsToUnionWith, std::vector<BSONObj>()}}});
+    auto bson = BSON("$unionWith" << nsToUnionWith.coll());
+    auto desugaredBson =
+        BSON("$unionWith" << BSON("coll" << nsToUnionWith.coll() << "pipeline" << BSONArray()));
+    auto unionWith = DocumentSourceUnionWith::createFromBson(bson.firstElement(), expCtx);
     ASSERT(unionWith->getSourceName() == DocumentSourceUnionWith::kStageName);
     std::vector<Value> serializedArray;
     unionWith->serializeToArray(serializedArray);
     auto serializedBson = serializedArray[0].getDocument().toBson();
     ASSERT_BSONOBJ_EQ(serializedBson, desugaredBson);
-    unionWith = DocumentSourceUnionWith::createFromBson(serializedBson.firstElement(), getExpCtx());
+    unionWith = DocumentSourceUnionWith::createFromBson(serializedBson.firstElement(), expCtx);
     ASSERT(unionWith != nullptr);
     ASSERT(unionWith->getSourceName() == DocumentSourceUnionWith::kStageName);
 }
 
 TEST_F(DocumentSourceUnionWithTest, SerializeAndParseWithoutPipelineExtraSubobject) {
-    auto bson = BSON("$unionWith" << BSON("coll"
-                                          << "foo"));
-    auto desugaredBson = BSON("$unionWith" << BSON("coll"
-                                                   << "foo"
-                                                   << "pipeline" << BSONArray()));
-    auto unionWith = DocumentSourceUnionWith::createFromBson(bson.firstElement(), getExpCtx());
+    auto expCtx = getExpCtx();
+    NamespaceString nsToUnionWith(expCtx->ns.db(), "coll");
+    expCtx->setResolvedNamespaces(StringMap<ExpressionContext::ResolvedNamespace>{
+        {nsToUnionWith.coll().toString(), {nsToUnionWith, std::vector<BSONObj>()}}});
+    auto bson = BSON("$unionWith" << BSON("coll" << nsToUnionWith.coll()));
+    auto desugaredBson =
+        BSON("$unionWith" << BSON("coll" << nsToUnionWith.coll() << "pipeline" << BSONArray()));
+    auto unionWith = DocumentSourceUnionWith::createFromBson(bson.firstElement(), expCtx);
     ASSERT(unionWith->getSourceName() == DocumentSourceUnionWith::kStageName);
     std::vector<Value> serializedArray;
     unionWith->serializeToArray(serializedArray);
@@ -221,50 +227,49 @@ TEST_F(DocumentSourceUnionWithTest, SerializeAndParseWithoutPipelineExtraSubobje
 }
 
 TEST_F(DocumentSourceUnionWithTest, ParseErrors) {
+    auto expCtx = getExpCtx();
+    NamespaceString nsToUnionWith(expCtx->ns.db(), "coll");
+    expCtx->setResolvedNamespaces(StringMap<ExpressionContext::ResolvedNamespace>{
+        {nsToUnionWith.coll().toString(), {nsToUnionWith, std::vector<BSONObj>()}}});
+    ASSERT_THROWS_CODE(
+        DocumentSourceUnionWith::createFromBson(BSON("$unionWith" << false).firstElement(), expCtx),
+        AssertionException,
+        ErrorCodes::FailedToParse);
     ASSERT_THROWS_CODE(DocumentSourceUnionWith::createFromBson(
-                           BSON("$unionWith" << false).firstElement(), getExpCtx()),
-                       AssertionException,
-                       ErrorCodes::FailedToParse);
-    ASSERT_THROWS_CODE(DocumentSourceUnionWith::createFromBson(BSON("$unionWith" << BSON("coll"
-                                                                                         << "foo"
-                                                                                         << "coll"
-                                                                                         << "bar"))
-                                                                   .firstElement(),
-                                                               getExpCtx()),
+                           BSON("$unionWith" << BSON("coll" << nsToUnionWith.coll() << "coll"
+                                                            << "bar"))
+                               .firstElement(),
+                           expCtx),
                        AssertionException,
                        40413);
     ASSERT_THROWS_CODE(
         DocumentSourceUnionWith::createFromBson(
-            BSON("$unionWith" << BSON("coll"
-                                      << "foo"
-                                      << "pipeline"
-                                      << BSON_ARRAY(BSON("$addFields" << BSON("a" << 3))) << "coll"
-                                      << "bar"))
+            BSON("$unionWith" << BSON("coll" << nsToUnionWith.coll() << "pipeline"
+                                             << BSON_ARRAY(BSON("$addFields" << BSON("a" << 3)))
+                                             << "coll"
+                                             << "bar"))
                 .firstElement(),
-            getExpCtx()),
+            expCtx),
         AssertionException,
         40413);
     ASSERT_THROWS_CODE(
         DocumentSourceUnionWith::createFromBson(
-            BSON("$unionWith" << BSON("coll"
-                                      << "foo"
-                                      << "pipeline"
-                                      << BSON_ARRAY(BSON("$addFields" << BSON("a" << 3))) << "myDog"
-                                      << "bar"))
+            BSON("$unionWith" << BSON("coll" << nsToUnionWith.coll() << "pipeline"
+                                             << BSON_ARRAY(BSON("$addFields" << BSON("a" << 3)))
+                                             << "myDog"
+                                             << "bar"))
                 .firstElement(),
-            getExpCtx()),
+            expCtx),
         AssertionException,
         40415);
-    ASSERT_THROWS_CODE(
-        DocumentSourceUnionWith::createFromBson(
-            BSON("$unionWith" << BSON("coll"
-                                      << "foo"
-                                      << "pipeline"
-                                      << BSON_ARRAY(BSON("$petMyDog" << BSON("myDog" << 3)))))
-                .firstElement(),
-            getExpCtx()),
-        AssertionException,
-        16436);
+    ASSERT_THROWS_CODE(DocumentSourceUnionWith::createFromBson(
+                           BSON("$unionWith" << BSON(
+                                    "coll" << nsToUnionWith.coll() << "pipeline"
+                                           << BSON_ARRAY(BSON("$petMyDog" << BSON("myDog" << 3)))))
+                               .firstElement(),
+                           getExpCtx()),
+                       AssertionException,
+                       16436);
     ASSERT_THROWS_CODE(
         DocumentSourceUnionWith::createFromBson(
             BSON("$unionWith" << BSON("coll" << BSON("not"
@@ -275,25 +280,21 @@ TEST_F(DocumentSourceUnionWithTest, ParseErrors) {
             getExpCtx()),
         AssertionException,
         ErrorCodes::TypeMismatch);
-    ASSERT_THROWS_CODE(
-        DocumentSourceUnionWith::createFromBson(BSON("$unionWith" << BSON("coll"
-                                                                          << "foo"
-                                                                          << "pipeline"
-                                                                          << "string"))
-                                                    .firstElement(),
-                                                getExpCtx()),
-        AssertionException,
-        10065);
-    ASSERT_THROWS_CODE(
-        DocumentSourceUnionWith::createFromBson(BSON("$unionWith" << BSON("coll"
-                                                                          << "foo"
-                                                                          << "pipeline"
-                                                                          << BSON("not"
-                                                                                  << "string")))
-                                                    .firstElement(),
-                                                getExpCtx()),
-        AssertionException,
-        40422);
+    ASSERT_THROWS_CODE(DocumentSourceUnionWith::createFromBson(
+                           BSON("$unionWith" << BSON("coll" << nsToUnionWith.coll() << "pipeline"
+                                                            << "string"))
+                               .firstElement(),
+                           getExpCtx()),
+                       AssertionException,
+                       10065);
+    ASSERT_THROWS_CODE(DocumentSourceUnionWith::createFromBson(
+                           BSON("$unionWith" << BSON("coll" << nsToUnionWith.coll() << "pipeline"
+                                                            << BSON("not"
+                                                                    << "string")))
+                               .firstElement(),
+                           getExpCtx()),
+                       AssertionException,
+                       40422);
 }
 
 TEST_F(DocumentSourceUnionWithTest, PropagatePauses) {
@@ -372,5 +373,69 @@ TEST_F(DocumentSourceUnionWithTest, DependencyAnalysisReportsReferencedFieldsBef
     ASSERT_FALSE(deps.needWholeDocument);
 }
 
+TEST_F(DocumentSourceUnionWithTest, RespectsViewDefinition) {
+    auto expCtx = getExpCtx();
+    NamespaceString nsToUnionWith(expCtx->ns.db(), "coll");
+    expCtx->setResolvedNamespaces(StringMap<ExpressionContext::ResolvedNamespace>{
+        {nsToUnionWith.coll().toString(),
+         {nsToUnionWith, std::vector<BSONObj>{fromjson("{$match: {_id: {$mod: [2, 0]}}}")}}}});
+
+    // Mock out the foreign collection.
+    std::deque<DocumentSource::GetNextResult> mockForeignContents{Document{{"_id", 1}},
+                                                                  Document{{"_id", 2}}};
+    expCtx->mongoProcessInterface =
+        std::make_shared<MockMongoInterface>(std::move(mockForeignContents));
+
+    auto bson = BSON("$unionWith" << nsToUnionWith.coll());
+    auto unionWith = DocumentSourceUnionWith::createFromBson(bson.firstElement(), expCtx);
+    const auto localMock = DocumentSourceMock::createForTest({Document{{"_id"_sd, "local"_sd}}});
+    unionWith->setSource(localMock.get());
+
+    auto result = unionWith->getNext();
+    ASSERT_TRUE(result.isAdvanced());
+    ASSERT_DOCUMENT_EQ(result.getDocument(), (Document{{"_id"_sd, "local"_sd}}));
+
+    result = unionWith->getNext();
+    ASSERT_TRUE(result.isAdvanced());
+    ASSERT_DOCUMENT_EQ(result.getDocument(), (Document{{"_id"_sd, 2}}));
+
+    ASSERT_TRUE(unionWith->getNext().isEOF());
+}
+
+TEST_F(DocumentSourceUnionWithTest, ConcatenatesViewDefinitionToPipeline) {
+    auto expCtx = getExpCtx();
+    NamespaceString viewNsToUnionWith(expCtx->ns.db(), "view");
+    NamespaceString nsToUnionWith(expCtx->ns.db(), "coll");
+    expCtx->setResolvedNamespaces(StringMap<ExpressionContext::ResolvedNamespace>{
+        {viewNsToUnionWith.coll().toString(),
+         {nsToUnionWith, std::vector<BSONObj>{fromjson("{$match: {_id: {$mod: [2, 0]}}}")}}}});
+
+    // Mock out the foreign collection.
+    std::deque<DocumentSource::GetNextResult> mockForeignContents{Document{{"_id", 1}},
+                                                                  Document{{"_id", 2}}};
+    expCtx->mongoProcessInterface =
+        std::make_shared<MockMongoInterface>(std::move(mockForeignContents));
+
+    const auto localMock = DocumentSourceMock::createForTest({Document{{"_id"_sd, "local"_sd}}});
+    auto bson = BSON("$unionWith" << BSON(
+                         "coll" << viewNsToUnionWith.coll() << "pipeline"
+                                << BSON_ARRAY(fromjson(
+                                       "{$set: {originalId: '$_id', _id: {$add: [1, '$_id']}}}"))));
+    auto unionWith = DocumentSourceUnionWith::createFromBson(bson.firstElement(), expCtx);
+    unionWith->setSource(localMock.get());
+
+    auto result = unionWith->getNext();
+    ASSERT_TRUE(result.isAdvanced());
+    ASSERT_DOCUMENT_EQ(result.getDocument(), (Document{{"_id"_sd, "local"_sd}}));
+
+    result = unionWith->getNext();
+    ASSERT_TRUE(result.isAdvanced());
+    // Assert we get the document that originally had an even _id. Note this proves that the view
+    // definition was _prepended_ on the pipeline, which is important.
+    ASSERT_DOCUMENT_EQ(result.getDocument(), (Document{{"_id"_sd, 3}, {"originalId"_sd, 2}}));
+
+    ASSERT_TRUE(unionWith->getNext().isEOF());
+}
+
 }  // namespace
 }  // namespace mongo