diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c8b332a9..c0192435 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -4,6 +4,8 @@ on: branches: [main] # A change to docs or Markdown alone cannot break a build, so it skips CI. # docs/coverage still runs it: the Coverage job regenerates the page there. + # docs/published runs it too: the coverage generator holds each published + # page to the invariants it cites, so a page-only edit must reach that check. pull_request: branches: [main] paths: @@ -11,6 +13,7 @@ on: - "!docs/**" - "!**.md" - "docs/coverage/**" + - "docs/published/**" jobs: # The Go engine (sqlflow), unit pass only. Fast feedback with no Docker: the # integration tests start containers, and the job below runs those. -short diff --git a/dev/config/examples/published/README.md b/dev/config/examples/published/README.md index b19104fc..f8752911 100644 --- a/dev/config/examples/published/README.md +++ b/dev/config/examples/published/README.md @@ -19,3 +19,8 @@ generated JSON schema. When a page changes, change the file here first and copy it out. When the engine changes, CI tells you which published page needs an edit before a reader finds out. + +The ClickHouse page itself lives in this repo too, at +[`docs/published/clickhouse.mdx`](../../../../docs/published/clickhouse.mdx). +Its YAML blocks must equal the two `clickhouse.*` files here byte for byte, +and the coverage generator fails when they drift. diff --git a/docs/coverage/clickhouse-types.mdx b/docs/coverage/clickhouse-types.mdx index 03e3e51f..528804cd 100644 --- a/docs/coverage/clickhouse-types.mdx +++ b/docs/coverage/clickhouse-types.mdx @@ -18,7 +18,7 @@ SQLFlow hands each batch to ClickHouse as rows built from the Arrow table DuckDB | `UBIGINT` | `UInt64` | | | `REAL` | `Float32` | | | `DOUBLE` | `Float64` | | -| `VARCHAR`, `UUID`, `JSON` | `String`, `LowCardinality(String)`, `DateTime64(3)`, `DateTime`, `Enum8('' = 0, 'a' = 1, 'b' = 2)`, `FixedString(36)`, `UUID`, `IPv4`, `Decimal(10, 2)` | | +| `VARCHAR` | `String`, `LowCardinality(String)`, `DateTime64(3)`, `DateTime`, `Enum8('' = 0, 'a' = 1, 'b' = 2)`, `FixedString(36)`, `UUID`, `IPv4`, `Decimal(10, 2)` | the text must already be the column's text form. `UUID` and `JSON` reach the sink as `VARCHAR`. | | `BLOB`, `BIT`, `VARINT` | `String` | | | `DATE` | `Date`, `Date32` | | | `TIMESTAMP_S` | `DateTime` | | @@ -46,7 +46,7 @@ SQLFlow hands each batch to ClickHouse as rows built from the Arrow table DuckDB | `TIMESTAMP_NS[]` | `Array(DateTime64(9))` | | | `TIMESTAMPTZ[]` | `Array(DateTime64(6))` | stored as the same UTC instant; the zone is dropped and the column's own zone governs rendering | | `BIGINT[][]` | `Array(Array(Int64))` | | -| `NULL` in any column | the column's own type | the column type's zero value; the DEFAULT expression is not evaluated because every produced column is named in the INSERT | +| `NULL` in any column | the column's own type | the column type's zero value; the DEFAULT expression is not evaluated because every produced column is named in the INSERT. For an Enum with no 0 member that value is not a member, and the row cannot be read | | `NULL` inside a `LIST` | the element's own type | the element type's zero value, because ClickHouse's Array(T) holds no nulls unless T is Nullable | Values are matched to columns **by name**, not by position. diff --git a/docs/coverage/integrations/sink.clickhouse.yml b/docs/coverage/integrations/sink.clickhouse.yml index fc1716d0..c387b517 100644 --- a/docs/coverage/integrations/sink.clickhouse.yml +++ b/docs/coverage/integrations/sink.clickhouse.yml @@ -82,6 +82,7 @@ types: # override the value the lattice sends. utf8: outcome: exact + rule: the text must already be the column's text form. columns: - type: String expect: "L'Ε’il πŸ‘ \"quoted\" back\\slash\ttab" @@ -374,7 +375,7 @@ types: nulls: default: outcome: coerced - rule: the column type's zero value; the DEFAULT expression is not evaluated because every produced column is named in the INSERT + rule: the column type's zero value; the DEFAULT expression is not evaluated because every produced column is named in the INSERT. For an Enum with no 0 member that value is not a member, and the row cannot be read # The third null position, and the one the published page does not # mention. arrowListValue appends the element type's zero value at # clickhouse.go:429-431, so [1, null, 3] lands as [1, 0, 3] with no diff --git a/docs/coverage/invariants.yml b/docs/coverage/invariants.yml index 130dc7df..307c3eab 100644 --- a/docs/coverage/invariants.yml +++ b/docs/coverage/invariants.yml @@ -7,7 +7,8 @@ # # `applies_to` is the interface the invariant is a property of. Every # integration of that kind must prove it, or be exempt in the registry -# with a reason. +# with a reason. It may instead name one integration, for a claim only that +# integration makes; the id then starts with the integration's namespace. # # `verified_by` says how evidence attaches: # @@ -19,10 +20,11 @@ # never implemented, and dropped as a third way to claim the same evidence. # scripts/coverage_matrix accepts the two above and nothing else. # -# `requires` lists the levels evidence must exist at. Every entry is empty in -# this revision: the matrix reports, and nothing fails. An invariant is -# enforced by filling one in, which is legal only once every non-exempt -# integration has a subject. +# `requires` lists the levels evidence must exist at. Every kind's entry is +# empty in this revision: the matrix reports, and nothing fails. An invariant +# is enforced by filling one in, which is legal only once every non-exempt +# integration has a subject. An invariant scoped to one integration, such as +# clickhouse.*, is written with its subject and requires its level. # # `tracked_by` marks an invariant the code does not yet satisfy. It renders as # declared-unenforced until that issue lands. @@ -396,6 +398,140 @@ invariants: verified_by: typetable requires: [] + # --- ClickHouse ---------------------------------------------------------- + # Claims only sink.clickhouse makes. Each one backs a sentence on + # docs/published/clickhouse.mdx, which cites it by id, and the page check + # fails when the two disagree. A failing test here is a page that is now + # wrong: fix the page or the sink, never the assertion. + - id: clickhouse.insert.by_name + family: clickhouse + class: safety + applies_to: sink.clickhouse + claim: > + Columns are matched by name. A produced column the table lacks + fails the batch as `system.sink.write_failed`, naming the column. + A column the handler omits takes the table's DEFAULT. + verified_by: harness + requires: [integration] + + - id: clickhouse.table.not_created + family: clickhouse + class: safety + applies_to: sink.clickhouse + claim: > + The sink never creates or alters a table. A table that does not + exist fails the batch as `system.sink.write_failed`. + verified_by: harness + requires: [integration] + + - id: clickhouse.enum.null_is_zero + family: clickhouse + class: safety + applies_to: sink.clickhouse + claim: > + A NULL into a non-Nullable Enum is stored as 0. With no 0 member + the INSERT succeeds and reads of the column fail with + `UNKNOWN_ELEMENT_OF_ENUM`. With a 0 member it reads as that + member. Nullable reads NULL. + verified_by: harness + requires: [integration] + + - id: clickhouse.encode.fails_once + family: clickhouse + class: safety + applies_to: sink.clickhouse + claim: > + A value the driver cannot encode fails the batch on the first + attempt as `user.sink.encode_failed`. It is not retried, and the + process exits with code 10. + verified_by: harness + requires: [integration] + + - id: clickhouse.temporal_text.two_shapes + family: clickhouse + class: safety + applies_to: sink.clickhouse + claim: > + Text into a DateTime column takes `YYYY-MM-DD hh:mm:ss`, read as UTC + whatever the column's zone, or the same with an offset, which is + applied. Any other shape fails as `user.sink.encode_failed`. + verified_by: harness + requires: [integration] + + - id: clickhouse.decimal.text_only + family: clickhouse + class: safety + applies_to: sink.clickhouse + claim: > + A Decimal column accepts its value as text. A float fails as + `user.sink.encode_failed`. + verified_by: harness + requires: [integration] + + - id: clickhouse.replay.duplicates_until_merge + family: clickhouse + class: safety + applies_to: sink.clickhouse + claim: > + A batch delivered twice into a ReplacingMergeTree reads twice + until a merge, and once with FINAL. An identical block inside the + table's deduplication window is dropped at insert. + verified_by: harness + requires: [integration] + + - id: clickhouse.flush.one_insert + family: clickhouse + class: safety + applies_to: sink.clickhouse + claim: > + One flush is one INSERT, however many batches it buffered. + verified_by: harness + requires: [integration] + + - id: clickhouse.dsn.scheme + family: clickhouse + class: safety + applies_to: sink.clickhouse + claim: > + The DSN scheme decides the protocol, TLS and the default port. `@` + and `:` in a password parse raw. `/`, `#` and `?` must be percent- + encoded. + verified_by: harness + requires: [unit] + + - id: clickhouse.tls.verifies + family: clickhouse + class: safety + applies_to: sink.clickhouse + claim: > + With TLS, the server certificate is verified against the system + roots. No option skips verification. + verified_by: harness + requires: [unit] + + - id: clickhouse.auth.fails_start + family: clickhouse + class: safety + applies_to: sink.clickhouse + claim: > + Wrong credentials fail the startup probe, before any message is + consumed, with ClickHouse's "Authentication failed: password is + incorrect, or there is no user with such name". The HTTP status and + code vary by server and user. + verified_by: harness + requires: [integration] + + - id: clickhouse.type_text.per_destination + family: clickhouse + class: safety + applies_to: sink.clickhouse + claim: > + Text into a non-String column must already be in that column's + text form. The type table proves each destination with its own + value. + verified_by: harness + requires: [integration] + # --- Lifecycle ----------------------------------------------------------- - id: lifecycle.drain.on_cancel family: lifecycle diff --git a/docs/coverage/matrix.md b/docs/coverage/matrix.md index c2f266e4..89014b35 100644 --- a/docs/coverage/matrix.md +++ b/docs/coverage/matrix.md @@ -82,7 +82,7 @@ integration behind it keeps a batch it could not deliver, or commits offsets only after a flush. Those are invariants, they are counted separately below, and the two numbers are not interchangeable. -**55 invariants declared: 47 safety and 8 liveness. Of 195 (invariant, integration) cells: 97 proven, 63 missing, 0 skipped, 0 failing, 35 exempt. 0 gap(s).** +**67 invariants declared: 59 safety and 8 liveness. Of 207 (invariant, integration) cells: 110 proven, 62 missing, 0 skipped, 0 failing, 35 exempt. 0 gap(s).** Safety says nothing bad happens. Liveness says something good eventually does, and the two are not interchangeable: a sink that @@ -122,7 +122,7 @@ until an invariant's `requires` is filled in, and none is yet. | `sink.flush.honours_context` | Flush returns ctx.Err() when the context ends. Rows stay buffered. *(violated once: #219)* | βœ… i | β€” exempt | β€” exempt | βœ… i | β€” exempt | βœ… i | β€” exempt | | `sink.flush.empty_is_noop` | Flush with nothing buffered returns nil and touches nothing. | βœ… i | βœ… u | βœ… u | βœ… i | β€” exempt | βœ… i | βœ… u | | `sink.buffer.reports_depth` | A sink reports how many rows it is holding, and the count rises when a flush fails and falls to zero when one succeeds. | βœ… i | βœ… u | βœ… u | βœ… i | β€” exempt | βœ… i | βœ… u | -| `sink.error.classifies` | The sink's errors classify as unreachable, rejected, or a user fault, so the retry ladder retries only the first. A value the sink's client cannot encode is a user fault: it never reaches the network and fails the same way every attempt. | ❌ missing | ❌ missing | ❌ missing | ❌ missing | β€” exempt | ❌ missing | ❌ missing | +| `sink.error.classifies` | The sink's errors classify as unreachable, rejected, or a user fault, so the retry ladder retries only the first. A value the sink's client cannot encode is a user fault: it never reaches the network and fails the same way every attempt. | βœ… i | ❌ missing | ❌ missing | ❌ missing | β€” exempt | ❌ missing | ❌ missing | | `sink.probe.fails_start` | A Prober whose destination is unreachable fails the start once, without retrying. | βœ… i | β€” exempt | β€” exempt | βœ… i | β€” exempt | βœ… i | β€” exempt | | `sink.flush.idempotent_on_key` | Delivering the same batch twice leaves the destination holding it once. The engine promises at-least-once; a sink that identifies rows by key is what turns that into exactly-once for the reader. A sink that declares no key is exempt, with a proof that it declares none. | β€” exempt | β€” exempt | β€” exempt | β€” exempt | β€” exempt | βœ… i | β€” exempt | @@ -224,6 +224,23 @@ drains. An invariant holds only if it holds on all four. | `serve.cache.errors_not_stored` | A request that fails stores nothing: the next request for its key runs a query. | βœ… u | | `serve.cache.one_fill` | Concurrent requests for one key run one query, and a caller leaving fails no other caller. | βœ… u | +## Safety invariants: clickhouse + +| Invariant | Claim | `sink.clickhouse` | +| --- | --- | --- | +| `clickhouse.insert.by_name` | Columns are matched by name. A produced column the table lacks fails the batch as `system.sink.write_failed`, naming the column. A column the handler omits takes the table's DEFAULT. | βœ… i | +| `clickhouse.table.not_created` | The sink never creates or alters a table. A table that does not exist fails the batch as `system.sink.write_failed`. | βœ… i | +| `clickhouse.enum.null_is_zero` | A NULL into a non-Nullable Enum is stored as 0. With no 0 member the INSERT succeeds and reads of the column fail with `UNKNOWN_ELEMENT_OF_ENUM`. With a 0 member it reads as that member. Nullable reads NULL. | βœ… i | +| `clickhouse.encode.fails_once` | A value the driver cannot encode fails the batch on the first attempt as `user.sink.encode_failed`. It is not retried, and the process exits with code 10. | βœ… ui | +| `clickhouse.temporal_text.two_shapes` | Text into a DateTime column takes `YYYY-MM-DD hh:mm:ss`, read as UTC whatever the column's zone, or the same with an offset, which is applied. Any other shape fails as `user.sink.encode_failed`. | βœ… i | +| `clickhouse.decimal.text_only` | A Decimal column accepts its value as text. A float fails as `user.sink.encode_failed`. | βœ… i | +| `clickhouse.replay.duplicates_until_merge` | A batch delivered twice into a ReplacingMergeTree reads twice until a merge, and once with FINAL. An identical block inside the table's deduplication window is dropped at insert. | βœ… i | +| `clickhouse.flush.one_insert` | One flush is one INSERT, however many batches it buffered. | βœ… i | +| `clickhouse.dsn.scheme` | The DSN scheme decides the protocol, TLS and the default port. `@` and `:` in a password parse raw. `/`, `#` and `?` must be percent- encoded. | βœ… u | +| `clickhouse.tls.verifies` | With TLS, the server certificate is verified against the system roots. No option skips verification. | βœ… u | +| `clickhouse.auth.fails_start` | Wrong credentials fail the startup probe, before any message is consumed, with ClickHouse's "Authentication failed: password is incorrect, or there is no user with such name". The HTTP status and code vary by server and user. | βœ… i | +| `clickhouse.type_text.per_destination` | Text into a non-String column must already be in that column's text form. The type table proves each destination with its own value. | βœ… i | + ## Liveness invariants: lifecycle | Invariant | Claim | `manager.watermark` | diff --git a/docs/coverage/status/sink.clickhouse.yml b/docs/coverage/status/sink.clickhouse.yml index fe68018c..662b2ce6 100644 --- a/docs/coverage/status/sink.clickhouse.yml +++ b/docs/coverage/status/sink.clickhouse.yml @@ -1,8 +1,20 @@ # Generated by `make coverage-matrix`. Do not edit by hand. # One line per invariant that applies to sink.clickhouse, sorted by id: the status of each level. +clickhouse.auth.fails_start: {unit: skipped, integration: covered, growth: missing, release: missing} +clickhouse.decimal.text_only: {unit: skipped, integration: covered, growth: missing, release: missing} +clickhouse.dsn.scheme: {unit: covered, integration: missing, growth: missing, release: missing} +clickhouse.encode.fails_once: {unit: covered, integration: covered, growth: missing, release: missing} +clickhouse.enum.null_is_zero: {unit: skipped, integration: covered, growth: missing, release: missing} +clickhouse.flush.one_insert: {unit: skipped, integration: covered, growth: missing, release: missing} +clickhouse.insert.by_name: {unit: skipped, integration: covered, growth: missing, release: missing} +clickhouse.replay.duplicates_until_merge: {unit: skipped, integration: covered, growth: missing, release: missing} +clickhouse.table.not_created: {unit: skipped, integration: covered, growth: missing, release: missing} +clickhouse.temporal_text.two_shapes: {unit: skipped, integration: covered, growth: missing, release: missing} +clickhouse.tls.verifies: {unit: covered, integration: missing, growth: missing, release: missing} +clickhouse.type_text.per_destination: {unit: skipped, integration: covered, growth: missing, release: missing} lifecycle.close.idempotent: {unit: missing, integration: covered, growth: missing, release: missing} sink.buffer.reports_depth: {unit: missing, integration: covered, growth: missing, release: missing} -sink.error.classifies: {unit: missing, integration: missing, growth: missing, release: missing} +sink.error.classifies: {unit: skipped, integration: covered, growth: missing, release: missing} sink.flush.empty_is_noop: {unit: missing, integration: covered, growth: missing, release: missing} sink.flush.honours_context: {unit: missing, integration: covered, growth: missing, release: missing} sink.flush.idempotent_on_key: {unit: exempt, integration: exempt, growth: exempt, release: exempt} diff --git a/docs/published/README.md b/docs/published/README.md new file mode 100644 index 00000000..4f3c0843 --- /dev/null +++ b/docs/published/README.md @@ -0,0 +1,36 @@ +# Published pages + +Pages SQLFlow publishes in third-party docs. Each page here is the source of +truth for what a vendor's docs say about SQLFlow. The vendor's repository +gets a copy, by hand, after review. + +A page carries an anchor comment beside each claim. MDX renders none of them: + +| Anchor | What it cites | What the check requires | +| --- | --- | --- | +| `{/* invariant: */}` | a claim in `docs/coverage/invariants.yml` | the id is declared | +| `{/* measured: */}` | a committed output file that names its versions | the file exists | +| `{/* config: */}` | the YAML block that follows | it equals `dev/config/examples/published/` | +| `{/* feature: */}` | a tooling claim, proven by a named test | the id is declared in `features.yml` | +| `{/* generated: type-table */}` | the type table that follows | it equals what `sink.clickhouse.yml` renders | +| `{/* generated: unsupported-types */}` | the list that follows | it equals the rendered unsupported list | + +A page named for a namespace cites every invariant in it: `clickhouse.mdx` +must cite each `clickhouse.*` invariant at least once. + +The coverage generator runs the check before it reads any test report, so +`make coverage-page`, `make coverage-check` and CI's Coverage job all fail on +a page that drifted from its evidence. An invariant whose test fails fails +the Integration job, which names the page claim through the invariant's id. + +## Pages + +| Page | Published in | Evidence | +| --- | --- | --- | +| [`clickhouse.mdx`](clickhouse.mdx) | ClickHouse docs, `docs/integrations/connectors/data-ingestion/community-integrations/sqlflow.mdx` | [`clickhouse/`](clickhouse/README.md) | + +## Changing a page + +1. Change the claim here, and its invariant, test or measurement with it. +2. Run `make coverage-page` and `uv run --locked pytest tests/tooling -q`. +3. After the pull request merges, copy the page to the vendor's repository. diff --git a/docs/published/clickhouse.mdx b/docs/published/clickhouse.mdx new file mode 100644 index 00000000..9715d866 --- /dev/null +++ b/docs/published/clickhouse.mdx @@ -0,0 +1,548 @@ +--- +sidebarTitle: 'SQLFlow' +keywords: ['clickhouse', 'sqlflow', 'duckdb', 'kafka', 'streaming', 'stream processing', 'sink', 'data ingestion', 'integrate', 'websocket'] +slug: /integrations/sqlflow +description: 'Stream Kafka, WebSocket and webhook data into ClickHouse, transformed with DuckDB SQL, using SQLFlow' +title: 'SQLFlow and ClickHouse' +doc_type: 'guide' +integration: + - support_level: 'community' + - category: 'data_ingestion' + - website: 'https://turbolytics.io' +--- + +import ConnectionDetails from '/snippets/_gather_your_details_http.mdx'; +import { CommunityMaintainedBadge } from "/snippets/components/CommunityMaintainedBadge/CommunityMaintainedBadge.jsx"; + + + +SQLFlow is an open-source (MIT) stream processing engine. It consumes a stream (Kafka, a WebSocket, or webhooks), runs DuckDB SQL over each micro-batch, and inserts the result into a ClickHouse table. A pipeline is a single YAML file, and SQLFlow ships as one binary and one Docker image, so there is no cluster to run. + +This page covers the ClickHouse sink: connecting to ClickHouse Cloud and self-hosted ClickHouse, which types map, an end-to-end example on the NYC taxi sample dataset, and the limits that were measured rather than assumed. + +## Prerequisites {#prerequisites} + +- SQLFlow **[v2026.10.08](https://github.com/turbolytics/sql-flow/releases/tag/v2026.10.08)** or later. Every result on this page was measured on that release unless it names another. +- ClickHouse Cloud, or self-hosted ClickHouse. This page was run against self-hosted 26.8.1 and Cloud 26.6.1. No minimum below those has been established. +- A Kafka broker the pipeline can reach, for the Kafka source used in the example. The WebSocket and webhook sources need no broker. +- Python 3 with the `confluent-kafka` package (`pip install confluent-kafka`), only for the script in the example that publishes the sample data to Kafka. SQLFlow itself needs no Python. +- Docker is the simplest install. `turbolytics/sql-flow` is published multi-arch for `linux/amd64` and `linux/arm64`. SQLFlow can also be built from source; it needs Go, a C toolchain, and `libduckdb` present at runtime. + +## Setup {#setup} + +SQLFlow connects with a single DSN. The scheme selects the protocol and whether TLS is used; the port defaults from the scheme if omitted. + +| Scheme | Protocol | TLS | Default port | Use for | +| ------------------------------- | -------- | --- | ------------ | ------- | +| `clickhouses://`, `https://` | HTTP | yes | 8443 | ClickHouse Cloud, or any TLS-terminated HTTP endpoint | +| `clickhouse://`, `http://` | HTTP | no | 8123 | Self-hosted ClickHouse | +| `natives://` | native | yes | 9440 | ClickHouse Cloud over the native protocol | +| `tcp://`, `native://` | native | no | 9000 | Self-hosted, native protocol | + +{/* invariant: clickhouse.dsn.scheme */} + +The target table must already exist. SQLFlow does not create tables and does not evolve schemas: it names the columns in its `INSERT` from the column names its SQL produces, so **the columns your handler SQL emits must match the table's column names**. Use aliases in the SQL to rename. {/* invariant: clickhouse.table.not_created */} {/* invariant: clickhouse.insert.by_name */} + +### Connect to ClickHouse Cloud {#connect-to-clickhouse-cloud} + + + +With those details, the DSN is: + +``` +clickhouses://default:@.clickhouse.cloud:8443/ +``` + +Keep it out of the YAML by passing it as an environment variable; the config's `dsn` field is a template: + +```yaml +sink: + type: clickhouse + clickhouse: + dsn: {{ SQLFLOW_CLICKHOUSE_DSN }} + table: trips_small +``` + +```bash +export SQLFLOW_CLICKHOUSE_DSN='clickhouses://default:@.clickhouse.cloud:8443/nyc_taxi' +``` + + +Use `clickhouses://` (with the `s`) for Cloud. `clickhouse://` sends plain HTTP to port 8443, the server closes the connection, and startup fails as `system.sink.unreachable` before any message is consumed. See [Troubleshooting](#troubleshooting). {/* measured: docs/published/clickhouse/cloud-reset.out */} + + +### Connect to self-hosted ClickHouse {#connect-to-self-hosted-clickhouse} + +Point the DSN at the HTTP interface, port 8123 by default: + +```bash +export SQLFLOW_CLICKHOUSE_DSN='clickhouse://default:@clickhouse.internal:8123/nyc_taxi' +``` + +If you have put TLS in front of the HTTP interface, use `clickhouses://` exactly as for Cloud. For the native protocol use `tcp://` (9000) or `natives://` (9440). + +When SQLFlow runs in Docker, `localhost` inside the container is the container itself, not your machine. Put SQLFlow and ClickHouse on the same Docker network and use the ClickHouse container's name as the host. + +## Authentication {#authentication} + +Username and password are carried in the DSN and, with `clickhouses://` or `natives://`, are sent over TLS. The server certificate is verified against the system trust store; there is no option to skip verification. {/* invariant: clickhouse.tls.verifies */} + +- **Special characters in the password.** The DSN is parsed as a URL, so `/`, `#` and `?` must be percent-encoded (`%2F`, `%23`, `%3F`); left raw they produce a confusing `invalid port ":p" after host`. `@` and `:` are handled correctly either way. {/* invariant: clickhouse.dsn.scheme */} +- **Cloud IP access lists.** If your service restricts source IPs, the address SQLFlow connects from has to be on the list. +- **Not supported:** mutual TLS / client certificates, and custom CA bundles. Use username and password. {/* invariant: clickhouse.tls.verifies */} + +## Example: streaming the NYC taxi dataset from Kafka {#example-nyc-taxi} + +This walks the [NYC taxi sample dataset](/get-started/sample-datasets/nyc-taxi) through Kafka into ClickHouse, transformed by SQLFlow on the way. It reproduces on Cloud or self-hosted; only the DSN changes. + +### 1. Create the target table {#example-create-table} + +The same table the sample-dataset page defines: + +```sql +CREATE DATABASE IF NOT EXISTS nyc_taxi; + +CREATE TABLE IF NOT EXISTS nyc_taxi.trips_small ( + trip_id UInt32, + pickup_datetime DateTime, + dropoff_datetime DateTime, + pickup_longitude Nullable(Float64), + pickup_latitude Nullable(Float64), + dropoff_longitude Nullable(Float64), + dropoff_latitude Nullable(Float64), + passenger_count UInt8, + trip_distance Float32, + fare_amount Float32, + extra Float32, + tip_amount Float32, + tolls_amount Float32, + total_amount Float32, + payment_type Enum('CSH' = 1, 'CRE' = 2, 'NOC' = 3, 'DIS' = 4, 'UNK' = 5), + pickup_ntaname LowCardinality(String), + dropoff_ntaname LowCardinality(String) +) +ENGINE = MergeTree +PRIMARY KEY (pickup_datetime, dropoff_datetime); +``` + +### 2. Publish the data to Kafka {#example-publish} + +The dataset is published as gzipped TSV. This step converts one file (about a million rows) to newline-delimited JSON and publishes it to a topic named `nyc-taxi-trips`. + +Install the publisher's dependency and download the file: + +```bash +pip install confluent-kafka +curl -O https://datasets-documentation.s3.eu-west-3.amazonaws.com/nyc-taxi/trips_0.gz +``` + +Create the topic. A stock Apache Kafka broker creates it on first publish, but managed and production clusters commonly set `auto.create.topics.enable=false`, where an uncreated topic fails the example before any row reaches ClickHouse: + +```bash +kafka-topics.sh --bootstrap-server localhost:9092 \ + --create --topic nyc-taxi-trips --partitions 1 --replication-factor 1 +``` + +Confluent's images name that command `kafka-topics`, without the `.sh`. + +Save the publisher as `publish.py`. Change `bootstrap.servers` if the broker is not on `localhost:9092`: + +```python +import gzip, json +from confluent_kafka import Producer + +COLUMNS = ["trip_id", "pickup_datetime", "dropoff_datetime", "pickup_longitude", + "pickup_latitude", "dropoff_longitude", "dropoff_latitude", "passenger_count", + "trip_distance", "fare_amount", "extra", "tip_amount", "tolls_amount", + "total_amount", "payment_type", "pickup_ntaname", "dropoff_ntaname"] +INTS = {"trip_id", "passenger_count"} +FLOATS = {"pickup_longitude", "pickup_latitude", "dropoff_longitude", "dropoff_latitude", + "trip_distance", "fare_amount", "extra", "tip_amount", "tolls_amount", "total_amount"} + +producer = Producer({"bootstrap.servers": "localhost:9092"}) + +def send(topic, value): + # The producer's queue holds 100,000 messages by default, and this file + # has a million rows; poll() drains it as it fills. + while True: + try: + producer.produce(topic, value) + return + except BufferError: + producer.poll(0.5) + +with gzip.open("trips_0.gz", "rb") as f: + header = f.readline().decode().rstrip("\n").split("\t") + index = {name: i for i, name in enumerate(header)} + for raw in f: + # Split on the raw bytes and decode only the columns being published. + # The file carries columns this example does not use that are not + # valid UTF-8, and decoding the whole line would have to corrupt them. + parts = raw.rstrip(b"\n").split(b"\t") + row = {} + for col in COLUMNS: + value = parts[index[col]].decode() + if value == "": + row[col] = None + elif col in INTS: + row[col] = int(float(value)) + elif col in FLOATS: + row[col] = float(value) + else: + row[col] = value + send("nyc-taxi-trips", json.dumps(row)) +producer.flush() +``` + +Run it. Publishing the million rows takes under a minute: {/* measured: docs/published/clickhouse/selfhosted-verify.out */} + +```bash +python publish.py +``` + +### 3. Write the pipeline {#example-pipeline} + +Save this as `taxi.yml`. The handler SQL runs in DuckDB over each batch of JSON messages; the casts produce Arrow types that map onto the table's ClickHouse types (see [Type mapping](#type-mapping)). + +{/* config: clickhouse.taxi.yml */} + +```yaml +pipeline: + name: nyc-taxi-clickhouse + batch_size: 20000 + + source: + type: kafka + kafka: + brokers: [{{ SQLFLOW_KAFKA_BROKERS|default('localhost:9092') }}] + group_id: nyc-taxi + auto_offset_reset: earliest + topics: + - nyc-taxi-trips + + handler: + type: 'handlers.InferredMemBatch' + sql: | + SELECT + CAST(trip_id AS UINTEGER) AS trip_id, + CAST(pickup_datetime AS TIMESTAMP) AS pickup_datetime, + CAST(dropoff_datetime AS TIMESTAMP) AS dropoff_datetime, + pickup_longitude, + pickup_latitude, + dropoff_longitude, + dropoff_latitude, + CAST(passenger_count AS UTINYINT) AS passenger_count, + CAST(trip_distance AS FLOAT) AS trip_distance, + CAST(fare_amount AS FLOAT) AS fare_amount, + CAST(extra AS FLOAT) AS extra, + CAST(tip_amount AS FLOAT) AS tip_amount, + CAST(tolls_amount AS FLOAT) AS tolls_amount, + CAST(total_amount AS FLOAT) AS total_amount, + payment_type, + pickup_ntaname, + dropoff_ntaname + FROM batch + + sink: + type: clickhouse + clickhouse: + dsn: {{ SQLFLOW_CLICKHOUSE_DSN }} + table: trips_small +``` + +`batch_size` is the number of messages per DuckDB batch and per ClickHouse `INSERT`. {/* invariant: clickhouse.flush.one_insert */} See [Performance](#performance) for why 20,000. + +### 4. Run it {#example-run} + +```bash +docker run --rm \ + --network \ + -v "$(pwd)":/conf \ + -e SQLFLOW_CLICKHOUSE_DSN \ + -e SQLFLOW_KAFKA_BROKERS= \ + turbolytics/sql-flow:v2026.10.08 \ + run /conf/taxi.yml --max-msgs=1000000 +``` + + +**The broker address must be one the container can reach.** A Kafka client is redirected to whatever the broker *advertises*, not the address you dial. A broker advertising `127.0.0.1:9092` for external clients sends the container to itself, and every fetch fails with `dial tcp 127.0.0.1:9092: connect: connection refused`, even though the same address works from your shell. Put the container on the broker's own network and use the address advertised there, or run the binary on the host instead: + +```bash +sqlflow run taxi.yml --max-msgs=1000000 +``` + + +`--max-msgs` stops after that many messages; leave it off to run continuously. Kafka offsets are committed only after each `INSERT` has been acknowledged by ClickHouse, and only up to the last message actually processed. {/* invariant: pipeline.commit.after_flush */} + +### 5. Verify {#example-verify} + +```sql +SELECT payment_type, count() AS trips, round(avg(total_amount), 2) AS avg_total +FROM nyc_taxi.trips_small +GROUP BY payment_type +ORDER BY trips DESC; +``` + +``` +β”Œβ”€payment_type─┬──trips─┬─avg_total─┐ +β”‚ CSH β”‚ 617237 β”‚ 17.82 β”‚ +β”‚ CRE β”‚ 377897 β”‚ 13.91 β”‚ +β”‚ NOC β”‚ 3573 β”‚ 13.89 β”‚ +β”‚ DIS β”‚ 1293 β”‚ 12.09 β”‚ +β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ +``` + +The same query gives the same counts against Cloud and self-hosted, which is a reasonable check that nothing was dropped or coerced on either path. {/* measured: docs/published/clickhouse/selfhosted-verify.out */} {/* measured: docs/published/clickhouse/cloudverify.out */} On Cloud, a read served by a replica that has not yet fetched the newest part can undercount for a moment after the run ends. Add `SETTINGS select_sequential_consistency = 1` to the query if the total comes up short. {/* measured: docs/published/clickhouse/perfcloud.out */} + +### A second example: WebSocket source, no Kafka {#example-websocket} + +The WebSocket source needs no broker, and this shows an `Array(String)` column being written and queried natively. It consumes the public Bluesky firehose. + +```sql +CREATE TABLE bluesky_posts ( + ts DateTime, + did String, + langs Array(String), + text String +) ENGINE = MergeTree ORDER BY ts; +``` + +{/* config: clickhouse.bluesky.yml */} + +```yaml +pipeline: + batch_size: 200 + source: + type: websocket + websocket: + uri: 'wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.feed.post' + handler: + type: 'handlers.InferredMemBatch' + sql: | + SELECT + to_timestamp(time_us / 1000000) AS ts, + did, + COALESCE(commit.record.langs, []) AS langs, + COALESCE(commit.record.text, '') AS text + FROM batch + sink: + type: clickhouse + clickhouse: + dsn: {{ SQLFLOW_CLICKHOUSE_DSN }} + table: bluesky_posts +``` + +Save it as `bluesky.yml` and run it. No `SQLFLOW_KAFKA_BROKERS` here: the source is the WebSocket, and the only variable the pipeline reads is the DSN. + +```bash +docker run --rm \ + -v "$(pwd)":/conf \ + -e SQLFLOW_CLICKHOUSE_DSN \ + turbolytics/sql-flow:v2026.10.08 \ + run /conf/bluesky.yml --max-msgs=2000 +``` + +Add `--network` as in the Kafka example if ClickHouse runs in Docker on the same machine. With the binary instead of the image: + +```bash +sqlflow run bluesky.yml --max-msgs=2000 +``` + +The firehose delivers about 50 posts a second, so 2,000 messages take under a minute. {/* measured: docs/published/clickhouse/bluesky.out */} Then: + +```sql +SELECT lang, count() AS posts +FROM (SELECT arrayJoin(langs) AS lang FROM bluesky_posts) +GROUP BY lang ORDER BY posts DESC LIMIT 5; +``` + +## Type mapping {#type-mapping} + +SQLFlow hands each batch to ClickHouse as rows built from the Arrow table DuckDB produces, so what matters is the Arrow type each SQL column ends up with. Cast in the handler SQL to control it. + +Every row below is generated from the engine's type matrix and proven by an integration test that writes the value, reads it back through the server, and compares. The matrix and the test live in the [SQLFlow repository](https://github.com/turbolytics/sql-flow/blob/main/docs/coverage/clickhouse-types.mdx). A pairing in the table is proven. A pairing missing from it is untested, not refused: the refused ones are listed under [Unsupported](#unsupported) and fail the batch with a code. Parameterised column types are listed as the instance the test wrote. + +{/* generated: type-table */} + +| DuckDB SQL type (cast to) | ClickHouse column types that accept it | Notes | +| --- | --- | --- | +| `BOOLEAN` | `Bool`, `UInt8` | | +| `TINYINT` | `Int8` | | +| `SMALLINT` | `Int16` | | +| `INTEGER` | `Int32` | | +| `BIGINT` | `Int64` | | +| `UTINYINT` | `UInt8` | | +| `USMALLINT` | `UInt16` | | +| `UINTEGER` | `UInt32` | | +| `UBIGINT` | `UInt64` | | +| `REAL` | `Float32` | | +| `DOUBLE` | `Float64` | | +| `VARCHAR` | `String`, `LowCardinality(String)`, `DateTime64(3)`, `DateTime`, `Enum8('' = 0, 'a' = 1, 'b' = 2)`, `FixedString(36)`, `UUID`, `IPv4`, `Decimal(10, 2)` | the text must already be the column's text form. `UUID` and `JSON` reach the sink as `VARCHAR`. | +| `BLOB`, `BIT`, `VARINT` | `String` | | +| `DATE` | `Date`, `Date32` | | +| `TIMESTAMP_S` | `DateTime` | | +| `TIMESTAMP_MS` | `DateTime64(3)` | | +| `TIMESTAMP` | `DateTime64(6)`, `DateTime` | DateTime keeps whole seconds; DateTime64(6) keeps all six digits | +| `TIMESTAMP_NS` | `DateTime64(9)` | | +| `TIMESTAMPTZ` | `DateTime64(6)`, `DateTime` | stored as the same UTC instant; the zone is dropped and the column's own zone governs rendering; DateTime keeps whole seconds | +| `BOOLEAN[]` | `Array(Bool)` | | +| `TINYINT[]` | `Array(Int8)` | | +| `SMALLINT[]` | `Array(Int16)` | | +| `INTEGER[]` | `Array(Int32)` | | +| `BIGINT[]` | `Array(Int64)` | | +| `UTINYINT[]` | `Array(UInt8)` | | +| `USMALLINT[]` | `Array(UInt16)` | | +| `UINTEGER[]` | `Array(UInt32)` | | +| `UBIGINT[]` | `Array(UInt64)` | | +| `REAL[]` | `Array(Float32)` | | +| `DOUBLE[]` | `Array(Float64)` | | +| `VARCHAR[]` | `Array(String)` | | +| `BLOB[]` | `Array(String)` | | +| `DATE[]` | `Array(Date32)` | | +| `TIMESTAMP_S[]` | `Array(DateTime)` | | +| `TIMESTAMP_MS[]` | `Array(DateTime64(3))` | | +| `TIMESTAMP[]` | `Array(DateTime64(6))` | | +| `TIMESTAMP_NS[]` | `Array(DateTime64(9))` | | +| `TIMESTAMPTZ[]` | `Array(DateTime64(6))` | stored as the same UTC instant; the zone is dropped and the column's own zone governs rendering | +| `BIGINT[][]` | `Array(Array(Int64))` | | +| `NULL` in any column | the column's own type | the column type's zero value; the DEFAULT expression is not evaluated because every produced column is named in the INSERT. For an Enum with no 0 member that value is not a member, and the row cannot be read | +| `NULL` inside a `LIST` | the element's own type | the element type's zero value, because ClickHouse's Array(T) holds no nulls unless T is Nullable | + +{/* invariant: type.roundtrip */} {/* invariant: type.null */} {/* invariant: type.nested */} {/* invariant: type.string.fidelity */} {/* invariant: clickhouse.type_text.per_destination */} + + +**A `VARCHAR` reaches a non-`String` column as text the column parses.** The matrix proves each destination with a value already in that destination's text form, not with one value for all of them. A UUID string does not become a `DateTime`. The forms tested: + +| Destination | Text the test wrote | +| --- | --- | +| `String`, `LowCardinality(String)` | `L'Ε’il πŸ‘ "quoted" back\slash` and a tab | +| `DateTime64(3)` | `2026-09-01 12:00:00.123` | +| `DateTime` | `2026-09-01 12:00:00` | +| `Enum8(...)` | a member name, such as `a` | +| `FixedString(36)`, `UUID` | `0b7e2c9a-4d31-4f6e-9a02-7c1de3f10b45` | +| `IPv4` | `192.168.1.1` | +| `Decimal(10, 2)` | `1234.56` | + +{/* invariant: clickhouse.type_text.per_destination */} + + + +**A `NULL` into a non-`Nullable` `Enum` with no `0` member writes a row that cannot be read.** The sink stores the zero value, `0`. The `INSERT` succeeds, and every later `SELECT` of the column fails with `UNKNOWN_ELEMENT_OF_ENUM`. Measured with `Enum8('a' = 1, 'b' = 2)`. An enum column that can receive a `NULL` needs a `0` member, as in `Enum8('' = 0, 'a' = 1, 'b' = 2)`, or `Nullable(Enum8(...))`. Otherwise `coalesce` the value in the handler SQL. {/* invariant: clickhouse.enum.null_is_zero */} + + +Values are matched to columns **by name**, not by position. {/* invariant: clickhouse.insert.by_name */} + + +**Prefer `CAST(ts AS TIMESTAMP)` over passing timestamps as strings.** When SQLFlow sends a `VARCHAR` to a `DateTime` column, the Go driver parses the string, not the server. It accepts exactly two shapes: + +- `'2026-09-01 12:00:00'`, which it reads as UTC. The column's timezone does not change the instant stored: the sink parses the text and sends the instant, so a `DateTime('Asia/Tokyo')` column holds the same instant as a `DateTime('UTC')` one. +- `'2026-09-01 12:00:00 +09:00'`, whose offset is applied. That row stores `2026-09-01 03:00:00 UTC`. + +Anything else fails the batch, including ISO 8601 with a `T` separator or a `Z` suffix. `'2026-09-01T12:00:00Z'` is rejected with `parsing time "2026-09-01T12:00:00Z" as "2006-01-02 15:04:05"`. {/* invariant: clickhouse.temporal_text.two_shapes */} {/* invariant: type.timestamp.instant */} + +This differs from writing the same literal in an `INSERT` statement yourself, where ClickHouse parses the string in the target column's timezone, and a plain `DateTime` column uses the server timezone. Casting in the handler SQL sidesteps both sets of rules. + +If the field can be `NULL`, a `DateTime` column stores it as `1970-01-01 00:00:00`. Use `Nullable(DateTime)`, or `coalesce(CAST(ts AS TIMESTAMP), now()::TIMESTAMP)` in the handler SQL. {/* invariant: type.null */} + + + +`Decimal` columns accept the value as a **string**, not as a `DOUBLE`. `CAST(amount AS VARCHAR)` works; `CAST(amount AS DOUBLE)` fails the batch with `converting float64 to Decimal(10, 2) is unsupported`. {/* invariant: clickhouse.decimal.text_only */} + + +### Unsupported {#unsupported} + +These fail the batch with `user.sink.type_unsupported` rather than reaching the table. Cast or flatten them in the handler SQL. + +{/* generated: unsupported-types */} + +- `DECIMAL(p, s)`, `HUGEINT`, `UHUGEINT` +- `TIME`, `TIMETZ` +- `TIME_NS` +- `INTERVAL` +- `DECIMAL(p, s)[]`, `HUGEINT[]` +- `TIME[]` +- `STRUCT(...)[]` +- `INTEGER[3]` +- `STRUCT(...)` +- `MAP(K, V)` +- `UNION(...)` +- `ENUM` + +{/* invariant: type.undeclared.fails_loud */} + +## Known limits {#known-limits} + +- **Unsupported types:** listed under [Unsupported](#unsupported). `DECIMAL`, `STRUCT`, `MAP`, `TIME` and `INTERVAL` are the ones a handler reaches for. Each fails the batch with `user.sink.type_unsupported` before any row is written. Flatten a `STRUCT` or `MAP` into scalar columns in the handler SQL, or serialise it with `to_json()` into a `String` column. Cast a `DECIMAL` to `VARCHAR`. `Nested` columns were not tested. {/* invariant: type.undeclared.fails_loud */} +- **No table creation or schema evolution.** The table must exist, and a column the handler emits that the table lacks fails the batch (see [Troubleshooting](#troubleshooting)). Columns the handler omits take the table's `DEFAULT`. {/* invariant: clickhouse.table.not_created */} {/* invariant: clickhouse.insert.by_name */} +- **A `NULL` written to a non-`Nullable` column is stored as the type's zero value**, not as the column's `DEFAULT` expression, and no error is raised. See [Type mapping](#type-mapping) for the workaround. For an `Enum` with no `0` member that value is not a member, and the row cannot be read back. See the warning under [Type mapping](#type-mapping). {/* invariant: type.null */} {/* invariant: clickhouse.enum.null_is_zero */} +- **At-least-once delivery.** Kafka offsets are committed after the `INSERT` is acknowledged. A crash between the two replays the batch, so duplicates are possible. {/* invariant: pipeline.commit.after_flush */} + + `ReplacingMergeTree` keyed on something that identifies the row removes them **eventually**, on a background merge. Until that merge, an ordinary read sees both copies: measured on self-hosted 26.8.1, a replayed block read 4 rows, and 2 with `FINAL`. Query with `FINAL` where a read must be exact now. {/* invariant: clickhouse.replay.duplicates_until_merge */} + + Insert-time block deduplication drops a second copy only when the block is byte-identical. A replicated table, which every Cloud table is, does that by default, and a non-replicated one does it with `non_replicated_deduplication_window`. A replay cuts its batches again from the committed offset, and a batch the flush interval cut short does not come back with the same boundaries. Treat insert-time deduplication as a partial measure, not a guarantee. {/* invariant: clickhouse.replay.duplicates_until_merge */} {/* measured: docs/published/clickhouse/cloud-dedup.out */} +- **Transient failures are retried.** A refused, reset, or timed-out connection to ClickHouse is retried with exponential backoff under a deadline, configured by the sink's `retry` block (`max_attempts`, `initial_backoff_ms`, `max_backoff_ms`, `deadline_seconds`). A rejected `INSERT` is not retried once the server has seen it. {/* invariant: sink.error.classifies */} {/* feature: sink.retry */} A value the driver itself cannot encode, such as the ISO 8601 timestamp above, is not retried: it fails at once as `user.sink.encode_failed`, and the process exits with code 10. The sink also dials ClickHouse once at startup, so an unreachable host fails before any message is consumed. {/* invariant: clickhouse.encode.fails_once */} {/* invariant: sink.probe.fails_start */} +- **One `INSERT` per batch.** However many messages a batch holds, it reaches ClickHouse as one `INSERT`. Throughput is governed by `batch_size`; see below. {/* invariant: clickhouse.flush.one_insert */} +- **Streaming aggregation** (tumbling windows) is supported. A window closes on an event-time watermark the engine asserts, never on wall clock. A record's event time is the payload field the source's `event_time` names, or the Kafka timestamp when it names none. The handler cuts the bucket as `time_bucket(INTERVAL '', event_time)`. A replay running behind real time therefore publishes each bucket once, whole: 1,000 events in one past one-minute bucket, read in ten batches of 100, published one row of 1,000. {/* invariant: pipeline.window.counts_every_row */} {/* measured: docs/published/clickhouse/window-replay.out */} Set `allowed_lateness_seconds` above zero and a late row republishes its bucket as a whole value, so the target must then replace the row by key, with a `ReplacingMergeTree` read through `FINAL`. `sqlflow validate` warns about this for a ClickHouse window sink. {/* feature: config.validation */} See [Tumbling windows](https://github.com/turbolytics/sql-flow#tumbling-windows) in the README. + +## Performance {#performance} + +Measured with the example above: one million 17-column rows, one SQLFlow process, Kafka co-located with SQLFlow, two runs per row. + +| Target | Rows/second | +| ------------------------------------------ | ----------- | +| Self-hosted ClickHouse, SQLFlow and Kafka on the same Docker network, v2026.10.08 | 57,900 and 55,900 | +| ClickHouse Cloud (us-east-2), SQLFlow beside the broker, over the public internet, v2026.10.08 | 13,700 and 13,300 | + +{/* measured: docs/published/clickhouse/perf.out */} {/* measured: docs/published/clickhouse/perfcloud.out */} + +The Cloud figure is the network between the machine and the service, not SQLFlow. Over the same link in the same hour, the previous release, v2026.09.17.1, ran 12,500 and 12,900 rows a second {/* measured: docs/published/clickhouse/perfcloud.out */}, and the same pipeline against a self-hosted server on one Docker network runs four times as fast. Run SQLFlow near the service, same region and ideally same VPC, for numbers closer to the self-hosted figure. + +`batch_size` is the lever on a nearby target, measured on the same Docker network: + +| `batch_size` | Rows/second | +| ------------ | ----------- | +| 5,000 | 34,000 | +| 20,000 | 56,900 | +| 50,000 | 57,600 | +| 100,000 | 56,900 | + +{/* measured: docs/published/clickhouse/perf.out */} + +Below about 20,000 rows per batch, per-`INSERT` overhead dominates; above it there is little further to gain, and memory per batch grows. 20,000 is a good default for rows of this width; verify against your own. + +## Troubleshooting {#troubleshooting} + +### `Post "http://…:8443?…": EOF` at startup {#trouble-reset} + +The DSN uses `clickhouse://` (plain HTTP) against a TLS port, and the server closes the connection. Startup fails as `system.sink.unreachable`: `failed to query server hello: … Post "http://…:8443?…": EOF`. Change the scheme to `clickhouses://`. {/* measured: docs/published/clickhouse/cloud-reset.out */} + +### `Authentication failed: password is incorrect, or there is no user with such name` {#trouble-auth} + +The credentials in the DSN are wrong, or a special character in the password was not URL-encoded. Startup fails before any message is consumed. The HTTP status and code depend on the server and the user: Cloud answers its `default` user with `[HTTP 401] code: 194`, and self-hosted 26.8 answers a named user with `[HTTP 403] code: 516`. On Cloud, the `default` user's password is shown once at service creation and can be reset from the console. If the password is right, check the service's IP access list. {/* invariant: clickhouse.auth.fails_start */} {/* measured: docs/published/clickhouse/cloud-auth.out */} + +### `column is not present in the table ` {#trouble-column} + +The full error starts with `[system.sink.write_failed] clickhouse sink: prepare batch:`. The handler SQL produced a column the target table does not have. SQLFlow inserts by column name, so either alias the column in the SQL to match the table, drop it from the `SELECT`, or add it to the table. {/* invariant: clickhouse.insert.by_name */} + +### `clickhouse sink: column "": unsupported arrow type ` {#trouble-type} + +The column's Arrow type has no ClickHouse mapping, such as a struct or a map value. Cast or flatten in the handler SQL; see [Type mapping](#type-mapping). {/* invariant: type.undeclared.fails_loud */} + +### `converting float64 to Decimal(10, 2) is unsupported` {#trouble-decimal} + +A `DOUBLE` was sent to a `Decimal` column. Cast to a string instead: `CAST(amount AS VARCHAR) AS amount`. The driver parses the decimal text exactly, with no floating-point rounding. {/* invariant: clickhouse.decimal.text_only */} + +### `parsing time "…" as "2006-01-02 15:04:05"` {#trouble-timestamp} + +A `VARCHAR` timestamp reached a `DateTime` column in a shape the driver does not accept. It takes `'2006-01-02 15:04:05'` and the same with a trailing offset, and nothing else. ISO 8601 with a `T` separator or a `Z` suffix fails. Cast in the handler SQL with `CAST(ts AS TIMESTAMP)`, or reformat the string. See [Type mapping](#type-mapping). + +The error carries the code `user.sink.encode_failed`, and the process exits with code 10. It is not retried, because the value would fail every attempt. {/* invariant: clickhouse.temporal_text.two_shapes */} {/* invariant: clickhouse.encode.fails_once */} + +### The pipeline runs but rows never appear {#trouble-no-rows} + +Check the batching. A batch is executed and inserted when `batch_size` messages have arrived, or when `flush_interval_seconds` elapses with a partial batch waiting. Whichever comes first wins. `flush_interval_seconds` defaults to `30`, so a partial batch on a low-volume topic waits up to 30 seconds. For faster inserts on a quiet topic, lower `flush_interval_seconds`, which takes any whole number of seconds, or lower `batch_size`. With `--max-msgs`, the final partial batch is inserted on exit. {/* invariant: pipeline.flush.eventually */} {/* invariant: lifecycle.drain.on_cancel */} {/* feature: cli.invocation */} + +## Further reading {#further-reading} + +- SQLFlow on GitHub: configuration reference, all sources and sinks, error policies, windowing. +- SQLFlow documentation: tutorials, including a longer ClickHouse walkthrough. diff --git a/docs/published/clickhouse/.gitignore b/docs/published/clickhouse/.gitignore new file mode 100644 index 00000000..94a123a8 --- /dev/null +++ b/docs/published/clickhouse/.gitignore @@ -0,0 +1,5 @@ +# What the scripts here write while they run. The .out files are the evidence. +*.log +taxi_*.yml +win_events.jsonl +win_late.yml diff --git a/docs/published/clickhouse/README.md b/docs/published/clickhouse/README.md new file mode 100644 index 00000000..e16306ca --- /dev/null +++ b/docs/published/clickhouse/README.md @@ -0,0 +1,27 @@ +# Evidence for the ClickHouse page + +The measurements [`../clickhouse.mdx`](../clickhouse.mdx) cites. Its other +claims cite `clickhouse.*` invariants, proven by +`internal/sinks/clickhouse_invariants_test.go`. + +| File | What it shows | Produced by | +| --- | --- | --- | +| `selfhosted-verify.out` | The taxi example's Verify counts on self-hosted 26.8.1, and the publish time | the page's steps, on 2026-10-09 | +| `cloudverify.out` | The same counts on Cloud 26.6.1 | the page's steps, then the Verify query | +| `bluesky.out` | The WebSocket example landing 2,000 posts in 40 seconds | the page's steps, on 2026-10-09 | +| `perf.out` | Self-hosted throughput at each `batch_size` | `perf.sh` | +| `perfcloud.out` | Cloud throughput, the previous release as a control, and one count read short from a lagging replica | `perfcloud.sh` | +| `throughput-bisect.out` | Why v2026.10.08 is slower than v2026.09.17.1: the Kafka fetch defaults and jsonparser 1.3.0 | host builds of each commit | +| `window-replay.out` | 1,000 events in one past bucket, read in ten batches, published as one row | `win.sh` with `win.yml` | +| `cloud-dedup.out` | Cloud dropping an identical second block at insert | the queries in its header | +| `cloud-reset.out` | `clickhouse://` against Cloud's TLS port failing at startup | the page's run step with a plain-HTTP DSN | +| `cloud-auth.out` | Cloud's answer to a wrong password: HTTP 401, code 194 | the query in its header | + +The scripts expect the dev stack (`make start-backing-services`): Kafka as +`kafka1` and ClickHouse as `clickhouse` on the `dev_default` network. Set +`IMG` to run another image. `perfcloud.sh` reads Cloud credentials from +`~/.sqlflow-clickhouse` or the file `CLICKHOUSE_CREDS` names, and keeps the +password out of the argument list. + +A measurement names the versions it ran on. When a release changes one of +these numbers, re-run the script, replace the file, and update the page. diff --git a/docs/published/clickhouse/bluesky.out b/docs/published/clickhouse/bluesky.out new file mode 100644 index 00000000..81fea348 --- /dev/null +++ b/docs/published/clickhouse/bluesky.out @@ -0,0 +1,12 @@ +# The Bluesky WebSocket example, run as the page prints it, against self-hosted ClickHouse 26.8.1.2041. +# turbolytics/sql-flow:v2026.10.08, --max-msgs=2000, 2026-10-09. Recorded in +# docs/superpowers/status/clickhouse-117740/STATUS.md on sql-flow PR #329. +exit=0 wall=40s errors=0 +count() 2000 +β”Œβ”€lang─┬─posts─┐ +β”‚ en β”‚ 1070 β”‚ +β”‚ ja β”‚ 236 β”‚ +β”‚ es β”‚ 63 β”‚ +β”‚ de β”‚ 53 β”‚ +β”‚ pt β”‚ 50 β”‚ +β””β”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β”€β”˜ diff --git a/docs/published/clickhouse/cloud-auth.out b/docs/published/clickhouse/cloud-auth.out new file mode 100644 index 00000000..c03ab9f2 --- /dev/null +++ b/docs/published/clickhouse/cloud-auth.out @@ -0,0 +1,14 @@ +# A wrong password against ClickHouse Cloud, 2026-10-11, user default, password 'wrong' +# server version: 26.6.1.2326 +Code: 194. DB::Exception: default: Authentication failed: password is incorrect, or there is no user with such name + +If you use ClickHouse Cloud, the password can be reset at https://clickhouse.cloud/ +on the settings page for the corresponding service. + +If you have installed ClickHouse and forgot password you can reset it in the configuration file. +The password for default user is typically located at /etc/clickhouse-server/users.d/default-password.xml +and deleting this file will reset the password. +See also /etc/clickhouse-server/users.xml on the server where ClickHouse is installed. + +. (REQUIRED_PASSWORD) +HTTP 401 diff --git a/docs/published/clickhouse/cloud-dedup.out b/docs/published/clickhouse/cloud-dedup.out new file mode 100644 index 00000000..9680cbec --- /dev/null +++ b/docs/published/clickhouse/cloud-dedup.out @@ -0,0 +1,8 @@ +# The same block inserted twice into a plain MergeTree on ClickHouse Cloud, 2026-10-11 +# server version: 26.6.1.2326 +# CREATE TABLE default.dedup_probe_1791713956 (id Int32, v String) ENGINE = MergeTree ORDER BY id +engine: SharedMergeTree +replicated_deduplication_window: 10000 +insert 1 done +insert 2 done +count(): 2 diff --git a/docs/published/clickhouse/cloud-reset.out b/docs/published/clickhouse/cloud-reset.out new file mode 100644 index 00000000..ac904f81 --- /dev/null +++ b/docs/published/clickhouse/cloud-reset.out @@ -0,0 +1,4 @@ +# A plain-HTTP clickhouse:// DSN against ClickHouse Cloud's TLS port 8443, 2026-10-11 +# turbolytics/sql-flow:v2026.10.08, Cloud 26.6.1.2326. Host and password redacted. +exit=12 +Error: [system.sink.unreachable] sink: the destination could not be reached at startup: failed to query server hello: failed to query server hello info: sendQuery: Post "http://default:***@:8443?client_protocol_version=54460&database=default&default_format=Native&max_execution_time=34": EOF diff --git a/docs/published/clickhouse/cloudverify.out b/docs/published/clickhouse/cloudverify.out new file mode 100644 index 00000000..95e07359 --- /dev/null +++ b/docs/published/clickhouse/cloudverify.out @@ -0,0 +1,8 @@ +# Verify query on Cloud 26.6.1.2326 + β”Œβ”€payment_type─┬──trips─┬─avg_total─┐ +1. β”‚ CSH β”‚ 617237 β”‚ 17.82 β”‚ +2. β”‚ CRE β”‚ 377897 β”‚ 13.91 β”‚ +3. β”‚ NOC β”‚ 3573 β”‚ 13.89 β”‚ +4. β”‚ DIS β”‚ 1293 β”‚ 12.09 β”‚ + β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ +fffd: 0 diff --git a/docs/published/clickhouse/perf.out b/docs/published/clickhouse/perf.out new file mode 100644 index 00000000..fd03ee8f --- /dev/null +++ b/docs/published/clickhouse/perf.out @@ -0,0 +1,17 @@ +# v2026.10.08 unless a comment names another image; perf.sh