Optimizing Large-Scale MongoDB Aggregation Pipelines: From Execution Plans to Production Tuning

// Streaming operators process documents one at a time (or in bounded batches) // - $match: filters documents as they arrive // - $project: transforms each document independently // - $addFields: enriches each document // - $sortByCount: wraps $group + $sort (blocking) // Blocking operators must consume ALL input before producing output // - $group: accumulates all matching documents into groups // - $sort: must buffer enough documents to produce sorted output // - $limit: buffers until enough documents are consumed // - $skip: discards documents until offset is reached 

// $sum: O(1) memory per group — accumulates into a single value // $push: O(n) memory per group — stores all elements in an array // $addToSet: O(k) memory per group where k = distinct values // $first/$last: O(1) memory per group // $avg: O(1) memory per group — maintains count + sum // $min/$max: O(1) memory per group // $stdDevPop/$stdDevSamp: O(1) memory per group — uses Welford's algorithm // DANGEROUS AT SCALE: { $group: { _id: "$customerId", // If cardinality = 50M, this materializes 50M groups orders: { $push: "$orderData" }, // Each group holds ALL its orders in memory total: { $sum: "$amount" } } } 

// Pipeline: // db.orders.aggregate([ // { $match: { status: "completed", region: "us-east", createdAt: { $gte: new Date("2024-01-01") } } }, // { $sort: { amount: -1 } }, // { $limit: 100 }, // { $group: { _id: "$customerId", total: { $sum: "$amount" } } } // ]) // OPTIMAL INDEX (supports $match + $sort + projection): db.orders.createIndex( { status: 1, region: 1, createdAt: -1, amount: -1 }, { name: "agg_status_region_date_amount" } ) // This index allows: // 1. Index range scan for status + region (equality) // 2. Sorted traversal by createdAt (range) // 3. Sorted traversal by amount within the range // 4. Covering projection if all needed fields are in the index 

// If the pipeline only needs fields present in the index: db.orders.aggregate([ { $match: { status: "completed", region: "us-east" } }, { $group: { _id: "$customerId", total: { $sum: "$amount" } } }, { $project: { _id: 1, total: 1 } } ]).explain("executionStats") // Required covering index: db.orders.createIndex( { status: 1, region: 1, customerId: 1, amount: 1 }, { name: "agg_covering_status_region_cust_amount" } ) // Verify with explain — look for "IXSCAN" with no "FETCH" stage 

// Document structure: // { orderId: 123, tags: ["electronics", "premium", "sale"], amount: 99.99 } // Multikey index on array field: db.orders.createIndex({ tags: 1, amount: 1 }) // WARNING: Multikey indexes prevent index-only scans for $group. // The engine cannot determine which index entry maps to which document // without fetching the full document. This forces FETCH stage. // Better approach: pre-compute a non-array grouping field // or use $unwind + $match before $group 

// Only index documents relevant to the aggregation db.orders.createIndex( { customerId: 1, amount: 1 }, { name: "agg_active_orders", partialFilterExpression: { status: { $in: ["completed", "refunded"] }, amount: { $gt: 0 } } } ) // Benefit: Smaller index = better B-tree depth = fewer disk pages in cache // A 500MB index vs a 5GB index can be the difference between // entirely in-memory scan vs repeated disk I/O 

// ANTI-PATTERN: Group before filtering // Materializes ALL groups, then discards most [ { $group: { _id: "$customerId", total: { $sum: "$amount" } } }, { $match: { total: { $gte: 1000 } } }, // Too late — all groups already materialized { $sort: { total: -1 } }, { $limit: 1000 } ] // OPTIMIZED: Push filtering before grouping where possible [ { $match: { amount: { $gte: 1 } } }, // Pre-filter cheaply (uses index) { $group: { _id: "$customerId", total: { $sum: "$amount" } } }, { $match: { total: { $gte: 1000 } } }, // Post-group filter (unavoidable) { $sort: { total: -1 } }, { $limit: 1000 } ] 

// Pipeline: Get top 100 customers by order count in a specific category // WITHOUT $sort+$limit: sorts ALL matching documents into memory [ { $match: { category: "electronics" } }, { $group: { _id: "$customerId", count: { $sum: 1 } } }, { $sort: { count: -1 } }, // Sorts ALL groups — potentially millions { $limit: 100 } ] // The engine DOES optimize $sort + $limit into a top-N heap, // but you can reduce input size further with pre-filtering: [ { $match: { category: "electronics", orderDate: { $gte: last30Days } } }, { $group: { _id: "$customerId", count: { $sum: 1 } } }, { $sort: { count: -1 } }, { $limit: 100 } ] 

// $facet materializes ALL input documents before processing // Only use when the input to $facet is already heavily filtered [ { $match: { region: "us-east", status: "completed" } }, // Heavy pre-filter { $facet: { revenueByMonth: [ { $group: { _id: { $month: "$createdAt" }, revenue: { $sum: "$amount" } } }, { $sort: { _id: 1 } } ], topProducts: [ { $group: { _id: "$productId", count: { $sum: 1 } } }, { $sort: { count: -1 } }, { $limit: 10 } ], summary: [ { $group: { _id: null, totalRevenue: { $sum: "$amount" }, orderCount: { $sum: 1 } } } ] }} ] // Memory cost: ALL documents matching $match are held in memory simultaneously // for ALL sub-pipelines. For 1M documents × 500 bytes = 500MB just for input buffering. 

// CRITICAL: allowDiskUse is NOT always safe // It trades memory for disk I/O — potentially making things 10-100x slower // Safe with allowDiskUse: predictable, bounded output // - $group with known cardinality // - $sort with known input size // DANGEROUS with allowDiskUse: unpredictable output size // - $push with unknown group sizes // - $facet with multiple sub-pipelines on large input // - $sort on high-cardinality data with no $limit // Production pattern: set allowDiskUse with monitoring [ { $match: { status: "completed" } }, { $group: { _id: "$region", revenue: { $sum: "$amount" } } }, { $sort: { revenue: -1 } } ], { allowDiskUse: true } 

// Basic $lookup (hash join on foreign key) db.orders.aggregate([ { $lookup: { from: "customers", localField: "customerId", foreignField: "_id", as: "customer" } } ]) // The engine builds a hash index on the foreign collection's foreignField. // For 50M customer documents, this hash index consumes ~2-4GB of RAM. // This is the #1 cause of OOM in aggregation pipelines. 

// PATTERN 1: Filter before lookup (reduce hash table size) [ { $match: { status: "completed" } }, // Reduce input documents { $lookup: { from: "customers", localField: "customerId", foreignField: "_id", as: "customer", // pipeline option allows filtering the foreign collection BEFORE hash build pipeline: [ { $match: { active: true, region: "us-east" } }, { $project: { name: 1, tier: 1 } } // Project only needed fields ] } } ] // PATTERN 2: Replace $lookup with pre-joined embedded data // If the lookup field is rarely updated, embed it at write time: // { orderId: 123, customerId: "c1", customerName: "Alice", customerTier: "gold" } // Eliminates the join entirely — read-only, zero join cost // PATTERN 3: Use $expr with $in for single-document lookups // When you only need to verify existence: { $match: { $expr: { $eq: ["$customerId", "$customerLookup"] // Pre-resolved } } } 

// Memory consumed by $lookup hash table: // Formula: numDocuments × (keySize + valueReferenceSize + overhead) // For 50M customers with ObjectId keys: // = 50M × (12 bytes ObjectId + 8 bytes pointer + ~32 bytes overhead) // = 50M × 52 bytes ≈ 2.6 GB // Mitigation: Use the pipeline option to reduce foreign collection size // BEFORE hash table construction: db.orders.aggregate([ { $lookup: { from: "customers", localField: "customerId", foreignField: "_id", as: "customer", pipeline: [ // This $match runs on the foreign collection BEFORE hash build { $match: { _id: { $in: topCustomerIds } } }, // If possible { $project: { name: 1, email: 1 } } // Slim the documents ] } } ]) // Even better: If you know the set of customerIds in advance, // pre-fetch customer data and use $addFields with $reduce/$arrayElemAt 

┌─────────────────────────────────────────────────────────────┐ │ mongos │ │ ┌───────────────────────────────────────────────────────┐ │ │ │ Global Planning Phase │ │ │ │ 1. Send pipeline to all shards for planning │ │ │ │ 2. Collect execution plans from each shard │ │ │ │ 3. Select merge strategy based on operator type │ │ │ └───────────────────────────────────────────────────────┘ │ │ ┌───────────────────────────────────────────────────────┐ │ │ │ Fan-Out Execution Phase │ │ │ │ Send pipeline to Shard 1 → returns partial results │ │ │ │ Send pipeline to Shard 2 → returns partial results │ │ │ │ Send pipeline to Shard N → returns partial results │ │ │ └───────────────────────────────────────────────────────┘ │ │ ┌───────────────────────────────────────────────────────┐ │ │ │ Merge Phase (at mongos) │ │ │ │ Merge all partial results into final output │ │ │ └───────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────┘ 

// Sharded collection: orders // Shard key: { customerId: "hashed" } // Pipeline 1: SINGLE SHARD (no fan-out) // $match on shard key routes to exactly one shard db.orders.aggregate([ { $match: { customerId: "customer_12345" } }, // Routes to 1 shard { $group: { _id: "$category", total: { $sum: "$amount" } } } ]) // Pipeline 2: FULL FAN-OUT (all shards) // No $match on shard key before first blocking operator db.orders.aggregate([ { $group: { _id: "$category", total: { $sum: "$amount" } } }, // ALL shards { $sort: { total: -1 } }, { $limit: 10 } ]) // Pipeline 3: PARTIAL FAN-OUT // $match on non-shard-key field fans out but reduces per-shard work db.orders.aggregate([ { $match: { status: "completed" } }, // All shards, but less data per shard { $group: { _id: "$region", total: { $sum: "$amount" } } }, { $sort: { total: -1 } } ]) 

// Two-phase group (default when group _id is not the shard key): db.orders.aggregate([ { $group: { _id: "$category", total: { $sum: "$amount" } } } // Phase 1: Each shard groups by category → returns partial results // Phase 2: mongos re-groups all partial results // If 10 shards × 50 categories = 500 intermediate docs to merge ]) // Single-phase group (when group _id IS the shard key): db.orders.aggregate([ { $match: { customerId: { $in: allCustomerIds } } }, // Routes to specific shards { $group: { _id: "$customerId", total: { $sum: "$amount" } } } // Each shard handles its own customerId groups — no merge needed ]) // Alternative: Pre-aggregate in application layer // Issue N separate queries (one per shard) and merge results in app code // Gives you explicit control over merge logic and error handling 

// For time-series data, zoned sharding keeps date ranges on single shards db.adminCommand({ shardCollection: "analytics.events", uniqueIndex: { key: { timestamp: 1, service: 1 }, partialFilterExpression: { timestamp: { $type: "date" } }, shardKeyPattern: { timestamp: { $gte: new Date("2024-01-01"), $lt: new Date("2024-02-01") } } } }) // Now: monthly aggregations hit a SINGLE shard [ { $match: { timestamp: { $gte: new Date("2024-01-01"), $lt: new DateSource：https://dev.to/tamizuddin/optimizing-large-scale-mongodb-aggregation-pipelines-from-execution-plans-to-production-tuning-16bm
My favorites