diff options
Diffstat (limited to 'src/mongo/s/query/cluster_aggregation_planner.cpp')
| -rw-r--r-- | src/mongo/s/query/cluster_aggregation_planner.cpp | 53 |
1 files changed, 39 insertions, 14 deletions
diff --git a/src/mongo/s/query/cluster_aggregation_planner.cpp b/src/mongo/s/query/cluster_aggregation_planner.cpp index 3f970f2f8c1..e124b63c4f9 100644 --- a/src/mongo/s/query/cluster_aggregation_planner.cpp +++ b/src/mongo/s/query/cluster_aggregation_planner.cpp @@ -65,6 +65,7 @@ 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; @@ -171,9 +172,13 @@ 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. - invariant(shardDispatchResults.splitPipeline); + tassert(6525900, + "tried to dispatch merge pipeline but the pipeline was not split", + shardDispatchResults.splitPipeline); auto* mergePipeline = shardDispatchResults.splitPipeline->mergePipeline.get(); - invariant(mergePipeline); + tassert(6525901, + "tried to dispatch merge pipeline but there was no merge portion of the split pipeline", + mergePipeline); auto* opCtx = expCtx->opCtx; std::vector<ShardId> targetedShards; @@ -232,9 +237,13 @@ Status dispatchMergingPipeline(const boost::intrusive_ptr<ExpressionContext>& ex privileges, expCtx->tailableMode)); - // 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. + // 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. auto mergeCursors = static_cast<DocumentSourceMergeCursors*>(mergePipeline->peekFront()); mergeCursors->dismissCursorOwnership(); @@ -374,6 +383,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())) { @@ -410,7 +422,8 @@ DispatchShardPipelineResults dispatchExchangeConsumerPipeline( serializedCommand, consumerPipelines.back(), boost::none, /* exchangeSpec */ - false /* needsMerge */); + false /* needsMerge */, + boost::none /* explain */); requests.emplace_back(shardDispatchResults->exchangeSpec->consumerShards[idx], consumerCmdObj); @@ -435,8 +448,11 @@ DispatchShardPipelineResults dispatchExchangeConsumerPipeline( SplitPipeline splitPipeline{nullptr, std::move(mergePipeline), boost::none}; - // Relinquish ownership of the local consumer pipelines' cursors as each shard is now - // responsible for its own producer cursors. + // 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. for (const auto& pipeline : consumerPipelines) { const auto& mergeCursors = static_cast<DocumentSourceMergeCursors*>(pipeline.shardsPipeline->peekFront()); @@ -563,6 +579,7 @@ AggregationTargeter AggregationTargeter::make( boost::optional<ChunkManager> cm, stdx::unordered_set<NamespaceString> involvedNamespaces, bool hasChangeStream, + bool startsWithDocuments, bool allowedToPassthrough, bool perShardCursor) { if (perShardCursor) { @@ -583,10 +600,11 @@ AggregationTargeter AggregationTargeter::make( // Determine whether this aggregation must be dispatched to all shards in the cluster. const bool mustRunOnAll = - sharded_agg_helpers::mustRunOnAllShards(executionNss, hasChangeStream); + sharded_agg_helpers::mustRunOnAllShards(executionNss, hasChangeStream, startsWithDocuments); - // If we don't have a routing table, then this is a $changeStream which must run on all shards. - invariant(cm || (mustRunOnAll && 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 || (mustRunOnAll && hasChangeStream) || (startsWithDocuments && !mustRunOnAll)); // A pipeline is allowed to passthrough to the primary shard iff the following conditions are // met: @@ -663,11 +681,16 @@ Status dispatchPipelineAndMerge(OperationContext* opCtx, const ClusterAggregate::Namespaces& namespaces, const PrivilegeVector& privileges, BSONObjBuilder* result, - bool hasChangeStream) { + bool hasChangeStream, + bool startsWithDocuments) { 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, std::move(targeter.pipeline)); + auto shardDispatchResults = + sharded_agg_helpers::dispatchShardPipeline(serializedCommand, + hasChangeStream, + startsWithDocuments, + 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. @@ -701,6 +724,8 @@ 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, |