> ## Documentation Index
> Fetch the complete documentation index at: https://private-7c7dfe99-vortex-format.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.
> Stream Kafka, WebSocket and webhook data into ClickHouse, transformed with DuckDB SQL, using SQLFlow
# SQLFlow and ClickHouse
export const CommunityMaintainedBadge = () => {
return ;
};
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
* 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
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 |
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.
Connect to ClickHouse Cloud
To connect to ClickHouse with HTTP(S) you need this information:
| Parameter(s) | Description |
| - | - |
| `HOST` and `PORT` | Typically, the port is 8443 when using TLS or 8123 when not using TLS. |
| `DATABASE NAME` | Out of the box, there is a database named `default`, use the name of the database that you want to connect to. |
| `USERNAME` and `PASSWORD` | Out of the box, the username is `default`. Use the username appropriate for your use case. |
The details for your ClickHouse Cloud service are available in the ClickHouse Cloud console.
Select a service and click **Connect**:
Choose **HTTPS**. Connection details are displayed in an example `curl` command.
If you're using self-managed ClickHouse, the connection details are set by your ClickHouse administrator.
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 theme={null}
sink:
type: clickhouse
clickhouse:
dsn: {{ SQLFLOW_CLICKHOUSE_DSN }}
table: trips_small
```
```bash theme={null}
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, and the connection is reset before any request is made. See [Troubleshooting](#troubleshooting).
Connect to self-hosted ClickHouse
Point the DSN at the HTTP interface, port 8123 by default:
```bash theme={null}
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
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.
* **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.
* **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.
Example: streaming the NYC taxi dataset from Kafka
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
The same table the sample-dataset page defines:
```sql theme={null}
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
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 theme={null}
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 theme={null}
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 theme={null}
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:
```bash theme={null}
python publish.py
```
3. Write the 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)).
```yaml theme={null}
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`. See [Performance](#performance) for why 20,000.
4. Run it
```bash theme={null}
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 theme={null}
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.
5. Verify
```sql theme={null}
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. 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.
A second example: WebSocket source, no Kafka
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 theme={null}
CREATE TABLE bluesky_posts (
ts DateTime,
did String,
langs Array(String),
text String
) ENGINE = MergeTree ORDER BY ts;
```
```yaml theme={null}
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 theme={null}
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 theme={null}
sqlflow run bluesky.yml --max-msgs=2000
```
The firehose delivers about 50 posts a second, so 2,000 messages take under a minute. Then:
```sql theme={null}
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
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.
| 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 string must already be the destination's text form; see the note below. `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. `Enum` is the exception: see the note below |
| `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 |
**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` |
**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.
Values are matched to columns **by name**, not by position.
**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 server's timezone and the column's timezone do not change the instant stored. Measured against a server running `Asia/Tokyo`: the row landed on `2026-09-01 12:00:00 UTC`, the same instant a `DateTime('UTC')` column received.
* `'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"`.
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.
`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`.
Unsupported
These fail the batch with `user.sink.type_unsupported` rather than reaching the table. Cast or flatten them in the handler SQL.
* `DECIMAL(p, s)`, `HUGEINT`, `UHUGEINT`
* `TIME`, `TIMETZ`
* `TIME_NS`
* `INTERVAL`
* `DECIMAL(p, s)[]`, `HUGEINT[]`
* `TIME[]`
* `STRUCT(...)[]`
* `INTEGER[3]`
* `STRUCT(...)`
* `MAP(K, V)`
* `UNION(...)`
* `ENUM`
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.
* **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`.
* **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).
* **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.
`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.
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.
* **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. 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.
* **One connection, one `INSERT` per batch.** There is no parallel writer. Throughput is governed by `batch_size`; see below.
* **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. 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`. See [Tumbling windows](https://github.com/turbolytics/sql-flow#tumbling-windows) in the README.
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 |
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, 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 |
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
`connection reset by peer` immediately, with `...:8443?...` in the message
The DSN uses `clickhouse://` (plain HTTP) against a TLS port. The full error looks like `failed to query server hello: … Post "http://…:8443?…": … connection reset by peer`. Change the scheme to `clickhouses://`.
`[HTTP 401] code: 194 … Authentication failed: password is incorrect, or there is no user with such name`
The credentials in the DSN are wrong, or a special character in the password was not URL-encoded. 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.
`column is not present in the table `
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.
`clickhouse sink: column "": unsupported arrow 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).
`converting float64 to Decimal(10, 2) is unsupported`
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.
`parsing time "…" as "2006-01-02 15:04:05"`
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.
The pipeline runs but rows never appear
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.
Further reading
* SQLFlow on GitHub: configuration reference, all sources and sinks, error policies, windowing.
* SQLFlow documentation: tutorials, including a longer ClickHouse walkthrough.