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

[tamnd]

conn.stream(...) runs a statement and hands its rows over as they are made, instead of building the whole answer first. 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. All three are the same statement read once.

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 }

How it is put together. The engine's shape is a push, a sink that says whether it wants more, 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 and DNS lookups gone. The queue is hand-written over a mutex and a condition variable, because a bounded send that gives up on a timeout is not on the stable side of std::sync::mpsc, and giving up is how a statement notices a connection closed underneath it.

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 on that connection 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.

What the summary says. The columns, the rows handed over, whether the reader stopped it, whether it was streamed rather than run whole and cut into batches afterwards, and the notices. ORDER BY, DISTINCT and the aggregates are the second kind, and the loop over them is the same either way, which is why streamed is there to tell them apart.

batchRows is a ceiling rather than a promise, and it 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.

The engine is pinned forward to 92c9a5e for tamnd/zu#338. Without it an AbortSignal on a stream ended the loop cleanly after 1024 of 60000 rows: an interrupt broke the morsel loop out through the same door a filled LIMIT does, so a truncated result came back reported as a whole one. Found by the signal test here.

Measured, 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, which is the point of it.

Checked. 69 tests including seventeen new ones: rows equal to query, batching, break, cancel(), await using, the Web Stream and its cancel, a signal fired mid-stream and one fired already, no listener left on a signal after eight streams, a statement that will not compile failing at the first read, a closed connection, four bad batch sizes, a statement that runs whole, and the first row arriving in a fifth of the time the whole scan takes. npm run check:types over both formats, npm run check:package, cargo fmt --check, cargo clippy --all-features, and the pack-and-install round trip, which now streams from the installed package too.