Aggregation Pipeline 2: $setWindowFields, $graphLookup, $bucket, $merge Materialized Views and Memory Limits

Version note: The examples in this post were run with MongoDB 9.0.8 Community Server and mongosh 2.6. The scripts gave the same result on MongoDB 8.0.38; the per-query memory limit exists only in 9.0, and the difference is in the relevant section.

Guess First

In the window functions post of the SQL Temelleri series, a 7-day moving average was computed for the December 2025 orders of Bookstore, the online bookstore used as the shared example across this site's series, and the correct value for 21 December, counting days without orders as 0, came out as 1.00. Let's do the same calculation in MongoDB with two windows: the last 7 documents and the last 7 days.

// 08 - guess first: 7-day moving average of daily orders, December 2025
// SQL Temelleri 14: avg_7_days on 2025-12-21 is 1.00 (days without orders count as 0)
db.orders.aggregate([
    { $match: { orderedAt: { $gte: ISODate("2025-12-01"), $lt: ISODate("2026-01-01") } } },
    { $group: { _id: { $dateTrunc: { date: "$orderedAt", unit: "day" } }, orders: { $sum: 1 } } },
    { $setWindowFields: {
        sortBy: { _id: 1 },
        output: {
            byDocuments: { $avg: "$orders", window: { documents: [-6, 0] } },
            byRange: { $avg: "$orders", window: { range: [-6, 0], unit: "day" } }
        }
    } },
    { $match: { _id: { $gte: ISODate("2025-12-20"), $lt: ISODate("2025-12-24") } } }
]).forEach(d => print(d._id.toISOString().slice(0, 10), "orders", d.orders,
    " documents:", d.byDocuments.toFixed(2), " range:", d.byRange.toFixed(2)));

The range window means "the last 7 days"; so which one gives 1.00?

2025-12-21 orders 2  documents: 1.43  range: 1.40
2025-12-22 orders 4  documents: 1.86  range: 2.00
2025-12-23 orders 1  documents: 1.71  range: 1.83

Neither. The documents window looks at the last 7 documents; it matches the ROWS BETWEEN 6 PRECEDING result in SQL (1.43) and, as we saw in that post, behaves as if days without orders did not exist. The range window picks the right days, but $avg divides by the number of documents that fall into the window: there are five days with orders between 15 and 21 December, so it divides 7 orders by 5 and gets 1.40. And 20 December is missing from the output altogether, because there were no orders that day and $group does not produce a document for a day that is not there. The problem is not the window but the data: if you want an average over calendar days, you first have to add the missing days. In this post we go through the advanced stages of the pipeline the same way, comparing the results with the numbers from SQL Temelleri. The scripts run as the application user; only the memory limit experiment needs admin.

$setWindowFields: Running Totals and Ranking within a Group

$setWindowFields is the counterpart of OVER (PARTITION BY ... ORDER BY ...) in SQL: without grouping the documents, it adds fields computed over a window to each document. partitionBy maps to PARTITION BY, sortBy to ORDER BY and window to the frame.

// Monthly revenue and year-to-date total (SQL Temelleri 14: SUM(...) OVER (ORDER BY month ROWS UNBOUNDED PRECEDING))
db.orders.aggregate([
    { $match: { status: { $ne: "Cancelled" }, orderedAt: { $gte: ISODate("2025-01-01"), $lt: ISODate("2026-01-01") } } },
    { $group: { _id: { $dateTrunc: { date: "$orderedAt", unit: "month" } }, revenue: { $sum: "$total" } } },
    { $setWindowFields: {
        sortBy: { _id: 1 },
        output: { ytd: { $sum: "$revenue", window: { documents: ["unbounded", "current"] } } }
    } }
]).forEach(m => print(m._id.toISOString().slice(0, 7), m.revenue.toString().padStart(9), m.ytd.toString().padStart(10)));

// Top-N per group (SQL Temelleri 13): the best-selling book of each category by units
db.orders.aggregate([
    { $match: { status: { $ne: "Cancelled" } } },
    { $unwind: "$items" },
    { $group: { _id: "$items.bookId", units: { $sum: "$items.quantity" } } },
    { $lookup: { from: "books", localField: "_id", foreignField: "_id", as: "book" } },
    { $set: { category: { $first: "$book.category.name" }, title: { $first: "$book.title" } } },
    { $setWindowFields: { partitionBy: "$category", sortBy: { units: -1 }, output: { rank: { $rank: {} } } } },
    { $match: { rank: 1, category: { $in: ["Science Fiction", "Programming", "Dystopian"] } } },
    { $sort: { category: 1 } }
]).forEach(b => print(b.category.padEnd(16), b.title.padEnd(26), "units", b.units));

2025-01   8959.20    8959.20
2025-02  13727.50   22686.70
2025-03  20367.60   43054.30
2025-04  18488.20   61542.50
2025-05  13679.60   75222.10
2025-06  15414.30   90636.40
2025-07  12866.70  103503.10
2025-08  12681.50  116184.60
2025-09  33349.23  149533.83
2025-10  27489.90  177023.73
2025-11  20325.80  197349.53
2025-12  41866.20  239215.73
Dystopian        Nineteen Eighty-Four       units 30
Programming      The Pragmatic Programmer   units 16
Science Fiction  Dune                       units 32

Both the revenue and the year-to-date total of all twelve months match the table in SQL Temelleri to the cent. documents: ["unbounded", "current"] is ROWS UNBOUNDED PRECEDING in SQL; in that post we saw that SQL Server's default RANGE frame runs slowly when you do not write ROWS; here you always write the frame explicitly. The second query is top-N per group: $rank gives the sales order within the category, and rank: 1 keeps each category's best seller. With a tie, $rank gives 1 to both books; if you want exactly one, use $documentNumber (ROW_NUMBER in SQL).

$densify and $fill: Adding the Missing Days

In SQL we generated a calendar table (GENERATE_SERIES in that post) for the missing days and did a LEFT JOIN. In MongoDB the counterpart is two stages: $densify adds documents for the missing values of a field at a given step, and $fill fills the empty fields in these new documents.

// 08 - $densify + $fill: add the days without orders, then average over calendar days
db.orders.aggregate([
    { $match: { orderedAt: { $gte: ISODate("2025-12-01"), $lt: ISODate("2026-01-01") } } },
    { $group: { _id: { $dateTrunc: { date: "$orderedAt", unit: "day" } }, orders: { $sum: 1 } } },
    { $densify: { field: "_id", range: { step: 1, unit: "day", bounds: [ISODate("2025-12-01"), ISODate("2026-01-01")] } } },
    { $fill: { output: { orders: { value: 0 } } } },
    { $setWindowFields: {
        sortBy: { _id: 1 },
        output: { avg7: { $avg: "$orders", window: { range: [-6, 0], unit: "day" } } }
    } },
    { $match: { _id: { $gte: ISODate("2025-12-20"), $lt: ISODate("2025-12-24") } } }
]).forEach(d => print(d._id.toISOString().slice(0, 10), "orders", d.orders, " avg7:", d.avg7.toFixed(2)));

2025-12-20 orders 0  avg7: 0.71
2025-12-21 orders 2  avg7: 1.00
2025-12-22 orders 4  avg7: 1.43
2025-12-23 orders 1  avg7: 1.57

20 December came back with 0 orders, and the average for 21 December is 1.00, like avg_7_days in SQL. bounds spreads the range across the whole month; with "full" it would only fill between the first and the last document, and days without orders at the start of the month would stay missing. $fill can fill not only with a constant but also with method: "locf" (carry the last known value) or "linear"; for a measurement such as stock level locf is right, for a count such as orders it is 0.

$graphLookup: the Category Tree

Categories form a tree through parentId: Fiction > Science Fiction > Dystopian. In the CTE post of SQL Temelleri we walked this tree with a recursive CTE. $graphLookup does the same job in one stage: it starts from a value and repeatedly follows the link between connectFromField and connectToField.

// Path of every category: walk up through parentId
db.categories.aggregate([
    { $graphLookup: { from: "categories", startWith: "$parentId", connectFromField: "parentId",
                      connectToField: "_id", as: "ancestors", depthField: "level" } },
    { $set: { depth: { $size: "$ancestors" },
              path: { $reduce: {
                  input: { $sortArray: { input: "$ancestors", sortBy: { level: -1 } } },
                  initialValue: "",
                  in: { $concat: ["$$value", "$$this.name", " > "] } } } } },
    { $set: { path: { $concat: ["$path", "$name"] } } },
    { $sort: { path: 1 } },
    { $limit: 8 }
]).forEach(c => print(String(c._id).padStart(2), c.depth, c.path));

// Subtree: every category under Non-Fiction (8), then the books in them
const subtree = db.categories.aggregate([
    { $match: { _id: 8 } },
    { $graphLookup: { from: "categories", startWith: "$_id", connectFromField: "_id",
                      connectToField: "parentId", as: "descendants" } },
    { $project: { ids: { $concatArrays: [["$_id"], "$descendants._id"] } } }
]).next().ids;
print("categories under Non-Fiction:", subtree.sort((a, b) => a - b).join(", "),
      " books:", db.books.countDocuments({ "category.categoryId": { $in: subtree } }));

13 0 Children
 1 0 Fiction
 2 1 Fiction > Classics
 5 1 Fiction > Fantasy
 6 1 Fiction > Mystery
 7 1 Fiction > Poetry
 3 1 Fiction > Science Fiction
 4 2 Fiction > Science Fiction > Dystopian
categories under Non-Fiction: 8, 9, 10, 11, 12  books: 16

The paths and depths match the output of the recursive CTE in SQL, and so do the 16 books under Non-Fiction. There are two directions: upwards (startWith: "$parentId", connectToField: "_id") gives the ancestors, downwards (startWith: "$_id", connectToField: "parentId") the subtree. $graphLookup returns the result as an array and does not guarantee order; building the path needed $sortArray by the level coming from depthField. You can limit the depth with maxDepth. Since the stage searches the foreign collection by connectToField again and again for every starting document, that field should be indexed; beyond 100 MB it writes temporary files. It suits small hierarchies such as a category tree, not large graphs. For a tree that is read very often, the approach from post 5 is cheaper: writing the list of ancestors into the category document.

$bucketAuto: Evenly Distributed Ranges

The $bucket in post 7 drew the boundaries at the values you gave it. To $bucketAuto you give only the number of buckets, and it chooses the boundaries itself so that roughly the same number of documents falls into each bucket:

// 08 - $bucketAuto: price ranges with roughly the same number of books
db.books.aggregate([
    { $bucketAuto: { groupBy: "$listPrice", buckets: 4, output: { books: { $sum: 1 } } } }
]).forEach(b => print(`${b._id.min} - ${b._id.max}`.padEnd(18), "books", b.books));

200.00 - 350.00    books 20
350.00 - 470.00    books 19
470.00 - 660.00    books 20
660.00 - 1450.00   books 18

The ranges are irregular but evenly filled; handy for generating price filter options from the real distribution in the catalog. If you want "round" boundaries, the granularity option (for example the R5 or 1-2-5 series) snaps the boundaries to known number series.

Materialized Views with $merge

In post 7 we suggested not tying heavy reports to screen requests. For a daily sales summary the result should be written to a collection that the screen reads from: an on-demand materialized view. $merge combines the pipeline's result with a collection; recomputing only the affected days is possible as well.

// 08 - $merge: a daily sales summary as a materialized view, refreshed on demand
function refresh(from) {
    db.orders.aggregate([
        { $match: { status: { $ne: "Cancelled" }, orderedAt: { $gte: from } } },
        { $group: { _id: { $dateTrunc: { date: "$orderedAt", unit: "day" } },
                    orders: { $sum: 1 }, revenue: { $sum: "$total" } } },
        { $set: { refreshedAt: "$$NOW" } },
        { $merge: { into: "daily_sales", on: "_id", whenMatched: "replace", whenNotMatched: "insert" } }
    ]);
}

refresh(ISODate("2000-01-01"));                       // full build
show("full build");

// A new order arrives; only the affected days are recomputed
db.orders.insertOne({ _id: 380, customer: { customerId: 3, name: "Grace Hopper" }, orderedAt: ISODate("2026-06-29T22:10:00Z"),
    status: "Pending", items: [{ bookId: 26, title: "Dune", quantity: 1, unitPrice: Decimal128("390.50") }],
    total: Decimal128("390.50"), currency: "TRY", payments: [] });
show("before refresh");
refresh(ISODate("2026-06-29"));
show("incremental refresh");
refresh(ISODate("2026-06-29"));
show("same refresh again");

full build             days 273  2026-06-29: 1 orders 440.00
before refresh         days 273  2026-06-29: 1 orders 440.00
incremental refresh    days 273  2026-06-29: 2 orders 830.50
same refresh again     days 273  2026-06-29: 2 orders 830.50
2026-06-29 2 830.50
2026-06-22 1 440.00
2026-06-20 1 290.90

The first call built the 273-day summary. When the new order arrived the summary went stale; refresh recomputed only from 29 June onwards and replaced that day's document; running it a second time broke nothing, so the refresh is idempotent. The screen now reads a few small documents through the _id index instead of 379 orders. The cost is staleness: the summary is as current as the last refresh, and the refreshedAt field shows it. Whether to tie the refresh to a scheduler (a full run every night, the last two days every 5 minutes) or to the change stream from post 10 is your call. At the end the script deletes the order and the summary collection.

$out is a simpler sibling: it builds the target collection from scratch with the result and atomically swaps it in for the old one when done; it cannot do incremental updates and cannot write to a sharded collection. $out for small reports that are fully rebuilt, $merge for incremental refresh. If you use a field other than _id as on, the target collection needs a unique index on those fields.

Memory Limits and the Per-Query Limit in 9.0

In post 7 we mentioned the 100 MB limit per stage and that a stage exceeding it writes temporary files (spills). MongoDB 9.0 added a limit on the total memory of a query on top of this: by default 1 GB or 20% of the server's memory, whichever is greater. In the lab I lowered this limit to 20 MB just for the test and ran a 30 MB $group:

// 08 - MongoDB 9.0: per-operation memory limit (needs admin; the limit is lowered only for this test)
const pipeline = [{ $group: { _id: "$bookId", payloads: { $push: "$payload" } } }];
const param = "internalQueryMaxMemoryUsageBytesPerOperation";

print("default limit:", Math.round(Number(original) / 1024 / 1024), "MB",
      " peak memory of the $group:", Number(bookstore.memory_test.explain("executionStats").aggregate(pipeline).peakTrackedMemBytes), "bytes");
db.adminCommand({ setParameter: 1, [param]: 20 * 1024 * 1024 });
run("20 MB limit, allowDiskUse: false", { allowDiskUse: false });
run("20 MB limit, allowDiskUse: true ", { allowDiskUse: true });
db.adminCommand({ setParameter: 1, [param]: original });
run("default,     allowDiskUse: false", { allowDiskUse: false });

default limit: 1567 MB  peak memory of the $group: 30225118 bytes
20 MB limit, allowDiskUse: false: QueryExceededMemoryLimitNoDiskUseAllowed
20 MB limit, allowDiskUse: true : ok, 77 groups
default,     allowDiskUse: false: ok, 77 groups

The memory_test collection is temporary data generated by the script: 30,000 documents of 1 KB each. The default limit in this lab is 1567 MB, which is 20% of the memory given to Docker's virtual machine; it will be different on your machine. The peakTrackedMemBytes field in explain, added in 9.0, shows that the $group used 30 MB: below the stage limit (100 MB) but above the query limit we lowered. With allowDiskUse: false the query stopped with QueryExceededMemoryLimitNoDiskUseAllowed; with allowDiskUse: true the stage spilled to disk and finished. Once the limit was back at the default, the same query ran again. In practice: a single expensive report can no longer consume all of the server's memory, but a very large query may hit this error for the first time after moving to 9.0. The cure, as the documentation suggests, is to move the filter forward, put a $limit after the $sort and let an index do the sorting.

The difference on MongoDB 8.0: the parameter does not exist. On 8.0.38 the getParameter call failed with no option found to get, and the same $group ran with allowDiskUse: false too, because 30 MB is below the stage limit.

Server-Side JavaScript: $function

$function and $accumulator run JavaScript inside a pipeline. They were deprecated in 8.0; according to the 9.0 release notes the deprecation was lifted, and JavaScript now runs in a WebAssembly (WASM) based engine with stronger isolation. Let's validate the ISBN-13 check digit both with JavaScript and with aggregation operators:

// 08 - server-side JavaScript with $function versus a native expression: ISBN-13 check digit
const viaFunction = db.books.aggregate([
    { $set: { valid: { $function: {
        body: function (isbn) {
            let sum = 0;
            for (let i = 0; i < 12; i++) sum += Number(isbn[i]) * (i % 2 === 0 ? 1 : 3);
            return (10 - sum % 10) % 10 === Number(isbn[12]);
        },
        args: ["$isbn"],
        lang: "js"
    } } } },
    { $group: { _id: "$valid", books: { $sum: 1 } } }
]).toArray();

// The same rule with aggregation operators: no JavaScript engine involved
const viaOperators = db.books.aggregate([
    { $set: { sum: { $reduce: {
        input: { $range: [0, 12] },
        initialValue: 0,
        in: { $add: ["$$value", { $multiply: [
            { $toInt: { $substrCP: ["$isbn", "$$this", 1] } },
            { $cond: [{ $eq: [{ $mod: ["$$this", 2] }, 0] }, 1, 3] }] }] }
    } } } },
    { $set: { valid: { $eq: [{ $mod: [{ $subtract: [10, { $mod: ["$sum", 10] }] }, 10] },
                              { $toInt: { $substrCP: ["$isbn", 12, 1] } }] } } },
    { $group: { _id: "$valid", books: { $sum: 1 } } }
]).toArray();

$function: [{"_id":true,"books":77}]
operators: [{"_id":true,"books":77}]

Both ways found all 77 ISBNs valid. In the 9.0.8 log the first call produced the line "Initialized shared WASM engine context" (98 ms), but a $function is deprecated warning (id 8996503) was written for the same call too; the release notes and the log message do not agree yet in this patch. 8.0.38 wrote the same warning and ran the query. Even with the deprecation lifted, the rule does not change: the documentation says $function can reduce performance and should be used only when the operators cannot express the logic. JavaScript is a black box for the optimizer and cannot be moved into the query engine, and there is a 9.0-specific detail as well: local Date operations inside JavaScript now run in UTC regardless of the server's time zone. The version written with operators looks longer, but it is the one to prefer.

In Production

Do not put the reporting load on the primary. Heavy pipelines such as monthly revenue, top-N and materialized view refreshes share the primary's CPU, memory and WiredTiger cache with the application's writes. The options: reading reports from a secondary through read preference (accepting staleness, post 12), a member set aside for reports that the application does not see (a hidden or tagged analytics node, post 13), or refreshing materialized views at quiet hours.

Manage a materialized view like any collection. Index it for its read pattern, and monitor how long refreshes take and when the last successful refresh happened; a field such as refreshedAt lets the screen say "data updated 5 minutes ago". If a refresh stops halfway, $merge does not roll back what it has written so far; that is why the refresh has to be idempotent, so that running it again is always safe.

Common Pitfalls

Forgetting the missing days. As in "Guess First", $group produces no document for days that are not there; that means gaps in a daily chart, wrong values in a moving average and missing rows in a "days without sales" report. In time series reports, treat the $densify and $fill step as the default.

Not writing the window frame. If you do not give window in $setWindowFields, operators such as $sum use the whole partition, even with sortBy; every row gets the grand total instead of a running total. Always write the frame explicitly.

Using $out for incremental refresh. $out rebuilds the target from scratch every time; recomputing the 273-day summary for a single new order is invisible with small data, but with years of accumulated data it turns every refresh into a full scan.

A Small Experiment

★☆☆ Number each customer's orders in date order ($documentNumber) and compute the number of days since the previous order ($shift). Find the longest gap in Ada Lovelace's orders.

★★☆ Produce the order count for every week of 2025 without gaps using $densify, and compute a 4-week moving average. Start the week on Monday with $dateTrunc (startOfWeek).

★★★ Build a materialized view per category and day in addition to daily_sales (with two fields as on; add the required unique index). Write a script that determines which documents must be recomputed when an order's status becomes Cancelled and refreshes only those.

With this post the groundwork on the mongosh side is done. In the next post we move to .NET: registering MongoClient as a singleton with DI, POCO mapping and conventions, the BSON counterparts of decimal, Guid and DateTime, and querying with LINQ.