Aggregation
Pipeline Intro

The Aggregation Pipeline is MongoDB's framework for data processing and transformation. It is modeled as a conveyor belt: documents enter the pipeline, pass through a series of sequential stages, and emerge as transformed output.

Each stage takes documents as input, applies one specific operation (filter, reshape, group, sort, join, etc.), and passes the result to the next stage.

Basic Syntax

db.collection.aggregate([
  { $stage1: { /* options */ } },
  { $stage2: { /* options */ } },
  { $stage3: { /* options */ } }
])
// Array of stage objects — executed top to bottom
// Output of each stage is the input of the next

Why Use the Aggregation Pipeline?

• Group documents and compute summaries (GROUP BY equivalent)
• Join multiple collections (LEFT JOIN equivalent)
• Compute derived fields with expressions
• Reshape documents — include, exclude, rename, compute fields
• Deconstruct arrays into individual documents
• Date/time extraction and formatting
• Statistical operations (standard deviation, etc.)

Execution is strictly sequential — top to bottom. The output of one stage becomes the direct input of the next. No stage has visibility into stages that follow it.

Key Implications

• Stages that reduce data volume ($match) should appear as early as possible
• Expensive operations ($lookup, $unwind) should operate on the smallest possible dataset
• Stage order is the primary performance lever — a misplaced $match can turn a 10ms query into a 10-second query
• Computed fields created in a stage are not available in prior stages

Recommended Stage Order

// Optimal pipeline ordering pattern
db.orders.aggregate([
  { $match: { ... } },        // 1. Filter first — enables index usage
  { $sort: { ... } },         // 2. Sort early if needed for $first/$last
  { $skip: N },               // 3. Skip (pagination)
  { $limit: N },              // 4. Limit — Top-K optimization with $sort
  { $lookup: { ... } },       // 5. Join — on smallest filtered set
  { $unwind: "$arr" },        // 6. Unwind — after filtering, before group
  { $group: { ... } },        // 7. Group — on reduced dataset
  { $project: { ... } }       // 8. Project last — shape final output
])

Index Usage in the Pipeline

Indexes are exploitable only in the earliest stages, before any transformation occurs. Once documents have been transformed (e.g., field renamed via $project), the original index on that field can no longer be used.

// Index on "status" IS used — $match is first
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $group: { _id: "$category", total: { $sum: "$amount" } } }
])

// Index on "status" is NOT used — $match is after $group
db.orders.aggregate([
  { $group: { _id: "$category", total: { $sum: "$amount" } } },
  { $match: { status: "completed" } }   // scans ALL grouped results
])

db.collection.aggregate(
  [ /* pipeline stages */ ],
  { /* options object */ }
)

Common Option Combinations

// Production-safe large aggregation
db.orders.aggregate(
  [ { $match: { ... } }, { $group: { ... } } ],
  {
    allowDiskUse: true,      // handle large datasets safely
    maxTimeMS: 30000,        // fail after 30 seconds
    comment: "monthly-sales-report"  // visible in slow query log
  }
)

// Debugging with explain
db.orders.aggregate(
  [ { $match: { status: "completed" } } ]
).explain("executionStats")

100MB Per-Stage RAM Limit

Any single pipeline stage ($group, $sort, $bucket, etc.) is limited to 100MB of RAM. Exceeding this limit throws an error unless allowDiskUse: true is set.

// Error: Exceeded memory limit for $group stage
// Fix: enable disk use
db.events.aggregate(
  [ { $group: { _id: "$userId", count: { $sum: 1 } } } ],
  { allowDiskUse: true }
)
// allowDiskUse: true allows temporary spill to disk at a performance cost
// This is a per-stage limit — two $sort stages = two separate 100MB limits

16MB BSON Document Limit

Any single document at any point during pipeline execution — including intermediate documents — cannot exceed 16MB. $lookup is the most common cause when the joined array is very large.

// PROBLEM — joining 50,000 orders into a user doc exceeds 16MB
db.users.aggregate([
  { $lookup: { from: "orders", localField: "_id", foreignField: "userId", as: "orders" } }
])

// FIX — pre-filter inside $lookup pipeline to keep the embedded array small
db.users.aggregate([
  {
    $lookup: {
      from: "orders",
      let: { uid: "$_id" },
      pipeline: [
        { $match: { $expr: { $eq: ["$userId", "$$uid"] }, status: "active" } },
        { $limit: 10 }
      ],
      as: "recentOrders"
    }
  }
])

Null and Missing Field Handling Across Stages

$expr — Using Aggregation Expressions in find()

// Standard find() cannot compare two fields from the same document
// $expr bridges aggregation expressions into query context
db.orders.find({
  $expr: { $gt: ["$sellingPrice", "$costPrice"] }
})
// Finds orders where selling price > cost price
// Also works inside $match in an aggregation pipeline

// Find top 3 product categories by total revenue from completed orders
db.orders.aggregate([

  // Stage 1: Only completed orders — uses index on "status", reduces set early
  { $match: { status: "completed" } },

  // Stage 2: Group by category, compute totals
  {
    $group: {
      _id: "$category",
      totalRevenue: { $sum: "$amount" },
      orderCount:   { $sum: 1 }
    }
  },

  // Stage 3: Sort by highest revenue — Top-K optimization with $limit next
  { $sort: { totalRevenue: -1 } },

  // Stage 4: Keep only top 3 — triggers priority queue, not full-sort memory
  { $limit: 3 },

  // Stage 5: Rename _id to "category" for clean output
  {
    $project: {
      _id: 0,
      category:     "$_id",
      totalRevenue: 1,
      orderCount:   1
    }
  }

],
{ allowDiskUse: true, maxTimeMS: 15000, comment: "top-3-categories" })

// Sample output
[
  { "category": "Electronics", "totalRevenue": 84200, "orderCount": 312 },
  { "category": "Furniture",   "totalRevenue": 61500, "orderCount": 198 },
  { "category": "Clothing",    "totalRevenue": 43800, "orderCount": 540 }
]