MongoDB Aggregation Pipeline: From Basic to Advanced

MongoDB Aggregation Pipeline

The aggregation pipeline processes documents through stages, transforming them at each step.

Basic Pipeline Stages

// db.orders.aggregate([...])

// $match: filter documents (uses indexes!)
{ $match: { status: 'completed', createdAt: { $gte: new Date('2026-01-01') } } }

// $group: aggregate by field
{ $group: {
  _id: '$userId',
  orderCount: { $sum: 1 },
  totalRevenue: { $sum: '$total' },
  avgOrder: { $avg: '$total' },
  firstOrder: { $min: '$createdAt' },
  lastOrder: { $max: '$createdAt' },
} }

// $sort: sort results
{ $sort: { totalRevenue: -1 } }

// $limit and $skip: pagination
{ $limit: 20 }

// $project: reshape documents
{ $project: {
  userId: '$_id',
  _id: 0,
  orderCount: 1,
  totalRevenue: { $round: ['$totalRevenue', 2] },
  avgOrder: { $round: ['$avgOrder', 2] },
} }

$lookup (Joins)

// Left join orders with users
db.orders.aggregate([
  { $match: { status: 'completed' } },
  
  // Simple join
  { $lookup: {
    from: 'users',
    localField: 'userId',
    foreignField: '_id',
    as: 'user',
  } },
  
  // Flatten the array (left join gives array)
  { $unwind: { path: '$user', preserveNullAndEmpty: true } },
  
  // Pipeline join (more powerful)
  { $lookup: {
    from: 'products',
    let: { itemIds: '$items.productId' },
    pipeline: [
      { $match: { $expr: { $in: ['$_id', '$itemIds'] } } },
      { $project: { name: 1, price: 1, category: 1 } },
    ],
    as: 'products',
  } },
])

$facet (Multiple Aggregations in One Query)

// Run multiple sub-pipelines on same input
db.products.aggregate([
  { $match: { active: true } },
  
  { $facet: {
    // Total count and stats
    metadata: [
      { $count: 'total' },
    ],
    
    // Price distribution
    priceRanges: [
      { $bucket: {
        groupBy: '$price',
        boundaries: [0, 25, 50, 100, 200, 500],
        default: '500+',
        output: { count: { $sum: 1 }, products: { $push: '$name' } }
      } },
    ],
    
    // Top categories
    categories: [
      { $group: { _id: '$category', count: { $sum: 1 } } },
      { $sort: { count: -1 } },
      { $limit: 5 },
    ],
  } },
])

Time Series Aggregations

// Daily revenue for the last 30 days
db.orders.aggregate([
  { $match: {
    createdAt: { $gte: new Date(Date.now() - 30 * 24 * 3600 * 1000) },
    status: 'completed',
  } },
  
  { $group: {
    _id: {
      year: { $year: '$createdAt' },
      month: { $month: '$createdAt' },
      day: { $dayOfMonth: '$createdAt' },
    },
    revenue: { $sum: '$total' },
    count: { $sum: 1 },
  } },
  
  { $sort: { '_id.year': 1, '_id.month': 1, '_id.day': 1 } },
  
  { $project: {
    date: {
      $dateFromParts: {
        year: '$_id.year',
        month: '$_id.month',
        day: '$_id.day',
      }
    },
    revenue: { $round: ['$revenue', 2] },
    count: 1,
    _id: 0,
  } },
])

Window Functions ($setWindowFields)

// Cumulative revenue with running total
db.orders.aggregate([
  { $match: { status: 'completed' } },
  
  { $setWindowFields: {
    partitionBy: '$userId',
    sortBy: { createdAt: 1 },
    output: {
      runningTotal: {
        $sum: '$total',
        window: { documents: ['unbounded', 'current'] },
      },
      previousOrderTotal: {
        $shift: { output: '$total', by: -1, default: 0 },
      },
      rankInUser: { $rank: {} },
    },
  } },
])

Performance: Use Indexes in Pipelines

// The $match stage MUST be first to use indexes
// Bad:
db.orders.aggregate([
  { $group: { _id: '$userId', total: { $sum: '$amount' } } },
  { $match: { total: { $gt: 1000 } } },  // Too late for index!
])

// Good:
db.orders.aggregate([
  { $match: { createdAt: { $gte: new Date('2026-01-01') } } },  // Uses index
  { $group: { _id: '$userId', total: { $sum: '$amount' } } },
])

// Use explain() to verify
db.orders.explain('executionStats').aggregate([...])

Atlas Search (Full-Text)

// Requires Atlas Search index
db.products.aggregate([
  { $search: {
    index: 'products_search',
    compound: {
      must: [
        { text: { query: 'wireless headphones', path: ['name', 'description'] } },
      ],
      should: [
        { range: { path: 'rating', gte: 4.0, boost: { value: 2 } } },
      ],
      filter: [
        { equals: { path: 'inStock', value: true } },
      ],
    },
  } },
  
  { $project: {
    name: 1,
    price: 1,
    score: { $meta: 'searchScore' },
  } },
  
  { $sort: { score: -1 } },
])