A result as Arrow, in the bytes every reader takes

Cargo.toml

@@ -18,8 +18,14 @@ 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 = "6753ded13a215bf5e8fe70ce41e69b35321e3e73" }
-zu-common = { git = "https://github.com/tamnd/zu", rev = "6753ded13a215bf5e8fe70ce41e69b35321e3e73" }
+zudb = { package = "zu", git = "https://github.com/tamnd/zu", rev = "0698a4eccd31670f0a875b6d097d7753440a51b3" }
+zu-common = { git = "https://github.com/tamnd/zu", rev = "0698a4eccd31670f0a875b6d097d7753440a51b3" }
+# The one translation from a result into Arrow, which lives in the engine
+# tree so that every client agrees about what a column becomes. `ipc` is
+# the only feature this client turns on: the C Data Interface hands over
+# a pointer and nothing in a JavaScript runtime can read one, so the way
+# out here is the bytes of an IPC stream.
+zu-arrow = { git = "https://github.com/tamnd/zu", rev = "0698a4eccd31670f0a875b6d097d7753440a51b3", features = ["ipc"] }
 # 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

bench/arrow.mjs

@@ -0,0 +1,96 @@
+// What a result costs as Arrow, against the two other ways out.
+//
+// Three calls over the same statement. `query` builds an object a row
+// and a value a cell. `columnar` moves one buffer a column and leaves
+// the reader to put a type around them. `arrow` writes those same
+// buffers into an IPC stream, which costs a copy of the values and a
+// header a batch, and buys a result every Arrow implementation reads.
+//
+// So the pair worth reading is `arrow` against `columnar`: the
+// difference between them is what the framing costs, and it is the
+// number that says whether a caller should take the bytes or take the
+// buffers. The line after each is the reader's side, since bytes nobody
+// decodes are not a result: `tableFromIPC` against the eleven lines the
+// README prints for wrapping the buffers.
+//
+// Run it against a release build, for the reason bench/query.mjs gives.
+//
+//   npm run build && npm run bench:arrow
+
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+
+import { tableFromIPC } from 'apache-arrow'
+import { connect } from 'zudb'
+
+const ROWS = Number(process.env.ZU_BENCH_ROWS ?? 1_000_000)
+const REPEATS = Number(process.env.ZU_BENCH_REPEATS ?? 5)
+
+const dir = await mkdtemp(join(tmpdir(), 'zu-bench-arrow-'))
+const conn = await connect(join(dir, 'bench.zu1'))
+
+await conn.exec("INSERT (p:person {uid: 1, score: 1.5, name: 'n1'})")
+{
+  const rows = await conn.appender('person')
+  for (let ix = 2; ix <= ROWS; ix++) rows.appendRow([BigInt(ix), ix / 3, `n${ix}`])
+  await rows.close()
+}
+
+// The fastest of `REPEATS` runs, in milliseconds, after one warmup, for
+// the reason bench/query.mjs gives: everything that makes a run slower
+// than the work itself is something that happened to it rather than
+// something about it.
+async function time(run) {
+  await run()
+  let best = Infinity
+  for (let round = 0; round < REPEATS; round++) {
+    const started = performance.now()
+    await run()
+    best = Math.min(best, performance.now() - started)
+  }
+  return best
+}
+
+function report(name, ms) {
+  const each = (ms * 1e6) / ROWS
+  console.log(
+    `${name.padEnd(30)} ${ms.toFixed(1).padStart(8)} ms  ${Math.round(each).toString().padStart(6)} ns/row`,
+  )
+}
+
+const ONE = 'MATCH (p:person) RETURN p.uid AS uid'
+const THREE = 'MATCH (p:person) RETURN p.uid AS uid, p.score AS score, p.name AS name'
+
+const cases = [
+  { name: 'one integer column, arrow', run: () => conn.arrow(ONE) },
+  { name: 'one integer column, columnar', run: () => conn.columnar(ONE) },
+  { name: 'one integer column, rows', run: () => conn.query(ONE) },
+  { name: 'three columns, arrow', run: () => conn.arrow(THREE) },
+  { name: 'three columns, columnar', run: () => conn.columnar(THREE) },
+  { name: 'three columns, rows', run: () => conn.query(THREE) },
+]
+
+console.log(`reading ${ROWS} rows, fastest of ${REPEATS}`)
+for (const { name, run } of cases) report(name, await time(run))
+
+// What a caller does next, which is the half the calls above do not
+// include. The bytes are read once here and decoded every round, so what
+// is timed is the decode and not the statement.
+const bytes = (await conn.arrow(THREE)).ipc
+
+console.log('')
+console.log('turning what came back into a table')
+report('tableFromIPC over the bytes', await time(async () => tableFromIPC(bytes)))
+
+// A batch is a slice of arrays that are already built, so the size is
+// about what the reader holds at once rather than about the write. This
+// says by how much, which is the answer to whether it is worth tuning.
+console.log('')
+console.log('the batch size the stream is cut into')
+for (const batchRows of [4_096, 65_536, 1_000_000]) {
+  report(`batchRows ${batchRows}`, await time(() => conn.arrow(THREE, null, { batchRows })))
+}
+
+await conn.close()
+await rm(dir, { recursive: true, force: true })

binding.d.cts

@@ -377,6 +377,46 @@ export interface ZuColumnar {
   readonly notices: ZuNotice[]
 }
 
+/**
+ * A whole result as Arrow, in the bytes Arrow ships between processes.
+ *
+ * The same buffers a columnar read hands over, with the schema written
+ * beside them, so `tableFromIPC(read.ipc)` is the whole of the reading
+ * code and every Arrow implementation is a reader. A result with no rows
+ * is a schema and one empty batch rather than nothing at all, so the
+ * columns are known either way.
+ */
+export interface ZuArrow {
+  /**
+   * The stream, as one buffer: a schema message and then a message a
+   * batch. It is the addon's own allocation handed over rather than
+   * copied, and it detaches when posted to a worker, which is what makes
+   * a result cross a thread without being cloned.
+   */
+  readonly ipc: Uint8Array
+  /** How many rows are in it, which the batches also add up to. */
+  readonly rows: number
+  readonly gqlstatus: string
+  readonly notices: ZuNotice[]
+}
+
+/**
+ * What a statement read as Arrow takes beside its parameters.
+ */
+export interface ZuArrowOptions extends ZuStatementOptions {
+  /**
+   * How many rows one record batch holds. Arrow's own 65,536 by
+   * default, which is what a reader expects and what keeps a batch
+   * inside a cache.
+   *
+   * The arrays are built whole either way and a batch is a slice of
+   * them, so this costs nothing to change and buys nothing to tune. It
+   * is worth naming when the reader on the other side has a size of its
+   * own, or when the batches are going somewhere one at a time.
+   */
+  readonly batchRows?: number
+}
+
 /**
  * The rows a statement gave back.
  *
@@ -998,6 +1038,29 @@ export declare class Connection {
    * JavaScript values on the way.
    */
   columnar(statement: string, params?: Record<string, ZuParam> | null, options?: ZuStatementOptions | null): Promise<ZuColumnar>
+  /**
+   * Runs one statement and gives back the bytes of an Arrow IPC
+   * stream.
+   *
+   * ```js
+   * import { tableFromIPC } from 'apache-arrow'
+   * const read = await conn.arrow('MATCH (a:account) RETURN a.name AS name, a.balance AS balance')
+   * const table = tableFromIPC(read.ipc)
+   * ```
+   *
+   * The same buffers [`Connection::columnar`] hands over, with the
+   * schema written beside them in the format every Arrow
+   * implementation already reads. That is the difference worth
+   * knowing: `columnar` is the fastest way out and leaves the caller
+   * to say what each buffer means, and this is the one where the
+   * result arrives as a table, a dataframe or a DuckDB relation with
+   * no code in between.
+   *
+   * The translation lives in the engine and is the same one the
+   * Python client exports through, so a node column names its table
+   * in both and a year-month duration is a month interval in both.
+   */
+  arrow(statement: string, params?: Record<string, ZuParam> | null, options?: ZuArrowOptions | null): Promise<ZuArrow>
   /**
    * Compiles a statement, pins it, and hands back something that
    * runs it.
@@ -1166,6 +1229,8 @@ export declare class Prepared {
   exec(params?: Record<string, ZuParam> | null, options?: ZuStatementOptions | null): Promise<void>
   /** Runs it and gives back its columns rather than its rows. */
   columnar(params?: Record<string, ZuParam> | null, options?: ZuStatementOptions | null): Promise<ZuColumnar>
+  /** Runs it and gives back the bytes of an Arrow IPC stream. */
+  arrow(params?: Record<string, ZuParam> | null, options?: ZuArrowOptions | null): Promise<ZuArrow>
   /**
    * Gives the id back to the session.
    *

etc/zudb.api.md

@@ -27,6 +27,7 @@ export function connect(path?: string | ConnectOptions | undefined | null, optio
 // @public
 export class Connection {
     appender(table: string): Promise<Appender>
+    arrow(statement: string, params?: Record<string, ZuParam> | null, options?: ZuArrowOptions | null): Promise<ZuArrow>
     close(): void
     columnar(statement: string, params?: Record<string, ZuParam> | null, options?: ZuStatementOptions | null): Promise<ZuColumnar>
     cursor(statement: string, params?: Record<string, ZuParam> | null, options?: ZuStreamOptions | null): ZuCursor
@@ -66,6 +67,7 @@ export function load(path: string, options: ZuLoadOptions): Promise<ZuLoadStats>
 
 // @public
 export class Prepared {
+    arrow(params?: Record<string, ZuParam> | null, options?: ZuArrowOptions | null): Promise<ZuArrow>
     close(): Promise<void>
     get closed(): boolean
     columnar(params?: Record<string, ZuParam> | null, options?: ZuStatementOptions | null): Promise<ZuColumnar>
@@ -101,6 +103,21 @@ export type ZuAppendValue =
 | ZuDuration
 | ZuTemporalValue
 
+// @public
+export interface ZuArrow {
+    // (undocumented)
+    readonly gqlstatus: string
+    readonly ipc: Uint8Array
+    // (undocumented)
+    readonly notices: ZuNotice[]
+    readonly rows: number
+}
+
+// @public
+export interface ZuArrowOptions extends ZuStatementOptions {
+    readonly batchRows?: number
+}
+
 // @public
 export interface ZuArrowTable {
     // (undocumented)

package.json

@@ -72,6 +72,7 @@
     "reference": "node tools/reference.mjs reference",
     "bench": "node bench/query.mjs",
     "bench:append": "node bench/append.mjs",
+    "bench:arrow": "node bench/arrow.mjs",
     "bench:columnar": "node bench/columnar.mjs",
     "bench:load": "node bench/load.mjs",
     "bench:prepared": "node bench/prepared.mjs",

src/arrow.rs

@@ -0,0 +1,107 @@
+//! A result as Arrow, in the bytes Arrow ships between processes.
+//!
+//! ```js
+//! import { tableFromIPC } from 'apache-arrow'
+//! const read = await conn.arrow('MATCH (p:person) RETURN p.name AS name, p.age AS age')
+//! const table = tableFromIPC(read.ipc)
+//! ```
+//!
+//! [`columnar`] hands over the buffers themselves and leaves the reader
+//! to put a type around them, which is the fastest way out and the one
+//! that costs a caller ten lines of Arrow before they have a table. This
+//! is the other way: the same buffers, with the schema written beside
+//! them, in the format every Arrow implementation already reads. What
+//! comes back is bytes, so `apache-arrow` reads it, DuckDB-Wasm reads
+//! it, a `fetch` response body carries it, and a worker gets it as a
+//! transferable rather than as a structured clone.
+//!
+//! The translation is not written here. `zu-arrow` in the engine tree is
+//! the one answer about what a zu column becomes in Arrow, shared with
+//! the Python client, because a second copy of it would be a second set
+//! of rules about what a year-month duration is. This module is the
+//! runtime's half: read the option, run the statement, hand the bytes to
+//! V8 without copying them again.
+//!
+//! Why bytes and not the C Data Interface, which is what the Python
+//! client takes and is a pointer rather than a serialization: nothing in
+//! a JavaScript runtime can dereference a pointer. An addon can, and
+//! this one does on the way in, but the value that reaches JavaScript
+//! has to be something V8 holds, and the only thing V8 holds that Arrow
+//! also speaks is a buffer of IPC bytes. The framing is the cost: a
+//! schema message, then a header a batch, which is kilobytes against a
+//! result of any size and the price of a format with readers.
+//!
+//! [`columnar`]: crate::columns
+use napi::bindgen_prelude::*;
+use napi::{Env, ScopedTask};
+use zudb::DiagnosticRecord;
+
+use crate::conn::{Failure, QueryTask, notices};
+
+/// One statement, read as Arrow.
+pub struct ArrowTask {
+    pub(crate) task: QueryTask,
+    /// How many rows go in a record batch, or why the caller's answer to
+    /// that could not be read.
+    pub(crate) batch: std::result::Result<usize, String>,
+}
+
+/// A whole result, as the bytes and what came with them.
+pub struct Read {
+    ipc: Vec<u8>,
+    rows: usize,
+    gqlstatus: &'static str,
+    notices: Vec<DiagnosticRecord>,
+}
+
+impl ArrowTask {
+    fn run(&mut self) -> std::result::Result<Read, Failure> {
+        // Before the statement, because a batch size nobody could read
+        // is a call that was wrong when it was written and not an answer
+        // worth running a scan for.
+        let batch = self
+            .batch
+            .as_ref()
+            .map_err(|message| Failure::Usage(message.clone()))?;
+        let batch = *batch;
+        let (result, shape) = self.task.run()?;
+        // The names are the ones the statement's own catalog gave, so a
+        // node column names its table rather than its id.
+        let ipc = zu_arrow::ipc(&result, shape.names(), batch)
+            .map_err(|err| Failure::Usage(err.to_string()))?;
+        Ok(Read {
+            ipc,
+            // Answered off the columns the sink filled, which is a read
+            // of a length and not a pass that builds rows nobody wants.
+            rows: result.rows.len(),
+            gqlstatus: result.status().code(),
+            notices: result.notices,
+        })
+    }
+}
+
+impl<'task> ScopedTask<'task> for ArrowTask {
+    type Output = std::result::Result<Read, Failure>;
+    type JsValue = Object<'task>;
+
+    fn compute(&mut self) -> Result<Self::Output> {
+        Ok(self.run())
+    }
+
+    fn resolve(&mut self, env: &'task Env, output: Self::Output) -> Result<Self::JsValue> {
+        let read = output.map_err(|failure| self.task.failed(env, failure))?;
+        let mut object = Object::new(env)?;
+        // The `Vec` is moved and not read: the pointer V8 is handed is
+        // the pointer the writer filled, and the allocation is freed when
+        // the typed array is collected.
+        object.set("ipc", Uint8Array::new(read.ipc))?;
+        object.set("rows", read.rows as f64)?;
+        object.set("gqlstatus", read.gqlstatus)?;
+        object.set("notices", notices(env, &read.notices)?)?;
+        Ok(object)
+    }
+
+    fn finally(mut self, env: Env) -> Result<()> {
+        self.task.release(&env)
+    }
+}