Aggregation Pipeline

MongoDB Aggregation Pipeline

find() can filter and project documents, but it can't compute anything: no totals, no grouping, no joins. For that MongoDB has the aggregation pipeline. Documents flow through a sequence of stages, and each stage transforms the stream of documents and passes the result to the next, like an assembly line or Unix pipes (cat | grep | sort | head).

orders collection ──▶ $match ──▶ $group ──▶ $sort ──▶ $limit ──▶ results
                      (filter)   (sum per    (order)    (top 10)
                                  customer)

Each stage only sees the output of the stage before it, which is why the order of stages matters for both correctness and speed.

A complete example: top customers by revenue

db.orders.aggregate([
  // 1. Filter first — like WHERE. Uses an index on {status, createdAt} if one exists.
  { $match: { status: "COMPLETED", createdAt: { $gte: ISODate("2026-01-01") } } },

  // 2. Group — like GROUP BY. _id is the grouping key; the other fields are accumulators.
  { $group: {
      _id: "$customerId",
      revenue:     { $sum: "$total" },
      orders:      { $sum: 1 },
      avgOrder:    { $avg: "$total" },
      lastOrderAt: { $max: "$createdAt" }
  }},

  // 3. Add computed fields
  { $set: { isVip: { $gte: ["$revenue", 1000] } } },

  // 4–5. Order and take the top 10
  { $sort: { revenue: -1 } },
  { $limit: 10 },

  // 6. Shape the output
  { $project: { _id: 0, customerId: "$_id", revenue: 1, orders: 1, isVip: 1 } }
])

"$total" (with a $) means the value of the field total in the current document. Without the $ it would be the literal string "total".

The stages you'll use most

A $match placed after a $group acts like SQL's HAVING: it filters on the computed totals.

$unwind: working with arrays

// { _id: 1, title: "Streams in Java", tags: ["java", "streams", "functional"] }
db.posts.aggregate([
  { $unwind: "$tags" },
  { $group: { _id: "$tags", posts: { $sum: 1 } } },     // how many posts per tag
  { $sort: { posts: -1 } }
])

$unwind turns one document with a 3-element array into 3 documents, one per tag, so the tags can be grouped. By default, documents where the array is missing or empty disappear from the pipeline. Use { $unwind: { path: "$tags", preserveNullAndEmptyArrays: true } } to keep them. Forgetting this silently drops data, for example posts with no tags vanish from a report.

$lookup: joining collections

db.orders.aggregate([
  { $match: { status: "COMPLETED" } },
  { $lookup: {
      from: "customers",           // the other collection
      localField: "customerId",    // field in orders
      foreignField: "_id",         // field in customers
      as: "customer"               // result is always an ARRAY of matches
  }},
  { $unwind: "$customer" },        // one customer per order → flatten the array
  { $project: { total: 1, "customer.name": 1, "customer.email": 1 } }
])

$lookup is a left outer join: orders with no matching customer are kept, with an empty customer array (and then dropped by a plain $unwind). It runs a lookup for every input document, so:

• index foreignField in the other collection, or each lookup scans it;
• $match before $lookup so you join as few documents as possible;
• if you join the same data on every read, reconsider the schema. In MongoDB, data that's read together should usually be embedded together (see Schema Design Patterns).

Pagination with a total count: $facet

A common API need is "page 3 of the results, and how many results there are in total". $facet runs both over the same filtered input in one round trip:

db.products.aggregate([
  { $match: { category: "electronics", status: "ACTIVE" } },
  { $facet: {
      items: [ { $sort: { price: 1 } }, { $skip: 40 }, { $limit: 20 } ],
      total: [ { $count: "count" } ]
  }}
])
// → { items: [...20 products...], total: [ { count: 137 } ] }

Note that $skip still walks past all skipped documents, so deep pages get slower. For very deep or infinite scrolling, range-based pagination ("give me 20 items with price > the last price I saw") scales better.

Performance rules

1. Put $match (and $sort) first. Only stages at the start of the pipeline can use indexes. Once a $group or $project has reshaped the documents, the original indexes no longer apply. The optimizer moves some stages automatically, but don't rely on it.
2. Shrink documents early. A $project that drops large unused fields reduces the memory every later stage needs.
3. Mind the 100 MB stage limit. Blocking stages like $group and $sort have a per-stage memory limit. Beyond it they must spill to disk. Since MongoDB 6.0 that's allowed by default; on older versions you pass { allowDiskUse: true }. Spilling is slow, so it's a sign to filter earlier or pre-aggregate.
4. Check the plan with db.orders.explain("executionStats").aggregate([...]). Look for IXSCAN (an index was used) rather than COLLSCAN (a full collection scan) in the first stage.
5. Pre-aggregate hot reports. A dashboard that sums a year of orders on every page load should instead read from a summary collection updated incrementally (or with $merge on a schedule).

Aggregations from Spring

Spring Data MongoDB offers two styles. For fixed pipelines, use a repository annotation:

public interface OrderRepository extends MongoRepository<Order, String> {
    @Aggregation(pipeline = {
        "{ $match: { status: 'COMPLETED' } }",
        "{ $group: { _id: '$customerId', revenue: { $sum: '$total' } } }",
        "{ $sort: { revenue: -1 } }",
        "{ $limit: ?0 }"
    })
    List<CustomerRevenue> topCustomers(int limit);
}

public record CustomerRevenue(@Id String customerId, BigDecimal revenue) {}

For pipelines built at runtime, use the fluent API with MongoTemplate:

Aggregation agg = Aggregation.newAggregation(
        Aggregation.match(Criteria.where("status").is("COMPLETED")),
        Aggregation.group("customerId").sum("total").as("revenue").count().as("orders"),
        Aggregation.sort(Sort.Direction.DESC, "revenue"),
        Aggregation.limit(10));

List<CustomerRevenue> top = mongoTemplate
        .aggregate(agg, "orders", CustomerRevenue.class)
        .getMappedResults();

Follow-up questions this topic invites — and their answers

Q: Why should $match come before $group? A: Two reasons. First, only stages at the start of the pipeline can use indexes, so an early $match can read just the matching documents instead of scanning the collection. Second, every later stage processes fewer documents, which saves CPU and memory, and $group is a blocking stage that has to hold its groups in memory.

Q: Is $lookup a sign of bad schema design? A: Not always. It's fine for occasional reports, admin screens, or joining small reference data. But if a core, high-traffic read needs a $lookup every time, the data is probably modelled like a relational database. Embedding the needed fields, or keeping a denormalized copy, usually serves MongoDB better.

Q: What happens to documents with an empty array in $unwind? A: By default they're dropped, because there's no element to emit. Set preserveNullAndEmptyArrays: true to keep them, with the field missing or null in the output. It's a common source of reports that silently under-count.

Q: Aggregation pipeline vs map-reduce? A: Map-reduce is deprecated (since MongoDB 5.0). The aggregation pipeline covers the same use cases, runs natively in the database engine instead of in JavaScript, and is much faster. For custom logic there are expression operators and, if really needed, $function.

Q: How can the results of an aggregation be stored? A: End the pipeline with $out (replaces a whole collection) or $merge (inserts or updates into an existing collection, document by document). $merge is the usual choice for incrementally maintained summary collections, such as refreshing daily revenue totals on a schedule.