Open History to see the last commit
1/**
2 * ClickHouse driver for Effect SQL, backed by `@clickhouse/client`.
3 *
4 * This module provides both the ClickHouse-specific {@link ClickhouseClient}
5 * service and the generic {@link Client.SqlClient} service. `make` creates a
6 * scoped client, checks the connection with `ping()`, maps ClickHouse errors
7 * to `SqlError`, and aborts in-flight queries when interrupted. The
8 * ClickHouse-specific service adds typed parameters, command execution, insert
9 * queries, query id and settings helpers, a statement compiler, and direct or
10 * config-backed layers.
11 *
12 * @stability unstable
13 * @since 4.0.0
14 */
15import * as Clickhouse from "@clickhouse/client"
16import * as NodeStream from "@effect/platform-node/NodeStream"
17import * as Config from "effect/Config"
18import * as Context from "effect/Context"
19import * as Duration from "effect/Duration"
20import * as Effect from "effect/Effect"
21import * as Fiber from "effect/Fiber"
22import { dual } from "effect/Function"
23import * as Layer from "effect/Layer"
24import * as Reactivity from "effect/reactivity/Reactivity"
25import type * as Scope from "effect/Scope"
26import * as Client from "effect/sql/SqlClient"
27import type { Connection } from "effect/sql/SqlConnection"
28import {
29 AuthenticationError,
30 AuthorizationError,
31 ConnectionError,
32 SqlError,
33 SqlSyntaxError,
34 StatementTimeoutError,
35 UnknownError
36} from "effect/sql/SqlError"
37import * as Statement from "effect/sql/Statement"
38import * as Stream from "effect/Stream"
39import * as Crypto from "node:crypto"
40import type { Readable } from "node:stream"
41
42const ATTR_DB_SYSTEM_NAME = "db.system.name"
43const ATTR_DB_NAMESPACE = "db.namespace"
44
45const clickhouseCodeFromCause = (cause: unknown): number | undefined => {
46 if (typeof cause !== "object" || cause === null || !("code" in cause)) {
47 return undefined
48 }
49 const code = cause.code
50 if (typeof code === "number") {
51 return code
52 }
53 if (typeof code === "string") {
54 const parsed = Number(code)
55 return Number.isNaN(parsed) ? undefined : parsed
56 }
57 return undefined
58}
59
60const clickhouseSyntaxErrorCodes = new Set([36, 60, 62, 242])
61
62const classifyError = (
63 cause: unknown,
64 message: string,
65 operation: string,
66 fallback: "connection" | "unknown" = "unknown"
67) => {
68 const props = { cause, message, operation }
69 const code = clickhouseCodeFromCause(cause)
70 if (code !== undefined) {
71 if (code === 516) {
72 return new AuthenticationError(props)
73 }
74 if (code === 497) {
75 return new AuthorizationError(props)
76 }
77 if (clickhouseSyntaxErrorCodes.has(code)) {
78 return new SqlSyntaxError(props)
79 }
80 if (code === 159 || code === 469) {
81 return new StatementTimeoutError(props)
82 }
83 }
84 return fallback === "connection" ? new ConnectionError(props) : new UnknownError(props)
85}
86
87const killQuery = (client: Clickhouse.ClickHouseClient, queryId: string) =>
88 Effect.promise(() =>
89 client.command({
90 query: "KILL QUERY WHERE query_id = {queryId:String}",
91 query_params: { queryId }
92 })
93 )
94
95/**
96 * Unique runtime identifier used to tag `ClickhouseClient` values.
97 *
98 * @stability unstable
99 * @category type IDs
100 * @since 4.0.0
101 */
102export const TypeId: TypeId = "~@effect/sql-clickhouse/ClickhouseClient"
103
104/**
105 * Type-level literal for the `ClickhouseClient` runtime identifier.
106 *
107 * @stability unstable
108 * @category type IDs
109 * @since 4.0.0
110 */
111export type TypeId = "~@effect/sql-clickhouse/ClickhouseClient"
112
113/**
114 * ClickHouse-specific `SqlClient` extension with access to its configuration,
115 * the underlying `@clickhouse/client` instance, typed parameter fragments,
116 * command-mode execution, insert queries, and per-effect query ID and
117 * ClickHouse settings.
118 *
119 * @stability unstable
120 * @category services
121 * @since 4.0.0
122 */
123export interface ClickhouseClient extends Client.SqlClient {
124 readonly [TypeId]: TypeId
125 readonly config: ClickhouseClientConfig
126 /**
127 * The `@clickhouse/client` instance behind this client, for driver features
128 * the SQL client does not wrap, such as `exec()` with raw output formats.
129 * Requests made on it bypass `withQueryId`, `withClickhouseSettings` and
130 * interruption handling; pass `query_id`, `clickhouse_settings` and
131 * `abort_signal` directly. It is closed with the client's scope.
132 */
133 readonly client: Clickhouse.ClickHouseClient
134 readonly param: (dataType: string, value: unknown) => Statement.Fragment
135 readonly asCommand: <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
136 readonly insertQuery: <T = unknown>(options: {
137 readonly table: string
138 readonly values: Clickhouse.InsertValues<Readable, T>
139 readonly format?: Clickhouse.DataFormat
140 readonly columns?: NonNullable<Clickhouse.InsertParams<Readable, T>["columns"]>
141 }) => Effect.Effect<Clickhouse.InsertResult, SqlError>
142 readonly withQueryId: {
143 (queryId: string): <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
144 <A, E, R>(effect: Effect.Effect<A, E, R>, queryId: string): Effect.Effect<A, E, R>
145 }
146 readonly withClickhouseSettings: {
147 (
148 settings: NonNullable<Clickhouse.BaseQueryParams["clickhouse_settings"]>
149 ): <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
150 <A, E, R>(
151 effect: Effect.Effect<A, E, R>,
152 settings: NonNullable<Clickhouse.BaseQueryParams["clickhouse_settings"]>
153 ): Effect.Effect<A, E, R>
154 }
155}
156
157/**
158 * Service tag for the active ClickHouse SQL client.
159 *
160 * **When to use**
161 *
162 * Use to access or provide a ClickHouse SQL client through the Effect context.
163 *
164 * @stability unstable
165 * @category services
166 * @since 4.0.0
167 */
168export const ClickhouseClient = Context.Service<ClickhouseClient>("@effect/sql-clickhouse/ClickhouseClient")
169
170/**
171 * Configuration for creating a ClickHouse client, combining
172 * `@clickhouse/client` options with optional span attributes and query/result
173 * name transforms.
174 *
175 * @stability unstable
176 * @category models
177 * @since 4.0.0
178 */
179export interface ClickhouseClientConfig extends Clickhouse.ClickHouseClientConfigOptions {
180 readonly spanAttributes?: Record<string, unknown> | undefined
181 readonly transformResultNames?: ((str: string) => string) | undefined
182 readonly transformQueryNames?: ((str: string) => string) | undefined
183}
184
185/**
186 * Creates a scoped `ClickhouseClient`, verifies connectivity with `ping()`,
187 * closes the underlying client when the scope ends, maps ClickHouse failures
188 * to `SqlError`, and aborts plus kills in-flight queries when interrupted.
189 *
190 * @stability unstable
191 * @category constructors
192 * @since 4.0.0
193 */
194export const make = (
195 options: ClickhouseClientConfig
196): Effect.Effect<ClickhouseClient, SqlError, Scope.Scope | Reactivity.Reactivity> =>
197 Effect.gen(function*() {
198 const compiler = makeCompiler(options.transformQueryNames)
199 const transformRows = options.transformResultNames
200 ? Statement.defaultTransforms(options.transformResultNames).array
201 : undefined
202
203 const client = yield* Effect.acquireRelease(
204 Effect.sync(() => Clickhouse.createClient(options)),
205 (client) => Effect.promise(() => client.close())
206 )
207
208 yield* Effect.tryPromise({
209 try: async () => {
210 const result = await client.ping()
211 if (!result.success) {
212 throw result.error
213 }
214 return result
215 },
216 catch: (cause) =>
217 new SqlError({ reason: classifyError(cause, "ClickhouseClient: Failed to connect", "connect", "connection") })
218 }).pipe(
219 Effect.timeoutOrElse({
220 duration: Duration.seconds(5),
221 orElse: () =>
222 Effect.fail(
223 new SqlError({
224 reason: new ConnectionError({
225 message: "ClickhouseClient: Connection timeout",
226 cause: new Error("connection timeout"),
227 operation: "connect"
228 })
229 })
230 )
231 })
232 )
233
234 class ConnectionImpl implements Connection {
235 private conn: Clickhouse.ClickHouseClient
236 constructor(conn: Clickhouse.ClickHouseClient) {
237 this.conn = conn
238 }
239
240 private runRaw(sql: string, params: ReadonlyArray<unknown>, format: Clickhouse.DataFormat = "JSON") {
241 const paramsObj: Record<string, unknown> = {}
242 for (let i = 0; i < params.length; i++) {
243 paramsObj[`p${i + 1}`] = params[i]
244 }
245 return Effect.withFiber<Clickhouse.ResultSet<"JSON"> | Clickhouse.CommandResult, SqlError>((fiber) => {
246 const method = fiber.getRef(ClientMethod)
247 return Effect.callback<Clickhouse.ResultSet<"JSON"> | Clickhouse.CommandResult, SqlError>((resume) => {
248 const queryId = fiber.getRef(QueryId) ?? Crypto.randomUUID()
249 const settings = fiber.getRef(ClickhouseSettings)
250 const controller = new AbortController()
251 if (method === "command") {
252 this.conn.command({
253 query: sql,
254 query_params: paramsObj,
255 abort_signal: controller.signal,
256 query_id: queryId,
257 clickhouse_settings: settings
258 }).then(
259 (result) => resume(Effect.succeed(result)),
260 (cause) =>
261 resume(
262 Effect.fail(
263 new SqlError({ reason: classifyError(cause, "Failed to execute statement", "execute") })
264 )
265 )
266 )
267 } else {
268 this.conn.query({
269 query: sql,
270 query_params: paramsObj,
271 abort_signal: controller.signal,
272 query_id: queryId,
273 clickhouse_settings: settings,
274 format
275 }).then(
276 (result) => resume(Effect.succeed(result)),
277 (cause) =>
278 resume(
279 Effect.fail(
280 new SqlError({ reason: classifyError(cause, "Failed to execute statement", "execute") })
281 )
282 )
283 )
284 }
285 return Effect.suspend(() => {
286 controller.abort()
287 return killQuery(this.conn, queryId)
288 })
289 })
290 })
291 }
292
293 private run(sql: string, params: ReadonlyArray<unknown>, format?: Clickhouse.DataFormat) {
294 return this.runRaw(sql, params, format).pipe(
295 Effect.flatMap((result) => {
296 if ("json" in result) {
297 return Effect.tryPromise({
298 try: () => result.json().then((result) => "data" in result ? result.data : result as any),
299 catch: (cause) =>
300 new SqlError({ reason: classifyError(cause, "Failed to parse result", "parseResult") })
301 })
302 }
303 return Effect.succeed([])
304 })
305 )
306 }
307
308 execute(
309 sql: string,
310 params: ReadonlyArray<unknown>,
311 transformRows: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
312 ) {
313 return transformRows
314 ? Effect.map(this.run(sql, params), transformRows)
315 : this.run(sql, params)
316 }
317 executeRaw(sql: string, params: ReadonlyArray<unknown>) {
318 return this.runRaw(sql, params)
319 }
320 executeValues(sql: string, params: ReadonlyArray<unknown>) {
321 return this.run(sql, params, "JSONCompact")
322 }
323 executeValuesUnprepared(sql: string, params: ReadonlyArray<unknown>) {
324 return this.executeValues(sql, params)
325 }
326 executeUnprepared(sql: string, params: ReadonlyArray<unknown>, transformRows?: any) {
327 return this.execute(sql, params, transformRows)
328 }
329 executeStream(
330 sql: string,
331 params: ReadonlyArray<unknown>,
332 transformRows: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
333 ) {
334 const conn = this.conn
335 return Effect.withFiber((fiber) => {
336 const queryId = fiber.getRef(QueryId) ?? Crypto.randomUUID()
337 return this.runRaw(sql, params, "JSONEachRow").pipe(
338 Effect.provideService(QueryId, queryId),
339 Effect.map((result) => {
340 if (!("stream" in result)) {
341 return Stream.empty
342 }
343 // Closing the socket does not stop a query the server is still
344 // computing, so kill it unless the response was read to the end.
345 let done = false
346 return NodeStream.fromReadable<ReadonlyArray<Clickhouse.Row<any, "JSONEachRow">>, SqlError>({
347 evaluate: () => result.stream() as any,
348 onError: (cause) => new SqlError({ reason: classifyError(cause, "Failed to execute stream", "stream") })
349 }).pipe(
350 Stream.onEnd(Effect.sync(() => {
351 done = true
352 })),
353 Stream.ensuring(Effect.suspend(() => done ? Effect.void : Effect.ignoreCause(killQuery(conn, queryId))))
354 )
355 })
356 )
357 }).pipe(
358 Stream.unwrap,
359 Stream.chunks,
360 Stream.mapEffect((chunk) => {
361 const promises: Array<Promise<any>> = []
362 for (const rows of chunk) {
363 for (const row of rows) {
364 promises.push(row.json())
365 }
366 }
367 return Effect.tryPromise({
368 try: () => Promise.all(promises).then((rows) => transformRows ? transformRows(rows) : rows),
369 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to parse row", "parseRow") })
370 })
371 }),
372 Stream.flattenIterable
373 )
374 }
375 }
376
377 const connection = new ConnectionImpl(client)
378
379 return Object.assign(
380 yield* Client.make({
381 driver: "clickhouse",
382 acquirer: Effect.succeed(connection),
383 compiler,
384 spanAttributes: [
385 ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
386 [ATTR_DB_SYSTEM_NAME, "clickhouse"],
387 [ATTR_DB_NAMESPACE, options.database ?? "default"]
388 ],
389 beginTransaction: "BEGIN TRANSACTION",
390 transformRows
391 }),
392 {
393 [TypeId]: TypeId as TypeId,
394 config: options,
395 client,
396 param(dataType: string, value: unknown) {
397 return Statement.fragment([clickhouseParam(dataType, value)])
398 },
399 asCommand<A, E, R>(effect: Effect.Effect<A, E, R>) {
400 return Effect.provideService(effect, ClientMethod, "command")
401 },
402 insertQuery<T = unknown>(options: {
403 readonly table: string
404 readonly values: Clickhouse.InsertValues<Readable, T>
405 readonly format?: Clickhouse.DataFormat
406 readonly columns?: NonNullable<Clickhouse.InsertParams<Readable, T>["columns"]>
407 }) {
408 return Effect.callback<Clickhouse.InsertResult, SqlError>((resume) => {
409 const fiber = Fiber.getCurrent()!
410 const queryId = fiber.getRef(QueryId) ?? Crypto.randomUUID()
411 const settings = fiber.getRef(ClickhouseSettings)
412 const controller = new AbortController()
413 client.insert({
414 format: "JSONEachRow",
415 ...options,
416 abort_signal: controller.signal,
417 query_id: queryId,
418 clickhouse_settings: settings
419 }).then(
420 (result) => resume(Effect.succeed(result)),
421 (cause) =>
422 resume(Effect.fail(new SqlError({ reason: classifyError(cause, "Failed to insert data", "insert") })))
423 )
424 return Effect.suspend(() => {
425 controller.abort()
426 return killQuery(client, queryId)
427 })
428 })
429 },
430 withQueryId: dual(
431 2,
432 <A, E, R>(effect: Effect.Effect<A, E, R>, queryId: string) => Effect.provideService(effect, QueryId, queryId)
433 ),
434 withClickhouseSettings: dual(
435 2,
436 <A, E, R>(
437 effect: Effect.Effect<A, E, R>,
438 settings: NonNullable<Clickhouse.BaseQueryParams["clickhouse_settings"]>
439 ) => Effect.provideService(effect, ClickhouseSettings, settings)
440 )
441 }
442 )
443 })
444
445/**
446 * Fiber reference read by the low-level ClickHouse connection to choose query
447 * or command execution for statements; defaults to `query`.
448 *
449 * @stability unstable
450 * @category services
451 * @since 4.0.0
452 */
453export const ClientMethod = Context.Reference<"query" | "command" | "insert">(
454 "@effect/sql-clickhouse/ClickhouseClient/ClientMethod",
455 {
456 defaultValue: () => "query"
457 }
458)
459
460/**
461 * Fiber reference for the ClickHouse `query_id` applied to queries and
462 * inserts; a random UUID is generated when no query ID is set.
463 *
464 * @stability unstable
465 * @category services
466 * @since 4.0.0
467 */
468export const QueryId = Context.Reference<string | undefined>(
469 "@effect/sql-clickhouse/ClickhouseClient/QueryId",
470 { defaultValue: () => undefined }
471)
472
473/**
474 * Fiber reference containing ClickHouse settings to attach to queries,
475 * commands, and inserts.
476 *
477 * @stability unstable
478 * @category services
479 * @since 4.0.0
480 */
481export const ClickhouseSettings: Context.Reference<
482 NonNullable<Clickhouse.BaseQueryParams["clickhouse_settings"]>
483> = Context.Reference("@effect/sql-clickhouse/ClickhouseClient/ClickhouseSettings", {
484 defaultValue: () => ({})
485})
486
487/**
488 * Provides both `ClickhouseClient` and generic `SqlClient` services from a
489 * `Config`-backed ClickHouse client configuration.
490 *
491 * @stability unstable
492 * @category layers
493 * @since 4.0.0
494 */
495export const layerConfig: (
496 config: Config.Wrap<ClickhouseClientConfig>
497) => Layer.Layer<ClickhouseClient | Client.SqlClient, Config.ConfigError | SqlError> = (
498 config: Config.Wrap<ClickhouseClientConfig>
499): Layer.Layer<ClickhouseClient | Client.SqlClient, Config.ConfigError | SqlError> =>
500 Layer.effectContext(
501 Config.unwrap(config).pipe(
502 Effect.flatMap(make),
503 Effect.map((client) =>
504 Context.make(ClickhouseClient, client).pipe(
505 Context.add(Client.SqlClient, client)
506 )
507 )
508 )
509 ).pipe(Layer.provide(Reactivity.layer))
510
511/**
512 * Provides both `ClickhouseClient` and generic `SqlClient` services from a
513 * ClickHouse client configuration.
514 *
515 * @stability unstable
516 * @category layers
517 * @since 4.0.0
518 */
519export const layer = (
520 config: ClickhouseClientConfig
521): Layer.Layer<ClickhouseClient | Client.SqlClient, Config.ConfigError | SqlError> =>
522 Layer.effectContext(
523 Effect.map(make(config), (client) =>
524 Context.make(ClickhouseClient, client).pipe(
525 Context.add(Client.SqlClient, client)
526 ))
527 ).pipe(Layer.provide(Reactivity.layer))
528
529const typeFromUnknown = (value: unknown): string => {
530 if (Statement.isFragment(value)) {
531 return typeFromUnknown(value.segments[0])
532 } else if (isClickhouseParam(value)) {
533 return value.paramA
534 } else if (Array.isArray(value)) {
535 return `Array(${typeFromUnknown(value[0])})`
536 }
537 switch (typeof value) {
538 case "number":
539 return "Float64"
540 case "bigint":
541 return "Int64"
542 case "boolean":
543 return "Bool"
544 case "object":
545 if (value instanceof Date) {
546 return "DateTime"
547 }
548 return "String"
549 default:
550 return "String"
551 }
552}
553
554/**
555 * Creates the SQL statement compiler for ClickHouse, emitting typed
556 * `{pN: Type}` placeholders and escaping identifiers with an optional query
557 * name transform.
558 *
559 * @stability unstable
560 * @category constructors
561 * @since 4.0.0
562 */
563export const makeCompiler = (transform?: (_: string) => string) =>
564 Statement.makeCompiler<ClickhouseCustom>({
565 dialect: "clickhouse",
566 placeholder(i, u) {
567 return `{p${i}: ${typeFromUnknown(u)}}`
568 },
569 onIdentifier: transform ?
570 function(value, withoutTransform) {
571 return withoutTransform ? escape(value) : escape(transform(value))
572 } :
573 escape,
574 onRecordUpdate() {
575 return ["", []]
576 },
577 onCustom(type, placeholder) {
578 return [placeholder(type), [type.paramB]]
579 }
580 })
581
582// compiler helpers
583
584const escape = Statement.defaultEscape("\"")
585
586/**
587 * Custom SQL fragment type used for ClickHouse typed parameters created by
588 * `ClickhouseClient.param`.
589 *
590 * @stability unstable
591 * @category models
592 * @since 4.0.0
593 */
594export type ClickhouseCustom = ClickhouseParam
595
596/**
597 * @category models
598 * @since 4.0.0
599 */
600interface ClickhouseParam extends Statement.Custom<"ClickhouseParam", string, unknown> {}
601
602const clickhouseParam = Statement.custom<ClickhouseParam>("ClickhouseParam")
603const isClickhouseParam = Statement.isCustom<ClickhouseParam>("ClickhouseParam")