From 8ee504e3f8a0337182305bdae96372b894614079 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Thu, 15 Oct 2026 14:37:21 +0200 Subject: [PATCH 1/3] feat: ClickHouse over fetch, and OpenTelemetry spans for every statement effect-orm compiled queries but left running them to Effect's @effect/sql-clickhouse, which wraps the Node client and cannot run in Cloudflare Workers; and the spans a query got were Effect's generic `sql.execute`, whose `db.operation.name` is the client method. - @maple-dev/effect-orm/clickhouse-http: a ClickHouse SqlClient over Effect's HttpClient. Interrupting a statement aborts its request and sends KILL QUERY for its query_id (a closed connection alone left the query running on the server); failures carry the SqlError reasons Effect's driver uses; credentials go in a header; asCommand, withQueryId and withSettings behave as there. - Database opens one client span per statement, per the OpenTelemetry database conventions: named after db.query.summary (`SELECT events`), with db.system.name, db.operation.name, db.collection.name, db.response.returned_rows and error.type. withQuerySummary names a group of statements. sql.execute nests inside with the query text. - CompiledQuery.collection: the FROM table of a SELECT, or the table a write changes; absent over a subquery, union or CTE. Tested offline against a fake HttpClient and a recording tracer on PGlite, and live against ClickHouse 26.6, including that an interrupted query leaves system.processes (it fails without the kill). Co-Authored-By: Claude Opus 5.5 --- .changeset/clickhouse-http-and-spans.md | 9 + README.md | 1 + docs/database.md | 25 ++ docs/reference.md | 2 + docs/running-queries.md | 52 +++- package.json | 7 +- src/ch/compile.ts | 22 +- src/clickhouse-http.ts | 14 ++ src/clickhouse-http/client.test.ts | 193 +++++++++++++++ src/clickhouse-http/client.ts | 296 +++++++++++++++++++++++ src/database.ts | 1 + src/database/database.ts | 30 ++- src/database/tracing.test.ts | 130 ++++++++++ src/database/tracing.ts | 82 +++++++ tests/clickhouse-http.clickhouse.test.ts | 91 +++++++ tsdown.config.ts | 1 + 16 files changed, 940 insertions(+), 16 deletions(-) create mode 100644 .changeset/clickhouse-http-and-spans.md create mode 100644 src/clickhouse-http.ts create mode 100644 src/clickhouse-http/client.test.ts create mode 100644 src/clickhouse-http/client.ts create mode 100644 src/database/tracing.test.ts create mode 100644 src/database/tracing.ts create mode 100644 tests/clickhouse-http.clickhouse.test.ts diff --git a/.changeset/clickhouse-http-and-spans.md b/.changeset/clickhouse-http-and-spans.md new file mode 100644 index 0000000..184802e --- /dev/null +++ b/.changeset/clickhouse-http-and-spans.md @@ -0,0 +1,9 @@ +--- +"@maple-dev/effect-orm": minor +--- + +A ClickHouse client for Cloudflare Workers and any `fetch` runtime, and OpenTelemetry-shaped spans for every statement. + +- `@maple-dev/effect-orm/clickhouse-http` is a ClickHouse `SqlClient` over Effect's `HttpClient`, so `Database` runs where `@effect/sql-clickhouse` (Node only) cannot. Interrupting a statement aborts its request and sends `KILL QUERY` for its `query_id`; failures carry the same `SqlError` reasons as Effect's driver; credentials travel in a header; `asCommand`, `withQueryId` and `withSettings` work as there. +- `Database` opens one client span per statement, named and attributed per the OpenTelemetry database conventions: `db.system.name`, `db.operation.name`, `db.collection.name`, `db.query.summary`, `db.response.returned_rows` and `error.type`. `withQuerySummary(name)` names the statements an effect runs. +- `CompiledQuery.collection` is the table a statement targets: a SELECT's `FROM` table, or the table a write changes. diff --git a/README.md b/README.md index 8d13f2e..20e0b25 100644 --- a/README.md +++ b/README.md @@ -228,6 +228,7 @@ regressions live in [`src/docs-examples.test.ts`](./src/docs-examples.test.ts). | `@maple-dev/effect-orm/schema` | Migration tooling over `table` values: DDL rendering, snapshots, the schema diff. Pure. | | `@maple-dev/effect-orm/kit`, `/migrate` | `effect-orm generate` and `check`; applying migrations through a driver you provide. See [Schema and migrations](./docs/migrations.md). | | `@maple-dev/effect-orm/database` | `Database` over your `SqlClient`: `run`, `execute`, `transaction` with retry. | +| `@maple-dev/effect-orm/clickhouse-http` | A ClickHouse `SqlClient` over `fetch`, for Cloudflare Workers and any runtime with `fetch`; give it to `Database`. | ## Extending with custom functions diff --git a/docs/database.md b/docs/database.md index a98b84f..2e0cccd 100644 --- a/docs/database.md +++ b/docs/database.md @@ -259,6 +259,31 @@ fails at BEGIN today: through the query path with a syntax error, and through `a `NOT_IMPLEMENTED`. The full measurements are in [`design/transactions.md`](../design/transactions.md). +## Tracing + +Each statement `Database` runs is one client span, named and attributed the way the +[OpenTelemetry database conventions](https://opentelemetry.io/docs/specs/semconv/database/database-spans/) +describe. A compiled query knows its operation and table, so the span needs no SQL parsing: + +| Attribute | Value | +| --------------------------- | -------------------------------------------------------------------- | +| `db.system.name` | `clickhouse`, or `postgresql` | +| `db.operation.name` | `SELECT`, `INSERT`, `UPDATE` or `DELETE` | +| `db.collection.name` | The `FROM` or written table; absent over a subquery, union or CTE | +| `db.query.summary` | `SELECT events`, or the name given with `withQuerySummary` | +| `db.response.returned_rows` | The rows it returned | +| `error.type` | On failure: the `DatabaseError` reason, such as `SqlSyntaxError` | + +The span is named after the summary, so a trace reads `SELECT events` rather than `sql.execute`. +Effect's own `sql.execute` span runs inside it and carries the query text. To give a group of +statements a stable name of your own, wrap them: + +```ts +Db.run(query, params).pipe(Db.withQuerySummary("dashboard.revenue_by_day")) +``` + +The table is also on the compiled query, as `compiled.collection`. + ## Writes The builder compiles SELECTs, [INSERTs](./inserts.md) and diff --git a/docs/reference.md b/docs/reference.md index ae09e57..89c2faf 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -513,6 +513,8 @@ placeholders, double-quoted identifiers. | `run`, `status`, `verify`, `baseline`, `MigrationDriver` and the migrate errors | `/migrate` | | `Database`, `run`, `sql`, `query`, `execute`, `transaction`, `requireTransaction`, `retryContention`, `Transaction` | `/database` | | `DatabaseError`, `TransactionCommitFailed`, `TransactionRollbackFailed` and the other transaction errors | `/database` | +| `withQuerySummary`, `QuerySummary` — the `db.query.summary` of the statements an effect runs, for spans | `/database` | +| `ClickhouseHttpClient`, `layer`, `make`, `ClickhouseHttpClientConfig` — a ClickHouse `SqlClient` over `fetch`, for Workers | `/clickhouse-http` | | `defineConfig`, `generate`, `check`, `loadSchema`, `readMigrations`, `analyze`, `KitError` | `/kit` | | `defineSuite`, `query`, `caseFromCompiled`, `runSuite`, `compareRuns`, `compareBudgets` | `/benchmark` | | `Suite`, `RunOutput`, `BenchmarkError` and benchmark contracts | `/benchmark` | diff --git a/docs/running-queries.md b/docs/running-queries.md index 0b020de..756f86a 100644 --- a/docs/running-queries.md +++ b/docs/running-queries.md @@ -60,11 +60,57 @@ SQL failures as `SqlError`. not validate or escape arbitrary SQL; keep untrusted values in `CH.param` bindings. Leave result name transforms disabled so aliases still match the compiled decoder. -`@effect/sql-clickhouse` wraps the official Node.js client. For Workers or other fetch-based -runtimes, use a runtime-compatible adapter such as `@clickhouse/client-web` with an Effect -integration. Keep database credentials on your server. +`@effect/sql-clickhouse` wraps the official Node.js client, so it does not run in Cloudflare +Workers, Deno or a browser. For those, and anywhere `fetch` exists, use +[`@maple-dev/effect-orm/clickhouse-http`](#on-workers-and-other-fetch-runtimes) below. Keep +database credentials on your server. See the [Effect ClickHouse driver source](https://github.com/Effect-TS/effect/blob/main/packages/sql/clickhouse/src/ClickhouseClient.ts). +### On Workers and other fetch runtimes + +`@maple-dev/effect-orm/clickhouse-http` is a ClickHouse `SqlClient` over Effect's `HttpClient`. +Provide `FetchHttpClient.layer` and it runs wherever `fetch` does; give it to `Database` like +any other client: + +```ts +import { Effect, Layer, Redacted } from "effect" +import { FetchHttpClient } from "effect/http" +import * as SqlClient from "effect/sql/SqlClient" +import * as CH from "@maple-dev/effect-orm/clickhouse" +import * as ClickhouseHttp from "@maple-dev/effect-orm/clickhouse-http" +import * as Db from "@maple-dev/effect-orm/database" + +const ClickHouse = ClickhouseHttp.layer({ + url: "https://abc.clickhouse.cloud:8443", + username: "reader", + password: Redacted.make("…"), + database: "analytics", +}).pipe(Layer.provide(FetchHttpClient.layer)) + +const DatabaseLive = Layer.effect( + Db.Database, + Effect.gen(function* () { + return Db.fromSqlClient(yield* SqlClient.SqlClient, { dialect: CH.clickhouseDialect }) + }), +).pipe(Layer.provideMerge(ClickHouse)) +``` + +What it does for you: + +- **Interrupting cancels the query.** A timeout or an interrupted fiber aborts the request and + sends `KILL QUERY` for its `query_id`: a closed connection alone does not stop a query + ClickHouse is still computing. Queries also send `cancel_http_readonly_queries_on_client_close`. +- **Failures carry Effect's `SqlError` reasons** (`SqlSyntaxError`, `AuthenticationError`, + `StatementTimeoutError`, …), read from ClickHouse's exception code, the same as + `@effect/sql-clickhouse`. +- **Credentials travel in a header,** never the URL; the URL may not contain them. +- `asCommand` runs DDL and writes, `withQueryId` sets the `query_id`, and `withSettings` adds + ClickHouse settings over the configured ones. Statements bind `sql` template values on the + server, as `{pN: Type}` placeholders. + +It opens no connection at startup, so a Worker's first request pays nothing extra. Its spans +are `Database`'s, described in [Running statements: tracing](./database.md#tracing). + For a client that implements Effect's `SqlClient`, the opt-in [`@maple-dev/effect-orm/database`](./database.md) entry point does this loop for you (`Db.run(compiled)`), and adds transactions on Postgres. diff --git a/package.json b/package.json index 59d405c..da976a3 100644 --- a/package.json +++ b/package.json @@ -38,6 +38,11 @@ "import": "./dist/clickhouse.mjs", "default": "./dist/clickhouse.mjs" }, + "./clickhouse-http": { + "types": "./dist/clickhouse-http.d.mts", + "import": "./dist/clickhouse-http.mjs", + "default": "./dist/clickhouse-http.mjs" + }, "./expr": { "types": "./dist/expr.d.mts", "import": "./dist/expr.mjs", @@ -132,4 +137,4 @@ "patchedDependencies": { "@effect/vitest@4.0.0": "patches/@effect%2Fvitest@4.0.0.patch" } -} \ No newline at end of file +} diff --git a/src/ch/compile.ts b/src/ch/compile.ts index 5709864..add77b4 100644 --- a/src/ch/compile.ts +++ b/src/ch/compile.ts @@ -270,6 +270,16 @@ interface CompiledQueryBase { * meant: the SQL may even run, with ClickHouse quoting and inlined params. */ readonly dialect?: string + /** + * The table the statement targets: a SELECT's `FROM` table, or the table an + * INSERT, UPDATE or DELETE writes. Absent when there is no single table, as + * for a `FROM` over a subquery, a union or a CTE, and for handwritten SQL. + * + * What a tracer reports as `db.collection.name`, and with `kind` the + * statement's `db.query.summary` (`SELECT events`), so a span names its + * table without parsing SQL. + */ + readonly collection?: string /** Runtime decode of raw query results. Queries built from handwritten SQL * should provide a row schema so schema drift is caught before consumers * read fields from `Record`. Without a schema this is an @@ -413,6 +423,7 @@ const makeCompiledQuery = ( kind: CompiledQueryBase["kind"] = "select", returning?: ReadonlyArray, crossTenantDeclared = false, + collection?: string, ): CompiledQuery => { let cachedDecodeRow: ((row: unknown) => Effect.Effect) | undefined let decoderBuilt = false @@ -489,6 +500,7 @@ const makeCompiledQuery = ( rowSchemaMismatch, ...(rawSql !== undefined ? { rawSql } : undefined), ...(dialect !== undefined ? { dialect } : undefined), + ...(collection !== undefined ? { collection } : undefined), ...(!(route === undefined) ? { route } : undefined), decodeRows, encodeRows, @@ -997,6 +1009,10 @@ function compileInner< "select", undefined, state.crossTenant === true, + // A table, not a CTE of the same name and not a subquery or union. + state.fromQuery || state.fromUnion || visibleCtes.some((c) => c.name === state.tableName) + ? undefined + : state.tableName, ), tenantScope === "single-tenant" ? scope.bound : undefined, ) @@ -1697,6 +1713,7 @@ const writeCompiledQuery = ( tenantBound: string | undefined, returning: ReturnType, crossTenant: boolean | undefined, + collection: string, ): CompiledQuery => { const returnedSchema = returning !== undefined && "schema" in returning.derived ? returning.derived.schema : undefined return withTenantBound( @@ -1714,6 +1731,7 @@ const writeCompiledQuery = ( kind, returning?.aliases, crossTenant === true, + collection, ), crossTenant === true ? undefined : tenantBound, ) @@ -1948,7 +1966,7 @@ function compileInsert(insert: CHInsert, params: Record +} + +type Reply = { readonly status?: number; readonly body: string; readonly headers?: Record } + +/** A ClickHouse HTTP client over a fake HttpClient that records each request. */ +const fake = (reply: (sent: Sent) => Reply, config: Partial = {}) => { + const sent: Array = [] + const http = HttpClient.make((request, url) => { + const body = request.body._tag === "Uint8Array" ? (request.body.text ?? new TextDecoder().decode(request.body.body)) : "" + const recorded = { url, body, headers: { ...request.headers } } + sent.push(recorded) + const { status = 200, body: text, headers } = reply(recorded) + return Effect.succeed(HttpClientResponse.fromWeb(request, new Response(text, { status, headers }))) + }) + const live = layer({ + url: "https://clickhouse.example:8443", + username: "reader", + password: Redacted.make("s3cret"), + database: "analytics", + settings: { max_execution_time: 30 }, + ...config, + }).pipe(Layer.provide(Layer.succeed(HttpClient.HttpClient, http))) + return { sent, live } +} + +const json = (data: ReadonlyArray): Reply => ({ body: JSON.stringify({ meta: [], data, rows: data.length }) }) + +describe("ClickhouseHttpClient", () => { + it.effect("POSTs the SQL with its format, credentials as a header, and settings as URL parameters", () => { + const { sent, live } = fake(() => json([{ x: 1 }])) + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + const rows = yield* sql.unsafe("SELECT 1 AS x") + expect(rows).toEqual([{ x: 1 }]) + const [request] = sent + assert(request !== undefined) + expect(request.body).toBe("SELECT 1 AS x\nFORMAT JSON") + expect(request.headers.authorization).toBe(`Basic ${btoa("reader:s3cret")}`) + expect(request.url.origin).toBe("https://clickhouse.example:8443") + expect(request.url.username).toBe("") + expect(request.url.searchParams.get("database")).toBe("analytics") + expect(request.url.searchParams.get("max_execution_time")).toBe("30") + // An interrupted query stops on the server, not only in the client. + expect(request.url.searchParams.get("cancel_http_readonly_queries_on_client_close")).toBe("1") + expect(request.url.searchParams.get("query_id")).toMatch(/^[0-9a-f-]{36}$/) + }).pipe(Effect.provide(live)) + }) + + it.effect("binds parameters server-side, typed, as param_pN", () => { + const { sent, live } = fake(() => json([])) + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + yield* sql`SELECT ${"o'brien"} AS name, ${42} AS n` + const [request] = sent + assert(request !== undefined) + expect(request.body).toBe("SELECT {p1: String} AS name, {p2: Float64} AS n\nFORMAT JSON") + expect(request.url.searchParams.get("param_p1")).toBe("o'brien") + expect(request.url.searchParams.get("param_p2")).toBe("42") + }).pipe(Effect.provide(live)) + }) + + it.effect("sends a command without a format, and returns no rows", () => { + const { sent, live } = fake(() => ({ body: "" })) + return Effect.gen(function* () { + const client = yield* ClickhouseHttpClient + const rows = yield* client.asCommand(client.unsafe("CREATE TABLE t (x UInt8) ENGINE = Memory")) + expect(rows).toEqual([]) + const [request] = sent + assert(request !== undefined) + expect(request.body).toBe("CREATE TABLE t (x UInt8) ENGINE = Memory") + expect(request.url.searchParams.has("cancel_http_readonly_queries_on_client_close")).toBe(false) + }).pipe(Effect.provide(live)) + }) + + it.effect("applies a query id and per-call settings over the configured ones", () => { + const { sent, live } = fake(() => json([])) + return Effect.gen(function* () { + const client = yield* ClickhouseHttpClient + yield* client.unsafe("SELECT 1").pipe(client.withSettings({ max_execution_time: 5 }), client.withQueryId("q-1")) + const [request] = sent + assert(request !== undefined) + expect(request.url.searchParams.get("query_id")).toBe("q-1") + expect(request.url.searchParams.get("max_execution_time")).toBe("5") + }).pipe(Effect.provide(live)) + }) + + it.effect.each([ + [62, "SqlSyntaxError"], + [516, "AuthenticationError"], + [497, "AuthorizationError"], + [159, "StatementTimeoutError"], + [241, "UnknownError"], + ] as const)("classifies ClickHouse code %s as %s", ([code, reason]) => { + const { live } = fake(() => ({ + status: 500, + body: `Code: ${code}. DB::Exception: fixture failure. (FIXTURE)`, + headers: { "x-clickhouse-exception-code": String(code) }, + })) + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + const exit = yield* Effect.exit(sql.unsafe("SELECT 1")) + assert(Exit.isFailure(exit)) + const error = exit.cause.reasons.find((r) => r._tag === "Fail") + assert(error?._tag === "Fail" && isSqlError(error.error)) + expect(error.error.reason._tag).toBe(reason) + expect(error.error.message).toContain("fixture failure") + }).pipe(Effect.provide(live)) + }) + + it.effect("reads the code from the body when the header is missing", () => { + const { live } = fake(() => ({ status: 400, body: "Code: 62. DB::Exception: Syntax error" })) + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + const error = yield* Effect.flip(sql.unsafe("SELEKT 1")) + expect(error.reason._tag).toBe("SqlSyntaxError") + }).pipe(Effect.provide(live)) + }) + + it.effect("streams JSONEachRow rows", () => { + const { sent, live } = fake(() => ({ body: '{"n":1}\n{"n":2}\n{"n":3}\n' })) + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + const rows = yield* Stream.runCollect(sql.unsafe("SELECT number AS n FROM numbers(3)").stream) + expect(rows).toEqual([{ n: 1 }, { n: 2 }, { n: 3 }]) + expect(sent[0]?.body).toBe("SELECT number AS n FROM numbers(3)\nFORMAT JSONEachRow") + }).pipe(Effect.provide(live)) + }) + + it.effect("runs effect-orm queries through Database, decoding 64-bit integers", () => { + const { sent, live } = fake(() => json([{ id: "18446744073709551615", n: "7" }])) + const Events = CH.table("events", { external: true, columns: { Id: CH.uint64, OrgId: CH.string } }) + const query = CH.from(Events) + .select(($) => ({ id: CH.toString($.Id), n: CH.count() })) + .where(($) => [$.OrgId.eq(CH.param.string("org"))]) + .groupBy("id") + return Effect.gen(function* () { + const rows = yield* Db.run(query, { org: "o'1" }) + expect(rows).toEqual([{ id: "18446744073709551615", n: 7 }]) + expect(sent[0]?.body).toContain("events.OrgId = 'o\\'1'") + }).pipe( + Effect.provide(Layer.effect(Db.Database, Effect.gen(function* () { + return Db.fromSqlClient(yield* SqlClient.SqlClient, { dialect: CH.clickhouseDialect }) + })).pipe(Layer.provideMerge(live))), + ) + }) + + it.effect("kills the query by id when its fiber is interrupted", () => { + const sent: Array = [] + const http = HttpClient.make((request, url) => { + const body = request.body._tag === "Uint8Array" ? (request.body.text ?? "") : "" + sent.push({ url, body, headers: { ...request.headers } }) + // The query never answers; the kill does. + return body.startsWith("KILL") + ? Effect.succeed(HttpClientResponse.fromWeb(request, new Response(""))) + : Effect.never + }) + const live = layer({ url: "https://clickhouse.example" }).pipe(Layer.provide(Layer.succeed(HttpClient.HttpClient, http))) + return Effect.gen(function* () { + const client = yield* ClickhouseHttpClient + const fiber = yield* Effect.forkChild(client.unsafe("SELECT sleep(3)").pipe(client.withQueryId("slow-1"))) + yield* Effect.yieldNow + yield* Fiber.interrupt(fiber) + const kill = sent.find((request) => request.body.startsWith("KILL")) + assert(kill !== undefined, "no KILL QUERY was sent") + expect(kill.body).toBe("KILL QUERY WHERE query_id = {id:String} ASYNC") + expect(kill.url.searchParams.get("param_id")).toBe("slow-1") + }).pipe(Effect.provide(live)) + }) + + it("refuses credentials, a query string or a fragment in the url", () => { + for (const url of ["https://u:p@clickhouse.example", "https://clickhouse.example/?x=1", "https://clickhouse.example/#f"]) { + const { live } = fake(() => json([]), { url }) + expect(() => Effect.runSync(Effect.scoped(Layer.build(live)))).toThrow(/url/) + } + }) +}) diff --git a/src/clickhouse-http/client.ts b/src/clickhouse-http/client.ts new file mode 100644 index 0000000..48400ab --- /dev/null +++ b/src/clickhouse-http/client.ts @@ -0,0 +1,296 @@ +// A ClickHouse `SqlClient` over Effect's `HttpClient`. +// +// Effect's own driver, `@effect/sql-clickhouse`, wraps the Node client, so it +// cannot run in Cloudflare Workers, Deno or a browser. This one speaks +// ClickHouse's HTTP interface through whatever `HttpClient` the caller provides +// (`FetchHttpClient.layer` anywhere `fetch` exists), and keeps that driver's +// shape: the same error reasons, `asCommand`, query ids and settings, so +// `Database` and the migration runner treat the two alike. +import { Context, Effect, Layer, Redacted, Stream } from "effect" +import { dual } from "effect/Function" +import * as Base64 from "effect/encoding/Base64" +import * as HttpClient from "effect/http/HttpClient" +import * as HttpClientRequest from "effect/http/HttpClientRequest" +import type * as HttpClientResponse from "effect/http/HttpClientResponse" +import * as Reactivity from "effect/reactivity/Reactivity" +import type * as Scope from "effect/Scope" +import * as Client from "effect/sql/SqlClient" +import type { Connection } from "effect/sql/SqlConnection" +import { + AuthenticationError, + AuthorizationError, + ConnectionError, + SqlError, + SqlSyntaxError, + StatementTimeoutError, + UnknownError, +} from "effect/sql/SqlError" +import * as Statement from "effect/sql/Statement" + +/** A ClickHouse setting value as the HTTP interface takes it. */ +export type ClickhouseSetting = string | number | boolean + +export interface ClickhouseHttpClientConfig { + /** The server, e.g. `https://abc.clickhouse.cloud:8443`. No credentials, path, query or fragment. */ + readonly url: string + readonly username?: string | undefined + readonly password?: Redacted.Redacted | string | undefined + /** The database unqualified names resolve in; `db.namespace` on spans. Defaults to `default`. */ + readonly database?: string | undefined + /** + * Settings sent with every statement, under the per-call ones + * (`withSettings`). Queries already send + * `cancel_http_readonly_queries_on_client_close=1`, so an interrupted fiber + * stops its query on the server as well as the request. + */ + readonly settings?: Readonly> | undefined + readonly spanAttributes?: Readonly> | undefined +} + +/** The ClickHouse HTTP client: a `SqlClient`, plus what statements over HTTP need. */ +export interface ClickhouseHttpClient extends Client.SqlClient { + readonly config: ClickhouseHttpClientConfig + /** Run the statements in `effect` as commands: DDL and writes, which send back no rows. */ + readonly asCommand: (effect: Effect.Effect) => Effect.Effect + /** The `query_id` of the statements in `effect`. Each statement gets a random one otherwise. */ + readonly withQueryId: { + (queryId: string): (effect: Effect.Effect) => Effect.Effect + (effect: Effect.Effect, queryId: string): Effect.Effect + } + /** ClickHouse settings for the statements in `effect`, over the configured ones. */ + readonly withSettings: { + (settings: Readonly>): (effect: Effect.Effect) => Effect.Effect + (effect: Effect.Effect, settings: Readonly>): Effect.Effect + } +} + +export const ClickhouseHttpClient = Context.Service("@maple-dev/effect-orm/ClickhouseHttpClient") + +const ClientMethod = Context.Reference<"query" | "command">("@maple-dev/effect-orm/ClickhouseHttpClient/ClientMethod", { + defaultValue: () => "query", +}) +const QueryId = Context.Reference("@maple-dev/effect-orm/ClickhouseHttpClient/QueryId", { + defaultValue: () => undefined, +}) +const Settings = Context.Reference>>( + "@maple-dev/effect-orm/ClickhouseHttpClient/Settings", + { defaultValue: () => ({}) }, +) + +// The codes Effect's Node driver classifies, so a failure means the same thing +// whichever client ran it. +const SYNTAX_ERROR_CODES = new Set([36, 60, 62, 242]) + +const classify = (code: number | undefined, message: string, cause: unknown, operation: string) => { + const props = { cause, message, operation } + if (code === 516) return new AuthenticationError(props) + if (code === 497) return new AuthorizationError(props) + if (code !== undefined && SYNTAX_ERROR_CODES.has(code)) return new SqlSyntaxError(props) + if (code === 159 || code === 469) return new StatementTimeoutError(props) + return new UnknownError(props) +} + +/** `Code: 62. DB::Exception: Syntax error …` when the header is missing. */ +const codeFromBody = (body: string): number | undefined => { + const match = /^Code:\s*(\d+)/.exec(body.trim()) + return match ? Number(match[1]) : undefined +} + +const validateUrl = (raw: string): URL => { + const url = new URL(raw) + if (url.username !== "" || url.password !== "") throw new Error("ClickhouseHttpClient: pass credentials as username and password, not in the url") + if (url.search !== "" || url.hash !== "") throw new Error("ClickhouseHttpClient: the url takes no query string or fragment") + return url +} + +const typeFromUnknown = (value: unknown): string => { + if (Array.isArray(value)) return `Array(${typeFromUnknown(value[0])})` + switch (typeof value) { + case "number": + return "Float64" + case "bigint": + return "Int64" + case "boolean": + return "Bool" + case "object": + return value instanceof Date ? "DateTime64(3)" : "String" + default: + return "String" + } +} + +/** `{pN: Type}` placeholders, sent as `param_pN`: ClickHouse's server-side binding. */ +const compiler = Statement.makeCompiler({ + dialect: "clickhouse", + placeholder: (i, value) => `{p${i}: ${typeFromUnknown(value)}}`, + onIdentifier: Statement.defaultEscape('"'), + onRecordUpdate: () => ["", []], + onCustom: () => ["", []], +}) + +const paramValue = (value: unknown): string => + value instanceof Date ? value.toISOString().replace("T", " ").replace("Z", "") : Array.isArray(value) ? JSON.stringify(value) : String(value) + +export const make = ( + config: ClickhouseHttpClientConfig, +): Effect.Effect => + Effect.gen(function* () { + const url = validateUrl(config.url) + const database = config.database ?? "default" + const password = config.password === undefined ? "" : Redacted.isRedacted(config.password) ? Redacted.value(config.password) : config.password + const authorization = + config.username === undefined ? undefined : `Basic ${Base64.encode(`${config.username}:${password}`)}` + const http = yield* HttpClient.HttpClient + + const post = (urlParams: Record, body: string) => { + const request = HttpClientRequest.post(url).pipe( + HttpClientRequest.setUrlParams(urlParams), + HttpClientRequest.bodyText(body, "text/plain; charset=utf-8"), + ) + return authorization === undefined ? request : HttpClientRequest.setHeader(request, "authorization", authorization) + } + // The statement span is the record of the request: an HTTP span would add + // nothing, and its URL carries bound parameter values. + const untraced = (effect: Effect.Effect) => + Effect.provideService(effect, HttpClient.TracerDisabledWhen, () => true) + // Best effort, and bounded: a kill that fails leaves the query to finish or + // to time out on its own, which is what would happen without it. + const kill = (queryId: string) => + untraced( + http.execute(post({ param_id: queryId }, "KILL QUERY WHERE query_id = {id:String} ASYNC")), + ).pipe(Effect.scoped, Effect.timeout("5 seconds"), Effect.ignore) + + const send = ( + sql: string, + params: ReadonlyArray, + format: "JSON" | "JSONCompact" | "JSONEachRow" | undefined, + operation: string, + ): Effect.Effect => + Effect.withFiber((fiber) => { + const command = fiber.getRef(ClientMethod) === "command" + const queryId = fiber.getRef(QueryId) ?? crypto.randomUUID() + const urlParams: Record = { database, query_id: queryId } + if (!command) urlParams.cancel_http_readonly_queries_on_client_close = "1" + for (const [key, value] of Object.entries({ ...config.settings, ...fiber.getRef(Settings) })) { + urlParams[key] = String(value) + } + params.forEach((value, i) => { + urlParams[`param_p${i + 1}`] = paramValue(value) + }) + const request = post(urlParams, command || format === undefined ? sql : `${sql}\nFORMAT ${format}`) + return untraced(http.execute(request)).pipe( + // Aborting the request does not stop the query: ClickHouse notices a + // closed connection only when it next writes to it, which a query still + // computing its first block may not do for a long time. So kill it by id, + // as Effect's Node driver does. + Effect.onInterrupt(() => kill(queryId)), + Effect.mapError( + (cause) => + new SqlError({ + reason: new ConnectionError({ cause, message: "ClickhouseHttpClient: the request failed", operation }), + }), + ), + Effect.flatMap((response) => + response.status === 200 + ? Effect.succeed(response) + : Effect.flatMap(Effect.orElseSucceed(response.text, () => ""), (body) => { + const header = response.headers["x-clickhouse-exception-code"] + const code = header === undefined ? codeFromBody(body) : Number(header) + const message = body.trim().slice(0, 1000) || `ClickHouse answered ${response.status}` + return Effect.fail(new SqlError({ reason: classify(code, message, { status: response.status, code }, operation) })) + }), + ), + ) + }) + + const parseFailure = (cause: unknown, operation: string) => + new SqlError({ reason: new UnknownError({ cause, message: "ClickhouseHttpClient: the response is not valid JSON", operation }) }) + + const json = (sql: string, params: ReadonlyArray, format: "JSON" | "JSONCompact") => + Effect.scoped( + Effect.withFiber((fiber) => + fiber.getRef(ClientMethod) === "command" + ? Effect.as(send(sql, params, undefined, "execute"), [] as ReadonlyArray) + : send(sql, params, format, "execute").pipe( + Effect.flatMap((response) => + Effect.mapError(response.text, (cause) => parseFailure(cause, "execute")), + ), + Effect.flatMap((text) => + Effect.try({ + try: () => (JSON.parse(text) as { readonly data: ReadonlyArray }).data, + catch: (cause) => parseFailure(cause, "execute"), + }), + ), + ), + ), + ) + + const connection: Connection = { + execute: (sql, params, transformRows) => + transformRows ? Effect.map(json(sql, params, "JSON"), transformRows) : json(sql, params, "JSON"), + executeRaw: (sql, params) => json(sql, params, "JSON"), + executeValues: (sql, params) => json(sql, params, "JSONCompact"), + executeValuesUnprepared: (sql, params) => json(sql, params, "JSONCompact"), + executeUnprepared: (sql, params, transformRows) => + transformRows ? Effect.map(json(sql, params, "JSON"), transformRows) : json(sql, params, "JSON"), + executeStream: (sql, params, transformRows) => + send(sql, params, "JSONEachRow", "stream").pipe( + Effect.map((response) => + response.stream.pipe( + Stream.mapError( + (cause) => + new SqlError({ + reason: new ConnectionError({ cause, message: "ClickhouseHttpClient: the stream failed", operation: "stream" }), + }), + ), + Stream.decodeText(), + Stream.splitLines, + Stream.filter((line) => line.length > 0), + Stream.mapEffect((line) => + Effect.try({ try: () => JSON.parse(line) as object, catch: (cause) => parseFailure(cause, "stream") }), + ), + Stream.map((row) => (transformRows ? transformRows([row])[0] : row)), + ), + ), + Stream.unwrap, + ), + } + + const client = yield* Client.make({ + acquirer: Effect.succeed(connection), + compiler, + spanAttributes: [ + ...Object.entries(config.spanAttributes ?? {}), + ["db.system.name", "clickhouse"], + ["db.namespace", database], + ], + // ClickHouse has no transactions over HTTP without a session; effect-orm's + // ClickHouse dialect refuses them before this is ever sent. + beginTransaction: "BEGIN TRANSACTION", + }) + + return Object.assign(client, { + config, + asCommand: (effect: Effect.Effect) => Effect.provideService(effect, ClientMethod, "command"), + withQueryId: dual(2, (effect: Effect.Effect, queryId: string) => + Effect.provideService(effect, QueryId, queryId), + ), + withSettings: dual(2, (effect: Effect.Effect, settings: Readonly>) => + Effect.provideService(effect, Settings, settings), + ), + }) as ClickhouseHttpClient + }) + +/** + * `ClickhouseHttpClient` and `SqlClient` from `config`. Provide an + * `HttpClient`: `FetchHttpClient.layer` from `effect/http` runs anywhere + * `fetch` does, Cloudflare Workers included. + */ +export const layer = ( + config: ClickhouseHttpClientConfig, +): Layer.Layer => + Layer.effectContext( + Effect.map(make(config), (client) => + Context.make(ClickhouseHttpClient, client).pipe(Context.add(Client.SqlClient, client)), + ), + ).pipe(Layer.provide(Reactivity.layer)) diff --git a/src/database.ts b/src/database.ts index d01d8ec..ba66164 100644 --- a/src/database.ts +++ b/src/database.ts @@ -38,5 +38,6 @@ export { TransactionUnsupported, type TransactionError, } from "./database/errors" +export { QuerySummary, withQuerySummary } from "./database/tracing" export { sql, type SqlIdentifier, type SqlRawText, type SqlTemplate } from "./database/sql" export type { DialectTransactions, IsolationLevel, TransactionSettings } from "./ch/dialect" diff --git a/src/database/database.ts b/src/database/database.ts index 8e2dd76..aaf89d4 100644 --- a/src/database/database.ts +++ b/src/database/database.ts @@ -31,6 +31,7 @@ import { } from "./errors" import { isSqlTemplate, renderTemplate, type SqlTemplate } from "./sql" import { firstLine, reasonOf, sqlStateOf, toDatabaseError } from "./sql-error" +import { traceStatement, type TracedStatement } from "./tracing" /** SQL and the values bound to its placeholders. A `CompiledQuery` is one. */ export interface Statement { @@ -243,11 +244,16 @@ export const fromSqlClient = (sql: SqlClient.SqlClient, options: FromSqlClientOp const resolve = (statement: StatementInput): Effect.Effect => isSqlTemplate(statement) ? renderTemplate(statement, dialect) : checkScope(statement) - const rows = (statement: Statement) => - guard(statement).pipe( - Effect.andThen(raw.unsafe>(statement.sql, statement.parameters ?? [])), - Effect.map((rows): ReadonlyArray> => rows), - Effect.mapError(toDatabaseError(statement.sql)), + const rows = (statement: Statement & TracedStatement) => + traceStatement( + dialect, + statement, + guard(statement).pipe( + Effect.andThen(raw.unsafe>(statement.sql, statement.parameters ?? [])), + Effect.map((rows): ReadonlyArray> => rows), + Effect.mapError(toDatabaseError(statement.sql)), + ), + (rows) => rows.length, ) const decodeWith = @@ -271,11 +277,15 @@ export const fromSqlClient = (sql: SqlClient.SqlClient, options: FromSqlClientOp )) as DatabaseApi["query"] const execute: DatabaseApi["execute"] = (statement) => - Effect.flatMap(resolve(statement), (resolved) => - guard(resolved).pipe( - Effect.andThen(command(raw.unsafe(resolved.sql, resolved.parameters ?? []))), - Effect.asVoid, - Effect.mapError(toDatabaseError(resolved.sql)), + Effect.flatMap(resolve(statement), (resolved: Statement & TracedStatement) => + traceStatement( + dialect, + resolved, + guard(resolved).pipe( + Effect.andThen(command(raw.unsafe(resolved.sql, resolved.parameters ?? []))), + Effect.asVoid, + Effect.mapError(toDatabaseError(resolved.sql)), + ), ), ) diff --git a/src/database/tracing.test.ts b/src/database/tracing.test.ts new file mode 100644 index 0000000..9bc2fa3 --- /dev/null +++ b/src/database/tracing.test.ts @@ -0,0 +1,130 @@ +// The client span `Database` opens per statement, checked on PGlite with a +// tracer that records spans, against the OpenTelemetry database conventions. +import { PgliteClient } from "@effect/sql-pglite" +import { assert, describe, expect, it, layer } from "@effect/vitest" +import { Effect, Exit, Layer, Option, Tracer } from "effect" +import * as SqlClient from "effect/sql/SqlClient" +import * as PG from "../postgres" +import { postgresDialect } from "../pg/dialect" +import * as Db from "../database" + +const spans: Array = [] +const recording = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan({ ...options, sampled: true }) + spans.push(span) + return span + }, +}) + +const Live = Layer.effect( + Db.Database, + Effect.gen(function* () { + return Db.fromSqlClient(yield* SqlClient.SqlClient, { dialect: postgresDialect }) + }), +).pipe(Layer.provideMerge(PgliteClient.layer({ postgresqlconf: "timezone = 'UTC'" }))) + +const Items = PG.table("traced_items", { columns: { id: PG.int4, name: PG.text }, primaryKey: ["id"] }) + +/** Runs `effect` traced, and returns the spans it opened, by name. */ +const traced = (effect: Effect.Effect) => + Effect.gen(function* () { + spans.length = 0 + const exit = yield* Effect.exit(Effect.withTracer(effect, recording)) + return { exit, spans: [...spans] } + }) + +const dbSpan = (recorded: ReadonlyArray) => { + const span = recorded.find((s) => s.attributes.has("db.system.name") && s.name !== "sql.execute") + assert(span !== undefined, `no statement span among ${recorded.map((s) => s.name).join(", ")}`) + return span +} + +layer(Live)("Database statement spans", (it) => { + it.effect("creates the table", () => + Db.execute(Db.sql`CREATE TABLE traced_items (id int4 PRIMARY KEY, name text NOT NULL)`), + ) + + it.effect("names a SELECT after its operation and table, as a client span", () => + Effect.gen(function* () { + const { exit, spans } = yield* traced( + Db.run(PG.from(Items).select(($) => ({ name: $.name })).where(($) => [$.id.eq(PG.param.int("id"))]), { id: 1 }), + ) + assert(Exit.isSuccess(exit)) + const span = dbSpan(spans) + expect(span.name).toBe("SELECT traced_items") + expect(span.kind).toBe("client") + expect(Object.fromEntries(span.attributes)).toMatchObject({ + "db.system.name": "postgresql", + "db.operation.name": "SELECT", + "db.collection.name": "traced_items", + "db.query.summary": "SELECT traced_items", + "db.response.returned_rows": 0, + }) + // Effect's own span runs inside it, carrying the query text. + const execute = spans.find((s) => s.name === "sql.execute") + assert(execute !== undefined) + expect(Option.getOrUndefined(execute.parent)).toBe(span) + }), + ) + + it.effect("names a write after its target table and counts RETURNING rows", () => + Effect.gen(function* () { + const { exit, spans } = yield* traced( + Db.run(PG.insertInto(Items).values({ id: 2, name: "two" }).returning(($) => ({ id: $.id }))), + ) + assert(Exit.isSuccess(exit)) + const span = dbSpan(spans) + expect(span.name).toBe("INSERT traced_items") + expect(span.attributes.get("db.collection.name")).toBe("traced_items") + expect(span.attributes.get("db.response.returned_rows")).toBe(1) + }), + ) + + it.effect("leaves the table out when there is no single one", () => + Effect.gen(function* () { + const inner = PG.from(Items).select(($) => ({ name: $.name })) + const { spans } = yield* traced(Db.run(PG.fromQuery(inner, "sub").select(($) => ({ name: $.name })))) + const span = dbSpan(spans) + expect(span.name).toBe("SELECT") + expect(span.attributes.has("db.collection.name")).toBe(false) + }), + ) + + it.effect("takes the summary from withQuerySummary", () => + Effect.gen(function* () { + const { spans } = yield* traced( + Db.run(PG.from(Items).select(($) => ({ id: $.id }))).pipe(Db.withQuerySummary("items.list")), + ) + const span = dbSpan(spans) + expect(span.name).toBe("items.list") + expect(span.attributes.get("db.query.summary")).toBe("items.list") + expect(span.attributes.get("db.operation.name")).toBe("SELECT") + }), + ) + + it.effect("records error.type from a DatabaseError's reason", () => + Effect.gen(function* () { + const { exit, spans } = yield* traced(Db.execute(Db.sql`SELEKT 1`)) + assert(Exit.isFailure(exit)) + const span = dbSpan(spans) + expect(span.attributes.get("error.type")).toBe("SqlSyntaxError") + expect(span.status._tag).toBe("Ended") + }), + ) +}) + +describe("collection", () => { + it("is the FROM table, or the written table, and absent otherwise", () => { + const select = PG.compileUnsafe(PG.from(Items).select(($) => ({ id: $.id }))) + expect(select.collection).toBe("traced_items") + const write = PG.compileUnsafe(PG.deleteFrom(Items).where(($) => [$.id.eq(1)])) + expect(write.collection).toBe("traced_items") + const cte = PG.compileUnsafe( + PG.from(PG.table("recent", { external: true, columns: { id: PG.int4 } })) + .withCTE("recent", PG.from(Items).select(($) => ({ id: $.id }))) + .select(($) => ({ id: $.id })), + ) + expect(cte.collection).toBeUndefined() + }) +}) diff --git a/src/database/tracing.ts b/src/database/tracing.ts new file mode 100644 index 0000000..97237ac --- /dev/null +++ b/src/database/tracing.ts @@ -0,0 +1,82 @@ +// One client span per statement, named and attributed the way the OpenTelemetry +// database semantic conventions describe: +// https://opentelemetry.io/docs/specs/semconv/database/database-spans/ +// +// Effect's own `sql.execute` span still runs inside it with the query text, but +// it cannot carry these: it names the operation after the client method +// (`execute`), and only a compiled query knows its table. +import { Context, Effect } from "effect" +import type { CompiledQuery } from "../ch/compile" +import type { Dialect } from "../ch/dialect" + +/** What the spans read off a statement. A `CompiledQuery` has all of it. */ +export type TracedStatement = { readonly sql: string } & Partial, "kind" | "collection">> + +/** + * A low-cardinality name for the statements run inside an effect, recorded as + * `db.query.summary` and used as the span name. Without one the summary is + * the operation and table (`SELECT events`), as the conventions recommend. + */ +export const QuerySummary = Context.Reference("@maple-dev/effect-orm/QuerySummary", { + defaultValue: () => undefined, +}) + +/** Names the statements `effect` runs, for spans: `db.query.summary`. */ +export const withQuerySummary: { + (summary: string): (effect: Effect.Effect) => Effect.Effect + (effect: Effect.Effect, summary: string): Effect.Effect +} = ((...args: ReadonlyArray) => + typeof args[0] === "string" + ? (effect: Effect.Effect) => Effect.provideService(effect, QuerySummary, args[0] as string) + : Effect.provideService(args[0] as Effect.Effect, QuerySummary, args[1] as string)) as never + +/** `db.system.name` values the conventions define, by dialect name. */ +const SYSTEM_NAMES: Record = { clickhouse: "clickhouse", postgres: "postgresql" } + +const OPERATIONS: Record, string> = { + select: "SELECT", + insert: "INSERT", + update: "UPDATE", + delete: "DELETE", +} + +/** + * Runs `effect`, which sends `statement`, in its client span. The span name is + * the summary when there is one, then `{operation} {table}`, then the system. + * `rowsOf` reads `db.response.returned_rows` from the result, when it has rows. + */ +export const traceStatement = ( + dialect: Dialect, + statement: TracedStatement, + effect: Effect.Effect, + rowsOf?: (result: A) => number, +): Effect.Effect => + Effect.withFiber((fiber) => { + const configured = fiber.getRef(QuerySummary) + const system = SYSTEM_NAMES[dialect.name] ?? dialect.name + const operation = statement.kind === undefined ? undefined : OPERATIONS[statement.kind] + const summary = + configured ?? (operation === undefined ? undefined : statement.collection === undefined ? operation : `${operation} ${statement.collection}`) + const attributes: Record = { "db.system.name": system } + if (operation !== undefined) attributes["db.operation.name"] = operation + if (statement.collection !== undefined) attributes["db.collection.name"] = statement.collection + if (summary !== undefined) attributes["db.query.summary"] = summary + return effect.pipe( + Effect.tap((result) => + rowsOf === undefined ? Effect.void : Effect.annotateCurrentSpan("db.response.returned_rows", rowsOf(result)), + ), + // `error.type` is the failure's class: a DatabaseError's reason, else its tag. + Effect.tapError((error) => { + const reason = + typeof error === "object" && error !== null + ? "reason" in error && typeof error.reason === "string" + ? error.reason + : "_tag" in error && typeof error._tag === "string" + ? error._tag + : undefined + : undefined + return reason === undefined ? Effect.void : Effect.annotateCurrentSpan("error.type", reason) + }), + Effect.withSpan(summary ?? system, { kind: "client", attributes }), + ) + }) diff --git a/tests/clickhouse-http.clickhouse.test.ts b/tests/clickhouse-http.clickhouse.test.ts new file mode 100644 index 0000000..909cc34 --- /dev/null +++ b/tests/clickhouse-http.clickhouse.test.ts @@ -0,0 +1,91 @@ +// The ClickHouse HTTP client against a real server, over `fetch`. Opt in with +// EFFECT_ORM_CLICKHOUSE_URL (and _USER, _PASSWORD); like the other live suites +// it creates no tables and writes no data. +import { assert, describe, expect, it } from "@effect/vitest" +import { Duration, Effect, Exit, Layer, Schedule, Stream } from "effect" +import { FetchHttpClient } from "effect/http" +import * as TestClock from "effect/testing/TestClock" +import * as SqlClient from "effect/sql/SqlClient" +import * as CH from "@maple-dev/effect-orm/clickhouse" +import * as Db from "@maple-dev/effect-orm/database" +import { ClickhouseHttpClient, layer } from "@maple-dev/effect-orm/clickhouse-http" +import { endpoint } from "./clickhouse-support" + +const Client = layer({ + url: endpoint ?? "http://127.0.0.1:8123", + username: process.env.EFFECT_ORM_CLICKHOUSE_USER ?? "default", + password: process.env.EFFECT_ORM_CLICKHOUSE_PASSWORD ?? "", + settings: { session_timezone: "UTC" }, +}).pipe(Layer.provide(FetchHttpClient.layer)) + +const Live = Layer.effect( + Db.Database, + Effect.gen(function* () { + return Db.fromSqlClient(yield* SqlClient.SqlClient, { dialect: CH.clickhouseDialect }) + }), +).pipe(Layer.provideMerge(Client)) + +const Numbers = CH.table("system.numbers", { external: true, columns: { number: CH.uint64 } }) + +describe.skipIf(!endpoint)("ClickhouseHttpClient on ClickHouse", () => { + it.layer(Live)((it) => { + it.effect("runs a builder query and decodes it", () => + Effect.gen(function* () { + const rows = yield* Db.run( + CH.from(Numbers) + .select(($) => ({ n: $.number })) + .where(($) => [$.number.lt(CH.param.int("limit"))]) + .limit(3), + { limit: 3 }, + ) + expect(rows).toEqual([{ n: 0 }, { n: 1 }, { n: 2 }]) + }), + ) + + it.effect("binds parameters on the server", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + const rows = yield* sql`SELECT ${"o'brien"} AS name, ${41} + 1 AS n` + expect(rows).toEqual([{ name: "o'brien", n: 42 }]) + }), + ) + + it.effect("streams rows", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + const rows = yield* Stream.runCollect(sql.unsafe("SELECT toUInt8(number) AS n FROM numbers(4)").stream) + expect(rows).toEqual([{ n: 0 }, { n: 1 }, { n: 2 }, { n: 3 }]) + }), + ) + + it.effect("classifies a syntax error", () => + Effect.gen(function* () { + const exit = yield* Effect.exit(Db.query({ sql: "SELEKT 1" })) + assert(Exit.isFailure(exit)) + const error = exit.cause.reasons.find((r) => r._tag === "Fail") + assert(error?._tag === "Fail" && error.error._tag === "@maple-dev/effect-orm/DatabaseError") + expect(error.error.reason).toBe("SqlSyntaxError") + }), + ) + + // On the real clock: the timeout below has to fire, which a TestClock never does. + it.effect("stops the query on the server when the fiber is interrupted", () => + TestClock.withLive(Effect.gen(function* () { + const client = yield* ClickhouseHttpClient + const queryId = `effect-orm-cancel-${crypto.randomUUID()}` + const slow = client + .unsafe("SELECT count() FROM numbers(1e12) WHERE NOT ignore(sleepEachRow(0))") + .pipe(client.withQueryId(queryId), client.withSettings({ max_block_size: 1 })) + yield* Effect.timeout(slow, Duration.millis(500)).pipe(Effect.ignore) + // Aborting the request is not enough on its own; the client kills the query by id. + const running = yield* client + .unsafe<{ readonly n: number }>(`SELECT count() AS n FROM system.processes WHERE query_id = '${queryId}'`) + .pipe( + Effect.map((rows) => Number(rows[0]?.n)), + Effect.repeat({ until: (n) => n === 0, schedule: Schedule.spaced(Duration.millis(100)), times: 30 }), + ) + expect(running).toBe(0) + })), + ) + }) +}) diff --git a/tsdown.config.ts b/tsdown.config.ts index eddaa4f..7e1f75c 100644 --- a/tsdown.config.ts +++ b/tsdown.config.ts @@ -3,6 +3,7 @@ import { defineConfig } from "tsdown" export default defineConfig({ entry: { clickhouse: "./src/clickhouse.ts", + "clickhouse-http": "./src/clickhouse-http.ts", expr: "./src/expr.ts", sql: "./src/sql/index.ts", postgres: "./src/postgres.ts", From fb494a4afa02c4412831a1a2232501c433ed1526 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Thu, 15 Oct 2026 19:42:24 +0200 Subject: [PATCH 2/3] fix(clickhouse-http): name the peer on spans; keep a url path - Spans carry server.address, server.port and peer.service ("clickhouse", overridable through spanAttributes, which now apply last). With the HTTP span off, nothing else named the server, so service maps drew no edge to it. - A path in `url` is kept, for ClickHouse behind a proxy prefix; the config doc said paths were refused, which the code never did. Co-Authored-By: Claude Opus 5.5 --- docs/running-queries.md | 5 +++- src/clickhouse-http/client.test.ts | 46 +++++++++++++++++++++++++++++- src/clickhouse-http/client.ts | 13 +++++++-- 3 files changed, 60 insertions(+), 4 deletions(-) diff --git a/docs/running-queries.md b/docs/running-queries.md index 756f86a..f820445 100644 --- a/docs/running-queries.md +++ b/docs/running-queries.md @@ -103,7 +103,10 @@ What it does for you: - **Failures carry Effect's `SqlError` reasons** (`SqlSyntaxError`, `AuthenticationError`, `StatementTimeoutError`, …), read from ClickHouse's exception code, the same as `@effect/sql-clickhouse`. -- **Credentials travel in a header,** never the URL; the URL may not contain them. +- **Credentials travel in a header,** never the URL; the URL may not contain them. A path in the + URL is kept, for a ClickHouse behind a proxy prefix. +- **Spans name the peer:** `server.address`, `server.port` and `peer.service` (`clickhouse`; + override it with `spanAttributes`), so a service map draws the edge to the database. - `asCommand` runs DDL and writes, `withQueryId` sets the `query_id`, and `withSettings` adds ClickHouse settings over the configured ones. Statements bind `sql` template values on the server, as `{pN: Type}` placeholders. diff --git a/src/clickhouse-http/client.test.ts b/src/clickhouse-http/client.test.ts index 6ab12c1..e4d9f23 100644 --- a/src/clickhouse-http/client.test.ts +++ b/src/clickhouse-http/client.test.ts @@ -2,7 +2,7 @@ // reads replies and failures, and that `Database` runs through it unchanged. // tests/clickhouse-http.clickhouse.test.ts runs the same client on a server. import { assert, describe, expect, it } from "@effect/vitest" -import { Effect, Exit, Fiber, Layer, Redacted, Stream } from "effect" +import { Effect, Exit, Fiber, Layer, Redacted, Stream, Tracer } from "effect" import * as HttpClient from "effect/http/HttpClient" import * as HttpClientResponse from "effect/http/HttpClientResponse" import * as SqlClient from "effect/sql/SqlClient" @@ -184,6 +184,50 @@ describe("ClickhouseHttpClient", () => { }).pipe(Effect.provide(live)) }) + it.effect("keeps a path in the url, for ClickHouse behind a proxy prefix", () => { + const { sent, live } = fake(() => json([]), { url: "https://gateway.example/clickhouse" }) + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + yield* sql.unsafe("SELECT 1") + expect(sent[0]?.url.pathname).toBe("/clickhouse") + }).pipe(Effect.provide(live)) + }) + + it.effect("names the peer on its spans, which the caller can override", () => { + const spans: Array = [] + const recording = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan({ ...options, sampled: true }) + spans.push(span) + return span + }, + }) + const run = (config: Partial) => { + const { live } = fake(() => json([]), config) + spans.length = 0 + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + yield* sql.unsafe("SELECT 1") + const span = spans.find((s) => s.name === "sql.execute") + assert(span !== undefined) + return Object.fromEntries(span.attributes) + }).pipe(Effect.provide(live), Effect.withTracer(recording)) + } + return Effect.gen(function* () { + expect(yield* run({})).toMatchObject({ + "db.system.name": "clickhouse", + "db.namespace": "analytics", + "server.address": "clickhouse.example", + "server.port": 8443, + "peer.service": "clickhouse", + }) + expect((yield* run({ url: "https://ch.internal", spanAttributes: { "peer.service": "events-clickhouse" } }))).toMatchObject({ + "server.port": 443, + "peer.service": "events-clickhouse", + }) + }) + }) + it("refuses credentials, a query string or a fragment in the url", () => { for (const url of ["https://u:p@clickhouse.example", "https://clickhouse.example/?x=1", "https://clickhouse.example/#f"]) { const { live } = fake(() => json([]), { url }) diff --git a/src/clickhouse-http/client.ts b/src/clickhouse-http/client.ts index 48400ab..2455d3c 100644 --- a/src/clickhouse-http/client.ts +++ b/src/clickhouse-http/client.ts @@ -31,7 +31,11 @@ import * as Statement from "effect/sql/Statement" export type ClickhouseSetting = string | number | boolean export interface ClickhouseHttpClientConfig { - /** The server, e.g. `https://abc.clickhouse.cloud:8443`. No credentials, path, query or fragment. */ + /** + * The server, e.g. `https://abc.clickhouse.cloud:8443`. A path is kept, for a + * ClickHouse behind a proxy prefix (`https://gateway/clickhouse`); credentials, + * a query string or a fragment are refused. + */ readonly url: string readonly username?: string | undefined readonly password?: Redacted.Redacted | string | undefined @@ -259,10 +263,15 @@ export const make = ( const client = yield* Client.make({ acquirer: Effect.succeed(connection), compiler, + // The peer, for service maps: with the HTTP span off, nothing else names it. + // The caller's attributes come last, so they can rename the peer. spanAttributes: [ - ...Object.entries(config.spanAttributes ?? {}), ["db.system.name", "clickhouse"], ["db.namespace", database], + ["server.address", url.hostname], + ["server.port", Number(url.port || (url.protocol === "https:" ? 443 : 80))], + ["peer.service", "clickhouse"], + ...Object.entries(config.spanAttributes ?? {}), ], // ClickHouse has no transactions over HTTP without a session; effect-orm's // ClickHouse dialect refuses them before this is ever sent. From 3cf5ef7fcc9a749863d4f470eada99a70bb83a13 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Thu, 15 Oct 2026 19:48:58 +0200 Subject: [PATCH 3/3] fix(clickhouse-http): kill the query when a body read or stream ends early The KILL QUERY guard covered only the request, which returns once ClickHouse sends its headers, i.e. with the first block. Reading the body afterwards, or a stream ended by interruption, failure or an early stop (`Stream.take`), left the query running on the server. `send` now hands back the query id; a JSON read kills on interrupt, and a stream kills on any ending other than reading its body to the end. Tests cover an early-stopped stream, an interrupted body read and a fully read stream (no kill); each fails without its guard. Co-Authored-By: Claude Opus 5.5 --- docs/running-queries.md | 4 ++- src/clickhouse-http/client.test.ts | 51 ++++++++++++++++++++++++++++++ src/clickhouse-http/client.ts | 28 ++++++++++------ 3 files changed, 73 insertions(+), 10 deletions(-) diff --git a/docs/running-queries.md b/docs/running-queries.md index f820445..717fde4 100644 --- a/docs/running-queries.md +++ b/docs/running-queries.md @@ -99,7 +99,9 @@ What it does for you: - **Interrupting cancels the query.** A timeout or an interrupted fiber aborts the request and sends `KILL QUERY` for its `query_id`: a closed connection alone does not stop a query - ClickHouse is still computing. Queries also send `cancel_http_readonly_queries_on_client_close`. + ClickHouse is still computing. That holds while the body is read too, and for a `.stream` + that ends before its body does (interrupted, failed, or stopped early with `Stream.take`). + Queries also send `cancel_http_readonly_queries_on_client_close`. - **Failures carry Effect's `SqlError` reasons** (`SqlSyntaxError`, `AuthenticationError`, `StatementTimeoutError`, …), read from ClickHouse's exception code, the same as `@effect/sql-clickhouse`. diff --git a/src/clickhouse-http/client.test.ts b/src/clickhouse-http/client.test.ts index e4d9f23..b94f8e5 100644 --- a/src/clickhouse-http/client.test.ts +++ b/src/clickhouse-http/client.test.ts @@ -184,6 +184,57 @@ describe("ClickhouseHttpClient", () => { }).pipe(Effect.provide(live)) }) + /** A client whose queries answer 200 with `first` and then a body that never ends. */ + const hanging = (first: string) => { + const sent: Array = [] + const http = HttpClient.make((request, url) => { + const body = request.body._tag === "Uint8Array" ? (request.body.text ?? "") : "" + sent.push({ url, body, headers: { ...request.headers } }) + if (body.startsWith("KILL")) return Effect.succeed(HttpClientResponse.fromWeb(request, new Response(""))) + const stream = new ReadableStream({ + start(controller) { + if (first !== "") controller.enqueue(new TextEncoder().encode(first)) + }, + }) + return Effect.succeed(HttpClientResponse.fromWeb(request, new Response(stream))) + }) + const live = layer({ url: "https://clickhouse.example" }).pipe(Layer.provide(Layer.succeed(HttpClient.HttpClient, http))) + const kills = () => sent.filter((request) => request.body.startsWith("KILL")) + return { live, kills } + } + + it.effect("kills the query when a stream is stopped before its body ends", () => { + const { live, kills } = hanging('{"n":1}\n') + return Effect.gen(function* () { + const client = yield* ClickhouseHttpClient + const rows = yield* Stream.runCollect( + client.unsafe("SELECT number AS n FROM numbers(1e9)").stream.pipe(Stream.take(1)), + ).pipe(client.withQueryId("early-1")) + expect(rows).toEqual([{ n: 1 }]) + expect(kills().map((kill) => kill.url.searchParams.get("param_id"))).toEqual(["early-1"]) + }).pipe(Effect.provide(live)) + }) + + it.effect("kills the query when interrupted while reading a response body", () => { + const { live, kills } = hanging("") + return Effect.gen(function* () { + const client = yield* ClickhouseHttpClient + const fiber = yield* Effect.forkChild(client.unsafe("SELECT sleep(3)").pipe(client.withQueryId("body-1"))) + for (let i = 0; i < 5; i++) yield* Effect.yieldNow + yield* Fiber.interrupt(fiber) + expect(kills().map((kill) => kill.url.searchParams.get("param_id"))).toEqual(["body-1"]) + }).pipe(Effect.provide(live)) + }) + + it.effect("sends no kill for a stream read to the end", () => { + const { sent, live } = fake(() => ({ body: '{"n":1}\n{"n":2}\n' })) + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + expect(yield* Stream.runCollect(sql.unsafe("SELECT 1").stream)).toEqual([{ n: 1 }, { n: 2 }]) + expect(sent.filter((request) => request.body.startsWith("KILL"))).toEqual([]) + }).pipe(Effect.provide(live)) + }) + it.effect("keeps a path in the url, for ClickHouse behind a proxy prefix", () => { const { sent, live } = fake(() => json([]), { url: "https://gateway.example/clickhouse" }) return Effect.gen(function* () { diff --git a/src/clickhouse-http/client.ts b/src/clickhouse-http/client.ts index 2455d3c..d771e33 100644 --- a/src/clickhouse-http/client.ts +++ b/src/clickhouse-http/client.ts @@ -169,7 +169,7 @@ export const make = ( params: ReadonlyArray, format: "JSON" | "JSONCompact" | "JSONEachRow" | undefined, operation: string, - ): Effect.Effect => + ): Effect.Effect<{ readonly response: HttpClientResponse.HttpClientResponse; readonly queryId: string }, SqlError, Scope.Scope> => Effect.withFiber((fiber) => { const command = fiber.getRef(ClientMethod) === "command" const queryId = fiber.getRef(QueryId) ?? crypto.randomUUID() @@ -186,7 +186,8 @@ export const make = ( // Aborting the request does not stop the query: ClickHouse notices a // closed connection only when it next writes to it, which a query still // computing its first block may not do for a long time. So kill it by id, - // as Effect's Node driver does. + // as Effect's Node driver does. The body is read after this returns (the + // headers come with the first block), so readers keep the same guard. Effect.onInterrupt(() => kill(queryId)), Effect.mapError( (cause) => @@ -196,7 +197,7 @@ export const make = ( ), Effect.flatMap((response) => response.status === 200 - ? Effect.succeed(response) + ? Effect.succeed({ response, queryId }) : Effect.flatMap(Effect.orElseSucceed(response.text, () => ""), (body) => { const header = response.headers["x-clickhouse-exception-code"] const code = header === undefined ? codeFromBody(body) : Number(header) @@ -216,8 +217,11 @@ export const make = ( fiber.getRef(ClientMethod) === "command" ? Effect.as(send(sql, params, undefined, "execute"), [] as ReadonlyArray) : send(sql, params, format, "execute").pipe( - Effect.flatMap((response) => - Effect.mapError(response.text, (cause) => parseFailure(cause, "execute")), + Effect.flatMap(({ response, queryId }) => + response.text.pipe( + Effect.mapError((cause) => parseFailure(cause, "execute")), + Effect.onInterrupt(() => kill(queryId)), + ), ), Effect.flatMap((text) => Effect.try({ @@ -239,8 +243,13 @@ export const make = ( transformRows ? Effect.map(json(sql, params, "JSON"), transformRows) : json(sql, params, "JSON"), executeStream: (sql, params, transformRows) => send(sql, params, "JSONEachRow", "stream").pipe( - Effect.map((response) => - response.stream.pipe( + Effect.map(({ response, queryId }) => { + // A stream can end before the body does: interrupted, failed, or stopped + // early by its consumer (`Stream.take`). Any of those leaves the query + // running on the server, so it is killed unless the body was read to the end. + let drained = false + return response.stream.pipe( + Stream.onEnd(Effect.sync(() => void (drained = true))), Stream.mapError( (cause) => new SqlError({ @@ -254,8 +263,9 @@ export const make = ( Effect.try({ try: () => JSON.parse(line) as object, catch: (cause) => parseFailure(cause, "stream") }), ), Stream.map((row) => (transformRows ? transformRows([row])[0] : row)), - ), - ), + Stream.ensuring(Effect.suspend(() => (drained ? Effect.void : kill(queryId)))), + ) + }), Stream.unwrap, ), }