1 files changed, 26 insertions, 61 deletions

diff --git a/src/mongo/s/query/cluster_aggregation_planner.cpp b/src/mongo/s/query/cluster_aggregation_planner.cpp
index 70adf70404a..3f970f2f8c1 100644
--- a/src/mongo/s/query/cluster_aggregation_planner.cpp
+++ b/src/mongo/s/query/cluster_aggregation_planner.cpp
@@ -65,7 +65,6 @@ namespace cluster_aggregation_planner {
 
 MONGO_FAIL_POINT_DEFINE(shardedAggregateFailToDispatchExchangeConsumerPipeline);
 MONGO_FAIL_POINT_DEFINE(shardedAggregateFailToEstablishMergingShardCursor);
-MONGO_FAIL_POINT_DEFINE(shardedAggregateHangBeforeDispatchMergingPipeline);
 
 using sharded_agg_helpers::DispatchShardPipelineResults;
 using sharded_agg_helpers::SplitPipeline;
@@ -172,13 +171,9 @@ Status dispatchMergingPipeline(const boost::intrusive_ptr<ExpressionContext>& ex
                                const PrivilegeVector& privileges,
                                bool hasChangeStream) {
     // We should never be in a situation where we call this function on a non-merge pipeline.
-    tassert(6525900,
-            "tried to dispatch merge pipeline but the pipeline was not split",
-            shardDispatchResults.splitPipeline);
+    invariant(shardDispatchResults.splitPipeline);
     auto* mergePipeline = shardDispatchResults.splitPipeline->mergePipeline.get();
-    tassert(6525901,
-            "tried to dispatch merge pipeline but there was no merge portion of the split pipeline",
-            mergePipeline);
+    invariant(mergePipeline);
     auto* opCtx = expCtx->opCtx;
 
     std::vector<ShardId> targetedShards;
@@ -237,13 +232,9 @@ Status dispatchMergingPipeline(const boost::intrusive_ptr<ExpressionContext>& ex
                             privileges,
                             expCtx->tailableMode));
 
-    // If the mergingShard returned an error and did not accept ownership it is our responsibility
-    // to kill the cursors.
-    uassertStatusOK(getStatusFromCommandResult(mergeResponse.swResponse.getValue().data));
-
-    // If we didn't get an error from the merging shard, ownership for the shard cursors has been
-    // transferred to the merging shard. Dismiss the ownership in the current merging pipeline such
-    // that when it goes out of scope it does not attempt to kill the cursors.
+    // Ownership for the shard cursors has been transferred to the merging shard. Dismiss the
+    // ownership in the current merging pipeline such that when it goes out of scope it does not
+    // attempt to kill the cursors.
     auto mergeCursors = static_cast<DocumentSourceMergeCursors*>(mergePipeline->peekFront());
     mergeCursors->dismissCursorOwnership();
 
@@ -345,27 +336,12 @@ BSONObj establishMergingMongosCursor(OperationContext* opCtx,
         responseBuilder.setPostBatchResumeToken(ccc->getPostBatchResumeToken());
     }
 
-    bool exhausted = cursorState != ClusterCursorManager::CursorState::NotExhausted;
-    int nShards = ccc->getNumRemotes();
-
-    auto&& opDebug = CurOp::get(opCtx)->debug();
-    // Fill out the aggregation metrics in CurOp, and record queryStats metrics, before detaching
-    // the cursor from its opCtx.
-    opDebug.nShards = std::max(opDebug.nShards, nShards);
-    opDebug.cursorExhausted = exhausted;
-    opDebug.additiveMetrics.nBatches = 1;
-    CurOp::get(opCtx)->setEndOfOpMetrics(responseBuilder.numDocs());
-    if (exhausted) {
-        collectQueryStatsMongos(opCtx, ccc->takeKey());
-    } else {
-        collectQueryStatsMongos(opCtx, ccc);
-    }
-
     ccc->detachFromOperationContext();
 
+    int nShards = ccc->getNumRemotes();
     CursorId clusterCursorId = 0;
 
-    if (!exhausted) {
+    if (cursorState == ClusterCursorManager::CursorState::NotExhausted) {
         auto authUsers = AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames();
         clusterCursorId = uassertStatusOK(Grid::get(opCtx)->getCursorManager()->registerCursor(
             opCtx,
@@ -374,9 +350,16 @@ BSONObj establishMergingMongosCursor(OperationContext* opCtx,
             ClusterCursorManager::CursorType::MultiTarget,
             ClusterCursorManager::CursorLifetime::Mortal,
             authUsers));
-        opDebug.cursorid = clusterCursorId;
     }
 
+    // Fill out the aggregation metrics in CurOp.
+    if (clusterCursorId > 0) {
+        CurOp::get(opCtx)->debug().cursorid = clusterCursorId;
+    }
+    CurOp::get(opCtx)->debug().nShards = std::max(CurOp::get(opCtx)->debug().nShards, nShards);
+    CurOp::get(opCtx)->debug().cursorExhausted = (clusterCursorId == 0);
+    CurOp::get(opCtx)->debug().nreturned = responseBuilder.numDocs();
+
     responseBuilder.done(clusterCursorId, requestedNss.ns());
 
     auto bodyBuilder = replyBuilder.getBodyBuilder();
@@ -391,9 +374,6 @@ 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())) {
@@ -430,8 +410,7 @@ DispatchShardPipelineResults dispatchExchangeConsumerPipeline(
                                                                 serializedCommand,
                                                                 consumerPipelines.back(),
                                                                 boost::none, /* exchangeSpec */
-                                                                false /* needsMerge */,
-                                                                boost::none /* explain */);
+                                                                false /* needsMerge */);
 
         requests.emplace_back(shardDispatchResults->exchangeSpec->consumerShards[idx],
                               consumerCmdObj);
@@ -456,11 +435,8 @@ DispatchShardPipelineResults dispatchExchangeConsumerPipeline(
 
     SplitPipeline splitPipeline{nullptr, std::move(mergePipeline), boost::none};
 
-    // Relinquish ownership of the consumer pipelines' cursors. These cursors are now set up to be
-    // merged by a set of $mergeCursors pipelines that we just dispatched to the shards above. Now
-    // that we've established those pipelines on the shards, we are no longer responsible for
-    // ensuring they are cleaned up. If there was a problem establishing the cursors then
-    // establishCursors() would have thrown and mongos would kill all the consumer cursors itself.
+    // Relinquish ownership of the local consumer pipelines' cursors as each shard is now
+    // responsible for its own producer cursors.
     for (const auto& pipeline : consumerPipelines) {
         const auto& mergeCursors =
             static_cast<DocumentSourceMergeCursors*>(pipeline.shardsPipeline->peekFront());
@@ -587,7 +563,6 @@ AggregationTargeter AggregationTargeter::make(
     boost::optional<ChunkManager> cm,
     stdx::unordered_set<NamespaceString> involvedNamespaces,
     bool hasChangeStream,
-    bool startsWithDocuments,
     bool allowedToPassthrough,
     bool perShardCursor) {
     if (perShardCursor) {
@@ -607,13 +582,11 @@ AggregationTargeter AggregationTargeter::make(
     }();
 
     // Determine whether this aggregation must be dispatched to all shards in the cluster.
-    const bool mustRunOnAllShards = sharded_agg_helpers::checkIfMustRunOnAllShards(
-        executionNss, hasChangeStream, startsWithDocuments);
+    const bool mustRunOnAll =
+        sharded_agg_helpers::mustRunOnAllShards(executionNss, hasChangeStream);
 
-    // If we don't have a routing table, then this is either a $changeStream which must run on all
-    // shards or a $documents stage which must not.
-    invariant(cm || (mustRunOnAllShards && hasChangeStream) ||
-              (startsWithDocuments && !mustRunOnAllShards));
+    // If we don't have a routing table, then this is a $changeStream which must run on all shards.
+    invariant(cm || (mustRunOnAll && hasChangeStream));
 
     // A pipeline is allowed to passthrough to the primary shard iff the following conditions are
     // met:
@@ -625,7 +598,7 @@ AggregationTargeter AggregationTargeter::make(
     //    $currentOp.
     // 4. Doesn't need transformation via DocumentSource::serialize(). For example, list sessions
     //    needs to include information about users that can only be deduced on mongos.
-    if (cm && !cm->isSharded() && !mustRunOnAllShards && allowedToPassthrough &&
+    if (cm && !cm->isSharded() && !mustRunOnAll && allowedToPassthrough &&
         !involvesShardedCollections) {
         return AggregationTargeter{TargetingPolicy::kPassthrough, nullptr, cm};
     } else {
@@ -690,16 +663,11 @@ Status dispatchPipelineAndMerge(OperationContext* opCtx,
                                 const ClusterAggregate::Namespaces& namespaces,
                                 const PrivilegeVector& privileges,
                                 BSONObjBuilder* result,
-                                bool hasChangeStream,
-                                bool startsWithDocuments) {
+                                bool hasChangeStream) {
     auto expCtx = targeter.pipeline->getContext();
     // If not, split the pipeline as necessary and dispatch to the relevant shards.
-    auto shardDispatchResults =
-        sharded_agg_helpers::dispatchShardPipeline(serializedCommand,
-                                                   hasChangeStream,
-                                                   startsWithDocuments,
-                                                   std::move(targeter.pipeline),
-                                                   expCtx->explain);
+    auto shardDispatchResults = sharded_agg_helpers::dispatchShardPipeline(
+        serializedCommand, hasChangeStream, std::move(targeter.pipeline));
 
     // 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.
@@ -733,8 +701,6 @@ Status dispatchPipelineAndMerge(OperationContext* opCtx,
             expCtx, namespaces.executionNss, serializedCommand, &shardDispatchResults);
     }
 
-    shardedAggregateHangBeforeDispatchMergingPipeline.pauseWhileSet();
-
     // If we reach here, we have a merge pipeline to dispatch.
     return dispatchMergingPipeline(expCtx,
                                    namespaces,
@@ -867,7 +833,6 @@ Status runPipelineOnSpecificShardOnly(const boost::intrusive_ptr<ExpressionConte
     if (explain) {
         // If this was an explain, then we get back an explain result object rather than a cursor.
         result = response.swResponse.getValue().data;
-        collectQueryStatsMongos(opCtx, std::move(CurOp::get(opCtx)->debug().queryStatsInfo.key));
     } else {
         result = uassertStatusOK(storePossibleCursor(
             opCtx,