Setting the file. One moment.
Streaming Row Bench · Clickhouse Js Node Rowbinary · ClickHouse/agent-skills · Skills Docs
Back to the top of the page
8 KB tests/ streamingRow.bench.ts
TypeScript · 227 lines · 8 KB
14 *
15 * To measure ONLY that mechanism, the model is deliberately stripped down:
16 *
17 * - The stream is ONE contiguous buffer whose AVAILABLE length grows. "More
18 * bytes arrived" = bump `avail` by a chunk. So no contender pays for
19 * stitching separate chunk buffers — that cost is identical for both in
20 * reality and would only add noise here.
21 * - The row shape is known (5 × little-endian UInt32 = 20 bytes), as it always
22 * is for a bespoke generated parser. So each parser checks whether the WHOLE
23 * row is available before reading any field. Consequence: neither parser
24 * ever re-reads a field on resume — `parseThrow` restarts from a clean row
25 * boundary, the generators suspend between rows. The "generators avoid
26 * re-parsing" argument therefore does NOT apply here; what's left is purely
27 * throw/catch unwinding vs generator resume.
28 *
29 * Contenders:
30 * - throw + restart — plain function, `throw MORE` when starved.
31 * - generator (yield* reader) — combinator style: `const a = yield* r.u32()`.
32 * Two levels of generator delegation per field;
33 * the "great on paper" form.
34 * - generator (inline yield) — one generator, row-level `while(...) yield`,
35 * fields read inline. The lean generator, shown
36 * so the combinator's overhead isn't mistaken
37 * for "all generators".
38 *
39 * Two chunk regimes are timed: a realistic large chunk (suspends rarely; steady
40 * state dominates) and a tiny sub-row chunk (suspends constantly; the mechanism
41 * cost dominates). Read the numbers on your own machine — that's the point.
42 */
43
44 const ROWS = 50_000 ;
45 const FIELDS_PER_ROW = 5 ;
46 const ROW_BYTES = FIELDS_PER_ROW * 4 ; // 5 × UInt32
47 const N = ROWS * FIELDS_PER_ROW ; // total field count
48 const TOTAL = ROWS * ROW_BYTES ;
49
50 // Build the payload once: field k (global index) holds the value k, so the
51 // expected checksum is the exact triangular sum and stays < 2^53 (no masking).
52 const PAYLOAD = new Uint8Array ( TOTAL );
53 {
54 const dv = new DataView ( PAYLOAD .buffer);
55 for ( let k = 0 ; k < N ; k ++ ) dv. setUint32 (k * 4 , k, true );
56 }
57 const EXPECTED = { rows: ROWS , sum: ( N * ( N - 1 )) / 2 };
58
59 type Result = { rows : number ; sum : number };
60
61 /** Singleton sentinel thrown on starvation — a bare value, so no Error stack is
62 * captured (that capture, not the unwind, is what makes throwing Errors slow). */
63 const MORE = Symbol ( "need-more-bytes" );
64
65 /**
66 * THROW approach. `avail` grows by `chunkSize` each time the parser starves.
67 * Reads restart from `committed` (the last completed row); because the whole-row
68 * check precedes any field read, a restart re-reads nothing.
69 */
70 function parseThrow ( bytes : Uint8Array , chunkSize : number ) : Result {
71 const dv = new DataView (bytes.buffer, bytes.byteOffset, bytes.byteLength);
72 let avail = 0 ;
73 let committed = 0 ;
74 let rows = 0 ;
75 let sum = 0 ;
76 for (;;) {
77 try {
78 let pos = committed;
79 while (pos < TOTAL ) {
80 if (pos + ROW_BYTES > avail) throw MORE ;
81 sum +=
82 dv. getUint32 (pos, true ) +
83 dv. getUint32 (pos + 4 , true ) +
84 dv. getUint32 (pos + 8 , true ) +
85 dv. getUint32 (pos + 12 , true ) +
86 dv. getUint32 (pos + 16 , true );
87 pos += ROW_BYTES ;
88 rows ++ ;
89 committed = pos;
90 }
91 return { rows, sum };
92 } catch (err) {
93 if (err !== MORE ) throw err;
94 avail = Math. min ( TOTAL , avail + chunkSize);
95 }
96 }
97 }
98
99 /**
100 * Combinator generator reader. Each `u32()` is itself a generator that suspends
101 * until its 4 bytes are available, so the parse body reads as if synchronous:
102 * `const a = yield* r.u32()`. `avail` is mutated on the reader between resumes.
103 */
104 function makeGenReader ( bytes : Uint8Array ) {
105 const dv = new DataView (bytes.buffer, bytes.byteOffset, bytes.byteLength);
106 return {
107 pos: 0 ,
108 avail: 0 ,
109 * u32 () : Generator < undefined , number , undefined > {
110 while ( this .pos + 4 > this .avail) yield ;
111 const v = dv. getUint32 ( this .pos, true );
112 this .pos += 4 ;
113 return v;
114 },
115 };
116 }
117
118 function* parseGenCombinator (
119 r : ReturnType < typeof makeGenReader>,
120 ) : Generator < undefined , Result , undefined > {
121 let rows = 0 ;
122 let sum = 0 ;
123 while (r.pos < TOTAL ) {
124 const a = yield* r. u32 ();
125 const b = yield* r. u32 ();
126 const c = yield* r. u32 ();
127 const d = yield* r. u32 ();
128 const e = yield* r. u32 ();
129 sum += a + b + c + d + e;
130 rows ++ ;
131 }
132 return { rows, sum };
133 }
134
135 /**
136 * Lean generator: a single generator, whole-row availability checked with an
137 * inline `while (...) yield`, fields read inline. No per-field delegation, so it
138 * runs at near-normal speed between the (rare) suspensions.
139 */
140 function* parseGenInline (
141 bytes : Uint8Array ,
142 box : { avail : number },
143 ) : Generator < undefined , Result , undefined > {
144 const dv = new DataView (bytes.buffer, bytes.byteOffset, bytes.byteLength);
145 let pos = 0 ;
146 let rows = 0 ;
147 let sum = 0 ;
148 while (pos < TOTAL ) {
149 while (pos + ROW_BYTES > box.avail) yield ;
150 sum +=
151 dv. getUint32 (pos, true ) +
152 dv. getUint32 (pos + 4 , true ) +
153 dv. getUint32 (pos + 8 , true ) +
154 dv. getUint32 (pos + 12 , true ) +
155 dv. getUint32 (pos + 16 , true );
156 pos += ROW_BYTES ;
157 rows ++ ;
158 }
159 return { rows, sum };
160 }
161
162 /** Drive a generator to completion, revealing `chunkSize` more bytes per
163 * suspension via the shared `holder.avail`. Returns the generator's `return`. */
164 function driveGen (
165 gen : Generator < undefined , Result , undefined >,
166 holder : { avail : number },
167 chunkSize : number ,
168 ) : Result {
169 let step = gen. next ();
170 while ( ! step.done) {
171 holder.avail = Math. min ( TOTAL , holder.avail + chunkSize);
172 step = gen. next ();
173 }
174 return step.value;
175 }
176
177 function runGenCombinator ( chunkSize : number ) : Result {
178 const r = makeGenReader ( PAYLOAD );
179 return driveGen ( parseGenCombinator (r), r, chunkSize);
180 }
181
182 function runGenInline ( chunkSize : number ) : Result {
183 const box = { avail: 0 };
184 return driveGen ( parseGenInline ( PAYLOAD , box), box, chunkSize);
185 }
186
187 // Equivalence guard: a faster wrong answer is worthless. Validate every
188 // contender at both chunk regimes before any timing runs.
189 function assertCorrect ( label : string , got : Result ) : void {
190 if (got.rows !== EXPECTED .rows || got.sum !== EXPECTED .sum) {
191 throw new Error (
192 `${ label } mismatch: got rows=${ got . rows } sum=${ got . sum }, ` +
193 `expected rows=${ EXPECTED . rows } sum=${ EXPECTED . sum }` ,
194 );
195 }
196 }
197 for ( const cs of [ 64 * 1024 , 8 ]) {
198 assertCorrect ( `throw cs=${ cs }` , parseThrow ( PAYLOAD , cs));
199 assertCorrect ( `gen-combinator cs=${ cs }` , runGenCombinator (cs));
200 assertCorrect ( `gen-inline cs=${ cs }` , runGenInline (cs));
201 }
202
203 describe ( "streaming need-more-bytes: 64 KB chunks (suspends rarely)" , () => {
204 const cs = 64 * 1024 ;
205 bench ( "throw + restart" , () => {
206 parseThrow ( PAYLOAD , cs);
207 });
208 bench ( "generator (yield* reader)" , () => {
209 runGenCombinator (cs);
210 });
211 bench ( "generator (inline yield)" , () => {
212 runGenInline (cs);
213 });
214 });
215
216 describe ( "streaming need-more-bytes: 8-byte chunks (suspends constantly)" , () => {
217 const cs = 8 ;
218 bench ( "throw + restart" , () => {
219 parseThrow ( PAYLOAD , cs);
220 });
221 bench ( "generator (yield* reader)" , () => {
222 runGenCombinator (cs);
223 });
224 bench ( "generator (inline yield)" , () => {
225 runGenInline (cs);
226 });
227 });