MongoDB Aggregation skill for Codex

Done well means: the pipeline returns the right numbers on edge-case data (missing fields, empty groups, duplicates after $unwind), its first stages use an index, each stage passes on only the fields later stages need, and its runtime is known from explain, not guessed.

Workflow

1. Read AGENTS.md; find the driver, server version (db.version()) and whether this is Atlas (enables $search and $vectorSearch, which must be the first stage).
2. State the result shape first: one example output document.
3. Write the pipeline in mongosh against sample data, stage by stage, checking the output after each stage (pipeline.slice(0, n)).
4. Order stages for index use and data reduction (rules below).
5. Run explain("executionStats"), fix the plan, then port the pipeline into code.
6. Add a test with fixture documents covering nulls, empty arrays and ties.

Stage order rules

• $match first, on indexed fields, then $sort (+$limit) while the index still applies. Once $group, $unwind, $project with computed fields or $lookup runs, later $match/$sort stages cannot use collection indexes.
• $sort immediately followed by $limit becomes a top-k sort with bounded memory.
• Drop unused fields early with $project or $unset; large documents through $group or $lookup dominate memory.
• $match on fields produced by $lookup comes after the $lookup, but filter the joined collection inside the $lookup pipeline so fewer documents are joined.
• $unwind multiplies documents; $group back by _id afterwards, or avoid unwinding by using array operators ($filter, $map, $reduce, $size).
• $count or $group: { _id: null, n: { $sum: 1 } } for totals; do not fetch all documents to count in application code.

Core example: revenue per product, last 30 days

db.orders.aggregate([
  { $match: { status: "paid", paidAt: { $gte: ISODate("2026-09-28") } } },   // index { status: 1, paidAt: 1 }
  { $project: { items: 1, paidAt: 1 } },
  { $unwind: "$items" },
  { $group: {
      _id: "$items.productId",
      revenueCents: { $sum: { $multiply: ["$items.qty", "$items.unitCents"] } },
      orders: { $addToSet: "$_id" },
      lastSale: { $max: "$paidAt" }
  }},
  { $project: { revenueCents: 1, lastSale: 1, orderCount: { $size: "$orders" } } },
  { $sort: { revenueCents: -1 } },
  { $limit: 20 },
  { $lookup: {
      from: "products",
      localField: "_id",
      foreignField: "_id",
      pipeline: [{ $project: { name: 1, sku: 1 } }],
      as: "product"
  }},
  { $set: { product: { $first: "$product" } } }
]);

Note the $lookup after $limit: join 20 documents, not 50 000.

$lookup decisions

Need	Form
Equality join	localField / foreignField (index foreignField on the joined collection)
Equality plus filter/projection on the joined side	localField/foreignField plus pipeline (MongoDB 5.0+)
Non-equality or multi-field conditions	let + pipeline with $match: { $expr: ... }
Called for every document on a hot read path	Reshape the schema (extended reference) instead

{ $lookup: {
    from: "payments",
    let: { orderId: "$_id", since: "$createdAt" },
    pipeline: [
      { $match: { $expr: { $and: [
        { $eq: ["$orderId", "$$orderId"] },
        { $gte: ["$at", "$$since"] }
      ] } } },
      { $sort: { at: -1 } },
      { $limit: 1 }
    ],
    as: "lastPayment"
}}

$expr equality matches can use an index on the joined collection; range comparisons inside $expr often cannot. Check the explain output for each $lookup.

Grouping, windows and buckets

• Top document per group: $group: { _id: "$customerId", latest: { $top: { sortBy: { createdAt: -1 }, output: "$$ROOT" } } } ($top/$topN/$bottomN in 5.2+). Otherwise $sort then $group with $first.
• Running totals, ranks, moving averages: $setWindowFields (5.0+).

{ $setWindowFields: {
    partitionBy: "$storeId",
    sortBy: { day: 1 },
    output: {
      runningRevenue: { $sum: "$revenue", window: { documents: ["unbounded", "current"] } },
      avg7d: { $avg: "$revenue", window: { range: [-6, 0], unit: "day" } }
    }
}}

• Histograms: $bucket with fixed boundaries and a default bucket, or $bucketAuto.
• Several aggregations over the same input (counts per facet on a search page): $facet. Each sub-pipeline output must fit in one 16 MB document; cap each with $limit.
• Date grouping: $dateTrunc: { date: "$at", unit: "day", timezone: "Europe/Berlin" }. Always pass the timezone the business reports in.
• Fill gaps in time series output: $densify then $fill.

Materialized results with $merge

db.orders.aggregate([
  { $match: { paidAt: { $gte: startOfDay, $lt: endOfDay } } },
  { $group: { _id: { day: startOfDay, productId: "$productId" }, revenue: { $sum: "$totalCents" } } },
  { $merge: { into: "daily_revenue", on: "_id", whenMatched: "replace", whenNotMatched: "insert" } }
]);

• Recompute only the changed window (a day, an hour) on a schedule; $merge upserts.
• $out replaces the whole target collection; use it only for full rebuilds, and confirm with the user before pointing it at an existing collection, since its contents are lost.
• Reports on large collections should read the merged collection, not aggregate raw data per request.

Reading explain

db.orders.explain("executionStats").aggregate(pipeline)

Check:

• The first stage(s) show IXSCAN (in $cursor / queryPlanner.winningPlan), not COLLSCAN.
• totalDocsExamined vs documents returned by the initial match: close means a selective index.
• For $lookup, collectionScans and indexesUsed per lookup stage (recent versions report them).
• $sort stages showing usedDisk: true or memory use near the limit.

Memory and limits

• Each blocking stage ($group, $sort, $bucketAuto, $setWindowFields) is limited to 100 MB of RAM. Since 6.0, allowDiskUseByDefault is on, so it spills to disk; spills are slow. Reduce input first; pass allowDiskUse: true explicitly on older servers.
• Output documents (including $facet and $group with $push) are capped at 16 MB.
• $addToSet/$push on large groups is the usual cause of both limits; aggregate counts or use $topN instead of collecting everything.
• Long-running pipelines on primaries compete with production traffic; use readPreference: "secondaryPreferred" for reports where slight staleness is acceptable.

Gotchas

• $unwind drops documents with missing or empty arrays unless preserveNullAndEmptyArrays: true.
• $sum of a missing field is 0; $avg ignores missing values. Decide which the report needs.
• $group key _id: "$field" groups missing and null together.
• Comparisons across BSON types follow type order (numbers < strings < objects); mixed types in a field produce surprising sorts and matches. Validate types at write time.
• ObjectId vs string ids: a $lookup between "abc..." and ObjectId("abc...") matches nothing.
• Driver code must pass dates as Date objects, not ISO strings, or $match compares strings.

For more stage recipes (pagination with totals, deduplication, pivoting, change-stream friendly pipelines, translating SQL clauses), read references/stage-cookbook.md when the pattern above does not cover the task.

Verification

• Fixture test: documents with missing fields, empty arrays, duplicates and ties; assert exact output.
• Compare totals with an independent computation on a sample (a simple find and sum in a script) before trusting a new report.
• explain("executionStats") before and after; record execution time and docs examined.

Output

**Pipeline:** <purpose>, collection <name>
**Result shape:** <example document>
**Indexes used / added:** <index> for stage 1
**Explain:** docs examined <n>, returned <n>, time <ms>, disk use <yes/no>
**Edge cases tested:** missing fields, empty arrays, ties, timezone boundaries