ClickHouse sink

The ClickHouse sink (etl-clickhouse, feature clickhouse on the etl facade) writes directly to shard-local tables over HTTP: rows are encoded to RowBinary on the pipeline threads, shipped to sink workers as pre-formatted frames, and inserted one INSERT ... FORMAT RowBinary per sealed batch with a deterministic insert_deduplication_token.

Construct it from the pipeline's opaque section; the result is a SinkBundle that drops straight into the builder:

let sink = etl::clickhouse::config::from_component_config(pipeline.config().sink_config("default")?)?;
// optional pre-flight — see the schema validation guide
let schema = pipeline.block_on(sink.validate_schema())?;
let pipeline = pipeline.sink(sink)?;

To write to several tables from one pipeline (one per record kind), see multi-table split.

Configuration

The sink: { clickhouse: ... } section deserializes into ClickHouseSinkConfig (crates/etl-clickhouse/src/config.rs); unknown fields are rejected with the offending key, and the retry/breaker values are validated at load (a bad multiplier or zero threshold fails startup instead of the write loop).

sink:
  clickhouse:
    table: orders_local            # or db.orders_local
    columns: [id, name, amount]    # MUST match the row struct's field order
    shards:
      - replicas: ["http://ch-0-0:8123", "http://ch-0-1:8123"]
      - replicas: ["http://ch-1-0:8123", "http://ch-1-1:8123"]
    user: default
    password: ${CLICKHOUSE_PASSWORD}
    batch: { max_rows: 500000, max_bytes: 128MiB, linger: 1s }
    inflight: { max_per_shard: 2 }
    compression: lz4               # off | lz4 | zstd | zstd:<1-22>
    validate_schema: full          # off | names | full

Key	Type	Default	Meaning
table	string	required	Target table, optionally database.table-qualified.
columns	string list	required	Insert column list. Order is the wire contract — must match the row struct's field declaration order. Duplicates rejected at load.
shards	list of { replicas: [url] }	required	One entry per shard, each with its replica HTTP(S) URLs.
database	string	none	Default database for unqualified tables.
user	string	none	Username (interpolate secrets via ${VAR}).
password	string	none	Password (interpolate secrets via ${VAR}).
settings	string map	{}	Extra per-insert ClickHouse settings. insert_deduplication_token, insert_deduplicate, and wait_end_of_query are managed by the sink and cannot be overridden.
batch / inflight / retry / breaker	sections	framework defaults	Shared sink-pool tuning (batching, in-flight limit, retry backoff, circuit breaker). Keys and defaults are in the configuration reference; the breaker is described under Replica rotation below.
timeouts.send	duration	30s	ClickHouse-only. Per-send timeout (one frame reaching the socket); configures the HTTP client, not the shared pool.
timeouts.end	duration	180s	End-to-end insert timeout: the server fully processing the insert, materialized views included.
compression	string	lz4	Transport (HTTP-body) compression: off/none, lz4, zstd (level 3), or zstd:N with N in 1..22. Unrelated to on-disk column codecs.
validate_schema	string	off	Startup + first-record schema validation: off, names, full. See Schema validation.
format	string	rowbinary	Insert wire format: rowbinary (row-wise, default) or native (columnar blocks). See Native format.

Security (TLS)

No feature flag is needed for TLS: point a replica at an https:// URL and the connection is encrypted (rustls is always compiled in), with user/password authenticating over it. Keep the password out of the file with ${VAR} interpolation.

sink:
  clickhouse:
    shards:
      - replicas: ["https://ch-1.internal:8443", "https://ch-2.internal:8443"]
    user: etl
    password: "${CLICKHOUSE_PASSWORD}"

Client-certificate (mTLS) authentication is not yet supported — a tracked follow-up. For the framework-wide view of transport security across connectors, see Securing connections.

RowBinary on pipeline threads

Encoding is CPU work, so it happens on the pinned pipeline threads — ClickHouseEncoder (the sink's RowEncoder) serializes each row to RowBinary using the crate's own serializer. Its type parameter is the record family, not the row type: owned rows are wrapped as Owned<Row> (e.g. ClickHouseEncoder::<Owned<Order>>::new()), while the encode path itself needs only Row: serde::Serialize. Sink workers on the I/O runtime only merge frames, seal batches, and drive inserts; they never touch a record.

info

RowBinary carries no column names — position is everything. The configured columns list and the row struct's field declaration order must match; reordering either is a breaking change to the pipeline. validate_schema: names (or full) plus ClickHouseEncoder::with_schema turns this contract into a fail-fast check — see Schema validation. The full type mapping is in the crate's rowbinary module docs.

Deduplication tokens — and the window warning

Every sealed batch carries a deterministic insert_deduplication_token, so a retry of the same batch (including on another replica after a timeout) is idempotent — the server drops the duplicate insert.

warning

Deduplication is silently off on plain MergeTree. Token deduplication only works if the server keeps a deduplication window. Replicated*MergeTree defaults to a window of 10000; plain MergeTree defaults to 0, and the token does nothing — no error, no warning, duplicates on every retry. Set it explicitly:

CREATE TABLE orders (...) ENGINE = MergeTree ORDER BY id
SETTINGS non_replicated_deduplication_window = 100;

And the honest limit: tokens cover same-batch retries only. Crash replay re-batches with different boundaries and different tokens, so replayed rows land again. Design target tables to tolerate duplicates — ReplacingMergeTree with a version column is the sanctioned pattern. See Delivery guarantees.

Replica rotation and the circuit breaker

Each shard has one worker that seals batches and dispatches up to inflight.max_per_shard concurrent writes, rotating round-robin across that shard's healthy replicas. A failed write retries the same sealed batch (same token) on the next healthy replica with capped exponential backoff. Per-replica circuit breakers quarantine a replica after failure_threshold consecutive failures for open_for, then admit half_open_probes trial writes before closing again. Per-shard workers (rather than per-replica) keep batches full-sized — ClickHouse wants few big inserts — while max_per_shard still provides replica parallelism.

Durability and replica loss

The sink writes to one healthy replica per shard and relies on ReplicatedMergeTree (via ClickHouse Keeper) to replicate the part to the shard's other replicas and to heal a replica that was down when it returns. This is the client-side equivalent of internal_replication=true: the sink never fans out a block to several replicas as a delivery strategy (that is the internal_replication=false footgun — replicas drift). Cross-replica retries are the one deliberate exception — a failed write re-sends the same sealed batch to the next healthy replica, and the deduplication token makes an ambiguous double-landing a server-side no-op rather than a duplicate.

Replication is asynchronous by default (insert_quorum=0), which is the right posture for high-throughput streams. The honest boundary this creates: a batch acknowledged by one replica that is destroyed before the part replicates is lost. That window — one confirming replica, no quorum — is the explicit edge of the sink's at-least-once guarantee. Pipelines that cannot tolerate it can require a write quorum by passing insert_quorum (and, if wanted, insert_quorum_parallel / insert_quorum_timeout) through the sink's settings passthrough — these are not among the three reserved settings (insert_deduplicate, wait_end_of_query, insert_deduplication_token), so they work today without framework changes. Quorum composes correctly with the deduplication the sink always enables: when a retried batch is recognised as a duplicate of an already-committed part, ClickHouse still waits for quorum on that existing part before acknowledging, and throws UNKNOWN_STATUS_OF_INSERT if it cannot verify — a deduplicated retry cannot silently bypass the quorum (verified against the ReplicatedMergeTreeSink implementation in ClickHouse 24.8 LTS and current master; treat older servers with suspicion). The cost is availability: a quorum insert fails with TOO_FEW_LIVE_REPLICAS when live replicas drop below the quorum, so a single replica outage can block writes to that shard. The framework itself stays async and takes no position on the tradeoff.

When no replica of a shard is circuit-closed, write attempts park until the next half-open probe window (each probe is a real batch write, retried every open_for), in-flight slots stay held, and the shard's queue fills, so backpressure reaches the source and the source watermark stalls rather than losing or misrouting data — records are never rerouted to another shard (that would break the record→shard placement and the dedup-token scheme). With the default unbounded retry.max_attempts this holds indefinitely; with a bounded one, exhausted batches fail their acks and replay after restart instead — the watermark stalls either way. Recovery is automatic once any probe succeeds. This whole-shard condition is surfaced distinctly as etl_sink_shard_healthy == 0 (with per-replica etl_sink_replica_errors_total to attribute the failures); see docs/METRICS.md.

Why direct-to-shard, not a Distributed table

Writes go to shard-local tables (with internal_replication=true topology) rather than through a Distributed table: bigger blocks, less merge pressure, and — crucially for checkpointing — a synchronous server acknowledgement. The sink always sets wait_end_of_query=1, so a successful write means the server fully processed the insert, materialized views included; that success is the durable-ack point that lets watermarks advance. Sharding across shards is the pipeline's job (the chain's router, e.g. KeyHashRouter), which also makes batch boundaries deterministic for the dedup tokens.

SinkBundle and the readiness probe

ClickHouseSink implements SinkBundle, so pipeline.sink(sink) derives everything: per-shard queues, per-shard metrics labeled by replica URL, the workers, the drain hook — and a readiness probe (probe_fn) over every replica of every shard that drives the sinks-connected half of /readyz. The probe deliberately uses its own client set: sharing the insert clients would report the write path healthy merely because probing keeps its connections warm. Manual assemblies hand probe_fn() to SinkRuntime.probe directly (see Manual assembly).

Related

• Multi-table split — write to one table per record kind from a single pipeline (the sinks: map + the split terminal).
• Native format — the opt-in columnar format: native: when to use it, the supported type matrix, and the measured trade-off (server CPU/wire vs client CPU).
• Performance tuning — batch/part size, writer parallelism, async_insert, and the reasoning behind each starting-point setting.
• Distributed parity — match the router's shard placement to a Distributed table's sharding key so cluster reads can prune shards, plus the opt-in startup check that keeps the two aligned.
• Permissions — the ClickHouse GRANTs the sink machinery needs (minimal INSERT, plus the conditional system-table reads).
• Schema validation — the validate_schema modes end to end.
• Tuning — batch sizing and its coupling to backpressure.max_inflight_bytes (the sizing rule).
• Graceful shutdown — what happens to in-flight batches at the drain deadline.