Arrow: readArrow, writeArrow

Vectro reads the Arrow IPC stream format and the file format (.arrow, .feather, Feather v2), LZ4-compressed bodies included, and writes both. Arrow types map onto Vectro’s column kinds:

Arrow	Vectro
ints, floats (16, 32, 64-bit), bools	number (bools as 0 and 1); nulls as NaN
dates, times, timestamps, durations	number in milliseconds; dates and timestamps become date columns
strings, dictionary-encoded strings	category; nulls as ''
lists, structs, decimals, binary	refused, unless left out with columns

64-bit integers are exact up to 2^53. ZSTD-compressed bodies and big-endian data aren’t supported. See CSV, JSON lines, Arrow.

import { readArrow, writeArrow, ArrowReader, ArrowEncoder } from 'vectrojs';   // also 'vectrojs/browser', 'vectrojs/portable'

Tables

engine.fromArrow(bytes, options?)

Loads Arrow bytes as a Table. Options: columns (the names to read; default all) and type (Float64Array or Float32Array). A split engine has fromArrowAsync(bytes, options?), which returns a split table.

t.toArrow(options?)

The table’s rows (those matching its filter) as Arrow bytes: the stream format, or with { file: true } the file format, which pandas.read_feather and pyarrow.feather open. Numbers are written as float64 (float32 for Float32Array tables), categories as dictionary-encoded strings (pandas opens them as Categorical), dates as Timestamp(ms, UTC). Also toArrowAsync, and on split tables toArrowAsync only.

import { Engine } from 'vectrojs';

const engine = new Engine();
const sales = engine.table({
  region: ['EU', 'US', 'EU'],
  amount: [10.5, 20, 7.25],
  day: [new Date('2024-05-01'), new Date('2024-05-02'), new Date('2024-05-02')],
});

const feather = sales.filter("region == 'EU'").toArrow({ file: true });
console.log(feather.length, 'bytes');

const back = engine.fromArrow(feather);
console.log(back.columnNames, back.toRows());
console.log(engine.fromArrow(sales.toArrow(), { columns: ['amount'] }).sum('amount'));
await engine.close();

Functions

readArrow(bytes, options?)

Reads a whole Arrow stream or file. bytes is a Uint8Array, another typed array view or an ArrayBuffer. Option: columns, the names to read (an unknown name, or a column of a type Vectro can’t read, is an error). Returns { rows, columns, dates }: columns maps each name to a Float64Array (numbers) or { codes: Uint32Array, labels } (text), ready for engine.table(), and dates lists the date and timestamp columns (values in milliseconds). Throws if the bytes have no schema or end in the middle of a message.

writeArrow(columns, options?)

Writes columns as Arrow bytes. columns maps names to a Float64Array, Float32Array (written as float32), number[], string[] or { codes, labels } (both written as dictionary-encoded strings). Options: file (default false: the stream format; true: the file format) and dates (names of columns holding milliseconds, written as Timestamp(ms, UTC)). Returns a Uint8Array.

Incremental reading and writing

For Arrow that arrives or leaves in pieces: a socket, a WebSocket, Arrow Flight. Streams use these internally for format: 'arrow' (see StreamTable and Emitter).

new ArrowReader(handlers)

A reader for bytes that arrive in pieces cut anywhere. The file format works too (its footer is ignored). handlers:

Name	Called with
onSchema(kinds, fields)	kinds: { name: 'number' or 'category' or null } (null: a type Vectro can’t read). fields: [{ name, kind, type, date }]. May return the names to read; by default every readable column.
onBatch(rows, columns)	columns[k], in schema order: { values: Float64Array }, { codes: Uint32Array, labels } (every label of that column so far; codes stay valid as labels are added), or null for a column not read.
onEnd()	At the end-of-stream marker. Optional.

The arrays may point into the bytes you pushed: copy what you keep.

reader.push(chunk)

Feeds a Uint8Array. Handlers run for every complete message; the rest is kept for the next push. Bytes after the end-of-stream marker are ignored.

new ArrowEncoder(columns)

Builds Arrow IPC stream messages for one connection or file. columns maps names to 'number' (float64), 'float32', 'category' (utf8, dictionary-encoded with int32 indices) or 'date' (Timestamp(ms, UTC); NaN as null). Labels new since the last batch go out as a delta dictionary.

encoder.schema()

The schema message, as bytes. Send it first.

encoder.encode(batch)

One batch as bytes: any new dictionary entries, then the record batch. batch maps each column to a Float64Array, number[], string[] or { codes, labels }, all the same length.

encoder.messages(batch)

The same as encode, as separate messages: [{ bytes, metaLength, bodyLength, dictionary }], dictionaries first. Arrow Flight sends each as its own message.

encoder.end()

The end-of-stream marker, as bytes.

import { ArrowEncoder, ArrowReader, readArrow } from 'vectrojs';

const enc = new ArrowEncoder({ price: 'number', region: 'category' });
const pieces = [
  enc.schema(),
  enc.encode({ price: [10, 20], region: ['EU', 'US'] }),
  enc.encode({ price: [30], region: { codes: [0], labels: ['APAC'] } }),
  enc.end(),
];

const reader = new ArrowReader({
  onSchema: (kinds) => console.log('schema', kinds),
  onBatch: (rows, [price, region]) => console.log(rows, Array.from(price.values), Array.from(region.codes, (c) => region.labels[c])),
  onEnd: () => console.log('end'),
});
const all = new Uint8Array(pieces.reduce((n, p) => n + p.length, 0));
let at = 0;
for (const p of pieces) { all.set(p, at); at += p.length; }
for (let i = 0; i < all.length; i += 7) reader.push(all.subarray(i, i + 7)); // cut anywhere

console.log(readArrow(all).rows);