Tamiz Uddin

Posted on Oct 28 Originally published at tamiz.pro

Optimizing Large-Scale MongoDB Aggregation Pipelines: A Deep Dive into Performance Engineering

#web #performance #optimizing #large

Originally published on tamiz.pro.

MongoDB's aggregation framework is powerful enough to replace entire ETL pipelines, but at scale it becomes a performance minefield. A pipeline that processes 10,000 documents in milliseconds can grind a replica set to a halt when fed 100 million. This deep dive explores how the aggregation engine executes pipelines, where bottlenecks hide, and the concrete optimization patterns that separate production-grade pipelines from prototypes.

1. How the Aggregation Engine Executes Pipelines
2. Index Strategy: The Foundation of Fast Aggregations
3. Pipeline Restructuring: Reorder, Fuse, and Eliminate
4. Memory Management: Spilling to Disk and the 100MB Limit
5. Sharded Cluster Considerations
6. Monitoring, Profiling, and Explain Plans
7. Real-World Patterns and Anti-Patterns
8. Frequently Asked Questions

1. How the Aggregation Engine Executes Pipelines

Before optimizing, you must understand what the server actually does. The aggregation pipeline is not a single monolithic operation — it's a sequence of stages, each transforming a stream of documents. The query optimizer builds an execution plan that can reorder, fuse, or eliminate stages, but only when certain conditions are met.

1.1 Pipeline Optimization Phases

MongoDB's optimizer runs in three phases:

Analysis: Determines which stages can use indexes, which can be moved earlier, and which are "blocking" (require all input before producing output).
Rewriting: Applies transformations like $match pushdown, $project elimination, and $sort + $limit fusion.
Execution: Runs the optimized plan, potentially spilling to disk for blocking stages.

1.2 Streaming vs Blocking Stages

This distinction dictates memory behavior:

Stage Type	Stages	Memory Profile
Streaming	`$match`, `$project`, `$addFields`, `$set`, `$unset`, `$lookup` (uncorrelated), `$unwind`	O(1) per document
Blocking	`$group`, `$sort`, `$bucket`, `$facet`, `$graphLookup`, `$setWindowFields`	O(n) — requires all input

Critical insight: A single blocking stage forces the entire preceding stream to materialize. Placing $match after $group is a classic anti-pattern because it prevents index usage and forces grouping of documents that will be discarded.

1.3 The Explain Plan: Your Primary Diagnostic Tool

Always start with explain("executionStats"):

// Run this in mongosh or your driver
db.orders.aggregate([
  { $match: { status: "completed", createdAt: { $gte: ISODate("2024-01-01") } } },
  { $group: { _id: "$customerId", total: { $sum: "$amount" } } },
  { $sort: { total: -1 } },
  { $limit: 10 }
]).explain("executionStats")

Key fields to examine:

executionStats.executionStages.stage: Look for COLLSCAN (collection scan) vs IXSCAN (index scan)
executionStats.executionStages.inputStage: Reveals the pipeline flow
executionStats.executionStages.usedDisk / spillToDisk: Indicates memory pressure
executionStats.executionStages.executionTimeMillis: Wall-clock time per stage
executionStats.nReturned vs executionStats.totalDocsExamined: Selectivity ratio

2. Index Strategy: The Foundation of Fast Aggregations

Indexes are the single highest-leverage optimization. The aggregation optimizer can only use an index if the pipeline's first stage is $match (or $geoNear / $search) and the filter matches an index prefix.

2.1 Compound Index Design for Aggregation

For a pipeline filtering on status and createdAt, then grouping by customerId:

// ❌ Bad: Index doesn't match filter prefix
db.orders.createIndex({ customerId: 1, status: 1, createdAt: 1 })

// ✅ Good: Filter fields first, then group/sort fields
db.orders.createIndex({ status: 1, createdAt: 1, customerId: 1 })

// ✅ Even better for this specific pipeline: include the aggregation field
db.orders.createIndex({ status: 1, createdAt: 1, customerId: 1, amount: 1 })

The last index is a covering index for this pipeline — MongoDB can satisfy the entire aggregation from the index without fetching documents.

2.2 Index Intersection Limitations

MongoDB can intersect multiple indexes for $match, but not for $group or $sort. If your pipeline needs both filtering and sorting/grouping on different fields, a single compound index is mandatory.

// These two indexes WON'T be intersected for the aggregation below
db.orders.createIndex({ status: 1 })
db.orders.createIndex({ customerId: 1 })

// Pipeline that can't use index intersection
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $group: { _id: "$customerId", count: { $sum: 1 } } },
  { $sort: { count: -1 } }
])

2.3 Partial Indexes for High-Cardinality Filters

When a filter value represents a small fraction of documents, partial indexes reduce size and maintenance cost:

db.orders.createIndex(
  { customerId: 1, createdAt: 1 },
  { partialFilterExpression: { status: "completed" } }
)

This index only contains completed orders, making it smaller and faster for the common case.

2.4 The `$lookup` Index Requirement

For $lookup to use an index on the foreign collection, the localField/foreignField must be indexed on the foreign collection:

// Orders collection: localField = "customerId"
// Customers collection: MUST have index on foreignField = "_id"
db.customers.createIndex({ _id: 1 }) // Default _id index usually suffices

// But for non-_id lookups:
db.products.createIndex({ sku: 1 })

db.orders.aggregate([
  { $match: { status: "completed" } },
  { $lookup: {
      from: "products",
      localField: "items.sku",
      foreignField: "sku",
      as: "productDetails"
    }
  }
])

3. Pipeline Restructuring: Reorder, Fuse, and Eliminate

The optimizer rewrites pipelines automatically, but only within strict rules. Understanding these rules lets you write pipelines that optimize themselves.

3.1 `$match` Pushdown: The Golden Rule

Always place $match as early as possible. The optimizer pushes $match before $project, $addFields, $lookup (uncorrelated), and $unwind, but not before $group, $sort, or $facet.

// ❌ Suboptimal: match after project
db.orders.aggregate([
  { $project: { customerId: 1, amount: 1, status: 1, createdAt: 1 } },
  { $match: { status: "completed", createdAt: { $gte: ISODate("2024-01-01") } } },
  { $group: { _id: "$customerId", total: { $sum: "$amount" } } }
])

// ✅ Optimal: match first
db.orders.aggregate([
  { $match: { status: "completed", createdAt: { $gte: ISODate("2024-01-01") } } },
  { $group: { _id: "$customerId", total: { $sum: "$amount" } } }
])

3.2 `$project` / `$addFields` Elimination

If a $project only renames or passes through fields, the optimizer removes it. But computed fields prevent elimination:

// ❌ Prevents optimization: computed field
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $addFields: { amountWithTax: { $multiply: ["$amount", 1.1] } } },
  { $group: { _id: "$customerId", total: { $sum: "$amountWithTax" } } }
])

// ✅ Better: compute in $group
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $group: {
      _id: "$customerId",
      total: { $sum: { $multiply: ["$amount", 1.1] } }
    }
  }
])

3.3 `$sort` + `$limit` Fusion

When $sort is immediately followed by $limit, MongoDB uses a top-K algorithm (priority queue) instead of full sort, reducing memory from O(n) to O(k):

// ✅ Fused: top-10 sort, O(k) memory
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $sort: { totalAmount: -1 } },
  { $limit: 10 }
])

// ❌ Not fused: $project breaks the chain
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $sort: { totalAmount: -1 } },
  { $project: { _id: 0, customerId: 1, totalAmount: 1 } },
  { $limit: 10 }
])

3.4 `$facet` Parallelization

$facet runs multiple sub-pipelines in parallel on the same input stream. Each facet gets a copy of the input — this is expensive. Use $facet only when facets share the exact same upstream work:

// ✅ Good: shared expensive match/group
db.orders.aggregate([
  { $match: { status: "completed", createdAt: { $gte: ISODate("2024-01-01") } } },
  { $facet: {
      byCustomer: [
        { $group: { _id: "$customerId", total: { $sum: "$amount" } } },
        { $sort: { total: -1 } },
        { $limit: 10 }
      ],
      byProduct: [
        { $unwind: "$items" },
        { $group: { _id: "$items.productId", total: { $sum: "$items.amount" } } },
        { $sort: { total: -1 } },
        { $limit: 10 }
      ],
      dailyTotals: [
        { $group: {
            _id: { $dateToString: { format: "%Y-%m-%d", date: "$createdAt" } },
            total: { $sum: "$amount" }
          }
        },
        { $sort: { _id: 1 } }
      ]
    }
  }
])

// ❌ Bad: facets have different match criteria — run separate aggregations
db.orders.aggregate([
  { $facet: {
      completed: [ { $match: { status: "completed" } }, { $count: "total" } ],
      pending: [ { $match: { status: "pending" } }, { $count: "total" } ]
    }
  }
])

3.5 `$lookup` Optimization: Uncorrelated vs Correlated

Uncorrelated $lookup (static pipeline): The sub-pipeline runs once, results cached. Can use indexes on foreign collection.

Correlated $lookup (uses let/pipeline with $$variables): Runs once per input document. Avoid at scale.

// ✅ Uncorrelated: efficient
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $lookup: {
      from: "customers",
      localField: "customerId",
      foreignField: "_id",
      as: "customer"
    }
  }
])

// ❌ Correlated: runs sub-pipeline per document
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $lookup: {
      from: "customers",
      let: { custId: "$customerId" },
      pipeline: [
        { $match: { $expr: { $eq: ["$_id", "$$custId"] } } },
        { $project: { name: 1, email: 1 } }
      ],
      as: "customer"
    }
  }
])

// ✅ Rewrite correlated as uncorrelated when possible
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $lookup: {
      from: "customers",
      localField: "customerId",
      foreignField: "_id",
      as: "customer"
    }
  },
  { $addFields: { customer: { $arrayElemAt: ["$customer", 0] } } },
  { $project: { "customer.name": 1, "customer.email": 1 } }
])

4. Memory Management: Spilling to Disk and the 100MB Limit

Each pipeline stage has a 100MB memory limit (configurable via internalQueryMaxBlockingSortMemoryUsageBytes for $sort, but hard-coded for $group). Exceeding it triggers spilling to disk — a dramatic performance cliff.

4.1 Detecting Spills

const result = await db.orders.aggregate([
  { $match: { status: "completed" } },
  { $group: { _id: "$customerId", total: { $sum: "$amount" } } },
  { $sort: { total: -1 } }
]).explain("executionStats")

// Check for spills
const sortStage = result.executionStats.executionStages.inputStage
console.log("Sort used disk:", sortStage.usedDisk)
console.log("Sort memory bytes:", sortStage.memoryUsageBytes)

4.2 Allow Disk Use: The Escape Hatch

For $sort and $group, you can enable disk spill explicitly:

db.orders.aggregate([
  { $match: { status: "completed" } },
  { $group: { _id: "$customerId", total: { $sum: "$amount" } } },
  { $sort: { total: -1 } }
], { allowDiskUse: true })

Warning: allowDiskUse turns OOM kills into slow queries. It's a band-aid, not a fix. The real fix is reducing the working set.

4.3 Reducing Working Set Size

4.3.1 Early `$project` to Drop Fields

// ❌ Carries all fields through $group
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $group: { _id: "$customerId", total: { $sum: "$amount" } } }
])

// ✅ Drop unused fields BEFORE blocking stage
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $project: { customerId: 1, amount: 1 } },
  { $group: { _id: "$customerId", total: { $sum: "$amount" } } }
])

4.3.2 `$limit` Before Blocking Stages

If you only need top-K results, limit before grouping when possible:

// For "top 10 customers by order count"
// ❌ Groups all, then sorts all, then limits
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $group: { _id: "$customerId", count: { $sum: 1 } } },
  { $sort: { count: -1 } },
  { $limit: 10 }
])

// ✅ If you have an index on { status: 1, customerId: 1 }
// Use a two-phase approach (application-side)
// Phase 1: Get candidate customerIds from a sampled or indexed query
// Phase 2: Aggregate only those customers

4.3.3 Approximate Aggregations with `$sample`

For exploratory analysis, $sample avoids full scans:

// Get approximate distribution of order amounts
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $sample: { size: 10000 } },
  { $bucket: {
      groupBy: "$amount",
      boundaries: [0, 50, 100, 250, 500, 1000, 10000],
      default: "10000+",
      output: { count: { $sum: 1 }, avgAmount: { $avg: "$amount" } }
    }
  }
])

4.4 `$setWindowFields` Memory Profile

$setWindowFields is a blocking stage that partitions data. Each partition must fit in memory:

// ❌ Dangerous: single partition = all documents in memory
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $setWindowFields: {
      partitionBy: null, // Single partition
      sortBy: { createdAt: 1 },
      output: { runningTotal: { $sum: "$amount", window: { documents: ["unbounded", "current"] } } }
    }
  }
])

// ✅ Partitioned: each customer's orders processed independently
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $setWindowFields: {
      partitionBy: "$customerId",
      sortBy: { createdAt: 1 },
      output: { runningTotal: { $sum: "$amount", window: { documents: ["unbounded", "current"] } } }
    }
  }
])

5. Sharded Cluster Considerations

On sharded clusters, aggregation behavior changes dramatically. The query router (mongos) splits the pipeline into shard-local and merge portions.

5.1 Pipeline Splitting Rules

Stages that can run on shards (shard-local):

$match, $project, $addFields, $set, $unset, $unwind, $lookup (uncorrelated, sharded collection), $group, $sort, $limit, $skip, $sample, $bucket, $facet

Stages that must run on mongos (merge stages):

$group (final), $sort (final), $limit (final), $skip (final), $out, $merge, $lookup (uncorrelated, unsharded collection)

5.2 The Shard Key Dictates Everything

If your $match includes the shard key, MongoDB targets specific shards (targeted operation). Otherwise, it broadcasts to all shards.

// Shard key: { customerId: 1 }

// ✅ Targeted: single shard
db.orders.aggregate([
  { $match: { customerId: "cust_123", status: "completed" } },
  { $group: { _id: "$productId", total: { $sum: "$amount" } } }
])

// ❌ Broadcast: all shards
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $group: { _id: "$customerId", total: { $sum: "$amount" } } }
])

5.3 Two-Phase `$group` Optimization

For broadcast operations, MongoDB automatically performs a two-phase group: each shard does a partial $group, mongos merges results. This works only if the accumulator is associative and commutative ($sum, $avg, $min, $max, $push with $first/$last on sorted input).

// ✅ Two-phase group works: $sum is associative
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $group: { _id: "$customerId", total: { $sum: "$amount" } } }
])

// ❌ Two-phase group FAILS: $stdDevPop not associative
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $group: { _id: "$customerId", stdDev: { $stdDevPop: "$amount" } } }
])
// Falls back: all documents sent to mongos for final group

5.4 `$sort` + `$limit` on Sharded Clusters

The optimization only works if the sort key includes the shard key prefix:

// Shard key: { customerId: 1, createdAt: 1 }

// ✅ Optimized: each shard returns top-K, mongos merges
db.orders.aggregate([
  { $match: { customerId: "cust_123" } },
  { $sort: { createdAt: -1 } },
  { $limit: 10 }
])

// ❌ Not optimized: mongos must merge all then sort
db.orders.aggregate([
  { $match: { customerId: "cust_123" } },
  { $sort: { amount: -1 } },
  { $limit: 10 }
])

5.5 `$merge` and `$out` for Materialized Views

For frequently-run heavy aggregations, pre-compute results:

// Run nightly via scheduler
db.orders.aggregate([
  { $match: { status: "completed", createdAt: { $gte: ISODate("2024-01-01") } } },
  { $group: { _id: "$customerId", total: { $sum: "$amount" }, orderCount: { $sum: 1 } } },
  { $merge: {
      into: "customer_totals_2024",
      on: "_id",
      whenMatched: "replace",
      whenNotMatched: "insert"
    }
  }
])

// Query the materialized view — milliseconds
db.customer_totals_2024.find().sort({ total: -1 }).limit(10)

6. Monitoring, Profiling, and Explain Plans

6.1 Profiling Slow Aggregations

Enable profiling for aggregations exceeding 100ms:

db.setProfilingLevel(1, { slowms: 100, filter: { "command.aggregate": { $exists: true } } })

// Query profile
db.system.profile.find({ "command.aggregate": "orders" }).sort({ ts: -1 }).limit(5)

6.2 Key Metrics to Alert On

Metric	Threshold	Action
`executionTimeMillis`	> 5000ms	Investigate pipeline
`usedDisk: true`	Any	Add index or restructure
`totalDocsExamined / nReturned`	> 1000	Poor selectivity, add index
`shards.total` vs `shards.targeted`	Broadcast > 50%	Review shard key
`memoryUsageBytes`	> 50MB	Reduce working set

6.3 `explain("allPlansExecution")` for Plan Comparison

db.orders.aggregate([
  { $match: { status: "completed" } },
  { $group: { _id: "$customerId", total: { $sum: "$amount" } } }
]).explain("allPlansExecution")

Shows all considered plans — useful for verifying index selection.

6.4 Atlas Performance Advisor & Query Profiler

If on MongoDB Atlas, use:

Performance Advisor: Suggests indexes from slow query patterns
Query Profiler: Visual explain plans with stage timing
Real-Time Performance Panel: Live aggregation monitoring

7. Real-World Patterns and Anti-Patterns

7.1 Pattern: Time-Series Rollups with `$dateTrunc` (MongoDB 5.0+)

db.events.aggregate([
  { $match: { timestamp: { $gte: ISODate("2024-01-01") } } },
  { $group: {
      _id: {
        hour: { $dateTrunc: { date: "$timestamp", unit: "hour" } },
        eventType: "$type"
      },
      count: { $sum: 1 },
      avgDuration: { $avg: "$duration" },
      p95Duration: { $percentile: { input: "$duration", p: [0.95], method: "approximate" } }
    }
  },
  { $sort: { "_id.hour": 1 } }
])

Index: { timestamp: 1, type: 1 } — enables targeted shard access if sharded by timestamp.

7.2 Pattern: Hierarchical Aggregation with `$facet`

db.orders.aggregate([
  { $match: { status: "completed", createdAt: { $gte: ISODate("2024-01-01") } } },
  { $facet: {
      summary: [
        { $group: { _id: null, totalRevenue: { $sum: "$amount" }, totalOrders: { $sum: 1 } } }
      ],
      byRegion: [
        { $group: { _id: "$region", revenue: { $sum: "$amount" }, orders: { $sum: 1 } } },
        { $sort: { revenue: -1 } }
      ],
      byChannel: [
        { $group: { _id: "$channel", revenue: { $sum: "$amount" }, orders: { $sum: 1 } } },
        { $sort: { revenue: -1 } }
      ],
      topCustomers: [
        { $group: { _id: "$customerId", revenue: { $sum: "$amount" } } },
        { $sort: { revenue: -1 } },
        { $limit: 20 }
      ]
    }
  }
])

Single scan, four analytical views.

7.3 Anti-Pattern: `$unwind` Explosion

// ❌ 1M orders × 10 items = 10M documents in memory
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $unwind: "$items" },
  { $group: { _id: "$items.productId", total: { $sum: "$items.amount" } } }
])

// ✅ Use $reduce or $map to compute per-document, then group
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $project: {
      productTotals: {
        $map: {
          input: "$items",
          as: "item",
          in: { productId: "$$item.productId", amount: "$$item.amount" }
        }
      }
    }
  },
  { $unwind: "$productTotals" },
  { $group: { _id: "$productTotals.productId", total: { $sum: "$productTotals.amount" } } }
])

// ✅ Better: $setWindowFields with partitionBy (MongoDB 5.0+)
// Avoids $unwind entirely for some patterns

7.4 Anti-Pattern: `$lookup` Inside `$facet`

// ❌ Each facet runs its own $lookup — multiplies foreign collection reads
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $facet: {
      withCustomer: [
        { $lookup: { from: "customers", localField: "customerId", foreignField: "_id", as: "cust" } },
        { $unwind: "$cust" }
      ],
      withProduct: [
        { $unwind: "$items" },
        { $lookup: { from: "products", localField: "items.productId", foreignField: "_id", as: "prod" } },
        { $unwind: "$prod" }
      ]
    }
  }
])

// ✅ Single $lookup before $facet, then reference
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $lookup: { from: "customers", localField: "customerId", foreignField: "_id", as: "customer" } },
  { $unwind: "$customer" },
  { $facet: { /* facets now reference $customer directly */ } }
])

7.5 Pattern: Incremental Aggregation with Change Streams

For real-time dashboards, avoid re-aggregating entire collections:

// Application-side: maintain materialized view
const changeStream = db.orders.watch([
  { $match: { "fullDocument.status": "completed" } }
])

for await (const change of changeStream) {
  if (change.operationType === "insert") {
    await db.customer_totals.updateOne(
      { _id: change.fullDocument.customerId },
      { $inc: { totalRevenue: change.fullDocument.amount, orderCount: 1 } },
      { upsert: true }
    )
  }
  // Handle update/delete similarly
}

This shifts compute from query-time to write-time — essential for sub-second dashboards on large datasets.

8. Frequently Asked Questions

Q: When should I use `allowDiskUse: true` vs fixing the pipeline?

Use allowDiskUse only as a temporary mitigation while you implement the real fix (index, pipeline restructure, or materialized view). Disk spill adds 10-100x latency. If a query requires disk spill to complete, it's not production-ready for interactive workloads.

Q: How do I optimize `$group` with high-cardinality `_id` (e.g., grouping by sessionId)?

High-cardinality groups create massive working sets. Options:

Pre-filter aggressively with $match using an index
Use $bucket with coarse boundaries instead of exact grouping
Approximate with $sample + extrapolation
Materialize incrementally via change streams
Consider a time-series collection (MongoDB 5.0+) with automatic bucketing

Q: Does `$merge` into a sharded collection work efficiently?

Yes, but the on field should be the shard key (or prefix) for targeted writes. If merging by a non-shard-key field, each upsert becomes a broadcast operation. Design the target collection's shard key to match your merge key.

Optimization is iterative: measure with explain, identify the bottleneck, apply the highest-leverage fix, repeat. The patterns above cover 90% of production aggregation performance issues. For the remaining 10%, the profiler and a reproducible test case are your best tools.

Advanced Patterns: The Remaining 10%

Pipeline Fusion for Microservice Architectures

When aggregations span service boundaries, network latency dominates. Fuse what you can:

// ❌ Three round-trips
const users = await db.users.aggregate([{ $match: { active: true } }]).toArray();
const orders = await db.orders.aggregate([{ $match: { userId: { $in: users.map(u => u._id) } } }]).toArray();
const payments = await db.payments.aggregate([{ $match: { orderId: { $in: orders.map(o => o._id) } } }]).toArray();

// ✅ Single pipeline with $lookup (if co-located)
const pipeline = [
  { $match: { active: true } },
  {
    $lookup: {
      from: 'orders',
      let: { userId: '$_id' },
      pipeline: [
        { $match: { $expr: { $eq: ['$userId', '$$userId'] } } },
        {
          $lookup: {
            from: 'payments',
            let: { orderId: '$_id' },
            pipeline: [{ $match: { $expr: { $eq: ['$orderId', '$$orderId'] } } }],
            as: 'payments'
          }
        }
      ],
      as: 'orders'
    }
  }
];

Rule: $lookup with uncorrelated sub-pipelines (MongoDB 3.6+) pushes work to the shard. Correlated sub-pipelines ($expr with $$vars) run on the mongos — avoid for high-cardinality joins.

Time-Series Collections: Built-In Optimization

MongoDB 5.0+ time-series collections auto-optimize for append-heavy, time-ordered workloads:

db.createCollection('metrics', {
  timeseries: {
    timeField: 'timestamp',
    metaField: 'sensorId',
    granularity: 'hours'
  },
  expireAfterSeconds: 2592000 // 30 days TTL
});

// Aggregation automatically uses columnar storage, compression, and
// predicate pushdown on timeField. No manual bucketing needed.
db.metrics.aggregate([
  { $match: { timestamp: { $gte: ISODate('2024-01-01'), $lt: ISODate('2024-02-01') } } },
  { $group: { _id: '$sensorId', avgTemp: { $avg: '$temperature' } } }
]);

Ceiling: Not suitable for random updates/deletes on historical data. ponytail: immutable append-only pattern; use regular collections if backfill/update frequency > 1%.

Atlas Search `$search` vs Aggregation `$text`

Scenario	Use
Full-text relevance, typo tolerance, faceted search	`$search` (Atlas only)
Simple prefix/exact match, self-hosted	`$text` index + `$match`
Regex on indexed field	`$match` with anchored regex (`/^prefix/`)

// Atlas Search: runs on dedicated search nodes, doesn't block primary
db.products.aggregate([
  {
    $search: {
      index: 'product_search',
      compound: {
        must: [{ text: { query: 'wireless headphones', path: 'name', fuzzy: { maxEdits: 1 } } }],
        filter: [{ range: { path: 'price', gte: 50, lte: 300 } }]
      }
    }
  },
  { $limit: 20 },
  { $project: { name: 1, price: 1, score: { $meta: 'searchScore' } } }
]);

Production Guardrails

Automated `explain` in CI

#!/usr/bin/env bash
# ci/check-aggregation-plans.sh
# Fails build if any aggregation lacks index coverage or exceeds 100ms estimated

THRESHOLD_MS=100
FAILED=0

for file in src/aggregations/*.js; do
  PIPELINE=$(cat "$file")
  EXPLAIN=$(mongosh --quiet --eval "db.getSiblingDB('prod').collection.explain('executionStats')($PIPELINE)")

  # Check for COLLSCAN
  if echo "$EXPLAIN" | jq -e '.executionStats.stages[] | select(.stage == "COLLSCAN")' > /dev/null; then
    echo "❌ $file: COLLSCAN detected"
    FAILED=1
  fi

  # Check estimated execution time
  EXEC_TIME=$(echo "$EXPLAIN" | jq '.executionStats.executionTimeMillis')
  if (( EXEC_TIME > THRESHOLD_MS )); then
    echo "⚠️  $file: ${EXEC_TIME}ms estimated (threshold: ${THRESHOLD_MS}ms)"
  fi
done

exit $FAILED

Slow Aggregation Alerting (Atlas/Cloud Manager)

{
  "alertConfig": {
    "eventTypeName": "QUERY_EXECUTOR_SCANNED_OBJECTS_RATIO",
    "matchers": [
      { "fieldName": "namespace", "operator": "CONTAINS", "value": "analytics" }
    ],
    "threshold": 1000,
    "notifications": [{ "typeName": "PAGER_DUTY", "intervalMin": 5 }]
  }
}

Key metrics to alert on:

scannedObjects / returnedObjects > 1000 — missing index or bad $match placement
executionTimeMillis > 5000 — pipeline too heavy for interactive use
memoryUsageBytes > 100MB — needs allowDiskUse or redesign

Testing Strategy: Reproducible Performance Tests

Deterministic Dataset Generator

// test/fixtures/generate.js
const { MongoClient } = require('mongodb');
const { faker } = require('@faker-js/faker');

async function seed(db, { users = 10000, ordersPerUser = 5, products = 500 } = {}) {
  // Deterministic seed for reproducible benchmarks
  faker.seed(12345);

  const productIds = Array.from({ length: products }, (_, i) => ({
    _id: `prod_${i}`,
    name: faker.commerce.productName(),
    category: faker.helpers.arrayElement(['electronics', 'clothing', 'home', 'sports']),
    price: Number(faker.commerce.price({ min: 10, max: 1000 })),
    tags: faker.helpers.arrayElements(['sale', 'new', 'popular', 'clearance'], 2)
  }));

  const users = Array.from({ length: users }, (_, i) => ({
    _id: `user_${i}`,
    email: faker.internet.email(),
    region: faker.helpers.arrayElement(['US-EAST', 'US-WEST', 'EU', 'APAC']),
    tier: faker.helpers.arrayElement(['free', 'pro', 'enterprise']),
    createdAt: faker.date.between({ from: '2020-01-01', to: '2024-01-01' })
  }));

  const orders = users.flatMap(u => 
    Array.from({ length: ordersPerUser }, () => ({
      _id: `order_${u._id}_${Math.random().toString(36).slice(2)}`,
      userId: u._id,
      productId: faker.helpers.arrayElement(productIds)._id,
      quantity: faker.number.int({ min: 1, max: 5 }),
      status: faker.helpers.arrayElement(['pending', 'shipped', 'delivered', 'returned']),
      createdAt: faker.date.between({ from: u.createdAt, to: '2024-12-31' })
    }))
  );

  await Promise.all([
    db.collection('products').insertMany(productIds),
    db.collection('users').insertMany(users),
    db.collection('orders').insertMany(orders)
  ]);

  // Indexes matching production
  await db.collection('orders').createIndexes([
    { key: { userId: 1, createdAt: -1 } },
    { key: { productId: 1, status: 1 } },
    { key: { status: 1, createdAt: -1 } }
  ]);
}

Benchmark Harness

// test/perf/aggregation.bench.js
const { performance } = require('perf_hooks');

async function benchmark(db, name, pipeline, { runs = 10, warmup = 3 } = {}) {
  // Warmup
  for (let i = 0; i < warmup; i++) {
    await db.collection('orders').aggregate(pipeline).toArray();
  }

  const times = [];
  for (let i = 0; i < runs; i++) {
    const start = performance.now();
    await db.collection('orders').aggregate(pipeline).toArray();
    times.push(performance.now() - start);
  }

  times.sort((a, b) => a - b);
  const p50 = times[Math.floor(runs * 0.5)];
  const p95 = times[Math.floor(runs * 0.95)];
  const p99 = times[Math.floor(runs * 0.99)];

  console.log(`${name}: p50=${p50.toFixed(2)}ms p95=${p95.toFixed(2)}ms p99=${p99.toFixed(2)}ms`);

  return { p50, p95, p99 };
}

// Usage
await benchmark(db, 'revenue-by-region', [
  { $match: { createdAt: { $gte: new Date('2024-01-01') } } },
  { $lookup: { from: 'users', localField: 'userId', foreignField: '_id', as: 'user' } },
  { $unwind: '$user' },
  { $group: { _id: '$user.region', revenue: { $sum: { $multiply: ['$quantity', '$price'] } } } }
]);

Migration Checklist: 4.x → 6.x / 7.x

Change	Action Required
`$group` accumulator changes	Verify `$first`/`$last` behavior with sort (now requires explicit `$sort` before)
`$dateToString` format tokens	`%L` (milliseconds) → `%3d`; `%f` (microseconds) → `%6d`
`allowDiskUse` default	Now `true` by default; remove explicit setting
Sharded `$sort` + `$limit`	Auto-optimized; remove manual `$sort` on shard key prefix
`$lookup` uncorrelated sub-pipeline	Preferred over correlated; refactor where possible
Time-series collections	Migrate append-heavy workloads; drop manual bucketing code

When to Stop Optimizing

Diminishing returns checklist — if all are true, ship it:

[ ] explain shows IXSCAN on all $match/$lookup paths
[ ] No stage exceeds 50% of total execution time
[ ] Memory < 50% of allowDiskUse threshold (or 50MB if in-memory)
[ ] P99 latency < SLA (typically 200ms for API, 5s for batch)
[ ] No COLLSCAN, SORT without index, or unbounded $unwind
[ ] CI regression test passes with 2× production data volume

Closing Thought

The fastest aggregation is the one you don't run. Cache computed results, pre-aggregate in change streams, push work to the write path. But when you must aggregate: index first, reduce early, measure always.

// The only comment that matters in production
// explain() or it didn't happen

Appendix: MongoDB Aggregation Pipeline Quick Reference • Explain Output Reference • Performance Best Practices

Table of Contents