chwire

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.