chwire ClickHouse HTTP/TCP client and Native wire format toolkit for TypeScript. Install npm install @maxjustus/chwire HTTP vs TCP TCP works better for long-running queries and gives you streaming telemetry (progress, logs, profile events). HTTP works anywhere (including browsers) and supports all ClickHouse input/output formats. Quick Start HTTP Client The HTTP client is stateless — query() and insert() are standalone functions that each make a single HTTP request. import { insert, query, streamEncodeJsonEachRow, collectText } from "@maxjustus/chwire"; const connectionConfig = { url: "http://localhost:8123/", auth: { username: "default", password: "" }, }; // Insert const { summary } = await insert( "INSERT INTO table FORMAT JSONEachRow", streamEncodeJsonEachRow([{ id: 1, name: "test" }]), connectionConfig, ); console.log(`Wrote ${summary.written_rows} rows`); // Query const json = await collectText(query("SELECT * FROM table FORMAT JSON", connectionConfig)); // DDL await query("CREATE TABLE ...", connectionConfig); TCP Client import { TcpClient } from "@maxjustus/chwire/tcp"; const client = new TcpClient({ host: "localhost", port: 9000 }); await client.connect(); // Query for await (const packet of client.query("SELECT * FROM table")) { if (packet.type === "Data") { for (const row of packet.batch) console.log(row.id, row.name); } } // Insert await client.insert("INSERT INTO table", [{ id: 1, name: "alice" }]); client.close(); HTTP Client query() returns a CollectableAsyncGenerator that yields Data packets (raw Uint8Array chunks), Progress packets, and a final Summary. Use await to collect all packets into an array, or pipe through helpers like collectText and streamDecodeJsonEachRow which consume the Data chunks for you. insert() returns Promise<InsertResult> — an object with summary (containing written_rows, written_bytes, elapsed_ns, etc.) and queryId. Query Parameters const result = await collectText( query("SELECT {id: UInt64} as id, {name: String} as name FORMAT JSON", { ...connectionConfig, params: { id: 42, name: "Alice" }, }), ); Parameters are type-safe and prevent SQL injection. The type annotation (e.g., {name: String}) tells ClickHouse how to parse the value. Placeholders without a supplied value are sent as-is and resolved by the server. This makes parameterized-view DDL and session-level SET param_x = ... bindings work; a genuinely unbound parameter fails server-side with UNKNOWN_QUERY_PARAMETER. Unknown root-level option keys are forwarded as raw ClickHouse URL params: const result = await collectText(query("SELECT 42 as value", { ...connectionConfig, default_format: "TSV", wait_end_of_query: 1, })); The transport keys url, auth, compression, compressQuery, signal, timeout, clientVersion, settings, params, externalTables, queryId, and sessionId are reserved and are not forwarded as raw URL params. Parsing Query Results The query() function yields raw Uint8Array chunks aligned to compression blocks, not rows. Use helpers to parse: import { query, streamText, streamLines, streamDecodeJsonEachRow, collectJsonEachRow, collectText, collectBytes, } from "@maxjustus/chwire"; // JSONEachRow - streaming parsed objects for await (const row of streamDecodeJsonEachRow( query("SELECT * FROM t FORMAT JSONEachRow", connectionConfig), )) { console.log(row.id, row.name); } const res = await collectJsonEachRow( query("SELECT * FROM t FORMAT JSONEachRow", connectionConfig), ); // CSV/TSV - streaming raw lines for await (const line of streamLines( query("SELECT * FROM t FORMAT CSV", connectionConfig), )) { const [id, name] = line.split(","); } // JSON format - buffer entire response const json = await collectText( query("SELECT * FROM t FORMAT JSON", connectionConfig), ); const data = JSON.parse(json); Streaming Large Inserts The HTTP insert function accepts Uint8Array, Uint8Array[], or AsyncIterable<Uint8Array>. Use streamEncodeJsonEachRow for JSON data: // Streaming JSON objects async function* generateRows() { for (let i = 0; i < 1000000; i++) { yield { id: i, value: `data_${i}` }; } } await insert( "INSERT INTO large_table FORMAT JSONEachRow", streamEncodeJsonEachRow(generateRows()), { ...connectionConfig, compression: { method: "zstd", level: 6 }, onProgress: (p) => console.log(`${p.bytesUncompressed} bytes`), }, ); // Streaming raw bytes (any format) async function* generateCsvChunks() { const encoder = new TextEncoder(); for (let batch = 0; batch < 1000; batch++) { let chunk = ""; for (let i = 0; i < 1000; i++) { chunk += `${batch * 1000 + i},value_${i}\n`; } yield encoder.encode(chunk); } } await insert( "INSERT INTO large_table FORMAT CSV", generateCsvChunks(), { ...connectionConfig, compression: "lz4" }, ); External Tables Send temporary in-memory tables with your query. Schema is auto-extracted from RecordBatch (see Native Format for construction): import { batchFromCols, getCodec, query, collectText } from "@maxjustus/chwire"; const users = batchFromCols({ id: getCodec("UInt32").fromValues(new Uint32Array([1, 2, 3])), name: getCodec("String").fromValues(["Alice", "Bob", "Charlie"]), }); const result = await collectText(query( "SELECT * FROM users WHERE id > 1 FORMAT JSON", { url, auth, externalTables: { users } } )); For raw TSV/CSV/JSON data, use the explicit structure form: const result = await collectText(query( "SELECT * FROM mydata ORDER BY id FORMAT JSON", { url, auth, externalTables: { mydata: { structure: "id UInt32, name String", format: "TabSeparated", // or JSONEachRow, CSV, etc. data: "1\tAlice\n2\tBob\n" } } } )); Timeout and Cancellation Configure with timeout (ms) or provide an AbortSignal for manual cancellation: // Custom timeout await insert(sql, data, { ...connectionConfig, timeout: 60_000 }); // Manual cancellation const controller = new AbortController(); setTimeout(() => controller.abort(), 5000); await insert(sql, data, { ...connectionConfig, signal: controller.signal }); // Both (whichever triggers first) await insert(sql, data, { ...connectionConfig, signal: controller.signal, timeout: 60_000, }); Error Handling The HTTP client throws ClickHouseException for server errors: import { ClickHouseException } from "@maxjustus/chwire"; try { for await (const _ of query("SELECT * FROM nonexistent", config)) {} } catch (err) { if (err instanceof ClickHouseException) { console.log(err.code); // 60 (UNKNOWN_TABLE) console.log(err.exceptionName); // "DB::Exception" console.log(err.message); // "Table ... doesn't exist" } } Insert errors follow the same pattern: try { await insert("INSERT INTO t FORMAT JSONEachRow", data, config); } catch (err) { if (err instanceof ClickHouseException) { console.log(err.code); console.log(err.message); } } TCP Client Uses ClickHouse's native TCP protocol. TCP Single connection per client; use separate clients for concurrent operations. Note that the TCP protocol only sends/recieves data in Native format. Basic Usage import { TcpClient } from "@maxjustus/chwire/tcp"; const client = new TcpClient({ host: "localhost", port: 9000, database: "default", user: "default", password: "", }); await client.connect(); for await (const packet of client.query("SELECT * FROM table")) { if (packet.type === "Data") { for (const row of packet.batch) { console.log(row.id, row.name); } } } // DDL await client.query("CREATE TABLE ..."); // Insert (await collects and discards packets; for progress tracking see "Insert Progress Tracking" below) await client.insert("INSERT INTO table", [{ id: 1, name: "alice" }]); client.close(); TCP connections hold a persistent socket — always call client.close() when done, or use await using for automatic cleanup. Connection Options const client = new TcpClient({ host: "localhost", port: 9000, database: "default", user: "default", password: "", compression: "lz4", // 'lz4' | 'zstd' | false | { method: 'zstd', level: 6 } connectTimeout: 10000, // ms queryTimeout: 30000, // ms tls: true, // or Node.js tls.ConnectionOptions for custom CA/certs. IE: tls: { ca: fs.readFileSync("/path/to/ca.pem"), rejectUnauthorized: true }, }); Streaming Results Query yields packets - handle by type: for await (const packet of client.query(sql, { settings: { send_logs_level: "trace" } })) { switch (packet.type) { case "Data": console.log(`${packet.batch.rowCount} rows`); break; case "Progress": console.log(`${packet.progress.readRows} rows read`); break; case "Log": for (const entry of packet.entries) { console.log(`[${entry.source}] ${entry.text}`); } break; case "ProfileInfo": console.log(`${packet.info.rows} total rows`); break; case "EndOfStream": break; } } Progress Tracking Progress packets contain delta values (increments since the last packet). The client accumulates these into running totals available via packet.accumulated: for await (const packet of client.query(sql)) { if (packet.type === "Progress") { const { accumulated } = packet; console.log(`${accumulated.percent}% complete`); console.log(`Read: ${accumulated.readRows} rows, ${accumulated.readBytes} bytes`); console.log(`Elapsed: ${Number(accumulated.elapsedNs) / 1e9}s`); } } ProfileEvents and Resource Metrics ProfileEvents provide execution metrics. Memory and CPU stats are merged into accumulated progress: for await (const packet of client.query(sql)) { if (packet.type === "Progress") { const { accumulated } = packet; console.log(`Memory: ${accumulated.memoryUsage} bytes`); console.log(`Peak memory: ${accumulated.peakMemoryUsage} bytes`); console.log(`CPU time: ${accumulated.cpuTimeMicroseconds}µs`); console.log(`CPU cores utilized: ${accumulated.cpuUsage.toFixed(1)}`); } if (packet.type === "ProfileEvents") { // Raw accumulated event counters console.log(`Selected rows: ${packet.accumulated.get("SelectedRows")}`); console.log(`Read bytes: ${packet.accumulated.get("ReadCompressedBytes")}`); } } memoryUsage is the latest value; peakMemoryUsage is the max seen. cpuUsage shows equivalent CPUs utilized. Insert API The insert() method accepts RecordBatches or row objects: // Single batch await client.insert("INSERT INTO t", batch); // Multiple batches await client.insert("INSERT INTO t", [batch1, batch2]); // Row objects with auto-coercion. Types are inferred from server schema. // Unknown keys ignored, omitted keys use defaults, incompatible provided types throw. await client.insert("INSERT INTO t", [ { id: 1, name: "alice" }, { id: 2, name: "bob" }, ]); // Streaming rows with generator async function* generateRows() { for (let i = 0; i < 1000000; i++) { yield { id: i, name: `user${i}` }; } } // batchSize dictates number of rows per RecordBatch (native insert block) sent await client.insert("INSERT INTO t", generateRows(), { batchSize: 10000 }); // Schema validation (fail fast if types don't match the schema the server sends for the insert table) await client.insert("INSERT INTO t", rows, { schema: [ { name: "id", type: "UInt32" }, { name: "name", type: "String" }, ], }); Insert Progress Tracking Both query() and insert() return a CollectableAsyncGenerator<Packet>: await gen collects all packets into an array for await streams packets one at a time the same generator instance is single-consumer and does not replay once drained // Collect all packets const packets = await client.insert("INSERT INTO t", rows); const progress = packets.findLast(p => p.type === "Progress"); if (progress?.type === "Progress") { console.log(`Wrote ${progress.accumulated.writtenRows} rows`); } // Stream packets (useful for real-time progress on large inserts) for await (const packet of client.insert("INSERT INTO t", generateRows())) { if (packet.type === "Progress") { console.log(`Written: ${packet.accumulated.writtenRows} rows`); } } Query Parameters for await (const packet of client.query( "SELECT * FROM users WHERE age > {min_age: UInt32}", { params: { min_age: 18 } } )) { /* ... */ } Same {name: Type} syntax as HTTP. Parameters are type-safe and prevent SQL injection. External Tables Pass RecordBatches directly: import { batchFromCols, getCodec } from "@maxjustus/chwire"; const users = batchFromCols({ id: getCodec("UInt32").fromValues(new Uint32Array([1, 2, 3])), name: getCodec("String").fromValues(["Alice", "Bob", "Charlie"]), }); for await (const packet of client.query( "SELECT * FROM users WHERE id > 1", { externalTables: { users } } )) { if (packet.type === "Data") { for (const row of packet.batch) console.log(row.name); } } Supports streaming via iterables/async iterables of RecordBatch: async function* generateBatches() { for (let i = 0; i < 10; i++) { yield batchFromCols({ id: getCodec("UInt32").fromValues([i]) }); } } await client.query("SELECT sum(id) FROM data", { externalTables: { data: generateBatches() } }); Cancellation const controller = new AbortController(); setTimeout(() => controller.abort(), 5000); await client.connect({ signal: controller.signal }); for await (const p of client.query(sql, { signal: controller.signal })) { // ... } Auto-Close on Scope Exit await using client = await TcpClient.connect(options); // automatically closed when scope exits Error Handling The TCP client throws ClickHouseException for server errors: import { TcpClient, ClickHouseException } from "@maxjustus/chwire/tcp"; try { for await (const _ of client.query("SELECT * FROM nonexistent")) {} } catch (err) { if (err instanceof ClickHouseException) { console.log(err.code); // 60 (UNKNOWN_TABLE) console.log(err.exceptionName); // "DB::Exception" console.log(err.message); // "Table ... doesn't exist" console.log(err.serverStackTrace); // Full server-side stack trace console.log(err.nested); // Nested exception if present } } Connection and protocol errors throw JavaScript's built-in Error (not ClickHouseException): try { await client.connect(); } catch (err) { // err.message: "Connection timeout after 10000ms" // err.message: "Not connected" // err.message: "Connection busy - cannot run concurrent operations..." } Native Format Native is ClickHouse's columnar binary wire format. It's generally faster and smaller to serialize/deserialize vs JSON (see Performance below). Data arrives as RecordBatch objects. RecordBatch wraps typed column arrays you can iterate by row or access by column. Use it when throughput matters; use JSON when you want plain objects and don't need the speed.