ClickHouse Sink

Streams CDC events into ClickHouse over the HTTP interface using the compact RowBinary format. It supports two consumption shapes with one uniform write path — the target table’s engine decides whether you get a change-log or a mirrored current-state table.

sinks:
  - type: clickhouse
    config:
      id: ch-orders
      url: "https://clickhouse:8443"     # HTTP(S) endpoint
      database: analytics
      table: orders
      mode: upsert                       # upsert | changelog
      user: default
      password: "${CLICKHOUSE_PASSWORD}" # ${ENV} expansion

Configuration

Modes

The sink always writes the same row shape — your source columns plus the meta columns _op, _version, _deleted, _source_ts. What differs is the target table’s engine, which the sink auto-creates for you (or you pre-create it and set auto_create: false).

upsert — mirror current state

CREATE TABLE analytics.orders
( id Int64, amount Decimal(12,2),
  _op LowCardinality(String), _version UInt64, _deleted UInt8,
  _source_ts DateTime64(3) )
ENGINE = ReplacingMergeTree(_version, _deleted)
ORDER BY id;                        -- ORDER BY = source primary key

ClickHouse’s merge keeps the row with the max _version per key and drops deletes. Query current state with FINAL:

SELECT * FROM analytics.orders FINAL;

changelog — retain every change

Same columns, ENGINE = MergeTree ORDER BY (id, _version). Every insert/update/ delete is kept as a row (audit / streaming). Reconstruct current state at read time:

SELECT argMax(amount, _version) FROM analytics.orders
WHERE _deleted = 0 GROUP BY id;

Delivery guarantees

At-least-once. The checkpoint advances only after an insert acks, so no events are lost; a crash between ack and checkpoint can replay a batch. On top of that:

• upsert mode is idempotent — duplicate or out-of-order inserts collapse by (key, _version), so the mirrored table converges to exactly the source’s current state regardless of retries. Read with FINAL for a consistent view. This is the recommended mode when correctness matters most.
• changelog mode is at-least-once — retries are suppressed by ClickHouse’s insert_deduplication_token for simple replays (within the dedup window; Replicated tables, or non-replicated with non_replicated_deduplication_window), but that is best-effort. Dedup at read time via argMax(_version) for exact current state.

End-to-end exactly-once (Kafka-EOS style) is not claimed — ClickHouse inserts are not transactional with the DeltaForge checkpoint.

_version defaults to the source LSN/binlog position (monotonic), which is what makes ReplacingMergeTree replacement correct. Only fall back to ts_ms when the source lacks a usable position (millisecond ties per key are undefined).

Requirements

The sink needs the source table’s column types + primary key (RowBinary is positional and auto-create derives the DDL from them). It loads these on demand the first time it sees a table: the schema resolver queries the source database’s catalog (MySQL INFORMATION_SCHEMA, Postgres pg_catalog) and caches the result. So the sink works under any snapshot mode, including snapshot: never — no snapshot or schema sensing is required. The only prerequisite is that the source database is reachable and the table exists, which is already true for CDC. The v1 sink targets a single table; capture one source table per ClickHouse sink (not a wildcard).

Auto table creation

By default the sink creates the target table on the first event, deriving column types from the source DDL and the engine/ORDER BY from the mode + primary key. Set auto_create: false to require a pre-created table (locked-down setups).

Type mapping

Nullable source columns map to Nullable(T). JSON is stored as text in v1.