Read a result a piece at a time: streams, batches and Web Streams

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

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<Row>` 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
 

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',

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

binding.d.cts

@@ -88,6 +88,66 @@ export interface ZuRows<Row = Record<string, ZuValue>> extends Array<Row> {
   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<Row = Record<string, ZuValue>> extends Array<Row> {
+  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<string, ZuParam> | null, options?: ZuStatementOptions | null): Promise<void>
+  /**
+   * 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<string, ZuParam> | null, options?: ZuStreamOptions | null): ZuCursor
   /**
    * Closes the connection and releases the database.
    *
@@ -199,6 +268,45 @@ export declare class Connection {
   dispose(): Promise<void>
 }
 
+/**
+ * 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<ZuBatch | null>
+  /**
+   * 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<void>
+  /**
+   * 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.
  *