Setting the file. One moment.
Columnar Test · Clickhouse Js Node Rowbinary · ClickHouse/agent-skills · Skills Docs
Back to the top of the page tests/columnar.test.ts
tests/ columnar.test.ts
TypeScript · 150 lines · 6 KB
12 * 1. correctness of every column, checked against a reference decode of the
13 * same buffer;
14 * 2. the streaming contract — chunk boundaries that fall mid-row must not
15 * corrupt or drop rows, and a truncated stream must throw;
16 * 3. THE COLUMNAR INVARIANT THAT MATTERS HERE: the Int64 (`ts`) column is filled
17 * by copying two 32-bit words, NOT via `getBigInt64`, so no bigint is
18 * allocated per row on the decode path. We prove it by spying on
19 * `DataView.prototype.getBigInt64` and asserting it is never called while
20 * decoding.
21 */
22
23 const STRIDE = 25 ;
24 const N = 1000 ;
25
26 // Deterministic, exactly representable values per column. `ts` is a DateTime64(3),
27 // whose wire form is Int64 millisecond ticks — here (1700000000 + i) * 1000.
28 const SELECT =
29 `SELECT toUInt32(number) AS sensor_id, ` +
30 `toDateTime64(1700000000 + number, 3) AS ts, ` +
31 `toFloat64(number) / 2 AS value, ` +
32 `toFloat32(number) AS quality, ` +
33 `toUInt8(number % 256) AS status ` +
34 `FROM numbers(${ N })` ;
35
36 const BUF = await query ( `${ SELECT } FORMAT RowBinary` );
37
38 /** Reference decode of the whole buffer — the test is free to allocate bigints. */
39 function reference ( buf : Buffer ) {
40 const view = new DataView (buf.buffer, buf.byteOffset, buf.byteLength);
41 const sensor_id : number [] = [];
42 const ts : bigint [] = [];
43 const value : number [] = [];
44 const quality : number [] = [];
45 const status : number [] = [];
46 for ( let o = 0 ; o + STRIDE <= buf. length ; o += STRIDE ) {
47 sensor_id. push (view. getUint32 (o, true ));
48 ts. push (view. getBigInt64 (o + 4 , true ));
49 value. push (view. getFloat64 (o + 12 , true ));
50 quality. push (view. getFloat32 (o + 20 , true ));
51 status. push (buf[o + 24 ] ! );
52 }
53 return { sensor_id, ts, value, quality, status };
54 }
55
56 /** Yield `buf` in chunks of the given repeating sizes (deliberately mid-row). */
57 async function* chunked (
58 buf : Buffer ,
59 sizes : number [],
60 ) : AsyncGenerator < Uint8Array > {
61 let o = 0 ;
62 let k = 0 ;
63 while (o < buf. length ) {
64 const len = sizes[k ++ % sizes. length ] ! ;
65 yield buf. subarray (o, Math. min (o + len, buf. length ));
66 o += len;
67 }
68 }
69
70 /** Drain the columnar stream into flat per-column arrays. */
71 async function collect ( chunks : AsyncIterable < Uint8Array >) {
72 const sensor_id : number [] = [];
73 const ts : bigint [] = [];
74 const value : number [] = [];
75 const quality : number [] = [];
76 const status : number [] = [];
77 let batches = 0 ;
78 for await ( const b of streamSensorColumns (chunks)) {
79 batches ++ ;
80 for ( let i = 0 ; i < b.rows; i ++ ) {
81 sensor_id. push (b.columns.sensor_id[i] ! );
82 ts. push (b.columns.ts[i] ! );
83 value. push (b.columns.value[i] ! );
84 quality. push (b.columns.quality[i] ! );
85 status. push (b.columns.status[i] ! );
86 }
87 }
88 return { sensor_id, ts, value, quality, status, batches };
89 }
90
91 describe ( "streamSensorColumns" , () => {
92 it ( "matches a reference decode of the live RowBinary buffer" , async () => {
93 const ref = reference ( BUF );
94 // One chunk = whole buffer.
95 const got = await collect ( chunked ( BUF , [ BUF . length ]));
96 expect (got.sensor_id). toEqual (ref.sensor_id);
97 expect (got.ts). toEqual (ref.ts);
98 expect (got.value). toEqual (ref.value);
99 expect (got.quality). toEqual (ref.quality);
100 expect (got.status). toEqual (ref.status);
101 expect (got.sensor_id. length ). toBe ( N );
102 });
103
104 it ( "survives chunk boundaries that split rows mid-field" , async () => {
105 const ref = reference ( BUF );
106 // Sizes coprime-ish to STRIDE (25) so boundaries land inside every field.
107 const got = await collect ( chunked ( BUF , [ 1 , 7 , 13 , 100 , 3 ]));
108 expect (got.batches). toBeGreaterThan ( 1 );
109 expect (got.sensor_id). toEqual (ref.sensor_id);
110 expect (got.ts). toEqual (ref.ts);
111 expect (got.value). toEqual (ref.value);
112 expect (got.quality). toEqual (ref.quality);
113 expect (got.status). toEqual (ref.status);
114 });
115
116 it ( "throws on a stream truncated mid-row" , async () => {
117 const truncated = BUF . subarray ( 0 , BUF . length - 3 );
118 await expect ( collect ( chunked (truncated, [ 256 ]))).rejects. toThrow ( /mid-row/ );
119 });
120
121 describe ( "Int64 column is not transferred through a bigint allocation" , () => {
122 const original = DataView . prototype .getBigInt64;
123 afterEach (() => {
124 DataView . prototype .getBigInt64 = original;
125 });
126
127 it ( "never calls getBigInt64 while decoding" , async () => {
128 let calls = 0 ;
129 // Spy that ALSO returns a correct value, so if the decoder regressed to
130 // using it the column would still be right — the test would fail only on
131 // the call count, pinpointing the allocation, not on a value mismatch.
132 DataView . prototype . getBigInt64 = function (
133 this : DataView ,
134 byteOffset : number ,
135 littleEndian ?: boolean ,
136 ) : bigint {
137 calls ++ ;
138 return original. call ( this , byteOffset, littleEndian);
139 };
140
141 const got = await collect ( chunked ( BUF , [ 1 , 7 , 13 , 100 , 3 ]));
142
143 expect (calls). toBe ( 0 );
144 // sanity: ts still decoded correctly via the two-word copy
145 expect (got.ts. length ). toBe ( N );
146 expect (got.ts[ 0 ]). toBe ( 1700000000 n * 1000 n );
147 expect (got.ts[ N - 1 ]). toBe ( BigInt ( 1700000000 + N - 1 ) * 1000 n );
148 });
149 });
150 });