SERVER-44013 MR Agg: Report plan stats and summary for currentOp/profiler/slow query logging

4 files changed, 29 insertions, 15 deletions

diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript
index c382ad4a62a..96e4d4c32a0 100644
--- a/src/mongo/db/commands/SConscript
+++ b/src/mongo/db/commands/SConscript
@@ -523,6 +523,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/commands/servers',
         '$BUILD_DIR/mongo/db/db_raii',
         '$BUILD_DIR/mongo/db/pipeline/mongo_process_interface',
+        '$BUILD_DIR/mongo/db/query_exec',
         '$BUILD_DIR/mongo/db/query/map_reduce_output_format',
         '$BUILD_DIR/mongo/idl/idl_parser',
         'map_reduce_parser'
diff --git a/src/mongo/db/commands/map_reduce_agg.cpp b/src/mongo/db/commands/map_reduce_agg.cpp
index 68ad9cb031d..8ca42eec7b0 100644
--- a/src/mongo/db/commands/map_reduce_agg.cpp
+++ b/src/mongo/db/commands/map_reduce_agg.cpp
@@ -43,9 +43,11 @@
 #include "mongo/db/commands/map_reduce_javascript_code.h"
 #include "mongo/db/commands/map_reduce_stats.h"
 #include "mongo/db/commands/mr_common.h"
+#include "mongo/db/curop.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_cursor.h"
 #include "mongo/db/pipeline/expression.h"
 #include "mongo/db/pipeline/pipeline_d.h"
 #include "mongo/db/query/map_reduce_output_format.h"
@@ -78,7 +80,7 @@ auto makeExpressionContext(OperationContext* opCtx, const MapReduce& parsedMr) {
     // 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>(
+    auto expCtx = make_intrusive<ExpressionContext>(
         opCtx,
         boost::none,  // explain
         false,        // fromMongos
@@ -91,6 +93,8 @@ auto makeExpressionContext(OperationContext* opCtx, const MapReduce& parsedMr) {
         MongoProcessInterface::create(opCtx),
         StringMap<ExpressionContext::ResolvedNamespace>{},  // resolvedNamespaces
         uuid);
+    expCtx->tempDir = storageGlobalParams.dbpath + "/_tmp";
+    return expCtx;
 }
 
 std::vector<CommonStats> extractStats(const Pipeline& pipeline) {
@@ -127,8 +131,19 @@ bool runAggregationMapReduce(OperationContext* opCtx,
             expCtx, pipeline.release());
     }();
 
+    {
+        auto planSummaryStr = PipelineD::getPlanSummaryStr(runnablePipeline.get());
+
+        stdx::lock_guard<Client> lk(*opCtx->getClient());
+        CurOp::get(opCtx)->setPlanSummary_inlock(std::move(planSummaryStr));
+    }
+
     auto resultArray = exhaustPipelineIntoBSONArray(runnablePipeline);
 
+    PlanSummaryStats planSummaryStats;
+    PipelineD::getPlanSummaryStats(runnablePipeline.get(), &planSummaryStats);
+    CurOp::get(opCtx)->debug().setPlanSummaryMetrics(planSummaryStats);
+
     MapReduceStats mapReduceStats(extractStats(*runnablePipeline),
                                   MapReduceStats::ResponseType::kUnsharded,
                                   boost::get_optional_value_or(parsedMr.getVerbose(), false),
@@ -147,6 +162,15 @@ bool runAggregationMapReduce(OperationContext* opCtx,
                                                     &result);
     }
 
+    // The aggregation pipeline may change the namespace of the curop and we need to set it back to
+    // the original namespace to correctly report command stats. One example when the namespace can
+    // be changed is when the pipeline contains an $out stage, which executes an internal command to
+    // create a temp collection, changing the curop namespace to the name of this temp collection.
+    {
+        stdx::lock_guard<Client> lk(*opCtx->getClient());
+        CurOp::get(opCtx)->setNS_inlock(parsedMr.getNamespace().ns());
+    }
+
     return true;
 }
 
diff --git a/src/mongo/db/pipeline/document_source_cursor.cpp b/src/mongo/db/pipeline/document_source_cursor.cpp
index 6ad22cce49d..c2f9c9d1a8d 100644
--- a/src/mongo/db/pipeline/document_source_cursor.cpp
+++ b/src/mongo/db/pipeline/document_source_cursor.cpp
@@ -161,14 +161,7 @@ void DocumentSourceCursor::_updateOplogTimestamp() {
 
 void DocumentSourceCursor::recordPlanSummaryStats() {
     invariant(_exec);
-    // Aggregation handles in-memory sort outside of the query sub-system. Given that we need to
-    // preserve the existing value of hasSortStage rather than overwrite with the underlying
-    // PlanExecutor's value.
-    auto hasSortStage = _planSummaryStats.hasSortStage;
-
     Explain::getSummaryStats(*_exec, &_planSummaryStats);
-
-    _planSummaryStats.hasSortStage = hasSortStage;
 }
 
 Value DocumentSourceCursor::serialize(boost::optional<ExplainOptions::Verbosity> verbosity) const {
diff --git a/src/mongo/db/pipeline/pipeline_d.cpp b/src/mongo/db/pipeline/pipeline_d.cpp
index 3bdd0a61fa2..d9c64f97707 100644
--- a/src/mongo/db/pipeline/pipeline_d.cpp
+++ b/src/mongo/db/pipeline/pipeline_d.cpp
@@ -851,18 +851,14 @@ void PipelineD::getPlanSummaryStats(const Pipeline* pipeline, PlanSummaryStats*
         *statsOut = docSourceCursor->getPlanSummaryStats();
     }
 
-    bool hasSortStage{false};
-    bool usedDisk{false};
     for (auto&& source : pipeline->_sources) {
         if (dynamic_cast<DocumentSourceSort*>(source.get()))
-            hasSortStage = true;
+            statsOut->hasSortStage = true;
 
-        usedDisk = usedDisk || source->usedDisk();
-        if (usedDisk && hasSortStage)
+        statsOut->usedDisk = statsOut->usedDisk || source->usedDisk();
+        if (statsOut->usedDisk && statsOut->hasSortStage)
             break;
     }
-    statsOut->hasSortStage = hasSortStage;
-    statsOut->usedDisk = usedDisk;
 }
 
 }  // namespace mongo