Setting the file. One moment.
Rows Write Test · Clickhouse Js Node Rowbinary · ClickHouse/agent-skills · Skills Docs
Back to the top of the page tests/Rows.write.test.ts
tests/ Rows.write.test.ts
TypeScript · 194 lines · 8 KB
from
"../src/writers/core.js"
;
10 import { writeTupleNamed } from "../src/writers/composite.js" ;
11 import { writeUInt64, writeUInt32 } from "../src/writers/integers.js" ;
12 import { writeString } from "../src/writers/strings.js" ;
13
14 type Row = {
15 id : bigint ;
16 n : number ;
17 name : string ;
18 };
19
20 const writeRow = writeTupleNamed < Row >({
21 id: writeUInt64,
22 n: writeUInt32,
23 name: writeString,
24 });
25
26 /**
27 * Drive `writeRows` to completion and concatenate every yielded buffer — the
28 * canonical driver loop. `bufferSize` sizes each (fixed) sink, so a small value
29 * forces the overflow + flush path; the default fits the rows in one buffer.
30 */
31 function encodeRows (
32 write : Writer < Row >,
33 rows : Iterable < Row >,
34 bufferSize = 4096 ,
35 ) : Buffer {
36 return Buffer. concat ([ ... writeRows (write)(rows, bufferSize)]);
37 }
38
39 describe ( "writeRows" , () => {
40 const rows : Row [] = Array. from ({ length: 5 }, ( _ , i ) => ({
41 id: BigInt (i),
42 n: i * 10 ,
43 name: `row${ i }` ,
44 }));
45 const sql =
46 "SELECT toUInt64(number) AS id, toUInt32(number * 10) AS n, concat('row', toString(number)) AS name " +
47 "FROM numbers(5) FORMAT RowBinary" ;
48
49 it ( "encodes a plain RowBinary result of several rows" , async () => {
50 expect ( encodeRows (writeRow, rows)). toEqual ( await query (sql));
51 });
52
53 it ( "writes nothing for an empty array" , () =>
54 expect ( encodeRows (writeRow, []). length ). toBe ( 0 ));
55
56 it ( "flushes on buffer overflow and resumes — same bytes across a tiny buffer" , async () => {
57 // A buffer far smaller than the whole result: the driver must flush full
58 // buffers mid-stream at row boundaries and reassemble to the identical bytes.
59 const expected = await query (sql);
60 const tiny = encodeRows (writeRow, rows, 20 ); // 20 holds one 17-byte row, not two
61 expect (tiny). toEqual (expected);
62 });
63
64 it ( "yields at row boundaries, never a half-written row" , () => {
65 // bufferSize holds two rows + change but not three, so the first yield must
66 // land exactly on a row boundary (a whole number of rows), not mid-row.
67 const gen = writeRows (writeRow)(rows, 40 );
68 const first = gen. next ();
69 expect (first.done). toBe ( false );
70 const flushed = first.value as Buffer ;
71 // A prefix check alone is NOT enough — a mid-row split is also a prefix. Prove
72 // the flush ends EXACTLY on a row boundary: collect the per-row cumulative byte
73 // offsets and assert the flushed length is one of them (i.e. a whole number of
74 // rows), AND that the bytes are the matching prefix of the full encoding.
75 const boundaries = new Set < number >();
76 let acc = 0 ;
77 for ( const row of rows) {
78 acc += encodeRows (writeRow, [row]). length ;
79 boundaries. add (acc);
80 }
81 expect (boundaries. has (flushed. length )). toBe ( true ); // ends on a row boundary
82 const full = encodeRows (writeRow, rows);
83 expect (full. subarray ( 0 , flushed. length )). toEqual (flushed); // and is that prefix
84 });
85
86 it ( "yields independent buffers — a flushed buffer survives the next iteration" , () => {
87 // Each flush gets a fresh buffer, so an earlier yield isn't clobbered when
88 // the generator resumes. Collect two buffers, then assert the first is intact.
89 const gen = writeRows (writeRow)(rows, 20 );
90 const a = gen. next ().value as Buffer ;
91 const snapshot = Buffer. from (a); // independent copy of what we saw first
92 gen. next (); // resume: writes the next row into a NEW buffer
93 expect (a). toEqual (snapshot); // `a` must be untouched
94 });
95
96 it ( "grows the buffer to fit a row larger than bufferSize, warning once" , async () => {
97 const expected = await query (sql);
98 const warn = vi. spyOn (console, "warn" ). mockImplementation (() => {});
99 try {
100 // bufferSize 4 can't hold even one 17-byte row: the buffer doubles
101 // (4→8→16→32) until the row fits — no data lost, nothing thrown — and it
102 // warns exactly once even though it grew several times.
103 expect ( encodeRows (writeRow, rows, 4 )). toEqual (expected);
104 expect (warn). toHaveBeenCalledTimes ( 1 );
105 expect (warn.mock.calls[ 0 ]?.[ 0 ]). toMatch ( /didn't fit bufferSize=4/ );
106 } finally {
107 warn. mockRestore ();
108 }
109 });
110
111 it ( "does not warn when every row fits the buffer" , () => {
112 const warn = vi. spyOn (console, "warn" ). mockImplementation (() => {});
113 try {
114 encodeRows (writeRow, rows); // default 4096 fits every row
115 expect (warn).not. toHaveBeenCalled ();
116 } finally {
117 warn. mockRestore ();
118 }
119 });
120
121 it ( "rejects a non-positive or non-integer bufferSize instead of looping forever" , () => {
122 // 0 / NaN would make the growth loop spin (size *= 2 never escapes 0/NaN);
123 // fail fast on first .next() with a clear error.
124 for ( const bad of [ 0 , - 1 , NaN , 1.5 , Infinity ]) {
125 const gen = writeRows (writeRow)(rows, bad);
126 expect (() => gen. next ()). toThrow ( /bufferSize must be a positive integer/ );
127 }
128 });
129
130 /** Run `body` with a subscriber on the flush channel, collecting every event. */
131 function withFlushEvents ( body : () => void ) : WriteRowsFlush [] {
132 const events : WriteRowsFlush [] = [];
133 const onMessage = ( msg : unknown ) => events. push (msg as WriteRowsFlush );
134 diagnostics_channel. subscribe ( FLUSH_CHANNEL_NAME , onMessage);
135 try {
136 body ();
137 } finally {
138 diagnostics_channel. unsubscribe ( FLUSH_CHANNEL_NAME , onMessage);
139 }
140 return events;
141 }
142
143 it ( "publishes a flush event per buffer — every 'full' batch fills its capacity, then one 'end'" , () => {
144 // bufferSize 20 holds one 17-byte row: rows 0..3 each flush a 'full' buffer
145 // when the next row overflows, row 4 comes out as the 'end' batch.
146 const events = withFlushEvents (() => encodeRows (writeRow, rows, 20 ));
147 expect (events. map (( e ) => e.reason)). toEqual ([
148 "full" ,
149 "full" ,
150 "full" ,
151 "full" ,
152 "end" ,
153 ]);
154 // Every buffer reports its real capacity and the configured size; used never
155 // exceeds capacity; nothing grew, so capacity stays at bufferSize.
156 for ( const e of events) {
157 expect (e.capacityBytes). toBe ( 20 );
158 expect (e.bufferSize). toBe ( 20 );
159 expect (e.usedBytes). toBeLessThanOrEqual (e.capacityBytes);
160 }
161 // The four mid-stream flushes each carried exactly one 17-byte row.
162 expect (events. slice ( 0 , 4 ). every (( e ) => e.usedBytes === 17 )). toBe ( true );
163 // Summed used bytes equal the whole payload — nothing is double-counted.
164 const total = events. reduce (( n , e ) => n + e.usedBytes, 0 );
165 expect (total). toBe ( encodeRows (writeRow, rows). length );
166 });
167
168 it ( "reports the grown capacity and original bufferSize so overflow is identifiable" , () => {
169 const warn = vi. spyOn (console, "warn" ). mockImplementation (() => {});
170 try {
171 // bufferSize 4 grows to 32 to fit a 17-byte row; the published capacity is
172 // the grown size while bufferSize stays 4, so `capacityBytes > bufferSize`
173 // flags the overflow and `usedBytes / bufferSize` gives its magnitude.
174 const events = withFlushEvents (() => encodeRows (writeRow, rows, 4 ));
175 expect (events. every (( e ) => e.capacityBytes === 32 )). toBe ( true );
176 expect (events. every (( e ) => e.bufferSize === 4 )). toBe ( true );
177 expect (events. every (( e ) => e.capacityBytes > e.bufferSize)). toBe ( true );
178 } finally {
179 warn. mockRestore ();
180 }
181 });
182
183 it ( "stops publishing to a subscriber once it unsubscribes" , () => {
184 const events : WriteRowsFlush [] = [];
185 const onMessage = ( msg : unknown ) => events. push (msg as WriteRowsFlush );
186 diagnostics_channel. subscribe ( FLUSH_CHANNEL_NAME , onMessage);
187 encodeRows (writeRow, rows, 20 );
188 const afterFirst = events. length ;
189 expect (afterFirst). toBeGreaterThan ( 0 );
190 diagnostics_channel. unsubscribe ( FLUSH_CHANNEL_NAME , onMessage);
191 encodeRows (writeRow, rows, 20 ); // a second run with no subscriber
192 expect (events. length ). toBe (afterFirst); // nothing more delivered
193 });
194 });