SplitEngine and SplitTable

A SplitEngine starts workers that each run their own one-thread engine. A split table’s rows are dealt to the workers in blocks of 4,096, round-robin, so a filter on sorted data (the last month of orders) still matches on every worker. A query sends its expression text to every worker; each answers for its own rows, and the calling thread merges the small partial results: sums and counts are added, a mean is a total sum over a total count, groups are merged by key, top-k keeps the best k of all workers. It needs no SharedArrayBuffer, so in a browser it gives every core to pages without cross-origin isolation headers. The API is async only and matches the async API of Table, so the same code runs on either. Split tables are also the base of streams.

import { SplitEngine } from 'vectrojs';           // Node: worker threads
import { SplitEngine } from 'vectrojs/browser';   // browsers: Web Workers, any page

SplitEngine

new SplitEngine(options?)

Starts the workers. They start up in the background; every method waits for them.

Name	Type	Default	Meaning
workers	number	one per core	How many workers.
spawn	function	spawnSplitWorker of the entry point	Starts one worker and returns { post(msg, transfer), onMessage(fn), terminate() }, optionally with onError(fn).
engineOptions	object	{}	Engine options for each worker’s engine.

Properties: size (the number of workers), workers (their handles; workers.map(...) is a convenient way to connect one stream per worker), ready (a promise that settles when every worker has started) and failed (failed[i] is the error that stopped worker i, if any). When a worker dies, often because a browser stopped it for using too much memory, every call waiting on it fails, and the tables on that engine are incomplete: make a new SplitEngine.

spawnSplitWorker()

The default spawn of each entry point: a worker thread in Node, a module Web Worker in browsers.

engine.tableAsync(columns, options?)

Deals columns out to the workers and returns a promise of a SplitTable. Columns as for engine.table(): number arrays, string[], arrays of Date objects, or { codes, labels }. Text is encoded once on this thread, and only number arrays are transferred to the workers. Options: type (Float32Array halves memory) and dates.

engine.parseCSVAsync(text, options?)

Cuts CSV text or bytes at row ends (quote-aware) and has every worker parse its pieces. Options as for parseCSV; the column types are guessed once, on this thread. Returns a promise of a SplitTable.

engine.fromArrowAsync(bytes, options?)

Reads Arrow bytes on this thread with readArrow and deals the columns out. Options: columns, type.

engine.loadCSV(url, options?)

Browser only: fetches a CSV and runs parseCSVAsync on it.

engine.streamAsync(options)

Opens a live table fed by CSV lines, JSON lines, Vectro frames or Arrow, spread over the workers. Returns a promise of a StreamTable.

engine.local

A getter: a one-thread Engine on this thread, made on first use, that holds results such as group-by and top-k tables.

engine.close()

Terminates the workers and closes local. Returns a promise.

SplitTable

A split table isn’t constructed directly: it comes from tableAsync, parseCSVAsync, fromArrowAsync, filter, collectAsync and deriveAsync. The class isn’t exported.

st.length

A getter: the rows the table holds, ignoring filters.

st.columnNames

A getter: the column names.

st.filter(expr, params?)

A lazy view with the condition added; the workers apply it on every query. Synchronous, like table.filter.

st.aggAsync(spec, params?)

Totals over the matching rows, aggregations as in table.agg. countDistinct merges the workers’ distinct values exactly; approxDistinct merges HyperLogLog sketches (within about 1.6%). Also sumAsync, meanAsync, minAsync, maxAsync, countAsync(expr?).

st.groupBy(keys)

Returns an object with aggAsync(spec, params?), which returns a promise of a normal Table in engine.local: one row per group, sorted by key as on a normal table, with the key columns first and any countDistinct or approxDistinct columns before the other aggregations. There is no sync agg.

st.deriveAsync(name, expr, params?)

A new split table with one more column, computed by every worker for its own rows.

st.collectAsync()

The matching rows as a new split table; each worker keeps its own matches. Without filters, returns this table.

st.topAsync(k, by, options?)

The first k rows by by (as table.top), as a normal table in engine.local: every worker sends its own first k and they are merged. Option: columns, to return (and read) only these columns of a wide table. Rows that tie on every key come in worker order. Split tables have no full sort.

st.toRowsAsync(limit?)

The matching rows as plain objects, in table order.

st.toColumnsAsync()

The matching rows as plain columns, in table order, as table.toColumns.

st.toArrowAsync(options?)

The matching rows as Arrow bytes; { file: true } for the file format.

st.vectorSearchAsync(query, options)

Each worker scores its own rows and sends its best hits; used by VectorIndex when it is made on a split engine. Use index.searchAsync instead.

st.dispose()

Frees the table’s rows on every worker. A filtered view owns nothing; disposing it does nothing. Returns a promise.

Joins aren’t available on split tables yet.

import { SplitEngine } from 'vectrojs';

const engine = new SplitEngine({ workers: 2 });
const n = 50_000;
const orders = await engine.tableAsync({
  region: Array.from({ length: n }, (_, i) => ['EU', 'US', 'APAC'][i % 3]),
  customer: Float64Array.from({ length: n }, (_, i) => i % 997),
  price: Float64Array.from({ length: n }, (_, i) => (i % 400) + 0.5),
});

console.log(engine.size, orders.length);
console.log(await orders.filter("region == 'EU'").aggAsync({ n: 'count()', revenue: 'sum(price)', customers: 'countDistinct(customer)' }));

const byRegion = await orders.groupBy('region').aggAsync({ avg: 'mean(price)', approx: 'approxDistinct(customer)' });
console.log(byRegion.toRows());

const withTax = await orders.deriveAsync('gross', 'price * 1.2');
const top = await withTax.topAsync(3, '-gross', { columns: ['customer', 'gross'] });
console.log(top.toRows());

await withTax.dispose();
await orders.dispose();
await engine.close();

import { SplitEngine } from 'vectrojs';

const engine = new SplitEngine({ workers: 2 });
const csv = 'day,region,amount\n1,EU,10\n1,US,20\n2,EU,5\n3,US,15\n';
const t = await engine.parseCSVAsync(csv);
const busy = await t.filter('amount >= 10').collectAsync();
console.log(await busy.toRowsAsync(), await t.sumAsync('amount'));

const copy = await engine.fromArrowAsync(await t.toArrowAsync());
console.log(await copy.toColumnsAsync());
await engine.close();