SERVER-42942 M/R Agg: Implement translation for cluster mapReduce command

Author: nzolnierzmdb

buildscripts/resmokeconfig/suites/sharding_last_stable_mongos_and_mixed_shards.yml

@@ -49,6 +49,7 @@ selector:
   - jstests/sharding/track_unsharded_collections_rename_collection.js
   - jstests/sharding/banned_txn_databases_sharded.js
   - jstests/sharding/split_large_key.js
+  - jstests/sharding/change_streams_establishment_finds_new_shards.js
   # Enable if SERVER-41813 is backported or 4.4 becomes last-stable
   - jstests/sharding/invalid_system_views_sharded_collection.js
 

buildscripts/resmokeconfig/suites/sharding_map_reduce_agg.yaml

@@ -4,8 +4,16 @@ test_kind: js_test
 
 selector:
   roots:
-  # Placeholder test to avoid failing in resmoke.
-  - jstests/sharding/accurate_count_with_predicate.js
+  - jstests/sharding/auth.js
+  - jstests/sharding/authCommands.js
+  - jstests/sharding/authmr.js
+  - jstests/sharding/causal_consistency_shell_support.js
+  - jstests/sharding/localhostAuthBypass.js
+  - jstests/sharding/max_time_ms_sharded.js
+  - jstests/sharding/mr_and_agg_versioning.js
+  - jstests/sharding/mr_shard_version.js
+  - jstests/sharding/query_config.js
+  - jstests/sharding/shard_targeting.js
 
 executor:
   config:

jstests/aggregation/sharded_agg_cleanup_on_error.js

@@ -88,7 +88,7 @@ try {
 try {
     // Enable the failpoint to fail on establishing a merging shard cursor.
     assert.commandWorked(mongosDB.adminCommand({
-        configureFailPoint: "clusterAggregateFailToEstablishMergingShardCursor",
+        configureFailPoint: "shardedAggregateFailToEstablishMergingShardCursor",
         mode: "alwaysOn"
     }));
 
@@ -103,13 +103,13 @@ try {
 
 } finally {
     assert.commandWorked(mongosDB.adminCommand(
-        {configureFailPoint: "clusterAggregateFailToEstablishMergingShardCursor", mode: "off"}));
+        {configureFailPoint: "shardedAggregateFailToEstablishMergingShardCursor", mode: "off"}));
 }
 
 // Test that aggregations involving $exchange correctly clean up the producer cursors.
 try {
     assert.commandWorked(mongosDB.adminCommand({
-        configureFailPoint: "clusterAggregateFailToDispatchExchangeConsumerPipeline",
+        configureFailPoint: "shardedAggregateFailToDispatchExchangeConsumerPipeline",
         mode: "alwaysOn"
     }));
 
@@ -133,7 +133,7 @@ try {
 
 } finally {
     assert.commandWorked(mongosDB.adminCommand({
-        configureFailPoint: "clusterAggregateFailToDispatchExchangeConsumerPipeline",
+        configureFailPoint: "shardedAggregateFailToDispatchExchangeConsumerPipeline",
         mode: "off"
     }));
 }

jstests/sharding/change_streams_establishment_finds_new_shards.js

@@ -40,13 +40,13 @@ assert.commandWorked(mongos.adminCommand({split: mongosColl.getFullName(), middl
 
 // Enable the failpoint.
 assert.commandWorked(mongos.adminCommand(
-    {configureFailPoint: "clusterAggregateHangBeforeEstablishingShardCursors", mode: "alwaysOn"}));
+    {configureFailPoint: "shardedAggregateHangBeforeEstablishingShardCursors", mode: "alwaysOn"}));
 
 // While opening the cursor, wait for the failpoint and add the new shard.
 const awaitNewShard = startParallelShell(`
         load("jstests/libs/check_log.js");
         checkLog.contains(db,
-            "clusterAggregateHangBeforeEstablishingShardCursors fail point enabled");
+            "shardedAggregateHangBeforeEstablishingShardCursors fail point enabled");
         assert.commandWorked(
             db.adminCommand({addShard: "${newShard.getURL()}", name: "${newShard.name}"}));
         // Migrate the [10, MaxKey] chunk to "newShard".
@@ -56,7 +56,7 @@ const awaitNewShard = startParallelShell(`
                                               _waitForDelete: true}));
         assert.commandWorked(
             db.adminCommand(
-                {configureFailPoint: "clusterAggregateHangBeforeEstablishingShardCursors",
+                {configureFailPoint: "shardedAggregateHangBeforeEstablishingShardCursors",
                  mode: "off"}));`,
                                              mongos.port);
 

src/mongo/db/commands/SConscript

@@ -157,6 +157,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/logical_session_cache',
         '$BUILD_DIR/mongo/db/logical_session_id_helpers',
         '$BUILD_DIR/mongo/db/logical_session_id',
+        '$BUILD_DIR/mongo/db/pipeline/pipeline',
         '$BUILD_DIR/mongo/db/repl/isself',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
         '$BUILD_DIR/mongo/db/session_catalog',
@@ -508,10 +509,10 @@ env.Library(
         'map_reduce_agg.cpp',
     ],
     LIBDEPS=[
-        '$BUILD_DIR/mongo/db/db_raii',
         '$BUILD_DIR/mongo/idl/idl_parser',
+        '$BUILD_DIR/mongo/db/commands/servers',
+        '$BUILD_DIR/mongo/db/db_raii',
         '$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'
     ]
@@ -541,6 +542,7 @@ env.CppUnitTest(
         "map_reduce_parse_test.cpp",
     ],
     LIBDEPS=[
+        '$BUILD_DIR/mongo/db/auth/authmocks',
         '$BUILD_DIR/mongo/db/query/query_test_service_context',
         'map_reduce_agg',
     ]

src/mongo/db/commands/map_reduce_agg.cpp

@@ -41,155 +41,18 @@
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/map_reduce_agg.h"
 #include "mongo/db/commands/map_reduce_javascript_code.h"
+#include "mongo/db/commands/mr_common.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/exec/document_value/value.h"
 #include "mongo/db/namespace_string.h"
-#include "mongo/db/pipeline/document_source.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/query/map_reduce_output_format.h"
-#include "mongo/db/query/util/make_data_structure.h"
-#include "mongo/util/intrusive_counter.h"
 
 namespace mongo::map_reduce_agg {
 
 namespace {
 
-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.
@@ -247,7 +110,7 @@ bool runAggregationMapReduce(OperationContext* opCtx,
     auto parsedMr = MapReduce::parse(IDLParserErrorContext("MapReduce"), cmd);
     auto expCtx = makeExpressionContext(opCtx, parsedMr);
     auto runnablePipeline = [&]() {
-        auto pipeline = translateFromMR(parsedMr, expCtx);
+        auto pipeline = map_reduce_common::translateFromMR(parsedMr, expCtx);
         return expCtx->mongoProcessInterface->attachCursorSourceToPipelineForLocalRead(
             expCtx, pipeline.release());
     }();
@@ -274,32 +137,4 @@ bool runAggregationMapReduce(OperationContext* opCtx,
     return true;
 }
 
-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

src/mongo/db/commands/map_reduce_agg.h

@@ -47,7 +47,4 @@ bool runAggregationMapReduce(OperationContext* opCtx,
                              std::string& errmsg,
                              BSONObjBuilder& result);
 
-std::unique_ptr<Pipeline, PipelineDeleter> translateFromMR(
-    MapReduce parsedMr, boost::intrusive_ptr<ExpressionContext> expCtx);
-
 }  // namespace mongo::map_reduce_agg