From d5b83df605fce7ed87c098eb3ff1695170570acd Mon Sep 17 00:00:00 2001 From: Tam Nguyen Duc <1218621+tamnd@users.noreply.github.com> Date: Tue, 18 Aug 2026 20:49:46 +0700 Subject: [PATCH] Read a result a piece at a time: streams, batches and Web Streams conn.stream(...) runs a statement and hands its rows over as they are made. Reading it is a for await over rows, a batches() over the arrays they crossed the boundary in, or a toReadableStream() for anything that already speaks Web Streams, and all three are the same statement read once. The engine's shape is a push and JavaScript wants a pull, so a thread of the statement's own sits between them with a queue of two batches. That queue is the whole of the buffering: a reader slower than the scan stops the scan rather than filling memory behind it. The thread is its own rather than libuv's, because a statement parked on the threadpool waiting for a loop body to come round again is a quarter of every other library's file reads gone. Stopping is the part worth the machinery. A break, a throw, a return() on the iterator, a cancel() or the end of an await using block all end the statement and wait for it to let go of the connection, so the next statement runs rather than queueing behind a scan nobody is reading. A cursor nobody holds closes its queue when it is collected, which is the backstop under a program that forgot. The statement starts on the first read and not before, so a stream made and never read holds nothing at all. The summary says what the statement did once it is over: the columns, the rows handed over, whether the reader stopped it, whether it was streamed rather than run whole and cut up afterwards, and the notices. ORDER BY, DISTINCT and the aggregates are the second kind, and the loop over them is the same either way. batchRows is read as a double and checked here rather than narrowed by the runtime, because JavaScript's own narrowing turns -1 into four billion and 1.5 into 1. Engine pinned forward to 92c9a5e for tamnd/zu#338, without which an interrupted stream ended cleanly and truncated instead of failing. 50k rows, fastest of nine: a stream read to the end costs 463ns a row against 372ns for query, a batch at a time 320ns, and reading the first batch and stopping 1.1ms against 18.6ms for the whole scan. --- Cargo.lock | 20 +- Cargo.toml | 4 +- README.md | 27 +- bench/query.mjs | 38 +++ binding.cjs | 1 + binding.d.cts | 108 +++++++ src/cancel.rs | 56 +++- src/conn.rs | 162 +++++++++-- src/lib.rs | 1 + src/stream.rs | 661 ++++++++++++++++++++++++++++++++++++++++++ test/exports.test.mjs | 2 + test/stream.test.mjs | 342 ++++++++++++++++++++++ test/types/cjs.cts | 14 +- test/types/esm.mts | 52 +++- tools/install.mjs | 9 + types/header.d.ts | 60 ++++ zudb.cjs | 130 +++++++++ zudb.d.cts | 64 +++- zudb.mjs | 2 + 19 files changed, 1695 insertions(+), 58 deletions(-) create mode 100644 src/stream.rs create mode 100644 test/stream.test.mjs 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