Rows

1import { NeedMoreData, type Reader } from "./core.js";
2
3/**
4 * Drive `readRow` over every row of a plain `RowBinary` result into an array.
5 * Curried: `readRows(readRow)` returns a `Reader<T[]>`. Rows are concatenated on
6 * the wire with no count, length prefix, or delimiter, so the result is exhausted
7 * only when the cursor reaches the buffer end.
8 *
9 * `readRow` must consume EXACTLY one row's bytes — a byte short or long compounds
10 * across rows and the cursor overshoots or never lands on `buf.length`. Returns
11 * `[]` for an empty buffer. When generating code, inline the per-column reads
12 * into the loop body:
13 *
14 *   function readRowsUser(s) {
15 *     const out = [];
16 *     while (s.pos < s.buf.length) {
17 *       out.push({ id: readUInt64(s), name: readString(s) });
18 *     }
19 *     return out;
20 *   }
21 *
22 * STREAMING (partial trailing row): a chunk of a still-arriving response may end
23 * mid-row. `pos` is committed only AFTER a row reads cleanly, so when a row
24 * starves and `readRow` throws {@link NeedMoreData}, this catches it, rewinds
25 * `pos` to the last complete row boundary, and returns the rows so far — never a
26 * half-built row. The cursor is left at the straddling row, a commit point the
27 * driver carries forward:
28 *
29 *   const drive = readRows(readRow);
30 *   let committed = 0;
31 *   for (const chunk of chunks) {              // chunk = growing prefix
32 *     const s = new Cursor(chunk);
33 *     s.pos = committed;
34 *     emit(drive(s));                          // complete rows in this chunk
35 *     committed = s.pos;                        // start of the straddling row
36 *   }
37 *
38 * On a complete buffer no read starves, so the catch never runs. Errors other
39 * than {@link NeedMoreData} are real decode faults and propagate. See also
40 * `streamRowBatches`, the async driver built on this.
41 */
42export function readRows<T>(readRow: Reader<T>): Reader<T[]> {
43  return (state) => {
44    const out: T[] = [];
45    let committed = state.pos;
46    try {
47      while (state.pos < state.buf.length) {
48        const row = readRow(state);
49        committed = state.pos; // row read cleanly — advance the commit point
50        out.push(row);
51      }
52    } catch (e) {
53      if (e !== NeedMoreData) throw e;
54      state.pos = committed; // drop the partial trailing row; resume next chunk
55    }
56    return out;
57  };
58}