Streams and live windows

A stream is a live table spread over the workers of a SplitEngine. Rows arrive as CSV lines, JSON lines or binary column frames; each worker parses its share into growing columns, and you query the rows kept so far with the same calls as any split table.

Streams are a prototype: the API may change, and the limits at the end of this page still apply.

import { SplitEngine } from 'vectrojs';

const engine = new SplitEngine({ workers: 2 });   // default: one worker per core
const orders = await engine.streamAsync({
  format: 'ndjson',
  columns: { ts: 'number', cust: 'number', price: 'number', region: 'category' },
  keepRows: 1_000_000,                            // the oldest rows are dropped past this
});

// JSON lines, in chunks cut anywhere (here: mid-line)
let lines = '';
for (let i = 0; i < 20_000; i++) {
  lines += JSON.stringify({ ts: i * 10, cust: i % 500, price: (i * 37) % 100, region: ['EU', 'US', 'APAC'][i % 3] }) + '\n';
}
await orders.push(lines.slice(0, 500_000));
await orders.push(lines.slice(500_000));
await orders.push('not a JSON object\n');           // counted in badLines, skipped

const { rows, badLines } = await orders.flushAsync();   // parse what has arrived
console.log({ rows, badLines });

console.log(await orders.filter("region == 'EU'").aggAsync({ n: 'count()', avg: 'mean(price)' }));
console.log((await orders.groupBy('region').aggAsync({ n: 'count()', revenue: 'sum(price)' })).toRows());
console.log((await orders.topAsync(2, '-price')).toRows());

await engine.close();

streamAsync takes:

Option	Meaning
columns	{ name: 'number' | 'category' | 'date' }; dates are milliseconds or ISO text
format	'csv' (lines with no header, fields in the order of columns), 'ndjson', 'vectro' (frames from an Emitter) or 'arrow' (an Arrow IPC stream)
keepRows	rows kept in total; the oldest are dropped
keepMs, timeColumn	keep only rows newer than the newest time seen minus keepMs

Feeding a stream

There are two ways in.

push(chunk) sends text or bytes from your own thread. Chunks can be cut anywhere; whole lines go to the workers in turn. The promise resolves once a worker has queued them.

connect(worker, target) has a worker open its own connection, so the data never passes through your thread:

• { url: 'wss://…' }: a WebSocket, in browsers and Node 22 and later. Each message must hold whole lines.
• { host, port }: raw TCP, Node only, cut anywhere.

await Promise.all(engine.workers.map((_, w) =>
  orders.connect(w, { url: `wss://feed.example.com/orders?part=${w}` })));

Rows are parsed in batches of 256 KB (or 20 ms after the first unparsed line), one batch per event-loop turn, so a query waits for at most one batch. Call flushAsync() to parse everything received before a query; it returns { rows, rowsIn, badLines, bytesIn, batches, unparsedBytes, perWorker, errors }, and statsAsync() returns the same without parsing.

JSON lines are parsed byte by byte straight into columns, with no object per row: keys can come in any order, extra and nested keys are skipped, null, missing values and numbers that don’t parse are empty, and lines that aren’t a JSON object are counted in badLines and skipped.

Backpressure. When a worker has more than 4 MB waiting to be parsed, it pauses its TCP socket (or delays its reply to push) until it is down to 1 MB, and TCP then slows the sender. WebSockets can’t be paused, so a WebSocket feed faster than the workers grows their backlog.

Querying

A stream has the split-table query calls: filter, aggAsync, countAsync, sumAsync and friends, groupBy().aggAsync, topAsync, toRowsAsync, toColumnsAsync, toArrowAsync. Each sees the rows parsed so far. A filtered stream (orders.filter(...)) is still live.

Rows have no global order across workers, so toRowsAsync() returns worker 0’s rows first. deriveAsync needs fixed rows: call snapshotAsync() first, which makes a normal split table from the rows kept right now. Zone maps are kept up to date as batches arrive, so “the last minute” reads only the newest blocks.

Live windows

windowAsync keeps running totals over a time window, updated as rows arrive. Each worker keeps one small total per (slice of every ms, group), so reading the window costs about a millisecond, not a scan, and the totals outlive rows dropped by keepRows. The window ends at the newest time seen and covers whole slices.

import { SplitEngine } from 'vectrojs';

const engine = new SplitEngine({ workers: 2 });
const orders = await engine.streamAsync({
  format: 'csv',
  columns: { ts: 'number', cust: 'number', price: 'number', region: 'category' },
});

const live = await orders.windowAsync({
  by: 'region',              // one category column, or leave it out
  every: 1000,               // slices of one second
  last: 10_000,              // the window: the last 10 seconds
  timeColumn: 'ts',
  agg: { orders: 'count()', revenue: 'sum(price)', avg: 'mean(price)', customers: 'approxDistinct(cust)' },
});

// 60 seconds of orders, 100 a second
let csv = '';
for (let i = 0; i < 6000; i++) csv += `${i * 10},${i % 400},${(i * 37) % 100},${['EU', 'US'][i % 2]}\n`;
await orders.push(csv);
await orders.flushAsync();

console.log(await live.readAsync());                          // per region, last 10 s
console.log((await live.readAsync({ series: true })).slice(0, 4)); // per second and region, for charts

await live.dispose();
await engine.close();

With series: true each row also has t, the slice’s start time. Windows keep totals, not rows, so countDistinct is refused there: use approxDistinct(x), which keeps a HyperLogLog sketch (4 KB, within about 1.6%) per slice and group. Sketches merge across slices and workers, so “distinct customers in the last 5 minutes” is a cheap read. DistinctSketch builds, merges and ships sketches yourself; see the DistinctSketch reference and distinct counts.

Binary frames: the Emitter

When you control both ends, skip text. An Emitter (Node) listens on a port and sends column batches as Vectro frames: numbers as raw float64, categories as codes with each label sent once per connection, every section 8-byte aligned. The receiving worker copies number columns straight into memory, with no parsing.

import { Emitter, SplitEngine } from 'vectrojs';

const columns = { ts: 'number', price: 'number', region: 'category' };
const feed = await Emitter.listen({ port: 0, columns });           // port 0: any free port

const engine = new SplitEngine({ workers: 2 });
const orders = await engine.streamAsync({ format: 'vectro', columns });
await Promise.all(engine.workers.map((_, w) => orders.connect(w, { port: feed.port })));
await feed.waitForConnections(2);

const n = 100_000;
await feed.write({
  ts: Float64Array.from({ length: n }, (_, i) => i),
  price: Float64Array.from({ length: n }, (_, i) => i % 100),
  region: { codes: Uint32Array.from({ length: n }, (_, i) => i % 3), labels: ['EU', 'US', 'APAC'] },
});
await feed.close();

// wait until the workers have parsed every row
let rows = 0;
for (let tries = 0; rows < n && tries < 200; tries++) {
  rows = (await orders.flushAsync()).rows;
  if (rows < n) await new Promise((r) => setTimeout(r, 10));
}
console.log(rows);
console.log((await orders.groupBy('region').aggAsync({ n: 'count()', avg: 'mean(price)' })).toRows());

await engine.close();

A category column can be plain strings, or { codes, labels } (codes into a labels array), which is faster. The emitter deals batches round-robin to the connections that aren’t backed up; write() waits when every connection has 8 MB unsent, and large batches go out in frames of at most 65,536 rows. format: 'arrow' on both ends sends an Arrow IPC stream instead, which pyarrow and Polars read too. Over WebSockets, send each frame as one binary message. BatchEncoder and FrameReader build and read frames without the Emitter; see the Emitter reference.

How fast

In a browser (Chrome, Web Workers each reading its own WebSocket of JSON lines): 4 workers kept up with the feed’s 2 million orders per second; 8 workers took in 3.5 to 3.6 million per second, where the test feed (one Node thread writing JSON) is the likely limit. The page never waited more than 4 ms, and the 30-second window read in 0.1 to 11 ms.

In Node (Apple M1 Pro, 7 workers, 10M rows kept, localhost TCP, a live query every 250 ms over every kept row):

Receiver	Rows/s taken in	Query median / max	Main thread stalled up to
Plain Node, 1 connection (split / Number parsing, a loop per query)	3.2–3.3M	19–20 / 23 ms	40–41 ms
Vectro, 1 connection, chunks pushed to workers	17.0–17.5M	13–14 / 22–32 ms	6–9 ms
Vectro, 1 connection per worker	16.7–18.5M	12–14 / 29–62 ms	12 ms

With JSON lines, Vectro took in 9.3 to 10.3M rows/s against 2.45M for plain Node calling JSON.parse per line. With an Emitter sending to 7 workers (Vectro frames or Arrow), 63 to 78M rows/s end to end, measured on a busy machine.

Limits

• Streams take flat records only (no nested JSON fields).
• Binary streams have number and category columns only.
• Windows group by at most one category column.
• No joins on streams yet, and no full sort: topAsync only.
• A SplitEngine worker that dies (often a browser stopping it for memory) fails every pending call; make a new engine.