From 8c3063c84745593c286029bbf5e11249c91a24e7 Mon Sep 17 00:00:00 2001 From: Tam Nguyen Duc <1218621+tamnd@users.noreply.github.com> Date: Wed, 19 Aug 2026 15:41:57 +0700 Subject: [PATCH] Load a whole database out of columns and an edge list An appender writes rows into a table that already exists, and no statement makes a rel table, so neither of them is a way to a graph with edges in it. `load(path, options)` is the other shape and the one the C ABI's loader has: a table's columns whole, an edge list whole, one file written once. It is a function rather than a method because there is no connection yet: the file it writes is the file a program connects to afterwards. The path must not exist, since a load builds a database rather than adding to one, and a path that already holds one is a caller who meant a different path. What comes back is what went in, as `{ nodes, rels, columns }`. Edges name rows by position, counting from zero, because at load time a row has no other name. They go in as pairs or as a flat `Int32Array` or `Uint32Array` for a program that built them in memory and would rather not make a million small arrays to hand them over. The same edge twice is one edge, and an edge naming a row the table has not got is refused rather than written. A column goes in as an array of values or as a typed array. The first value of an array settles what the column holds and every value after it has to agree, which is the appender's rule, with the same one widening. A typed array is read as the numbers it already holds, which is one pass over memory rather than a runtime call per value. Everything the caller passed is read on the thread that owns the runtime, and everything after that runs on the threadpool: the edges are sorted, the graph is built, and every column is encoded and written to disk. So the event loop is free for the whole of the expensive part, which on a load is all of it. Over a million rows a typed array column costs 51 ns a row against a plain array's 92, and two columns with an edge each cost 1097 ns a row against the appender's 2125 for the same two columns and no edges at all. --- README.md | 44 +++- bench/load.mjs | 110 ++++++++ binding.cjs | 1 + binding.d.cts | 45 ++++ etc/zudb.api.md | 25 ++ package.json | 1 + src/conn.rs | 2 +- src/lib.rs | 1 + src/load.rs | 580 ++++++++++++++++++++++++++++++++++++++++++ test/exports.test.mjs | 1 + test/load.test.mjs | 407 +++++++++++++++++++++++++++++ test/readme.test.mjs | 2 +- test/types/cjs.cts | 10 + test/types/esm.mts | 26 ++ types/header.d.ts | 42 +++ zudb.cjs | 1 + zudb.mjs | 1 + 17 files changed, 1295 insertions(+), 4 deletions(-) create mode 100644 bench/load.mjs create mode 100644 src/load.rs create mode 100644 test/load.test.mjs diff --git a/README.md b/README.md index c157287..c8e9f22 100644 --- a/README.md +++ b/README.md @@ -36,7 +36,7 @@ The rows are an array, so iterating them is `for (const row of rows)` and nothin ## What works today -`connect`, `query`, `exec`, `stream`, `close`, `dispose` and `await using`. Named parameters both ways, including lists, records and nesting. Every scalar the engine has, plus nodes, edges and paths with their tables named rather than numbered, and `ZuDate`, `ZuTime`, `ZuTimestamp` and `ZuDuration`, with `{ temporal: true }` and `toTemporal()` for the runtimes that have `Temporal`. Read-only connections, memory and thread limits. `bigIntMode`, per statement or per connection. An `AbortSignal` on any statement. The full error surface above, and `isZuError` to recognize it. Streaming, as an async iterable, as batches and as a Web Stream. Transactions, with `inTransaction` on the connection. An appender, for loading rows a batch at a time. Registered frames, so an Arrow table or an object of typed arrays is something a statement can match on without the rows being copied. Both module formats, typed separately. +`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, and `load` for building a whole database out of columns and an edge list. Registered frames, so an Arrow table or an object of typed arrays is something a statement can match on without the rows being copied. Both module formats, typed separately. Build it with `npm run build`, and run the suite with `npm test`. Nothing is published yet, so `npm i zudb` is not a thing you can type at anybody's terminal, but everything it will do is built and installed on every run of the release workflow. @@ -124,6 +124,46 @@ Two more things are worth knowing before a load. A flush issued while one is sti A rel table has no property columns. A row of one is the two ends of an edge, as offsets into the tables it runs between, so `conn.appender("knows")` takes two columns and the flush checks that both rows are there before it writes anything. That check is here rather than the engine's, because the engine's comes after the write is durable. +## Building a graph out of columns and an edge list + +An appender writes rows into a table that already exists, and no statement makes a rel table, so neither of them is a way to a graph with edges in it. `load` is the other shape and the one the C ABI's loader has: a table's columns whole, an edge list whole, one file written once. + +```ts +import { load } from "zudb"; + +const uid = new BigInt64Array([1n, 2n, 3n]); +const name = ["ada", "grace", "kay"]; + +const stats = await load("social.zu1", { + nodes: "person", + rels: "knows", + columns: { uid, name }, + edges: [ + [0, 1], + [1, 2], + ], +}); +console.log(stats); // { nodes: 3, rels: 2, columns: 2 } +``` + +It is a function rather than a method because there is no connection yet: the file it writes is the file a program connects to afterwards. The path must not exist, since a load builds a database rather than adding to one, and a path that already holds one is a caller who meant a different path. What comes back is what went in, as `{ nodes, rels, columns }`. + +Edges name rows by position, counting from zero in the order the columns were written, because at load time a row has no other name. They go in as pairs, `[[0, 1], [1, 2]]`, or as a flat `Int32Array` or `Uint32Array` of two elements an edge for a program that built them in memory and would rather not make a million small arrays to hand them over. The same edge twice is one edge, and an edge naming a row the table has not got is refused rather than written, because a builder handed one would either invent the row or lose the edge. + +A column goes in as an array of values or as a typed array. The first value of an array settles what the column holds and every value after it has to agree, which is the appender's rule, with the same one widening: `[1, 2, 2.5]` is a column of floats. A typed array is read as the numbers it already holds, which is one pass over memory rather than a runtime call per value, and every integer width lands as the INT64 the store keeps. On this machine, with `npm run bench:load` over a million rows: + +``` +one column, typed array 51.3 ms 51 ns/row +one column, plain array 92.2 ms 92 ns/row +two columns, with names 1060.1 ms 1060 ns/row +two columns and an edge each 1096.9 ms 1097 ns/row +the appender, for contrast 2124.5 ms 2125 ns/row +``` + +The last line is the same two columns through the appender, which is the closest comparison there is: a load is about twice as quick and is the only one of the two that can write the edges. The strings are where the rest of the time goes, which is the store encoding them rather than anything on this side of the boundary. + +Everything the caller passed is read on the thread that owns the runtime, because that is the only thread allowed to read a JavaScript value, and everything after that runs on the threadpool: the edges are sorted, the graph is built, and every column is encoded and written to disk. So the event loop is free for the whole of the expensive part, which on a load is all of it. + ## Matching on columns a program already has Columns a program is already holding become something a statement can match on, under a name the program picks. @@ -238,7 +278,7 @@ typedoc rather than api-documenter, which would have been the obvious pick since Anything outside that table has no binary and no source build to fall back on, so the install resolves nothing and the first `require` says so. The browser and the platforms nobody builds for are what the WASM target answers, later. -`npm run bench` measures what this package adds to the engine, which is a row object and one JavaScript value per column: the same scan with the rows dropped is the floor, and the difference between the two is what the boundary costs. Run it against a release build, since a debug build of the engine moves the floor by an order of magnitude and not the rest of it. `npm run bench:append` does the same for the load path and `npm run bench:register` for registered frames, where what is being watched is that the registration does not scale with the rows. +`npm run bench` measures what this package adds to the engine, which is a row object and one JavaScript value per column: the same scan with the rows dropped is the floor, and the difference between the two is what the boundary costs. Run it against a release build, since a debug build of the engine moves the floor by an order of magnitude and not the rest of it. `npm run bench:append` does the same for the appender, `npm run bench:load` for building a database out of columns, and `npm run bench:register` for registered frames, where what is being watched is that the registration does not scale with the rows. ## Still to come diff --git a/bench/load.mjs b/bench/load.mjs new file mode 100644 index 0000000..e221d32 --- /dev/null +++ b/bench/load.mjs @@ -0,0 +1,110 @@ +// What building a database out of columns costs. +// +// `load` is the other way in, and it is the only way to a graph with +// edges in it. The appender writes rows into a table that already +// exists; this writes the file, so the numbers here are not the +// appender's numbers with a different name on them and the first two +// lines are what says so. +// +// The third and fourth are about the way the columns were handed over. +// A typed array is read as the numbers it already holds, which is one +// pass over memory, and an ordinary array is read a value at a time +// through the runtime. The gap between them is the whole argument for +// reaching for a typed array on a load this size. +// +// Run it against a release build, for the reason bench/query.mjs gives. +// +// npm run build && npm run bench:load + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { connect, load } from 'zudb' + +const ROWS = Number(process.env.ZU_BENCH_ROWS ?? 1_000_000) +const REPEATS = Number(process.env.ZU_BENCH_REPEATS ?? 3) + +const dir = await mkdtemp(join(tmpdir(), 'zu-bench-load-')) + +const wide = BigInt64Array.from({ length: ROWS }, (_, ix) => BigInt(ix)) +const plain = Array.from({ length: ROWS }, (_, ix) => ix) +const name = Array.from({ length: ROWS }, (_, ix) => `n${ix}`) +// Out of order on purpose, because sorting them is half the work of +// building the graph and an edge list a program produced by walking +// something else is never in order. +const edges = new Uint32Array(ROWS * 2) +for (let ix = 0; ix < ROWS; ix++) { + edges[ix * 2] = ix + edges[ix * 2 + 1] = (ix * 7 + 1) % ROWS +} + +let counter = 0 + +/// The fastest of `REPEATS` runs, in milliseconds, after one warmup. +/// +/// A fresh path every time, because a load will not write over a +/// database that is there and because a second load into a warm page +/// cache is not the load being measured. +async function time(run) { + await run(join(dir, `warm-${counter++}.zu1`)) + let best = Infinity + for (let round = 0; round < REPEATS; round++) { + const path = join(dir, `bench-${counter++}.zu1`) + const started = performance.now() + await run(path) + best = Math.min(best, performance.now() - started) + } + return best +} + +const cases = [ + { + // One column of whole numbers and nothing else, which is the floor: + // the file, the table and one run of words. + name: 'one column, typed array', + run: (path) => load(path, { nodes: 'person', columns: { uid: wide } }), + }, + { + // The same column written as an ordinary array, which is a runtime + // call per value on the way in. + name: 'one column, plain array', + run: (path) => load(path, { nodes: 'person', columns: { uid: plain } }), + }, + { + // A column of strings, which is where the bytes are. + name: 'two columns, with names', + run: (path) => load(path, { nodes: 'person', columns: { uid: wide, name } }), + }, + { + // The graph: the same columns and an edge per row, sorted and built + // into the CSRs a pattern walks. + name: 'two columns and an edge each', + run: (path) => load(path, { nodes: 'person', rels: 'knows', columns: { uid: wide, name }, edges }), + }, + { + // The other way in, for the one case both can do: rows into a table + // that a first insert declared. No edges, since no statement makes + // a rel table. + name: 'the appender, for contrast', + run: async (path) => { + const conn = await connect(path) + await conn.exec("INSERT (p:person {uid: 0, name: 'n0'})") + const rows = await conn.appender('person') + for (let ix = 1; ix < ROWS; ix++) rows.appendRow([wide[ix], name[ix]]) + await rows.close() + conn.close() + }, + }, +] + +console.log(`${ROWS} rows, fastest of ${REPEATS}`) +for (const { name, run } of cases) { + const ms = await time(run) + const each = (ms * 1e6) / ROWS + console.log( + `${name.padEnd(30)} ${ms.toFixed(1).padStart(9)} ms ${Math.round(each).toString().padStart(6)} ns/row`, + ) +} + +await rm(dir, { recursive: true, force: true }) diff --git a/binding.cjs b/binding.cjs index 2929d71..37ad67f 100644 --- a/binding.cjs +++ b/binding.cjs @@ -712,4 +712,5 @@ module.exports.ZuTime = nativeBinding.ZuTime module.exports.ZuTimestamp = nativeBinding.ZuTimestamp module.exports.abiVersion = nativeBinding.abiVersion module.exports.connect = nativeBinding.connect +module.exports.load = nativeBinding.load module.exports.version = nativeBinding.version diff --git a/binding.d.cts b/binding.d.cts index 992f273..6e6da73 100644 --- a/binding.d.cts +++ b/binding.d.cts @@ -221,6 +221,48 @@ export interface ZuArrowTable { */ export type ZuFrame = ZuArrowTable | Record +/** + * An edge list, as the pairs of row numbers it is. + * + * Rows are numbered from zero in the order their columns were written, + * because at load time a row has no other name. The flat spelling is + * there for a program that built its edges in memory and would rather + * not make a million two element arrays to hand them over: two elements + * an edge, read in one pass. + */ +export type ZuEdges = readonly (readonly [number, number])[] | Int32Array | Uint32Array + +/** + * What a load writes, and how much of it. + */ +export interface ZuLoadOptions { + /** The node table, which gets a row per element of every column. */ + readonly nodes: string + /** The rel table holding the edges between those rows. `rel` by default. */ + readonly rels?: string + /** The node table's properties, as column name to values. */ + readonly columns?: Readonly> + /** The edges, as pairs of row numbers. */ + readonly edges?: ZuEdges | null + /** + * How many rows the node table has. + * + * Read off the columns when there are any, so this is for the load + * that writes a graph with no properties at all, and a check on the + * columns when both are given. + */ + readonly rows?: number +} + +/** + * What went into a load. + */ +export interface ZuLoadStats { + readonly nodes: number + readonly rels: number + readonly columns: number +} + /** * A walk through the graph: nodes and edges, alternating, a node at * each end. @@ -1024,5 +1066,8 @@ export interface ConnectOptions { temporal?: boolean } +/** Writes a new database at `path` and answers what went into it. */ +export declare function load(path: string, options: ZuLoadOptions): Promise + /** The version of the client. */ export declare function version(): string diff --git a/etc/zudb.api.md b/etc/zudb.api.md index dfda7c1..b2f0199 100644 --- a/etc/zudb.api.md +++ b/etc/zudb.api.md @@ -54,6 +54,9 @@ export interface ConnectOptions { // @public export function isZuError(value: unknown): value is ZuError +// @public +export function load(path: string, options: ZuLoadOptions): Promise + // @public export class Transaction { commit(): Promise @@ -123,6 +126,9 @@ export class ZuDuration { toTemporal(): ZuTemporalDuration } +// @public +export type ZuEdges = readonly (readonly [number, number])[] | Int32Array | Uint32Array + // @public export interface ZuError extends Error { readonly code?: string @@ -171,6 +177,25 @@ export type ZuFrameValue = | ZuDuration | ZuTemporalValue +// @public +export interface ZuLoadOptions { + readonly columns?: Readonly> + readonly edges?: ZuEdges | null + readonly nodes: string + readonly rels?: string + readonly rows?: number +} + +// @public +export interface ZuLoadStats { + // (undocumented) + readonly columns: number + // (undocumented) + readonly nodes: number + // (undocumented) + readonly rels: number +} + // @public export class ZuNode { // (undocumented) diff --git a/package.json b/package.json index 3fbb23b..8a7a645 100644 --- a/package.json +++ b/package.json @@ -72,6 +72,7 @@ "reference": "node tools/reference.mjs reference", "bench": "node bench/query.mjs", "bench:append": "node bench/append.mjs", + "bench:load": "node bench/load.mjs", "bench:register": "node bench/register.mjs", "bench:temporal": "node --harmony-temporal bench/query.mjs" }, diff --git a/src/conn.rs b/src/conn.rs index 84efea5..b1304ad 100644 --- a/src/conn.rs +++ b/src/conn.rs @@ -956,7 +956,7 @@ fn int_mode(options: Option<&Object<'_>>, connection: Ints) -> std::result::Resu /// from a method whose every other failure is a rejection is a method /// callers have to wrap twice. So the argument arrives unread and this /// is what reads it. -fn text(value: &Unknown<'_>, what: &str) -> std::result::Result { +pub(crate) fn text(value: &Unknown<'_>, what: &str) -> std::result::Result { match value.get_type().map_err(|err| err.reason)? { ValueType::String => String::from_unknown(*value).map_err(|err| err.reason), other => Err(format!( diff --git a/src/lib.rs b/src/lib.rs index 0183b6b..ca9e455 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -22,6 +22,7 @@ mod cancel; mod conn; mod error; mod frame; +mod load; mod register; mod stream; mod temporal; diff --git a/src/load.rs b/src/load.rs new file mode 100644 index 0000000..7b49ab4 --- /dev/null +++ b/src/load.rs @@ -0,0 +1,580 @@ +//! Building a database out of columns and an edge list. +//! +//! ```js +//! await load('social.zu1', { nodes: 'person', rels: 'knows', columns, edges }) +//! ``` +//! +//! A row at a time through `INSERT` is the wrong shape for loading data +//! and the wrong shape for making a graph: every row is parsed, bound and +//! committed, and a rel table cannot be made that way at all, because the +//! statement that would make one says which two tables it joins only for +//! the edge it is writing. This is the other shape, and it is the one the +//! C ABI's loader has: a table's columns whole, an edge list whole, one +//! file written once. +//! +//! What it writes is a node table with a row per element of every column, +//! a rel table holding the edges between those rows, and a primary-key +//! index over the rows so a lookup by key does not scan. Edges name rows +//! by position, counting from zero, because at load time a row has no +//! other name. +//! +//! It is a function rather than a method because there is no connection +//! yet: the file it writes is the file a program connects to afterwards. +//! The path must not exist, since a load builds a database rather than +//! adding to one, and a path that already holds one is a caller who meant +//! a different path. +//! +//! Everything the caller passed is read on the thread that owns the +//! runtime, because that is the only thread allowed to read a JavaScript +//! value, and everything after that runs on the threadpool: the edges are +//! sorted, the graph is built, and every column is encoded and written to +//! disk. So the event loop is free for the whole of the expensive part, +//! which on a load is all of it. + +use std::path::PathBuf; + +use napi::bindgen_prelude::*; +use napi::{Env, ScopedTask, ValueType}; +use napi_derive::napi; +use zudb::zu1::file::Zu1File; +use zudb::zu1::graph::bulk_load_keyed; +use zudb::zu1::props::{PropValues, store_props}; + +use crate::buffer::{Column, Mismatch, named}; +use crate::conn::{Failure, failed, text}; +use crate::register::identifier; + +/// Writes a new database at `path` and answers what went into it. +#[napi( + ts_args_type = "path: string, options: ZuLoadOptions", + ts_return_type = "Promise" +)] +pub fn load(env: &Env, path: Unknown<'_>, options: Unknown<'_>) -> AsyncTask { + match plan(env, path, options) { + Ok(plan) => AsyncTask::new(LoadTask { + plan: Some(plan), + refused: None, + }), + Err(message) => AsyncTask::new(LoadTask { + plan: None, + refused: Some(message), + }), + } +} + +/// Everything the load needs, read off the caller's objects and owned +/// here, so that the write below owes the runtime nothing. +pub struct Plan { + path: PathBuf, + nodes: String, + rels: String, + rows: u64, + columns: Vec<(String, Column)>, + pairs: Vec<(u32, u32)>, +} + +/// What went in, which is what the caller gets back. +pub struct Stats { + nodes: u64, + rels: u64, + columns: u64, +} + +pub struct LoadTask { + plan: Option, + /// What this client refused the call with, before any of it ran. + refused: Option, +} + +/// What a task says when it is run a second time, which nothing this +/// client writes does: a task is built by one call and driven once. +const TWICE: &str = "this load has already run, and a database is written once"; + +impl LoadTask { + fn run(&mut self) -> std::result::Result { + if let Some(message) = self.refused.take() { + return Err(Failure::Usage(message)); + } + let Plan { + path, + nodes, + rels, + rows, + columns, + mut pairs, + } = self.plan.take().ok_or(Failure::Usage(TWICE.to_string()))?; + let mut db = Zu1File::create(&path)?; + // Sorted and deduplicated because the builder wants them that + // way, and because an edge list a program produced by walking + // something else is neither. + pairs.sort_unstable(); + pairs.dedup(); + bulk_load_keyed(&mut db, &nodes, &rels, rows, &pairs, None)?; + let written = columns.len() as u64; + if !columns.is_empty() { + // The store wants a slice of slices for a column of strings, + // which a vector of strings is not, so the row borrows are + // built first and handed over after. + let runs: Vec> = columns + .iter() + .map(|(_, column)| match column { + Column::Str(v) => v.iter().map(String::as_bytes).collect(), + Column::Bytes(v) => v.iter().map(Vec::as_slice).collect(), + _ => Vec::new(), + }) + .collect(); + let props: Vec<(&str, PropValues<'_>)> = columns + .iter() + .zip(&runs) + .map(|((name, column), runs)| { + let values = match column { + Column::Str(_) => PropValues::Str(runs), + Column::Bytes(_) => PropValues::Bytes(runs), + Column::Int(v) => PropValues::Int(words(v)), + Column::Float(v) => PropValues::Float(v), + Column::Bool(v) => PropValues::Bool(v), + Column::Date(v) => PropValues::Date(v), + Column::LocalTime(v) => PropValues::LocalTime(v), + Column::LocalDatetime(v) => PropValues::LocalDatetime(v), + Column::Duration(kind, v) => PropValues::Duration(*kind, v), + }; + (name.as_str(), values) + }) + .collect(); + store_props(&mut db, &nodes, &props)?; + } + Ok(Stats { + nodes: rows, + rels: pairs.len() as u64, + columns: written, + }) + } +} + +/// A column of whole numbers as the words the store keeps them in. +/// +/// The store's integer column is a run of 64 bit words and this client +/// buffers signed ones, which is the same run of bytes read with a +/// different sign: the two types have the same size and alignment and +/// every bit pattern is a value of both. +fn words(values: &[i64]) -> &[u64] { + unsafe { std::slice::from_raw_parts(values.as_ptr().cast::(), values.len()) } +} + +impl<'task> ScopedTask<'task> for LoadTask { + type Output = std::result::Result; + type JsValue = Object<'task>; + + fn compute(&mut self) -> Result { + Ok(self.run()) + } + + fn resolve(&mut self, env: &'task Env, output: Self::Output) -> Result { + let stats = output.map_err(|failure| failed(env, failure, None))?; + // Numbers rather than bigints, because these are counts this + // client made rather than INT64 columns a statement gave back, + // and no load fits 2^53 rows in memory to be counted wrongly. + let mut object = Object::new(env)?; + object.set_named_property("nodes", stats.nodes as f64)?; + object.set_named_property("rels", stats.rels as f64)?; + object.set_named_property("columns", stats.columns as f64)?; + Ok(object) + } +} + +/// The whole call, read into a [`Plan`], or the one sentence it was +/// refused with. +fn plan(env: &Env, path: Unknown<'_>, options: Unknown<'_>) -> std::result::Result { + read(env, path, options).map_err(|err| err.reason) +} + +fn read(env: &Env, path: Unknown<'_>, options: Unknown<'_>) -> Result { + let path = text(&path, "path").map_err(|message| crate::error::usage(env, message))?; + if options.get_type()? != ValueType::Object { + return Err(crate::error::usage( + env, + format!( + "the options are {}, and a load is told at least which node table it is writing", + named(&options) + ), + )); + } + let options = Object::from_unknown(options)?; + let refuse = |message: String| crate::error::usage(env, message); + + let nodes: Option> = options.get("nodes")?; + let Some(nodes) = nodes else { + return Err(refuse( + "a load names the node table it is writing, and this one names none".to_string(), + )); + }; + let nodes = text(&nodes, "node table").map_err(refuse)?; + // A default rather than a required name, because a graph with no + // edges still gets a rel table and a caller who wrote none has no + // opinion about what it is called. + let rels: Option> = options.get("rels")?; + let rels = match rels { + Some(rels) => text(&rels, "rel table").map_err(refuse)?, + None => "rel".to_string(), + }; + for (name, what) in [(&nodes, "a node table"), (&rels, "a rel table")] { + identifier(name, what).map_err(refuse)?; + } + + let columns = built(env, &options)?; + let asked = counted(env, &options)?; + let rows = match (asked, columns.first()) { + (Some(rows), _) if rows < 0.0 || rows.fract() != 0.0 => { + return Err(refuse(format!( + "a load of {rows} rows is a load of a number of rows that is not a whole one" + ))); + } + (Some(rows), Some((name, column))) if column.len() as f64 != rows => { + return Err(refuse(format!( + "column '{name}' holds {} values against the {rows} rows this load asks for", + column.len() + ))); + } + (Some(rows), _) => rows as u64, + (None, Some((_, column))) => column.len() as u64, + (None, None) => { + return Err(refuse( + "a load with no columns has no rows to count, so it has to be told how many" + .to_string(), + )); + } + }; + let pairs = pairs(env, &options, rows)?; + + Ok(Plan { + path: PathBuf::from(path), + nodes, + rels, + rows, + columns, + pairs, + }) +} + +/// How many rows the caller said the table has, written either way a +/// count is written. +fn counted(env: &Env, options: &Object<'_>) -> Result> { + let Some(rows) = options.get::>("rows")? else { + return Ok(None); + }; + Ok(match rows.get_type()? { + ValueType::Null | ValueType::Undefined => None, + ValueType::Number => Some(f64::from_unknown(rows)?), + ValueType::BigInt => Some(BigInt::from_unknown(rows)?.get_i64().0 as f64), + _ => { + return Err(crate::error::usage( + env, + format!( + "the row count is {}, and a row count is a number", + named(&rows) + ), + )); + } + }) +} + +/// Every column, in the order the object holds them, which is the order +/// they were written. +fn built(env: &Env, options: &Object<'_>) -> Result> { + let Some(columns) = options.get::>("columns")? else { + return Ok(Vec::new()); + }; + let names = Object::keys(&columns)?; + let mut built: Vec<(String, Column)> = Vec::with_capacity(names.len()); + for name in names { + identifier(&name, "a column").map_err(|message| crate::error::usage(env, message))?; + let column = column( + env, + &name, + columns.get_named_property::>(&name)?, + )?; + // The store takes a column of bytes and every statement that + // reads one back refuses it, so a load that wrote one would be + // writing data the caller cannot get at again. Refused here until + // the read side catches up, at which point this goes and nothing + // else has to change. + if matches!(column, Column::Bytes(_)) { + return Err(crate::error::usage( + env, + format!( + "column '{name}' holds byte strings, and no statement can read one back yet, \ + so a load will not write a column of them" + ), + )); + } + if let Some((first, had)) = built.first().map(|(name, column)| (name, column.len())) + && column.len() != had + { + return Err(crate::error::usage( + env, + format!( + "column '{name}' holds {} values and column '{first}' holds {had}, and a table \ + is as wide as it is long", + column.len() + ), + )); + } + built.push((name, column)); + } + Ok(built) +} + +/// One column, read out of whatever the caller wrote it as. +/// +/// A typed array is read as the numbers it already holds, which is one +/// pass over memory and no runtime call per value. An ordinary array is +/// read a value at a time, where the first value settles what the column +/// is and every value after it has to agree, which is [`Column`]'s rule +/// and is the appender's rule too. +fn column(env: &Env, name: &str, values: Unknown<'_>) -> Result { + let refuse = |message: String| crate::error::usage(env, message); + if values.get_type()? == ValueType::Object && values.is_typedarray()? { + return filled(env, name, values); + } + if !values.is_array()? { + return Err(refuse(format!( + "column '{name}' is {}, and a column is an array or a typed array of its values", + named(&values) + ))); + } + let values = Object::from_unknown(values)?; + let rows = values.get_array_length()?; + let mut column: Option = None; + for row in 0..rows { + let value: Unknown<'_> = values.get_element(row)?; + match column.as_mut() { + Some(column) => column.widening_push(env, value).map_err(|why| match why { + Mismatch::Wanted(holds) => refuse(format!( + "column '{name}' holds {holds} and row {row} is {}", + named(&value) + )), + Mismatch::Says(reason) => refuse(format!( + "row {row} does not go in column '{name}': {reason}" + )), + Mismatch::Boundary(err) => err, + })?, + None => { + column = Some( + Column::start(env, value) + .map_err(|why| match why { + Mismatch::Boundary(err) => err, + Mismatch::Wanted(holds) => refuse(format!( + "column '{name}' holds {holds} and row {row} is {}", + named(&value) + )), + Mismatch::Says(reason) => refuse(format!( + "row {row} does not go in column '{name}': {reason}" + )), + })? + .ok_or_else(|| { + refuse(format!( + "column '{name}' starts at row {row} with {}, and a loaded column \ + holds booleans, integers, floats, strings, dates, times, \ + datetimes or durations", + named(&value) + )) + })?, + ); + } + } + } + column.ok_or_else(|| { + refuse(format!( + "column '{name}' is empty, and an empty column says nothing about what it would hold" + )) + }) +} + +/// A column read straight out of a typed array. +/// +/// Every width goes in as the INT64 or the FLOAT64 the store keeps, so +/// what a caller saves by handing over an `Int32Array` is the runtime +/// call per value rather than the width on disk. A load writes the file a +/// statement will read, and a statement reads whole numbers and floats. +fn filled(env: &Env, name: &str, values: Unknown<'_>) -> Result { + let kind = TypedArray::from_unknown(values)?.typed_array_type; + let widen = |v: &[i64]| Column::Int(v.to_vec()); + Ok(match kind { + TypedArrayType::Int8 => Column::Int(cast(Int8Array::from_unknown(values)?.as_ref())), + TypedArrayType::Uint8 | TypedArrayType::Uint8Clamped => { + Column::Int(cast(Uint8Array::from_unknown(values)?.as_ref())) + } + TypedArrayType::Int16 => Column::Int(cast(Int16Array::from_unknown(values)?.as_ref())), + TypedArrayType::Uint16 => Column::Int(cast(Uint16Array::from_unknown(values)?.as_ref())), + TypedArrayType::Int32 => Column::Int(cast(Int32Array::from_unknown(values)?.as_ref())), + TypedArrayType::Uint32 => Column::Int(cast(Uint32Array::from_unknown(values)?.as_ref())), + TypedArrayType::BigInt64 => widen(BigInt64Array::from_unknown(values)?.as_ref()), + TypedArrayType::BigUint64 => { + let array = BigUint64Array::from_unknown(values)?; + let mut out = Vec::with_capacity(array.as_ref().len()); + for (row, &n) in array.as_ref().iter().enumerate() { + // Refused by the row that holds it, because a column of + // unsigned words is one a caller may well have built + // without ever going near 2^63 and the one value that + // did is the thing worth naming. + if n > i64::MAX as u64 { + return Err(crate::error::usage( + env, + format!( + "column '{name}' holds {n} at row {row}, which is past what INT64 \ + holds, and every whole number this engine stores is an INT64" + ), + )); + } + out.push(n as i64); + } + Column::Int(out) + } + TypedArrayType::Float32 => Column::Float( + Float32Array::from_unknown(values)? + .as_ref() + .iter() + .map(|&n| f64::from(n)) + .collect(), + ), + TypedArrayType::Float64 => Column::Float(Float64Array::from_unknown(values)?.to_vec()), + _ => { + return Err(crate::error::usage( + env, + format!( + "column '{name}' is a typed array of a kind this client does not read, and a \ + column of numbers is one of the ten integer and float widths" + ), + )); + } + }) +} + +/// Every element of a narrower integer array, as the INT64 it becomes. +fn cast>(values: &[T]) -> Vec { + values.iter().map(|&n| n.into()).collect() +} + +/// The edge list, as the pairs of row numbers it is. +/// +/// An edge naming a row the table has not got is refused here rather than +/// written, because a graph builder handed one would either invent the +/// row or lose the edge and neither is what the caller meant. +fn pairs(env: &Env, options: &Object<'_>, rows: u64) -> Result> { + let Some(edges) = options.get::>("edges")? else { + return Ok(Vec::new()); + }; + if edges.get_type()? == ValueType::Null { + return Ok(Vec::new()); + } + let refuse = |message: String| crate::error::usage(env, message); + let within = |at: usize, end: i64| -> Result { + match end < 0 || end as u64 >= rows { + true => Err(refuse(format!( + "edge {at} joins row {end} of a table with {rows} rows in it" + ))), + false => Ok(end as u32), + } + }; + + // A flat array of row numbers, which is the shape a program that + // built its edges in memory already has and the one that costs + // nothing to read: two elements an edge, no object per edge, and one + // pass over a buffer the runtime never has to be asked about again. + if edges.get_type()? == ValueType::Object && edges.is_typedarray()? { + let flat = flat(edges)?.ok_or_else(|| { + refuse( + "the edge list is a typed array of a kind this client does not read, and a flat \ + edge list is an Int32Array or a Uint32Array of row numbers" + .to_string(), + ) + })?; + if !flat.len().is_multiple_of(2) { + return Err(refuse(format!( + "the edge list is a flat array of {} row numbers, and an edge is a pair of them", + flat.len() + ))); + } + let mut pairs = Vec::with_capacity(flat.len() / 2); + for (at, edge) in flat.chunks_exact(2).enumerate() { + pairs.push((within(at, edge[0])?, within(at, edge[1])?)); + } + return Ok(pairs); + } + + if !edges.is_array()? { + return Err(refuse(format!( + "the edge list is {}, and an edge list is an array of pairs of row numbers or a flat \ + typed array of them", + named(&edges) + ))); + } + let edges = Object::from_unknown(edges)?; + let count = edges.get_array_length()?; + let mut pairs = Vec::with_capacity(count as usize); + for at in 0..count { + let edge: Unknown<'_> = edges.get_element(at)?; + let at = at as usize; + let pair = match edge.is_array()? { + true => Object::from_unknown(edge)?, + false => { + return Err(refuse(format!( + "edge {at} is {}, and an edge is a pair of row numbers", + named(&edge) + ))); + } + }; + if pair.get_array_length()? != 2 { + return Err(refuse(format!( + "edge {at} holds {} row numbers, and an edge is a pair of them", + pair.get_array_length()? + ))); + } + let mut ends = [0u32; 2]; + for (end, slot) in ends.iter_mut().enumerate() { + let value: Unknown<'_> = pair.get_element(end as u32)?; + let n = match value.get_type()? { + ValueType::Number => f64::from_unknown(value)?, + ValueType::BigInt => { + let (n, lossless) = BigInt::from_unknown(value)?.get_i64(); + match lossless { + true => n as f64, + false => { + return Err(refuse(format!( + "edge {at} joins a row past what INT64 holds" + ))); + } + } + } + _ => { + return Err(refuse(format!( + "edge {at} joins {}, and a row is named by its number", + named(&value) + ))); + } + }; + if n.fract() != 0.0 { + return Err(refuse(format!( + "edge {at} joins row {n}, and a row number is a whole one" + ))); + } + *slot = within(at, n as i64)?; + } + pairs.push((ends[0], ends[1])); + } + Ok(pairs) +} + +/// A flat edge list as the row numbers it holds, or `None` for a typed +/// array that is not a run of them. +/// +/// The two widths a row number is written in and no others: a row is +/// numbered from zero and a table has fewer than 2^32 rows, so a +/// `Float64Array` of edges is a caller who meant something else. +fn flat(edges: Unknown<'_>) -> Result>> { + Ok(match TypedArray::from_unknown(edges)?.typed_array_type { + TypedArrayType::Int32 => Some(cast(Int32Array::from_unknown(edges)?.as_ref())), + TypedArrayType::Uint32 => Some(cast(Uint32Array::from_unknown(edges)?.as_ref())), + _ => None, + }) +} diff --git a/test/exports.test.mjs b/test/exports.test.mjs index cfba292..83edef9 100644 --- a/test/exports.test.mjs +++ b/test/exports.test.mjs @@ -18,6 +18,7 @@ const cjs = require('../zudb.cjs') // exactly what a test derived from one of them cannot see. const SURFACE = [ 'connect', + 'load', 'version', 'abiVersion', 'isZuError', diff --git a/test/load.test.mjs b/test/load.test.mjs new file mode 100644 index 0000000..9caaa90 --- /dev/null +++ b/test/load.test.mjs @@ -0,0 +1,407 @@ +// Building a database out of columns and an edge list. +// +// A load is the only way a JavaScript program makes a graph with edges +// in it, so these check both halves: that what went in comes back out +// through statements, and that a load which cannot mean anything is +// refused where the mistake is rather than written to disk and found +// later. + +import assert from 'node:assert/strict' +import { mkdtemp, rm, stat } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import test from 'node:test' + +import { ZuDate, ZuDuration, ZuTime, ZuTimestamp, connect, load } from 'zudb' + +import { isZuError } from './helper.mjs' + +// A path in a directory of its own that nothing has written to yet, +// which is what a load wants: it builds a database rather than adding +// to one. +async function spot(t, name = 'g.zu1') { + const dir = await mkdtemp(join(tmpdir(), 'zu-node-load-')) + t.after(() => rm(dir, { recursive: true, force: true })) + return join(dir, name) +} + +// A connection to a database a load just wrote, closed when the test +// ends. Read-only, because nothing here writes to one afterwards and a +// read-only open is the one a reader would use. +async function opened(t, path) { + const conn = await connect(path, { readOnly: true }) + t.after(() => conn.close()) + return conn +} + +// The graph most of these ask questions about: three people and the two +// edges between them, in the order they were loaded. +async function three(t) { + const path = await spot(t) + await load(path, { + nodes: 'person', + rels: 'knows', + columns: { uid: [10, 20, 30], name: ['ada', 'grace', 'kay'] }, + edges: [ + [0, 1], + [1, 2], + ], + }) + return { path, conn: await opened(t, path) } +} + +test('a load says what it wrote', async (t) => { + const stats = await load(await spot(t), { + nodes: 'person', + rels: 'knows', + columns: { uid: [1, 2, 3], name: ['ada', 'grace', 'kay'] }, + edges: [ + [0, 1], + [1, 2], + ], + }) + assert.deepEqual(stats, { nodes: 3, rels: 2, columns: 2 }) +}) + +test('the rows read back in the order they went in', async (t) => { + const { conn } = await three(t) + const rows = await conn.query('MATCH (p:person) RETURN p.uid AS uid, p.name AS name') + assert.deepEqual( + rows.map((row) => [row.uid, row.name]), + [ + [10n, 'ada'], + [20n, 'grace'], + [30n, 'kay'], + ], + ) +}) + +test('the edges are a table a pattern can walk', async (t) => { + const { conn } = await three(t) + const rows = await conn.query( + 'MATCH (a:person)-[:knows]->(b:person) RETURN a.name AS a, b.name AS b', + ) + assert.deepEqual( + rows.map((row) => [row.a, row.b]), + [ + ['ada', 'grace'], + ['grace', 'kay'], + ], + ) +}) + +test('an edge comes back as a rel that knows its table', async (t) => { + const { conn } = await three(t) + const rows = await conn.query('MATCH ()-[r:knows]->() RETURN r AS r') + // The ordinal is the edge's place in the load, which is where its + // properties sit, so the second edge loaded is the second ordinal. + assert.deepEqual( + rows.map((row) => [row.r.table, row.r.src, row.r.dst, row.r.ord]), + [ + ['knows', 0n, 1n, 0n], + ['knows', 1n, 2n, 1n], + ], + ) +}) + +test('a walk comes back as a path', async (t) => { + const { conn } = await three(t) + const rows = await conn.query( + 'MATCH q = (a:person)-[:knows]->()-[:knows]->(c:person) RETURN q AS q', + ) + const walk = rows[0].q + assert.deepEqual( + walk.nodes.map((node) => node.offset), + [0n, 1n, 2n], + ) + assert.deepEqual( + walk.rels.map((rel) => [rel.src, rel.dst]), + [ + [0n, 1n], + [1n, 2n], + ], + ) +}) + +test('a load with no edges is a graph with none', async (t) => { + const path = await spot(t) + const stats = await load(path, { nodes: 'person', rels: 'knows', columns: { uid: [1, 2] } }) + assert.equal(stats.rels, 0) + const conn = await opened(t, path) + const rows = await conn.query('MATCH ()-[r:knows]->() RETURN count(r) AS n') + assert.equal(rows[0].n, 0n) +}) + +test('a load with no columns still has rows', async (t) => { + const path = await spot(t) + const stats = await load(path, { nodes: 'person', rels: 'knows', rows: 4, edges: [[0, 3]] }) + assert.deepEqual(stats, { nodes: 4, rels: 1, columns: 0 }) + const conn = await opened(t, path) + const rows = await conn.query('MATCH (p:person) RETURN count(p) AS n') + assert.equal(rows[0].n, 4n) +}) + +test('the rel table is called rel when it is not named', async (t) => { + const path = await spot(t) + await load(path, { nodes: 'person', columns: { uid: [1, 2] }, edges: [[0, 1]] }) + const conn = await opened(t, path) + const rows = await conn.query('MATCH ()-[r]->() RETURN r AS r') + assert.equal(rows[0].r.table, 'rel') +}) + +test('the same edge twice is one edge', async (t) => { + const stats = await load(await spot(t), { + nodes: 'person', + rels: 'knows', + columns: { uid: [1, 2] }, + edges: [ + [0, 1], + [0, 1], + [0, 1], + ], + }) + assert.equal(stats.rels, 1) +}) + +test('an edge list is read out of a flat typed array too', async (t) => { + // Two elements an edge, which is the shape a program that built its + // edges in memory already has, and it means the same graph as the + // arrays of pairs above. + const path = await spot(t) + const stats = await load(path, { + nodes: 'person', + rels: 'knows', + columns: { uid: [1, 2, 3] }, + edges: new Uint32Array([0, 1, 1, 2]), + }) + assert.equal(stats.rels, 2) + const conn = await opened(t, path) + const rows = await conn.query('MATCH (a:person)-[:knows]->(b:person) RETURN b.uid AS uid') + assert.deepEqual( + rows.map((row) => row.uid), + [2n, 3n], + ) +}) + +test('a column is read out of a typed array as the numbers it holds', async (t) => { + const path = await spot(t) + await load(path, { + nodes: 'person', + rels: 'knows', + columns: { + small: new Int8Array([-1, 2]), + wide: new BigInt64Array([1n << 40n, -5n]), + ratio: new Float32Array([0.5, 1.25]), + exact: new Float64Array([0.1, 2.5]), + }, + }) + const conn = await opened(t, path) + const rows = await conn.query( + 'MATCH (p:person) RETURN p.small AS small, p.wide AS wide, p.ratio AS ratio, p.exact AS exact', + ) + assert.deepEqual( + rows.map((row) => [row.small, row.wide, row.ratio, row.exact]), + [ + [-1n, 1n << 40n, 0.5, 0.1], + [2n, -5n, 1.25, 2.5], + ], + ) +}) + +test('a column of every kind reads back as what it was', async (t) => { + const path = await spot(t) + const columns = { + count: [1, -2], + ratio: [1.5, -0.25], + flag: [true, false], + name: ['ada', 'grace'], + born: [new ZuDate(-56_312), new ZuDate(-23_032)], + woke: [new ZuTime(23_400_000_000_000n), new ZuTime(86_399_000_000_000n)], + seen: [new ZuTimestamp(1_704_164_645_000_006_000n), new ZuTimestamp(0n)], + took: [ZuDuration.ofNanos(86_402_000_000_000n), ZuDuration.ofNanos(0n)], + aged: [ZuDuration.ofMonths(14n), ZuDuration.ofMonths(-1n)], + } + await load(path, { nodes: 'person', rels: 'knows', columns }) + const conn = await opened(t, path) + const names = Object.keys(columns) + const rows = await conn.query( + `MATCH (p:person) RETURN ${names.map((name) => `p.${name} AS ${name}`).join(', ')}`, + ) + assert.equal(rows[0].count, 1n) + assert.equal(rows[0].ratio, 1.5) + assert.equal(rows[0].flag, true) + assert.equal(rows[0].name, 'ada') + assert.equal(rows[0].born.days, -56_312) + assert.equal(rows[0].woke.nanos, 23_400_000_000_000n) + assert.equal(rows[0].seen.nanos, 1_704_164_645_000_006_000n) + assert.equal(rows[0].took.nanos, 86_402_000_000_000n) + assert.equal(rows[0].aged.months, 14n) + assert.deepEqual( + [rows[1].count, rows[1].ratio, rows[1].flag, rows[1].name], + [-2n, -0.25, false, 'grace'], + ) +}) + +test('a column of whole numbers that meets a fractional one widens', async (t) => { + const path = await spot(t) + await load(path, { nodes: 'person', rels: 'knows', columns: { n: [1, 2, 2.5] } }) + const conn = await opened(t, path) + const rows = await conn.query('MATCH (p:person) RETURN p.n AS n') + assert.deepEqual( + rows.map((row) => row.n), + [1, 2, 2.5], + ) +}) + +test('a load never writes over a database that is there', async (t) => { + const path = await spot(t) + await load(path, { nodes: 'person', rels: 'knows', columns: { uid: [1] } }) + await assert.rejects( + () => load(path, { nodes: 'person', rels: 'knows', columns: { uid: [2] } }), + (err) => isZuError(err, 'ZuConnectionError') || isZuError(err, 'ZuIOError'), + ) + const conn = await opened(t, path) + const rows = await conn.query('MATCH (p:person) RETURN p.uid AS uid') + assert.deepEqual( + rows.map((row) => row.uid), + [1n], + ) +}) + +// Every way of writing a load that cannot mean anything, refused before +// anything is written. The file is checked afterwards in each case, +// because a refusal that left a database behind would be a refusal that +// made the next call fail for a reason the caller could not see. +const refusals = [ + [{ nodes: '', rels: 'knows', rows: 1 }, /not a name a statement can carry/], + [{ nodes: 'person', rels: '', rows: 1 }, /not a name a statement can carry/], + [{ nodes: 'person', rels: 'knows' }, /has to be told how many/], + [{ nodes: 'person', rels: 'knows', columns: { a: [1, 2], b: [3] } }, /as wide as it is long/], + [{ nodes: 'person', rels: 'knows', columns: { a: [] } }, /is empty/], + [{ nodes: 'person', rels: 'knows', columns: { a: [1, 2] }, rows: 3 }, /against the 3 rows/], + [ + { nodes: 'person', rels: 'knows', columns: { a: [1, 2] }, edges: [[0, 5]] }, + /row 5 of a table with 2 rows/, + ], + [ + { nodes: 'person', rels: 'knows', columns: { a: [1, 2] }, edges: [[0, -1]] }, + /row -1 of a table/, + ], + [ + { nodes: 'person', rels: 'knows', columns: { a: [1, 2] }, edges: [[0, 1], 7] }, + /edge 1 is a number, and an edge is a pair of row numbers/, + ], + [ + { nodes: 'person', rels: 'knows', columns: { a: [1, 2] }, edges: new Uint32Array([0, 1, 1]) }, + /flat array of 3 row numbers/, + ], + [{ nodes: 'person', rels: 'knows', columns: { '2legs': [1] } }, /not a name a statement can carry/], + [{ rels: 'knows', rows: 1 }, /names the node table/], + [{ nodes: 'person', rels: 'knows', columns: { a: [new Uint8Array([1])] } }, /byte strings/], + [{ nodes: 'person', rels: 'knows', columns: { a: 7 } }, /a column is an array or a typed array/], + [{ nodes: 'person', rels: 'knows', rows: 1.5 }, /not a whole one/], +] + +for (const [options, message] of refusals) { + test(`a load that cannot mean anything is refused: ${message.source}`, async (t) => { + const path = await spot(t) + await assert.rejects( + () => load(path, options), + (err) => isZuError(err, 'ZuUsageError') && message.test(err.message), + ) + await assert.rejects(() => stat(path), { code: 'ENOENT' }) + }) +} + +// A column holds one kind of value, and the first value is what says +// which kind. The message names the column and the row, because a +// million row load that stops at row 700_000 is worth telling where. +const mixed = [ + [[1, true], /column 'a' holds whole numbers and row 1 is a boolean/], + [[true, 1], /column 'a' holds booleans and row 1 is a number/], + [[1, 'ada'], /column 'a' holds whole numbers and row 1 is a string/], + [[1.5, 'ada'], /column 'a' holds floats and row 1 is a string/], + [['ada', 1], /column 'a' holds strings and row 1 is a number/], + [ + [new ZuDate(0), new ZuTimestamp(0n)], + /column 'a' holds dates and row 1 is a ZuTimestamp/, + ], + [ + [ZuDuration.ofMonths(1n), ZuDuration.ofNanos(1n)], + /column 'a' holds year-month durations and row 1 is a ZuDuration/, + ], + [[null], /starts at row 0 with null/], + [[{}], /starts at row 0 with an Object/], +] + +for (const [values, message] of mixed) { + test(`a column holds one kind of value: ${message.source}`, async (t) => { + const path = await spot(t) + await assert.rejects( + () => load(path, { nodes: 'person', rels: 'knows', columns: { a: values } }), + (err) => isZuError(err, 'ZuUsageError') && message.test(err.message), + ) + }) +} + +test('a whole number past what INT64 holds is refused by the row that holds it', async (t) => { + const path = await spot(t) + await assert.rejects( + () => + load(path, { + nodes: 'person', + rels: 'knows', + columns: { a: new BigUint64Array([1n, 1n << 63n]) }, + }), + (err) => isZuError(err, 'ZuUsageError') && /at row 1, which is past what INT64 holds/.test(err.message), + ) +}) + +test('the path is a string and the options are an object', async (t) => { + const path = await spot(t) + await assert.rejects( + () => load(7, { nodes: 'person', rows: 1 }), + (err) => isZuError(err, 'ZuUsageError') && /the path is a Number/.test(err.message), + ) + await assert.rejects( + () => load(path, 'person'), + (err) => isZuError(err, 'ZuUsageError') && /the options are a string/.test(err.message), + ) +}) + +test('the event loop keeps turning while a load runs', async (t) => { + // Big enough that the write takes long enough to watch, and shaped so + // the edges are out of order and have to be sorted, which is the other + // half of the work the threadpool is for. + const rows = 200_000 + const uid = BigInt64Array.from({ length: rows }, (_, ix) => BigInt(ix)) + const name = Array.from({ length: rows }, (_, ix) => `p${ix}`) + const edges = new Uint32Array(rows * 2) + for (let ix = 0; ix < rows; ix++) { + edges[ix * 2] = ix + edges[ix * 2 + 1] = (ix * 7 + 1) % rows + } + + let ticks = 0 + const tick = () => { + ticks++ + return new Promise((resolve) => setImmediate(resolve)) + } + const writing = load(await spot(t, 'big.zu1'), { + nodes: 'person', + rels: 'knows', + columns: { uid, name }, + edges, + }) + let done = false + writing.then(() => { + done = true + }) + while (!done) await tick() + + const stats = await writing + assert.equal(stats.nodes, rows) + // An event loop held for the length of the write would get no turns + // at all, since the whole of it happens inside the one call. + assert.ok(ticks > 50, `the loop only got ${ticks} turns`) +}) diff --git a/test/readme.test.mjs b/test/readme.test.mjs index 54c37c6..c6e34d3 100644 --- a/test/readme.test.mjs +++ b/test/readme.test.mjs @@ -81,7 +81,7 @@ async function installed(t, program) { } test('the README prints programs and fragments and knows which is which', async () => { - assert.equal((await programs()).length, 1, "the README's whole programs") + assert.equal((await programs()).length, 2, "the README's whole programs") assert.ok((await blocks('ts')).length > (await programs()).length, 'and its fragments') }) diff --git a/test/types/cjs.cts b/test/types/cjs.cts index b32ac33..01e3734 100644 --- a/test/types/cjs.cts +++ b/test/types/cjs.cts @@ -5,9 +5,11 @@ import { connect, isZuError, + load, ZuTimestamp, type ZuAppendValue, type ZuArrowTable, + type ZuLoadOptions, type ZuParam, type ZuStream, type ZuTransactionOptions, @@ -88,6 +90,14 @@ export async function bulk(path: string, batch: readonly ZuAppendValue[][]): Pro } } +export async function graph(path: string, options: ZuLoadOptions): Promise { + // The options are an object of their own, so a caller can build one + // and pass it around, which is what a loader reading a manifest ends + // up doing. + const stats = await load(path, options) + return stats.nodes +} + export async function scanned(path: string, table: ZuArrowTable): Promise { // A table is taken by shape rather than by class, so a caller holding // an `apache-arrow` Table passes it here without this package having diff --git a/test/types/esm.mts b/test/types/esm.mts index 3595497..6791796 100644 --- a/test/types/esm.mts +++ b/test/types/esm.mts @@ -5,14 +5,17 @@ import { connect, isZuError, + load, ZuDate, type Appender, type ZuAppendValue, type ZuBatch, type ZuBigIntMode, type ZuError, + type ZuEdges, type ZuFrame, type ZuFrameColumn, + type ZuLoadStats, type ZuPlainDate, type ZuRows, type ZuStream, @@ -134,6 +137,29 @@ export async function loaded(path: string, people: [bigint, string][]): Promise< return written + rows.committed } +export async function built(path: string, uid: BigInt64Array): Promise { + // Both spellings of an edge list are the one type, so a program that + // has its edges flat and a program that has them in pairs both + // compile without either of them saying which they meant. + const pairs: ZuEdges = [ + [0, 1], + [1, 2], + ] + const flat: ZuEdges = new Uint32Array([0, 1, 1, 2]) + + const stats: ZuLoadStats = await load(path, { + nodes: 'person', + rels: 'knows', + columns: { uid, name: ['ada', 'grace', 'kay'] }, + edges: uid.length === 3 ? pairs : flat, + }) + + // Numbers rather than bigints, which is the one place in this package + // a count is not a `bigint`, since these are counts the client made + // rather than an INT64 a statement gave back. + return stats.nodes + stats.rels + stats.columns +} + export async function matched(path: string, ages: Int32Array): Promise { await using conn = await connect(path) diff --git a/types/header.d.ts b/types/header.d.ts index dfd8bb5..127f4bc 100644 --- a/types/header.d.ts +++ b/types/header.d.ts @@ -221,6 +221,48 @@ export interface ZuArrowTable { */ export type ZuFrame = ZuArrowTable | Record +/** + * An edge list, as the pairs of row numbers it is. + * + * Rows are numbered from zero in the order their columns were written, + * because at load time a row has no other name. The flat spelling is + * there for a program that built its edges in memory and would rather + * not make a million two element arrays to hand them over: two elements + * an edge, read in one pass. + */ +export type ZuEdges = readonly (readonly [number, number])[] | Int32Array | Uint32Array + +/** + * What a load writes, and how much of it. + */ +export interface ZuLoadOptions { + /** The node table, which gets a row per element of every column. */ + readonly nodes: string + /** The rel table holding the edges between those rows. `rel` by default. */ + readonly rels?: string + /** The node table's properties, as column name to values. */ + readonly columns?: Readonly> + /** The edges, as pairs of row numbers. */ + readonly edges?: ZuEdges | null + /** + * How many rows the node table has. + * + * Read off the columns when there are any, so this is for the load + * that writes a graph with no properties at all, and a check on the + * columns when both are given. + */ + readonly rows?: number +} + +/** + * What went into a load. + */ +export interface ZuLoadStats { + readonly nodes: number + readonly rels: number + readonly columns: number +} + /** * A walk through the graph: nodes and edges, alternating, a node at * each end. diff --git a/zudb.cjs b/zudb.cjs index 548e789..0212986 100644 --- a/zudb.cjs +++ b/zudb.cjs @@ -159,6 +159,7 @@ function isZuError(value) { module.exports = { connect: binding.connect, + load: binding.load, version: binding.version, abiVersion: binding.abiVersion, isZuError, diff --git a/zudb.mjs b/zudb.mjs index afc7b83..3167250 100644 --- a/zudb.mjs +++ b/zudb.mjs @@ -17,6 +17,7 @@ const require = createRequire(import.meta.url) const zudb = require('./zudb.cjs') export const connect = zudb.connect +export const load = zudb.load export const version = zudb.version export const abiVersion = zudb.abiVersion export const isZuError = zudb.isZuError