Aggregation Pipeline

The previous lesson built the evaluator that says whether a document satisfies a condition. Most of the library’s questions do not end there: “how many copies of each author’s books are under repair,” “the five most-borrowed books,” “the number of shelved copies per branch.” What these questions have in common is that the answer sits not in individual documents but in the aggregate of documents, and a condition expression cannot aggregate.

The document model does this work with an aggregation pipeline: a chain of ordered stages. Each stage takes a stream of records, transforms it, and hands it to the next. The word is the same one used for the pipeline that chains shell commands together, and the resemblance is real; where they part is the unit of the stream — what flows through a shell is bytes, what flows through here is records, and the stages operate on records. This lesson’s question is not what the stages do, but what their order costs.

Streaming Stage and Blocking Stage

Stages split into two classes. A streaming stage takes a record, processes it, releases it: filtering, unwinding an array, and projection all work this way, holding a single record in memory at a time. A blocking stage must see the entire stream to produce its result: sorting holds the whole stream in memory, grouping holds the group table. A pipeline’s memory cost is the number of records its blocking stages hold at any one time.

// pipeline.mjs — staged transformation. Each stage takes a stream of records, transforms
// it, and hands it to the next. The records in/out per stage and the records that stage
// holds in memory are counted.
export function run(source, stages) {
  let stream = source, peak = 0, processed = 0;
  const report = [];
  for (const stage of stages) {
    const name = Object.keys(stage)[0], recordsIn = stream.length;
    let memory = 1;                                    // streaming stage: holds a single record
    if (name === "filter") stream = stream.filter(stage.filter);
    else if (name === "unwind") {                       // opens an array into its elements
      const field = stage.unwind;
      stream = stream.flatMap((b) => (b[field] ?? []).map((o) => ({ ...b, [field]: o })));
    } else if (name === "group") {
      const table = new Map();
      for (const b of stream) {
        const key = stage.group.key(b);
        table.set(key, stage.group.combine(table.get(key) ?? stage.group.empty(key), b));
      }
      stream = [...table.values()];
      memory = table.size;                              // blocking: the group table sits in memory
    } else if (name === "sort") {
      memory = recordsIn;                               // blocking: the entire stream sits in memory
      stream = [...stream].sort(stage.sort);
    } else if (name === "limit") stream = stream.slice(0, stage.limit);
    else if (name === "project") stream = stream.map(stage.project);
    processed += recordsIn;
    peak = Math.max(peak, memory);
    report.push({ name, recordsIn, recordsOut: stream.length, memory });
  }
  return { result: stream, report, processed, peak };
}

export function print(label, stats) {
  console.log(`${label}: records processed ${stats.processed}, peak memory ${stats.peak} records`);
  for (const r of stats.report)
    console.log(`  ${r.name.padEnd(8)} in ${String(r.recordsIn).padStart(6)}` +
      ` -> out ${String(r.recordsOut).padStart(6)}   memory ${r.memory}`);
}

The unwind stage is specific to the document model, and it is the pipeline’s most expensive stage: a book document carrying a copy array turns into as many records as the array has elements. 20,000 book documents, once unwound, become 59,494 records. Every stage after this one processes about three times as many records.

Same Question, Three Orders

NS7 (assumption): the catalog is 20,000 book documents, each book has 1–5 copies and 2–4 tags, and the seed is 424242. The question is fixed: the five authors ranked highest by number of copies under repair at the Kadikoy branch. Three pipelines answer this question. A puts the filter at the very start; B unwinds the array first and filters afterward; C groups without filtering at all, and filters over the groups. All three must give the correct answer — what is measured is not correctness, but cost.

// order-measurement.mjs — the same question is answered with three different stage
// orders. The result is the same, records processed and peak memory are not.
// pipeline.mjs is in the same directory.
import { run, print } from "./pipeline.mjs";

let seed = 424242;                                        // visible seed
const random = () => (seed = (seed * 1103515245 + 12345) % 2147483648) / 2147483648;
const BRANCH = ["Central", "Bahcelievler", "Kadikoy", "Beyoglu", "Konak", "Nilufer"];
const STATUS = ["shelved", "checked_out", "in_repair"];
const TAGS = ["fiction", "history", "children", "poetry", "science", "reference"];

const CATALOG = [];
for (let i = 1; i <= 20000; i += 1) {
  const copy = [], tag = [];
  for (let j = 0, n = 1 + Math.floor(random() * 5); j < n; j += 1)
    copy.push({ barcode: `B${String(i * 10 + j).padStart(7, "0")}`,
      branch: BRANCH[Math.floor(random() * 6)], status: STATUS[Math.floor(random() * 3)] });
  for (let j = 0, n = 2 + Math.floor(random() * 3); j < n; j += 1)
    tag.push(TAGS[Math.floor(random() * 6)]);
  CATALOG.push({ _k: `K-${String(i).padStart(5, "0")}`, author: `Author ${i % 4000}`,
    publication_year: 1950 + (i % 75), tag, copy });
}
console.log(`catalog ${CATALOG.length} books, ` +
  `${CATALOG.reduce((t, b) => t + b.copy.length, 0)} copies`);

// Question: top 5 authors by number of copies under repair at the Kadikoy branch.
const REPAIR = (k) => k.branch === "Kadikoy" && k.status === "in_repair";
const GROUP = { key: (b) => b.author, empty: (k) => ({ author: k, count: 0 }),
  combine: (g, b) => ({ ...g, count: g.count + 1 }) };
const SORT = (a, b) => b.count - a.count || (a.author < b.author ? -1 : 1);

const A = run(CATALOG, [
  { filter: (b) => b.copy.some(REPAIR) },               // filter first: at the book level
  { unwind: "copy" },
  { filter: (b) => REPAIR(b.copy) },                    // filter the unwound element
  { group: GROUP }, { sort: SORT }, { limit: 5 },
]);
const B = run(CATALOG, [
  { unwind: "copy" },
  { filter: (b) => REPAIR(b.copy) },
  { group: GROUP }, { sort: SORT }, { limit: 5 },
]);
const C = run(CATALOG, [
  { unwind: "copy" },
  { group: { key: (b) => `${b.author}|${b.copy.branch}|${b.copy.status}`,
    empty: (k) => ({ author: k.split("|")[0], branch: k.split("|")[1],
      status: k.split("|")[2], count: 0 }),
    combine: (g, b) => ({ ...g, count: g.count + 1 }) } },
  { filter: (g) => g.branch === "Kadikoy" && g.status === "in_repair" },
  { sort: SORT }, { limit: 5 },
]);

print("A filter first", A);
print("B unwind then filter", B);
print("C group first then filter", C);
const summary = (r) => r.result.map((g) => `${g.author}=${g.count}`).join(" ");
console.log(`A result: ${summary(A)}`);
console.log(`all three the same: ${summary(A) === summary(B) && summary(B) === summary(C)}`);
console.log(`records processed  A ${A.processed}  B ${B.processed}  C ${C.processed}` +
  `   B/A ${(B.processed / A.processed).toFixed(2)}  C/A ${(C.processed / A.processed).toFixed(2)}`);
console.log(`peak memory   A ${A.peak}  B ${B.peak}  C ${C.peak}` +
  `   C/A ${(C.peak / A.peak).toFixed(1)}x`);

catalog 20000 books, 59494 copies
A filter first: records processed 43610, peak memory 2418 records
  filter   in  20000 -> out   3340   memory 1
  unwind   in   3340 -> out  11926   memory 1
  filter   in  11926 -> out   3508   memory 1
  group    in   3508 -> out   2418   memory 2418
  sort     in   2418 -> out   2418   memory 2418
  limit    in   2418 -> out      5   memory 1
B unwind then filter: records processed 87838, peak memory 2418 records
  unwind   in  20000 -> out  59494   memory 1
  filter   in  59494 -> out   3508   memory 1
  group    in   3508 -> out   2418   memory 2418
  sort     in   2418 -> out   2418   memory 2418
  limit    in   2418 -> out      5   memory 1
C group first then filter: records processed 124606, peak memory 40276 records
  unwind   in  20000 -> out  59494   memory 1
  group    in  59494 -> out  40276   memory 40276
  filter   in  40276 -> out   2418   memory 1
  sort     in   2418 -> out   2418   memory 2418
  limit    in   2418 -> out      5   memory 1
A result: Author 1068=4 Author 111=4 Author 1530=4 Author 1580=4 Author 1618=4
all three the same: true
records processed  A 43610  B 87838  C 124606   B/A 2.01  C/A 2.86
peak memory   A 2418  B 2418  C 40276   C/A 16.7x

The Cost of the Order

The three pipelines return the same five authors with the same counts; the all three the same line confirms it. Where they part is cost, and cost is measured with two separate numbers.

Records processed is 43,610 in A, 87,838 in B, 124,606 in C. The difference between A and B comes only from the first filter: A feeds the unwind stage 3,340 of the 20,000 books, B feeds it all of them. The unwind stage produces 11,926 records in A, 59,494 in B — five times as many. Against that, A runs one extra stage, and its first filter looks at 20,000 documents. Overall the ratio is 2.01. The rule is this: a filter is placed not where it will narrow the stream the most, but before the stage that multiplies records; unwind is a multiplier.

A’s first filter is at the document level, not the element level, and it is the loose form the previous lesson measured: it returns 3,340 documents. Of the 11,926 records that unwinding then produces, only 3,508 pass the second filter. This looseness is not a mistake, it is a deliberate choice — a document-level filter does not miss the correct result, it only lets extra records through. This is the only condition on a filter placed early in a pipeline: it must be inclusive, it does not need to be narrowing.

The second number gives memory. Peak memory in A and B is 2,418 records — the size of the group table, that is, the number of distinct authors with a copy under repair at Kadikoy. In C the same number is 40,276, 16.7 times as many. C’s mistake is placing the grouping before the filter: the grouping now holds not just the authors of interest, but every combination of the author–branch–status triple. This number grows with the product of the group key as the collection grows, and a blocking stage exceeding its memory limit leads to the pipeline spilling to disk or being rejected outright.

The final limit stage sits in the same place in all three and reduces no work in any of them: sorting has already seen all 2,418 records. Limiting does not shrink the work of the blocking stage that precedes it — only a form of sorting that itself keeps just the top five can do that, and that is a decision inside the sort stage, not in the pipeline’s order.

Summary

• An aggregation pipeline is made of ordered stages; streaming stages hold a single record, blocking stages (grouping, sorting) hold the entire stream or the group table in memory.
• The unwind stage multiplies records: 20,000 book documents turn into 59,494 records.
• The same question gives the same answer under three orders, but processes 43,610, 87,838, and 124,606 records; moving the filter before the multiplying stage roughly halves the records processed.
• It is enough for an early filter to be inclusive; A’s document-level filter lets 3,340 documents through, and of the 11,926 records that unwinding them produces, only 3,508 remain in the result.
• Placing grouping before filtering raises peak memory from 2,418 records to 40,276 (16.7×).

Next Step

One thing the three pipelines have in common went unnoticed: all three read the entire collection, 20,000 documents, in their first stage. The stage order changed the records processed and the memory held, not the number of documents read. The only thing that changes that number is the index. In the document model, the index is not structurally different from the index measured in the relational course — but the data it is built over is different: an array field produces more than one entry for a document, an embedded field is addressed by a dotted path, and an index may be asked to cover only part of the documents. The next lesson builds these three cases and counts the index size and the write cost.