SERVER-43013 Translate Map Reduce to Agg Pipeline

16 files changed, 621 insertions, 111 deletions

diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript
index 689859e0d3c..a3ae15cb440 100644
--- a/src/mongo/db/commands/SConscript
+++ b/src/mongo/db/commands/SConscript
@@ -353,7 +353,6 @@ env.Library(
         "dbhash.cpp",
         "driverHelpers.cpp",
         "haystack.cpp",
-        "map_reduce_agg.cpp",
         "map_reduce_command.cpp",
         "map_reduce_finish_command.cpp",
         "mr.cpp",
@@ -397,7 +396,6 @@ env.Library(
         '$BUILD_DIR/mongo/db/index_builds_coordinator_interface',
         '$BUILD_DIR/mongo/db/pipeline/mongo_process_interface',
         '$BUILD_DIR/mongo/db/pipeline/pipeline',
-        '$BUILD_DIR/mongo/db/query/map_reduce_output_format',
         '$BUILD_DIR/mongo/db/repl/dbcheck',
         '$BUILD_DIR/mongo/db/repl/oplog',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
@@ -409,7 +407,7 @@ env.Library(
         '$BUILD_DIR/mongo/util/net/ssl_manager',
         'core',
         'kill_common',
-        'map_reduce_parser',
+        'map_reduce_agg',
         'mongod_fcv',
         'mongod_fsync',
         'profile_common',
@@ -503,6 +501,21 @@ env.Library(
     ]
 )
 
+env.Library(
+    target='map_reduce_agg',
+    source=[
+        'map_reduce_agg.cpp',
+    ],
+    LIBDEPS=[
+        '$BUILD_DIR/mongo/db/db_raii',
+        '$BUILD_DIR/mongo/idl/idl_parser',
+        '$BUILD_DIR/mongo/db/pipeline/mongo_process_interface',
+        '$BUILD_DIR/mongo/db/pipeline/pipeline',
+        '$BUILD_DIR/mongo/db/query/map_reduce_output_format',
+        'map_reduce_parser'
+    ]
+)
+
 if has_option('use-cpu-profiler'):
     profEnv = env.Clone()
     profEnv.InjectThirdParty('gperftools')
@@ -521,14 +534,14 @@ if has_option('use-cpu-profiler'):
     )
 
 env.CppUnitTest(
-    target="map_reduce_parse_test",
+    target="map_reduce_agg_test",
     source=[
+        "map_reduce_agg_test.cpp",
         "map_reduce_parse_test.cpp",
     ],
     LIBDEPS=[
-        '$BUILD_DIR/mongo/db/write_concern_options',
-        '$BUILD_DIR/mongo/idl/idl_parser',
-        'map_reduce_parser'
+        '$BUILD_DIR/mongo/db/query/query_test_service_context',
+        'map_reduce_agg',
     ]
 )
 
diff --git a/src/mongo/db/commands/map_reduce_agg.cpp b/src/mongo/db/commands/map_reduce_agg.cpp
index 6dd26ee46c3..a1866c9372c 100644
--- a/src/mongo/db/commands/map_reduce_agg.cpp
+++ b/src/mongo/db/commands/map_reduce_agg.cpp
@@ -29,32 +29,170 @@
 
 #include "mongo/platform/basic.h"
 
+#include <boost/intrusive_ptr.hpp>
+#include <boost/optional.hpp>
+#include <cstdint>
+#include <string>
+#include <utility>
+
+#include "mongo/base/string_data.h"
 #include "mongo/bson/bsonobj.h"
 #include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/map_reduce_agg.h"
-#include "mongo/db/commands/map_reduce_gen.h"
+#include "mongo/db/commands/map_reduce_javascript_code.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/pipeline/document_source.h"
-#include "mongo/db/pipeline/expression_context.h"
-#include "mongo/db/pipeline/pipeline.h"
+#include "mongo/db/pipeline/document_source_group.h"
+#include "mongo/db/pipeline/document_source_limit.h"
+#include "mongo/db/pipeline/document_source_match.h"
+#include "mongo/db/pipeline/document_source_merge.h"
+#include "mongo/db/pipeline/document_source_out.h"
+#include "mongo/db/pipeline/document_source_project.h"
+#include "mongo/db/pipeline/document_source_single_document_transformation.h"
+#include "mongo/db/pipeline/document_source_sort.h"
+#include "mongo/db/pipeline/document_source_unwind.h"
+#include "mongo/db/pipeline/expression.h"
+#include "mongo/db/pipeline/expression_javascript.h"
+#include "mongo/db/pipeline/parsed_aggregation_projection_node.h"
+#include "mongo/db/pipeline/parsed_inclusion_projection.h"
 #include "mongo/db/pipeline/pipeline_d.h"
+#include "mongo/db/pipeline/value.h"
 #include "mongo/db/query/map_reduce_output_format.h"
+#include "mongo/db/query/util/make_data_structure.h"
 #include "mongo/util/intrusive_counter.h"
 
-namespace mongo {
+namespace mongo::map_reduce_agg {
 
 namespace {
 
-std::unique_ptr<Pipeline, PipelineDeleter> translateFromMR(
-    MapReduce parsedMr, boost::intrusive_ptr<ExpressionContext> expCtx) {
-    return uassertStatusOK(Pipeline::create({}, expCtx));
+using namespace std::string_literals;
+
+auto translateSort(boost::intrusive_ptr<ExpressionContext> expCtx,
+                   const BSONObj& sort,
+                   const boost::optional<std::int64_t>& limit) {
+    return DocumentSourceSort::create(expCtx, sort, limit.get_value_or(-1));
+}
+
+auto translateMap(boost::intrusive_ptr<ExpressionContext> expCtx, std::string code) {
+    auto emitExpression = ExpressionInternalJsEmit::create(
+        expCtx, ExpressionFieldPath::parse(expCtx, "$$ROOT", expCtx->variablesParseState), code);
+    auto node = std::make_unique<parsed_aggregation_projection::InclusionNode>(
+        ProjectionPolicies{ProjectionPolicies::DefaultIdPolicy::kExcludeId});
+    node->addExpressionForPath(FieldPath{"emits"s}, std::move(emitExpression));
+    auto inclusion = std::unique_ptr<TransformerInterface>{
+        std::make_unique<parsed_aggregation_projection::ParsedInclusionProjection>(
+            expCtx,
+            ProjectionPolicies{ProjectionPolicies::DefaultIdPolicy::kExcludeId},
+            std::move(node))};
+    return make_intrusive<DocumentSourceSingleDocumentTransformation>(
+        expCtx, std::move(inclusion), DocumentSourceProject::kStageName, false);
+}
+
+auto translateReduce(boost::intrusive_ptr<ExpressionContext> expCtx, std::string code) {
+    auto accumulatorArguments = ExpressionObject::create(
+        expCtx,
+        make_vector<std::pair<std::string, boost::intrusive_ptr<Expression>>>(
+            std::pair{"data"s,
+                      ExpressionFieldPath::parse(expCtx, "$emits", expCtx->variablesParseState)},
+            std::pair{"eval"s, ExpressionConstant::create(expCtx, Value{code})}));
+    auto jsReduce = AccumulationStatement{
+        "value",
+        std::move(accumulatorArguments),
+        AccumulationStatement::getFactory(AccumulatorInternalJsReduce::kAccumulatorName)};
+    auto groupExpr = ExpressionFieldPath::parse(expCtx, "$emits.k", expCtx->variablesParseState);
+    return DocumentSourceGroup::create(expCtx,
+                                       std::move(groupExpr),
+                                       make_vector<AccumulationStatement>(std::move(jsReduce)),
+                                       boost::none);
+}
+
+auto translateFinalize(boost::intrusive_ptr<ExpressionContext> expCtx, std::string code) {
+    auto jsExpression = ExpressionInternalJs::create(
+        expCtx,
+        ExpressionArray::create(
+            expCtx,
+            make_vector<boost::intrusive_ptr<Expression>>(
+                ExpressionFieldPath::parse(expCtx, "$_id", expCtx->variablesParseState),
+                ExpressionFieldPath::parse(expCtx, "$value", expCtx->variablesParseState))),
+        code);
+    auto node = std::make_unique<parsed_aggregation_projection::InclusionNode>(
+        ProjectionPolicies{ProjectionPolicies::DefaultIdPolicy::kExcludeId});
+    node->addExpressionForPath(FieldPath{"value"s}, std::move(jsExpression));
+    auto inclusion = std::unique_ptr<TransformerInterface>{
+        std::make_unique<parsed_aggregation_projection::ParsedInclusionProjection>(
+            expCtx,
+            ProjectionPolicies{ProjectionPolicies::DefaultIdPolicy::kExcludeId},
+            std::move(node))};
+    return make_intrusive<DocumentSourceSingleDocumentTransformation>(
+        expCtx, std::move(inclusion), DocumentSourceProject::kStageName, false);
+}
+
+auto translateOutReplace(boost::intrusive_ptr<ExpressionContext> expCtx,
+                         const StringData inputDatabase,
+                         NamespaceString targetNss) {
+    uassert(31278,
+            "MapReduce must output to the database belonging to its input collection - Input: "s +
+                inputDatabase + "Output: " + targetNss.db(),
+            inputDatabase == targetNss.db());
+    return DocumentSourceOut::create(std::move(targetNss), expCtx);
+}
+
+auto translateOutMerge(boost::intrusive_ptr<ExpressionContext> expCtx, NamespaceString targetNss) {
+    return DocumentSourceMerge::create(targetNss,
+                                       expCtx,
+                                       MergeWhenMatchedModeEnum::kReplace,
+                                       MergeWhenNotMatchedModeEnum::kInsert,
+                                       boost::none,  // Let variables
+                                       boost::none,  // pipeline
+                                       std::set<FieldPath>{FieldPath("_id"s)},
+                                       boost::none);  // targetCollectionVersion
+}
+
+auto translateOutReduce(boost::intrusive_ptr<ExpressionContext> expCtx,
+                        NamespaceString targetNss,
+                        std::string code) {
+    // Because of communication for sharding, $merge must hold on to a serializable BSON object
+    // at the moment so we reparse here.
+    auto reduceObj = BSON("args" << BSON_ARRAY("$value"
+                                               << "$$new.value")
+                                 << "eval" << code);
+
+    auto finalProjectSpec =
+        BSON(DocumentSourceProject::kStageName
+             << BSON("value" << BSON(ExpressionInternalJs::kExpressionName << reduceObj)));
+    auto pipelineSpec = boost::make_optional(std::vector<BSONObj>{finalProjectSpec});
+    return DocumentSourceMerge::create(targetNss,
+                                       expCtx,
+                                       MergeWhenMatchedModeEnum::kPipeline,
+                                       MergeWhenNotMatchedModeEnum::kInsert,
+                                       boost::none,  // Let variables
+                                       pipelineSpec,
+                                       std::set<FieldPath>{FieldPath("_id"s)},
+                                       boost::none);  // targetCollectionVersion
+}
+
+auto translateOut(boost::intrusive_ptr<ExpressionContext> expCtx,
+                  const OutputType outputType,
+                  const StringData inputDatabase,
+                  NamespaceString targetNss,
+                  std::string reduceCode) {
+    switch (outputType) {
+        case OutputType::Replace:
+            return boost::make_optional(translateOutReplace(expCtx, inputDatabase, targetNss));
+        case OutputType::Merge:
+            return boost::make_optional(translateOutMerge(expCtx, targetNss));
+        case OutputType::Reduce:
+            return boost::make_optional(translateOutReduce(expCtx, targetNss, reduceCode));
+        case OutputType::InMemory:;
+    }
+    return boost::optional<boost::intrusive_ptr<mongo::DocumentSource>>{};
 }
 
 auto makeExpressionContext(OperationContext* opCtx, const MapReduce& parsedMr) {
-    // AutoGetCollectionForReadCommand will throw if the sharding version for this connection is out
-    // of date.
+    // AutoGetCollectionForReadCommand will throw if the sharding version for this connection is
+    // out of date.
     AutoGetCollectionForReadCommand ctx(
         opCtx, parsedMr.getNamespace(), AutoGetCollection::ViewMode::kViewsPermitted);
     uassert(ErrorCodes::CommandNotSupportedOnView,
@@ -68,9 +206,9 @@ auto makeExpressionContext(OperationContext* opCtx, const MapReduce& parsedMr) {
     auto uuid =
         ctx.getCollection() ? boost::make_optional(ctx.getCollection()->uuid()) : boost::none;
 
-    // Manually build an ExpressionContext with the desired options for the translated aggregation.
-    // The one option worth noting here is allowDiskUse, which is required to allow the $group stage
-    // of the translated pipeline to spill to disk.
+    // Manually build an ExpressionContext with the desired options for the translated
+    // aggregation. The one option worth noting here is allowDiskUse, which is required to allow
+    // the $group stage of the translated pipeline to spill to disk.
     return make_intrusive<ExpressionContext>(
         opCtx,
         boost::none,    // explain
@@ -90,7 +228,6 @@ auto makeExpressionContext(OperationContext* opCtx, const MapReduce& parsedMr) {
 
 }  // namespace
 
-// Update MapReduceFormatter
 bool runAggregationMapReduce(OperationContext* opCtx,
                              const std::string& dbname,
                              const BSONObj& cmd,
@@ -124,4 +261,31 @@ bool runAggregationMapReduce(OperationContext* opCtx,
     return true;
 }
 
-}  // namespace mongo
+std::unique_ptr<Pipeline, PipelineDeleter> translateFromMR(
+    MapReduce parsedMr, boost::intrusive_ptr<ExpressionContext> expCtx) {
+    // TODO: It would be good to figure out what kind of errors this would produce in the Status.
+    // It would be better not to produce something incomprehensible out of an internal translation.
+    return uassertStatusOK(Pipeline::create(
+        makeFlattenedList<boost::intrusive_ptr<DocumentSource>>(
+            parsedMr.getQuery().map(
+                [&](auto&& query) { return DocumentSourceMatch::create(query, expCtx); }),
+            parsedMr.getSort().map(
+                [&](auto&& sort) { return translateSort(expCtx, sort, parsedMr.getLimit()); }),
+            translateMap(expCtx, parsedMr.getMap().getCode()),
+            DocumentSourceUnwind::create(expCtx, "emits", false, boost::none),
+            translateReduce(expCtx, parsedMr.getReduce().getCode()),
+            parsedMr.getFinalize().map([&](auto&& finalize) {
+                return translateFinalize(expCtx, parsedMr.getFinalize()->getCode());
+            }),
+            translateOut(expCtx,
+                         parsedMr.getOutOptions().getOutputType(),
+                         parsedMr.getNamespace().db(),
+                         NamespaceString{parsedMr.getOutOptions().getDatabaseName()
+                                             ? *parsedMr.getOutOptions().getDatabaseName()
+                                             : parsedMr.getNamespace().db(),
+                                         parsedMr.getOutOptions().getCollectionName()},
+                         parsedMr.getReduce().getCode())),
+        expCtx));
+}
+
+}  // namespace mongo::map_reduce_agg
diff --git a/src/mongo/db/commands/map_reduce_agg.h b/src/mongo/db/commands/map_reduce_agg.h
index 8ac5d0ead5f..573119c0393 100644
--- a/src/mongo/db/commands/map_reduce_agg.h
+++ b/src/mongo/db/commands/map_reduce_agg.h
@@ -27,7 +27,19 @@
  *    it in the license file.
  */
 
-namespace mongo {
+#pragma once
+
+#include <memory>
+#include <string>
+
+#include "mongo/bson/bsonobj.h"
+#include "mongo/bson/bsonobjbuilder.h"
+#include "mongo/db/commands/map_reduce_gen.h"
+#include "mongo/db/operation_context.h"
+#include "mongo/db/pipeline/expression_context.h"
+#include "mongo/db/pipeline/pipeline.h"
+
+namespace mongo::map_reduce_agg {
 
 bool runAggregationMapReduce(OperationContext* opCtx,
                              const std::string& dbname,
@@ -35,4 +47,7 @@ bool runAggregationMapReduce(OperationContext* opCtx,
                              std::string& errmsg,
                              BSONObjBuilder& result);
 
-}  // namespace mongo
+std::unique_ptr<Pipeline, PipelineDeleter> translateFromMR(
+    MapReduce parsedMr, boost::intrusive_ptr<ExpressionContext> expCtx);
+
+}  // namespace mongo::map_reduce_agg
diff --git a/src/mongo/db/commands/map_reduce_agg_test.cpp b/src/mongo/db/commands/map_reduce_agg_test.cpp
new file mode 100644
index 00000000000..17b5f2520ba
--- /dev/null
+++ b/src/mongo/db/commands/map_reduce_agg_test.cpp
@@ -0,0 +1,215 @@
+/**
+ *    Copyright (C) 2019-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    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
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    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 Server Side 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.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include <boost/optional.hpp>
+#include <string>
+
+#include "mongo/base/string_data.h"
+#include "mongo/bson/bsonobj.h"
+#include "mongo/db/commands/map_reduce_agg.h"
+#include "mongo/db/pipeline/document_source_group.h"
+#include "mongo/db/pipeline/document_source_match.h"
+#include "mongo/db/pipeline/document_source_merge.h"
+#include "mongo/db/pipeline/document_source_out.h"
+#include "mongo/db/pipeline/document_source_single_document_transformation.h"
+#include "mongo/db/pipeline/document_source_sort.h"
+#include "mongo/db/pipeline/document_source_unwind.h"
+#include "mongo/db/pipeline/expression_context_for_test.h"
+#include "mongo/unittest/unittest.h"
+
+namespace mongo {
+namespace {
+
+using namespace std::string_literals;
+using namespace map_reduce_agg;
+
+// The translator treats Javascript objects as black boxes so there's no need for realistic examples
+// here.
+constexpr auto initJavascript = "init!"_sd;
+constexpr auto mapJavascript = "map!"_sd;
+constexpr auto reduceJavascript = "reduce!"_sd;
+constexpr auto finalizeJavascript = "finalize!"_sd;
+
+TEST(MapReduceAggTest, testBasicTranslate) {
+    auto nss = NamespaceString{"db", "coll"};
+    auto mr = MapReduce{nss,
+                        MapReduceJavascriptCode{mapJavascript.toString()},
+                        MapReduceJavascriptCode{reduceJavascript.toString()},
+                        MapReduceOutOptions{boost::none, "", OutputType::InMemory, false}};
+    boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest(nss));
+    auto pipeline = translateFromMR(mr, expCtx);
+    auto& sources = pipeline->getSources();
+    ASSERT_EQ(3u, sources.size());
+    auto iter = sources.begin();
+    ASSERT(typeid(DocumentSourceSingleDocumentTransformation) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceUnwind) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceGroup) == typeid(**iter));
+}
+
+TEST(MapReduceAggTest, testSortWithoutLimit) {
+    auto nss = NamespaceString{"db", "coll"};
+    auto mr = MapReduce{nss,
+                        MapReduceJavascriptCode{mapJavascript.toString()},
+                        MapReduceJavascriptCode{reduceJavascript.toString()},
+                        MapReduceOutOptions{boost::none, "", OutputType::InMemory, false}};
+    mr.setSort(BSON("foo" << 1));
+    boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest(nss));
+    auto pipeline = translateFromMR(mr, expCtx);
+    auto& sources = pipeline->getSources();
+    ASSERT_EQ(4u, sources.size());
+    auto iter = sources.begin();
+    ASSERT(typeid(DocumentSourceSort) == typeid(**iter));
+    auto& sort = dynamic_cast<DocumentSourceSort&>(**iter++);
+    ASSERT_EQ(-1ll, sort.getLimit());
+    ASSERT(typeid(DocumentSourceSingleDocumentTransformation) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceUnwind) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceGroup) == typeid(**iter));
+}
+
+TEST(MapReduceAggTest, testSortWithLimit) {
+    auto nss = NamespaceString{"db", "coll"};
+    auto mr = MapReduce{nss,
+                        MapReduceJavascriptCode{mapJavascript.toString()},
+                        MapReduceJavascriptCode{reduceJavascript.toString()},
+                        MapReduceOutOptions{boost::none, "", OutputType::InMemory, false}};
+    mr.setSort(BSON("foo" << 1));
+    mr.setLimit(23);
+    boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest(nss));
+    auto pipeline = translateFromMR(mr, expCtx);
+    auto& sources = pipeline->getSources();
+    ASSERT_EQ(4u, sources.size());
+    auto iter = sources.begin();
+    ASSERT(typeid(DocumentSourceSort) == typeid(**iter));
+    auto& sort = dynamic_cast<DocumentSourceSort&>(**iter++);
+    ASSERT_EQ(23ll, sort.getLimit());
+    ASSERT(typeid(DocumentSourceSingleDocumentTransformation) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceUnwind) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceGroup) == typeid(**iter));
+}
+
+TEST(MapReduceAggTest, testFeatureLadenTranslate) {
+    auto nss = NamespaceString{"db", "coll"};
+    auto mr = MapReduce{
+        nss,
+        MapReduceJavascriptCode{mapJavascript.toString()},
+        MapReduceJavascriptCode{reduceJavascript.toString()},
+        MapReduceOutOptions{boost::make_optional("db"s), "coll2", OutputType::Replace, false}};
+    mr.setSort(BSON("foo" << 1));
+    mr.setQuery(BSON("foo"
+                     << "fooval"));
+    mr.setFinalize(boost::make_optional(MapReduceJavascriptCode{finalizeJavascript.toString()}));
+    boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest(nss));
+    auto pipeline = translateFromMR(mr, expCtx);
+    auto& sources = pipeline->getSources();
+    ASSERT_EQ(7u, sources.size());
+    auto iter = sources.begin();
+    ASSERT(typeid(DocumentSourceMatch) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceSort) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceSingleDocumentTransformation) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceUnwind) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceGroup) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceSingleDocumentTransformation) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceOut) == typeid(**iter));
+}
+
+TEST(MapReduceAggTest, testOutMergeTranslate) {
+    auto nss = NamespaceString{"db", "coll"};
+    auto mr = MapReduce{
+        nss,
+        MapReduceJavascriptCode{mapJavascript.toString()},
+        MapReduceJavascriptCode{reduceJavascript.toString()},
+        MapReduceOutOptions{boost::make_optional("db"s), "coll2", OutputType::Merge, false}};
+    boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest(nss));
+    auto pipeline = translateFromMR(mr, expCtx);
+    auto& sources = pipeline->getSources();
+    ASSERT_EQ(sources.size(), 4u);
+    auto iter = sources.begin();
+    ASSERT(typeid(DocumentSourceSingleDocumentTransformation) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceUnwind) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceGroup) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceMerge) == typeid(**iter));
+    auto& merge = dynamic_cast<DocumentSourceMerge&>(**iter);
+    ASSERT_FALSE(merge.getPipeline());
+}
+
+TEST(MapReduceAggTest, testOutReduceTranslate) {
+    auto nss = NamespaceString{"db", "coll"};
+    auto mr = MapReduce{
+        nss,
+        MapReduceJavascriptCode{mapJavascript.toString()},
+        MapReduceJavascriptCode{reduceJavascript.toString()},
+        MapReduceOutOptions{boost::make_optional("db"s), "coll2", OutputType::Reduce, false}};
+    boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest(nss));
+    auto pipeline = translateFromMR(mr, expCtx);
+    auto& sources = pipeline->getSources();
+    ASSERT_EQ(sources.size(), 4u);
+    auto iter = sources.begin();
+    ASSERT(typeid(DocumentSourceSingleDocumentTransformation) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceUnwind) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceGroup) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceMerge) == typeid(**iter));
+    auto& merge = dynamic_cast<DocumentSourceMerge&>(**iter);
+    auto subpipeline = merge.getPipeline();
+    ASSERT_EQ(1u, subpipeline->size());
+    ASSERT_EQ("$project"s, (*subpipeline)[0].firstElement().fieldName());
+}
+
+TEST(MapReduceAggTest, testOutDifferentDBFails) {
+    auto nss = NamespaceString{"db", "coll"};
+    auto mr = MapReduce{
+        nss,
+        MapReduceJavascriptCode{mapJavascript.toString()},
+        MapReduceJavascriptCode{reduceJavascript.toString()},
+        MapReduceOutOptions{boost::make_optional("db2"s), "coll2", OutputType::Replace, false}};
+    boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest(nss));
+    ASSERT_THROWS_CODE(translateFromMR(mr, expCtx), AssertionException, 31278);
+}
+
+TEST(MapReduceAggTest, testOutSameCollection) {
+    auto nss = NamespaceString{"db", "coll"};
+    auto mr = MapReduce{
+        nss,
+        MapReduceJavascriptCode{mapJavascript.toString()},
+        MapReduceJavascriptCode{reduceJavascript.toString()},
+        MapReduceOutOptions{boost::make_optional("db"s), "coll", OutputType::Replace, false}};
+    boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest(nss));
+    auto pipeline = translateFromMR(mr, expCtx);
+    auto& sources = pipeline->getSources();
+    ASSERT_EQ(sources.size(), 4u);
+    auto iter = sources.begin();
+    ASSERT(typeid(DocumentSourceSingleDocumentTransformation) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceUnwind) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceGroup) == typeid(**iter++));
+    ASSERT(typeid(DocumentSourceOut) == typeid(**iter));
+}
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/db/commands/map_reduce_command.cpp b/src/mongo/db/commands/map_reduce_command.cpp
index 0da1795f799..34975cc550f 100644
--- a/src/mongo/db/commands/map_reduce_command.cpp
+++ b/src/mongo/db/commands/map_reduce_command.cpp
@@ -70,7 +70,7 @@ private:
                   std::string& errmsg,
                   BSONObjBuilder& result) final {
         if (getTestCommandsEnabled() && internalQueryUseAggMapReduce.load()) {
-            return runAggregationMapReduce(opCtx, dbname, cmd, errmsg, result);
+            return map_reduce_agg::runAggregationMapReduce(opCtx, dbname, cmd, errmsg, result);
         }
         return mr::runMapReduce(opCtx, dbname, cmd, errmsg, result);
     }
diff --git a/src/mongo/db/commands/map_reduce_javascript_code.h b/src/mongo/db/commands/map_reduce_javascript_code.h
index 0ab0752ab45..2fcf2e27b6b 100644
--- a/src/mongo/db/commands/map_reduce_javascript_code.h
+++ b/src/mongo/db/commands/map_reduce_javascript_code.h
@@ -31,6 +31,7 @@
 
 #include <string>
 
+#include "mongo/base/string_data.h"
 #include "mongo/bson/bsonelement.h"
 #include "mongo/bson/bsonobj.h"
 
@@ -43,22 +44,24 @@ class MapReduceJavascriptCode {
 public:
     static MapReduceJavascriptCode parseFromBSON(const BSONElement& element) {
         uassert(ErrorCodes::BadValue,
-                "'scope' must be a string, code or 'code with scope'",
-                element.type() == String || element.type() == Code || element.type() == CodeWScope);
-        return MapReduceJavascriptCode(element);
+                "'scope' must be of string or code type",
+                element.type() == String || element.type() == Code);
+        return MapReduceJavascriptCode(element._asCode());
     }
 
     MapReduceJavascriptCode() = default;
-    MapReduceJavascriptCode(const BSONElement& element) : code(element.wrap()) {}
+    MapReduceJavascriptCode(std::string&& code) : code(code) {}
 
     void serializeToBSON(StringData fieldName, BSONObjBuilder* builder) const {
-        (*builder) << fieldName << code[0];
+        (*builder) << fieldName << code;
+    }
+
+    auto getCode() const {
+        return code;
     }
 
 private:
-    // We have to save a local copy of the BSON for reserialization since it's difficult to rebuild
-    // once it has been turned into a code type.
-    BSONObj code;
+    std::string code;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/document_source_group_test.cpp b/src/mongo/db/pipeline/document_source_group_test.cpp
index 42411f52660..05f7d23ff2c 100644
--- a/src/mongo/db/pipeline/document_source_group_test.cpp
+++ b/src/mongo/db/pipeline/document_source_group_test.cpp
@@ -210,20 +210,7 @@ TEST_F(DocumentSourceGroupTest, ShouldReportMultipleFieldGroupKeysAsARename) {
     VariablesParseState vps = expCtx->variablesParseState;
     auto x = ExpressionFieldPath::parse(expCtx, "$x", vps);
     auto y = ExpressionFieldPath::parse(expCtx, "$y", vps);
-    auto groupByExpression = [&]() {
-        std::vector<boost::intrusive_ptr<Expression>> children;
-        std::vector<std::pair<std::string, boost::intrusive_ptr<Expression>&>> expressions;
-        auto doc = std::vector<std::pair<std::string, boost::intrusive_ptr<Expression>>>{{"x", x},
-                                                                                         {"y", y}};
-        for (auto& [unused, expression] : doc)
-            children.push_back(std::move(expression));
-        std::vector<boost::intrusive_ptr<Expression>>::size_type index = 0;
-        for (auto& [fieldName, unused] : doc) {
-            expressions.emplace_back(fieldName, children[index]);
-            ++index;
-        }
-        return ExpressionObject::create(expCtx, std::move(children), std::move(expressions));
-    }();
+    auto groupByExpression = ExpressionObject::create(expCtx, {{"x", x}, {"y", y}});
     auto group = DocumentSourceGroup::create(expCtx, groupByExpression, {});
     auto modifiedPathsRet = group->getModifiedPaths();
     ASSERT(modifiedPathsRet.type == DocumentSource::GetModPathsReturn::Type::kAllExcept);
diff --git a/src/mongo/db/pipeline/document_source_merge.h b/src/mongo/db/pipeline/document_source_merge.h
index f7889528930..05b7fe7606f 100644
--- a/src/mongo/db/pipeline/document_source_merge.h
+++ b/src/mongo/db/pipeline/document_source_merge.h
@@ -143,6 +143,10 @@ public:
     static boost::intrusive_ptr<DocumentSource> createFromBson(
         BSONElement elem, const boost::intrusive_ptr<ExpressionContext>& pExpCtx);
 
+    auto getPipeline() const {
+        return _pipeline;
+    }
+
 private:
     /**
      * Builds a new $merge stage which will merge all documents into 'outputNs'. If
diff --git a/src/mongo/db/pipeline/expression.cpp b/src/mongo/db/pipeline/expression.cpp
index 62b2d1e5018..9d21614c1ed 100644
--- a/src/mongo/db/pipeline/expression.cpp
+++ b/src/mongo/db/pipeline/expression.cpp
@@ -93,7 +93,7 @@ intrusive_ptr<Expression> Expression::parseObject(
     BSONObj obj,
     const VariablesParseState& vps) {
     if (obj.isEmpty()) {
-        return ExpressionObject::create(expCtx, {}, {});
+        return ExpressionObject::create(expCtx, {});
     }
 
     if (obj.firstElementFieldName()[0] == '$') {
@@ -1882,11 +1882,23 @@ ExpressionObject::ExpressionObject(const boost::intrusive_ptr<ExpressionContext>
                                    vector<pair<string, intrusive_ptr<Expression>&>>&& expressions)
     : Expression(expCtx, std::move(_children)), _expressions(std::move(expressions)) {}
 
-intrusive_ptr<ExpressionObject> ExpressionObject::create(
+boost::intrusive_ptr<ExpressionObject> ExpressionObject::create(
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
-    std::vector<boost::intrusive_ptr<Expression>> _children,
-    vector<pair<string, intrusive_ptr<Expression>&>>&& expressions) {
-    return new ExpressionObject(expCtx, std::move(_children), std::move(expressions));
+    std::vector<std::pair<std::string, boost::intrusive_ptr<Expression>>>&&
+        expressionsWithChildrenInPlace) {
+    std::vector<boost::intrusive_ptr<Expression>> children;
+    std::vector<std::pair<std::string, boost::intrusive_ptr<Expression>&>> expressions;
+    for (auto& [unused, expression] : expressionsWithChildrenInPlace)
+        // These 'push_back's must complete before we insert references to the 'children' vector
+        // into the 'expressions' vector since 'push_back' invalidates references.
+        children.push_back(std::move(expression));
+    std::vector<boost::intrusive_ptr<Expression>>::size_type index = 0;
+    for (auto& [fieldName, unused] : expressionsWithChildrenInPlace) {
+        expressions.emplace_back(fieldName, children[index]);
+        ++index;
+    }
+    // It is safe to 'std::move' 'children' since the standard guarantees the references are stable.
+    return new ExpressionObject(expCtx, std::move(children), std::move(expressions));
 }
 
 intrusive_ptr<ExpressionObject> ExpressionObject::parse(
diff --git a/src/mongo/db/pipeline/expression.h b/src/mongo/db/pipeline/expression.h
index 8c8bbc54ae9..85872dc675b 100644
--- a/src/mongo/db/pipeline/expression.h
+++ b/src/mongo/db/pipeline/expression.h
@@ -822,8 +822,21 @@ public:
     explicit ExpressionArray(const boost::intrusive_ptr<ExpressionContext>& expCtx)
         : ExpressionVariadic<ExpressionArray>(expCtx) {}
 
+    ExpressionArray(const boost::intrusive_ptr<ExpressionContext>& expCtx,
+                    std::vector<boost::intrusive_ptr<Expression>>&& children)
+        : ExpressionVariadic<ExpressionArray>(expCtx) {
+        _children = std::move(children);
+    }
+
     Value evaluate(const Document& root, Variables* variables) const final;
     Value serialize(bool explain) const final;
+
+    static boost::intrusive_ptr<ExpressionArray> create(
+        const boost::intrusive_ptr<ExpressionContext>& expCtx,
+        std::vector<boost::intrusive_ptr<Expression>>&& children) {
+        return make_intrusive<ExpressionArray>(expCtx, std::move(children));
+    }
+
     boost::intrusive_ptr<Expression> optimize() final;
     const char* getOpName() const final;
 
@@ -1788,8 +1801,8 @@ public:
 
     static boost::intrusive_ptr<ExpressionObject> create(
         const boost::intrusive_ptr<ExpressionContext>& expCtx,
-        std::vector<boost::intrusive_ptr<Expression>> children,
-        std::vector<std::pair<std::string, boost::intrusive_ptr<Expression>&>>&& expressions);
+        std::vector<std::pair<std::string, boost::intrusive_ptr<Expression>>>&&
+            expressionsWithChildrenInPlace);
 
     /**
      * Parses and constructs an ExpressionObject from 'obj'.
diff --git a/src/mongo/db/pipeline/expression_javascript.h b/src/mongo/db/pipeline/expression_javascript.h
index bf39dcf1227..1a3e973e952 100644
--- a/src/mongo/db/pipeline/expression_javascript.h
+++ b/src/mongo/db/pipeline/expression_javascript.h
@@ -49,6 +49,13 @@ public:
         BSONElement expr,
         const VariablesParseState& vps);
 
+    static boost::intrusive_ptr<ExpressionInternalJsEmit> create(
+        const boost::intrusive_ptr<ExpressionContext>& expCtx,
+        boost::intrusive_ptr<Expression> thisRef,
+        std::string funcSourceString) {
+        return new ExpressionInternalJsEmit{expCtx, thisRef, std::move(funcSourceString)};
+    }
+
     Value evaluate(const Document& root, Variables* variables) const final;
 
     Value serialize(bool explain) const final;
@@ -81,6 +88,13 @@ public:
         BSONElement expr,
         const VariablesParseState& vps);
 
+    static boost::intrusive_ptr<ExpressionInternalJs> create(
+        const boost::intrusive_ptr<ExpressionContext>& expCtx,
+        boost::intrusive_ptr<Expression> passedArgs,
+        std::string funcSourceString) {
+        return new ExpressionInternalJs{expCtx, passedArgs, std::move(funcSourceString)};
+    }
+
     Value evaluate(const Document& root, Variables* variables) const final;
 
     Value serialize(bool explain) const final;
@@ -89,6 +103,8 @@ public:
         return visitor->visit(this);
     }
 
+    static constexpr auto kExpressionName = "$_internalJs"_sd;
+
 private:
     ExpressionInternalJs(const boost::intrusive_ptr<ExpressionContext>& expCtx,
                          boost::intrusive_ptr<Expression> passedArgs,
@@ -97,6 +113,5 @@ private:
 
     const boost::intrusive_ptr<Expression>& _passedArgs;
     std::string _funcSource;
-    static constexpr auto kExpressionName = "$_internalJs"_sd;
 };
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/expression_test.cpp b/src/mongo/db/pipeline/expression_test.cpp
index 38b676067f4..9ed9e3a03b8 100644
--- a/src/mongo/db/pipeline/expression_test.cpp
+++ b/src/mongo/db/pipeline/expression_test.cpp
@@ -3367,32 +3367,9 @@ TEST(ParseObject, ShouldRejectExpressionAsTheSecondField) {
 // Evaluation.
 //
 
-namespace {
-/**
- * ExpressionObject builds two vectors within it's ::parse() method, one owning and one with names
- * and references to the former. Since the ::create() method bypasses this step, we have to mimic
- * the behavior here.
- */
-auto expressionObjectCreateHelper(
-    const boost::intrusive_ptr<ExpressionContext>& expCtx,
-    std::vector<std::pair<std::string, boost::intrusive_ptr<Expression>>>&&
-        expressionsWithChildrenInPlace) {
-    std::vector<boost::intrusive_ptr<Expression>> children;
-    std::vector<std::pair<std::string, boost::intrusive_ptr<Expression>&>> expressions;
-    for (auto& [unused, expression] : expressionsWithChildrenInPlace)
-        children.push_back(std::move(expression));
-    std::vector<boost::intrusive_ptr<Expression>>::size_type index = 0;
-    for (auto& [fieldName, unused] : expressionsWithChildrenInPlace) {
-        expressions.emplace_back(fieldName, children[index]);
-        ++index;
-    }
-    return ExpressionObject::create(expCtx, std::move(children), std::move(expressions));
-}
-}  // namespace
-
 TEST(ExpressionObjectEvaluate, EmptyObjectShouldEvaluateToEmptyDocument) {
     intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    auto object = expressionObjectCreateHelper(expCtx, {});
+    auto object = ExpressionObject::create(expCtx, {});
     ASSERT_VALUE_EQ(Value(Document()), object->evaluate(Document(), &(expCtx->variables)));
     ASSERT_VALUE_EQ(Value(Document()), object->evaluate(Document{{"a", 1}}, &(expCtx->variables)));
     ASSERT_VALUE_EQ(Value(Document()),
@@ -3402,7 +3379,7 @@ TEST(ExpressionObjectEvaluate, EmptyObjectShouldEvaluateToEmptyDocument) {
 TEST(ExpressionObjectEvaluate, ShouldEvaluateEachField) {
     intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
     auto object =
-        expressionObjectCreateHelper(expCtx, {{"a", makeConstant(1)}, {"b", makeConstant(5)}});
+        ExpressionObject::create(expCtx, {{"a", makeConstant(1)}, {"b", makeConstant(5)}});
     ASSERT_VALUE_EQ(Value(Document{{"a", 1}, {"b", 5}}),
                     object->evaluate(Document(), &(expCtx->variables)));
     ASSERT_VALUE_EQ(Value(Document{{"a", 1}, {"b", 5}}),
@@ -3413,10 +3390,10 @@ TEST(ExpressionObjectEvaluate, ShouldEvaluateEachField) {
 
 TEST(ExpressionObjectEvaluate, OrderOfFieldsInOutputShouldMatchOrderInSpecification) {
     intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    auto object = expressionObjectCreateHelper(expCtx,
-                                               {{"a", ExpressionFieldPath::create(expCtx, "a")},
-                                                {"b", ExpressionFieldPath::create(expCtx, "b")},
-                                                {"c", ExpressionFieldPath::create(expCtx, "c")}});
+    auto object = ExpressionObject::create(expCtx,
+                                           {{"a", ExpressionFieldPath::create(expCtx, "a")},
+                                            {"b", ExpressionFieldPath::create(expCtx, "b")},
+                                            {"c", ExpressionFieldPath::create(expCtx, "c")}});
     ASSERT_VALUE_EQ(
         Value(Document{{"a", "A"_sd}, {"b", "B"_sd}, {"c", "C"_sd}}),
         object->evaluate(Document{{"c", "C"_sd}, {"a", "A"_sd}, {"b", "B"_sd}, {"_id", "ID"_sd}},
@@ -3425,20 +3402,19 @@ TEST(ExpressionObjectEvaluate, OrderOfFieldsInOutputShouldMatchOrderInSpecificat
 
 TEST(ExpressionObjectEvaluate, ShouldRemoveFieldsThatHaveMissingValues) {
     intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    auto object =
-        expressionObjectCreateHelper(expCtx,
-                                     {{"a", ExpressionFieldPath::create(expCtx, "a.b")},
-                                      {"b", ExpressionFieldPath::create(expCtx, "missing")}});
+    auto object = ExpressionObject::create(expCtx,
+                                           {{"a", ExpressionFieldPath::create(expCtx, "a.b")},
+                                            {"b", ExpressionFieldPath::create(expCtx, "missing")}});
     ASSERT_VALUE_EQ(Value(Document{}), object->evaluate(Document(), &(expCtx->variables)));
     ASSERT_VALUE_EQ(Value(Document{}), object->evaluate(Document{{"a", 1}}, &(expCtx->variables)));
 }
 
 TEST(ExpressionObjectEvaluate, ShouldEvaluateFieldsWithinNestedObject) {
     intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    auto object = expressionObjectCreateHelper(
+    auto object = ExpressionObject::create(
         expCtx,
         {{"a",
-          expressionObjectCreateHelper(
+          ExpressionObject::create(
               expCtx,
               {{"b", makeConstant(1)}, {"c", ExpressionFieldPath::create(expCtx, "_id")}})}});
     ASSERT_VALUE_EQ(Value(Document{{"a", Document{{"b", 1}}}}),
@@ -3449,11 +3425,11 @@ TEST(ExpressionObjectEvaluate, ShouldEvaluateFieldsWithinNestedObject) {
 
 TEST(ExpressionObjectEvaluate, ShouldEvaluateToEmptyDocumentIfAllFieldsAreMissing) {
     intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    auto object = expressionObjectCreateHelper(
-        expCtx, {{"a", ExpressionFieldPath::create(expCtx, "missing")}});
+    auto object =
+        ExpressionObject::create(expCtx, {{"a", ExpressionFieldPath::create(expCtx, "missing")}});
     ASSERT_VALUE_EQ(Value(Document{}), object->evaluate(Document(), &(expCtx->variables)));
 
-    auto objectWithNestedObject = expressionObjectCreateHelper(expCtx, {{"nested", object}});
+    auto objectWithNestedObject = ExpressionObject::create(expCtx, {{"nested", object}});
     ASSERT_VALUE_EQ(Value(Document{{"nested", Document{}}}),
                     objectWithNestedObject->evaluate(Document(), &(expCtx->variables)));
 }
@@ -3464,7 +3440,7 @@ TEST(ExpressionObjectEvaluate, ShouldEvaluateToEmptyDocumentIfAllFieldsAreMissin
 
 TEST(ExpressionObjectDependencies, ConstantValuesShouldNotBeAddedToDependencies) {
     intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    auto object = expressionObjectCreateHelper(expCtx, {{"a", makeConstant(5)}});
+    auto object = ExpressionObject::create(expCtx, {{"a", makeConstant(5)}});
     DepsTracker deps;
     object->addDependencies(&deps);
     ASSERT_EQ(deps.fields.size(), 0UL);
@@ -3473,7 +3449,7 @@ TEST(ExpressionObjectDependencies, ConstantValuesShouldNotBeAddedToDependencies)
 TEST(ExpressionObjectDependencies, FieldPathsShouldBeAddedToDependencies) {
     intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
     auto object =
-        expressionObjectCreateHelper(expCtx, {{"x", ExpressionFieldPath::create(expCtx, "c.d")}});
+        ExpressionObject::create(expCtx, {{"x", ExpressionFieldPath::create(expCtx, "c.d")}});
     DepsTracker deps;
     object->addDependencies(&deps);
     ASSERT_EQ(deps.fields.size(), 1UL);
@@ -3552,7 +3528,7 @@ TEST(ExpressionObjectOptimizations, OptimizingAnObjectShouldOptimizeSubExpressio
     VariablesParseState vps = expCtx->variablesParseState;
     auto addExpression =
         ExpressionAdd::parse(expCtx, BSON("$add" << BSON_ARRAY(1 << 2)).firstElement(), vps);
-    auto object = expressionObjectCreateHelper(expCtx, {{"a", addExpression}});
+    auto object = ExpressionObject::create(expCtx, {{"a", addExpression}});
     ASSERT_EQ(object->getChildExpressions().size(), 1UL);
 
     auto optimized = object->optimize();
diff --git a/src/mongo/db/pipeline/parsed_inclusion_projection.h b/src/mongo/db/pipeline/parsed_inclusion_projection.h
index 46c93bf6a29..25740372acd 100644
--- a/src/mongo/db/pipeline/parsed_inclusion_projection.h
+++ b/src/mongo/db/pipeline/parsed_inclusion_projection.h
@@ -90,8 +90,13 @@ protected:
 class ParsedInclusionProjection : public ParsedAggregationProjection {
 public:
     ParsedInclusionProjection(const boost::intrusive_ptr<ExpressionContext>& expCtx,
+                              ProjectionPolicies policies,
+                              std::unique_ptr<InclusionNode> root)
+        : ParsedAggregationProjection(expCtx, policies), _root(std::move(root)) {}
+
+    ParsedInclusionProjection(const boost::intrusive_ptr<ExpressionContext>& expCtx,
                               ProjectionPolicies policies)
-        : ParsedAggregationProjection(expCtx, policies), _root(new InclusionNode(policies)) {}
+        : ParsedInclusionProjection(expCtx, policies, std::make_unique<InclusionNode>(policies)) {}
 
     TransformerType getType() const final {
         return TransformerType::kInclusionProjection;
diff --git a/src/mongo/db/pipeline/pipeline_d.cpp b/src/mongo/db/pipeline/pipeline_d.cpp
index 12c95222fe3..14584438e03 100644
--- a/src/mongo/db/pipeline/pipeline_d.cpp
+++ b/src/mongo/db/pipeline/pipeline_d.cpp
@@ -67,7 +67,6 @@
 #include "mongo/db/pipeline/document_source_single_document_transformation.h"
 #include "mongo/db/pipeline/document_source_sort.h"
 #include "mongo/db/pipeline/pipeline.h"
-#include "mongo/db/query/collation/collator_factory_interface.h"
 #include "mongo/db/query/collation/collator_interface.h"
 #include "mongo/db/query/get_executor.h"
 #include "mongo/db/query/plan_summary_stats.h"
@@ -880,17 +879,4 @@ void PipelineD::getPlanSummaryStats(const Pipeline* pipeline, PlanSummaryStats*
     statsOut->usedDisk = usedDisk;
 }
 
-std::unique_ptr<CollatorInterface> PipelineD::resolveCollator(OperationContext* opCtx,
-                                                              BSONObj userCollation,
-                                                              const Collection* collection) {
-    if (!userCollation.isEmpty()) {
-        return uassertStatusOK(
-            CollatorFactoryInterface::get(opCtx->getServiceContext())->makeFromBSON(userCollation));
-    }
-
-    return (collection && collection->getDefaultCollator()
-                ? collection->getDefaultCollator()->clone()
-                : nullptr);
-}
-
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/pipeline_d.h b/src/mongo/db/pipeline/pipeline_d.h
index 45bd9218484..606c91a1067 100644
--- a/src/mongo/db/pipeline/pipeline_d.h
+++ b/src/mongo/db/pipeline/pipeline_d.h
@@ -38,6 +38,7 @@
 #include "mongo/db/pipeline/dependencies.h"
 #include "mongo/db/pipeline/document_source_cursor.h"
 #include "mongo/db/pipeline/document_source_group.h"
+#include "mongo/db/query/collation/collator_factory_interface.h"
 #include "mongo/db/query/plan_executor.h"
 
 namespace mongo {
@@ -129,7 +130,16 @@ public:
      */
     static std::unique_ptr<CollatorInterface> resolveCollator(OperationContext* opCtx,
                                                               BSONObj userCollation,
-                                                              const Collection* collection);
+                                                              const Collection* collection) {
+        if (!userCollation.isEmpty()) {
+            return uassertStatusOK(CollatorFactoryInterface::get(opCtx->getServiceContext())
+                                       ->makeFromBSON(userCollation));
+        }
+
+        return (collection && collection->getDefaultCollator()
+                    ? collection->getDefaultCollator()->clone()
+                    : nullptr);
+    }
 
 private:
     PipelineD();  // does not exist:  prevent instantiation
diff --git a/src/mongo/db/query/util/make_data_structure.h b/src/mongo/db/query/util/make_data_structure.h
new file mode 100644
index 00000000000..32634c49041
--- /dev/null
+++ b/src/mongo/db/query/util/make_data_structure.h
@@ -0,0 +1,92 @@
+/**
+ *    Copyright (C) 2019-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    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
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    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 Server Side 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 <boost/optional.hpp>
+#include <list>
+#include <type_traits>
+
+namespace mongo {
+
+/**
+ * Create a vector. unlike an initializer list, this function will allow passing elements by Rvalue
+ * reference.
+ */
+template <typename T, typename... Args>
+auto make_vector(Args&&... args) {
+    std::vector<T> v;
+    v.reserve(sizeof...(Args));
+    (v.push_back(std::forward<Args>(args)), ...);
+    return v;
+}
+
+namespace detail {
+
+/**
+ * Appends an element with an operator* to the end of a data structure. If the operator* produces
+ * the data structure's element type, it will be called first unless the element argument is boolean
+ * convertable to false. If this is the case, this function will perform no action. The bool
+ * argument gives this function overload priority.
+ */
+template <typename T, typename DS, typename Arg>
+auto pushBackUnlessNone(DS&& ds, Arg&& arg, bool) -> decltype(*arg, void()) {
+    if constexpr (std::is_convertible_v<std::decay_t<decltype(*arg)>, T>) {
+        if (arg)
+            ds.push_back(*std::forward<Arg>(arg));
+    } else {
+        ds.push_back(std::forward<Arg>(arg));
+    }
+}
+
+/**
+ * Appends an element to the end of a data structure. SFINAE backup for elements without an
+ * operator*. The ... arguments casue this function to lose overload priority.
+ */
+template <typename T, typename DS, typename Arg>
+void pushBackUnlessNone(DS&& ds, Arg&& arg, ...) {
+    ds.push_back(std::forward<Arg>(arg));
+}
+
+}  // namespace detail
+
+/**
+ * Create a list. unlike an initializer list, this function will allow passing elements by Rvalue
+ * reference. If an argument is dereferenceable (operator*) to produce the new list's element type,
+ * the dereferencing will be performed before the argument is passed to the list. If an argument is
+ * dereferenceable and boolean convertable to false, it will be skipped.
+ */
+template <typename T, typename... Args>
+auto makeFlattenedList(Args&&... args) {
+    std::list<T> l;
+    (detail::pushBackUnlessNone<T>(l, std::forward<Args>(args), true), ...);
+    return l;
+}
+
+}  // namespace mongo