refactor: replace NDJSON spill/streaming with columnar binary format

Data was being serialized as row-based JSON (NDJSON) for spill storage
and streaming despite the source data being columnar (Lance/Parquet).
This introduced unnecessary overhead: column name repetition per row,
text encoding, and BigInt workarounds (__bigint__ prefix hack).

New columnar binary format (QMCB): header + per-column typed arrays
(Float64Array, BigInt64Array, packed bools, offset+data strings,
dimensioned float vectors) with null bitmaps. Applied to:
- R2SpillBackend (edge spill)
- FsSpillBackend (local spill)
- streamRpc + handleQueryStream (DO → client streaming)
- executeStream client-side parsing (frame reassembly)

Author: teamchong

src/client.ts

@@ -185,7 +185,7 @@ export class TableQuery<T extends Row = Row> {
     return this._executor.execute(this.toDescriptor()) as Promise<QueryResult<T>>;
   }
 
-  /** Execute and return an NDJSON stream of rows. Only works with RemoteExecutor. */
+  /** Execute and return a columnar binary stream of rows. Only works with RemoteExecutor. */
   async execStream(): Promise<ReadableStream<Row>> {
     if (!this._executor.executeStream) {
       throw new Error("execStream() requires a remote executor with streaming support");

src/index.ts

@@ -139,36 +139,47 @@ class RemoteExecutor implements QueryExecutor {
     return masterRpc.appendRpc(table, rows);
   }
 
-  /** Stream query results via RPC — native ReadableStream, no HTTP wrapper. */
+  /** Stream query results via RPC — columnar binary framed stream. */
   async executeStream(query: QueryDescriptor): Promise<ReadableStream<Row>> {
+    const { decodeColumnarRun } = await import("./r2-spill.js");
     const rpc = this.getQueryHandle();
     const byteStream = await rpc.streamRpc(query);
 
-    // Parse NDJSON byte stream into Row objects
-    const decoder = new TextDecoder();
-    const MAX_STREAM_BUFFER = 10 * 1024 * 1024;
-    let buffer = "";
-
+    // Parse length-prefixed columnar binary frames into Row objects
     return new ReadableStream<Row>({
       async start(controller) {
         const reader = byteStream.getReader();
+        let pending = new Uint8Array(0);
+
+        const concat = (a: Uint8Array, b: Uint8Array): Uint8Array => {
+          const out = new Uint8Array(a.length + b.length);
+          out.set(a, 0);
+          out.set(b, a.length);
+          return out;
+        };
+
         try {
           while (true) {
             const { done, value } = await reader.read();
             if (done) break;
-            buffer += decoder.decode(value, { stream: true });
-            if (buffer.length > MAX_STREAM_BUFFER) {
-              controller.error(new Error("Stream buffer exceeded 10MB"));
-              reader.cancel();
-              return;
-            }
-            const lines = buffer.split("\n");
-            buffer = lines.pop()!;
-            for (const line of lines) {
-              if (line.trim()) controller.enqueue(JSON.parse(line) as Row);
+            pending = pending.length > 0 ? concat(pending, value) : value;
+
+            // Process complete frames
+            while (pending.length >= 4) {
+              const frameLen = new DataView(pending.buffer, pending.byteOffset).getUint32(0, true);
+              if (pending.length < 4 + frameLen) break; // wait for more data
+
+              const frameBuf = pending.buffer.slice(
+                pending.byteOffset + 4,
+                pending.byteOffset + 4 + frameLen,
+              );
+              pending = pending.subarray(4 + frameLen);
+
+              for (const row of decodeColumnarRun(frameBuf)) {
+                controller.enqueue(row);
+              }
             }
           }
-          if (buffer.trim()) controller.enqueue(JSON.parse(buffer) as Row);
           controller.close();
         } catch (err) {
           controller.error(err);

src/operators.ts

@@ -891,25 +891,12 @@ export function estimateRowSize(row: Row): number {
   return size;
 }
 
-/** JSON replacer for bigint values in NDJSON runs. */
-function bigIntJsonReplacer(_key: string, value: unknown): unknown {
-  return typeof value === "bigint" ? `__bigint__${value.toString()}` : value;
-}
-
-/** Parse a single NDJSON line, restoring bigint values. */
-function parseNdjsonRow(line: string): Row {
-  return JSON.parse(line, (_key, value) => {
-    if (typeof value === "string" && value.startsWith("__bigint__")) {
-      return BigInt(value.slice(10));
-    }
-    return value;
-  }) as Row;
-}
-
-// Re-export SpillBackend from r2-spill so consumers can import from either place
+// Re-export SpillBackend + columnar codec from r2-spill
 export type { SpillBackend } from "./r2-spill.js";
+export { encodeColumnarRun, decodeColumnarRun } from "./r2-spill.js";
+import { encodeColumnarRun, decodeColumnarRun } from "./r2-spill.js";
 
-/** Filesystem-backed spill for Node/Bun environments. */
+/** Filesystem-backed spill for Node/Bun environments using columnar binary format. */
 export class FsSpillBackend {
   private runFiles: string[] = [];
   private tmpDir: string | null = null;
@@ -925,30 +912,20 @@ export class FsSpillBackend {
     }
     const path = await import("node:path");
     const fs = await import("node:fs/promises");
-    const runPath = path.join(this.tmpDir!, `run_${this.runFiles.length}.ndjson`);
-    const lines = rows.map(row => JSON.stringify(row, bigIntJsonReplacer));
-    const body = lines.join("\n") + "\n";
-    this.bytesWritten += Buffer.byteLength(body, "utf8");
-    await fs.writeFile(runPath, body);
+    const runPath = path.join(this.tmpDir!, `run_${this.runFiles.length}.bin`);
+    const buf = encodeColumnarRun(rows);
+    this.bytesWritten += buf.byteLength;
+    await fs.writeFile(runPath, new Uint8Array(buf));
     this.runFiles.push(runPath);
     return runPath;
   }
 
   async *streamRun(spillId: string): AsyncGenerator<Row> {
-    const nodeFs = await import("node:fs");
-    const readline = await import("node:readline");
-    const stream = nodeFs.createReadStream(spillId);
-    const rl = readline.createInterface({ input: stream, crlfDelay: Infinity });
-    try {
-      for await (const line of rl) {
-        if (line.length === 0) continue;
-        this.bytesRead += Buffer.byteLength(line, "utf8") + 1; // +1 for newline
-        yield parseNdjsonRow(line);
-      }
-    } finally {
-      rl.close();
-      stream.destroy();
-    }
+    const fs = await import("node:fs/promises");
+    const fileData = await fs.readFile(spillId);
+    const buf = fileData.buffer.slice(fileData.byteOffset, fileData.byteOffset + fileData.byteLength);
+    this.bytesRead += buf.byteLength;
+    yield* decodeColumnarRun(buf);
   }
 
   async cleanup(): Promise<void> {

src/query-do.ts

@@ -9,7 +9,7 @@ import { decodeParquetColumnChunk } from "./parquet-decode.js";
 import { instantiateWasm, type WasmEngine } from "./wasm-engine.js";
 import { mergeQueryResults } from "./merge.js";
 import { coalesceRanges, fetchBounded, withRetry, withTimeout } from "./coalesce.js";
-import { R2SpillBackend } from "./r2-spill.js";
+import { R2SpillBackend, encodeColumnarRun } from "./r2-spill.js";
 import {
   type Operator, type RowBatch,
   buildEdgePipeline, drainPipeline, estimateRowSize,
@@ -1806,30 +1806,35 @@ export class QueryDO extends DurableObject<Env> {
     }
   }
 
-  /** Stream query results as NDJSON. */
+  /** Stream query results as columnar binary batches. */
   private async handleQueryStream(request: Request): Promise<Response> {
     const body = await request.json();
     let query: QueryDescriptor;
     try { query = this.parseQuery(body); } catch (err) { return this.json({ error: this.errMsg(err) }, 400); }
     let result: QueryResult;
     try { result = await this.executeQuery(query); } catch (err) { return this.json({ error: this.errMsg(err) }, 500); }
 
-    const { readable, writable } = new TransformStream<Uint8Array>();
-    const writer = writable.getWriter();
-    const encoder = new TextEncoder();
-
-    (async () => {
-      try {
-        for (const row of result.rows) {
-          await writer.write(encoder.encode(JSON.stringify(row, bigIntReplacer) + "\n"));
-        }
-      } finally {
-        await writer.close();
-      }
-    })();
+    // Encode as columnar binary with length-prefixed batches for streaming
+    const STREAM_BATCH_SIZE = 4096;
+    const rows = result.rows;
+    let idx = 0;
+    const stream = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        if (idx >= rows.length) { controller.close(); return; }
+        const end = Math.min(idx + STREAM_BATCH_SIZE, rows.length);
+        const batch = rows.slice(idx, end);
+        idx = end;
+        const buf = encodeColumnarRun(batch);
+        // Length-prefix each batch: 4 bytes little-endian uint32 + columnar payload
+        const frame = new Uint8Array(4 + buf.byteLength);
+        new DataView(frame.buffer).setUint32(0, buf.byteLength, true);
+        frame.set(new Uint8Array(buf), 4);
+        controller.enqueue(frame);
+      },
+    });
 
-    return new Response(readable, {
-      headers: { "content-type": "application/x-ndjson" },
+    return new Response(stream, {
+      headers: { "content-type": "application/x-querymode-columnar" },
     });
   }
 
@@ -1873,13 +1878,21 @@ export class QueryDO extends DurableObject<Env> {
 
   async streamRpc(descriptor: unknown): Promise<ReadableStream<Uint8Array>> {
     const result = await this.executeQuery(await this.rpcParseQuery(descriptor));
-    const encoder = new TextEncoder();
     const rows = result.rows;
+    const STREAM_BATCH_SIZE = 4096;
     let idx = 0;
     return new ReadableStream<Uint8Array>({
       pull(controller) {
         if (idx >= rows.length) { controller.close(); return; }
-        controller.enqueue(encoder.encode(JSON.stringify(rows[idx++], bigIntReplacer) + "\n"));
+        const end = Math.min(idx + STREAM_BATCH_SIZE, rows.length);
+        const batch = rows.slice(idx, end);
+        idx = end;
+        const buf = encodeColumnarRun(batch);
+        // Length-prefix: 4 bytes LE uint32 + payload
+        const frame = new Uint8Array(4 + buf.byteLength);
+        new DataView(frame.buffer).setUint32(0, buf.byteLength, true);
+        frame.set(new Uint8Array(buf), 4);
+        controller.enqueue(frame);
       },
     });
   }