From 1582b996dd6b01b49b684061754db4ce74bc960a Mon Sep 17 00:00:00 2001 From: Tam Nguyen Duc <1218621+tamnd@users.noreply.github.com> Date: Wed, 19 Aug 2026 14:50:59 +0700 Subject: [PATCH] Loading rows into a table that is already there INSERT is the wrong shape for a load. Every row is parsed, bound, planned and committed, and the commit is the expensive part, so a million rows is a million commits and the load is spent on durability nobody asked for. An appender is the right shape: rows go into columns in memory and a flush turns the whole buffer into one commit. On this machine a hundred thousand rows take 13 seconds one INSERT at a time, 1.3 seconds five hundred to an INSERT, and 136 milliseconds through the appender. appendRow is the one synchronous call in this client, and it is synchronous because it reaches nothing: it converts the values in front of it and pushes them onto a vector, with no file and no lock at the end of it. Making it a promise would put a microtask between the loop and a memcpy and allocate a million promises to describe work that had already finished. Being synchronous it throws rather than rejecting, with the same ZuUsageError shape everything else here rejects with, so isZuError recognizes it either way. What is buffered is typed from the table's own columns, read when the appender opened, so a value that does not belong in a column is refused by the call that appended it rather than a million rows later by the flush that would have carried it. A refused row is a row that never happened: the columns that did take a value give it back. In a batch the refusal names which row it was and keeps the ones before it, since nothing here is a transaction until the flush. The counts sit outside the buffers' lock, for the reason open and inTransaction do on a connection: asking how many rows are buffered should not queue behind the commit that is writing them. A flush issued while one is running is refused rather than queued, and so is an append, because waiting for either would be the event loop waiting for a write to disk. await using flushes, which is the opposite of what a transaction's disposal does here. The two differ because the question differs: a transaction that leaves its scope unfinished is a unit of work nobody completed, and a buffer that leaves its scope unwritten is a loader that read a million rows and threw them away. discard() is there for the caller who meant that, and the Python client answers the same way. A rel table has no property columns, so a row of one is the two ends of an edge as offsets into the tables it runs between, and the flush checks 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, and the frame it leaves behind is refused again by every writer that opens the database afterwards. Milestone: tamnd/zu#169, item 3 --- README.md | 24 +- bench/append.mjs | 185 ++++++++++ binding.cjs | 1 + binding.d.cts | 147 ++++++++ etc/zudb.api.md | 28 ++ package.json | 1 + src/append.rs | 816 +++++++++++++++++++++++++++++++++++++++++ src/buffer.rs | 366 ++++++++++++++++++ src/conn.rs | 33 ++ src/lib.rs | 2 + src/value.rs | 2 +- test/appender.test.mjs | 545 +++++++++++++++++++++++++++ test/exports.test.mjs | 1 + test/types/cjs.cts | 17 + test/types/esm.mts | 24 ++ types/header.d.ts | 28 ++ zudb.cjs | 1 + zudb.d.cts | 11 + zudb.mjs | 1 + 19 files changed, 2231 insertions(+), 2 deletions(-) create mode 100644 bench/append.mjs create mode 100644 src/append.rs create mode 100644 src/buffer.rs create mode 100644 test/appender.test.mjs diff --git a/README.md b/README.md index c7940d8..289bb0a 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. 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. 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. @@ -102,6 +102,28 @@ The statements are still the connection's, because the span is the connection's A block that ends well and forgets to commit loses its work, which is a loud kind of wrong and shows up the first time the code runs. The alternative was a block that failed and kept half of what it did, which is a quiet kind and shows up in production. Committing or rolling back twice is refused as a `ZuUsageError` rather than ignored, since the statements after the first end belong to no transaction of yours. Leaving the block of a transaction whose connection has already been closed says nothing, because a closed connection took the unwritten span with it and there is nothing left to undo. +## Loading a lot of rows + +`INSERT` is the wrong shape for loading. Every row is parsed, bound, planned and committed, and the commit is the expensive part, so a million rows is a million commits and the load is spent on durability nobody asked for. An appender is the right shape: rows go into columns in memory, and a flush turns the whole buffer into one commit. + +```ts +await using rows = await conn.appender("Person"); +for (const [id, name] of people) rows.appendRow([id, name]); +await rows.flush(); +``` + +A row is every column of the table, in the order the table declares them, and a column is a position rather than a name. Naming the columns per row would cost a lookup per value on the one path where per-value cost is the whole story, and a loader knows its own column order. `appendRows` takes an array of them, which is one check for the batch rather than one per row. + +`appendRow` is the one synchronous call in this client, and it is synchronous because it reaches nothing. It converts the values in front of it and pushes them onto a vector, bounded by the width of one row, with no file and no lock at the end of it. Making it a promise would put a microtask between the loop and a memcpy and allocate a million promises to describe work that had already finished. Being synchronous it throws rather than rejecting, with the same `ZuUsageError` everything else here rejects with, so `isZuError(caught)` recognizes it either way. Everything that touches the file, which is `flush`, `close` and the disposal, is a promise like the rest of the client. + +What is buffered is typed from the table's own columns, read when the appender opened, so a value that does not belong in a column is refused by the call that appended it rather than a million rows later by the flush that would have carried it. The message names the column and the position: `value 0 of this row is a string and column 'id' of 'Person' holds whole numbers`. A refused row is a row that never happened, so the columns that did take a value give it back and the appender is usable as soon as the caller has fixed the row. In a batch the refusal says which row it was and keeps the ones before it, since nothing here is a transaction until the flush. + +`await using rows` flushes. That is the opposite of what a transaction's disposal does, and the two differ because the question differs: a transaction that leaves its scope unfinished is a unit of work nobody completed, and a buffer that leaves its scope unwritten is a loader that read a million rows and threw them away. `discard()` is there for the caller who meant exactly that, and it answers how many rows it dropped. + +Two more things are worth knowing before a load. A flush issued while one is still running is refused rather than queued, and so is an append, because waiting for either would be the event loop waiting for a write to disk: `await` the flush. And rows an appender writes are not part of an open transaction, since it writes through the file rather than through the session, so a `ROLLBACK` after a flush does not take them back. A load and a transaction are two different things to reach for. + +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. + ## 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: diff --git a/bench/append.mjs b/bench/append.mjs new file mode 100644 index 0000000..326929c --- /dev/null +++ b/bench/append.mjs @@ -0,0 +1,185 @@ +// What loading rows costs, and what it costs to load them the other way. +// +// The appender exists because `INSERT` is the wrong shape for a load: +// every row is parsed, bound, planned and committed, and the commit is +// the expensive part. So the first number here is the one to read the +// rest against, and the ratio between it and the last is the whole +// argument for the class. +// +// The other thing being measured is the boundary itself. `appendRow` is +// the one synchronous call in this client, and what it does is convert a +// value per column and push it onto a vector, so its number should be +// tens of nanoseconds rather than hundreds. When it is not, something on +// the way in started allocating. +// +// Run it against a release build, for the reason bench/query.mjs gives. +// +// npm run build && npm run bench:append + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { connect } from 'zudb' + +const ROWS = Number(process.env.ZU_BENCH_ROWS ?? 100_000) +const REPEATS = Number(process.env.ZU_BENCH_REPEATS ?? 5) +// How many rows one `INSERT` carries in the batched case. Enough to +// amortize the commit and small enough that the statement it builds is +// one a parser can still be asked to read. +const BATCH = 500 +// One statement per row is slow enough that measuring the whole table +// that way would dominate the run, so that case is measured over a +// smaller table and reported per row like the others. +const SLOW = Number(process.env.ZU_BENCH_SLOW_ROWS ?? 2_000) + +const dir = await mkdtemp(join(tmpdir(), 'zu-bench-append-')) + +/// A database with the table declared and nothing else in it. +/// +/// A fresh one per case, because a load into a table that already holds +/// a million rows is not the same load as one into a table that holds +/// two, and what is being compared is the way in rather than the size of +/// what is already there. +async function blank() { + const path = join(dir, `bench-${counter++}.zu1`) + const conn = await connect(path) + // The declaring insert is written with literals, because that is what + // tells the engine what each column holds. + await conn.exec("INSERT (p:person {id: 0, name: 'n0'})") + return conn +} + +let counter = 0 + +/// 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(rows, run) { + await run(await blank()) + let best = Infinity + for (let round = 0; round < REPEATS; round++) { + const conn = await blank() + const started = performance.now() + await run(conn) + best = Math.min(best, performance.now() - started) + conn.close() + } + return best +} + +const cases = [ + { + // One statement per row, which is a parse, a plan and a commit per + // row. The number every other line here is asking to be read + // against. + name: 'INSERT, one row each', + rows: SLOW, + run: async (conn) => { + for (let ix = 1; ix <= SLOW; ix++) { + await conn.exec(`INSERT (p:person {id: ${ix}, name: 'n${ix}'})`) + } + }, + }, + { + // The same statement carrying five hundred rows, which is what a + // loader without an appender ends up writing by hand. + name: 'INSERT, 500 rows each', + rows: ROWS, + run: async (conn) => { + for (let start = 1; start <= ROWS; start += BATCH) { + const parts = [] + for (let ix = start; ix < Math.min(start + BATCH, ROWS + 1); ix++) { + parts.push(`(p${ix}:person {id: ${ix}, name: 'n${ix}'})`) + } + await conn.exec(`INSERT ${parts.join(', ')}`) + } + }, + }, + { + // Every row buffered and one commit at the end, which is the shape + // the class is for. + name: 'appender, one flush', + rows: ROWS, + run: async (conn) => { + const rows = await conn.appender('person') + for (let ix = 1; ix <= ROWS; ix++) rows.appendRow([BigInt(ix), `n${ix}`]) + await rows.close() + }, + }, + { + // The same rows with a flush every ten thousand, which is what a + // loader that cannot hold the whole file in memory writes. The + // difference from the line above is what the extra commits cost. + name: 'appender, flush every 10k', + rows: ROWS, + run: async (conn) => { + const rows = await conn.appender('person') + for (let ix = 1; ix <= ROWS; ix++) { + rows.appendRow([BigInt(ix), `n${ix}`]) + if (ix % 10_000 === 0) await rows.flush() + } + await rows.close() + }, + }, + { + // Rows handed over in arrays of a hundred, which is one boundary + // crossing per hundred rows rather than one per row. What it saves + // is the call and the checks around it, and what it costs is the + // arrays. + name: 'appender, appendRows(100)', + rows: ROWS, + run: async (conn) => { + const rows = await conn.appender('person') + let batch = [] + for (let ix = 1; ix <= ROWS; ix++) { + batch.push([BigInt(ix), `n${ix}`]) + if (batch.length === 100) { + rows.appendRows(batch) + batch = [] + } + } + if (batch.length) rows.appendRows(batch) + await rows.close() + }, + }, + { + // A whole number rather than a `bigint`, which is what a caller + // writing row literals writes. It costs a check that the number is + // whole and saves whatever the runtime charges for a `bigint`. + name: 'appender, number ids', + rows: ROWS, + run: async (conn) => { + const rows = await conn.appender('person') + for (let ix = 1; ix <= ROWS; ix++) rows.appendRow([ix, `n${ix}`]) + await rows.close() + }, + }, + { + // The buffers on their own, with the commit taken out of the + // measurement: everything is appended and then thrown away. What is + // left is the conversion and the push, which is what `appendRow` + // does and all it does. + name: 'appender, buffered only', + rows: ROWS, + run: async (conn) => { + const rows = await conn.appender('person') + for (let ix = 1; ix <= ROWS; ix++) rows.appendRow([BigInt(ix), `n${ix}`]) + rows.discard() + }, + }, +] + +console.log(`${ROWS} rows, fastest of ${REPEATS}`) +for (const { name, rows, run } of cases) { + const ms = await time(rows, run) + const each = (ms * 1e6) / rows + const scale = rows === ROWS ? '' : ` (over ${rows})` + console.log( + `${name.padEnd(26)} ${ms.toFixed(2).padStart(9)} ms ${Math.round(each).toString().padStart(8)} ns/row${scale}`, + ) +} + +await rm(dir, { recursive: true, force: true }) diff --git a/binding.cjs b/binding.cjs index 3fb4b67..2929d71 100644 --- a/binding.cjs +++ b/binding.cjs @@ -700,6 +700,7 @@ if (!nativeBinding) { } module.exports = nativeBinding +module.exports.Appender = nativeBinding.Appender module.exports.Connection = nativeBinding.Connection module.exports.Transaction = nativeBinding.Transaction module.exports.ZuCursor = nativeBinding.ZuCursor diff --git a/binding.d.cts b/binding.d.cts index b2df584..aaa07ef 100644 --- a/binding.d.cts +++ b/binding.d.cts @@ -123,6 +123,34 @@ export type ZuParam = | ZuParam[] | { [field: string]: ZuParam } +/** + * A value an appender takes, which is narrower than what a statement + * takes. + * + * A column of an appender has one type, read from the table when the + * appender opened, and every value in it is that type. So there is no + * `null` here: a column that holds nulls cannot be appended to at all + * and the appender says so when it opens, and `undefined` in a row is a + * value the caller forgot rather than a null they meant. There are no + * lists and no objects either, because a property column holds a scalar. + * + * BYTES is a `Uint8Array`, which is the one type here that no statement + * parameter can be, and INT64 is a `bigint` or a whole `number` below + * 2^53. A number past that is refused rather than rounded, because past + * 2^53 a number no longer names one integer. + */ +export type ZuAppendValue = + | boolean + | number + | bigint + | string + | Uint8Array + | ZuDate + | ZuTime + | ZuTimestamp + | ZuDuration + | ZuTemporalValue + /** * A walk through the graph: nodes and edges, alternating, a node at * each end. @@ -333,6 +361,105 @@ export interface ZuError extends Error { /** The whole line `column` indexes into, for underlining it. */ readonly excerpt?: string } +/** + * Rows on their way into a table, buffered until they are flushed. + * + * Take one with `Connection.appender`, append rows to it, and close + * it. What is buffered is columnar and typed from the table's own + * columns, read when the appender opened, so a value that does not + * belong in a column is refused by the call that appended it rather + * than at the flush that would have carried it, and the message names + * the column it did not fit. + */ +export declare class Appender { + /** The table these rows are going into. */ + get table(): string + /** Rows buffered and not yet written. */ + get buffered(): number + /** Rows this appender has committed, across every flush. */ + get committed(): number + /** Whether this appender has been closed. */ + get closed(): boolean + /** + * Appends one row, which is one value per column of the table, in + * the order the table declares them. + * + * Synchronous, and the only synchronous call in this client: the + * values go into memory and nothing else happens, so this is a + * conversion and a push per column. Being synchronous it throws + * rather than rejecting, with the same `ZuUsageError` every other + * refusal here carries. + * + * A row of the wrong width, or with a value that does not fit the + * column, is refused with nothing of it kept, so the appender is + * still usable once the caller has fixed the row. + */ + appendRow(row: readonly ZuAppendValue[]): void + /** + * Appends every row of an array of rows. + * + * The same thing in a loop, and worth a call of its own because it + * is one check and one lock for the batch rather than one per row. + * A row that is refused stops the call where it was refused and the + * rows before it stay buffered: nothing here is a transaction until + * the flush, and throwing away work the caller can keep would not + * make it one. What it answers is how many rows went in, which is + * where a caller who caught the refusal starts again. + */ + appendRows(rows: readonly (readonly ZuAppendValue[])[]): number + /** + * Writes every buffered row and makes it readable, and answers how + * many rows this appender has committed in all. + * + * One commit, whatever the buffer holds: the values are sealed into + * the file, one frame naming them is synced to the log, and the + * fold that follows puts them where every query looks. On return + * the buffer is empty and the rows are there. A flush with nothing + * buffered touches no file, so a loader can flush on a timer + * without writing empty commits. + * + * A flush that fails keeps its rows, so that what did not go in is + * still there to be looked at and tried again. + */ + flush(): Promise + /** + * Flushes what is left and answers how many rows this appender + * committed in all. + * + * Closing twice is not an error and writes nothing the second + * time, because an `await using` that closed early would otherwise + * fail on the way out. + */ + close(): Promise + /** + * The close `await using` calls, which is the intended way to + * scope an appender. + * + * It flushes, whether the block ended well or badly, which is the + * opposite of what the disposal of a transaction here does and is + * the same answer the Python client gives. The two differ because + * the question differs: a transaction that leaves its scope + * unfinished is a unit of work nobody completed, and a buffer that + * leaves its scope unwritten is a loader that read a million rows + * and threw them away. A caller who wants the rows gone writes + * `discard()` and gets exactly that. + * + * It is also reachable as `Symbol.asyncDispose`, which is what + * `await using` actually looks for and which [`wire_disposal`] puts + * on every appender as it is made. + */ + dispose(): Promise + /** + * Throws away what is buffered and answers how many rows that was. + * + * The way out of a load that went wrong halfway. A caller who has + * noticed that the rows are wrong wants them gone, and closing + * would write them. Rows an earlier flush committed are committed, + * and this does not reach them. + */ + discard(): number +} + /** * One connection to one database. * @@ -384,6 +511,26 @@ export declare class Connection { * commit half of the work of a block that failed. */ transaction(options?: ZuTransactionOptions | null): Promise + /** + * Opens an appender on `table` and hands it back. + * + * The bulk-load path. A load written as statements pays a commit + * per row, and an appender pays one per flush, which is the whole + * difference between loading a million rows in an afternoon and + * loading them in a minute. + * + * ```js + * await using rows = await conn.appender('person') + * for (const [id, name] of people) rows.appendRow([id, name]) + * await rows.flush() + * ``` + * + * The table has to exist, and its columns are read here, so a + * table nothing declares and a column of a type the ingest cannot + * carry are both refused at this call rather than at the flush a + * million rows later. + */ + appender(table: string): Promise /** * Runs one statement and gives back its rows. * diff --git a/etc/zudb.api.md b/etc/zudb.api.md index 1cb7684..0eb63bd 100644 --- a/etc/zudb.api.md +++ b/etc/zudb.api.md @@ -7,11 +7,26 @@ // @public export function abiVersion(): string +// @public +export class Appender { + appendRow(row: readonly ZuAppendValue[]): void + appendRows(rows: readonly (readonly ZuAppendValue[])[]): number + get buffered(): number + close(): Promise + get closed(): boolean + get committed(): number + discard(): number + dispose(): Promise + flush(): Promise + get table(): string +} + // @public export function connect(path: string, options?: ConnectOptions | undefined | null): Promise // @public export class Connection { + appender(table: string): Promise close(): void cursor(statement: string, params?: Record | null, options?: ZuStreamOptions | null): ZuCursor dispose(): Promise @@ -48,6 +63,19 @@ export class Transaction { // @public export function version(): string +// @public +export type ZuAppendValue = +| boolean +| number +| bigint +| string +| Uint8Array +| ZuDate +| ZuTime +| ZuTimestamp +| ZuDuration +| ZuTemporalValue + // @public export interface ZuBatch> extends Array { // (undocumented) diff --git a/package.json b/package.json index 9bd046a..abc69f1 100644 --- a/package.json +++ b/package.json @@ -71,6 +71,7 @@ "check:api:update": "api-extractor run --local", "reference": "node tools/reference.mjs reference", "bench": "node bench/query.mjs", + "bench:append": "node bench/append.mjs", "bench:temporal": "node --harmony-temporal bench/query.mjs" }, "devDependencies": { diff --git a/src/append.rs b/src/append.rs new file mode 100644 index 0000000..8d1ba14 --- /dev/null +++ b/src/append.rs @@ -0,0 +1,816 @@ +//! Appending rows to a table that already exists. +//! +//! `INSERT` is the wrong shape for loading data. Every row is parsed, +//! bound, planned and committed, and the commit is the expensive part, +//! so a million rows is a million commits and the load is dominated by +//! durability work nobody asked for. An appender is the right shape: +//! rows go into per-column buffers in memory, and a flush turns the +//! whole buffer into one commit. +//! +//! ```js +//! await using rows = await conn.appender('person') +//! for (const [id, name] of people) rows.appendRow([id, name]) +//! await rows.flush() +//! ``` +//! +//! A row is every column of the table, in the order the table declares +//! them, and a column is a position rather than a name: naming the +//! columns per row would cost a lookup per value on the one path where +//! per-value cost is the whole story, and a loader knows its own column +//! order. +//! +//! ## The one synchronous call in this client +//! +//! `appendRow` is not a promise. Everything else here is, because +//! everything else reaches the database and a native call that reaches +//! a database on the event loop is a production incident. An append +//! reaches nothing: it converts the values in front of it and pushes +//! them onto a vector, which is bounded by the width of one row and +//! cannot wait on a file, a lock the engine holds, or another thread. +//! Making it a promise would put a microtask between the loop and a +//! memcpy, and a million-row load would allocate a million promises to +//! describe work that had already finished. +//! +//! So it is synchronous, and being synchronous it throws rather than +//! rejecting. The exception is the same `ZuUsageError` every other +//! refusal here is, so `isZuError(caught)` recognizes it in a `catch` +//! either way, and everything that touches the file, which is `flush`, +//! `close` and the disposal, is a promise like the rest of the client. +//! +//! ## Why the buffers are here and not in the engine +//! +//! The engine's appender borrows the connection for as long as it +//! lives, which is a promise a JavaScript object cannot make, so this +//! one buffers here and opens an engine appender for the length of a +//! flush. That is a catalog read per flush, against a commit and a fold +//! that cost time proportional to the table, so it is not where a load +//! spends its time. What it buys is an appender that can be held in a +//! variable, passed to a function and closed by an `await using`. +//! +//! Rows an appender writes are not part of an open transaction. It +//! writes through the file rather than through the session, so a +//! `ROLLBACK` after a flush does not take them back. That is the +//! engine's shape today rather than a decision made here, and it is the +//! reason a load and a transaction are two different things to reach +//! for. + +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, MutexGuard}; + +use napi::bindgen_prelude::*; +use napi::{Env, ScopedTask}; +use napi_derive::napi; +use zudb::Field; +use zudb::zu1::catalog::Catalog; + +use crate::buffer::{Column, Mismatch, named}; +use crate::conn::{CLOSED, Failure, POISONED, failed, wire_disposal, with}; +use crate::error::usage; + +/// Rows on their way into a table, buffered until they are flushed. +/// +/// Take one with `Connection.appender`, append rows to it, and close +/// it. What is buffered is columnar and typed from the table's own +/// columns, read when the appender opened, so a value that does not +/// belong in a column is refused by the call that appended it rather +/// than at the flush that would have carried it, and the message names +/// the column it did not fit. +#[napi] +pub struct Appender { + /// The same three handles every statement on this connection uses, + /// rather than a reference to the JavaScript object, so an appender + /// whose `Connection` was collected still has somewhere to write. + inner: Arc>>, + alive: Arc, + in_txn: Arc, + table: String, + state: Arc, + /// Whether a flush is in flight. + /// + /// It is set on the thread that owns the runtime, before the task + /// is handed back, and cleared on that same thread when the task + /// ends. So a call that finds it false knows no flush can start + /// before its own turn is over, which is what lets an append take + /// the buffers' lock without ever waiting for one: the only other + /// holder of that lock is a flush, and there is none. + busy: Arc, +} + +/// What is buffered, and how much of it has gone in. +/// +/// The three counts sit outside the lock rather than inside it, for the +/// reason `open` and `inTransaction` do on a connection: asking how many +/// rows are buffered should not queue behind the commit that is writing +/// them, and a getter that could wait is a getter that can stop the +/// event loop. +struct State { + /// One buffer per column of the table, in the order the table + /// declares them, built when the appender opened. A flush empties + /// these and keeps them, since the next batch is the same shape as + /// the last. + cols: Mutex>, + /// Rows buffered and not yet written, kept beside the columns so + /// that a table with no columns can still answer for itself. + buffered: AtomicU64, + /// Rows this appender has committed, across every flush. + committed: AtomicU64, + open: AtomicBool, +} + +/// One column of the table, and what has been buffered for it. +pub struct Buffer { + name: String, + values: Column, + /// The node table whose rows this column names, for the two columns + /// of a rel table and for nothing else: a row of one is an offset + /// into the table the edge runs from and an offset into the table + /// it runs to. A negative offset is no row of anything and is + /// refused where it was appended; whether the row is there at all + /// is a question only the flush can answer, since the table may be + /// being appended to at the same time. + ends: Option, +} + +#[napi] +impl Appender { + /// The table these rows are going into. + #[napi(getter)] + pub fn table(&self) -> String { + self.table.clone() + } + + /// Rows buffered and not yet written. + #[napi(getter)] + pub fn buffered(&self) -> f64 { + self.state.buffered.load(Ordering::Acquire) as f64 + } + + /// Rows this appender has committed, across every flush. + #[napi(getter)] + pub fn committed(&self) -> f64 { + self.state.committed.load(Ordering::Acquire) as f64 + } + + /// Whether this appender has been closed. + #[napi(getter)] + pub fn closed(&self) -> bool { + !self.state.open.load(Ordering::Acquire) + } + + /// Appends one row, which is one value per column of the table, in + /// the order the table declares them. + /// + /// Synchronous, and the only synchronous call in this client: the + /// values go into memory and nothing else happens, so this is a + /// conversion and a push per column. Being synchronous it throws + /// rather than rejecting, with the same `ZuUsageError` every other + /// refusal here carries. + /// + /// A row of the wrong width, or with a value that does not fit the + /// column, is refused with nothing of it kept, so the appender is + /// still usable once the caller has fixed the row. + #[napi(ts_args_type = "row: readonly ZuAppendValue[]")] + pub fn append_row(&self, env: &Env, row: Unknown<'_>) -> Result<()> { + let mut cols = self.writable(env)?; + self.state + .append(env, &mut cols, &self.table, row) + .map_err(|why| raised(env, why, "")) + } + + /// Appends every row of an array of rows. + /// + /// The same thing in a loop, and worth a call of its own because it + /// is one check and one lock for the batch rather than one per row. + /// A row that is refused stops the call where it was refused and the + /// rows before it stay buffered: nothing here is a transaction until + /// the flush, and throwing away work the caller can keep would not + /// make it one. What it answers is how many rows went in, which is + /// where a caller who caught the refusal starts again. + #[napi(ts_args_type = "rows: readonly (readonly ZuAppendValue[])[]")] + pub fn append_rows(&self, env: &Env, rows: Unknown<'_>) -> Result { + let mut cols = self.writable(env)?; + if !rows.is_array()? { + return Err(usage( + env, + format!( + "the rows are {}, and rows are an array of arrays, one value per column", + named(&rows) + ), + )); + } + let rows = Object::from_unknown(rows)?; + let len = rows.get_array_length()?; + for ix in 0..len { + let row: Unknown<'_> = rows.get_element(ix)?; + if let Err(why) = self.state.append(env, &mut cols, &self.table, row) { + // The index is what a caller needs and the only thing + // this call knows that the row itself does not, since + // the row that failed is somewhere inside an array they + // handed over whole. + return Err(raised(env, why, &format!("row {ix} of these: "))); + } + } + Ok(len as f64) + } + + /// Writes every buffered row and makes it readable, and answers how + /// many rows this appender has committed in all. + /// + /// One commit, whatever the buffer holds: the values are sealed into + /// the file, one frame naming them is synced to the log, and the + /// fold that follows puts them where every query looks. On return + /// the buffer is empty and the rows are there. A flush with nothing + /// buffered touches no file, so a loader can flush on a timer + /// without writing empty commits. + /// + /// A flush that fails keeps its rows, so that what did not go in is + /// still there to be looked at and tried again. + #[napi(ts_return_type = "Promise")] + pub fn flush(&self) -> AsyncTask { + self.write(false) + } + + /// Flushes what is left and answers how many rows this appender + /// committed in all. + /// + /// Closing twice is not an error and writes nothing the second + /// time, because an `await using` that closed early would otherwise + /// fail on the way out. + #[napi(ts_return_type = "Promise")] + pub fn close(&self) -> AsyncTask { + self.write(true) + } + + /// The close `await using` calls, which is the intended way to + /// scope an appender. + /// + /// It flushes, whether the block ended well or badly, which is the + /// opposite of what the disposal of a transaction here does and is + /// the same answer the Python client gives. The two differ because + /// the question differs: a transaction that leaves its scope + /// unfinished is a unit of work nobody completed, and a buffer that + /// leaves its scope unwritten is a loader that read a million rows + /// and threw them away. A caller who wants the rows gone writes + /// `discard()` and gets exactly that. + /// + /// It is also reachable as `Symbol.asyncDispose`, which is what + /// `await using` actually looks for and which [`wire_disposal`] puts + /// on every appender as it is made. + #[napi(ts_return_type = "Promise")] + pub fn dispose(&self) -> AsyncTask { + self.write(true) + } + + /// Throws away what is buffered and answers how many rows that was. + /// + /// The way out of a load that went wrong halfway. A caller who has + /// noticed that the rows are wrong wants them gone, and closing + /// would write them. Rows an earlier flush committed are committed, + /// and this does not reach them. + #[napi] + pub fn discard(&self, env: &Env) -> Result { + let mut cols = self.writable(env)?; + Ok(self.state.empty(&mut cols) as f64) + } + + /// The task a flush runs as, whether or not it is going to work. + fn write(&self, closing: bool) -> AsyncTask { + AsyncTask::new(self.flushing(closing)) + } + + /// The flush itself, built where the claim on the appender is taken. + fn flushing(&self, closing: bool) -> FlushTask { + // Claimed here, on the thread that owns the runtime, so that a + // second flush issued before the first has answered is refused + // rather than queued behind it on a threadpool thread. Two + // commits of the same buffer would be two writes whose order + // nobody chose, and a caller who wanted them overlapped wanted + // two appenders. + let refused = match self.busy.swap(true, Ordering::AcqRel) { + true => Some(BUSY.to_string()), + false => None, + }; + FlushTask { + inner: Arc::clone(&self.inner), + alive: Arc::clone(&self.alive), + in_txn: Arc::clone(&self.in_txn), + state: Arc::clone(&self.state), + // Given back only by the task that took it, so the one that + // was refused does not release the one that is running. + busy: match refused { + Some(_) => None, + None => Some(Arc::clone(&self.busy)), + }, + table: self.table.clone(), + closing, + refused, + } + } + + /// The buffers, for a call that writes to them. + /// + /// The lock is never waited for. A flush is the only other thing + /// that takes it, a flush is in flight exactly while `busy` is set, + /// and `busy` is set and cleared on this thread, so a call that gets + /// past the first line has the lock to itself. That is the whole + /// reason for refusing rather than queueing: an append that waited + /// for a flush would be an event loop waiting for a write to disk. + /// + /// The connection is checked as well as the appender, because a row + /// appended through a closed connection has nowhere to go and the + /// buffer is the only thing that would take it. Left to the flush, + /// the same call would be refused or not depending on whether the + /// batch happened to fill, which is a rule nobody can hold in their + /// head. It is refused here instead, at the call that made the + /// mistake, whatever the buffer is holding. + fn writable(&self, env: &Env) -> Result>> { + if self.busy.load(Ordering::Acquire) { + return Err(usage(env, BUSY)); + } + if !self.state.open.load(Ordering::Acquire) { + return Err(usage(env, FINISHED)); + } + if !self.alive.load(Ordering::Acquire) { + return Err(usage(env, CLOSED)); + } + self.state.cols.lock().map_err(|_| usage(env, POISONED)) + } +} + +impl State { + /// One row into the buffers, or nothing at all. + fn append( + &self, + env: &Env, + cols: &mut [Buffer], + table: &str, + row: Unknown<'_>, + ) -> std::result::Result<(), Refused> { + // One call rather than a type and then a kind, because an array + // is an object and asking twice is a boundary crossing per row on + // the one path where crossings are the cost. + if !row.is_array()? { + return Err(Refused::Row(format!( + "this row is {}, and a row is an array of one value per column of '{table}': {}", + named(&row), + names(cols) + ))); + } + let row = Object::from_unknown(row)?; + let len = row.get_array_length()?; + let width = cols.len() as u32; + if len != width { + return Err(Refused::Row(format!( + "this row carries {len} value{} and '{table}' takes {width}: {}", + match len { + 1 => "", + _ => "s", + }, + names(cols) + ))); + } + for at in 0..len { + let value: Unknown<'_> = row.get_element(at)?; + if let Err(why) = cols[at as usize].take(env, &value, table, at) { + return Err(refuse(cols, at, why)); + } + } + self.buffered.fetch_add(1, Ordering::AcqRel); + Ok(()) + } + + /// The write itself, off the event loop. + fn write( + &self, + cols: &mut [Buffer], + inner: &Mutex>, + alive: &AtomicBool, + in_txn: &AtomicBool, + table: &str, + ) -> std::result::Result { + let rows = self.buffered.load(Ordering::Acquire); + if rows == 0 { + return Ok(self.committed.load(Ordering::Acquire)); + } + with(inner, alive, in_txn, |conn| { + self.reachable(cols, conn, rows)?; + let mut appender = conn.appender(table)?; + // One vector, refilled per row rather than allocated per + // row, which over a million rows is one allocation rather + // than a million. The fields borrow the buffers, which is + // what keeps a string column to one copy on the way in and + // one on the way out rather than three. + let mut row: Vec> = Vec::with_capacity(cols.len()); + for at in 0..rows as usize { + row.clear(); + row.extend(cols.iter().map(|column| column.values.field(at))); + appender.append_row(&row[..]).map_err(|err| { + // The engine reports the value and the column; + // which row of the batch it was is the part only + // this side knows, and it is the part that says + // where to look. + Failure::Usage(format!("row {at} of this batch: {err}")) + })?; + } + appender.close()?; + Ok(()) + })?; + self.empty(cols); + Ok(self.committed.fetch_add(rows, Ordering::AcqRel) + rows) + } + + /// Every edge joins two rows that are there, checked against the row + /// counts as they stand at the flush. + /// + /// This is the flush's own check and not the engine's, because the + /// engine's comes too late: an edge to a row that is not there is + /// refused when the write is folded into the graph, which is after + /// the write is durable, and the frame it leaves behind is refused + /// again by every writer that opens the database afterwards. Caught + /// here, the batch is refused and the file is untouched. + /// + /// The counts are read at the flush and not when the appender + /// opened, because the rows a later edge names may be written by an + /// earlier flush of another appender on the same connection, and an + /// edge to a row that arrived in the meantime is a good edge. + fn reachable( + &self, + cols: &[Buffer], + conn: &mut zudb::Connection, + rows: u64, + ) -> std::result::Result<(), Failure> { + if cols.iter().all(|column| column.ends.is_none()) { + return Ok(()); + } + let catalog = catalog(conn)?; + for column in cols { + let Some(end) = column.ends else { continue }; + let Some(node) = catalog.node_by_id(end) else { + continue; + }; + for at in 0..rows as usize { + let Field::Int(offset) = column.values.field(at) else { + continue; + }; + if offset as u64 >= node.node_count { + return Err(Failure::Usage(format!( + "row {at} of this batch joins row {offset} of '{}', which has {} rows \ + in it, so the rows an edge joins have to be written before the edge is", + node.name, node.node_count + ))); + } + } + } + Ok(()) + } + + /// Empties the buffers and answers how many rows were dropped, which + /// is what `discard` reports and what a flush has already written. + fn empty(&self, cols: &mut [Buffer]) -> u64 { + cols.iter_mut().for_each(|column| column.values.clear()); + self.buffered.swap(0, Ordering::AcqRel) + } +} + +/// The columns of the table, named, for a message about a row that is the +/// wrong shape. A caller who miscounted wants to see what the count was +/// supposed to be made of. +fn names(cols: &[Buffer]) -> String { + cols.iter() + .map(|column| column.name.as_str()) + .collect::>() + .join(", ") +} + +/// Takes back the values a refused row managed to write, so that a +/// refused row is a row that never happened rather than half of one +/// nobody can find. A ragged buffer would be refused by the ingest at the +/// flush, a long way from the row that caused it. +fn refuse(cols: &mut [Buffer], written: u32, err: Refused) -> Refused { + for column in cols.iter_mut().take(written as usize) { + column.values.pop(); + } + err +} + +impl Buffer { + /// One value into this column, or the reason it does not go there. + /// + /// The column's own name is in the message, and its position too, + /// because a row is written by position and read by name and a + /// caller who has them the wrong way round needs both to see it. + fn take( + &mut self, + env: &Env, + value: &Unknown<'_>, + table: &str, + at: u32, + ) -> std::result::Result<(), Refused> { + self.values.push(env, *value).map_err(|why| match why { + Mismatch::Wanted(holds) => Refused::Row(format!( + "value {at} of this row is {} and column '{}' of '{table}' holds {holds}", + named(value), + self.name + )), + Mismatch::Says(reason) => Refused::Row(format!( + "value {at} of this row does not go in column '{}' of '{table}': {reason}", + self.name + )), + Mismatch::Boundary(err) => Refused::Boundary(err), + })?; + // Checked after the value is read rather than before, because + // what makes an offset negative is the number it turned into and + // a `bigint` and a `number` are two different ways of arriving at + // the same one. Taken back here, so the column is as it was. + if self.ends.is_some() + && let Column::Int(offsets) = &self.values + && let Some(&offset) = offsets.last() + && offset < 0 + { + self.values.pop(); + return Err(Refused::Row(format!( + "value {at} of this row is {offset}, and column '{}' of '{table}' holds row \ + offsets, which count from zero", + self.name + ))); + } + Ok(()) + } +} + +/// A row that did not go in, and why. +/// +/// Words rather than an exception, because the call that made the row is +/// not always the call that reports it: `appendRows` knows which row of +/// the batch it was and the row itself does not, and a sentence can have +/// that put in front of it where an exception that has already been built +/// cannot. +enum Refused { + Row(String), + /// The boundary itself failed, which is already an exception and is + /// not this side's to word. + Boundary(Error), +} + +impl From for Refused { + fn from(err: Error) -> Refused { + Refused::Boundary(err) + } +} + +/// The exception for a refused row, with a sentence in front of it for a +/// refusal being reported from further out than it was made. +fn raised(env: &Env, why: Refused, opening: &str) -> Error { + match why { + Refused::Row(reason) => usage(env, format!("{opening}{reason}")), + Refused::Boundary(err) => err, + } +} + +/// Opening one, which is the call that reads the table's shape. +pub struct OpenTask { + inner: Arc>>, + alive: Arc, + in_txn: Arc, + table: String, + /// Why this is not going to run, when it is not. + refused: Option, +} + +impl OpenTask { + pub(crate) fn new( + inner: Arc>>, + alive: Arc, + in_txn: Arc, + table: String, + refused: Option, + ) -> OpenTask { + OpenTask { + inner, + alive, + in_txn, + table, + refused, + } + } +} + +impl<'task> ScopedTask<'task> for OpenTask { + type Output = std::result::Result, Failure>; + type JsValue = ClassInstance<'task, Appender>; + + fn compute(&mut self) -> Result { + if let Some(message) = self.refused.take() { + return Ok(Err(Failure::Usage(message))); + } + let table = self.table.clone(); + Ok(with(&self.inner, &self.alive, &self.in_txn, |conn| { + let cols = shape(conn, &table)?; + // Opened and dropped, purely to find out whether it can be + // opened at all: a column that holds a null, a table a keyed + // rel table is built over, and a read-only connection are all + // refused here rather than at the first flush. A caller about + // to buffer a million rows wants to hear about them now. + // + // After the shape rather than before it, because a table that + // is not there is the common mistake and the words for it + // belong to the client that knows it is opening an appender. + conn.appender(&table).map(drop)?; + Ok(cols) + })) + } + + fn resolve(&mut self, env: &'task Env, output: Self::Output) -> Result { + let cols = output.map_err(|failure| failed(env, failure, None))?; + let mut instance = Appender { + inner: Arc::clone(&self.inner), + alive: Arc::clone(&self.alive), + in_txn: Arc::clone(&self.in_txn), + table: self.table.clone(), + state: Arc::new(State { + cols: Mutex::new(cols), + buffered: AtomicU64::new(0), + committed: AtomicU64::new(0), + open: AtomicBool::new(true), + }), + busy: Arc::new(AtomicBool::new(false)), + } + .into_instance(env)?; + wire_disposal(env, &mut instance, "dispose")?; + Ok(instance) + } +} + +/// Writing a batch, which is the one commit a load is made of. +pub struct FlushTask { + inner: Arc>>, + alive: Arc, + in_txn: Arc, + state: Arc, + /// The claim on the appender, held by the task that took it and + /// given back when it ends. + busy: Option>, + table: String, + /// Whether this flush is also the last one. + closing: bool, + refused: Option, +} + +impl<'task> ScopedTask<'task> for FlushTask { + type Output = std::result::Result; + type JsValue = f64; + + fn compute(&mut self) -> Result { + if let Some(message) = self.refused.take() { + return Ok(Err(Failure::Usage(message))); + } + let Ok(mut cols) = self.state.cols.lock() else { + return Ok(Err(Failure::Usage(POISONED.to_string()))); + }; + if !self.state.open.load(Ordering::Acquire) { + // A close of a closed appender writes nothing and says + // nothing, which is what makes an early close and an `await + // using` work together. A flush is a caller asking for a + // write and is owed the answer that there is nowhere to + // write it. + return Ok(match self.closing { + true => Ok(self.state.committed.load(Ordering::Acquire)), + false => Err(Failure::Usage(FINISHED.to_string())), + }); + } + let written = self.state.write( + &mut cols, + &self.inner, + &self.alive, + &self.in_txn, + &self.table, + ); + // Left open when the write failed, because the rows are still + // buffered and an appender that could not be written to is one a + // caller may want to try again, whereas a closed one has nowhere + // to put them. + if self.closing && written.is_ok() { + self.state.open.store(false, Ordering::Release); + } + Ok(written) + } + + fn resolve(&mut self, env: &'task Env, output: Self::Output) -> Result { + output + .map(|committed| committed as f64) + .map_err(|failure| failed(env, failure, None)) + } + + fn finally(self, _env: Env) -> Result<()> { + if let Some(busy) = &self.busy { + busy.store(false, Ordering::Release); + } + Ok(()) + } +} + +/// The catalog as the file has it, which is not always the one the +/// session is holding. +/// +/// A session reloads its catalog when a statement runs, and an appender +/// is not a statement: a flush commits its rows and folds them in +/// without the session hearing about it, so the row counts the session +/// remembers are the counts from the last statement. Loading it costs a +/// read of a block chain, once per appender opened and once per flush of +/// a rel table, which is nothing beside the commit either of them is +/// about to do. +fn catalog(conn: &mut zudb::Connection) -> std::result::Result { + let file = conn.session_mut().file_mut()?; + Ok(Catalog::load(file)?) +} + +/// The columns of the table an appender was opened on, in the order it +/// declares them. +/// +/// A node table's columns are the ones the property store holds, with +/// the types it holds them as, which is what the engine's appender +/// checks a row against. 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 those are the two columns and they are named for what +/// they are. +/// +/// Read here rather than left to the flush so that a value that does not +/// belong in a column is refused by the call that appended it. A row is +/// refused a million rows before the flush that would have carried it, +/// and the message names the column rather than guessing at it from the +/// values that came before. +fn shape(conn: &mut zudb::Connection, table: &str) -> std::result::Result, Failure> { + let catalog = catalog(conn)?; + if let Some(rel) = catalog.rel_by_name(table) { + let ends = [rel.from, rel.to]; + let named = |id: u32, fallback: &str| { + catalog + .node_tables() + .iter() + .find(|node| node.id == id) + .map_or_else(|| fallback.to_string(), |node| node.name.clone()) + }; + // Named for the tables the edge runs between, since that is what + // a row of a rel table is and there is nothing else to call the + // two columns. + return Ok(vec![ + Buffer { + name: format!("from {}", named(ends[0], "the source table")), + values: Column::Int(Vec::new()), + ends: Some(ends[0]), + }, + Buffer { + name: format!("to {}", named(ends[1], "the destination table")), + values: Column::Int(Vec::new()), + ends: Some(ends[1]), + }, + ]); + } + let id = catalog + .node_by_name(table) + .map(|node| node.id) + .ok_or_else(|| { + Failure::Usage(format!( + "there is no table '{table}' in this database, and an appender writes into a \ + table that is already there" + )) + })?; + let file = conn.session_mut().file_mut()?; + let directory = zudb::zu1::props::load_props(file, id)?.ok_or_else(|| { + Failure::Usage(format!( + "'{table}' stores no properties, so it has no columns to append to" + )) + })?; + directory + .columns + .iter() + .map(|column| { + Ok(Buffer { + name: column.name.clone(), + values: Column::for_type(&column.ty).ok_or_else(|| { + Failure::Usage(format!( + "column '{}' of '{table}' holds {}, which this client cannot yet \ + append to", + column.name, column.ty + )) + })?, + ends: None, + }) + }) + .collect() +} + +/// What an appender that has already been closed says. +const FINISHED: &str = "this appender is closed, and a closed appender has already written \ + everything it was given"; + +/// What an appender says while a flush of its own is still running. +/// +/// A flush holds the buffers for as long as the commit takes, and the +/// commit is off the event loop where waiting is allowed. Waiting for it +/// here is not: an append that blocked on a flush would stop the loop +/// for the length of a write to disk. So it is refused, the same way a +/// statement behind a half-read stream is, and awaiting the flush is the +/// answer to both. +const BUSY: &str = "a flush of this appender has not finished, and its rows are not yours to \ + add to until it has: await the flush"; diff --git a/src/buffer.rs b/src/buffer.rs new file mode 100644 index 0000000..04eb0da --- /dev/null +++ b/src/buffer.rs @@ -0,0 +1,366 @@ +//! JavaScript values, buffered as the columns the engine stores. +//! +//! A row appended is not a row written. It goes into a vector per +//! column, in the shape the property store keeps that column in, and a +//! flush hands the whole batch over as one commit. So the work per +//! value is a conversion and a push, and the conversion is where a +//! value that does not belong in a column is caught. +//! +//! What a column holds is settled by the table being written to, which +//! is read when the appender opens. There is no null: a column that +//! holds one cannot be appended to at all, so a null here could only +//! ever be refused, and refusing it at the row that wrote it is better +//! than refusing it at the flush a million rows later. +//! +//! The rules are the parameter binder's, in [`crate::value`], and they +//! are deliberately the same: a `bigint` is an INT64, a whole `number` +//! is an INT64 too because `[1, 'ada']` is what a caller writes, and +//! the four temporal classes and the `Temporal` values are read the +//! same way in both places. What is different is that a column has +//! already said what it holds, so a value of the wrong type is refused +//! here where the binder would have bound it and let the engine +//! decide. + +use napi::bindgen_prelude::*; +use napi::{Env, ValueType}; +use zu_common::{DurationKind, FloatBits, IntBits, LogicalType, Temporal}; +use zudb::Field; + +use crate::temporal; +use crate::value::temporal_from; + +/// One column's values, in the shape the property store wants them. +/// +/// Owned rather than borrowed from the caller's arrays, because a +/// JavaScript array holds values of the runtime's own and the store +/// holds numbers: there is nothing here to borrow. The arms are the +/// storage arms, so a flush hands a buffer over with no pass to convert +/// it. +pub enum Column { + Int(Vec), + Float(Vec), + Bool(Vec), + /// Kept as strings rather than as bytes, because the store wants + /// the bytes and the appender wants the `&str`, and a `String` + /// lends out either without a copy. + Str(Vec), + Bytes(Vec>), + Date(Vec), + LocalTime(Vec), + LocalDatetime(Vec), + Duration(DurationKind, Vec), +} + +/// Why a value did not go in. +/// +/// A column that wanted something else says what it holds and lets the +/// caller word the rest, since the caller knows the column's name and +/// where in the row it sits. A value that is of the right kind and +/// still wrong says so itself, as a phrase the caller puts its subject +/// in front of, because "a bigint outside what INT64 holds" is not +/// something a list of column types can express. +pub enum Mismatch { + Wanted(&'static str), + Says(String), + /// The boundary itself failed, which is not the caller's mistake + /// and is passed along as it is. + Boundary(Error), +} + +impl From for Mismatch { + fn from(err: Error) -> Mismatch { + Mismatch::Boundary(err) + } +} + +impl Column { + /// The buffer a column of this declared type appends into, or + /// `None` for a type the ingest path cannot carry. + /// + /// The match is on the exact declared type and not on its family, + /// because that is what the ingest checks: it compares the stored + /// column's type against the type its values claim, so an `INT32` + /// column or a `VARCHAR(20)` one has no buffer here even though its + /// bits would fit the same lane. This is the engine appender's own + /// table, kept in step with it, because a buffer it would refuse is + /// better refused before a million rows go into it. + pub fn for_type(ty: &LogicalType) -> Option { + Some(match ty { + LogicalType::Int { + signed: true, + bits: IntBits::B64, + precision: None, + } => Column::Int(Vec::new()), + LogicalType::Bool => Column::Bool(Vec::new()), + LogicalType::Float { + bits: FloatBits::B64, + precision: None, + } => Column::Float(Vec::new()), + LogicalType::Date => Column::Date(Vec::new()), + LogicalType::LocalTime => Column::LocalTime(Vec::new()), + LogicalType::LocalDatetime => Column::LocalDatetime(Vec::new()), + LogicalType::Duration(kind) => Column::Duration(*kind, Vec::new()), + LogicalType::Str { + min: None, + max: None, + fixed: false, + } => Column::Str(Vec::new()), + LogicalType::Bytes { + min: None, + max: None, + fixed: false, + } => Column::Bytes(Vec::new()), + _ => return None, + }) + } + + /// 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(); + let wanted = || Mismatch::Wanted(holds); + match self { + Column::Bool(v) => v.push(read_bool(value).ok_or_else(wanted)?), + Column::Int(v) => v.push(read_int(value)?.ok_or_else(wanted)?), + Column::Float(v) => v.push(read_float(value)?.ok_or_else(wanted)?), + Column::Str(v) => v.push(read_str(value)?.ok_or_else(wanted)?), + Column::Bytes(v) => v.push(read_bytes(value)?.ok_or_else(wanted)?), + Column::Date(v) => match moment(env, &value)? { + Some(Temporal::Date(days)) => v.push(days), + _ => return Err(wanted()), + }, + Column::LocalTime(v) => match moment(env, &value)? { + Some(Temporal::LocalTime(nanos)) => v.push(nanos), + // A time that carries an offset is a different type and + // not a time this column can hold, and saying so beats + // dropping the offset or writing it as though it were + // local. + Some(Temporal::ZonedTime { .. }) => { + return Err(Mismatch::Says( + "it carries an offset, and this column holds local times".to_string(), + )); + } + _ => return Err(wanted()), + }, + Column::LocalDatetime(v) => match moment(env, &value)? { + Some(Temporal::LocalDatetime(nanos)) => v.push(nanos), + Some(Temporal::ZonedDatetime { .. }) => { + return Err(Mismatch::Says( + "it carries an offset, and this column holds local datetimes".to_string(), + )); + } + _ => return Err(wanted()), + }, + Column::Duration(kind, v) => match moment(env, &value)? { + // The two kinds do not mix: a column of months has no + // room for a count of nanoseconds and the other way + // about, and a duration of the other kind is refused + // rather than converted through a month of some length + // nobody chose. + Some(Temporal::Duration(found, count)) if found == *kind => v.push(count), + Some(Temporal::Duration(..)) => return Err(wanted()), + _ => return Err(wanted()), + }, + } + Ok(()) + } + + /// What this column holds, for the message when it was handed + /// something else. Plural, because it is the column that is being + /// described and not the value. + pub fn holds(&self) -> &'static str { + match self { + Column::Int(_) => "whole numbers", + Column::Float(_) => "floats", + Column::Bool(_) => "booleans", + Column::Str(_) => "strings", + Column::Bytes(_) => "byte strings", + Column::Date(_) => "dates", + Column::LocalTime(_) => "times", + Column::LocalDatetime(_) => "datetimes", + Column::Duration(DurationKind::YearMonth, _) => "year-month durations", + Column::Duration(DurationKind::DayTime, _) => "day-time durations", + } + } + + /// Drops the value written last, which is how a refused row takes + /// back the fields it managed to write before the one that failed. + pub fn pop(&mut self) { + match self { + Column::Int(v) => drop(v.pop()), + Column::Float(v) => drop(v.pop()), + Column::Bool(v) => drop(v.pop()), + Column::Str(v) => drop(v.pop()), + Column::Bytes(v) => drop(v.pop()), + Column::Date(v) => drop(v.pop()), + Column::LocalTime(v) | Column::LocalDatetime(v) => drop(v.pop()), + Column::Duration(_, v) => drop(v.pop()), + } + } + + pub fn clear(&mut self) { + match self { + Column::Int(v) => v.clear(), + Column::Float(v) => v.clear(), + Column::Bool(v) => v.clear(), + Column::Str(v) => v.clear(), + Column::Bytes(v) => v.clear(), + Column::Date(v) => v.clear(), + Column::LocalTime(v) | Column::LocalDatetime(v) => v.clear(), + Column::Duration(_, v) => v.clear(), + } + } + + /// One value of this column as the engine's appender takes it. + /// + /// A field borrows rather than owning, which is the point of it on + /// a string column: the buffer already holds the bytes and the + /// appender is about to copy them into its own, so lending them is + /// the difference between one copy per row and two. + pub fn field(&self, row: usize) -> Field<'_> { + match self { + Column::Int(v) => Field::Int(v[row]), + Column::Float(v) => Field::Float(v[row]), + Column::Bool(v) => Field::Bool(v[row]), + Column::Str(v) => Field::Str(&v[row]), + Column::Bytes(v) => Field::Bytes(&v[row]), + Column::Date(v) => Field::Temporal(Temporal::Date(v[row])), + Column::LocalTime(v) => Field::Temporal(Temporal::LocalTime(v[row])), + Column::LocalDatetime(v) => Field::Temporal(Temporal::LocalDatetime(v[row])), + Column::Duration(kind, v) => Field::Temporal(Temporal::Duration(*kind, v[row])), + } + } +} + +fn read_bool(value: Unknown<'_>) -> Option { + match value.get_type().ok()? { + ValueType::Boolean => bool::from_unknown(value).ok(), + _ => None, + } +} + +/// A `bigint`, or a `number` that is a whole one. +/// +/// The `number` is the reason this is not one line. A caller writing +/// row literals writes `[1, 'ada']`, and refusing that because the +/// column is INT64 would be pedantry, so a whole number goes in. One +/// that is not whole does not: rounding it would put a value in the +/// database that the caller never wrote, and a number past 2^53 is one +/// whose neighbours it can no longer be told from, so both are refused +/// where they were written. +fn read_int(value: Unknown<'_>) -> std::result::Result, Mismatch> { + match value.get_type()? { + ValueType::BigInt => { + let (n, lossless) = BigInt::from_unknown(value)?.get_i64(); + match lossless { + true => Ok(Some(n)), + false => Err(Mismatch::Says( + "it is a bigint outside what INT64 holds, which is -2^63 up to 2^63 - 1" + .to_string(), + )), + } + } + ValueType::Number => { + let n = f64::from_unknown(value)?; + if n.fract() != 0.0 { + return Err(Mismatch::Says(format!( + "it is {n}, which is not a whole number" + ))); + } + if n.abs() > 9_007_199_254_740_992.0 { + return Err(Mismatch::Says(format!( + "it is {n}, which is past 2^53, where a number can no longer be told from \ + its neighbours: write it as a bigint" + ))); + } + Ok(Some(n as i64)) + } + _ => Ok(None), + } +} + +fn read_float(value: Unknown<'_>) -> std::result::Result, Mismatch> { + match value.get_type()? { + ValueType::Number => Ok(Some(f64::from_unknown(value)?)), + _ => Ok(None), + } +} + +fn read_str(value: Unknown<'_>) -> std::result::Result, Mismatch> { + match value.get_type()? { + ValueType::String => Ok(Some(String::from_unknown(value)?)), + _ => Ok(None), + } +} + +/// The bytes of a `Uint8Array`, which is what a `Buffer` is too. +fn read_bytes(value: Unknown<'_>) -> std::result::Result>, Mismatch> { + if value.get_type()? != ValueType::Object || !value.is_typedarray()? { + return Ok(None); + } + // A typed array of some other width is not bytes, and it says so + // the way every other wrong value does rather than as a boundary + // failure about a conversion the caller never wrote. + Ok(Uint8Array::from_unknown(value) + .ok() + .map(|array| array.to_vec())) +} + +/// The instant a value carries, whichever of the two spellings it is +/// written in. +/// +/// The four classes first and `Temporal` second, which is the order +/// [`crate::value`] binds a parameter in and for the same reason: a +/// `Temporal` value has no own enumerable properties, so anything that +/// reads it as a plain object reads it as empty. +fn moment(env: &Env, value: &Unknown<'_>) -> std::result::Result, Mismatch> { + if value.get_type()? != ValueType::Object { + return Ok(None); + } + if let Some(found) = temporal_from(env, value)? { + return Ok(Some(found)); + } + Ok(temporal::from_temporal(env, "this value", value)?) +} + +/// What arrived, in the words a person writing JavaScript uses for it. +/// +/// A class name where there is one, because `a ZuTimestamp` in a +/// message about a column of dates is the whole explanation, and `an +/// object` is a message that sends the reader back to their own code to +/// work out which object. +pub fn named(value: &Unknown<'_>) -> String { + let kind = match value.get_type() { + Ok(kind) => kind, + Err(_) => return "a value this client could not read".to_string(), + }; + match kind { + ValueType::Undefined => "undefined".to_string(), + ValueType::Null => "null".to_string(), + ValueType::Object => class_of(value).map_or_else( + || "an object".to_string(), + |name| match name.starts_with(['A', 'E', 'I', 'O', 'U']) { + true => format!("an {name}"), + false => format!("a {name}"), + }, + ), + other => format!("a {other}").to_lowercase(), + } +} + +/// The name of the class a value was made by, when it has one that is +/// worth printing. +fn class_of(value: &Unknown<'_>) -> Option { + let object = Object::from_unknown(*value).ok()?; + let constructor: Unknown<'_> = object.get_named_property("constructor").ok()?; + if constructor.get_type().ok()? != ValueType::Function { + return None; + } + let name: String = Object::from_unknown(constructor) + .ok()? + .get_named_property("name") + .ok()?; + match name.is_empty() { + true => None, + false => Some(name), + } +} diff --git a/src/conn.rs b/src/conn.rs index 7dab81c..e77888c 100644 --- a/src/conn.rs +++ b/src/conn.rs @@ -32,6 +32,7 @@ use napi_derive::napi; use zudb::query::{QueryResult, Value}; use zudb::{Config, Database, DiagnosticRecord, Interrupt, ZuError}; +use crate::append::OpenTask; use crate::cancel::Watch; use crate::error::{aborted, raise, usage}; use crate::stream::{self, Started, ZuCursor}; @@ -369,6 +370,38 @@ impl Connection { AsyncTask::new(self.start(read_only, refused)) } + /// Opens an appender on `table` and hands it back. + /// + /// The bulk-load path. A load written as statements pays a commit + /// per row, and an appender pays one per flush, which is the whole + /// difference between loading a million rows in an afternoon and + /// loading them in a minute. + /// + /// ```js + /// await using rows = await conn.appender('person') + /// for (const [id, name] of people) rows.appendRow([id, name]) + /// await rows.flush() + /// ``` + /// + /// The table has to exist, and its columns are read here, so a + /// table nothing declares and a column of a type the ingest cannot + /// carry are both refused at this call rather than at the flush a + /// million rows later. + #[napi(ts_args_type = "table: string", ts_return_type = "Promise")] + pub fn appender(&self, table: Unknown<'_>) -> AsyncTask { + let named = match self.alive.load(Ordering::Acquire) { + true => text(&table, "table"), + false => Err(CLOSED.to_string()), + }; + AsyncTask::new(OpenTask::new( + Arc::clone(&self.inner), + Arc::clone(&self.alive), + Arc::clone(&self.in_txn), + named.as_deref().unwrap_or_default().to_string(), + named.err(), + )) + } + /// 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/lib.rs b/src/lib.rs index 213337f..8c50457 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -16,6 +16,8 @@ //! ABI has nothing to say about and a JavaScript program cannot do //! without. +mod append; +mod buffer; mod cancel; mod conn; mod error; diff --git a/src/value.rs b/src/value.rs index efc1cdc..b7da1a1 100644 --- a/src/value.rs +++ b/src/value.rs @@ -827,7 +827,7 @@ fn from_object(env: &Env, name: &str, value: Unknown<'_>, depth: usize) -> Resul /// is one of its own rather than by reading a tag off the object. /// A plain object shaped like a date is a record, and only an instance /// is a date. -fn temporal_from(env: &Env, value: &Unknown<'_>) -> Result> { +pub(crate) fn temporal_from(env: &Env, value: &Unknown<'_>) -> Result> { if ZuDate::instance_of(env, value)? { let days: i32 = Object::from_unknown(*value)?.get_named_property("days")?; return Ok(Some(Temporal::Date(days))); diff --git a/test/appender.test.mjs b/test/appender.test.mjs new file mode 100644 index 0000000..1aa0832 --- /dev/null +++ b/test/appender.test.mjs @@ -0,0 +1,545 @@ +// Loading rows into a table that is already there. +// +// What is being asserted is the shape of a loader rather than the shape +// of a write. `INSERT` already writes a row and is already atomic; an +// appender exists because a million of them is a million commits, and +// what it trades for one commit is that the rows are in memory until the +// flush. So these tests are mostly about where a row is at each moment: +// buffered, written, refused, or thrown away. +// +// The other half is the one synchronous call in this client. `appendRow` +// throws rather than rejecting, because it reaches nothing that can +// wait, and a caller who wraps it in a `try` still catches the same +// `ZuUsageError` the promises reject with. + +import assert from 'node:assert/strict' +import test from 'node:test' + +import { ZuDate, ZuDuration, ZuTimestamp } from 'zudb' + +import { fresh, isZuError, twoPeople } from './helper.mjs' + +const COUNT = 'MATCH (p:person) RETURN count(*) AS n' +const PEOPLE = 'MATCH (p:person) RETURN p.id AS id, p.name AS name' + +async function people(conn) { + const rows = await conn.query(COUNT) + return Number(rows[0].n) +} + +test('rows go in on the flush and not before it', async (t) => { + const { conn } = await twoPeople(t) + + const rows = await conn.appender('person') + assert.equal(rows.table, 'person') + assert.equal(rows.buffered, 0) + assert.equal(rows.committed, 0) + assert.equal(rows.closed, false) + + rows.appendRow([3n, 'ida']) + rows.appendRow([4n, 'eve']) + // Buffered here rather than in the database, which is the whole + // bargain: a query run now sees the two it started with. + assert.equal(rows.buffered, 2) + assert.equal(await people(conn), 2) + + assert.equal(await rows.flush(), 2) + assert.equal(rows.buffered, 0) + assert.equal(rows.committed, 2) + assert.equal(await people(conn), 4) +}) + +test('what was appended is what comes back', async (t) => { + const { conn } = await twoPeople(t) + + const rows = await conn.appender('person') + rows.appendRows([ + [3n, 'ida'], + [4n, 'eve'], + ]) + await rows.close() + + const found = await conn.query(PEOPLE) + assert.deepEqual( + found.map((row) => row.name), + ['ada', 'zoe', 'ida', 'eve'], + ) + assert.deepEqual( + found.map((row) => row.id), + [1n, 2n, 3n, 4n], + ) +}) + +test('a batch is one commit and several are several', async (t) => { + const { conn } = await twoPeople(t) + + const rows = await conn.appender('person') + for (const batch of [0, 1, 2]) { + for (let at = 0; at < 100; at += 1) rows.appendRow([BigInt(batch * 100 + at), `p${at}`]) + // Each flush answers the running total rather than what it wrote + // itself, because the total is the number a loader reports and the + // batch is a number it already knows. + assert.equal(await rows.flush(), (batch + 1) * 100) + } + assert.equal(await rows.close(), 300) + + assert.equal(await people(conn), 302) +}) + +test('appendRows answers how many rows went in', async (t) => { + const { conn } = await twoPeople(t) + + const rows = await conn.appender('person') + assert.equal(rows.appendRows([]), 0) + assert.equal( + rows.appendRows([ + [3n, 'ida'], + [4n, 'eve'], + [5n, 'ora'], + ]), + 3, + ) + assert.equal(rows.buffered, 3) + await rows.close() + assert.equal(await people(conn), 5) +}) + +test('a flush with nothing buffered writes nothing and says so', async (t) => { + const { conn } = await twoPeople(t) + + const rows = await conn.appender('person') + // What a loader flushing on a timer does between batches, and it had + // better not be a commit. + assert.equal(await rows.flush(), 0) + assert.equal(await rows.flush(), 0) + assert.equal(await people(conn), 2) +}) + +test('await using flushes what the block left buffered', async (t) => { + const { conn } = await twoPeople(t) + + { + await using rows = await conn.appender('person') + rows.appendRow([3n, 'ida']) + } + + // The opposite of what a transaction's disposal does here, because + // the question is a different one: a buffer that left its scope + // unwritten is a loader that read its rows and threw them away. + assert.equal(await people(conn), 3) +}) + +test('discard is how a block leaves without writing', async (t) => { + const { conn } = await twoPeople(t) + + { + await using rows = await conn.appender('person') + rows.appendRow([3n, 'ida']) + rows.appendRow([4n, 'eve']) + assert.equal(rows.discard(), 2) + assert.equal(rows.buffered, 0) + } + + assert.equal(await people(conn), 2) +}) + +test('discard leaves what an earlier flush committed', async (t) => { + const { conn } = await twoPeople(t) + + const rows = await conn.appender('person') + rows.appendRow([3n, 'ida']) + await rows.flush() + rows.appendRow([4n, 'eve']) + assert.equal(rows.discard(), 1) + await rows.close() + + assert.equal(await people(conn), 3) + assert.equal(rows.committed, 1) +}) + +test('closing twice writes once', async (t) => { + const { conn } = await twoPeople(t) + + const rows = await conn.appender('person') + rows.appendRow([3n, 'ida']) + + assert.equal(await rows.close(), 1) + assert.equal(rows.closed, true) + // The second one is what an `await using` runs after a block that + // closed early, and it has to be quiet rather than a failure out of + // code that did everything right. + assert.equal(await rows.close(), 1) + assert.equal(await people(conn), 3) +}) + +test('a closed appender refuses a row and a flush', async (t) => { + const { conn } = await twoPeople(t) + + const rows = await conn.appender('person') + await rows.close() + + assert.throws( + () => rows.appendRow([3n, 'ida']), + (err) => { + assert.ok(isZuError(err, 'ZuUsageError'), err.message) + assert.match(err.message, /this appender is closed/) + return true + }, + ) + // A flush is a caller asking for a write, so it is owed the answer + // that there is nowhere to write it, where a close is not. + await assert.rejects(() => rows.flush(), (err) => isZuError(err, 'ZuUsageError')) +}) + +test('a row of the wrong width is refused and names the columns', async (t) => { + const { conn } = await twoPeople(t) + + const rows = await conn.appender('person') + assert.throws( + () => rows.appendRow([3n]), + (err) => { + assert.ok(isZuError(err, 'ZuUsageError'), err.message) + assert.match(err.message, /carries 1 value and 'person' takes 2: id, name/) + return true + }, + ) + // A synchronous refusal, so there is no GQLSTATUS: nothing reached + // the engine. + assert.equal(rows.buffered, 0) +}) + +test('a value that does not fit its column is refused where it was written', async (t) => { + const { conn } = await twoPeople(t) + + const rows = await conn.appender('person') + assert.throws( + () => rows.appendRow(['three', 'ida']), + (err) => { + assert.match(err.message, /value 0 of this row is a string/) + assert.match(err.message, /column 'id' of 'person' holds whole numbers/) + return true + }, + ) + + // Nothing of the refused row is kept, so the column that did take its + // value has given it back and the next row is a whole one. + rows.appendRow([3n, 'ida']) + assert.equal(rows.buffered, 1) + await rows.close() + + const found = await conn.query(PEOPLE) + assert.deepEqual( + found.map((row) => row.name), + ['ada', 'zoe', 'ida'], + ) +}) + +test('a whole number goes into an INT64 column and a fraction does not', async (t) => { + const { conn } = await twoPeople(t) + + const rows = await conn.appender('person') + // A caller writing row literals writes 3 rather than 3n, and refusing + // that would be pedantry. + rows.appendRow([3, 'ida']) + + assert.throws( + () => rows.appendRow([3.5, 'eve']), + (err) => { + assert.match(err.message, /it is 3.5, which is not a whole number/) + return true + }, + ) + assert.throws( + () => rows.appendRow([2 ** 60, 'eve']), + (err) => { + assert.match(err.message, /past 2\^53/) + assert.match(err.message, /write it as a bigint/) + return true + }, + ) + + await rows.close() + const found = await conn.query(PEOPLE) + assert.equal(found[2].id, 3n) +}) + +test('one bad row in a batch says which one it was', async (t) => { + const { conn } = await twoPeople(t) + + const rows = await conn.appender('person') + assert.throws( + () => + rows.appendRows([ + [3n, 'ida'], + [4n, 'eve'], + [5n, 6n], + ]), + (err) => { + assert.ok(isZuError(err, 'ZuUsageError'), err.message) + assert.match(err.message, /^row 2 of these: /) + assert.match(err.message, /column 'name' of 'person' holds strings/) + return true + }, + ) + + // The rows before it stay: nothing here is a transaction until the + // flush, and throwing away work the caller can keep would not make it + // one. The count is where they start again. + assert.equal(rows.buffered, 2) + await rows.close() + assert.equal(await people(conn), 4) +}) + +test('something that is not an array of rows is refused', async (t) => { + const { conn } = await twoPeople(t) + + const rows = await conn.appender('person') + assert.throws( + () => rows.appendRows({ id: 3n, name: 'ida' }), + (err) => { + assert.match(err.message, /rows are an array of arrays/) + return true + }, + ) + assert.throws( + () => rows.appendRow({ id: 3n, name: 'ida' }), + (err) => { + assert.match(err.message, /a row is an array of one value per column of 'person'/) + return true + }, + ) +}) + +test('every column type the engine stores takes a value', async (t) => { + const { conn } = await fresh(t) + await conn.exec( + "INSERT (e:event {id: 1, on: DATE '2024-01-02', at: LOCAL DATETIME '2024-01-02T03:04:05', " + + "took: DURATION 'PT1H', hot: true, ratio: 1.5})", + ) + + const rows = await conn.appender('event') + rows.appendRow([ + 2n, + new ZuDate(20_000), + new ZuTimestamp(1_700_000_000_000_000_000n), + ZuDuration.ofNanos(7_200_000_000_000n), + false, + 2.5, + ]) + assert.equal(await rows.close(), 1) + + const found = await conn.query( + 'MATCH (e:event) RETURN e.id AS id, e.on AS on, e.at AS at, e.took AS took, e.hot AS hot, ' + + 'e.ratio AS ratio', + ) + assert.equal(found.length, 2) + assert.equal(found[1].on.days, 20_000) + assert.equal(found[1].at.nanos, 1_700_000_000_000_000_000n) + assert.equal(found[1].took.nanos, 7_200_000_000_000n) + assert.equal(found[1].hot, false) + assert.equal(found[1].ratio, 2.5) +}) + +test('a timestamp with an offset does not go in a column of local ones', async (t) => { + const { conn } = await fresh(t) + await conn.exec("INSERT (e:event {id: 1, at: LOCAL DATETIME '2024-01-02T03:04:05'})") + + const rows = await conn.appender('event') + assert.throws( + () => rows.appendRow([2n, new ZuTimestamp(0n, 120)]), + (err) => { + // Saying so beats dropping the offset or writing it as though the + // instant were local, which are the two ways to be quietly wrong. + assert.match(err.message, /it carries an offset, and this column holds local datetimes/) + return true + }, + ) +}) + +test('a rel table takes the two ends of an edge', async (t) => { + const { conn } = await twoPeople(t) + await conn.exec('MATCH (a:person), (b:person) INSERT (a)-[:knows]->(b)') + + const edges = await conn.appender('knows') + // Named for the tables the edge runs between, since that is what a + // row of a rel table is. + assert.equal(edges.table, 'knows') + edges.appendRow([1n, 0n]) + assert.equal(await edges.close(), 1) + + const found = await conn.query('MATCH (a:person)-[:knows]->(b:person) RETURN a.name AS a, b.name AS b') + assert.ok(found.some((row) => row.a === 'zoe' && row.b === 'ada'), JSON.stringify(found)) +}) + +test('an edge to a row that is not there is refused before anything is written', async (t) => { + const { conn } = await twoPeople(t) + await conn.exec('MATCH (a:person), (b:person) INSERT (a)-[:knows]->(b)') + const before = await conn.query('MATCH ()-[r:knows]->() RETURN count(*) AS n') + + const edges = await conn.appender('knows') + edges.appendRow([0n, 99n]) + await assert.rejects( + () => edges.close(), + (err) => { + assert.ok(isZuError(err, 'ZuUsageError'), err.message) + assert.match(err.message, /joins row 99 of 'person', which has 2 rows in it/) + return true + }, + ) + + // The batch was refused and the file was not touched, which is the + // point of checking it here: the engine's own check comes after the + // write is durable. + const after = await conn.query('MATCH ()-[r:knows]->() RETURN count(*) AS n') + assert.equal(after[0].n, before[0].n) + // And the rows are still buffered, for a caller who wants to look at + // what did not go in. + assert.equal(edges.buffered, 1) +}) + +test('a negative offset is refused where it was appended', async (t) => { + const { conn } = await twoPeople(t) + await conn.exec('MATCH (a:person), (b:person) INSERT (a)-[:knows]->(b)') + + const edges = await conn.appender('knows') + assert.throws( + () => edges.appendRow([0n, -1n]), + (err) => { + assert.match(err.message, /holds row offsets, which count from zero/) + return true + }, + ) +}) + +test('a table that is not there is refused when the appender opens', async (t) => { + const { conn } = await twoPeople(t) + + await assert.rejects( + () => conn.appender('nobody'), + (err) => { + assert.ok(isZuError(err, 'ZuUsageError'), err.message) + assert.match(err.message, /there is no table 'nobody' in this database/) + return true + }, + ) +}) + +test('an appender on a closed connection is refused as a rejection', async (t) => { + const { conn } = await fresh(t) + conn.close() + + await assert.rejects( + () => conn.appender('person'), + (err) => { + assert.ok(isZuError(err, 'ZuUsageError'), err.message) + assert.match(err.message, /the connection is closed/) + return true + }, + ) +}) + +test('a connection closed under an open appender refuses the row that follows', async (t) => { + const { conn } = await twoPeople(t) + + const rows = await conn.appender('person') + conn.close() + + // Refused at the append rather than at the flush, because otherwise + // whether a row was taken would depend on whether the batch happened + // to fill, which is a rule nobody can hold in their head. + assert.throws( + () => rows.appendRow([3n, 'ida']), + (err) => { + assert.ok(isZuError(err, 'ZuUsageError'), err.message) + assert.match(err.message, /the connection is closed/) + return true + }, + ) +}) + +test('a table name that is not a string is refused', async (t) => { + const { conn } = await twoPeople(t) + + await assert.rejects( + () => conn.appender(7), + (err) => { + assert.ok(isZuError(err, 'ZuUsageError'), err.message) + return true + }, + ) +}) + +test('a second flush while one is running is refused rather than queued', async (t) => { + const { conn } = await twoPeople(t) + + const rows = await conn.appender('person') + for (let at = 0; at < 500; at += 1) rows.appendRow([BigInt(at), `p${at}`]) + + const running = rows.flush() + // Issued before the first has answered, so it is refused here on the + // runtime thread rather than queued behind a commit on a threadpool + // one. Two commits of the same buffer would be two writes whose order + // nobody chose. + await assert.rejects( + () => rows.flush(), + (err) => { + assert.ok(isZuError(err, 'ZuUsageError'), err.message) + assert.match(err.message, /await the flush/) + return true + }, + ) + // And an append is refused for the same reason a statement behind a + // half-read stream is: waiting for it would be the event loop waiting + // for a write to disk. + assert.throws(() => rows.appendRow([500n, 'late']), (err) => isZuError(err, 'ZuUsageError')) + + assert.equal(await running, 500) + // The refusal released nothing that was running, so the appender is + // usable the moment the flush has answered. + rows.appendRow([500n, 'late']) + assert.equal(await rows.close(), 501) + assert.equal(await people(conn), 503) +}) + +test('an appender writes what a rollback does not take back', async (t) => { + const { conn } = await twoPeople(t) + + const rows = await conn.appender('person') + const tx = await conn.transaction() + rows.appendRow([3n, 'ida']) + await rows.flush() + await tx.rollback() + await rows.close() + + // The engine's shape today rather than a decision made here: an + // appender writes through the file rather than through the session, + // so a load and a transaction are two different things to reach for. + assert.equal(await people(conn), 3) +}) + +test('two appenders on one connection each write their own rows', async (t) => { + const { conn } = await twoPeople(t) + + const first = await conn.appender('person') + const second = await conn.appender('person') + first.appendRow([3n, 'ida']) + second.appendRow([4n, 'eve']) + + assert.equal(await first.close(), 1) + assert.equal(await second.close(), 1) + assert.equal(await people(conn), 4) +}) + +test('a statement runs between one append and the next', async (t) => { + const { conn } = await twoPeople(t) + + const rows = await conn.appender('person') + rows.appendRow([3n, 'ida']) + // Nothing is held while rows are buffered, which is what the buffers + // being on this side buys: the connection is free between flushes. + assert.equal(await people(conn), 2) + rows.appendRow([4n, 'eve']) + await rows.close() + + assert.equal(await people(conn), 4) +}) diff --git a/test/exports.test.mjs b/test/exports.test.mjs index 5d786c3..cfba292 100644 --- a/test/exports.test.mjs +++ b/test/exports.test.mjs @@ -23,6 +23,7 @@ const SURFACE = [ 'isZuError', 'Connection', 'Transaction', + 'Appender', 'ZuStream', 'ZuCursor', 'ZuDate', diff --git a/test/types/cjs.cts b/test/types/cjs.cts index 02eba05..b20aedd 100644 --- a/test/types/cjs.cts +++ b/test/types/cjs.cts @@ -6,6 +6,7 @@ import { connect, isZuError, ZuTimestamp, + type ZuAppendValue, type ZuParam, type ZuStream, type ZuTransactionOptions, @@ -70,6 +71,22 @@ export async function span(path: string, options: ZuTransactionOptions): Promise } } +export async function bulk(path: string, batch: readonly ZuAppendValue[][]): Promise { + // The `try` and `finally` spelling again, and the word in the + // `finally` is `close` rather than `discard`, since a load that got + // this far means to keep what it read. + const conn = await connect(path) + const rows = await conn.appender('person') + try { + const taken: number = rows.appendRows(batch) + if (taken !== batch.length) throw new Error('a row went missing') + return await rows.flush() + } finally { + if (!rows.closed) await rows.close() + 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 5447a0f..a4680dd 100644 --- a/test/types/esm.mts +++ b/test/types/esm.mts @@ -6,6 +6,8 @@ import { connect, isZuError, ZuDate, + type Appender, + type ZuAppendValue, type ZuBatch, type ZuBigIntMode, type ZuError, @@ -108,6 +110,28 @@ export async function moved(path: string, from: bigint, to: bigint): Promise { + await using conn = await connect(path) + + // `await using` on an appender needs the same declaration the other + // two need, and this one flushes rather than undoing what it holds. + await using rows: Appender = await conn.appender('person') + for (const person of people) { + // A row is an array of values and nothing wider: a null in one does + // not compile, because a column of an appender has one type and + // every value in it is that type. + const row: readonly ZuAppendValue[] = person + rows.appendRow(row) + } + + // Synchronous, so it is a number rather than a promise, and the two + // counts beside it are numbers too. + const buffered: number = rows.buffered + if (buffered !== people.length) throw new Error('a row went missing') + const written: number = await rows.flush() + return written + rows.committed +} + 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 b432221..6777f98 100644 --- a/types/header.d.ts +++ b/types/header.d.ts @@ -123,6 +123,34 @@ export type ZuParam = | ZuParam[] | { [field: string]: ZuParam } +/** + * A value an appender takes, which is narrower than what a statement + * takes. + * + * A column of an appender has one type, read from the table when the + * appender opened, and every value in it is that type. So there is no + * `null` here: a column that holds nulls cannot be appended to at all + * and the appender says so when it opens, and `undefined` in a row is a + * value the caller forgot rather than a null they meant. There are no + * lists and no objects either, because a property column holds a scalar. + * + * BYTES is a `Uint8Array`, which is the one type here that no statement + * parameter can be, and INT64 is a `bigint` or a whole `number` below + * 2^53. A number past that is refused rather than rounded, because past + * 2^53 a number no longer names one integer. + */ +export type ZuAppendValue = + | boolean + | number + | bigint + | string + | Uint8Array + | ZuDate + | ZuTime + | ZuTimestamp + | ZuDuration + | ZuTemporalValue + /** * A walk through the graph: nodes and edges, alternating, a node at * each end. diff --git a/zudb.cjs b/zudb.cjs index cba3074..548e789 100644 --- a/zudb.cjs +++ b/zudb.cjs @@ -164,6 +164,7 @@ module.exports = { isZuError, Connection: binding.Connection, Transaction: binding.Transaction, + Appender: binding.Appender, ZuStream, // The pull underneath a stream, which `conn.cursor(...)` hands back // and almost nobody should be holding. It is here because it is in diff --git a/zudb.d.cts b/zudb.d.cts index 1092625..f22764e 100644 --- a/zudb.d.cts +++ b/zudb.d.cts @@ -42,6 +42,17 @@ declare module './binding.cjs' { * spelling and the commit the word a caller writes. */ interface Transaction extends AsyncDisposable {} + + /** + * The disposal of an appender, declared here for the same reason the + * other two are. + * + * It flushes, which is the opposite of what a transaction's does. A + * buffer that left its scope unwritten would be a loader that read a + * million rows and threw them away, and `discard()` is there for the + * caller who meant exactly that. + */ + interface Appender extends AsyncDisposable {} } /** diff --git a/zudb.mjs b/zudb.mjs index 88e4ace..afc7b83 100644 --- a/zudb.mjs +++ b/zudb.mjs @@ -22,6 +22,7 @@ export const abiVersion = zudb.abiVersion export const isZuError = zudb.isZuError export const Connection = zudb.Connection export const Transaction = zudb.Transaction +export const Appender = zudb.Appender export const ZuStream = zudb.ZuStream export const ZuCursor = zudb.ZuCursor export const ZuDate = zudb.ZuDate