aggregate

aggregate

Group sorted child streams under a master by key, folding each child's per-key group into one result. SQL-style GROUP BY over streams: one output row per key, streaming and single-pass (only one group's accumulators are live at a time — and a scalar fold like a count or sum holds nothing).

import aggregate from 'stream-sorting/sorted/aggregate.js';

// `departments` and `employees`, each sorted by department id
for await (const dept of aggregate(
  {input: departments, key: d => d.id},
  {employees: {input: employees, key: e => e.deptId}}
)) {
  // dept === {...departmentRow, employees: [ ...its employees ]}
}

The result is an AsyncIterable<R>; convert at the boundary you need (readableFrom(...), ReadableStream.from(...)). Every input must be sorted by its key under compareKey.

The master

The first argument is the master — the spine that drives one output row per distinct key. It comes in two forms:

• {input, key} descriptor — an external master stream (a dimension table). Its rows carry data into the output, and its keys define which rows emit: a master key with no matching children still emits (with empty child results); child items with no master are dropped. This is the "enrich a table with aggregated children" case.
• key => object function — the base object is synthesized from the group key alone (group-by). The spine is then the children's own distinct keys.

// group-by: no master stream, base built from the key
aggregate(deptId => ({id: deptId}), {employees: {input: employees, key: e => e.deptId}});
// → {id, employees: [...]} per distinct deptId

A descriptor master also accepts init / fold / finalize (below): duplicate master rows at one key are folded into a single base — first row wins by default — so a fold can merge them instead.

Children

Each value in the children map is a descriptor that folds its per-key group:

Field	Type	Default	Description
input	AsyncIterable | Iterable	—	The sorted child stream. Required.
key	(item) => K	identity	Extracts the group key.
init	() => A	() => []	Creates the accumulator for a new group (no arguments).
fold	(acc, item) => A	push the item	Folds one item into the accumulator.
finalize	(acc) => R	identity	Turns the accumulator into the result for combine.
required	boolean	false	When true, masters whose group for this child is empty are dropped.

The fold lifecycle is scoped to the master's key boundaries: new key → init(), each item → fold, key ends → finalize. A master with no items for a child gets finalize(init()) (e.g. [], or 0 for a count) — which is why init takes no arguments. Every scalar SQL aggregate is just a fold:

aggregate(
  {input: departments, key: d => d.id},
  {
    employees: {input: employees, key: e => e.deptId}, // default: collect to an array
    headcount: {input: employees, key: e => e.deptId, init: () => 0, fold: n => n + 1},
    avgSalary: {
      input: salaries,
      key: s => s.deptId,
      init: () => ({sum: 0, n: 0}),
      fold: (a, s) => ({sum: a.sum + s.amount, n: a.n + 1}),
      finalize: a => (a.n ? a.sum / a.n : 0)
    }
  }
);

ARRAY_AGG ordering is free — children arrive pre-sorted, so the collected array is already in key order.

Options

Option	Type	Default	Description
compareKey	(a, b) => number	natural order	Orders keys. Provide this or lessKey. Composite (tuple) keys need an explicit one.
lessKey	(a, b) => boolean	—	Strict-less key comparator. Provide this or compareKey.
combine	(base, parts) => R	(base, parts) => ({...base, ...parts})	Builds each output row from the base (master) and the named bag of child results. undefined drops the row.
maxGroupSize	number	—	Throw if any single group folds more than this many items.

combine receives the master base and parts — a named bag keyed by the child names ({employees, headcount, …}). The default spreads them together.

Multi-level nesting

Nesting (department → employee → equipment) is composition, not a built-in: enrich the deep stream's foreign keys with a join, re-sort it to the ancestor key path ([deptId, empId]), then run a flat aggregate per level — each level stays linear, and aggregate emits in key order so its output feeds the next level still sorted.

Notes

• Inputs must be sorted by key; an out-of-order master throws, and out-of-order children are caught within the consumed range.
• Composite keys: return a tuple from key and pass a matching compareKey (e.g. (x, y) => x[0] - y[0] || x[1] - y[1]).

See also

join, sort, ObjectStreamWrapper.