DocumentDB Aggregation Pipeline for MCP Servers — Analytics, $lookup Joins, and $facet

Pipeline execution model

DocumentDB processes aggregation stages as a sequential pipeline. Data flows through each stage in order; a later stage can only see fields that earlier stages passed through. The critical performance rule is that only $match (and to a lesser extent $sort) at the beginning of the pipeline can use a collection index. Any $match placed after a $group or $unwind forces a full scan of the intermediate result set in memory.

const db = (await getClient()).db(DB);
const coll = db.collection('events');

// ---- Correct: $match first to use index, then transform ----
const result = await coll.aggregate([
  { $match: { userId: userId, createdAt: { $gte: startDate, $lte: endDate } } },
  { $group: {
      _id: '$eventType',
      count:   { $sum: 1 },
      lastSeen: { $max: '$createdAt' },
  }},
  { $sort: { count: -1 } },
  { $limit: 20 },
], {
  allowDiskUse: true,
  cursor: { batchSize: 100 },
}).toArray();

// ---- Wrong: $match after $group cannot use index ----
const slow = await coll.aggregate([
  { $group: { _id: '$userId', total: { $sum: 1 } } },
  { $match: { total: { $gt: 10 } } },  // full scan of group result — no index
]).toArray();

Memory limit: without allowDiskUse: true, each pipeline stage is limited to 100 MB of RAM. A $group that produces a large intermediate document set, or a $sort over millions of documents, will throw MongoServerError: Exceeded memory limit for $group/$sort. Setting allowDiskUse: true causes DocumentDB to spill intermediate results to storage when the 100 MB limit is exceeded — slower but it completes.

$group — aggregation accumulator patterns

// Time-bucketed event counts — common analytics MCP tool pattern
const dailyCounts = await coll.aggregate([
  { $match: { createdAt: { $gte: new Date('2026-11-01'), $lt: new Date('2026-12-01') } } },
  { $group: {
      _id: {
        year:  { $year:  '$createdAt' },
        month: { $month: '$createdAt' },
        day:   { $dayOfMonth: '$createdAt' },
      },
      count:        { $sum: 1 },
      totalRevenue: { $sum: '$amount' },
      avgAmount:    { $avg: '$amount' },
      minAmount:    { $min: '$amount' },
      maxAmount:    { $max: '$amount' },
      uniqueUsers:  { $addToSet: '$userId' },  // array of distinct userIds — large!
  }},
  { $project: {
      _id: 1,
      count: 1,
      totalRevenue: 1,
      avgAmount: { $round: ['$avgAmount', 2] },
      uniqueUserCount: { $size: '$uniqueUsers' },  // $size on the addToSet array
      // $uniqueUsers itself is omitted — we only need the count
  }},
  { $sort: { '_id.year': 1, '_id.month': 1, '_id.day': 1 } },
], { allowDiskUse: true }).toArray();

// Multi-field group with conditional counting
const statusBreakdown = await coll.aggregate([
  { $match: { orgId: orgId } },
  { $group: {
      _id: '$status',
      total:    { $sum: 1 },
      hasNotes: { $sum: { $cond: [{ $gt: [{ $strLenCP: { $ifNull: ['$notes', ''] } }, 0] }, 1, 0] } },
      firstSeen: { $min: '$createdAt' },
      lastSeen:  { $max: '$createdAt' },
  }},
  { $sort: { total: -1 } },
]).toArray();

$addToSet memory warning: $addToSet accumulates all distinct values in an in-memory array within the $group stage. For high-cardinality fields (user IDs, session IDs), this array grows to the full distinct value count and can exhaust the 100 MB stage limit. Use $addToSet only for low-cardinality fields; for counting distinct high-cardinality values, the HyperLogLog-based $approximateDistinctCount is not available in DocumentDB — either use a two-pass approach or accept approximate counts from sampled queries.

$unwind — decomposing arrays

$unwind deconstructs an array field into multiple documents, one per array element. It is the gateway for aggregating over nested arrays — without it, $group treats the entire array as a single value.

// Documents have shape: { userId, tags: ['billing', 'support', 'enterprise'] }
// Count how many documents each tag appears in

const tagCounts = await coll.aggregate([
  { $match: { createdAt: { $gte: thirtyDaysAgo } } },
  { $unwind: { path: '$tags', preserveNullAndEmpty: false } },
  // preserveNullAndEmpty: false (default) — skips docs where tags is null/missing/empty array
  { $group: { _id: '$tags', count: { $sum: 1 } } },
  { $sort: { count: -1 } },
  { $limit: 50 },
]).toArray();

// $unwind with includeArrayIndex — keep track of original array position
const withIndex = await coll.aggregate([
  { $unwind: { path: '$lineItems', includeArrayIndex: 'lineItemIndex' } },
  { $match: { 'lineItems.sku': targetSku } },
  { $project: { orderId: '$_id', lineItemIndex: 1, lineItem: '$lineItems' } },
]).toArray();

$lookup — left outer joins between collections

$lookup performs a left outer join with another collection in the same database. DocumentDB supports the basic equality join syntax in all versions and the pipeline-based correlated subquery syntax in 4.0+, but with limitations.

// Basic equality join — supported in all DocumentDB versions
// Join orders (localField: productId) with products (_id)
const ordersWithProducts = await db.collection('orders').aggregate([
  { $match: { status: 'pending', createdAt: { $gte: yesterday } } },
  { $lookup: {
      from:         'products',
      localField:   'productId',    // field in orders collection
      foreignField: '_id',          // field in products collection
      as:           'productDetails',  // output array field name
  }},
  // $lookup always returns an array; for 1:1 joins, unwind to get the object
  { $unwind: { path: '$productDetails', preserveNullAndEmpty: true } },
  // preserveNullAndEmpty: true = keep orders with no matching product (outer join)
  { $project: {
      orderId: '$_id',
      amount: 1,
      status: 1,
      productName:  '$productDetails.name',
      productPrice: '$productDetails.price',
  }},
]).toArray();

// Pipeline-based correlated $lookup — DocumentDB 4.0+ (test before relying on this)
// Useful for complex join conditions that go beyond simple equality
const withFilteredHistory = await db.collection('users').aggregate([
  { $match: { active: true } },
  { $lookup: {
      from: 'events',
      let: { uid: '$_id' },
      pipeline: [
        { $match: {
            $expr: { $eq: ['$userId', '$$uid'] },
        }},
        { $match: { eventType: 'purchase' } },
        { $sort: { createdAt: -1 } },
        { $limit: 5 },
      ],
      as: 'recentPurchases',
  }},
], { allowDiskUse: true }).toArray();

$lookup performance: each document in the input stage triggers a lookup into the joined collection. For a pipeline that processes 10,000 orders, that is 10,000 lookups into the products collection. Ensure the foreignField has an index — without an index, each lookup is a collection scan. For large join volumes, consider pre-joining with application-layer batching instead.

DocumentDB $lookup with pipeline limitation: in DocumentDB 4.0, the pipeline-based correlated $lookup (let + pipeline syntax) works for many use cases but does not support all MongoDB 5.x+ features within the sub-pipeline (e.g., nested $lookup within the sub-pipeline). Test your specific pipeline in a staging environment before deploying.

$facet — multi-faceted queries in a single round trip

$facet runs multiple sub-pipelines over the same input documents and returns all results in a single document. It is the standard pattern for search results pages that return paginated results alongside category counts, date histograms, or price ranges.

// Product search — return paginated results + facet counts + total count
// in a single aggregation query
const searchResults = await db.collection('products').aggregate([
  // 1. Filter: apply search criteria BEFORE $facet
  { $match: {
      category: { $in: selectedCategories },
      price: { $gte: minPrice, $lte: maxPrice },
      inStock: true,
  }},
  // 2. $facet runs three sub-pipelines over the filtered set
  { $facet: {
      // Sub-pipeline 1: paginated, sorted results
      results: [
        { $sort: { score: -1, createdAt: -1 } },
        { $skip:  page * pageSize },
        { $limit: pageSize },
        { $project: { name: 1, price: 1, category: 1, imageUrl: 1 } },
      ],
      // Sub-pipeline 2: category breakdown with counts
      categoryFacets: [
        { $group: { _id: '$category', count: { $sum: 1 } } },
        { $sort: { count: -1 } },
      ],
      // Sub-pipeline 3: total count (for pagination metadata)
      totalCount: [
        { $count: 'n' },
      ],
      // Sub-pipeline 4: price range histogram
      priceHistogram: [
        { $bucket: {
            groupBy:    '$price',
            boundaries: [0, 25, 50, 100, 200, 500, 1000],
            default:    'over_1000',
            output: { count: { $sum: 1 } },
        }},
      ],
  }},
], { allowDiskUse: true }).toArray();

const { results, categoryFacets, totalCount, priceHistogram } = searchResults[0];
const total = totalCount[0]?.n ?? 0;

// Return structured response to MCP caller
return {
  results,
  pagination: { total, page, pageSize, totalPages: Math.ceil(total / pageSize) },
  facets: { categories: categoryFacets, priceHistogram },
};

$bucket note: DocumentDB supports $bucket (uniform boundaries) and $bucketAuto (automatically computed equal-frequency boundaries). $bucket requires the boundaries array to be sorted and cover the full value range; documents outside all boundaries go to the default bucket if specified, or cause an error if not.

explain() — verifying index usage

// Run explain on an aggregation pipeline — returns query plan without executing
const explanation = await db.command({
  aggregate: 'events',
  pipeline: [
    { $match: { userId: userId, createdAt: { $gte: startDate } } },
    { $group: { _id: '$eventType', count: { $sum: 1 } } },
  ],
  cursor: {},
  explain: 'queryPlanner',  // or 'executionStats' for actual execution metrics
});

// Look for IXSCAN (index scan) in the winning plan — not COLLSCAN (collection scan)
// A $match that uses no index will show:
//   winningPlan: { stage: 'COLLSCAN' }
// A properly indexed $match will show:
//   winningPlan: { stage: 'FETCH', inputStage: { stage: 'IXSCAN', indexName: '...' } }

console.log(JSON.stringify(explanation, null, 2));

// Alternatively, use the cursor's explain() method
const cursor = db.collection('events').aggregate(pipeline, { allowDiskUse: true });
const plan = await cursor.explain('executionStats');
// plan.stages[0].$cursor.executionStats shows:
//   nReturned, executionTimeMillis, totalKeysExamined, totalDocsExamined
// totalDocsExamined >> nReturned = missing index

Reading the execution stats: totalKeysExamined is the number of index entries scanned; totalDocsExamined is the number of documents fetched from storage; nReturned is the number of documents emitted by the stage. A healthy query has totalDocsExamined ≈ nReturned. When totalDocsExamined is orders of magnitude larger than nReturned, add a more selective index or reorder the $match filter fields to match an existing compound index (index fields must appear in left-to-right order relative to the query predicate).

Supported aggregation operators reference

DocumentDB supports the following core pipeline stages and operators (verified as of DocumentDB 5.0):

Common errors