SERVER-71636 Fix explain of $lookup when inner collection is sharded

Before the fix, the system would erroneously dispatch
explain commands across the wire when executing the inner
side and throw away the results. After the fix, the system
will actually run the agg command itself against the inner
side. This improves the runtime stats that we expose in the
final explain output, and avoids the parsing-related error
originally reported under this ticket.

(cherry picked from commit 1ff4d25cae73aa0c26e7a2f09da363401ba48d33)

10 files changed, 300 insertions, 44 deletions

diff --git a/jstests/aggregation/sources/lookup/lookup_query_stats.js b/jstests/aggregation/sources/lookup/lookup_query_stats.js
index d1a096e54d1..8c598b6c2f2 100644
--- a/jstests/aggregation/sources/lookup/lookup_query_stats.js
+++ b/jstests/aggregation/sources/lookup/lookup_query_stats.js
@@ -252,7 +252,12 @@ let testQueryExecutorStatsWithCollectionScan = function(params) {
         checkExplainOutputForAllVerbosityLevels(
             localColl,
             fromColl,
-            {totalDocsExamined: 20, totalKeysExamined: 0, collectionScans: 4, indexesUsed: []},
+            {
+                totalDocsExamined: localDocCount * foreignDocCount,
+                totalKeysExamined: 0,
+                collectionScans: localDocCount,
+                indexesUsed: []
+            },
             {allowDiskUse: false},
             params.withUnwind);
     }
diff --git a/jstests/noPassthrough/explain_unionwith_lookup_sharded.js b/jstests/noPassthrough/explain_unionwith_lookup_sharded.js
new file mode 100644
index 00000000000..ef4b4a42bec
--- /dev/null
+++ b/jstests/noPassthrough/explain_unionwith_lookup_sharded.js
@@ -0,0 +1,216 @@
+/**
+ * Test that explain of $lookup and $unionWith works correctly when the involved collections are
+ * sharded.
+ *
+ * This test was originally designed to reproduce SERVER-71636.
+ */
+(function() {
+"use strict";
+
+load("jstests/libs/analyze_plan.js");
+
+const dbName = "test";
+
+const st = new ShardingTest({shards: 2});
+const db = st.s.getDB(dbName);
+
+const outerColl = db["outer"];
+const innerColl = db["inner"];
+
+(function createOuterColl() {
+    outerColl.drop();
+    assert.commandWorked(outerColl.insert([{_id: 1, x: "foo"}, {_id: 3, x: "foo"}]));
+}());
+
+function createInnerColl() {
+    innerColl.drop();
+    assert.commandWorked(
+        innerColl.insert([{_id: 1, x: "foo", y: "a"}, {_id: 2, x: "foo", y: "z"}]));
+}
+createInnerColl();
+
+function explainStage(stage, stageName) {
+    const pipeline = [stage];
+    let explain = outerColl.explain("executionStats").aggregate(pipeline);
+    let stageExplain = getAggPlanStage(explain, stageName);
+    assert.neq(stageExplain, null, explain);
+    return stageExplain;
+}
+
+// Explain of $lookup when neither collection is sharded.
+const lookupStage =
+        {
+            $lookup: {
+                from: innerColl.getName(),
+                let: {
+                    myX: "$x",
+                },
+                pipeline: [
+                    {$match: {$expr: {$eq: ["$x", "$$myX"]}}},
+                ],
+                as: "as"
+            }
+        };
+let stageExplain = explainStage(lookupStage, "$lookup");
+assert.eq(stageExplain.nReturned, 2, stageExplain);
+// The two documents in the inner collection are scanned twice, leading to four docs examined in
+// total.
+assert.eq(stageExplain.collectionScans, 2, stageExplain);
+assert.eq(stageExplain.totalDocsExamined, 4, stageExplain);
+assert.eq(stageExplain.totalKeysExamined, 0, stageExplain);
+assert.eq(stageExplain.indexesUsed, [], stageExplain);
+
+const unionWithStage = {
+    $unionWith: {
+        coll: innerColl.getName(),
+        pipeline: [
+            {$match: {$expr: {$eq: ["$x", "foo"]}}},
+        ]
+    }
+};
+
+const nestedUnionWithStage = {
+    $unionWith: {coll: innerColl.getName(), pipeline: [unionWithStage]}
+};
+
+// Explain of $unionWith when neither collection is sharded.
+stageExplain = explainStage(unionWithStage, "$unionWith");
+assert.eq(stageExplain.nReturned, 4, stageExplain);
+
+// Explain of nested $unionWith when neither collection is sharded.
+stageExplain = explainStage(nestedUnionWithStage, "$unionWith");
+assert.eq(stageExplain.nReturned, 6, stageExplain);
+
+// Shard the inner collection.
+assert.commandWorked(innerColl.createIndex({y: 1, x: 1}));
+st.shardColl(innerColl.getName(),
+             {y: 1, x: 1} /* shard key */,
+             {y: "b", x: "b"} /* split at */,
+             {y: "c", x: "c"} /* move */,
+             dbName,
+             true);
+
+// Explain of $lookup when outer collection is unsharded and inner collection is sharded.
+stageExplain = explainStage(lookupStage, "$lookup");
+assert.eq(stageExplain.nReturned, 2, stageExplain);
+// Now that the inner collection is sharded, the execution of the $lookup requires dispatching
+// commands across the wire for the inner collection. The runtime stats currently do not reflect the
+// work done by these dispatched subcommands. We could improve this in the future to more accurately
+// reflect docs examined, keys examined, collection scans, etc accrued when executing the
+// subpipeline.
+assert.eq(stageExplain.totalDocsExamined, 0, stageExplain);
+assert.eq(stageExplain.totalKeysExamined, 0, stageExplain);
+assert.eq(stageExplain.collectionScans, 0, stageExplain);
+assert.eq(stageExplain.indexesUsed, [], stageExplain);
+
+// Explain of $unionWith when outer collection is unsharded and inner collection is sharded.
+stageExplain = explainStage(unionWithStage, "$unionWith");
+assert.eq(stageExplain.nReturned, 4, stageExplain);
+// The $unionWith explain format currently shows explains for the inner pipeline from both
+// targeted shards.
+assert(stageExplain.$unionWith.hasOwnProperty("pipeline"), stageExplain);
+const pipelineExplain = stageExplain.$unionWith.pipeline;
+assert(pipelineExplain.hasOwnProperty("shards"), stageExplain);
+const shardNames = Object.keys(pipelineExplain.shards);
+assert.eq(shardNames.length, 2, stageExplain);
+// Each shard should have returned one document.
+assert.eq(pipelineExplain.shards[shardNames[0]].executionStats.nReturned, 1, stageExplain);
+assert.eq(pipelineExplain.shards[shardNames[1]].executionStats.nReturned, 1, stageExplain);
+
+// Explain of nested $unionWith when outer collection is unsharded and inner collection is sharded.
+stageExplain = explainStage(nestedUnionWithStage, "$unionWith");
+assert.eq(stageExplain.nReturned, 6, stageExplain);
+
+// Shard the outer collection.
+st.shardColl(outerColl.getName(),
+             {_id: 1} /* shard key */,
+             {_id: 2} /* split at */,
+             {_id: 3} /* move */,
+             dbName,
+             true);
+
+// A variant of 'explainStage()' when the stage is expected to appear twice because it runs on
+// two shards.
+function explainStageTwoShards(stage, stageName) {
+    const pipeline = [stage];
+    let explain = outerColl.explain("executionStats").aggregate(pipeline);
+    let stageExplain = getAggPlanStages(explain, stageName);
+    assert.eq(stageExplain.length, 2, stageExplain);
+    return stageExplain;
+}
+
+// Explain of $lookup when inner and outer collections are both sharded.
+stageExplain = explainStageTwoShards(lookupStage, "$lookup");
+for (let explain of stageExplain) {
+    assert.eq(explain.nReturned, 1, stageExplain);
+    // As above, the inner collection is sharded. We don't currently ship execution stats across the
+    // wire alongside the query results themselves. As a result, the docs examined, total keys
+    // examined, etc. will currently always be reported as zero when the inner collection is
+    // sharded. We could improve this in the future to report the stats more accurately.
+    assert.eq(explain.totalDocsExamined, 0, stageExplain);
+    assert.eq(explain.totalKeysExamined, 0, stageExplain);
+    assert.eq(explain.collectionScans, 0, stageExplain);
+    assert.eq(explain.indexesUsed, [], stageExplain);
+}
+
+// Asserts that 'explain' is for a split pipeline with an empty shards part and a merger part with
+// two stages. Asserts that the first merging stage is a $mergeCursors and then returns the second
+// stage in the merging pipeline.
+function getStageFromMergerPart(explain) {
+    assert(explain.hasOwnProperty("splitPipeline"));
+    assert(explain.splitPipeline.hasOwnProperty("shardsPart"));
+    assert.eq(explain.splitPipeline.shardsPart, [], explain);
+    assert(explain.splitPipeline.hasOwnProperty("mergerPart"));
+    let mergerPart = explain.splitPipeline.mergerPart;
+    assert.eq(mergerPart.length, 2, explain);
+    assert(mergerPart[0].hasOwnProperty("$mergeCursors"), explain);
+    return mergerPart[1];
+}
+
+function assertStageDoesNotHaveRuntimeStats(stageExplain) {
+    assert(!stageExplain.hasOwnProperty("nReturned"), stageExplain);
+    assert(!stageExplain.hasOwnProperty("totalDocsExamined"), stageExplain);
+    assert(!stageExplain.hasOwnProperty("totalKeysExamined"), stageExplain);
+    assert(!stageExplain.hasOwnProperty("collectionScans"), stageExplain);
+    assert(!stageExplain.hasOwnProperty("indexesUsed"), stageExplain);
+}
+
+// Explain of $unionWith when inner and outer collections are both sharded. We expect the $unionWith
+// to be part of the merging pipeline rather than pushed down to the shards.
+let explain = outerColl.explain("executionStats").aggregate([unionWithStage]);
+stageExplain = getStageFromMergerPart(explain);
+assert(stageExplain.hasOwnProperty("$unionWith"), explain);
+assertStageDoesNotHaveRuntimeStats(stageExplain);
+
+// Nested $unionWith when inner and outer collections are sharded.
+explain = outerColl.explain("executionStats").aggregate([nestedUnionWithStage]);
+stageExplain = getStageFromMergerPart(explain);
+assert(stageExplain.hasOwnProperty("$unionWith"), explain);
+assertStageDoesNotHaveRuntimeStats(stageExplain);
+
+// Drop and recreate the inner collection. Re-test when the outer collection is sharded but the
+// inner collection is unsharded.
+createInnerColl();
+
+// Explain of $lookup when outer collection is sharded and inner collection is unsharded. In this
+// case we expect the $lookup operation to execute on the primary shard as part of the merging
+// pipeline.
+explain = outerColl.explain("executionStats").aggregate([lookupStage]);
+stageExplain = getStageFromMergerPart(explain);
+assert(stageExplain.hasOwnProperty("$lookup"), explain);
+assertStageDoesNotHaveRuntimeStats(stageExplain);
+
+// Explain of $unionWith when the outer collection is sharded and the inner collection is unsharded.
+explain = outerColl.explain("executionStats").aggregate([unionWithStage]);
+stageExplain = getStageFromMergerPart(explain);
+assert(stageExplain.hasOwnProperty("$unionWith"), explain);
+assertStageDoesNotHaveRuntimeStats(stageExplain);
+
+// Nested $unionWith when the outer collection is sharded and the inner collection is unsharded.
+explain = outerColl.explain("executionStats").aggregate([nestedUnionWithStage]);
+stageExplain = getStageFromMergerPart(explain);
+assert(stageExplain.hasOwnProperty("$unionWith"), explain);
+assertStageDoesNotHaveRuntimeStats(stageExplain);
+
+st.stop();
+}());
diff --git a/src/mongo/db/pipeline/dispatch_shard_pipeline_test.cpp b/src/mongo/db/pipeline/dispatch_shard_pipeline_test.cpp
index 0203dddb7a5..b7333014a14 100644
--- a/src/mongo/db/pipeline/dispatch_shard_pipeline_test.cpp
+++ b/src/mongo/db/pipeline/dispatch_shard_pipeline_test.cpp
@@ -61,7 +61,8 @@ TEST_F(DispatchShardPipelineTest, DoesNotSplitPipelineIfTargetingOneShard) {
                                                                   hasChangeStream,
                                                                   startsWithDocuments,
                                                                   eligibleForSampling,
-                                                                  std::move(pipeline));
+                                                                  std::move(pipeline),
+                                                                  boost::none /*explain*/);
         ASSERT_EQ(results.remoteCursors.size(), 1UL);
         ASSERT(!results.splitPipeline);
     });
@@ -97,7 +98,8 @@ TEST_F(DispatchShardPipelineTest, DoesSplitPipelineIfMatchSpansTwoShards) {
                                                                   hasChangeStream,
                                                                   startsWithDocuments,
                                                                   eligibleForSampling,
-                                                                  std::move(pipeline));
+                                                                  std::move(pipeline),
+                                                                  boost::none /*explain*/);
         ASSERT_EQ(results.remoteCursors.size(), 2UL);
         ASSERT(bool(results.splitPipeline));
     });
@@ -136,7 +138,8 @@ TEST_F(DispatchShardPipelineTest, DispatchShardPipelineRetriesOnNetworkError) {
                                                                   hasChangeStream,
                                                                   startsWithDocuments,
                                                                   eligibleForSampling,
-                                                                  std::move(pipeline));
+                                                                  std::move(pipeline),
+                                                                  boost::none /*explain*/);
         ASSERT_EQ(results.remoteCursors.size(), 2UL);
         ASSERT(bool(results.splitPipeline));
     });
@@ -186,7 +189,8 @@ TEST_F(DispatchShardPipelineTest, DispatchShardPipelineDoesNotRetryOnStaleConfig
                                                                       hasChangeStream,
                                                                       startsWithDocuments,
                                                                       eligibleForSampling,
-                                                                      std::move(pipeline)),
+                                                                      std::move(pipeline),
+                                                                      boost::none /*explain*/),
                            AssertionException,
                            ErrorCodes::StaleConfig);
     });
@@ -225,16 +229,17 @@ TEST_F(DispatchShardPipelineTest, WrappedDispatchDoesRetryOnStaleConfigError) {
     auto future = launchAsync([&] {
         // Shouldn't throw.
         sharding::router::CollectionRouter router(getServiceContext(), kTestAggregateNss);
-        auto results =
-            router.route(operationContext(),
-                         "dispatch shard pipeline"_sd,
-                         [&](OperationContext* opCtx, const CollectionRoutingInfo& cri) {
-                             return sharded_agg_helpers::dispatchShardPipeline(serializedCommand,
-                                                                               hasChangeStream,
-                                                                               startsWithDocuments,
-                                                                               eligibleForSampling,
-                                                                               pipeline->clone());
-                         });
+        auto results = router.route(operationContext(),
+                                    "dispatch shard pipeline"_sd,
+                                    [&](OperationContext* opCtx, const CollectionRoutingInfo& cri) {
+                                        return sharded_agg_helpers::dispatchShardPipeline(
+                                            serializedCommand,
+                                            hasChangeStream,
+                                            startsWithDocuments,
+                                            eligibleForSampling,
+                                            pipeline->clone(),
+                                            boost::none /*explain*/);
+                                    });
         ASSERT_EQ(results.remoteCursors.size(), 1UL);
         ASSERT(!bool(results.splitPipeline));
     });
diff --git a/src/mongo/db/pipeline/document_source_change_stream_handle_topology_change.cpp b/src/mongo/db/pipeline/document_source_change_stream_handle_topology_change.cpp
index cfd2a66e854..4b4e62ae492 100644
--- a/src/mongo/db/pipeline/document_source_change_stream_handle_topology_change.cpp
+++ b/src/mongo/db/pipeline/document_source_change_stream_handle_topology_change.cpp
@@ -226,7 +226,8 @@ BSONObj DocumentSourceChangeStreamHandleTopologyChange::createUpdatedCommandForN
                                                                Document{shardCommand},
                                                                splitPipelines,
                                                                boost::none, /* exhangeSpec */
-                                                               true /* needsMerge */);
+                                                               true /* needsMerge */,
+                                                               boost::none /* explain */);
 }
 
 BSONObj DocumentSourceChangeStreamHandleTopologyChange::replaceResumeTokenInCommand(
diff --git a/src/mongo/db/pipeline/expression_context.h b/src/mongo/db/pipeline/expression_context.h
index a0f04003479..e7d0621bc9e 100644
--- a/src/mongo/db/pipeline/expression_context.h
+++ b/src/mongo/db/pipeline/expression_context.h
@@ -301,11 +301,11 @@ public:
     }
 
     /**
-     * Returns true if the pipeline is eligible for query sampling. That is, it is not an explain
-     * and either it is not nested or it is nested inside $lookup, $graphLookup and $unionWith.
+     * Returns true if the pipeline is eligible for query sampling for the purpose of shard key
+     * selection metrics.
      */
     bool eligibleForSampling() const {
-        return !explain && (subPipelineDepth == 0 || inLookup || inUnionWith);
+        return !explain;
     }
 
     void setResolvedNamespaces(StringMap<ResolvedNamespace> resolvedNamespaces) {
diff --git a/src/mongo/db/pipeline/sharded_agg_helpers.cpp b/src/mongo/db/pipeline/sharded_agg_helpers.cpp
index b4f16cb7dde..aca3354a9b4 100644
--- a/src/mongo/db/pipeline/sharded_agg_helpers.cpp
+++ b/src/mongo/db/pipeline/sharded_agg_helpers.cpp
@@ -784,6 +784,10 @@ std::unique_ptr<Pipeline, PipelineDeleter> targetShardsAndAddMergeCursors(
                               startsWithDocuments,
                               expCtx->eligibleForSampling(),
                               std::move(pipeline),
+                              // Even if the overall operation is an explain, callers of this
+                              // function always intend to actually execute a regular agg command
+                              // and merge the results with $mergeCursors.
+                              boost::none /*explain*/,
                               shardTargetingPolicy,
                               std::move(readConcern));
 
@@ -966,6 +970,7 @@ BSONObj createCommandForTargetedShards(const boost::intrusive_ptr<ExpressionCont
                                        const SplitPipeline& splitPipeline,
                                        const boost::optional<ShardedExchangePolicy> exchangeSpec,
                                        bool needsMerge,
+                                       boost::optional<ExplainOptions::Verbosity> explain,
                                        boost::optional<BSONObj> readConcern) {
     // Create the command for the shards.
     MutableDocument targetedCmd(serializedCommand);
@@ -997,30 +1002,23 @@ BSONObj createCommandForTargetedShards(const boost::intrusive_ptr<ExpressionCont
     targetedCmd[AggregateCommandRequest::kExchangeFieldName] =
         exchangeSpec ? Value(exchangeSpec->exchangeSpec.toBSON()) : Value();
 
-    auto shardCommand = genericTransformForShards(std::move(targetedCmd),
-                                                  expCtx,
-                                                  expCtx->explain,
-                                                  expCtx->getCollatorBSON(),
-                                                  std::move(readConcern));
+    auto shardCommand = genericTransformForShards(
+        std::move(targetedCmd), expCtx, explain, expCtx->getCollatorBSON(), std::move(readConcern));
 
     // Apply RW concern to the final shard command.
     return applyReadWriteConcern(expCtx->opCtx,
-                                 true,             /* appendRC */
-                                 !expCtx->explain, /* appendWC */
+                                 true,     /* appendRC */
+                                 !explain, /* appendWC */
                                  shardCommand);
 }
 
-/**
- * Targets shards for the pipeline and returns a struct with the remote cursors or results, and
- * the pipeline that will need to be executed to merge the results from the remotes. If a stale
- * shard version is encountered, refreshes the routing table and tries again.
- */
 DispatchShardPipelineResults dispatchShardPipeline(
     Document serializedCommand,
     bool hasChangeStream,
     bool startsWithDocuments,
     bool eligibleForSampling,
     std::unique_ptr<Pipeline, PipelineDeleter> pipeline,
+    boost::optional<ExplainOptions::Verbosity> explain,
     ShardTargetingPolicy shardTargetingPolicy,
     boost::optional<BSONObj> readConcern) {
     auto expCtx = pipeline->getContext();
@@ -1108,10 +1106,11 @@ DispatchShardPipelineResults dispatchShardPipeline(
                                                          *splitPipelines,
                                                          exchangeSpec,
                                                          true /* needsMerge */,
+                                                         explain,
                                                          std::move(readConcern))
                         : createPassthroughCommandForShard(expCtx,
                                                            serializedCommand,
-                                                           expCtx->explain,
+                                                           explain,
                                                            pipeline.get(),
                                                            expCtx->getCollatorBSON(),
                                                            std::move(readConcern),
@@ -1147,7 +1146,7 @@ DispatchShardPipelineResults dispatchShardPipeline(
             shardIds.size() > 0);
 
     // Explain does not produce a cursor, so instead we scatter-gather commands to the shards.
-    if (expCtx->explain) {
+    if (explain) {
         if (mustRunOnAll) {
             // Some stages (such as $currentOp) need to be broadcast to all shards, and
             // should not participate in the shard version protocol.
@@ -1490,7 +1489,8 @@ BSONObj targetShardsForExplain(Pipeline* ownedPipeline) {
                               hasChangeStream,
                               startsWithDocuments,
                               expCtx->eligibleForSampling(),
-                              std::move(pipeline));
+                              std::move(pipeline),
+                              expCtx->explain);
     BSONObjBuilder explainBuilder;
     auto appendStatus =
         appendExplainResults(std::move(shardDispatchResults), expCtx, &explainBuilder);
diff --git a/src/mongo/db/pipeline/sharded_agg_helpers.h b/src/mongo/db/pipeline/sharded_agg_helpers.h
index 66c494a7e77..ecd82ee944a 100644
--- a/src/mongo/db/pipeline/sharded_agg_helpers.h
+++ b/src/mongo/db/pipeline/sharded_agg_helpers.h
@@ -117,12 +117,25 @@ boost::optional<ShardedExchangePolicy> checkIfEligibleForExchange(OperationConte
 SplitPipeline splitPipeline(std::unique_ptr<Pipeline, PipelineDeleter> pipeline);
 
 /**
- * Targets shards for the pipeline and returns a struct with the remote cursors or results, and
- * the pipeline that will need to be executed to merge the results from the remotes. If a stale
- * shard version is encountered, refreshes the routing table and tries again. If the command is
- * eligible for sampling, attaches a unique sample id to the request for one of the targeted shards
- * if the collection has query sampling enabled and the rate-limited sampler successfully generates
- * a sample id for it.
+ * Targets shards for the pipeline and returns a struct with the remote cursors or results, and the
+ * pipeline that will need to be executed to merge the results from the remotes. If a stale shard
+ * version is encountered, refreshes the routing table and tries again. If the command is eligible
+ * for sampling, attaches a unique sample id to the request for one of the targeted shards if the
+ * collection has query sampling enabled and the rate-limited sampler successfully generates a
+ * sample id for it.
+ *
+ * Although the 'pipeline' has an 'ExpressionContext' which indicates whether this operation is an
+ * explain (and if it is an explain what the verbosity is), the caller must explicitly indicate
+ * whether it wishes to dispatch a regular aggregate command or an explain command using the
+ * explicit 'explain' parameter. The reason for this is that in some contexts, the caller wishes to
+ * dispatch a regular agg command rather than an explain command even if the top-level operation is
+ * an explain. Consider the example of an explain that contains a stage like this:
+ *
+ *     {$unionWith: {coll: "innerShardedColl", pipeline: <sub-pipeline>}}
+ *
+ * The explain works by first executing the inner and outer subpipelines in order to gather runtime
+ * statistics. While dispatching the inner pipeline, we must dispatch it not as an explain but as a
+ * regular agg command so that the runtime stats are accurate.
  */
 DispatchShardPipelineResults dispatchShardPipeline(
     Document serializedCommand,
@@ -130,6 +143,7 @@ DispatchShardPipelineResults dispatchShardPipeline(
     bool startsWithDocuments,
     bool eligibleForSampling,
     std::unique_ptr<Pipeline, PipelineDeleter> pipeline,
+    boost::optional<ExplainOptions::Verbosity> explain,
     ShardTargetingPolicy shardTargetingPolicy = ShardTargetingPolicy::kAllowed,
     boost::optional<BSONObj> readConcern = boost::none);
 
@@ -147,6 +161,7 @@ BSONObj createCommandForTargetedShards(const boost::intrusive_ptr<ExpressionCont
                                        const SplitPipeline& splitPipeline,
                                        boost::optional<ShardedExchangePolicy> exchangeSpec,
                                        bool needsMerge,
+                                       boost::optional<ExplainOptions::Verbosity> explain,
                                        boost::optional<BSONObj> readConcern = boost::none);
 
 /**
@@ -207,6 +222,11 @@ std::unique_ptr<Pipeline, PipelineDeleter> attachCursorToPipeline(
  * beginning with that DocumentSourceMergeCursors stage. Note that one of the 'remote' cursors might
  * be this node itself.
  *
+ * Even if the ExpressionContext indicates that this operation is explain, this function still
+ * dispatches the pipeline as a non-explain, since it must open cursors on the remote nodes and
+ * merge them with a $mergeCursors. If the caller's intent is to dispatch an explain command, it
+ * must use a different helper.
+ *
  * Use the AggregateCommandRequest alternative for 'targetRequest' to explicitly specify command
  * options (e.g. read concern) to the shards when establishing remote cursors. Note that doing so
  * incurs the cost of parsing the pipeline.
diff --git a/src/mongo/db/query/plan_explainer_impl.cpp b/src/mongo/db/query/plan_explainer_impl.cpp
index 23ab0939ee6..68db5567105 100644
--- a/src/mongo/db/query/plan_explainer_impl.cpp
+++ b/src/mongo/db/query/plan_explainer_impl.cpp
@@ -672,6 +672,9 @@ void PlanExplainerImpl::getSummaryStats(PlanSummaryStats* statsOut) const {
 
     statsOut->totalKeysExamined = 0;
     statsOut->totalDocsExamined = 0;
+    statsOut->indexesUsed.clear();
+    statsOut->collectionScans = 0;
+    statsOut->collectionScansNonTailable = 0;
 
     for (size_t i = 0; i < stages.size(); i++) {
         statsOut->totalKeysExamined +=
diff --git a/src/mongo/db/query/plan_explainer_sbe.cpp b/src/mongo/db/query/plan_explainer_sbe.cpp
index f37382a23b6..d9422367aca 100644
--- a/src/mongo/db/query/plan_explainer_sbe.cpp
+++ b/src/mongo/db/query/plan_explainer_sbe.cpp
@@ -405,9 +405,10 @@ void PlanExplainerSBE::getSummaryStats(PlanSummaryStats* statsOut) const {
 
     // Use the pre-computed summary stats instead of traversing the QuerySolution tree.
     const auto& indexesUsed = _debugInfo->mainStats.indexesUsed;
+    statsOut->indexesUsed.clear();
     statsOut->indexesUsed.insert(indexesUsed.begin(), indexesUsed.end());
-    statsOut->collectionScans += _debugInfo->mainStats.collectionScans;
-    statsOut->collectionScansNonTailable += _debugInfo->mainStats.collectionScansNonTailable;
+    statsOut->collectionScans = _debugInfo->mainStats.collectionScans;
+    statsOut->collectionScansNonTailable = _debugInfo->mainStats.collectionScansNonTailable;
 }
 
 void PlanExplainerSBE::getSecondarySummaryStats(std::string secondaryColl,
diff --git a/src/mongo/s/query/cluster_aggregation_planner.cpp b/src/mongo/s/query/cluster_aggregation_planner.cpp
index bf9af2e0be1..cb4534b1db1 100644
--- a/src/mongo/s/query/cluster_aggregation_planner.cpp
+++ b/src/mongo/s/query/cluster_aggregation_planner.cpp
@@ -386,6 +386,9 @@ DispatchShardPipelineResults dispatchExchangeConsumerPipeline(
     const NamespaceString& executionNss,
     Document serializedCommand,
     DispatchShardPipelineResults* shardDispatchResults) {
+    tassert(7163600,
+            "dispatchExchangeConsumerPipeline() must not be called for explain operation",
+            !expCtx->explain);
     auto opCtx = expCtx->opCtx;
 
     if (MONGO_unlikely(shardedAggregateFailToDispatchExchangeConsumerPipeline.shouldFail())) {
@@ -422,7 +425,8 @@ DispatchShardPipelineResults dispatchExchangeConsumerPipeline(
                                                                 serializedCommand,
                                                                 consumerPipelines.back(),
                                                                 boost::none, /* exchangeSpec */
-                                                                false /* needsMerge */);
+                                                                false /* needsMerge */,
+                                                                boost::none /* explain */);
 
         requests.emplace_back(shardDispatchResults->exchangeSpec->consumerShards[idx],
                               consumerCmdObj);
@@ -691,7 +695,8 @@ Status dispatchPipelineAndMerge(OperationContext* opCtx,
                                                    hasChangeStream,
                                                    startsWithDocuments,
                                                    eligibleForSampling,
-                                                   std::move(targeter.pipeline));
+                                                   std::move(targeter.pipeline),
+                                                   expCtx->explain);
 
     // If the operation is an explain, then we verify that it succeeded on all targeted
     // shards, write the results to the output builder, and return immediately.