How is an aggregation pipeline split between the shards and a merging node in a sharded cluster?

## Why a pipeline has to be split An aggregation over a sharded collection reads data that lives on many shards, but stages like `$group`, `$sort` and `$limit` are defined over the **whole** result. So the planner divides the pipeline into two halves and runs them in different places. - The **shards part** runs on every participating shard, in parallel, against that shard's own data. The planner pushes as much down here as it safely can: the leading `$match` (which also determines *which* shards participate at all), `$project` to cut document size early, and a partial form of accumulating stages. - The **merger part** runs on a single node. It combines what the shards produced — finishing a `$group` by merging partial accumulator states, applying the final ordering, then the skip and limit. `$group` is the interesting case: it appears on **both** sides. Each shard groups its own documents locally and emits partial results; the merger combines the partials for the same key. That is what makes cross-shard aggregation scale at all — shipping raw documents to one node would not. ## Reading the split from explain Run `explain()` on the aggregation through `mongos`. The output contains a `splitPipeline` object with: - **`shardsPart`** — the stages executed on each shard. - **`mergerPart`** — the stages executed on the merging node. - **`mergeType`** — where the merger half runs. Alongside it, a `shards` map shows each participating shard's own query plan, so you can check index usage per shard the same way you would for a `find()`. This output is the fastest way to answer the two questions that matter in a review: *did my `$match` actually get pushed to the shards*, and *did the pipeline fan out at all*. ## When there is no split If the leading `$match` contains the shard key, the router targets one shard and ships the **entire** pipeline there. `splitPipeline` is then null and there is no merger half: no partial groups, no cross-shard transfer of intermediate results, no merge-node bottleneck. This is by far the most effective optimisation available for a sharded aggregation, and it is a modelling and query-shape decision, not a tuning knob. ## What pins the merge to a particular shard Normally the merger half can run on any participating shard. Some pipelines cannot, because a stage needs access to an **unsharded** collection, which lives on the database's **primary shard**: - `$lookup` whose foreign collection is unsharded. - `$out`, which writes its results into a collection. Those force the merge onto the primary shard, which turns one node into a serialisation point for the whole pipeline's output. On a heavy pipeline that shows up as one shard pinned at high CPU while the others idle after finishing their halves. ## $out versus $merge `$out` **cannot write to a sharded collection**. `$merge` can, and it can also insert, replace or update matched documents rather than replacing a whole collection. For materialising a cross-shard result at any real scale, `$merge` is the stage you want: it writes into a sharded target, and it lets you refresh a summary incrementally instead of rebuilding it. This is the shape of the standard fix for an expensive cross-shard pipeline on a request path: run it in the background, `$merge` the result into a collection whose shard key matches how the UI reads it, and serve the request with a targeted `find()`. ## $lookup across shards Recent MongoDB versions (5.1 and later) allow the foreign collection of a `$lookup` to be sharded. That removes an old hard restriction, but it does not make the join cheap: there is no co-location guarantee, because the two collections have different shard keys, so resolving the join for a given input document may itself require reaching other shards. Treat a large cross-shard `$lookup` as a fan-out per input document and design around it — by embedding the needed fields, or by materialising the joined shape — rather than assuming the planner will make it local. ## What to take into an interview Three points carry the answer: the pipeline is split into a shards half and a merger half; `explain()` names both halves and where the merger runs; and a leading `$match` on the shard key removes the split entirely. Add the `$out`-versus-`$merge` distinction and the primary-shard pinning, and you have covered what an interviewer is actually probing — whether you have looked at a sharded pipeline's explain output rather than only written pipelines.