diff --git a/Cargo.lock b/Cargo.lock index c2b0355..219fbba 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1277,7 +1277,7 @@ dependencies = [ [[package]] name = "zu" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=1b6575a9f258d467d7fce56ebd594b42d247789f#1b6575a9f258d467d7fce56ebd594b42d247789f" +source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" dependencies = [ "zu-common", "zu-encoding", @@ -1293,7 +1293,7 @@ dependencies = [ [[package]] name = "zu-common" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=1b6575a9f258d467d7fce56ebd594b42d247789f#1b6575a9f258d467d7fce56ebd594b42d247789f" +source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" dependencies = [ "thiserror", ] @@ -1301,7 +1301,7 @@ dependencies = [ [[package]] name = "zu-encoding" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=1b6575a9f258d467d7fce56ebd594b42d247789f#1b6575a9f258d467d7fce56ebd594b42d247789f" +source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" dependencies = [ "ruzstd", "zu-common", @@ -1310,7 +1310,7 @@ dependencies = [ [[package]] name = "zu-exec" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=1b6575a9f258d467d7fce56ebd594b42d247789f#1b6575a9f258d467d7fce56ebd594b42d247789f" +source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" dependencies = [ "zu-common", "zu-query", @@ -1320,7 +1320,7 @@ dependencies = [ [[package]] name = "zu-query" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=1b6575a9f258d467d7fce56ebd594b42d247789f#1b6575a9f258d467d7fce56ebd594b42d247789f" +source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" dependencies = [ "crossbeam-deque", "zu-common", @@ -1331,7 +1331,7 @@ dependencies = [ [[package]] name = "zu-s3" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=1b6575a9f258d467d7fce56ebd594b42d247789f#1b6575a9f258d467d7fce56ebd594b42d247789f" +source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" dependencies = [ "crc32c", "object_store", @@ -1342,7 +1342,7 @@ dependencies = [ [[package]] name = "zu-sqlite" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=1b6575a9f258d467d7fce56ebd594b42d247789f#1b6575a9f258d467d7fce56ebd594b42d247789f" +source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" dependencies = [ "rusqlite", "zu-common", @@ -1352,7 +1352,7 @@ dependencies = [ [[package]] name = "zu-storage" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=1b6575a9f258d467d7fce56ebd594b42d247789f#1b6575a9f258d467d7fce56ebd594b42d247789f" +source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" dependencies = [ "zu-common", "zu-encoding", @@ -1361,7 +1361,7 @@ dependencies = [ [[package]] name = "zu-vector" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=1b6575a9f258d467d7fce56ebd594b42d247789f#1b6575a9f258d467d7fce56ebd594b42d247789f" +source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" dependencies = [ "zu-common", ] @@ -1369,7 +1369,7 @@ dependencies = [ [[package]] name = "zu-zu1" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=1b6575a9f258d467d7fce56ebd594b42d247789f#1b6575a9f258d467d7fce56ebd594b42d247789f" +source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" dependencies = [ "crc32c", "loom", diff --git a/Cargo.toml b/Cargo.toml index 6d03df7..432f1e0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,8 +18,8 @@ crate-type = ["cdylib"] # with (ADR 0002), so a revision is the honest way to say which one. # A local checkout is used instead with a `paths` override in # `.cargo/config.toml`, which is untracked on purpose. -zudb = { package = "zu", git = "https://github.com/tamnd/zu", rev = "1b6575a9f258d467d7fce56ebd594b42d247789f" } -zu-common = { git = "https://github.com/tamnd/zu", rev = "1b6575a9f258d467d7fce56ebd594b42d247789f" } +zudb = { package = "zu", git = "https://github.com/tamnd/zu", rev = "92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" } +zu-common = { git = "https://github.com/tamnd/zu", rev = "92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" } # N-API by way of napi-rs (ADR 0002). `napi9` is the version of N-API # this addon declares it needs, which is what makes one binary work # across Node 24, Node 26, Electron and Bun without a rebuild: the diff --git a/README.md b/README.md index 315b463..70814f6 100644 --- a/README.md +++ b/README.md @@ -35,7 +35,7 @@ The rows are an array, so iterating them is `for (const row of rows)` and nothin ## What works today -`connect`, `query`, `exec`, `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`. Read-only connections, memory and thread limits. An `AbortSignal` on any statement. The full error surface above, and `isZuError` to recognize it. 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`. Read-only connections, memory and thread limits. 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. 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. @@ -51,6 +51,29 @@ It is the signal JavaScript already has, so a timeout written like the one above What the promise rejects with is the signal's own reason, which is what `fetch` does: `AbortSignal.timeout(50)` rejects with the runtime's `TimeoutError`, `controller.abort(new RequestGone())` rejects with the `RequestGone` you made, and a bare `controller.abort()` rejects with the runtime's `AbortError`. A signal that has already fired stops the statement before the engine sees it at all. A signal that never fires costs one listener, taken off again when the statement ends, whether it answered, failed or was stopped. +## Reading a result a piece at a time + +`conn.stream(...)` runs the same statement and hands the rows over as they are made, instead of building the whole answer first: + +```ts +await using stream = conn.stream<{ id: bigint; name: string }>( + `MATCH (p:Person) RETURN p.id AS id, p.name AS name`, +); +for await (const { id, name } of stream) { + if (name === "ada") break; // the scan under it stops here +} + +stream.summary; // { columns, rows, stopped, streamed, notices } +``` + +Three ways to read it, all the same statement read once. `for await` over the stream gives one row at a time. `stream.batches()` gives the array the rows crossed the boundary in, with `columns` beside it, which is what to reach for when the work is per batch rather than per row. `stream.toReadableStream()` gives a `ReadableStream` for anything that already speaks Web Streams, and its backpressure is the reader's: nothing is pulled from the database until what is in front of it has drained. + +Ending early is the case worth knowing about, because it is the reason streaming is different from `query`. A `break`, a `throw`, a `return()` on the iterator, a `cancel()`, or leaving the block of an `await using` all stop the statement and wait for it to let go of the connection, so the next statement on that connection runs rather than queueing behind a scan nobody is reading. The rows already read stand, and `summary.stopped` says the reader stopped it. The statement itself does not start until the first read, so a stream made and never read is not a scan holding anything. + +Between the statement and the loop sit two batches, which is the whole of the buffering: a reader slower than the scan stops the scan rather than filling memory behind it. `{ batchRows: 512 }` sets what a batch may hold, which is what to name when the rows are going somewhere with a size of its own. On 50k rows here a stream costs about 460ns a row against 370ns for `query`, reading a batch at a time costs about 320ns, and reading the first batch and stopping costs 1.1ms against 18.6ms for the whole scan, which is what the whole thing is for. + +A statement that has to see every row before it can give one, which is `ORDER BY`, `DISTINCT` and the aggregates, runs whole and is handed over in batches afterwards. The loop is the same either way and `summary.streamed` is what tells them apart. + ## Importing it, either way ```ts @@ -83,7 +106,7 @@ Anything outside that table has no binary and no source build to fall back on, s ## Still to come -`AsyncIterable` and Web Streams over a result. `bigIntMode`. `toTemporal()` and `{ temporal: true }`, for the runtimes where Temporal is unflagged: it reached Stage 4 in March 2026 and is unflagged in Node 26, but Node 24 is still the active LTS and Safari is still behind a flag, which is why the stable types are the four classes above. Bun and Deno in CI, and the WASM build for the browser. +`bigIntMode`. `toTemporal()` and `{ temporal: true }`, for the runtimes where Temporal is unflagged: it reached Stage 4 in March 2026 and is unflagged in Node 26, but Node 24 is still the active LTS and Safari is still behind a flag, which is why the stable types are the four classes above. Bun and Deno in CI, and the WASM build for the browser. ## Runtimes diff --git a/bench/query.mjs b/bench/query.mjs index 8c09355..3f1c5dc 100644 --- a/bench/query.mjs +++ b/bench/query.mjs @@ -83,6 +83,44 @@ const cases = [ per: 'row', run: () => conn.exec('MATCH (p:person) RETURN p.id AS id, p.name AS name'), }, + { + // The same rows through the streaming path, which is the number to + // read the two scans above against: what streaming costs is a + // thread, a queue and a promise per batch, and what it saves is + // holding the whole result. A stream read to the end is the worst + // case for it, since nothing was saved and everything was paid. + name: 'stream, two columns', + per: 'row', + run: async () => { + const stream = conn.stream('MATCH (p:person) RETURN p.id AS id, p.name AS name') + for await (const row of stream) void row + }, + }, + { + // A batch at a time rather than a row at a time, which is the same + // rows with one less iterator between them and the loop. + name: 'stream, batch at a time', + per: 'row', + run: async () => { + const stream = conn.stream('MATCH (p:person) RETURN p.id AS id, p.name AS name') + for await (const batch of stream.batches()) void batch + }, + }, + { + // What a reader that stops after one batch pays, which is what a + // stream is for: the scan under it ends, so this is a statement + // whose cost is the batch rather than the table. Per statement, + // because the rows it read are a batch and not the table. + name: 'stream, first batch only', + per: 'statement', + run: async () => { + const stream = conn.stream('MATCH (p:person) RETURN p.id AS id, p.name AS name') + for await (const batch of stream.batches()) { + void batch + break + } + }, + }, { name: 'aggregate, one row out', per: 'statement', diff --git a/binding.cjs b/binding.cjs index 527648d..9446558 100644 --- a/binding.cjs +++ b/binding.cjs @@ -701,6 +701,7 @@ if (!nativeBinding) { module.exports = nativeBinding module.exports.Connection = nativeBinding.Connection +module.exports.ZuCursor = nativeBinding.ZuCursor module.exports.ZuDate = nativeBinding.ZuDate module.exports.ZuDuration = nativeBinding.ZuDuration module.exports.ZuNode = nativeBinding.ZuNode diff --git a/binding.d.cts b/binding.d.cts index 0c7fca5..ac674f7 100644 --- a/binding.d.cts +++ b/binding.d.cts @@ -88,6 +88,66 @@ export interface ZuRows> extends Array { readonly notices: ZuNotice[] } +/** + * One batch of a streamed result. + * + * The rows of a whole result with the same array trick and one fewer + * property: `columns` is the statement's projection and is the same on + * every batch of one stream, and what a statement completed with is not + * known until it has, so it is on the summary rather than here. + */ +export interface ZuBatch> extends Array { + readonly columns: string[] +} + +/** + * What a streamed statement did, known once it has ended. + * + * The rows are gone by then, which is the point of streaming, so this + * is what is worth keeping about a result nobody held: what it + * projected, how much of it was read, whether the reader stopped it + * early, and what the engine wanted to say along the way. + */ +export interface ZuSummary { + readonly columns: string[] + /** How many rows were handed over, which is fewer than the statement + * would have returned when the reader stopped early. */ + readonly rows: number + readonly stopped: boolean + /** + * Whether the rows arrived as they were made, rather than the + * statement running whole and being handed over in batches + * afterwards. A statement that has to see every row before it can + * give one, which is `ORDER BY`, `DISTINCT`, the aggregates and + * anything that writes, is the second kind, and so is a plan the + * pipeline executor does not take. The loop over it reads the same + * either way, so this is here for a caller measuring where the time + * went rather than for one deciding what to do next. + */ + readonly streamed: boolean + readonly notices: ZuNotice[] +} + +/** + * What a streamed statement takes beside its parameters. + */ +export interface ZuStreamOptions extends ZuStatementOptions { + /** + * How many rows a batch may hold. The engine's own vector by + * default, which is the unit it already works in and the one that + * costs nothing to hand over. Name a size when the rows are going + * somewhere with a size of its own, an Arrow record batch or an + * HTTP chunk. + * + * A ceiling and not a promise: batches are cut out of rows that have + * already been made, so the last piece of a run of them is whatever + * was left, and a size above the engine's vector gets the vector. It + * is what bounds how much a reader holds at once, which is the + * question a caller is asking when they name one. + */ + readonly batchRows?: number +} + /** * What a statement takes beside its parameters. * @@ -179,6 +239,15 @@ export declare class Connection { * reads still costs a row object per row on the way out. */ exec(statement: string, params?: Record | null, options?: ZuStatementOptions | null): Promise + /** + * Runs one statement and gives back a cursor over its rows. + * + * The pull underneath `stream`, which is what a program uses. The + * statement does not start here: it starts on the first read, so + * that a cursor made and not read is not a scan holding the + * connection against every statement after it. + */ + cursor(statement: string, params?: Record | null, options?: ZuStreamOptions | null): ZuCursor /** * Closes the connection and releases the database. * @@ -199,6 +268,45 @@ export declare class Connection { dispose(): Promise } +/** + * One statement, read a batch at a time. + * + * This is the pull underneath the stream and not the shape a program + * should be reaching for: `conn.stream(...)` gives back something that + * is async-iterable, turns into the web's own `ReadableStream`, and + * stops itself when the loop over it breaks. + */ +export declare class ZuCursor { + /** + * The next batch of rows, or `null` once the statement has ended. + * + * An array of row objects with the column names beside it, which + * is the value `query` gives for a whole result and for the same + * reason: what a caller does with rows is iterate them. + */ + next(): Promise + /** + * Stops the statement and waits for it to let go of the + * connection. + * + * Waits, because a stream abandoned while the next statement on + * the same connection is already being asked for would make that + * statement queue behind a scan nobody is reading. Stopping is not + * a failure: the rows already handed over stand, and the summary + * says the statement was stopped. + */ + cancel(): Promise + /** + * What the statement did, once it has ended, and `null` until + * then. + * + * The column names are here as well as beside every batch, because + * a statement that gave back no rows gave back no batches either + * and its projection is still worth knowing. + */ + get summary(): ZuSummary | null +} + /** * A date, as days from 1970-01-01. * diff --git a/src/cancel.rs b/src/cancel.rs index 1e27088..33b0513 100644 --- a/src/cancel.rs +++ b/src/cancel.rs @@ -52,6 +52,44 @@ impl Shared { } } +/// The half of a watch a running statement uses, which is the half +/// that can be held on another thread. +/// +/// A statement in this client runs somewhere the runtime is not: on +/// libuv's threadpool, or on a thread of its own for a stream. The two +/// words it sets are plain atomics and the interrupt is the engine's, +/// so all three travel. The references to the signal and the listener +/// do not travel with them, because touching either is something only +/// the thread that owns the runtime may do. +#[derive(Clone)] +pub struct Guard(Arc); + +impl Guard { + /// The statement is about to run. `false` means it must not: the + /// signal fired first. + pub fn enter(&self) -> bool { + self.0.running.store(true, Ordering::SeqCst); + !self.0.asked.load(Ordering::SeqCst) + } + + /// The statement is done with the connection. + /// + /// The flag goes down first, so a signal firing now raises nothing, + /// and then the interrupt is put back down, so a stop that landed in + /// the moment between the statement finishing and this call cannot + /// end whatever runs next on the same connection. + pub fn leave(&self) { + self.0.running.store(false, Ordering::SeqCst); + self.0.interrupt.clear(); + } + + /// Whether the signal fired at all, which is what tells an interrupt + /// the caller asked for apart from one they did not. + pub fn asked(&self) -> bool { + self.0.asked.load(Ordering::SeqCst) + } +} + /// A signal watching one statement. /// /// It is made on the thread that owns the runtime, because that is the @@ -111,28 +149,26 @@ impl Watch { }) } + /// The part of this a statement takes with it to wherever it runs. + pub fn guard(&self) -> Guard { + Guard(Arc::clone(&self.shared)) + } + /// The statement is about to run. `false` means it must not: the /// signal fired first. pub fn enter(&self) -> bool { - self.shared.running.store(true, Ordering::SeqCst); - !self.shared.asked.load(Ordering::SeqCst) + self.guard().enter() } /// The statement is done with the connection. - /// - /// The flag goes down first, so a signal firing now raises nothing, - /// and then the interrupt is put back down, so a stop that landed in - /// the moment between the statement finishing and this call cannot - /// end whatever runs next on the same connection. pub fn leave(&self) { - self.shared.running.store(false, Ordering::SeqCst); - self.shared.interrupt.clear(); + self.guard().leave() } /// Whether the signal fired at all, which is what tells an interrupt /// the caller asked for apart from one they did not. pub fn asked(&self) -> bool { - self.shared.asked.load(Ordering::SeqCst) + self.guard().asked() } /// What the signal gives as its reason, if it gives one. diff --git a/src/conn.rs b/src/conn.rs index 85dc061..3f92a54 100644 --- a/src/conn.rs +++ b/src/conn.rs @@ -23,10 +23,11 @@ use napi::bindgen_prelude::*; use napi::{Env, ScopedTask, ValueType}; use napi_derive::napi; use zudb::query::{QueryResult, Value}; -use zudb::{Config, Database, Interrupt, ZuError}; +use zudb::{Config, Database, DiagnosticRecord, Interrupt, ZuError}; use crate::cancel::Watch; use crate::error::{aborted, raise, usage}; +use crate::stream::{self, Started, ZuCursor}; use crate::value::{Names, from_js, to_js}; /// What a connection can be opened with. @@ -242,6 +243,56 @@ impl Connection { AsyncTask::new(ExecTask(self.task(env, statement, params, options))) } + /// Runs one statement and gives back a cursor over its rows. + /// + /// The pull underneath `stream`, which is what a program uses. The + /// statement does not start here: it starts on the first read, so + /// that a cursor made and not read is not a scan holding the + /// connection against every statement after it. + #[napi( + ts_args_type = "statement: string, params?: Record | null, options?: ZuStreamOptions | null", + ts_return_type = "ZuCursor" + )] + pub fn cursor( + &self, + env: &Env, + statement: String, + params: Option>, + options: Option>, + ) -> ZuCursor { + // Read here rather than on the statement's thread, because + // reading a JavaScript value is something only the thread that + // owns the runtime may do. So is adding the listener the signal + // is watched through. + let bound = if self.alive.load(Ordering::Acquire) { + batch_rows(options.as_ref()).and_then(|batch_rows| { + Ok(( + bind(env, params)?, + batch_rows, + watch(env, options, self.interrupt.clone())?, + )) + }) + } else { + Err(CLOSED.to_string()) + }; + let (params, batch_rows, watch, refused) = match bound { + Ok((params, batch_rows, watch)) => (params, batch_rows, watch, None), + Err(message) => (Vec::new(), None, None, Some(message)), + }; + stream::open( + Started { + inner: Arc::clone(&self.inner), + alive: Arc::clone(&self.alive), + statement, + params, + batch_rows, + guard: watch.as_ref().map(Watch::guard), + }, + watch, + refused, + ) + } + /// The task one statement runs as, whether or not it is going to /// work. /// @@ -329,6 +380,23 @@ pub enum Failure { Aborted, } +/// The exception a failed statement rejects the caller's promise with. +/// +/// An abort rejects with the signal's own reason, which is what `fetch` +/// does: a caller who wrote `AbortSignal.timeout(50)` gets back the +/// `TimeoutError` that signal carries, and one who wrote +/// `controller.abort(new MyError())` gets their own object rather than +/// a description of it. +pub(crate) fn failed(env: &Env, failure: Failure, watch: Option<&Watch>) -> Error { + match failure { + Failure::Engine(err) => raise(env, err), + Failure::Usage(message) => usage(env, message), + Failure::Aborted => watch + .and_then(|watch| watch.reason(env)) + .map_or_else(|| aborted(env, ABORTED), Error::from), + } +} + impl From for Failure { fn from(err: ZuError) -> Self { Failure::Engine(err) @@ -336,7 +404,8 @@ impl From for Failure { } /// What a closed connection says, wherever it is noticed. -const CLOSED: &str = "the connection is closed, so there is nothing left to run a statement on"; +pub(crate) const CLOSED: &str = + "the connection is closed, so there is nothing left to run a statement on"; /// What an abort says when the signal that fired named no reason of its /// own, which is a signal built by hand rather than by a runtime. @@ -349,7 +418,7 @@ const ABORTED: &str = "the statement was stopped by the signal it was given"; /// `signal` that is not an `AbortSignal` is refused here rather than /// where the listener fails to be added, because the caller's mistake is /// the value they passed. -fn watch( +pub(crate) fn watch( env: &Env, options: Option>, interrupt: Interrupt, @@ -380,13 +449,52 @@ fn watch( } } +/// Reads `options.batchRows`, which is how many rows a caller wants in +/// a batch. +/// +/// Absent means the engine's own vector, which is the unit it already +/// works in and the one that costs nothing to hand over. A caller names +/// a size when the rows are going somewhere with a size of its own, an +/// HTTP chunk or a write of a fixed length, and a size of zero is a +/// stream that could never hand anything over rather than a default. +fn batch_rows(options: Option<&Object<'_>>) -> std::result::Result, String> { + let Some(options) = options else { + return Ok(None); + }; + let rows: Option = options + .get_named_property::>("batchRows") + .map_err(|err| err.reason) + .and_then(|rows| match rows.get_type().map_err(|err| err.reason)? { + ValueType::Undefined | ValueType::Null => Ok(None), + ValueType::Number => rows + .coerce_to_number() + .and_then(|rows| rows.get_double()) + .map(Some) + .map_err(|_| "batchRows is a number that cannot be read".to_string()), + other => Err(format!("batchRows is a {other}, which is not a number")), + })?; + // Read as a double and checked here rather than converted, because + // JavaScript's own narrowing to an unsigned integer turns -1 into + // four billion and 1.5 into 1, and a batch size nobody asked for is + // worse than a call that says no. + match rows { + None => Ok(None), + Some(rows) if rows.fract() == 0.0 && (1.0..=f64::from(u32::MAX)).contains(&rows) => { + Ok(Some(rows as u32)) + } + Some(rows) => Err(format!( + "batchRows is {rows}, and a batch holds a whole number of rows, one at the least" + )), + } +} + /// Reads the parameter object into the values the engine binds. /// /// Every failure comes back as the message to refuse the call with, /// including a boundary failure, because a caller who cannot be given /// the value they passed is being told the same thing either way and /// would rather hear it as a rejection than as a throw. -fn bind( +pub(crate) fn bind( env: &Env, params: Option>, ) -> std::result::Result, String> { @@ -473,22 +581,8 @@ impl QueryTask { } /// The exception this rejects the caller's promise with. - /// - /// An abort rejects with the signal's own reason, which is what - /// `fetch` does: a caller who wrote `AbortSignal.timeout(50)` gets - /// back the `TimeoutError` that signal carries, and one who wrote - /// `controller.abort(new MyError())` gets their own object rather - /// than a description of it. fn failed(&self, env: &Env, failure: Failure) -> Error { - match failure { - Failure::Engine(err) => raise(env, err), - Failure::Usage(message) => usage(env, message), - Failure::Aborted => self - .watch - .as_ref() - .and_then(|watch| watch.reason(env)) - .map_or_else(|| aborted(env, ABORTED), Error::from), - } + failed(env, failure, self.watch.as_ref()) } /// Takes the listener back off the signal, whatever happened. @@ -538,15 +632,7 @@ fn rows<'env>(env: &'env Env, result: &QueryResult, names: &Names) -> Result(env: &'env Env, result: &QueryResult, names: &Names) -> Result(env: &'env Env, raised: &[DiagnosticRecord]) -> Result> { + let mut array = env.create_array(raised.len() as u32)?; + for (ix, notice) in raised.iter().enumerate() { + let mut record = Object::new(env)?; + record.set("code", notice.status.code())?; + record.set("condition", notice.status.standard_text())?; + record.set("message", notice.detail.as_str())?; + record.set("docUrl", notice.doc_url())?; + array.set(ix as u32, record)?; + } + Ok(array) +} + /// One property that rides beside the rows rather than among them. /// /// Readable and replaceable like any other, but not enumerable, which /// is the whole difference between a result that is an array and one /// that merely looks like one. -fn beside(env: &Env, name: &str, value: T) -> Result { +pub(crate) fn beside(env: &Env, name: &str, value: T) -> Result { Ok(Property::new() .with_utf8_name(name)? .with_napi_value(env, value)? diff --git a/src/lib.rs b/src/lib.rs index accf897..801cf8c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -19,6 +19,7 @@ mod cancel; mod conn; mod error; +mod stream; mod value; /// The version of the client. diff --git a/src/stream.rs b/src/stream.rs new file mode 100644 index 0000000..5335677 --- /dev/null +++ b/src/stream.rs @@ -0,0 +1,661 @@ +//! A statement read a batch at a time, instead of all at once. +//! +//! A result that does not fit in memory is the reason this exists, and +//! a result the caller will not read all of is the reason it stops +//! properly. The engine's shape for both is a sink: it hands over a +//! batch of rows, the sink says whether it wants more, and a sink that +//! says no ends the scan at the boundary an interrupt is answered at. +//! That is a push, and JavaScript wants a pull, so the two are joined +//! by a channel of two batches and a thread of this statement's own. +//! +//! The thread rather than libuv's threadpool, because a stream lasts as +//! long as the reader takes and the threadpool has four threads in it +//! by default: a statement parked there waiting for a `for await` body +//! to come round again is a quarter of every other library's file reads +//! and DNS lookups gone. The channel is bounded because that is what +//! backpressure is. A reader that stops reading stops the scan two +//! batches later rather than buffering a database into memory behind +//! it. +//! +//! Nothing here waits on the thread that owns the runtime. The rows are +//! copied out of the batch on the statement's thread, turned into +//! JavaScript values on the runtime's thread, and every wait happens on +//! libuv's threadpool, where waiting is what the thread is for. + +use std::collections::VecDeque; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Condvar, Mutex}; +use std::time::Duration; + +use napi::bindgen_prelude::*; +use napi::{Env, ScopedTask}; +use napi_derive::napi; +use zudb::query::Value; +use zudb::{Batch, Flow, Streamed, ZuError}; + +use crate::cancel::{Guard, Watch}; +use crate::conn::{CLOSED, Failure, beside, failed, notices}; +use crate::value::{Names, to_js}; + +/// How many batches may sit between the statement and the reader. +/// +/// Two, so that one is being turned into JavaScript values while the +/// next is being made, and no more, because everything past that is +/// memory spent to hide a reader slower than the scan. It is the whole +/// of the backpressure and it is deliberately small. +const QUEUE: usize = 2; + +/// How long the statement's thread waits on a full queue before it +/// looks up to see whether anybody still wants the rows. +/// +/// Room in the queue wakes it exactly, so this is not how a batch is +/// handed over: it is how a connection closed underneath a stream gets +/// noticed, which is the one thing that changes without anybody being +/// able to say so through the queue. Short enough that closing returns +/// promptly, long enough that it is never the reason for a wakeup. +const PAUSE: Duration = Duration::from_millis(25); + +/// What a lock says when the thread that held it panicked. +const POISONED: &str = "the stream was left in an unknown state by a thread that panicked"; + +/// What every batch of one statement shares. +/// +/// The column names and the table names are the statement's rather than +/// the batch's, so they are made once and each batch carries a share of +/// them instead of a copy. +pub struct Head { + columns: Vec, + names: Arc, +} + +/// What the thread running the statement sends back. +enum Chunk { + /// Rows, copied out of the batch they were borrowed in. + Rows { + head: Arc, + rows: Vec>, + }, + /// The statement ended, and this is what it ended as. Always the + /// last thing sent. + Done(std::result::Result), +} + +/// The queue between the statement and the reader, with both ends able +/// to walk away. +/// +/// A bounded channel is what this is, and the standard library's own +/// would do but for one thing: a send that waits for room and gives up +/// on a timeout is not on the stable side of it, and giving up is how a +/// statement notices a connection closed underneath it. So the queue is +/// written out. A condition variable is what makes room and arrival +/// exact rather than polled: nothing here sleeps waiting for a batch or +/// for space, and the timeout is only the backstop for the one thing +/// that happens without a word being said. +struct Pipe { + held: Mutex, + /// One variable for both directions. There is one writer and, in + /// any program worth calling correct, one reader, so waking both is + /// waking at most one of each. + moved: Condvar, +} + +struct Held { + queue: VecDeque, + /// The reader is gone, so the statement stops. + closed: bool, + /// The statement is over, so a reader that finds nothing is at the + /// end rather than early. + done: bool, +} + +impl Pipe { + fn new() -> Pipe { + Pipe { + held: Mutex::new(Held { + queue: VecDeque::with_capacity(QUEUE), + closed: false, + done: false, + }), + moved: Condvar::new(), + } + } + + /// Hands a batch over, waiting for room. `false` means nobody is + /// reading any more. + /// + /// `alive` is the connection, and it can go false while this waits, + /// which is why the wait has a timeout at all: a close takes the + /// same lock the statement holds, so a stream that waited forever + /// for a reader that has gone would be a connection that could + /// never be closed. + fn send(&self, chunk: Chunk, alive: &AtomicBool) -> bool { + let Ok(mut held) = self.held.lock() else { + return false; + }; + while held.queue.len() >= QUEUE && !held.closed { + if !alive.load(Ordering::Acquire) { + return false; + } + let Ok((next, _)) = self.moved.wait_timeout(held, PAUSE) else { + return false; + }; + held = next; + } + if held.closed { + return false; + } + held.queue.push_back(chunk); + drop(held); + self.moved.notify_all(); + true + } + + /// The next batch, waiting for one. `None` is the end. + fn recv(&self) -> Option { + let mut held = self.held.lock().ok()?; + loop { + if let Some(chunk) = held.queue.pop_front() { + drop(held); + self.moved.notify_all(); + return Some(chunk); + } + if held.done || held.closed { + return None; + } + held = self.moved.wait(held).ok()?; + } + } + + /// The reader is gone. What is queued is dropped, because nobody + /// will ever ask for it and it is a database's worth of rows. + fn close(&self) { + if let Ok(mut held) = self.held.lock() { + held.closed = true; + held.queue.clear(); + } + self.moved.notify_all(); + } + + /// The statement is over and nothing else will be sent. + fn finish(&self) { + if let Ok(mut held) = self.held.lock() { + held.done = true; + } + self.moved.notify_all(); + } +} + +/// Everything the statement needs, held until the first read. +/// +/// A stream starts when it is first read rather than when it is asked +/// for. A statement holds the connection for as long as it runs, and a +/// stream nobody has read yet holding it would make `conn.stream(...)` +/// on a line of its own stop every other statement on that connection +/// until the garbage collector noticed. +pub struct Started { + pub inner: Arc>>, + pub alive: Arc, + pub statement: String, + pub params: Vec<(String, Value)>, + pub batch_rows: Option, + pub guard: Option, +} + +/// What a cursor and the thread feeding it share. +struct Live { + /// Where batches are handed over. + pipe: Pipe, + /// Set by a reader that wants no more rows. The statement reads it + /// between batches and stops the scan. + stopped: AtomicBool, + /// Taken by the first read, which is what starts the statement. + start: Mutex>, + /// What the statement ended as, once it has. A failure is taken out + /// to be thrown; a summary stays to be read. + ended: Mutex>>, + /// The signal watching this stream, kept here because taking it off + /// again is something only the runtime's thread may do, and a read + /// is the only thing here that happens on that thread. + watch: Mutex>, +} + +/// One statement, read a batch at a time. +/// +/// This is the pull underneath the stream and not the shape a program +/// should be reaching for: `conn.stream(...)` gives back something that +/// is async-iterable, turns into the web's own `ReadableStream`, and +/// stops itself when the loop over it breaks. +#[napi] +pub struct ZuCursor { + live: Arc, +} + +/// Makes a cursor over a statement that has not started yet. +/// +/// `refused` is the mistake this client caught before the engine saw +/// it. It is carried rather than thrown, so that a caller hears about +/// it where they hear about everything else a statement can do, which +/// is the first read. +pub(crate) fn open(started: Started, watch: Option, refused: Option) -> ZuCursor { + let (start, ended) = match refused { + Some(message) => (None, Some(Err(Failure::Usage(message)))), + None => (Some(started), None), + }; + let live = Live { + pipe: Pipe::new(), + stopped: AtomicBool::new(false), + start: Mutex::new(start), + ended: Mutex::new(ended), + watch: Mutex::new(watch), + }; + // A refused statement never runs and never sends, so the queue is + // over before it began and the first read finds the end of a stream + // that had nothing in it, and the reason why. + if live.start.lock().is_ok_and(|start| start.is_none()) { + live.pipe.finish(); + } + ZuCursor { + live: Arc::new(live), + } +} + +#[napi] +impl ZuCursor { + /// The next batch of rows, or `null` once the statement has ended. + /// + /// An array of row objects with the column names beside it, which + /// is the value `query` gives for a whole result and for the same + /// reason: what a caller does with rows is iterate them. + #[napi(ts_return_type = "Promise")] + pub fn next(&self) -> AsyncTask { + AsyncTask::new(NextTask { + live: Arc::clone(&self.live), + }) + } + + /// Stops the statement and waits for it to let go of the + /// connection. + /// + /// Waits, because a stream abandoned while the next statement on + /// the same connection is already being asked for would make that + /// statement queue behind a scan nobody is reading. Stopping is not + /// a failure: the rows already handed over stand, and the summary + /// says the statement was stopped. + #[napi(ts_return_type = "Promise")] + pub fn cancel(&self) -> AsyncTask { + // Before the task rather than inside it, so that the statement + // sees the word at the next batch instead of at the next free + // thread on the threadpool. + self.live.stopped.store(true, Ordering::Release); + AsyncTask::new(CancelTask { + live: Arc::clone(&self.live), + }) + } + + /// What the statement did, once it has ended, and `null` until + /// then. + /// + /// The column names are here as well as beside every batch, because + /// a statement that gave back no rows gave back no batches either + /// and its projection is still worth knowing. + #[napi(getter, ts_return_type = "ZuSummary | null")] + pub fn summary<'env>(&self, env: &'env Env) -> Result>> { + let ended = self + .live + .ended + .lock() + .map_err(|_| Error::from_reason(POISONED))?; + let Some(Ok(streamed)) = ended.as_ref() else { + return Ok(None); + }; + let mut summary = Object::new(env)?; + summary.set( + "columns", + Array::from_ref_vec_string(env, &streamed.columns)?, + )?; + // A count of what this client handed over, not an INT64 out of + // the database, which is why it is a `number` and everything + // that came out of a row is a `bigint`. + summary.set("rows", streamed.rows as f64)?; + summary.set("stopped", streamed.stopped)?; + summary.set("streamed", streamed.streamed)?; + summary.set("notices", notices(env, &streamed.notices)?)?; + Ok(Some(summary)) + } +} + +/// One read: the next batch, whatever it takes to get one. +pub struct NextTask { + live: Arc, +} + +/// What a read found. +pub enum Pulled { + /// A batch, with what it takes to read it. + Rows(Arc, Vec>), + /// The statement has ended. Why is in `ended`. + End, +} + +impl NextTask { + /// Starts the statement if this is the first read, then waits for a + /// batch. + fn pull(&self) -> Result { + self.begin()?; + match self.live.pipe.recv() { + Some(Chunk::Rows { head, rows }) => Ok(Pulled::Rows(head, rows)), + Some(Chunk::Done(ended)) => { + self.ended(ended)?; + Ok(Pulled::End) + } + // The queue is over, which happens only after the statement + // has said what it ended as. So this is a read past the end + // rather than a result that went missing. + None => Ok(Pulled::End), + } + } + + /// Runs the statement on a thread of its own, once. + /// + /// Nothing to start is a stream that already started, or one + /// somebody cancelled before it ever did, and both of those are a + /// stream to read rather than one to run. + fn begin(&self) -> Result<()> { + let Some(started) = self + .live + .start + .lock() + .map_err(|_| Error::from_reason(POISONED))? + .take() + else { + return Ok(()); + }; + let live = Arc::clone(&self.live); + std::thread::Builder::new() + .name("zu-stream".to_string()) + .spawn(move || started.run(&live)) + .map_err(|err| { + // The thread is the statement, so a machine that cannot + // give one out has a stream that never runs. Said as a + // failure of this read, which is the call that asked. + self.live.pipe.finish(); + Error::from_reason(format!("the stream could not be given a thread: {err}")) + })?; + Ok(()) + } + + fn ended(&self, ended: std::result::Result) -> Result<()> { + let mut slot = self + .live + .ended + .lock() + .map_err(|_| Error::from_reason(POISONED))?; + if slot.is_none() { + *slot = Some(ended); + } + Ok(()) + } + + /// The end of the stream, as the caller sees it: nothing, or the + /// failure that ended it. + fn ending<'env>(&self, env: &'env Env) -> Result>> { + let mut ended = self + .live + .ended + .lock() + .map_err(|_| Error::from_reason(POISONED))?; + if !matches!(ended.as_ref(), Some(Err(_))) { + return Ok(None); + } + // Taken out, because a failure is thrown once and a second read + // past the end is an end rather than the same exception again. + let Some(Err(failure)) = ended.take() else { + return Ok(None); + }; + let watch = self + .live + .watch + .lock() + .map_err(|_| Error::from_reason(POISONED))?; + Err(failed(env, failure, watch.as_ref())) + } + + /// Takes the listener back off the signal, once the statement it + /// was watching has ended. + /// + /// Only then, because a stream outlives every one of its reads and + /// the read that ends it is the last of many. + fn release(&self, env: &Env) -> Result<()> { + let ended = self + .live + .ended + .lock() + .map_err(|_| Error::from_reason(POISONED))? + .is_some(); + match ended { + true => release(&self.live, env), + false => Ok(()), + } + } +} + +/// Takes the listener off the signal and lets both references go. +/// +/// On the thread that owns the runtime, which is where a task's +/// `finally` runs and the only place a reference may be touched. A +/// signal outlives the stream it stopped, often by a whole request, and +/// a listener left on one is a stream's worth of memory that never goes +/// away. +fn release(live: &Live, env: &Env) -> Result<()> { + let watch = live + .watch + .lock() + .map_err(|_| Error::from_reason(POISONED))? + .take(); + match watch { + Some(watch) => watch.release(env), + None => Ok(()), + } +} + +impl<'task> ScopedTask<'task> for NextTask { + type Output = Pulled; + type JsValue = Option>; + + fn compute(&mut self) -> Result { + self.pull() + } + + fn resolve(&mut self, env: &'task Env, output: Self::Output) -> Result { + match output { + Pulled::Rows(head, rows) => batch(env, &head, &rows).map(Some), + Pulled::End => self.ending(env), + } + } + + fn finally(self, env: Env) -> Result<()> { + self.release(&env) + } +} + +/// One batch, as an array of row objects with the column names beside +/// them. +/// +/// The names are not enumerable, exactly as they are not on a whole +/// result, so a batch spreads, stringifies and compares as the plain +/// array of rows it is. +fn batch<'env>(env: &'env Env, head: &Head, rows: &[Vec]) -> Result> { + let mut array = env.create_array(rows.len() as u32)?; + for (ix, row) in rows.iter().enumerate() { + let mut object = Object::new(env)?; + for (column, value) in head.columns.iter().zip(row) { + object.set(column.as_str(), to_js(env, value, &head.names)?)?; + } + array.set(ix as u32, object)?; + } + let mut object = array.coerce_to_object()?; + object.define_properties(&[beside( + env, + "columns", + Array::from_ref_vec_string(env, &head.columns)?, + )?])?; + Ok(array) +} + +/// Stopping: say so, then read what is left until the statement ends. +/// +/// The reading is the point. The thread hands a batch over and waits +/// for room, so a stream that was stopped and then abandoned would +/// leave a thread holding the connection until it next looked up. +/// Draining means that when this resolves, the statement is over and +/// the connection is free. +pub struct CancelTask { + live: Arc, +} + +impl<'task> ScopedTask<'task> for CancelTask { + type Output = (); + type JsValue = (); + + fn compute(&mut self) -> Result { + // A stream nobody read has no thread to stop and no statement + // to end, so taking the start away ends it here: there is + // nothing left for a later read to begin, and the queue is over + // before anything was put in it. + let never = self + .live + .start + .lock() + .is_ok_and(|mut start| start.take().is_some()); + if never { + self.live.pipe.finish(); + return Ok(()); + } + // Everything already handed over, read and dropped, until the + // statement says it is over. Which it does: a stopped scan ends + // at the next batch, and the end is the last thing sent. + while let Some(chunk) = self.live.pipe.recv() { + let Chunk::Done(ended) = chunk else { + continue; + }; + if let Ok(mut slot) = self.live.ended.lock() + && slot.is_none() + { + // A stream stopped after it had already failed keeps + // the failure, because that is what a read waiting + // behind this one is owed. + *slot = Some(ended); + } + break; + } + Ok(()) + } + + fn resolve(&mut self, _env: &'task Env, _output: Self::Output) -> Result { + Ok(()) + } + + /// Nothing can happen to a cancelled stream afterwards, so the + /// signal is let go of here whether the statement had started or + /// not. + fn finally(self, env: Env) -> Result<()> { + release(&self.live, &env) + } +} + +/// A cursor nobody holds any more is a reader that has gone. +/// +/// The garbage collector is what says so, which is late, but late is +/// not never and the alternative is a statement scanning a database +/// into a queue that will never be read. Everything else about ending a +/// stream is deliberate: this is the backstop under a program that +/// forgot. +impl Drop for ZuCursor { + fn drop(&mut self) { + self.live.stopped.store(true, Ordering::Release); + self.live.pipe.close(); + } +} + +impl Started { + /// The statement, on the thread that owns it for its whole life. + fn run(self, live: &Live) { + let ended = self.stream(live); + // The last thing put in the queue, and then the queue is over, + // which is what a read past the end finds. + live.pipe.send(Chunk::Done(ended), &self.alive); + live.pipe.finish(); + } + + /// Takes the connection, runs the statement, and hands every batch + /// over. + fn stream(&self, live: &Live) -> std::result::Result { + let mut held = self + .inner + .lock() + .map_err(|_| Failure::Usage(POISONED.to_string()))?; + let Some(conn) = held.as_mut() else { + return Err(Failure::Usage(CLOSED.to_string())); + }; + // From here the connection is this statement's, so this is + // where a signal can start stopping it. A signal that fired + // first ends the statement without the engine ever seeing it. + if let Some(guard) = &self.guard + && !guard.enter() + { + guard.leave(); + return Err(Failure::Aborted); + } + let names = Arc::new(Names::of(conn.session_mut().catalog())); + let params: Vec<(&str, Value)> = self + .params + .iter() + .map(|(name, value)| (name.as_str(), value.clone())) + .collect(); + + let mut head: Option> = None; + let mut sink = |batch: Batch<'_>| { + // The columns are the same for every batch of one + // statement, and the first batch is where they are known. + let head = Arc::clone(head.get_or_insert_with(|| { + Arc::new(Head { + columns: batch.columns().to_vec(), + names: Arc::clone(&names), + }) + })); + Ok(self.hand(live, head, batch.rows().to_vec())) + }; + let result = match self.batch_rows { + Some(rows) => { + conn.query_stream_batched(&self.statement, ¶ms, rows as usize, &mut sink) + } + None => conn.query_stream(&self.statement, ¶ms, &mut sink), + }; + + if let Some(guard) = &self.guard { + guard.leave(); + // An interrupt is the engine's answer to somebody having + // asked, and the only somebody here is the caller's signal. + if guard.asked() && matches!(result, Err(ZuError::Interrupted)) { + return Err(Failure::Aborted); + } + } + Ok(result?) + } + + /// Hands one batch over, waiting for room, and says whether the + /// scan should carry on. + /// + /// Three things end a stream from this side: a reader that asked it + /// to stop, a connection closed underneath it, and a reader that + /// let go of the cursor entirely, which closes the queue. + fn hand(&self, live: &Live, head: Arc, rows: Vec>) -> Flow { + if live.stopped.load(Ordering::Acquire) { + return Flow::Stop; + } + match live.pipe.send(Chunk::Rows { head, rows }, &self.alive) { + true => Flow::More, + false => Flow::Stop, + } + } +} diff --git a/test/exports.test.mjs b/test/exports.test.mjs index e082156..9160baf 100644 --- a/test/exports.test.mjs +++ b/test/exports.test.mjs @@ -22,6 +22,8 @@ const SURFACE = [ 'abiVersion', 'isZuError', 'Connection', + 'ZuStream', + 'ZuCursor', 'ZuDate', 'ZuTime', 'ZuTimestamp', diff --git a/test/stream.test.mjs b/test/stream.test.mjs new file mode 100644 index 0000000..1b4287c --- /dev/null +++ b/test/stream.test.mjs @@ -0,0 +1,342 @@ +// A statement read a batch at a time. +// +// The interesting cases are not the ones where every row is read. They +// are the ones where the reader stops early, which is the whole reason +// streaming is different from a query: a scan has to end, the +// connection has to come back, and the statement after it has to run +// without waiting for a database nobody was reading. + +import assert from 'node:assert/strict' +import { getEventListeners } from 'node:events' +import test from 'node:test' + +import { isZuError, ZuStream } from 'zudb' + +import { fresh, twoPeople } from './helper.mjs' + +// Enough rows to arrive in more than one batch of a named size, and few +// enough that a test that reads all of them is not a benchmark. +const PEOPLE = 40 + +// Enough rows that a scan of them takes long enough to stop halfway, +// which is what the tests about stopping need and what a handful of +// rows cannot give them: a statement that has already finished is a +// statement no interrupt can catch. +const MANY = 60_000 + +async function people(t, count = PEOPLE) { + const made = await fresh(t) + // One statement per row is a write per row, so the rows past the + // first are written in batches. The first is written on its own with + // literals, because that is the insert that declares the table. + await made.conn.exec("INSERT (p:person {id: 1, name: 'p1'})") + for (let start = 2; start <= count; start += 500) { + const parts = [] + for (let id = start; id < Math.min(start + 500, count + 1); id++) { + parts.push(`(p${id}:person {id: ${id}, name: 'p${id}'})`) + } + await made.conn.exec(`INSERT ${parts.join(', ')}`) + } + return made +} + +test('a stream gives every row, in order, and says what it did', async (t) => { + const { conn } = await people(t) + + const stream = conn.stream('MATCH (p:person) RETURN p.id AS id, p.name AS name') + const seen = [] + for await (const row of stream) seen.push(row) + + assert.equal(seen.length, PEOPLE) + assert.deepEqual(seen[0], { id: 1n, name: 'p1' }) + assert.deepEqual(stream.columns, ['id', 'name']) + assert.deepEqual(stream.summary, { + columns: ['id', 'name'], + rows: PEOPLE, + stopped: false, + streamed: true, + notices: [], + }) +}) + +test('a batch is an array of rows with the columns beside it', async (t) => { + const { conn } = await people(t) + + const stream = conn.stream('MATCH (p:person) RETURN p.id AS id', null, { batchRows: 15 }) + const sizes = [] + for await (const batch of stream.batches()) { + // The same array trick a whole result uses: the names are there to + // read and out of the way of everything that iterates. + assert.deepEqual(batch.columns, ['id']) + assert.deepEqual(Object.keys(batch), [...batch.keys()].map(String)) + sizes.push(batch.length) + } + + // A size is a ceiling rather than a promise, because the engine cuts + // batches out of the rows it has already made and the last piece of + // one is whatever is left of it. What the size does promise is what + // a caller wants from it: nothing bigger than this is ever held. + assert.equal( + sizes.reduce((total, size) => total + size, 0), + PEOPLE, + ) + assert.ok( + sizes.every((size) => size > 0 && size <= 15), + `${sizes} is not a run of batches of at most 15`, + ) + assert.ok(sizes.length > 1, `expected more than one batch, got ${sizes.length}`) +}) + +test('a statement with no rows still says what it projected', async (t) => { + const { conn } = await twoPeople(t) + + const stream = conn.stream("MATCH (p:person) WHERE p.name = 'nobody' RETURN p.name AS name") + const seen = [] + for await (const row of stream) seen.push(row) + + assert.deepEqual(seen, []) + // No batch ever arrived, so this is the summary talking and not a + // batch, which is the reason the columns are on both. + assert.deepEqual(stream.columns, ['name']) + assert.equal(stream.summary.rows, 0) + assert.equal(stream.summary.stopped, false) +}) + +test('breaking out of the loop stops the statement and gives the connection back', async (t) => { + const { conn } = await people(t) + + const stream = conn.stream('MATCH (p:person) RETURN p.id AS id', null, { batchRows: 2 }) + for await (const row of stream) { + assert.equal(row.id, 1n) + break + } + + assert.equal(stream.summary.stopped, true) + assert.ok(stream.summary.rows < PEOPLE, `${stream.summary.rows} rows is the whole scan`) + // The point of waiting inside the break: the next statement runs + // rather than queueing behind a scan nobody is reading. + const rows = await conn.query('MATCH (p:person) RETURN count(*) AS n') + assert.equal(rows[0].n, BigInt(PEOPLE)) +}) + +test('a stream stops itself at the end of an await using block', async (t) => { + const { conn } = await people(t) + + let stream + { + await using held = conn.stream('MATCH (p:person) RETURN p.id AS id', null, { batchRows: 2 }) + stream = held + for await (const row of held) { + assert.equal(row.id, 1n) + break + } + } + + assert.equal(stream.summary.stopped, true) + assert.equal((await conn.query('RETURN 1 AS n'))[0].n, 1n) +}) + +test('cancelling twice is cancelling once, and reading afterwards is the end', async (t) => { + const { conn } = await people(t) + + const stream = conn.stream('MATCH (p:person) RETURN p.id AS id') + await stream.cancel() + await stream.cancel() + + const seen = [] + for await (const row of stream) seen.push(row) + assert.deepEqual(seen, []) + assert.equal((await conn.query('RETURN 1 AS n'))[0].n, 1n) +}) + +test('a stream is a ReadableStream when something wants one', async (t) => { + const { conn } = await people(t) + + const web = conn.stream('MATCH (p:person) RETURN p.id AS id').toReadableStream() + const seen = [] + for await (const row of web) seen.push(row.id) + + assert.equal(seen.length, PEOPLE) + assert.equal(seen[0], 1n) +}) + +test('cancelling the ReadableStream stops the statement underneath it', async (t) => { + const { conn } = await people(t) + + const stream = conn.stream('MATCH (p:person) RETURN p.id AS id', null, { batchRows: 2 }) + const web = stream.toReadableStream() + const reader = web.getReader() + const first = await reader.read() + assert.equal(first.value.id, 1n) + await reader.cancel() + + assert.equal(stream.summary.stopped, true) + assert.equal((await conn.query('RETURN 1 AS n'))[0].n, 1n) +}) + +test('a signal stops a stream, and the connection is exactly as it was', async (t) => { + const { conn } = await people(t, MANY) + + const controller = new AbortController() + const stream = conn.stream('MATCH (p:person) RETURN p.id AS id', null, { + batchRows: 1, + signal: controller.signal, + }) + const mine = new Error('enough') + + const caught = await (async () => { + try { + for await (const row of stream) { + if (row.id === 3n) controller.abort(mine) + } + } catch (err) { + return err + } + return null + })() + + // The caller's own reason, which is what `fetch` does and what a + // program that wrote the object wants back. + assert.equal(caught, mine) + assert.equal((await conn.query('MATCH (p:person) RETURN count(*) AS n'))[0].n, BigInt(MANY)) +}) + +test('a signal that has already fired stops a stream before the engine sees it', async (t) => { + const { conn } = await people(t) + + const controller = new AbortController() + controller.abort() + const stream = conn.stream('MATCH (p:person) RETURN p.id AS id', null, { + signal: controller.signal, + }) + + await assert.rejects( + async () => { + for await (const row of stream) assert.fail(`read ${row.id} from a stopped stream`) + }, + (err) => err.name === 'AbortError', + ) +}) + +test('a stream leaves no listener on the signal it was given', async (t) => { + const { conn } = await people(t) + + const controller = new AbortController() + for (let round = 0; round < 8; round++) { + const stream = conn.stream('MATCH (p:person) RETURN p.id AS id', null, { + signal: controller.signal, + batchRows: 4, + }) + // Half read to the end, half stopped early, because the two end a + // stream in different places and both have to take the listener off. + if (round % 2 === 0) { + for await (const _ of stream) void _ + } else { + for await (const _ of stream) break + } + } + + assert.deepEqual(getEventListeners(controller.signal, 'abort'), []) +}) + +test('a statement that will not compile fails at the first read', async (t) => { + const { conn } = await fresh(t) + + // Not at the call, which is the point: a stream is asked for and + // read, and everything a statement can do it does where the caller + // has an await to catch it. + const stream = conn.stream('MATCH (') + await assert.rejects( + async () => { + for await (const row of stream) void row + }, + (err) => isZuError(err) && err.name === 'ZuSyntaxError', + ) +}) + +test('a closed connection refuses a stream where every other statement is refused', async (t) => { + const { conn } = await twoPeople(t) + const stream = conn.stream('MATCH (p:person) RETURN p.id AS id') + conn.close() + + await assert.rejects( + async () => { + for await (const row of stream) void row + }, + (err) => isZuError(err) && err.name === 'ZuUsageError', + ) +}) + +test('a batch size that is not one is refused as the caller\'s mistake', async (t) => { + const { conn } = await twoPeople(t) + + for (const batchRows of [0, 'lots', {}, -1]) { + const stream = conn.stream('MATCH (p:person) RETURN p.id AS id', null, { batchRows }) + await assert.rejects( + async () => { + for await (const row of stream) void row + }, + (err) => isZuError(err) && err.name === 'ZuUsageError' && /batchRows/.test(err.message), + `batchRows: ${JSON.stringify(batchRows)} was accepted`, + ) + } +}) + +test('a stream is the class the package exports', async (t) => { + const { conn } = await twoPeople(t) + const stream = conn.stream('MATCH (p:person) RETURN p.id AS id') + assert.ok(stream instanceof ZuStream) + await stream.cancel() +}) + +test('the first row arrives long before the last one is made', async (t) => { + const { conn } = await people(t, MANY) + + const scan = 'MATCH (p:person) RETURN p.id AS id, p.name AS name' + const whole = process.hrtime.bigint() + const rows = await conn.query(scan) + const plain = Number(process.hrtime.bigint() - whole) / 1e6 + assert.equal(rows.length, MANY) + + const early = process.hrtime.bigint() + const stream = conn.stream(scan, null, { batchRows: 256 }) + for await (const row of stream) { + void row + break + } + const first = Number(process.hrtime.bigint() - early) / 1e6 + + // The whole point of streaming, measured rather than asserted about: + // a client that buffered the result would take as long to hand over + // the first row as to hand over all of them. A fifth is a wide margin + // around a first batch that arrives in about a hundredth of the scan, + // because a loaded machine slows the first batch and the whole scan + // by different amounts. + assert.ok( + first < plain / 5, + `the first row took ${first.toFixed(1)}ms of the ${plain.toFixed(1)}ms whole scan`, + ) + assert.equal(stream.summary.stopped, true) + assert.equal(stream.summary.streamed, true) + assert.ok(stream.summary.rows < MANY / 10, `${stream.summary.rows} rows is most of the scan`) +}) + +test('a statement that has to run whole is read the same way and says so', async (t) => { + const { conn } = await people(t) + + // Sorting is the clearest of them: the last row is the one that + // decides where the first goes, so there is nothing to hand over + // until it has all been made. The engine runs it whole and hands the + // result over in batches, which is a different thing from streaming + // and the reason the summary tells them apart. + const stream = conn.stream('MATCH (p:person) RETURN p.id AS id ORDER BY id DESC', null, { + batchRows: 8, + }) + const seen = [] + for await (const row of stream) seen.push(row.id) + + assert.equal(seen.length, PEOPLE) + assert.equal(seen[0], BigInt(PEOPLE)) + assert.equal(stream.summary.streamed, false) + assert.equal(stream.summary.stopped, false) +}) diff --git a/test/types/cjs.cts b/test/types/cjs.cts index f60ac8b..91b52a4 100644 --- a/test/types/cjs.cts +++ b/test/types/cjs.cts @@ -2,7 +2,7 @@ // resolution through a different condition to a different file, and so // is worth compiling separately rather than assuming. -import { connect, isZuError, type ZuParam } from 'zudb' +import { connect, isZuError, type ZuParam, type ZuStream } from 'zudb' export async function insert(path: string, values: Record): Promise { const conn = await connect(path) @@ -17,3 +17,15 @@ export async function insert(path: string, values: Record): Pro 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') + const out: bigint[] = [] + try { + for await (const row of stream) out.push(row.id) + } finally { + conn.close() + } + return out +} diff --git a/test/types/esm.mts b/test/types/esm.mts index 955ab67..8a26bd2 100644 --- a/test/types/esm.mts +++ b/test/types/esm.mts @@ -2,7 +2,16 @@ // being asserted is that it compiles, which is the half of an API that // a test suite in JavaScript cannot reach. -import { connect, isZuError, ZuDate, type ZuError, type ZuRows } from 'zudb' +import { + connect, + isZuError, + ZuDate, + type ZuBatch, + type ZuError, + type ZuRows, + type ZuStream, + type ZuSummary, +} from 'zudb' export async function people(path: string, name: string): Promise { // `await using` is the intended scoping, and it needs the connection @@ -25,6 +34,47 @@ export async function people(path: string, name: string): Promise { return rows.map((row) => `${row.name} ${row.id.toString()}`) } +type Person = { id: bigint; name: string } + +export async function names(path: string): Promise { + await using conn = await connect(path, { readOnly: true }) + + // The row type is the stream's, so what comes out of the loop is + // typed without a cast anywhere, and so is what comes out of a batch. + await using stream: ZuStream = conn.stream( + 'MATCH (p:person) RETURN p.id AS id, p.name AS name', + null, + { batchRows: 512, signal: AbortSignal.timeout(50) }, + ) + + let longest = 0 + for await (const row of stream) longest = Math.max(longest, row.name.length) + + // Both of these are null until there is something to read in them, + // which is what makes a caller check rather than believe. + const summary: ZuSummary | null = stream.summary + if (summary && !summary.streamed) longest += 0 + + return longest +} + +export async function counted(path: string): Promise { + const conn = await connect(path) + const stream = conn.stream('MATCH (p:person) RETURN p.id AS id, p.name AS name') + let rows = 0 + for await (const batch of stream.batches()) { + const typed: ZuBatch = batch + rows += typed.length + } + // A Web Stream of the same rows, for anything that already speaks one. + const web: ReadableStream = conn + .stream('MATCH (p:person) RETURN p.id AS id') + .toReadableStream() + await web.cancel() + await conn.close() + return rows +} + 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/tools/install.mjs b/tools/install.mjs index d31adce..4577d10 100644 --- a/tools/install.mjs +++ b/tools/install.mjs @@ -64,6 +64,15 @@ try { "assert.deepEqual(rows.columns, ['name'])", "assert.equal(rows[0].name, 'ada')", "assert.equal(isZuError(await conn.query('MATCH (').catch((err) => err)), true)", + // The streamed path as well, because it is the one part of the + // surface that is written in JavaScript over the addon rather + // than by the addon, and an installed package is where a missing + // file in `files` turns up. + "const stream = conn.stream('MATCH (p:person) RETURN p.name AS name')", + 'const names = []', + 'for await (const row of stream) names.push(row.name)', + "assert.deepEqual(names, ['ada'])", + 'assert.equal(stream.summary.rows, 1)', 'await conn.close()', ].join('\n'), ) diff --git a/types/header.d.ts b/types/header.d.ts index c9ca687..b823edd 100644 --- a/types/header.d.ts +++ b/types/header.d.ts @@ -88,6 +88,66 @@ export interface ZuRows> extends Array { readonly notices: ZuNotice[] } +/** + * One batch of a streamed result. + * + * The rows of a whole result with the same array trick and one fewer + * property: `columns` is the statement's projection and is the same on + * every batch of one stream, and what a statement completed with is not + * known until it has, so it is on the summary rather than here. + */ +export interface ZuBatch> extends Array { + readonly columns: string[] +} + +/** + * What a streamed statement did, known once it has ended. + * + * The rows are gone by then, which is the point of streaming, so this + * is what is worth keeping about a result nobody held: what it + * projected, how much of it was read, whether the reader stopped it + * early, and what the engine wanted to say along the way. + */ +export interface ZuSummary { + readonly columns: string[] + /** How many rows were handed over, which is fewer than the statement + * would have returned when the reader stopped early. */ + readonly rows: number + readonly stopped: boolean + /** + * Whether the rows arrived as they were made, rather than the + * statement running whole and being handed over in batches + * afterwards. A statement that has to see every row before it can + * give one, which is `ORDER BY`, `DISTINCT`, the aggregates and + * anything that writes, is the second kind, and so is a plan the + * pipeline executor does not take. The loop over it reads the same + * either way, so this is here for a caller measuring where the time + * went rather than for one deciding what to do next. + */ + readonly streamed: boolean + readonly notices: ZuNotice[] +} + +/** + * What a streamed statement takes beside its parameters. + */ +export interface ZuStreamOptions extends ZuStatementOptions { + /** + * How many rows a batch may hold. The engine's own vector by + * default, which is the unit it already works in and the one that + * costs nothing to hand over. Name a size when the rows are going + * somewhere with a size of its own, an Arrow record batch or an + * HTTP chunk. + * + * A ceiling and not a promise: batches are cut out of rows that have + * already been made, so the last piece of a run of them is whatever + * was left, and a size above the engine's vector gets the vector. It + * is what bounds how much a reader holds at once, which is the + * question a caller is asking when they name one. + */ + readonly batchRows?: number +} + /** * What a statement takes beside its parameters. * diff --git a/zudb.cjs b/zudb.cjs index f4ea505..235be94 100644 --- a/zudb.cjs +++ b/zudb.cjs @@ -14,6 +14,130 @@ const binding = require('./binding.cjs') +/** + * A statement read a batch at a time. + * + * The cursor underneath is a pull: one call, one batch, `null` at the + * end. Everything a program actually writes is here rather than in + * Rust, because iteration, early exit and Web Streams are JavaScript + * shapes, and the native side of a `for await` that breaks halfway is + * a lot of machinery for something the language already has. + * + * A stream is read once. All three ways of reading it are the same + * statement, so taking two of them gets two halves of one result. + */ +class ZuStream { + #cursor + #columns = null + #summary = null + #ended = false + + constructor(cursor) { + this.#cursor = cursor + } + + /** The projection, once the first batch has arrived. */ + get columns() { + return this.#columns + } + + /** What the statement did, once it has ended. */ + get summary() { + return this.#summary + } + + /** + * The rows a batch at a time, which is the shape they cross the + * boundary in. + * + * The `finally` is the whole reason this is a generator: a `break` + * out of a `for await`, a `throw` inside it and a `return()` on the + * iterator all end up there, and all three mean the reader has gone + * and the statement should stop rather than scan a database nobody + * is reading. + */ + async *batches() { + try { + for (;;) { + const batch = await this.#cursor.next() + if (batch === null) { + this.#done() + return + } + this.#columns ??= batch.columns + yield batch + } + } finally { + if (!this.#ended) await this.cancel() + } + } + + async *[Symbol.asyncIterator]() { + for await (const batch of this.batches()) yield* batch + } + + /** + * The same rows as a Web Stream, one row per chunk. + * + * Pulled a batch at a time and enqueued a row at a time, so the + * backpressure is the reader's: nothing is pulled from the database + * until the queue in front of it has drained. A `cancel()` on the + * stream, which is what a `pipeTo` that fails does, stops the + * statement. + */ + toReadableStream() { + const batches = this.batches() + return new ReadableStream({ + async pull(controller) { + const { value, done } = await batches.next() + if (done) { + controller.close() + return + } + for (const row of value) controller.enqueue(row) + }, + async cancel() { + await batches.return() + }, + }) + } + + /** Stops the statement and waits for it to let go of the connection. */ + async cancel() { + this.#ended = true + await this.#cursor.cancel() + this.#summary ??= this.#cursor.summary + } + + async [Symbol.asyncDispose]() { + await this.cancel() + } + + #done() { + this.#ended = true + this.#summary = this.#cursor.summary + this.#columns ??= this.#summary?.columns ?? null + } +} + +/** + * `conn.stream(...)`, which is a method of the native class written in + * JavaScript. + * + * On the prototype rather than in the class, because the class is + * registered by the addon and there is nowhere in Rust to write a + * method whose body is a JavaScript generator. Not enumerable, like + * every other method of a class, so it does not turn up in a + * `for...in` over a connection. + */ +Object.defineProperty(binding.Connection.prototype, 'stream', { + value: function stream(statement, params, options) { + return new ZuStream(this.cursor(statement, params, options)) + }, + writable: true, + configurable: true, +}) + /** * Whether a caught value is a failure from this client. * @@ -39,6 +163,12 @@ module.exports = { abiVersion: binding.abiVersion, isZuError, Connection: binding.Connection, + ZuStream, + // The pull underneath a stream, which `conn.cursor(...)` hands back + // and almost nobody should be holding. It is here because it is in + // the types either way, and a name that types can see and `require` + // cannot is a program that compiles and then throws. + ZuCursor: binding.ZuCursor, ZuDate: binding.ZuDate, ZuTime: binding.ZuTime, ZuTimestamp: binding.ZuTimestamp, diff --git a/zudb.d.cts b/zudb.d.cts index 34267c4..f8a688e 100644 --- a/zudb.d.cts +++ b/zudb.d.cts @@ -1,19 +1,77 @@ /* The types for `require('zudb')`. */ -import type { ZuError } from './binding.cjs' +import type { ZuBatch, ZuError, ZuParam, ZuStreamOptions, ZuSummary, ZuValue } from './binding.cjs' export * from './binding.cjs' /** - * `await using` on a connection, in the types as well as at runtime. + * The two things about a connection that are not written in Rust. * * The disposal is put on every connection as it is made, under the key * `await using` looks up. It cannot be declared where the rest of the * class is, because the generator writes that file from the Rust and a * method's name there is a string while this key is a symbol. + * + * `stream` is on the prototype for a plainer reason: its body is an + * async generator, and there is nowhere in Rust to write one. */ declare module './binding.cjs' { - interface Connection extends AsyncDisposable {} + interface Connection extends AsyncDisposable { + /** + * Runs one statement and reads it a batch at a time. + * + * The statement starts on the first read rather than here, so a + * stream that is made and never read is not a scan holding the + * connection against everything after it. Ending the read early, + * by a `break` or a `cancel()`, stops the statement. + */ + stream>( + statement: string, + params?: Record | null, + options?: ZuStreamOptions | null, + ): ZuStream + } +} + +/** + * A statement read a batch at a time. + * + * Iterating it row by row is the ordinary way, `batches()` is the fast + * way when the work is per batch rather than per row, and + * `toReadableStream()` is for handing rows to anything that already + * speaks Web Streams. All three read the same statement once, so a + * stream is used one of the three ways rather than two. + * + * Ending it early is the case worth knowing about: a `break` out of the + * loop, a `return()` on the iterator, a `cancel()`, or leaving the block + * of an `await using` all stop the statement and wait for it to let go + * of the connection. + * + * A class rather than an interface, so that a program passing streams + * around can name the type and ask `instanceof`. It is not constructed + * directly: a stream is a statement, and a statement comes from a + * connection. + */ +export declare class ZuStream> implements AsyncIterable, AsyncDisposable { + private constructor() + /** The projection, `null` until the first batch has arrived. */ + readonly columns: string[] | null + /** What the statement did, `null` until it has ended. */ + readonly summary: ZuSummary | null + /** + * The rows a batch at a time, which is the shape they cross the + * boundary in. + */ + batches(): AsyncIterableIterator> + [Symbol.asyncIterator](): AsyncIterableIterator + /** + * The same rows as a Web Stream, one row per chunk, pulled a batch at + * a time. + */ + toReadableStream(): ReadableStream + /** Stops the statement and waits for it to let go of the connection. */ + cancel(): Promise + [Symbol.asyncDispose](): Promise } /** diff --git a/zudb.mjs b/zudb.mjs index c5fc0fa..ac80a24 100644 --- a/zudb.mjs +++ b/zudb.mjs @@ -21,6 +21,8 @@ export const version = zudb.version export const abiVersion = zudb.abiVersion export const isZuError = zudb.isZuError export const Connection = zudb.Connection +export const ZuStream = zudb.ZuStream +export const ZuCursor = zudb.ZuCursor export const ZuDate = zudb.ZuDate export const ZuTime = zudb.ZuTime export const ZuTimestamp = zudb.ZuTimestamp