Aggregation Pipeline

What is aggregation?

Aggregation is MongoDB's analytics engine. Documents flow through a pipeline of stages — each stage transforms the stream and passes results to the next.

Think SQL: $match ≈ WHERE, $group ≈ GROUP BY, $sort ≈ ORDER BY, $project ≈ SELECT columns, $lookup ≈ JOIN.

Real-life example: An aggregation pipeline is a factory assembly line — raw orders enter, get filtered, grouped, labelled, and boxed reports come out the end.

Your first pipeline: $match → $group → $sort

Start every analytics query by trimming documents early with $match — less data for later stages means faster runs.

$group collapses rows by a key (_id) and computes totals with accumulators like $sum and $avg.

Real-life example: A Rishtaara shop owner asks 'Which category sold the most this month?' — three stages answer that without exporting CSV files.

db.orders.aggregate([
  // Stage 1: only completed orders in July
  {
    $match: {
      status: "completed",
      createdAt: {
        $gte: ISODate("2025-07-01"),
        $lt: ISODate("2025-08-01")
      }
    }
  },
  // Stage 2: one row per category
  {
    $group: {
      _id: "$category",
      totalRevenue: { $sum: "$total" },
      orderCount: { $sum: 1 },
      avgOrder: { $avg: "$total" }
    }
  },
  // Stage 3: bestsellers first
  { $sort: { totalRevenue: -1 } }
])

Pipeline mental model

Each stage receives documents from the previous stage and outputs a new stream. Order matters: $match before $group is almost always cheaper than grouping first.

The aggregation cursor returns batches — use allowDiskUse: true for heavy pipelines that might exceed memory limits.

Real-life example: Putting $match first is like filtering rice before cooking — you do not boil the whole sack then throw away stones.

• Filter early with $match near the start
• Shape output with $project or $addFields after computing fields
• Use $limit / $skip for pagination after $sort
• Test stages one at a time in mongosh — comment out later stages while debugging

Updates with aggregation pipeline

Since MongoDB 4.2, updateOne and updateMany accept an aggregation pipeline as the update body — not just $set/$inc objects.

This lets you compute new field values from existing fields, conditionally branch, and reshape arrays in one atomic update.

Real-life example: Pipeline updates are a smart cash register — it reads old price, applies tax rules, and writes new totals in one transaction.

// Recalculate line totals and order total from items array
db.orders.updateOne(
  { _id: orderId },
  [
    {
      $set: {
        items: {
          $map: {
            input: "$items",
            as: "line",
            in: {
              $mergeObjects: [
                "$$line",
                {
                  subtotal: {
                    $multiply: ["$$line.qty", "$$line.price"]
                  }
                }
              ]
            }
          }
        }
      }
    },
    {
      $set: {
        total: { $sum: "$items.subtotal" },
        updatedAt: new Date()
      }
    }
  ]
)

// Promote to VIP when lifetime spend crosses threshold
db.users.updateMany(
  { lifetimeSpend: { $gte: 50000 } },
  [
    {
      $set: {
        tier: "VIP",
        perks: { $concatArrays: [{ $ifNull: ["$perks", []] }, ["free_shipping"]] }
      }
    }
  ]
)