Crate spate_clickhouse

Expand description

ClickHouse sink for the Spate framework.

Writes directly to shard-local tables through the official clickhouse crate: rows are encoded to RowBinary on the pipeline threads (this crate’s own serializer), shipped as pre-formatted frames, and inserted one INSERT ... FORMAT RowBinary per sealed batch with a deterministic insert_deduplication_token. Direct-to-shard writes beat Distributed-table inserts for an ETL writer: bigger blocks, less merge pressure, and — crucially for checkpointing — a synchronous server acknowledgement (wait_end_of_query=1; write_batch returning Ok covers materialized views too).

§⚠ Deduplication requires a window — silently off on plain MergeTree

Retries reuse the batch’s deduplication token, making them idempotent — but only if the server keeps a deduplication window:

Replicated*MergeTree: deduplication is on by default (replicated_deduplication_window = 10000).
Plain MergeTree: the window defaults to 0 and token deduplication silently does nothing. Set it explicitly:

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

§At-least-once, honestly

Tokens cover same-batch retries (including on another replica after a timeout). They do not cover crash replay: after a restart, data is re-batched with different boundaries and different tokens, and replayed rows will land again. Design target tables to tolerate duplicates — ReplacingMergeTree with a version column is the sanctioned pattern.

Re-homing is a different hazard entirely: a key that moves to another shard — changed shard weights, a changed sharding key, a changed shard count — leaves its already-written rows behind on the old shard, and nothing collapses them there. Dedup tokens are shard-scoped, and ReplacingMergeTree merges within one shard’s local table only, so the old copies persist indefinitely: unpruned reads through a Distributed table return both homes’ rows, while pruned reads return only the new home’s. Treat a re-homing change as a planned migration — backfill or delete the moved keys’ old-shard rows — not as replay noise that settles. A plain restart with unchanged topology is the benign case: replayed rows land on the same shard, where the patterns above apply.

§Shard-affinity routing (Distributed parity)

DistributedRouter (built via ClickHouseSink::router) places each record on the same shard a ClickHouse Distributed table with sharding expression xxHash64(<key column>) would — so SELECTs through the Distributed table can prune shards (optimize_skip_unused_shards=1) while inserts keep going direct-to-local. The ordering contract is the operator’s: shards[i] in the sink config must be the cluster’s shard_num = i + 1 (the remote_servers order), with matching weights. The opt-in distributed_check config block verifies shard count, weights, and the DDL’s sharding expression at startup (ClickHouseSink::validate_distributed); the ordering itself it can only warn about. See router for the algorithm.

§Column order is the wire contract

RowBinary carries no column names. The configured columns list and the row struct’s field declaration order must match; reordering either is a breaking change to the pipeline. See rowbinary for the full type mapping.

§Wiring

sink:
  clickhouse:
    table: orders_local
    columns: [id, name, amount]
    shards:
      - replicas: ["http://ch-0-0:8123", "http://ch-0-1:8123"]
      - replicas: ["http://ch-1-0:8123", "http://ch-1-1:8123"]
    # Only when inserting into JSON columns (as String fields of JSON
    # text — see the rowbinary type table):
    settings: { input_format_binary_read_json_as_string: "1" }

config::from_component_config turns that section into a ClickHouseWriter (the framework’s ShardWriter), per-shard ClickHouseEndpoints, and the sink-pool configuration; ClickHouseEncoder is the matching RowEncoder, parameterized by a record family: owned rows use Owned<Row>, and the encode path itself needs only Row: serde::Serialize.

§Wire format

The default is row-wise RowBinary (rowbinary). An opt-in columnar Native encoder (native, format: native) transposes each chunk into one self-describing block: it costs more client CPU but cuts server parse CPU and compressed wire size substantially — build it from a fetched schema via ClickHouseSink::native_schema and NativeEncoder.

Re-exports§

pub use config::ClickHouseSink;
pub use config::ClickHouseSinkConfig;
pub use config::Compression;
pub use config::DistributedCheckSection;
pub use config::Format;
pub use config::SchemaValidation;
pub use config::from_component_config;
pub use native::NativeEncoder;
pub use native::NativeError;
pub use native::NativeSchema;
pub use router::DistributedRouter;
pub use router::KeyExtractor;
pub use router::ShardKey;
pub use rowbinary::RowBinaryError;
pub use rowbinary::serialize_row;

Modules§

config: Opaque-section configuration and the sink factory.
native: ClickHouse Native (columnar, block-framed) encoder.
router: Distributed-parity shard routing: place each record on the same shard a ClickHouse Distributed table with sharding expression xxHash64(<key column>) would.
rowbinary: A serde Serializer producing ClickHouse RowBinary.
serde: Field-attribute serde modules for ClickHouse column types, mirroring the clickhouse crate’s clickhouse::serde::* modules.

Structs§

ClickHouseEncoder: Encodes a record family’s Serialize rows into RowBinary via this crate’s serializer. Runs inside the chain’s terminal stage on pinned pipeline threads; sink workers ship the resulting frames as-is.
ClickHouseEndpoint: One connected ClickHouse replica. Wraps a clickhouse::Client (its own hyper connection pool) — the crate type stays private so its 0.x breaking releases never become ours.
ClickHouseWriter: Writes sealed batches: one INSERT ... FORMAT <fmt> per batch (the format — RowBinary or Native — is baked into insert_sql), carrying the batch’s deduplication token so retries — including on other replicas — are idempotent within the server’s dedup window. The transport is format-agnostic: frames concatenate to the request body (RowBinary rows or a stream of Native blocks). write_batch returning Ok is the durable-ack point (the server confirmed the insert, materialized views included, thanks to wait_end_of_query=1).
Date32Days: Days since the Unix epoch (signed), matching a Date32 column’s wire representation (Int32; the server accepts 1900-01-01 through 2299-12-31).
DateDays: Days since the Unix epoch, matching a Date column’s wire representation (UInt16, so 1970-01-01 through 2149-06-06).
DateTime64Micros: Microseconds since the Unix epoch, matching a DateTime64(6) column’s wire representation (Int64).
DateTime64Millis: Milliseconds since the Unix epoch, matching a DateTime64(3) column’s wire representation (Int64).
DateTime64Nanos: Nanoseconds since the Unix epoch, matching a DateTime64(9) column’s wire representation (Int64).
DateTime64Secs: Seconds since the Unix epoch, matching a DateTime64(0) column’s wire representation (Int64).
DateTimeSeconds: Seconds since the Unix epoch, matching a DateTime column’s wire representation (UInt32).
Decimal32: A pre-scaled Decimal32(S) / Decimal(P ≤ 9, S) value: the inner i32 is value × 10^SCALE, written little-endian.
Decimal64: A pre-scaled Decimal64(S) / Decimal(P ≤ 18, S) value: the inner i64 is value × 10^SCALE, written little-endian.
Decimal128: A pre-scaled Decimal128(S) / Decimal(P ≤ 38, S) value: the inner i128 is value × 10^SCALE, written little-endian.
Int256: An Int256 column value: 32 bytes, little-endian, two’s complement.
PreEncodedRows: Passthrough for records that are already RowBinary-encoded rows (Vec<u8> payloads): appends the bytes verbatim. For pipelines that encode upstream or replicate pre-encoded data. One record must be exactly one encoded row — the framework counts rows by records.
RowSchema: The validated, config-ordered expected schema. Produced by crate::ClickHouseSink::validate_schema; consumed by crate::ClickHouseEncoder::with_schema for the first-record check. Opaque: holds no clickhouse crate types.
Time64Micros: Microseconds, matching a Time64(6) column’s wire representation (Int64).
Time64Millis: Milliseconds, matching a Time64(3) column’s wire representation (Int64).
Time64Nanos: Nanoseconds, matching a Time64(9) column’s wire representation (Int64).
Time64Secs: Seconds, matching a Time64(0) column’s wire representation (Int64).
TimeSeconds: Seconds, matching a Time column’s wire representation (Int32; the server accepts -999:59:59 through 999:59:59).
UInt256: A UInt256 column value: 32 bytes, little-endian.

Enums§

DistributedCheckError: Startup DDL-parity guard failed. Fatal: fix the config or the DDL.
SchemaError: Schema validation failed at sink startup.

Type Aliases§

LineString: A LineString column: an open sequence of points (Array(Point)).
MultiLineString: A MultiLineString column: Array(LineString).
MultiPolygon: A MultiPolygon column: Array(Polygon).
Point: A Point column: (x, y) as two Float64s.
Polygon: A Polygon column: an outer ring plus hole rings (Array(Ring)).
Ring: A Ring column: a closed sequence of points (Array(Point)).