Aggregation pipeline

The aggregation pipeline is MongoDB's primary data transformation language. A pipeline is an ordered list of stages; each stage consumes documents from the previous stage and produces documents for the next. The implementation lives at src/mongo/db/pipeline/, with execution under src/mongo/db/exec/agg/ and shared expression evaluation in src/mongo/db/exec/expression/.

Purpose

Pipelines provide:

• A composable transformation language richer than find.
• Stage-level optimizations (predicate pushdown, sort/limit reordering, index intersection).
• Cross-collection operations via $lookup and $unionWith.
• Streaming results for change streams (which are a special pipeline) and collation-aware operators.

A pipeline like:

db.orders.aggregate([
  { $match: { status: 'shipped' } },
  { $group: { _id: '$region', total: { $sum: '$amount' } } },
  { $sort: { total: -1 } },
  { $limit: 5 },
]);

…is parsed into a Pipeline of DocumentSources, optimized, and then executed.

Stages

Each stage is a DocumentSource subclass under src/mongo/db/pipeline/. Common categories:

Category	Examples
Filtering	$match, $redact, $sample
Projection	$project, $replaceRoot, $set, $unset, $addFields
Grouping / accumulation	$group, $bucket, $bucketAuto, $count
Joins	$lookup, $graphLookup, $unionWith
Reshaping	$unwind, $facet, $densify, $fill
Window functions	$setWindowFields (see src/mongo/db/pipeline/window_function/)
Sorting / paging	$sort, $limit, $skip
Output	$out, $merge
Diagnostic	$indexStats, $collStats, $planCacheStats
Change streams	$changeStream family (see Change streams)
Search	$search, $searchMeta, $vectorSearch (Atlas Search; in src/mongo/db/pipeline/search/)
Time-series rewrites	$_internalUnpackBucket (see Time-series)

A new stage is implemented by subclassing DocumentSource, providing a parse() factory, and registering with REGISTER_DOCUMENT_SOURCE. The MongoDB Manual is the canonical reference for stage semantics.

Optimization

The pipeline optimizer in src/mongo/db/pipeline/optimization/ applies a series of rewrites:

• Coalescing — adjacent compatible stages merge ($match + $match, $project + $project).
• Pushdown — $match moves before $lookup when independent of the joined data.
• Index hints — $match followed by $sort may be served by an index that already orders the collection.
• Pipeline absorption — early stages can be absorbed into the underlying find/PlanStage execution.
• SBE lowering — eligible prefixes lower into the SBE engine for better per-document cost.

The optimizer is the largest source of complexity — many tickets concern subtle interactions between rewrites.

Execution

Once optimized, a pipeline is executed by the agg execution engine in src/mongo/db/exec/agg/. Each DocumentSource is wrapped by an executor stage that pulls documents from its child via getNext(). The engine integrates with:

• The document/value model (src/mongo/db/exec/document_value/) — a hashmap-style document representation distinct from BSON, optimized for in-memory transformation.
• The expression evaluator (src/mongo/db/exec/expression/) — handles the full $expr/$add/$concat/$function family.
• The process interface (src/mongo/db/pipeline/process_interface/) — abstracts shard- vs router-side capabilities so the same DocumentSource can run on either.

Spilling

Stages that aggregate large data sets ($group, $sort, $setWindowFields, $bucketAuto) can spill to disk via src/mongo/db/pipeline/spilling/. The spill machinery is shared with the SBE engine and uses the storage engine's record store as a temporary backing store.

Cluster execution

In a sharded cluster a pipeline runs in two halves:

graph LR
    subgraph Mongos
        Router[ClusterPipelineRouter]
        MergePart[Merge part]
    end
    subgraph ShardA
        SAprefix[Shard prefix]
    end
    subgraph ShardB
        SBprefix[Shard prefix]
    end

    Router --> SAprefix
    Router --> SBprefix
    SAprefix --> MergePart
    SBprefix --> MergePart
    MergePart --> Client

The router decides which prefix of the pipeline runs on each shard and which merger runs on mongos. The split is computed by ClusterAggregate::splitPipeline in src/mongo/s/. Many stages are split-aware: a $group can run partial-aggregation on shards and final aggregation on the merger.

$lookup and $unionWith

$lookup and $unionWith perform cross-collection or cross-cluster reads. They go through the process interface to issue secondary commands; on a sharded cluster they may target the foreign collection's owning shards directly.

Window functions

$setWindowFields introduces SQL-style window functions (SUM OVER ..., AVG OVER ..., RANK). The implementation under src/mongo/db/pipeline/window_function/ maintains sliding-window state and supports the standard partition/order/frame semantics.

Search

$search, $searchMeta, and $vectorSearch integrate with Atlas Search via a sidecar process (mongot). The plumbing is in src/mongo/db/pipeline/search/ and src/mongo/db/query/search/.

Key source files

File	Purpose
src/mongo/db/pipeline/pipeline.cpp	The Pipeline container and orchestration.
src/mongo/db/pipeline/document_source.h	Base class for stages.
src/mongo/db/pipeline/document_source_*.cpp	One file per built-in stage.
src/mongo/db/pipeline/optimization/	Pipeline rewriter.
src/mongo/db/exec/agg/	Pipeline execution engine.
src/mongo/db/exec/document_value/	Document/Value runtime model.
src/mongo/db/exec/expression/	Expression evaluator.
src/mongo/db/pipeline/window_function/	Window function implementations.
src/mongo/db/pipeline/spilling/	Spill-to-disk machinery.

Integration points

• The query engine handles find/update/delete and is the source for many aggregation prefixes.
• The change streams feature is a special pipeline rooted at $_internalChangeStream*.
• Time-series collections inject internal stages ($_internalUnpackBucket) to unpack buckets into measurement documents.
• Sharding splits pipelines across shards.

Entry points for modification

Adding a stage means a new document_source_<name>.cpp plus a REGISTER_DOCUMENT_SOURCE call. Adding an expression operator means a new subclass of Expression and a REGISTER_EXPRESSION registration. Optimizer rewrites land in src/mongo/db/pipeline/optimization/ — they should be paired with explicit tests because their interactions are subtle. The aggregation resmoke suite plus the aggregation_* jstests are the primary CI gate.