From b3734aa19442e20bc3b6af3525619f38a050adf9 Mon Sep 17 00:00:00 2001 From: Tam Nguyen Duc <1218621+tamnd@users.noreply.github.com> Date: Wed, 19 Aug 2026 15:27:56 +0700 Subject: [PATCH] Register frames, so a statement can match on columns a program already holds An Arrow table or an object of typed arrays goes in under a name, and a statement that names it reads the caller's arrays where they lie. The engine is told where each column is, how wide its values are and what they mean, so registering costs what describing the columns costs and not what the rows cost: a million rows in 26 microseconds against ten rows in 34, both of them mostly the promise. apache-arrow is a dev dependency and not a dependency. A table is recognized by its shape rather than by its class, so any library that speaks that shape takes the same path and a caller who never registers a frame installs nothing. Three things copy and all three are said rather than hidden. A string column is walked once, to check every offset at registration so that reading it afterwards cannot fail. A table that arrived as several record batches is concatenated, because a column of a frame is one run of bytes and two batches are two of them. A column given as a plain array is read into a buffer of this client's own, because an array holds JavaScript values rather than numbers and there is nothing in it to point at. The awkward case is a sliced column, and arrow-js splits the difference in a way worth writing down. Slicing narrows the values buffer and the offsets buffer, so those two already start at row zero of the chunk. It leaves the validity bitmap and a boolean column's bits whole, because both count in bits and a bit is not a place a typed array can start. So the chunk's offset means those two buffers and nothing else, and the test that reads a slice back is what caught it. All three calls are asynchronous, including registered(), because all three take the connection's lock and a call that waits on the event loop is the thing this client does not do. --- README.md | 28 +- bench/register.mjs | 158 ++++++ binding.d.cts | 117 ++++ etc/zudb.api.md | 41 ++ package-lock.json | 44 +- package.json | 2 + src/buffer.rs | 76 +++ src/conn.rs | 96 ++++ src/frame.rs | 1205 ++++++++++++++++++++++++++++++++++++++++ src/lib.rs | 2 + src/register.rs | 279 ++++++++++ test/register.test.mjs | 524 +++++++++++++++++ test/types/cjs.cts | 14 + test/types/esm.mts | 28 + types/header.d.ts | 70 +++ 15 files changed, 2679 insertions(+), 5 deletions(-) create mode 100644 bench/register.mjs create mode 100644 src/frame.rs create mode 100644 src/register.rs create mode 100644 test/register.test.mjs diff --git a/README.md b/README.md index 289bb0a..c157287 100644 --- a/README.md +++ b/README.md @@ -36,7 +36,7 @@ The rows are an array, so iterating them is `for (const row of rows)` and nothin ## What works today -`connect`, `query`, `exec`, `stream`, `close`, `dispose` and `await using`. Named parameters both ways, including lists, records and nesting. Every scalar the engine has, plus nodes, edges and paths with their tables named rather than numbered, and `ZuDate`, `ZuTime`, `ZuTimestamp` and `ZuDuration`, with `{ temporal: true }` and `toTemporal()` for the runtimes that have `Temporal`. Read-only connections, memory and thread limits. `bigIntMode`, per statement or per connection. An `AbortSignal` on any statement. The full error surface above, and `isZuError` to recognize it. Streaming, as an async iterable, as batches and as a Web Stream. Transactions, with `inTransaction` on the connection. An appender, for loading rows a batch at a time. Both module formats, typed separately. +`connect`, `query`, `exec`, `stream`, `close`, `dispose` and `await using`. Named parameters both ways, including lists, records and nesting. Every scalar the engine has, plus nodes, edges and paths with their tables named rather than numbered, and `ZuDate`, `ZuTime`, `ZuTimestamp` and `ZuDuration`, with `{ temporal: true }` and `toTemporal()` for the runtimes that have `Temporal`. Read-only connections, memory and thread limits. `bigIntMode`, per statement or per connection. An `AbortSignal` on any statement. The full error surface above, and `isZuError` to recognize it. Streaming, as an async iterable, as batches and as a Web Stream. Transactions, with `inTransaction` on the connection. An appender, for loading rows a batch at a time. Registered frames, so an Arrow table or an object of typed arrays is something a statement can match on without the rows being copied. Both module formats, typed separately. Build it with `npm run build`, and run the suite with `npm test`. Nothing is published yet, so `npm i zudb` is not a thing you can type at anybody's terminal, but everything it will do is built and installed on every run of the release workflow. @@ -124,6 +124,28 @@ Two more things are worth knowing before a load. A flush issued while one is sti A rel table has no property columns. A row of one is the two ends of an edge, as offsets into the tables it runs between, so `conn.appender("knows")` takes two columns and the flush checks that both rows are there before it writes anything. That check is here rather than the engine's, because the engine's comes after the write is durable. +## Matching on columns a program already has + +Columns a program is already holding become something a statement can match on, under a name the program picks. + +```ts +await conn.register("people", table); +const rows = await conn.query(`MATCH (p:people) WHERE p.age > 40 RETURN p.name AS name`); +await conn.unregister("people"); +``` + +An Arrow table goes in, which is what `apache-arrow` and everything built on it hands out, and so does an object of column name to typed array for a caller with none of that installed. `apache-arrow` is not a dependency of this package and is not imported by it: a table is recognized by its shape, so any library that speaks that shape takes the same path. + +Nothing is copied. What the engine is told is where each column is, how wide its values are and what they mean, and a statement that names the frame builds vectors pointing straight at the caller's arrays. So registering costs what describing the columns costs and not what the rows cost: on this machine a frame of a million rows registers in 26 microseconds and one of ten rows in 34, both of which are mostly the promise, since the round trip on its own is 16. + +The one column that is walked is a string column, and it is walked once. Every offset is checked at registration so that reading the frame afterwards cannot fail, which is 1.2 ms for a million strings. Two other things copy and both are said rather than hidden: a table that arrived as several record batches is concatenated into one, because a column of a frame is one run of bytes and two batches are two of them, and a column given as a plain array is read into a buffer of this client's own, because an array holds JavaScript values rather than numbers and there is nothing in it to point at. That last one is the expensive way in at 127 ns a row, and it is there so that a caller with an array is not stuck rather than because it is the way to do this. + +Because it is not a copy, a registered frame is a view and not a snapshot. Write into the typed array behind it and the next statement answers what is there now, which is the thing to know about the call and the reason it is worth having. Reading one is as fast as reading a table of the database and faster where the database has to decode: over a million rows here, summing an integer column takes 1.3 ms against a stored table's 1.6, and finding one row by a string takes 2.2 ms against 5.8. + +The frame belongs to the connection it was registered on and goes when that connection does. Nothing is written to the file, so another program opening the same database has never heard of it, and nothing writes to it either: a statement that inserts into or deletes from a registered name is refused with the reason, because that memory is the caller's array. `unregister(name)` takes the name away and hands the arrays back, which is not always that instant, since a statement still reading the frame holds it until it ends. `registered()` says what is registered here, and it is a method rather than a getter because it takes the connection's lock like everything else and nothing here blocks the event loop. + +Registering the same name again replaces what it stands for, columns and all. Registering over a table the database already holds is refused, since a statement naming it would mean the stored one. A frame with no rows is a table to match on and answers nothing, because a frame knows its columns without being told by a row. A null anywhere is refused by column and row, since a property that is null is one no row of this engine holds, and registering inside a transaction is refused because a frame is registered on the session, which is the thing the transaction is running on. + ## Asking for numbers instead of bigints `bigIntMode` says how INT64 is spelled on the way out. It goes on one statement, or on a connection for all of them, and a statement on a connection that named one may still name the other: @@ -216,7 +238,7 @@ typedoc rather than api-documenter, which would have been the obvious pick since Anything outside that table has no binary and no source build to fall back on, so the install resolves nothing and the first `require` says so. The browser and the platforms nobody builds for are what the WASM target answers, later. -`npm run bench` measures what this package adds to the engine, which is a row object and one JavaScript value per column: the same scan with the rows dropped is the floor, and the difference between the two is what the boundary costs. Run it against a release build, since a debug build of the engine moves the floor by an order of magnitude and not the rest of it. +`npm run bench` measures what this package adds to the engine, which is a row object and one JavaScript value per column: the same scan with the rows dropped is the floor, and the difference between the two is what the boundary costs. Run it against a release build, since a debug build of the engine moves the floor by an order of magnitude and not the rest of it. `npm run bench:append` does the same for the load path and `npm run bench:register` for registered frames, where what is being watched is that the registration does not scale with the rows. ## Still to come @@ -232,7 +254,7 @@ Bun and Deno in CI, and the WASM build for the browser. | Browser and edge | `zudb/wasm` | read-mostly, over OPFS or HTTP range requests | | Electron | the same binary | N-API is ABI-stable across Electron versions, so no per-Electron rebuild | -`apache-arrow` is an optional peer dependency behind the `zudb/arrow` entry point, so the base package stays small. +`apache-arrow` is a dev dependency and not a dependency, and it is one so that the tests can build the tables the register path reads. Nothing in the package imports it, so a caller who never registers a frame never installs it. One binary serves all three, because N-API is the ABI all three implement, and the whole suite runs on each of them in CI rather than the other two being assumed from Node passing. `npm run test:bun` and `npm run test:deno` run it locally. What the three do not agree on is what a native error carries: V8 writes a `stack` when the error is made and JavaScriptCore writes none at all through N-API, so this client writes the header line itself when it finds none, non-enumerably, and `err.stack` starts with the condition's name on all of them. diff --git a/bench/register.mjs b/bench/register.mjs new file mode 100644 index 0000000..94357af --- /dev/null +++ b/bench/register.mjs @@ -0,0 +1,158 @@ +// What registering a frame costs, and what reading one costs after. +// +// The claim the call makes is that nothing is copied, so the first block +// here is the one that has to hold: registering ten rows and registering +// ten million should cost the same, because what happens is that the +// engine is told where the columns are. A line in that block that scales +// with the rows is a line that copied them. +// +// Three cases do copy and all three are here rather than hidden. A string +// column is walked once, to check every offset at registration so that +// reading it afterwards cannot fail. A table that arrived as several +// batches is concatenated, because a column of a frame is one run of +// bytes and two batches are two of them. A plain JavaScript array is read +// into a buffer of this client's own, because an array holds values +// rather than numbers and there is nothing in it to point at. +// +// The second block is the reason to register at all: a statement reading +// a frame against the same statement reading a table of the database. +// +// Run it against a release build, for the reason bench/query.mjs gives. +// +// npm run build && npm run bench:register + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { Table, Utf8, tableFromArrays, vectorFromArray } from 'apache-arrow' +import { connect } from 'zudb' + +const ROWS = Number(process.env.ZU_BENCH_ROWS ?? 1_000_000) +const REPEATS = Number(process.env.ZU_BENCH_REPEATS ?? 5) +// The small frame, for the pair of lines that says registration does not +// scale with the rows. Ten is small enough that anything proportional to +// the rows disappears from it. +const FEW = 10 + +const dir = await mkdtemp(join(tmpdir(), 'zu-bench-register-')) + +const uid = BigInt64Array.from({ length: ROWS }, (_, ix) => BigInt(ix)) +const score = Float64Array.from({ length: ROWS }, (_, ix) => ix / 3) +const name = Array.from({ length: ROWS }, (_, ix) => `n${ix}`) +const plain = Array.from({ length: ROWS }, (_, ix) => ix) + +const wide = { uid, score } +const small = { uid: uid.subarray(0, FEW), score: score.subarray(0, FEW) } +const words = tableFromArrays({ name: vectorFromArray(name, new Utf8()) }) +const arrow = tableFromArrays({ uid, score }) +const halves = (() => { + const cut = ROWS >> 1 + const first = tableFromArrays({ uid: uid.slice(0, cut) }).batches[0] + const second = tableFromArrays({ uid: uid.slice(cut) }).batches[0] + return new Table([first, second]) +})() + +let counter = 0 + +/// A connection with nothing in it. +async function blank() { + return await connect(join(dir, `bench-${counter++}.zu1`)) +} + +/// The fastest of `REPEATS` runs, in milliseconds, after one warmup. +/// +/// The fastest for the reason bench/query.mjs gives: everything that +/// makes a run slower than the work itself is something that happened to +/// it rather than something about it. +async function time(conn, run) { + await run(conn) + let best = Infinity + for (let round = 0; round < REPEATS; round++) { + const started = performance.now() + await run(conn) + best = Math.min(best, performance.now() - started) + } + return best +} + +/// One registration, and what it costs is one promise round trip as well +/// as the work, because nothing here runs on the thread that called it. +/// The rounds after the first replace the name rather than taking it +/// away, which is the same description being built again and is what a +/// program rerunning the same cell does anyway. +function once(frame) { + return async (conn) => { + await conn.register('frame', frame) + } +} + +const registering = [ + // The floor, which is a call that takes the lock and answers a list of + // one name. Every line under it carries the same round trip, so this + // is what to subtract before believing any of them. + { name: 'the round trip alone', rows: 1, run: (conn) => conn.registered() }, + { name: `typed arrays, ${FEW} rows`, rows: FEW, run: once(small) }, + { name: 'typed arrays', rows: ROWS, run: once(wide) }, + { name: 'arrow table', rows: ROWS, run: once(arrow) }, + { name: 'arrow, two batches', rows: ROWS, run: once(halves) }, + { name: 'arrow strings', rows: ROWS, run: once(words) }, + { name: 'plain array', rows: ROWS, run: once({ n: plain }) }, +] + +const conn = await blank() +console.log(`registering ${ROWS} rows, fastest of ${REPEATS}`) +for (const { name, rows, run } of registering) { + const ms = await time(conn, run) + if (rows === 1) { + console.log(`${name.padEnd(24)} ${ms.toFixed(3).padStart(9)} ms`) + continue + } + const each = (ms * 1e6) / rows + const scale = rows === ROWS ? '' : ` (over ${rows})` + console.log( + `${name.padEnd(24)} ${ms.toFixed(3).padStart(9)} ms ${Math.round(each).toString().padStart(7)} ns/row${scale}`, + ) +} +conn.close() + +// The same rows twice, once as a frame the caller holds and once as a +// table the database holds, so that the two lines of each pair are the +// same statement over the same values. +const stored = await blank() +{ + await stored.exec("INSERT (p:person {uid: 0, name: 'n0'})") + const rows = await stored.appender('person') + for (let ix = 1; ix < ROWS; ix++) rows.appendRow([uid[ix], name[ix]]) + await rows.close() +} +await stored.register('frame', { + uid, + name, +}) + +const hunted = `n${ROWS - 1}` +const reading = [ + { + name: 'sum an integer column', + frame: 'MATCH (p:frame) RETURN sum(p.uid) AS total', + table: 'MATCH (p:person) RETURN sum(p.uid) AS total', + }, + { + name: 'find a row by string', + frame: `MATCH (p:frame) WHERE p.name = '${hunted}' RETURN p.uid AS uid`, + table: `MATCH (p:person) WHERE p.name = '${hunted}' RETURN p.uid AS uid`, + }, +] + +console.log(`\nreading ${ROWS} rows, fastest of ${REPEATS}`) +for (const { name, frame, table } of reading) { + const asFrame = await time(stored, (conn) => conn.query(frame)) + const asTable = await time(stored, (conn) => conn.query(table)) + console.log( + `${name.padEnd(24)} ${asFrame.toFixed(3).padStart(9)} ms frame ${asTable.toFixed(3).padStart(9)} ms table`, + ) +} + +stored.close() +await rm(dir, { recursive: true, force: true }) diff --git a/binding.d.cts b/binding.d.cts index aaa07ef..992f273 100644 --- a/binding.d.cts +++ b/binding.d.cts @@ -151,6 +151,76 @@ export type ZuAppendValue = | ZuDuration | ZuTemporalValue +/** + * One value of a registered frame's column, when the column is written + * as a plain array. + * + * The same values an appender takes, without the bytes: a column of + * BYTES is a column no statement can read back yet, so registering one + * would be naming data the caller cannot get at. There is no `null` + * either, for the reason there is none in a row of an appender. + */ +export type ZuFrameValue = + | boolean + | number + | bigint + | string + | ZuDate + | ZuTime + | ZuTimestamp + | ZuDuration + | ZuTemporalValue + +/** + * One column of a registered frame. + * + * A typed array is the shape that costs nothing: the engine reads it + * where it lies and no byte of it is copied. A plain array is read into + * buffers of this client's own, because an array holds values of the + * runtime rather than numbers, and its first value settles what the + * column holds. + */ +export type ZuFrameColumn = + | Int8Array + | Uint8Array + | Uint8ClampedArray + | Int16Array + | Uint16Array + | Int32Array + | Uint32Array + | Float32Array + | Float64Array + | BigInt64Array + | BigUint64Array + | readonly ZuFrameValue[] + +/** + * An Arrow table or record batch, described by its shape rather than by + * its class. + * + * Structural on purpose. `apache-arrow` is not a dependency of this + * client and should not have to be: recognizing a table by the two + * things every version of it has means a caller's copy of that library + * and this client's are never two copies of one package disagreeing + * about `instanceof`, and it means anything else that speaks the same + * shape works too. + */ +export interface ZuArrowTable { + readonly schema: { readonly fields: readonly { readonly name: string }[] } + getChildAt(index: number): unknown +} + +/** + * Columns the caller already holds, ready to be registered under a name. + * + * An Arrow table, or an object of column name to values. Both are read + * where they lie wherever there is one run of bytes to read: the two + * cases that copy are an Arrow column that arrived in several chunks, + * which is concatenated once, and a plain JavaScript array, which was + * never a column of numbers to begin with. + */ +export type ZuFrame = ZuArrowTable | Record + /** * A walk through the graph: nodes and edges, alternating, a node at * each end. @@ -531,6 +601,53 @@ export declare class Connection { * million rows later. */ appender(table: string): Promise + /** + * Registers columns the caller already holds as a table called + * `name`, and answers how many rows it has. + * + * The zero-copy way in. Nothing is read into the database: the + * engine is told where the caller's buffers are, and a statement + * that matches the name scans them where they lie, so registering + * ten million rows costs a description of their columns rather than + * ten million writes. + * + * ```js + * await conn.register('people', arrow.tableFromArrays({ id, name })) + * const rows = await conn.query('MATCH (p:people) RETURN p.name AS name') + * ``` + * + * An Arrow table or record batch, which is what `apache-arrow` and + * everything built on it hands out, or an object of column name to + * values. The values of that object are typed arrays where the + * caller has them, which is the zero-copy shape, and plain arrays + * where they do not, which is read into buffers of this client's + * own because an array holds values of the runtime rather than + * numbers. + * + * A frame is a view and not a snapshot: write into the array behind + * it and the next statement answers what is there now. It belongs + * to this connection, is never written to the database, and no + * other program opening the same file sees it. Nothing writes to + * one either, so a statement that inserts into a registered name is + * refused with the reason. + */ + register(name: string, data: ZuFrame): Promise + /** + * Takes a registered frame's name away and gives the bytes back. + * + * The bytes go when the last statement reading them lets go, which + * is usually now and is never before: a frame a running statement + * is still scanning is held until it ends. + */ + unregister(name: string): Promise + /** + * The names frames are registered under on this connection, sorted. + * + * A method rather than a getter, and asynchronous like everything + * else here, because reading them takes the connection's lock and + * nothing on this class waits on the event loop. + */ + registered(): Promise /** * Runs one statement and gives back its rows. * diff --git a/etc/zudb.api.md b/etc/zudb.api.md index 0eb63bd..dfda7c1 100644 --- a/etc/zudb.api.md +++ b/etc/zudb.api.md @@ -36,7 +36,10 @@ export class Connection { get path(): string query>(statement: string, params?: Record | null, options?: ZuStatementOptions | null): Promise> get readOnly(): boolean + register(name: string, data: ZuFrame): Promise + registered(): Promise transaction(options?: ZuTransactionOptions | null): Promise + unregister(name: string): Promise } // @public @@ -76,6 +79,14 @@ export type ZuAppendValue = | ZuDuration | ZuTemporalValue +// @public +export interface ZuArrowTable { + // (undocumented) + getChildAt(index: number): unknown + // (undocumented) + readonly schema: { readonly fields: readonly { readonly name: string }[] } +} + // @public export interface ZuBatch> extends Array { // (undocumented) @@ -130,6 +141,36 @@ export interface ZuError extends Error { readonly severity?: 'success' | 'noData' | 'warning' | 'informational' | 'exception' } +// @public +export type ZuFrame = ZuArrowTable | Record + +// @public +export type ZuFrameColumn = +| Int8Array +| Uint8Array +| Uint8ClampedArray +| Int16Array +| Uint16Array +| Int32Array +| Uint32Array +| Float32Array +| Float64Array +| BigInt64Array +| BigUint64Array +| readonly ZuFrameValue[] + +// @public +export type ZuFrameValue = +| boolean +| number +| bigint +| string +| ZuDate +| ZuTime +| ZuTimestamp +| ZuDuration +| ZuTemporalValue + // @public export class ZuNode { // (undocumented) diff --git a/package-lock.json b/package-lock.json index de799ee..1fdbbc8 100644 --- a/package-lock.json +++ b/package-lock.json @@ -13,6 +13,7 @@ "@microsoft/api-extractor": "^7.58.12", "@napi-rs/cli": "^3.8.6", "@types/node": "^26.2.0", + "apache-arrow": "^21.2.0", "typedoc": "^0.28.20", "typescript": "^5.9.3" }, @@ -2227,6 +2228,39 @@ "dev": true, "license": "MIT" }, + "node_modules/apache-arrow": { + "version": "21.2.0", + "resolved": "https://registry.npmjs.org/apache-arrow/-/apache-arrow-21.2.0.tgz", + "integrity": "sha512-Hxe6Agq26gQOM954qpzYSllJBPJl+e16U5CkfuMUhLrNba+5nKkttIVlflaovN6oaTratqMGAO8H5u/aNhmHWQ==", + "dev": true, + "license": "Apache-2.0", + "dependencies": { + "@types/node": "^25.2.0", + "flatbuffers": "^25.1.24", + "json-with-bigint": "^3.5.3", + "tslib": "^2.6.2" + }, + "bin": { + "arrow2csv": "bin/arrow2csv.js" + } + }, + "node_modules/apache-arrow/node_modules/@types/node": { + "version": "25.9.5", + "resolved": "https://registry.npmjs.org/@types/node/-/node-25.9.5.tgz", + "integrity": "sha512-OScDchr2fwuUmWdf4kZ9h7PcJiYDVInhJizG/biAq3cAvqwYktuy/TYGGdZNMtNTFUP7rnb0NU4TUdm82kt4Rg==", + "dev": true, + "license": "MIT", + "dependencies": { + "undici-types": ">=7.24.0 <7.24.7" + } + }, + "node_modules/apache-arrow/node_modules/undici-types": { + "version": "7.24.6", + "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-7.24.6.tgz", + "integrity": "sha512-WRNW+sJgj5OBN4/0JpHFqtqzhpbnV0GuB+OozA9gCL7a993SmU+1JBZCzLNxYsbMfIeDL+lTsphD5jN5N+n0zg==", + "dev": true, + "license": "MIT" + }, "node_modules/argparse": { "version": "2.0.1", "resolved": "https://registry.npmjs.org/argparse/-/argparse-2.0.1.tgz", @@ -2591,6 +2625,13 @@ "dev": true, "license": "MIT" }, + "node_modules/flatbuffers": { + "version": "25.9.23", + "resolved": "https://registry.npmjs.org/flatbuffers/-/flatbuffers-25.9.23.tgz", + "integrity": "sha512-MI1qs7Lo4Syw0EOzUl0xjs2lsoeqFku44KpngfIduHBYvzm8h2+7K8YMQh1JtVVVrUvhLpNwqVi4DERegUJhPQ==", + "dev": true, + "license": "Apache-2.0" + }, "node_modules/fs-extra": { "version": "11.3.6", "resolved": "https://registry.npmjs.org/fs-extra/-/fs-extra-11.3.6.tgz", @@ -3246,8 +3287,7 @@ "resolved": "https://registry.npmjs.org/tslib/-/tslib-2.8.1.tgz", "integrity": "sha512-oJFu94HQb+KVduSUQL7wnpmqnfmLsOA/nAh6b6EH0wCEoK0/mPeXU6c3wKDV83MkOuHPRHtSXKKU99IBazS/2w==", "dev": true, - "license": "0BSD", - "optional": true + "license": "0BSD" }, "node_modules/typanion": { "version": "3.14.0", diff --git a/package.json b/package.json index abc69f1..3fbb23b 100644 --- a/package.json +++ b/package.json @@ -72,6 +72,7 @@ "reference": "node tools/reference.mjs reference", "bench": "node bench/query.mjs", "bench:append": "node bench/append.mjs", + "bench:register": "node bench/register.mjs", "bench:temporal": "node --harmony-temporal bench/query.mjs" }, "devDependencies": { @@ -79,6 +80,7 @@ "@microsoft/api-extractor": "^7.58.12", "@napi-rs/cli": "^3.8.6", "@types/node": "^26.2.0", + "apache-arrow": "^21.2.0", "typedoc": "^0.28.20", "typescript": "^5.9.3" } diff --git a/src/buffer.rs b/src/buffer.rs index 04eb0da..4e69e94 100644 --- a/src/buffer.rs +++ b/src/buffer.rs @@ -114,6 +114,82 @@ impl Column { }) } + /// The buffer a column starts as, going by its first value, or + /// `None` for a value no column holds. + /// + /// The other way round from [`Column::for_type`]: there is no table + /// saying what the column is, so the first value says it. The order + /// of the arms is the parameter binder's, and the one rule worth + /// knowing is the same one: a whole `number` starts a column of + /// whole numbers, because `[1, 2, 3]` is what a caller writes, and a + /// fractional one starts a column of floats. + pub fn start(env: &Env, value: Unknown<'_>) -> std::result::Result, Mismatch> { + let mut column = match value.get_type()? { + ValueType::Boolean => Column::Bool(Vec::new()), + ValueType::String => Column::Str(Vec::new()), + ValueType::BigInt => Column::Int(Vec::new()), + ValueType::Number => match f64::from_unknown(value)?.fract() == 0.0 { + true => Column::Int(Vec::new()), + false => Column::Float(Vec::new()), + }, + ValueType::Object => match moment(env, &value)? { + Some(Temporal::Date(_)) => Column::Date(Vec::new()), + Some(Temporal::LocalTime(_)) => Column::LocalTime(Vec::new()), + Some(Temporal::LocalDatetime(_)) => Column::LocalDatetime(Vec::new()), + Some(Temporal::Duration(kind, _)) => Column::Duration(kind, Vec::new()), + // A zoned time or datetime is a type this engine has + // nowhere to keep, and the push below is what words that, + // so the column starts as the local one it is closest to + // and refuses its own first value. + Some(Temporal::ZonedTime { .. }) => Column::LocalTime(Vec::new()), + Some(Temporal::ZonedDatetime { .. }) => Column::LocalDatetime(Vec::new()), + None => match value.is_typedarray()? { + true => Column::Bytes(Vec::new()), + false => return Ok(None), + }, + }, + _ => return Ok(None), + }; + column.push(env, value)?; + Ok(Some(column)) + } + + /// Takes one more value, widening the column if the value asks it to. + /// + /// The one widening there is: a column of whole numbers that meets a + /// fractional one becomes a column of floats. `[1, 2, 2.5]` is a + /// column of numbers however it was written, and settling that on the + /// first value alone would refuse it at the third. + pub fn widening_push( + &mut self, + env: &Env, + value: Unknown<'_>, + ) -> std::result::Result<(), Mismatch> { + if let Column::Int(whole) = self + && value.get_type()? == ValueType::Number + { + let n = f64::from_unknown(value)?; + if n.fract() != 0.0 { + *self = Column::Float(whole.iter().map(|&n| n as f64).collect()); + } + } + self.push(env, value) + } + + /// How many values have gone in. + pub fn len(&self) -> usize { + match self { + Column::Int(v) => v.len(), + Column::Float(v) => v.len(), + Column::Bool(v) => v.len(), + Column::Str(v) => v.len(), + Column::Bytes(v) => v.len(), + Column::Date(v) => v.len(), + Column::LocalTime(v) | Column::LocalDatetime(v) => v.len(), + Column::Duration(_, v) => v.len(), + } + } + /// Takes one more value, or says why this column would not have it. pub fn push(&mut self, env: &Env, value: Unknown<'_>) -> std::result::Result<(), Mismatch> { let holds = self.holds(); diff --git a/src/conn.rs b/src/conn.rs index e77888c..84efea5 100644 --- a/src/conn.rs +++ b/src/conn.rs @@ -35,6 +35,7 @@ use zudb::{Config, Database, DiagnosticRecord, Interrupt, ZuError}; use crate::append::OpenTask; use crate::cancel::Watch; use crate::error::{aborted, raise, usage}; +use crate::register::{self, RegisterTask, RegisteredTask, UnregisterTask}; use crate::stream::{self, Started, ZuCursor}; use crate::temporal; use crate::txn::StartTask; @@ -402,6 +403,101 @@ impl Connection { )) } + /// Registers columns the caller already holds as a table called + /// `name`, and answers how many rows it has. + /// + /// The zero-copy way in. Nothing is read into the database: the + /// engine is told where the caller's buffers are, and a statement + /// that matches the name scans them where they lie, so registering + /// ten million rows costs a description of their columns rather than + /// ten million writes. + /// + /// ```js + /// await conn.register('people', arrow.tableFromArrays({ id, name })) + /// const rows = await conn.query('MATCH (p:people) RETURN p.name AS name') + /// ``` + /// + /// An Arrow table or record batch, which is what `apache-arrow` and + /// everything built on it hands out, or an object of column name to + /// values. The values of that object are typed arrays where the + /// caller has them, which is the zero-copy shape, and plain arrays + /// where they do not, which is read into buffers of this client's + /// own because an array holds values of the runtime rather than + /// numbers. + /// + /// A frame is a view and not a snapshot: write into the array behind + /// it and the next statement answers what is there now. It belongs + /// to this connection, is never written to the database, and no + /// other program opening the same file sees it. Nothing writes to + /// one either, so a statement that inserts into a registered name is + /// refused with the reason. + #[napi( + ts_args_type = "name: string, data: ZuFrame", + ts_return_type = "Promise" + )] + pub fn register( + &self, + env: &Env, + name: Unknown<'_>, + data: Unknown<'_>, + ) -> AsyncTask { + // The frame is read here, on the thread that owns the runtime, + // because reading a JavaScript value is something no other + // thread may do. What travels is the description and the + // references keeping the buffers alive. + let read = match self.alive.load(Ordering::Acquire) { + true => text(&name, "name") + .and_then(|name| register::read(env, &name, data).map(|frame| (name, frame))), + false => Err(register::closed()), + }; + let (name, described, refused) = match read { + Ok((name, described)) => (name, Some(described), None), + Err(message) => (String::new(), None, Some(message)), + }; + AsyncTask::new(RegisterTask::new(self.frames(), name, described, refused)) + } + + /// Takes a registered frame's name away and gives the bytes back. + /// + /// The bytes go when the last statement reading them lets go, which + /// is usually now and is never before: a frame a running statement + /// is still scanning is held until it ends. + #[napi(ts_args_type = "name: string", ts_return_type = "Promise")] + pub fn unregister(&self, name: Unknown<'_>) -> AsyncTask { + let named = match self.alive.load(Ordering::Acquire) { + true => text(&name, "name"), + false => Err(register::closed()), + }; + AsyncTask::new(UnregisterTask::new( + self.frames(), + named.as_deref().unwrap_or_default().to_string(), + named.err(), + )) + } + + /// The names frames are registered under on this connection, sorted. + /// + /// A method rather than a getter, and asynchronous like everything + /// else here, because reading them takes the connection's lock and + /// nothing on this class waits on the event loop. + #[napi(ts_return_type = "Promise")] + pub fn registered(&self) -> AsyncTask { + let refused = match self.alive.load(Ordering::Acquire) { + true => None, + false => Some(register::closed()), + }; + AsyncTask::new(RegisteredTask::new(self.frames(), refused)) + } + + /// The three handles a frame call runs against. + fn frames(&self) -> register::Held { + register::Held::new( + Arc::clone(&self.inner), + Arc::clone(&self.alive), + Arc::clone(&self.in_txn), + ) + } + /// The task that starts one, whether or not it is going to work. fn start(&self, read_only: bool, refused: Option) -> StartTask { StartTask::new( diff --git a/src/frame.rs b/src/frame.rs new file mode 100644 index 0000000..2fe4a1b --- /dev/null +++ b/src/frame.rs @@ -0,0 +1,1205 @@ +//! A frame of columns, described where the caller keeps them. +//! +//! The way in for data that is already in columns. What a JavaScript +//! program holds columns in is a typed array, and what an +//! `apache-arrow` table holds them in is typed arrays too: eight-byte +//! words back to back, one bit a row for a boolean, characters end to +//! end with offsets cutting them up. That is how this engine lays a +//! column out as well, so what this module produces is not a copy of any +//! of it but a description: where each column is, how wide its values +//! are, and what they mean. +//! +//! A typed array is not a Rust buffer, and the difference is who may +//! collect it. So every array a description points at is held here as a +//! reference the runtime counts, which is what [`Held`] is, and the +//! engine drops that when the last table naming those bytes goes. The +//! drop may happen on a thread that is not the runtime's, which is the +//! one thing about this that needs saying: napi-rs routes the release +//! back to the thread that owns the array rather than touching V8 from +//! wherever the frame died. +//! +//! Two things copy, and both are said rather than hidden. A column an +//! Arrow table holds in more than one chunk is concatenated, because a +//! column of a table is one run of bytes and three chunks are three of +//! them; that is a memcpy per column and it happens once. A plain array +//! of JavaScript values is read into buffers of this module's own, +//! because an array holds values of the runtime and a column holds +//! numbers, so there is nothing there to point at. +//! +//! There is no null anywhere in it. A property that is null is one no +//! row of this engine can hold, so a column with a gap in it can only +//! ever be refused, and refusing it by name and row number is the +//! difference between a caller who knows which cell to fix and one who +//! knows only that something somewhere was empty. + +use std::ptr::NonNull; +use std::sync::Arc; + +use napi::bindgen_prelude::*; +use napi::{Env, ValueType}; +use zu_common::{DurationKind, FloatBits, IntBits, LogicalType}; +use zudb::{Column as Described_, Layout}; + +use crate::buffer::{self, Mismatch, named}; + +/// What a registered frame's bytes are, and the thing whose life is +/// their life. +/// +/// Both halves are filled as the columns need them. The lent arrays are +/// the caller's own, held so the runtime cannot collect them while a +/// statement is reading them; the owned buffers are what this client +/// built out of JavaScript values, for a caller with no frame library +/// and for the columns an Arrow table did not hand over whole. +pub struct Held { + lent: Vec, + owned: Vec, +} + +impl Held { + fn new() -> Held { + Held { + lent: Vec::new(), + owned: Vec::new(), + } + } + + /// Keeps one of the caller's arrays and says where its bytes are. + fn lend(&mut self, array: Lent) -> NonNull { + let ptr = array.ptr(); + self.lent.push(array); + ptr + } + + /// Keeps one buffer of this module's own and says where it is. + /// + /// The pointer survives everything after it, because what moves when + /// the vector of them grows is the enum and not the allocation it + /// names. + fn own(&mut self, bytes: Bytes) -> NonNull { + let ptr = bytes.ptr(); + self.owned.push(bytes); + ptr + } +} + +/// One of the caller's typed arrays, held by reference. +/// +/// Ten variants rather than one, because napi hands a typed array back +/// as the Rust type that matches its elements and each of those is a +/// different type that owns a different reference. What this module +/// wants from all of them is the same three things, which is what the +/// methods under it answer. +enum Lent { + I8(Int8Array), + U8(Uint8Array), + I16(Int16Array), + U16(Uint16Array), + I32(Int32Array), + U32(Uint32Array), + I64(BigInt64Array), + U64(BigUint64Array), + F32(Float32Array), + F64(Float64Array), +} + +impl Lent { + /// The array a value is, or `None` for a value that is not one. + /// + /// The kind is read first, off a borrowed view that takes no + /// reference, so the array is claimed once and by the arm that + /// wanted it rather than tried against each of ten in turn. + fn of(value: Unknown<'_>) -> std::result::Result, Error> { + if value.get_type()? != ValueType::Object || !value.is_typedarray()? { + return Ok(None); + } + let kind = TypedArray::from_unknown(value)?.typed_array_type; + Ok(Some(match kind { + TypedArrayType::Int8 => Lent::I8(Int8Array::from_unknown(value)?), + // A clamped array holds bytes like any other and clamping is + // a rule about writing into one, which nothing here does. + TypedArrayType::Uint8 | TypedArrayType::Uint8Clamped => { + Lent::U8(Uint8Array::from_unknown(value)?) + } + TypedArrayType::Int16 => Lent::I16(Int16Array::from_unknown(value)?), + TypedArrayType::Uint16 => Lent::U16(Uint16Array::from_unknown(value)?), + TypedArrayType::Int32 => Lent::I32(Int32Array::from_unknown(value)?), + TypedArrayType::Uint32 => Lent::U32(Uint32Array::from_unknown(value)?), + TypedArrayType::BigInt64 => Lent::I64(BigInt64Array::from_unknown(value)?), + TypedArrayType::BigUint64 => Lent::U64(BigUint64Array::from_unknown(value)?), + TypedArrayType::Float32 => Lent::F32(Float32Array::from_unknown(value)?), + TypedArrayType::Float64 => Lent::F64(Float64Array::from_unknown(value)?), + _ => return Ok(None), + })) + } + + /// Every byte of it, whatever its elements are. + fn raw(&self) -> &[u8] { + match self { + Lent::I8(v) => flat(v.as_ref()), + Lent::U8(v) => v.as_ref(), + Lent::I16(v) => flat(v.as_ref()), + Lent::U16(v) => flat(v.as_ref()), + Lent::I32(v) => flat(v.as_ref()), + Lent::U32(v) => flat(v.as_ref()), + Lent::I64(v) => flat(v.as_ref()), + Lent::U64(v) => flat(v.as_ref()), + Lent::F32(v) => flat(v.as_ref()), + Lent::F64(v) => flat(v.as_ref()), + } + } + + /// Where its first element is. + /// + /// Never null: an array of no elements lends out the address an + /// empty slice has, which is aligned and is not zero, and no row is + /// ever read through it anyway. + fn ptr(&self) -> NonNull { + NonNull::new(self.raw().as_ptr() as *mut u8).expect("an empty slice is not at address zero") + } + + /// What one element of it is, in the terms a layout is written in. + fn elem(&self) -> Elem { + let int = |bits, signed| Elem::Int { + bits, + signed, + scale: 1, + }; + match self { + Lent::I8(_) => int(IntBits::B8, true), + Lent::U8(_) => int(IntBits::B8, false), + Lent::I16(_) => int(IntBits::B16, true), + Lent::U16(_) => int(IntBits::B16, false), + Lent::I32(_) => int(IntBits::B32, true), + Lent::U32(_) => int(IntBits::B32, false), + Lent::I64(_) => int(IntBits::B64, true), + Lent::U64(_) => int(IntBits::B64, false), + Lent::F32(_) => Elem::Float(FloatBits::B32), + Lent::F64(_) => Elem::Float(FloatBits::B64), + } + } + + /// How many elements it holds. + fn len(&self) -> usize { + self.raw().len() / self.elem().width() + } + + /// One offset of a string column's offset array, which is 32 bits + /// wide in Arrow's `Utf8` and 64 in its `LargeUtf8`. + fn offset(&self, at: usize) -> Option { + match self { + Lent::I32(v) => v.as_ref().get(at).map(|&n| i64::from(n)), + Lent::I64(v) => v.as_ref().get(at).copied(), + _ => None, + } + } +} + +/// A slice of values as the bytes under them. +/// +/// Whole elements either way, so the length is exact and the alignment +/// only ever loosens. +fn flat(values: &[T]) -> &[u8] { + unsafe { + std::slice::from_raw_parts(values.as_ptr().cast::(), std::mem::size_of_val(values)) + } +} + +/// One buffer this client built, because a JavaScript value is not a +/// column and something has to hold the bytes. +/// +/// The variants are widths rather than types: what a run of eight-byte +/// words means is said by the logical type recorded beside it, and a +/// count of nanoseconds and a number are the same eight bytes. Each is a +/// vector of its own width so that the allocation is aligned for the +/// values that will be read out of it. +enum Bytes { + Eight(Vec), + Four(Vec), + Two(Vec), + One(Vec), + /// The `rows + 1` offsets that cut a string column up, always 32 + /// bits wide: a buffer this module builds is one it also sized, and + /// it refuses to build one past what a 32-bit offset reaches. + Offsets(Vec), +} + +impl Bytes { + /// An empty buffer of the width `elem` reads through. + fn empty(elem: &Elem) -> Bytes { + match elem.width() { + 8 => Bytes::Eight(Vec::new()), + 4 => Bytes::Four(Vec::new()), + 2 => Bytes::Two(Vec::new()), + _ => Bytes::One(Vec::new()), + } + } + + /// A buffer of the width `elem` reads through, with room for `rows`. + fn with_room(elem: &Elem, rows: usize) -> Bytes { + match elem.width() { + 8 => Bytes::Eight(Vec::with_capacity(rows)), + 4 => Bytes::Four(Vec::with_capacity(rows)), + 2 => Bytes::Two(Vec::with_capacity(rows)), + _ => Bytes::One(Vec::with_capacity(rows)), + } + } + + /// Copies whole elements onto the end of it. + /// + /// The bytes go in as they lie, because what is being copied is a + /// column of this machine's own memory into another buffer of it: + /// what the words mean is the layout's business and does not change + /// on the way. + fn extend(&mut self, raw: &[u8]) { + match self { + Bytes::Eight(v) => v.extend( + raw.chunks_exact(8) + .map(|word| u64::from_ne_bytes(word.try_into().expect("eight bytes"))), + ), + Bytes::Four(v) => v.extend( + raw.chunks_exact(4) + .map(|word| u32::from_ne_bytes(word.try_into().expect("four bytes"))), + ), + Bytes::Two(v) => v.extend( + raw.chunks_exact(2) + .map(|word| u16::from_ne_bytes(word.try_into().expect("two bytes"))), + ), + Bytes::One(v) => v.extend_from_slice(raw), + Bytes::Offsets(_) => {} + } + } + + /// Where it starts. + /// + /// A vector that has never allocated lends out an aligned address + /// rather than a real one, which is never zero and is never read, + /// since a buffer with nothing in it belongs to a column with no + /// rows. + fn ptr(&self) -> NonNull { + let ptr = match self { + Bytes::Eight(v) => v.as_ptr().cast::(), + Bytes::Four(v) => v.as_ptr().cast::(), + Bytes::Two(v) => v.as_ptr().cast::(), + Bytes::One(v) => v.as_ptr(), + Bytes::Offsets(v) => v.as_ptr().cast::(), + }; + NonNull::new(ptr as *mut u8).expect("a buffer of this process is never at address zero") + } +} + +/// What one value of a column is, in the terms the engine reads it in. +/// +/// This is the layout and not the meaning: a date and a count of days +/// are the same `Int`, and the logical type beside it is what tells them +/// apart. +#[derive(Clone, Copy)] +enum Elem { + Int { + bits: IntBits, + signed: bool, + /// What one value is multiplied by to reach the unit its meaning + /// counts in, which is where Arrow's microseconds meet this + /// engine's nanoseconds. Nothing is converted here: the + /// multiplication happens per scanned chunk, on the rows a + /// statement actually reads. + scale: i64, + }, + Float(FloatBits), + /// One bit a row, low bit of the first byte first. + Bool, + /// Characters end to end with offsets cutting them up. + Str { + /// Whether the offsets are 64 bits wide, which is Arrow's + /// `LargeUtf8` against its `Utf8`. + wide: bool, + }, +} + +impl Elem { + /// How many bytes one value takes, which for a boolean is the byte + /// eight of them share. + fn width(&self) -> usize { + match self { + Elem::Int { bits, .. } => bits.bits() as usize / 8, + Elem::Float(bits) => bits.bits() as usize / 8, + Elem::Bool => 1, + Elem::Str { wide } => match wide { + true => 8, + false => 4, + }, + } + } +} + +/// A frame as the engine is about to be told about it. +/// +/// The columns hold raw pointers into what `held` keeps alive, which is +/// why the two travel together and why neither is any use without the +/// other. +pub struct Described { + pub columns: Vec, + pub rows: u64, + held: Arc, +} + +// A pointer is not `Send`, because Rust cannot know what it addresses. +// These address buffers the `Arc` in the same struct keeps alive, and +// that `Arc` is `Send` and `Sync`, so a description travels wherever the +// thing it describes does. Nothing writes through them. +unsafe impl Send for Described {} + +impl Described { + /// Registers this as a table named `name` on `engine`. + /// + /// Every pointer in it addresses a buffer the `Arc` handed over with + /// them keeps alive, which is what building one of these promises + /// and what nothing between there and here undoes, so the `unsafe` + /// that [`zudb::Frame::new`] asks for is discharged where the buffers + /// were described rather than here. + pub fn register(self, engine: &mut zudb::Connection, name: &str) -> zudb::Result<()> { + let Described { + columns, + rows, + held, + } = self; + let frame = unsafe { zudb::Frame::new(name, rows, columns, held) }?; + engine.register(frame) + } +} + +/// Reads whatever the caller handed over into a description of it. +/// +/// Two shapes, and both of them are shapes a JavaScript program already +/// has: an Arrow table, which is what every columnar library in this +/// ecosystem hands out, and an object of column name to values, whose +/// values are typed arrays where the caller has them and plain arrays +/// where they do not. Anything else is refused here rather than iterated +/// hopefully, because the message a caller wants is the list of what +/// would have worked. +/// +/// Every failure comes back as the sentence to refuse the call with, +/// including a boundary failure, because a caller who cannot be given +/// the value they passed is being told the same thing either way. +pub fn read(env: &Env, data: Unknown<'_>) -> std::result::Result { + inner(env, data).map_err(|err| err.reason) +} + +fn inner(env: &Env, data: Unknown<'_>) -> Result { + if data.get_type()? != ValueType::Object { + return Err(refuse(env, &data)); + } + let object = Object::from_unknown(data)?; + if arrowish(&object)? { + return from_arrow(env, &object); + } + if object.is_array()? || data.is_typedarray()? { + return Err(refuse(env, &data)); + } + from_object(env, &object) +} + +fn refuse(env: &Env, data: &Unknown<'_>) -> Error { + crate::error::usage( + env, + format!( + "a frame is an Arrow table, which is what `apache-arrow` and everything built on it \ + hands out, or an object of column name to values, and this is {}", + named(data) + ), + ) +} + +/// Whether this looks like an Arrow table or record batch. +/// +/// Asked of the object rather than of a package this client would have +/// to depend on. What is being recognized is the shape every Arrow table +/// in this ecosystem has, a schema listing its fields and a call that +/// hands a column back by position, and recognizing it that way means a +/// caller's `apache-arrow` and this client's are never two copies of one +/// library disagreeing about `instanceof`. +fn arrowish(object: &Object<'_>) -> Result { + let schema: Unknown<'_> = object.get_named_property("schema")?; + if schema.get_type()? != ValueType::Object { + return Ok(false); + } + let fields: Unknown<'_> = Object::from_unknown(schema)?.get_named_property("fields")?; + if fields.get_type()? != ValueType::Object || !fields.is_array()? { + return Ok(false); + } + let child: Unknown<'_> = object.get_named_property("getChildAt")?; + Ok(child.get_type()? == ValueType::Function) +} + +/// Reads an Arrow table where it lies. +/// +/// A column is read off the vector the table hands back rather than off +/// the table's own insides, because a vector is the part of that library +/// that has stayed the same: the chunks it is made of, the buffers under +/// each chunk, and the type they are read through. +fn from_arrow(env: &Env, table: &Object<'_>) -> Result { + let schema: Object<'_> = table.get_named_property("schema")?; + let fields: Object<'_> = schema.get_named_property("fields")?; + let width = fields.get_array_length()?; + let child: Function<'_, u32, Unknown<'_>> = table.get_named_property("getChildAt")?; + + let mut held = Held::new(); + let mut columns = Vec::with_capacity(width as usize); + let mut rows: Option = None; + for at in 0..width { + let field: Object<'_> = fields.get_element(at)?; + let name: String = field.get_named_property("name")?; + let vector: Unknown<'_> = child.apply(*table, at)?; + if vector.get_type()? != ValueType::Object { + return Err(crate::error::usage( + env, + format!("this table names a column '{name}' and then has no column there"), + )); + } + let vector = Object::from_unknown(vector)?; + let (ty, elem) = meaning(env, &name, &vector)?; + let (rows_here, layout) = column(env, &name, elem, &vector, &mut held)?; + match rows { + Some(had) if had != rows_here => { + return Err(crate::error::usage( + env, + format!( + "column '{name}' holds {rows_here} values and the column before it holds \ + {had}, and a table is as wide as it is long" + ), + )); + } + _ => rows = Some(rows_here), + } + columns.push(Described_ { name, ty, layout }); + } + Ok(Described { + columns, + rows: rows.unwrap_or(0), + held: Arc::new(held), + }) +} + +/// What one Arrow column holds, and how wide its values are. +/// +/// The type ids are Arrow's own and the numbers are part of its format +/// rather than of any one library, which is why they are matched on +/// here: a table that came over IPC and a table built in JavaScript +/// carry the same ones. +fn meaning(env: &Env, name: &str, vector: &Object<'_>) -> Result<(LogicalType, Elem)> { + let ty: Object<'_> = vector.get_named_property("type")?; + let id: i32 = ty.get_named_property("typeId")?; + let printed: String = printed(&ty).unwrap_or_else(|| format!("type {id}")); + let refused = |instead: &str| { + crate::error::usage(env, format!("column '{name}' is {printed}, and {instead}")) + }; + let unit = || -> Result { ty.get_named_property::("unit").map(|n| n as i64) }; + let counts = |bits, signed| LogicalType::Int { + signed, + bits, + precision: None, + }; + let characters = LogicalType::Str { + min: None, + max: None, + fixed: false, + }; + let word = |bits, scale, means| { + ( + means, + Elem::Int { + bits, + signed: true, + scale, + }, + ) + }; + Ok(match id { + // Integers, whose width and signedness are on the type rather + // than in the id. + 2 => { + let width: f64 = ty.get_named_property("bitWidth")?; + let signed: bool = ty.get_named_property("isSigned")?; + let bits = match width as i64 { + 8 => IntBits::B8, + 16 => IntBits::B16, + 32 => IntBits::B32, + 64 => IntBits::B64, + _ => { + return Err(refused( + "this engine reads integers of 8, 16, 32 and 64 bits", + )); + } + }; + ( + counts(bits, signed), + Elem::Int { + bits, + signed, + scale: 1, + }, + ) + } + 3 => { + let precision: f64 = ty.get_named_property("precision")?; + let bits = match precision as i64 { + 1 => FloatBits::B32, + 2 => FloatBits::B64, + _ => return Err(refused("this engine reads 32 and 64 bit floats")), + }; + ( + LogicalType::Float { + bits, + precision: None, + }, + Elem::Float(bits), + ) + } + 6 => (LogicalType::Bool, Elem::Bool), + // The two string layouts that cut one buffer up with offsets. + // A view column is Arrow's third and is not one of these: its + // chunks are several buffers a row can point into, which this + // client does not read yet. + 5 => (characters, Elem::Str { wide: false }), + 20 => (characters, Elem::Str { wide: true }), + // A date is days when its unit is days and milliseconds when it + // is not, and the second of those is eight bytes wide. + 8 => match unit()? { + 0 => word(IntBits::B32, 1, LogicalType::Date), + _ => { + return Err(refused( + "a date of this engine counts whole days, so cast it to Date32 or read it as \ + a datetime", + )); + } + }, + // A time, a datetime and a duration are all counts, and what + // differs is the unit they count in and what the count means. + 9 => match unit()? { + 0 => word(IntBits::B32, 1_000_000_000, LogicalType::LocalTime), + 1 => word(IntBits::B32, 1_000_000, LogicalType::LocalTime), + 2 => word(IntBits::B64, 1_000, LogicalType::LocalTime), + _ => word(IntBits::B64, 1, LogicalType::LocalTime), + }, + 10 => { + let zone: Unknown<'_> = ty.get_named_property("timezone")?; + if zone.get_type()? == ValueType::String { + let zone = String::from_unknown(zone)?; + return Err(refused(&format!( + "a column of this table has nowhere to keep '{zone}', so drop the zone once \ + the values are in the zone you want them in, or write it as a string" + ))); + } + word(IntBits::B64, nanos(unit()?), LogicalType::LocalDatetime) + } + 18 => word( + IntBits::B64, + nanos(unit()?), + LogicalType::Duration(DurationKind::DayTime), + ), + 11 => match unit()? { + 0 => word( + IntBits::B32, + 1, + LogicalType::Duration(DurationKind::YearMonth), + ), + _ => { + return Err(refused( + "a duration of this engine is months or it is nanoseconds, so cast it to \ + Interval or to a Duration", + )); + } + }, + -1 => { + return Err(refused( + "a dictionary is a layout rather than a type here, so cast it to what it holds \ + first", + )); + } + 4 | 19 | 23 => { + return Err(refused( + "no statement can read a column of bytes back yet, so registering one would be \ + naming data the caller cannot get at", + )); + } + _ => { + return Err(refused( + "a column holds booleans, integers, floats, strings, dates, times, datetimes or \ + durations", + )); + } + }) +} + +/// What the Arrow type calls itself, for a message about a column this +/// client will not take. +fn printed(ty: &Object<'_>) -> Option { + let name: Function<'_, (), String> = ty.get_named_property("toString").ok()?; + name.apply(*ty, ()).ok() +} + +/// How many nanoseconds one count of an Arrow time unit is. +fn nanos(unit: i64) -> i64 { + match unit { + 0 => 1_000_000_000, + 1 => 1_000_000, + 2 => 1_000, + _ => 1, + } +} + +/// One Arrow column, as the layout the engine reads it through. +/// +/// A vector is chunks, and a column of a table is one run of bytes, so a +/// vector of one chunk is described where it lies and a vector of +/// several is concatenated first. The common case is the first: a table +/// built in this process, or read from one batch, holds a column whole. +fn column( + env: &Env, + name: &str, + elem: Elem, + vector: &Object<'_>, + held: &mut Held, +) -> Result<(u64, Layout)> { + let chunks: Object<'_> = vector.get_named_property("data")?; + let count = chunks.get_array_length()?; + let mut pieces = Vec::with_capacity(count as usize); + let mut row = 0usize; + let nulls: f64 = vector.get_named_property("nullCount")?; + for at in 0..count { + let chunk: Object<'_> = chunks.get_element(at)?; + // Named by the row that is empty rather than by how many are, + // because the caller's next move is to go and look at that cell. + if nulls != 0.0 + && let Some(empty) = hole(&chunk)? + { + return Err(crate::error::usage( + env, + format!( + "column '{name}' has no value at row {}, and every column of a row holds one", + row + empty + ), + )); + } + let piece = piece(env, name, &elem, &chunk)?; + row += piece.rows; + pieces.push(piece); + } + let rows: usize = pieces.iter().map(|piece| piece.rows).sum(); + Ok((rows as u64, laid(env, name, elem, pieces, rows, held)?)) +} + +/// Which row of this chunk holds nothing, when one does. +/// +/// The validity bitmap is a bit a row, set where there is a value. A +/// chunk with none of them clear is a chunk whose bitmap is empty, which +/// is what a column with no nulls in it carries even when a sibling +/// chunk has some. +fn hole(chunk: &Object<'_>) -> Result> { + let Some(bitmap) = Lent::of(chunk.get_named_property("nullBitmap")?)? else { + return Ok(None); + }; + let raw = bitmap.raw(); + if raw.is_empty() { + return Ok(None); + } + let rows: f64 = chunk.get_named_property("length")?; + let at: f64 = chunk.get_named_property("offset")?; + let at = at as usize; + Ok((0..rows as usize).find(|row| { + let bit = at + row; + raw.get(bit / 8) + .is_none_or(|byte| byte >> (bit % 8) & 1 == 0) + })) +} + +/// One chunk of one Arrow column. +/// +/// A chunk of a sliced column is the awkward one, and arrow-js splits +/// the difference in a way worth writing down. Slicing narrows the +/// values buffer and the offsets buffer to the rows that were kept, so +/// those two already start at row zero of the chunk. It leaves the +/// validity bitmap and a boolean column's bits alone, because both count +/// in bits and a bit is not a place a typed array can start. So `offset` +/// stays on the chunk, and it means those two buffers and nothing else. +struct Piece { + rows: usize, + /// Which bit of the validity bitmap, and of a boolean column's data, + /// this chunk's first row is. + at: usize, + values: Lent, + offsets: Option, +} + +fn piece(env: &Env, name: &str, elem: &Elem, chunk: &Object<'_>) -> Result { + let missing = || { + crate::error::usage( + env, + format!("column '{name}' is a shape this client cannot find the values of"), + ) + }; + let rows: f64 = chunk.get_named_property("length")?; + let at: f64 = chunk.get_named_property("offset")?; + let values = Lent::of(chunk.get_named_property("values")?)?.ok_or_else(missing)?; + let offsets = match elem { + Elem::Str { .. } => { + Some(Lent::of(chunk.get_named_property("valueOffsets")?)?.ok_or_else(missing)?) + } + _ => None, + }; + Ok(Piece { + rows: rows as usize, + at: at as usize, + values, + offsets, + }) +} + +/// Where a column's bytes are, once it is known whether they are the +/// caller's or this client's. +/// +/// One chunk is described where it lies, which is the whole point of +/// registering a frame. Several are put together first, into a buffer of +/// this module's own, because a column of a table is one run of bytes +/// and three chunks are three of them. +fn laid( + env: &Env, + name: &str, + elem: Elem, + pieces: Vec, + rows: usize, + held: &mut Held, +) -> Result { + if rows == 0 { + // No row is ever read out of it, and what a layout still needs + // is a pointer that is not null and is aligned for the width it + // claims. An empty buffer of this module's own is both. + return Ok(match elem { + Elem::Str { .. } => Layout::Str { + offsets: held.own(Bytes::Offsets(Vec::new())), + wide: false, + data: held.own(Bytes::One(Vec::new())), + data_len: 0, + }, + _ => plain(elem, held.own(Bytes::empty(&elem))), + }); + } + match elem { + Elem::Str { wide } => strings(env, name, wide, pieces, rows, held), + Elem::Bool => Ok(Layout::Bool { + ptr: bits(pieces, rows, held), + }), + _ => Ok(plain(elem, words(elem, pieces, held))), + } +} + +/// The layout a fixed width column of this element is read through. +fn plain(elem: Elem, ptr: NonNull) -> Layout { + match elem { + Elem::Float(bits) => Layout::Float { ptr, bits }, + Elem::Int { + bits, + signed, + scale, + } => Layout::Int { + ptr, + bits, + signed, + scale, + }, + // A boolean and a string are laid out by the two functions that + // know their shapes, and neither of them comes through here. + _ => Layout::Bool { ptr }, + } +} + +/// Where a fixed width column's words are. +/// +/// The values buffer of a chunk already starts at that chunk's first +/// row, sliced or not, so one chunk is lent exactly as it arrived. +fn words(elem: Elem, mut pieces: Vec, held: &mut Held) -> NonNull { + let width = elem.width(); + if pieces.len() == 1 { + return held.lend(pieces.pop().expect("the one chunk").values); + } + let mut built = Bytes::with_room(&elem, pieces.iter().map(|piece| piece.rows).sum()); + for piece in &pieces { + built.extend(&piece.values.raw()[..piece.rows * width]); + } + held.own(built) +} + +/// Where a boolean column's bitmap is. +/// +/// A chunk whose first row is a whole byte in is described where it +/// lies, because the engine reads a bitmap from a byte boundary. One +/// that starts partway through a byte, or a column of several chunks, is +/// rebuilt, which is a bit a row and happens once. +fn bits(mut pieces: Vec, rows: usize, held: &mut Held) -> NonNull { + if pieces.len() == 1 && pieces[0].at.is_multiple_of(8) { + let one = pieces.pop().expect("the one chunk"); + let byte = one.at / 8; + let ptr = held.lend(one.values); + return unsafe { NonNull::new_unchecked(ptr.as_ptr().add(byte)) }; + } + let mut built = vec![0u8; rows.div_ceil(8).max(1)]; + let mut row = 0usize; + for piece in &pieces { + let raw = piece.values.raw(); + for at in 0..piece.rows { + let bit = piece.at + at; + if raw[bit / 8] >> (bit % 8) & 1 == 1 { + built[row / 8] |= 1 << (row % 8); + } + row += 1; + } + } + held.own(Bytes::One(built)) +} + +/// Where a string column's characters and offsets are. +/// +/// The characters are never copied. What may be is the offsets: the +/// engine reads a column whose first offset is zero, and a chunk that is +/// a slice of a longer array starts somewhere else, so the offsets are +/// rebased and the characters are pointed at from where that chunk's +/// first one begins. A column of several chunks is the one case where +/// the characters do move, because three runs of bytes are not one. +fn strings( + env: &Env, + name: &str, + wide: bool, + mut pieces: Vec, + rows: usize, + held: &mut Held, +) -> Result { + let span = |piece: &Piece| -> Result<(usize, usize)> { + let offsets = piece.offsets.as_ref().ok_or_else(|| { + crate::error::usage( + env, + format!("column '{name}' holds strings and does not say where they are cut"), + ) + })?; + let at = |row: usize| { + offsets.offset(row).ok_or_else(|| { + crate::error::usage( + env, + format!("column '{name}' is cut by offsets this client cannot read"), + ) + }) + }; + Ok((at(0)? as usize, at(piece.rows)? as usize)) + }; + if pieces.len() == 1 { + let one = pieces.pop().expect("the one chunk"); + let (from, to) = span(&one)?; + let Piece { + values, offsets, .. + } = one; + let offsets = offsets.expect("a string column is cut by offsets"); + if from == 0 { + let data = held.lend(values); + return Ok(Layout::Str { + offsets: held.lend(offsets), + wide, + data, + data_len: to, + }); + } + // Rebased rather than copied down: the characters stay where + // they are and what is rebuilt is the `rows + 1` numbers that + // cut them up, which is four bytes a row against however many + // characters a row holds. A sliced column is where this happens: + // its offsets were kept whole and so still count from the row + // the whole column started at. + let mut rebased = Vec::with_capacity(rows + 1); + for row in 0..=rows { + let raw = offsets.offset(row).unwrap_or(0) as usize - from; + rebased.push(narrow(env, name, raw)?); + } + let data = held.lend(values); + return Ok(Layout::Str { + offsets: held.own(Bytes::Offsets(rebased)), + wide: false, + data: unsafe { NonNull::new_unchecked(data.as_ptr().add(from)) }, + data_len: to - from, + }); + } + let mut data: Vec = Vec::new(); + let mut offsets: Vec = Vec::with_capacity(rows + 1); + offsets.push(0); + for piece in &pieces { + let (from, to) = span(piece)?; + let raw = piece.values.raw(); + let base = data.len(); + data.extend_from_slice(&raw[from..to]); + let cuts = piece.offsets.as_ref().expect("the offsets"); + for row in 1..=piece.rows { + let end = cuts.offset(row).unwrap_or(0) as usize - from + base; + offsets.push(narrow(env, name, end)?); + } + } + let data_len = data.len(); + // The characters first, so that a pointer taken from either one is + // taken after everything that could have moved the vector has. + let data = held.own(Bytes::One(data)); + Ok(Layout::Str { + offsets: held.own(Bytes::Offsets(offsets)), + wide: false, + data, + data_len, + }) +} + +/// One offset of a rebuilt string column, which is 32 bits wide. +fn narrow(env: &Env, name: &str, offset: usize) -> Result { + i32::try_from(offset).map_err(|_| { + crate::error::usage( + env, + format!( + "column '{name}' holds more than two gigabytes of characters once its chunks are \ + put together, which is further than the offsets of a frame reach" + ), + ) + }) +} + +/// Reads an object of column name to values. +/// +/// The shape a program with no frame library writes, and the shape one +/// with typed arrays writes too: a typed array is a column already and +/// is described where it lies, and a plain array is read into a buffer +/// of this module's own, because an array of JavaScript values is a +/// column of objects and a column of a table is a run of numbers. +fn from_object(env: &Env, object: &Object<'_>) -> Result { + let mut held = Held::new(); + let mut columns = Vec::new(); + let mut rows: Option = None; + for name in Object::keys(object)? { + let value: Unknown<'_> = object.get_named_property(name.as_str())?; + let (rows_here, ty, layout) = match Lent::of(value)? { + Some(array) => { + let elem = array.elem(); + let ty = match elem { + Elem::Float(bits) => LogicalType::Float { + bits, + precision: None, + }, + Elem::Int { bits, signed, .. } => LogicalType::Int { + signed, + bits, + precision: None, + }, + _ => unreachable!("a typed array is words or floats"), + }; + let rows = array.len() as u64; + match rows { + 0 => (0, ty, plain(elem, held.own(Bytes::empty(&elem)))), + _ => (rows, ty, plain(elem, held.lend(array))), + } + } + None => { + let built = values(env, &name, value)?; + let rows = built.len() as u64; + let (ty, layout) = packed(env, &name, built, &mut held)?; + (rows, ty, layout) + } + }; + match rows { + Some(had) if had != rows_here => { + return Err(crate::error::usage( + env, + format!( + "column '{name}' holds {rows_here} values and the column before it holds \ + {had}, and a table is as wide as it is long" + ), + )); + } + _ => rows = Some(rows_here), + } + columns.push(Described_ { name, ty, layout }); + } + Ok(Described { + columns, + rows: rows.unwrap_or(0), + held: Arc::new(held), + }) +} + +/// One column, read out of an array of JavaScript values. +/// +/// The first value settles what the column is and every value after it +/// has to agree, which is the loader's rule in the Python client and is +/// this client's appender's rule too. What it adds is the widening: a +/// column that started as whole numbers and meets a fractional one +/// becomes a column of floats, because `[1, 2, 2.5]` is a column of +/// numbers however it was written. +fn values(env: &Env, name: &str, value: Unknown<'_>) -> Result { + if value.get_type()? != ValueType::Object || !value.is_array()? { + return Err(crate::error::usage( + env, + format!( + "column '{name}' is {}, and a column is an array of values or a typed array", + named(&value) + ), + )); + } + let array = Object::from_unknown(value)?; + let len = array.get_array_length()?; + let mut column: Option = None; + for row in 0..len { + let value: Unknown<'_> = array.get_element(row)?; + match column.as_mut() { + Some(column) => column.widening_push(env, value).map_err(|why| match why { + Mismatch::Wanted(holds) => crate::error::usage( + env, + format!( + "column '{name}' holds {holds} and row {row} is {}", + named(&value) + ), + ), + Mismatch::Says(reason) => crate::error::usage( + env, + format!("row {row} does not go in column '{name}': {reason}"), + ), + Mismatch::Boundary(err) => err, + })?, + None => { + column = Some( + buffer::Column::start(env, value) + .map_err(|why| match why { + Mismatch::Boundary(err) => err, + Mismatch::Wanted(_) | Mismatch::Says(_) => unreachable!( + "a column that has not started has nothing to disagree with" + ), + })? + .ok_or_else(|| { + crate::error::usage( + env, + format!( + "column '{name}' starts at row {row} with {}, and a column \ + holds booleans, integers, floats, strings, dates, times, \ + datetimes or durations", + named(&value) + ), + ) + })?, + ); + } + } + } + column.ok_or_else(|| { + crate::error::usage( + env, + format!("column '{name}' is empty, and an empty column says nothing about what it would hold"), + ) + }) +} + +/// One column of JavaScript values, as the bytes a frame reads and what +/// they mean. +fn packed( + env: &Env, + name: &str, + column: buffer::Column, + held: &mut Held, +) -> Result<(LogicalType, Layout)> { + let counts = LogicalType::Int { + signed: true, + bits: IntBits::B64, + precision: None, + }; + let word = Elem::Int { + bits: IntBits::B64, + signed: true, + scale: 1, + }; + // The temporal buffers count in `i64` and the integer one holds the + // bits of one in a `u64`, and both of them are the eight-byte lane + // the engine reads through, so this cast is the identity on the + // bytes and the logical type beside them is what says how to read + // them. + let same = |v: Vec| Bytes::Eight(v.into_iter().map(|n| n as u64).collect()); + Ok(match column { + buffer::Column::Int(v) => (counts, plain(word, held.own(same(v)))), + buffer::Column::Float(v) => { + let bits = FloatBits::B64; + ( + LogicalType::Float { + bits, + precision: None, + }, + plain( + Elem::Float(bits), + held.own(Bytes::Eight(v.into_iter().map(f64::to_bits).collect())), + ), + ) + } + buffer::Column::Bool(v) => { + // At least one byte, so the pointer is an allocation and not + // the address an empty vector lends out. + let mut packed = vec![0u8; v.len().div_ceil(8).max(1)]; + for (row, &yes) in v.iter().enumerate() { + if yes { + packed[row / 8] |= 1 << (row % 8); + } + } + ( + LogicalType::Bool, + Layout::Bool { + ptr: held.own(Bytes::One(packed)), + }, + ) + } + buffer::Column::Str(v) => { + let mut data = Vec::with_capacity(v.iter().map(String::len).sum::().max(1)); + let mut offsets = Vec::with_capacity(v.len() + 1); + offsets.push(0i32); + for word in &v { + data.extend_from_slice(word.as_bytes()); + offsets.push(narrow(env, name, data.len())?); + } + let data_len = data.len(); + let data = held.own(Bytes::One(data)); + ( + LogicalType::Str { + min: None, + max: None, + fixed: false, + }, + Layout::Str { + offsets: held.own(Bytes::Offsets(offsets)), + wide: false, + data, + data_len, + }, + ) + } + buffer::Column::Bytes(_) => { + return Err(crate::error::usage( + env, + format!( + "column '{name}' holds byte strings, and no statement can read a column of \ + bytes back yet" + ), + )); + } + buffer::Column::Date(v) => ( + LogicalType::Date, + plain( + Elem::Int { + bits: IntBits::B32, + signed: true, + scale: 1, + }, + held.own(Bytes::Four(v.into_iter().map(|n| n as u32).collect())), + ), + ), + buffer::Column::LocalTime(v) => (LogicalType::LocalTime, plain(word, held.own(same(v)))), + buffer::Column::LocalDatetime(v) => { + (LogicalType::LocalDatetime, plain(word, held.own(same(v)))) + } + buffer::Column::Duration(kind, v) => { + (LogicalType::Duration(kind), plain(word, held.own(same(v)))) + } + }) +} diff --git a/src/lib.rs b/src/lib.rs index 8c50457..0183b6b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -21,6 +21,8 @@ mod buffer; mod cancel; mod conn; mod error; +mod frame; +mod register; mod stream; mod temporal; mod txn; diff --git a/src/register.rs b/src/register.rs new file mode 100644 index 0000000..16be538 --- /dev/null +++ b/src/register.rs @@ -0,0 +1,279 @@ +//! Frames, registered under a name a statement can match on. +//! +//! ```js +//! await conn.register('people', table) +//! await conn.query('MATCH (p:people) WHERE p.age > 40 RETURN p.name AS name') +//! ``` +//! +//! This is the replacement scan, and it copies nothing. What the engine +//! is told is where the caller's columns are, how wide their values are +//! and what they mean; a statement that matches the name builds vectors +//! that point straight at those buffers, so a frame of ten million rows +//! is registered in the time it takes to describe its columns and read +//! at the speed of the memory it already sits in. +//! +//! Because it is not a copy, a registered frame is a view and not a +//! snapshot: write into the array behind it and the next statement +//! answers what is there now. The exceptions are the two [`crate::frame`] +//! names, an Arrow column that arrived in several chunks and an object of +//! plain JavaScript arrays, and both of them are copied because there was +//! no one run of bytes to point at in the first place. +//! +//! A frame belongs to the connection it was registered on and goes when +//! that connection does. It is not written to the database, no other +//! program opening the same file sees it, and nothing writes to it: a +//! statement that tries to insert into or delete from one is refused +//! with the reason. `unregister` takes the name away entirely, and the +//! bytes go back to the caller when the last statement reading them has +//! finished with them. +//! +//! All three calls are asynchronous, because all three take the +//! connection's lock and a call that waits on the event loop is the +//! thing this client does not do. That includes `registered()`, which is +//! why it is a method and not a getter. + +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::atomic::AtomicBool; + +use napi::bindgen_prelude::*; +use napi::{Env, ScopedTask}; +use zudb::ZuError; +use zudb::zu1::catalog::Catalog; + +use crate::conn::{CLOSED, Failure, failed, with}; +use crate::frame::Described; + +/// What the three tasks share: the connection, and whether it is still +/// there to be had. +pub struct Held { + inner: Arc>>, + alive: Arc, + in_txn: Arc, +} + +impl Held { + pub fn new( + inner: Arc>>, + alive: Arc, + in_txn: Arc, + ) -> Held { + Held { + inner, + alive, + in_txn, + } + } + + fn run( + &self, + run: impl FnOnce(&mut zudb::Connection) -> std::result::Result, + ) -> std::result::Result { + with(&self.inner, &self.alive, &self.in_txn, run) + } +} + +/// Registers a frame as a table called `name`. +pub struct RegisterTask { + held: Held, + name: String, + /// The frame, read on the thread that owns the runtime because that + /// is the only thread allowed to read a JavaScript value. `None` + /// once the task has taken it, and `None` from the start when the + /// call was refused before there was anything to read. + described: Option, + refused: Option, +} + +impl RegisterTask { + pub fn new( + held: Held, + name: String, + described: Option, + refused: Option, + ) -> RegisterTask { + RegisterTask { + held, + name, + described, + refused, + } + } + + fn run(&mut self) -> std::result::Result { + if let Some(message) = self.refused.take() { + return Err(Failure::Usage(message)); + } + let described = self + .described + .take() + .ok_or_else(|| Failure::Usage(TWICE.to_string()))?; + let name = self.name.clone(); + let rows = described.rows as i64; + self.held.run(move |engine| { + // Refused here rather than at the statement that would have + // hit it. The engine keeps frames in an id space of their + // own and would take this name happily; what it could not do + // is bind it, because a label in a statement is one thing and + // the stored table would win. Better said at the call that + // made the clash. + if has_table(engine, &name)? { + return Err(Failure::Usage(format!( + "'{name}' is already a table of this database, and registering over one \ + would hide rows this frame knows nothing about" + ))); + } + // The walk `Frame::new` does over the unsigned, scaled and + // string columns happens in here, on the threadpool thread, + // which is the reason a frame of ten million rows does not + // stall the event loop while its offsets are checked. + described.register(engine, &name)?; + Ok(rows) + }) + } +} + +impl<'task> ScopedTask<'task> for RegisterTask { + type Output = std::result::Result; + type JsValue = i64; + + fn compute(&mut self) -> Result { + Ok(self.run()) + } + + fn resolve(&mut self, env: &'task Env, output: Self::Output) -> Result { + output.map_err(|failure| failed(env, failure, None)) + } +} + +/// What a task says when it is run a second time, which nothing this +/// client writes does: a task is built by one call and driven once. +const TWICE: &str = "this registration has already run, and a frame is registered once"; + +/// Takes a registered frame's name away and gives the bytes back. +pub struct UnregisterTask { + held: Held, + name: String, + refused: Option, +} + +impl UnregisterTask { + pub fn new(held: Held, name: String, refused: Option) -> UnregisterTask { + UnregisterTask { + held, + name, + refused, + } + } +} + +impl<'task> ScopedTask<'task> for UnregisterTask { + type Output = std::result::Result<(), Failure>; + type JsValue = (); + + fn compute(&mut self) -> Result { + if let Some(message) = self.refused.take() { + return Ok(Err(Failure::Usage(message))); + } + let name = self.name.clone(); + Ok(self + .held + .run(move |engine| match engine.unregister(&name)? { + true => Ok(()), + false => Err(Failure::Usage(format!( + "nothing is registered here as '{name}', and a name this connection did not \ + register is a table of the database rather than a frame of the caller's" + ))), + })) + } + + fn resolve(&mut self, env: &'task Env, output: Self::Output) -> Result { + output.map_err(|failure| failed(env, failure, None)) + } +} + +/// The names frames are registered under on this connection, sorted. +pub struct RegisteredTask { + held: Held, + refused: Option, +} + +impl RegisteredTask { + pub fn new(held: Held, refused: Option) -> RegisteredTask { + RegisteredTask { held, refused } + } +} + +impl<'task> ScopedTask<'task> for RegisteredTask { + type Output = std::result::Result, Failure>; + type JsValue = Vec; + + fn compute(&mut self) -> Result { + if let Some(message) = self.refused.take() { + return Ok(Err(Failure::Usage(message))); + } + Ok(self.held.run(|engine| Ok(engine.registered()))) + } + + fn resolve(&mut self, env: &'task Env, output: Self::Output) -> Result { + output.map_err(|failure| failed(env, failure, None)) + } +} + +/// Whether the database already holds a table of this name. +/// +/// Node tables and rel tables both, because a name is a label in a +/// statement either way and registering over either of them would be the +/// same mistake. +fn has_table(conn: &mut zudb::Connection, name: &str) -> std::result::Result { + let file = conn.session_mut().file_mut()?; + let catalog = Catalog::load(file)?; + Ok(catalog.node_by_name(name).is_some() || catalog.rel_by_name(name).is_some()) +} + +/// Refuses a name a statement could not carry. +/// +/// A table name goes into a statement as itself, so a name that is not +/// an identifier is a name that would either fail to parse or parse as +/// something else, and the second of those is the one worth refusing +/// for. +pub fn identifier(name: &str, what: &str) -> std::result::Result<(), String> { + let mut chars = name.chars(); + let starts = chars + .next() + .is_some_and(|first| first.is_ascii_alphabetic() || first == '_'); + if !starts || !chars.all(|c| c.is_ascii_alphanumeric() || c == '_') { + return Err(format!( + "'{name}' is not a name a statement can carry, and {what} is named by a letter or an \ + underscore followed by letters, digits or underscores" + )); + } + Ok(()) +} + +/// What this client refuses a frame for, before the connection is even +/// asked for. +/// +/// Read on the thread that owns the runtime, because reading the frame +/// is reading JavaScript values. What comes back is the description and +/// the sentence to refuse the call with, and never both. +pub fn read(env: &Env, name: &str, data: Unknown<'_>) -> std::result::Result { + identifier(name, "a registered frame")?; + let described = crate::frame::read(env, data)?; + if described.columns.is_empty() { + return Err( + "this frame has no columns, and a table whose rows hold nothing is not a table" + .to_string(), + ); + } + for column in &described.columns { + identifier(&column.name, "a column of a registered frame")?; + } + Ok(described) +} + +/// What a closed connection says, kept here so the three calls say it +/// once. +pub fn closed() -> String { + CLOSED.to_string() +} diff --git a/test/register.test.mjs b/test/register.test.mjs new file mode 100644 index 0000000..80528d6 --- /dev/null +++ b/test/register.test.mjs @@ -0,0 +1,524 @@ +// Frames, registered under a name a statement can match on. +// +// The point of the call is that columns a program already has become +// something a statement can read, and that reading them costs nothing: +// the engine is told where they are and reads them where they lie. So +// most of these register one and then match it, and the ones that matter +// most prove the two halves of that claim, which are that the bytes are +// never copied and that they are handed back when the name goes. +// +// `apache-arrow` is a dev dependency and not a dependency: the client +// recognizes a table by its shape rather than by its class, so what +// these tests exercise is the same path any other library that speaks +// that shape would take. + +import assert from 'node:assert/strict' +import test from 'node:test' + +import { + Binary, + Bool, + RecordBatch, + Table, + TimeUnit, + Timestamp, + Utf8, + tableFromArrays, + vectorFromArray, +} from 'apache-arrow' +import { ZuDate, ZuDuration, ZuTimestamp, connect } from 'zudb' + +import { fresh, isZuError, twoPeople } from './helper.mjs' + +// The `name` column of a registered frame, in the order it went in. +async function names(conn, table) { + const rows = await conn.query(`MATCH (f:${table}) RETURN f.name AS name`) + return rows.map((row) => row.name) +} + +test('an object of arrays is a frame', async (t) => { + const { conn } = await fresh(t) + + // The way in for a program with no columnar library at all. + assert.equal(await conn.register('people', { uid: [1, 2, 3], name: ['ada', 'grace', 'lynn'] }), 3) + assert.deepEqual(await names(conn, 'people'), ['ada', 'grace', 'lynn']) +}) + +test('a statement reads a registered frame like any other table', async (t) => { + const { conn } = await fresh(t) + await conn.register('people', { + uid: [1, 2, 3], + name: ['ada', 'grace', 'lynn'], + age: [36, 45, 52], + }) + + const rows = await conn.query('MATCH (p:people) WHERE p.age > 40 RETURN p.name AS name') + assert.deepEqual( + rows.map((row) => row.name), + ['grace', 'lynn'], + ) +}) + +test('an object of typed arrays is read where it lies', async (t) => { + const { conn } = await fresh(t) + + const uid = new BigInt64Array([1n, 2n, 3n]) + assert.equal(await conn.register('numbers', { uid }), 3) + const before = await conn.query('MATCH (n:numbers) RETURN sum(n.uid) AS total') + assert.equal(before[0].total, 6n) + + // Written into between two statements, and the second one answers the + // new number. No copy taken at registration could do that. + uid[0] = 1000n + const after = await conn.query('MATCH (n:numbers) RETURN sum(n.uid) AS total') + assert.equal(after[0].total, 1005n) +}) + +test('an arrow table goes in', async (t) => { + const { conn } = await fresh(t) + + const table = tableFromArrays({ + uid: new BigInt64Array([1n, 2n]), + name: vectorFromArray(['ada', 'grace'], new Utf8()), + }) + assert.equal(await conn.register('people', table), 2) + assert.deepEqual(await names(conn, 'people'), ['ada', 'grace']) +}) + +test('a record batch is a frame too', async (t) => { + const { conn } = await fresh(t) + + const table = tableFromArrays({ + uid: new BigInt64Array([1n, 2n]), + name: vectorFromArray(['ada', 'grace'], new Utf8()), + }) + const [batch] = table.batches + assert.ok(batch instanceof RecordBatch) + assert.equal(await conn.register('people', batch), 2) + assert.deepEqual(await names(conn, 'people'), ['ada', 'grace']) +}) + +test('a column of several chunks is one column', async (t) => { + const { conn } = await fresh(t) + + // A column of a table is one run of bytes, so this is the one arrow + // shape that costs a memcpy per column on the way in. + const batch = (uid, name) => + tableFromArrays({ + uid: new BigInt64Array(uid), + name: vectorFromArray(name, new Utf8()), + }).batches[0] + const joined = new Table([batch([1n, 2n], ['ada', 'grace']), batch([3n], ['lynn'])]) + assert.equal(joined.getChildAt(0).data.length, 2) + + assert.equal(await conn.register('people', joined), 3) + assert.deepEqual(await names(conn, 'people'), ['ada', 'grace', 'lynn']) +}) + +test('a sliced column is read from the row it starts at', async (t) => { + const { conn } = await fresh(t) + + // A slice is a column with a row offset, which is the one thing a bare + // pointer cannot say, so the words are pointed at further in and the + // offsets of a string column are rebased. + const table = tableFromArrays({ + uid: new BigInt64Array([1n, 2n, 3n, 4n]), + name: vectorFromArray(['a', 'b', 'c', 'd'], new Utf8()), + yes: vectorFromArray([true, false, true, false], new Bool()), + }) + assert.equal(await conn.register('people', table.slice(1, 3)), 2) + assert.deepEqual(await names(conn, 'people'), ['b', 'c']) + + // The numbers as well as the words, because a slice moves a column of + // words and a column of characters in two different ways and only one + // of them shows up in the names. + const rows = await conn.query( + 'MATCH (p:people) RETURN p.uid AS uid, p.name AS name, p.yes AS yes', + ) + assert.deepEqual( + rows.map((row) => [row.uid, row.name, row.yes]), + [ + [2n, 'b', false], + [3n, 'c', true], + ], + ) +}) + +test('a boolean column sliced partway through a byte is rebuilt', async (t) => { + const { conn } = await fresh(t) + + // A bitmap is read from a byte boundary, so a chunk that starts at + // row three is the one case where the bits are laid out again. + const yes = [true, false, true, false, true, true, false, true, false, true] + const table = tableFromArrays({ yes: vectorFromArray(yes, new Bool()) }) + assert.equal(await conn.register('flags', table.slice(3, 9)), 6) + + const rows = await conn.query('MATCH (f:flags) RETURN f.yes AS yes') + assert.deepEqual( + rows.map((row) => row.yes), + yes.slice(3, 9), + ) +}) + +test('every kind of column a row can hold arrives as itself', async (t) => { + const { conn } = await fresh(t) + + await conn.register('kinds', { + yes: [true, false], + small: new Int8Array([1, 2]), + wide: new Uint32Array([3, 4]), + narrow: new Float32Array([1.5, 2.5]), + word: ['a', 'b'], + day: [new ZuDate(19723), new ZuDate(19754)], + moment: [new ZuTimestamp(1_704_070_923_000_000_000n), new ZuTimestamp(0n)], + span: [ZuDuration.ofNanos(90_000_000_000n), ZuDuration.ofNanos(0n)], + }) + + const rows = await conn.query( + 'MATCH (k:kinds) RETURN k.yes AS yes, k.small AS small, k.wide AS wide, ' + + 'k.narrow AS narrow, k.word AS word, k.day AS day, k.moment AS moment, k.span AS span', + ) + assert.equal(rows[0].yes, true) + assert.equal(rows[0].small, 1n) + assert.equal(rows[0].wide, 3n) + assert.equal(rows[0].narrow, 1.5) + assert.equal(rows[0].word, 'a') + assert.equal(rows[0].day.days, 19723) + assert.equal(rows[0].moment.nanos, 1_704_070_923_000_000_000n) + assert.equal(rows[0].span.nanos, 90_000_000_000n) + assert.equal(rows[1].yes, false) +}) + +test('an arrow column of every width arrives as itself', async (t) => { + const { conn } = await fresh(t) + + const table = tableFromArrays({ + small: new Int8Array([1, 2]), + wide: new Uint32Array([3, 4]), + narrow: new Float32Array([1.5, 2.5]), + big: new Float64Array([1.25, 2.25]), + }) + assert.equal(await conn.register('widths', table), 2) + const rows = await conn.query( + 'MATCH (w:widths) RETURN w.small AS small, w.wide AS wide, w.narrow AS narrow, w.big AS big', + ) + assert.equal(rows[0].small, 1n) + assert.equal(rows[0].wide, 3n) + assert.equal(rows[0].narrow, 1.5) + assert.equal(rows[0].big, 1.25) +}) + +test('an object of plain arrays is copied because an array is not a column', async (t) => { + const { conn } = await fresh(t) + + // The one way in that does copy, and the reason it has to. + const name = ['ada'] + await conn.register('people', { uid: [1], name }) + name[0] = 'grace' + assert.deepEqual(await names(conn, 'people'), ['ada']) +}) + +test('a column of whole numbers that meets a fractional one widens', async (t) => { + const { conn } = await fresh(t) + + await conn.register('numbers', { n: [1, 2, 2.5] }) + const rows = await conn.query('MATCH (x:numbers) RETURN sum(x.n) AS total') + assert.equal(rows[0].total, 5.5) +}) + +test('a frame belongs to the connection that registered it', async (t) => { + const { conn, path } = await fresh(t) + await conn.register('people', { uid: [1], name: ['ada'] }) + + // Nothing is written to the database, so another connection to the + // same file has never heard of it. + const other = await connect(path) + t.after(() => other.close()) + assert.deepEqual(await other.registered(), []) + assert.deepEqual(await names(other, 'people'), []) +}) + +test('registered says what is registered here', async (t) => { + const { conn } = await fresh(t) + + assert.deepEqual(await conn.registered(), []) + await conn.register('second', { a: [1] }) + await conn.register('first', { a: [1] }) + assert.deepEqual(await conn.registered(), ['first', 'second']) +}) + +test('registering a name again replaces what it stands for', async (t) => { + const { conn } = await fresh(t) + + await conn.register('people', { uid: [1, 2], name: ['ada', 'grace'] }) + assert.equal(await conn.register('people', { uid: [3], name: ['lynn'] }), 1) + assert.deepEqual(await names(conn, 'people'), ['lynn']) +}) + +test('a name registered again may hold a different shape', async (t) => { + const { conn } = await fresh(t) + + // A frame is not a table, so nothing about the first registration + // survives the second. + await conn.register('people', { uid: [1], name: ['ada'] }) + await conn.register('people', { name: ['grace'], age: [45] }) + const rows = await conn.query('MATCH (p:people) RETURN p.name AS name, p.age AS age') + assert.equal(rows[0].name, 'grace') + assert.equal(rows[0].age, 45n) +}) + +test('unregister takes the name away', async (t) => { + const { conn } = await fresh(t) + + await conn.register('people', { uid: [1, 2], name: ['ada', 'grace'] }) + await conn.unregister('people') + assert.deepEqual(await conn.registered(), []) + assert.deepEqual(await names(conn, 'people'), []) +}) + +test('a name that was unregistered can be registered again', async (t) => { + const { conn } = await fresh(t) + + await conn.register('people', { uid: [1], name: ['ada'] }) + await conn.unregister('people') + assert.equal(await conn.register('people', { uid: [2], name: ['grace'] }), 1) + assert.deepEqual(await names(conn, 'people'), ['grace']) +}) + +test('unregistering twice is refused', async (t) => { + const { conn } = await fresh(t) + + await conn.register('people', { a: [1] }) + await conn.unregister('people') + await assert.rejects(conn.unregister('people'), (err) => { + assert.ok(isZuError(err, 'ZuUsageError')) + assert.match(err.message, /nothing is registered here/) + return true + }) +}) + +test('unregistering a table nobody registered is refused', async (t) => { + const { conn } = await twoPeople(t) + + await assert.rejects(conn.unregister('person'), (err) => { + assert.match(err.message, /nothing is registered here/) + return true + }) +}) + +test('registering over a table of the database is refused', async (t) => { + const { conn } = await twoPeople(t) + + // A statement naming it would mean the stored one. + await assert.rejects(conn.register('person', { uid: [1] }), (err) => { + assert.match(err.message, /already a table of this database/) + return true + }) +}) + +test('nothing writes to a registered frame', async (t) => { + const { conn } = await fresh(t) + + // It is the caller's memory, read where it lies, and a statement that + // wrote into it would be writing into the caller's array. + await conn.register('people', { uid: [1], name: ['ada'] }) + await assert.rejects(conn.exec("INSERT (p:people {uid: 2, name: 'grace'})"), (err) => { + assert.match(err.message, /never written/) + return true + }) + await assert.rejects(conn.exec('MATCH (p:people) DETACH DELETE p'), (err) => { + assert.match(err.message, /never written/) + return true + }) +}) + +test('a null anywhere is refused by column and row', async (t) => { + const { conn } = await fresh(t) + + // A property that is null is one no row of this engine can hold. + const table = tableFromArrays({ + uid: new BigInt64Array([1n, 2n, 3n]), + name: vectorFromArray(['ada', null, 'lynn'], new Utf8()), + }) + await assert.rejects(conn.register('people', table), (err) => { + assert.match(err.message, /column 'name' has no value at row 1/) + return true + }) +}) + +test('a frame with no rows registers and matches nothing', async (t) => { + const { conn } = await fresh(t) + + // A frame knows what its columns are without being told by a row, so + // a filter that came back empty is still a table to match on. + assert.equal(await conn.register('people', { uid: new BigInt64Array(0) }), 0) + const rows = await conn.query('MATCH (p:people) RETURN count(*) AS n') + assert.equal(rows[0].n, 0n) +}) + +test('a frame with no columns is refused', async (t) => { + const { conn } = await fresh(t) + + await assert.rejects(conn.register('people', {}), (err) => { + assert.match(err.message, /no columns/) + return true + }) +}) + +test('an empty column says nothing about what it would hold', async (t) => { + const { conn } = await fresh(t) + + await assert.rejects(conn.register('people', { uid: [] }), (err) => { + assert.match(err.message, /column 'uid' is empty/) + return true + }) +}) + +test('a name a statement could not carry is refused', async (t) => { + const { conn } = await fresh(t) + + await assert.rejects(conn.register('two words', { a: [1] }), (err) => { + assert.match(err.message, /not a name a statement can carry/) + return true + }) + await assert.rejects(conn.register('people', { 'two words': [1] }), (err) => { + assert.match(err.message, /a column of a registered frame/) + return true + }) +}) + +test('a zoned timestamp is refused with what to do about it', async (t) => { + const { conn } = await fresh(t) + + const table = tableFromArrays({ + when: vectorFromArray( + [new Date(1_700_000_000_000)], + new Timestamp(TimeUnit.MILLISECOND, 'UTC'), + ), + }) + await assert.rejects(conn.register('moments', table), (err) => { + assert.match(err.message, /nowhere to keep/) + return true + }) +}) + +test('a dictionary is a layout rather than a type', async (t) => { + const { conn } = await fresh(t) + + // What `tableFromArrays` makes of a plain array of strings, which is + // the mistake a caller is most likely to make by accident. + const table = tableFromArrays({ name: ['ada', 'grace'] }) + await assert.rejects(conn.register('people', table), (err) => { + assert.match(err.message, /a dictionary is a layout rather than a type/) + return true + }) +}) + +test('a column of bytes is refused', async (t) => { + const { conn } = await fresh(t) + + // Naming one would be naming data no statement reads back. + const table = tableFromArrays({ raw: vectorFromArray([new Uint8Array([1])], new Binary()) }) + await assert.rejects(conn.register('blobs', table), (err) => { + assert.match(err.message, /column of bytes/) + return true + }) +}) + +test('an integer too large for a column is refused by row', async (t) => { + const { conn } = await fresh(t) + + // Checked once, at registration, so that reading a frame cannot fail: + // the engine's lane is signed and this value is not in it. + await assert.rejects( + conn.register('numbers', { big: new BigUint64Array([1n, 1n << 63n]) }), + (err) => { + assert.match(err.message, /at row 1/) + return true + }, + ) +}) + +test('a column that is as long as the one before it', async (t) => { + const { conn } = await fresh(t) + + await assert.rejects(conn.register('people', { uid: [1, 2], name: ['ada'] }), (err) => { + assert.match(err.message, /a table is as wide as it is long/) + return true + }) +}) + +test('something that is not a frame at all is refused with the list', async (t) => { + const { conn } = await fresh(t) + + await assert.rejects(conn.register('people', [1, 2, 3]), (err) => { + assert.match(err.message, /a frame is an Arrow table/) + return true + }) + await assert.rejects(conn.register('people', 'ada'), (err) => { + assert.match(err.message, /a frame is an Arrow table/) + return true + }) +}) + +test('a column that is neither an array nor a typed array is refused', async (t) => { + const { conn } = await fresh(t) + + await assert.rejects(conn.register('people', { uid: 1 }), (err) => { + assert.match(err.message, /column 'uid' is a number/) + return true + }) +}) + +test('registering inside a transaction is refused', async (t) => { + const { conn } = await fresh(t) + + // A frame is registered on the session, which is the thing a + // transaction is running on, and a rollback has nothing to say about + // memory the caller owns. + await using tx = await conn.transaction() + assert.ok(tx) + await assert.rejects(conn.register('people', { a: [1, 2] }), (err) => { + assert.match(err.message, /not inside a transaction/) + return true + }) +}) + +test('a closed connection registers nothing', async (t) => { + const { conn } = await fresh(t) + conn.close() + + await assert.rejects(conn.register('people', { a: [1] }), (err) => { + assert.ok(isZuError(err, 'ZuUsageError')) + assert.match(err.message, /closed/) + return true + }) + await assert.rejects(conn.unregister('people'), (err) => { + assert.match(err.message, /closed/) + return true + }) + await assert.rejects(conn.registered(), (err) => { + assert.match(err.message, /closed/) + return true + }) +}) + +test('registering costs the same whatever the frame holds', async (t) => { + const { conn } = await fresh(t) + + // Nothing is copied, so nothing about the call is per row. Five + // million rows against ten, and the budget is loose by a wide margin: + // it is here to catch a way in that started walking the rows rather + // than to hold a number. + const best = {} + for (const rows of [10, 5_000_000]) { + const n = new BigInt64Array(rows) + best[rows] = Infinity + for (let round = 0; round < 5; round += 1) { + const started = process.hrtime.bigint() + await conn.register('numbers', { n }) + best[rows] = Math.min(best[rows], Number(process.hrtime.bigint() - started) / 1e6) + } + } + assert.ok(best[5_000_000] < 2, `registering 5m rows took ${best[5_000_000].toFixed(2)} ms`) +}) diff --git a/test/types/cjs.cts b/test/types/cjs.cts index b20aedd..b32ac33 100644 --- a/test/types/cjs.cts +++ b/test/types/cjs.cts @@ -7,6 +7,7 @@ import { isZuError, ZuTimestamp, type ZuAppendValue, + type ZuArrowTable, type ZuParam, type ZuStream, type ZuTransactionOptions, @@ -87,6 +88,19 @@ export async function bulk(path: string, batch: readonly ZuAppendValue[][]): Pro } } +export async function scanned(path: string, table: ZuArrowTable): Promise { + // A table is taken by shape rather than by class, so a caller holding + // an `apache-arrow` Table passes it here without this package having + // any opinion about which version of that package they installed. + const conn = await connect(path) + try { + await conn.register('frame', table) + return await conn.registered() + } finally { + conn.close() + } +} + export async function ids(path: string): Promise { const conn = await connect(path) const stream: ZuStream<{ id: bigint }> = conn.stream('MATCH (p:person) RETURN p.id AS id') diff --git a/test/types/esm.mts b/test/types/esm.mts index a4680dd..3595497 100644 --- a/test/types/esm.mts +++ b/test/types/esm.mts @@ -11,6 +11,8 @@ import { type ZuBatch, type ZuBigIntMode, type ZuError, + type ZuFrame, + type ZuFrameColumn, type ZuPlainDate, type ZuRows, type ZuStream, @@ -132,6 +134,32 @@ export async function loaded(path: string, people: [bigint, string][]): Promise< return written + rows.committed } +export async function matched(path: string, ages: Int32Array): Promise { + await using conn = await connect(path) + + // An object of columns is a frame, and a typed array is a column, so + // this compiles without the caller reaching for a cast. An Arrow table + // is the other half of the union and is structural, so a table from + // `apache-arrow` is one without this package importing that package. + const column: ZuFrameColumn = ages + const frame: ZuFrame = { age: column, name: ['ada', 'grace'] } + + // A count of rows rather than nothing, so a caller can check that what + // went in is what they meant. + const rows: number = await conn.register('people', frame) + + const found = await conn.query<{ n: bigint }>( + 'MATCH (p:people) WHERE p.age > 40 RETURN count(*) AS n', + ) + + // A method rather than a getter, because it takes the lock like the + // rest of them, so it needs the `await` to typecheck. + const names: string[] = await conn.registered() + await conn.unregister('people') + + return rows + names.length + Number(found[0]?.n ?? 0n) +} + export function retryable(caught: unknown): boolean { // `catch` gives `unknown`, and the guard is what narrows it. Reading // `caught.retryable` without it does not compile. diff --git a/types/header.d.ts b/types/header.d.ts index 6777f98..dfd8bb5 100644 --- a/types/header.d.ts +++ b/types/header.d.ts @@ -151,6 +151,76 @@ export type ZuAppendValue = | ZuDuration | ZuTemporalValue +/** + * One value of a registered frame's column, when the column is written + * as a plain array. + * + * The same values an appender takes, without the bytes: a column of + * BYTES is a column no statement can read back yet, so registering one + * would be naming data the caller cannot get at. There is no `null` + * either, for the reason there is none in a row of an appender. + */ +export type ZuFrameValue = + | boolean + | number + | bigint + | string + | ZuDate + | ZuTime + | ZuTimestamp + | ZuDuration + | ZuTemporalValue + +/** + * One column of a registered frame. + * + * A typed array is the shape that costs nothing: the engine reads it + * where it lies and no byte of it is copied. A plain array is read into + * buffers of this client's own, because an array holds values of the + * runtime rather than numbers, and its first value settles what the + * column holds. + */ +export type ZuFrameColumn = + | Int8Array + | Uint8Array + | Uint8ClampedArray + | Int16Array + | Uint16Array + | Int32Array + | Uint32Array + | Float32Array + | Float64Array + | BigInt64Array + | BigUint64Array + | readonly ZuFrameValue[] + +/** + * An Arrow table or record batch, described by its shape rather than by + * its class. + * + * Structural on purpose. `apache-arrow` is not a dependency of this + * client and should not have to be: recognizing a table by the two + * things every version of it has means a caller's copy of that library + * and this client's are never two copies of one package disagreeing + * about `instanceof`, and it means anything else that speaks the same + * shape works too. + */ +export interface ZuArrowTable { + readonly schema: { readonly fields: readonly { readonly name: string }[] } + getChildAt(index: number): unknown +} + +/** + * Columns the caller already holds, ready to be registered under a name. + * + * An Arrow table, or an object of column name to values. Both are read + * where they lie wherever there is one run of bytes to read: the two + * cases that copy are an Arrow column that arrived in several chunks, + * which is concatenated once, and a plain JavaScript array, which was + * never a column of numbers to begin with. + */ +export type ZuFrame = ZuArrowTable | Record + /** * A walk through the graph: nodes and edges, alternating, a node at * each end.