Skip to main content

Schema validation

A ClickHouse pipeline has three descriptions of the same row: the columns list in your YAML, the field order of your Rust row struct, and the live table's schema. RowBinary — the wire format — carries no column names, so nothing catches a disagreement at the protocol level: misaligned columns are silent data corruption, not an error. Opt-in schema validation closes that gap at startup and at the first record, before any misaligned batch can be sent.

The wire contract​

The sink sends INSERT INTO t (col1, col2, ...) FORMAT RowBinary with rows your struct serialized field by field, in declaration order. The server maps bytes to columns purely by position:

/// Field order here MUST match `columns: [id, customer, amount_cents, ts_ms]`.
#[derive(Serialize)]
struct Order {
    id: u64,
    customer: String,
    amount_cents: i64,
    ts_ms: i64,
}

Reordering either the struct fields or the columns list is a breaking change to the pipeline. The type mapping lives in the etl-clickhouse crate's rowbinary module docs.

The three modes​

validate_schema in the sink: { clickhouse: ... } section:

ModeStartup (before any thread spawns)First record (per pipeline thread)
off (default)Nothing — no queries issued.Nothing.
namesEvery configured column exists and is insertable on every replica of every shard; replica drift and missing tables fail with a readable diff.Struct field names and order match the configured columns.
fullEverything names does.names plus a class-based type-compatibility check per position.

The full type check is deliberately permissive: a u32 may feed UInt32, DateTime, or IPv4; server types the client does not know always pass. One mismatch always fails regardless: Nullable on the server versus a non-Option field (or the reverse) — that one is wire corruption, not a judgment call.

Wiring it up​

Validation is an async pre-flight step you run on the builder's I/O runtime via pipeline.block_on, after building the sink and before handing it to .sink(...). A failure exits the process before any pipeline thread or sink worker exists. From crates/etl/examples/kafka_avro_to_clickhouse.rs:

let sink = etl::clickhouse::config::from_component_config(&pipeline.config().sink)?;

// `off` returns Ok(None) instantly and issues no queries.
let encoder = match pipeline.block_on(sink.validate_schema())? {
    Some(schema) => ClickHouseEncoder::<Owned<Order>>::with_schema(schema),
    None => ClickHouseEncoder::<Owned<Order>>::new(),
};

let report = pipeline.sink(sink)?/* .chains(...).run(...) */;

with the matching YAML:

sink:
  clickhouse:
    table: orders
    columns: [id, customer, amount_cents, ts_ms]
    validate_schema: full     # off | names | full

The two halves, and why there are two​

Startup half — ClickHouseSink::validate_schema() fetches system.columns from every replica and checks the configured columns against the live tables. It returns the parsed schema as an Arc of RowSchema (or None when off).

First-record half — ClickHouseEncoder::with_schema(schema) carries that expected schema into the encoder. The row struct only exists as serde behavior, so it cannot be inspected at startup; instead the encoder checks the first record each pipeline thread encodes: field names and order always, type classes in full mode. A mismatch is a fatal error that stops the pipeline before the batch is sent. Steady-state cost after the first record is one predictable branch — nothing on the per-record hot path.

Together the two halves pin all three descriptions to each other: columns ↔ live table at startup, struct ↔ columns at the first record.

[!NOTE] Validation checks every replica independently, so it also catches replica drift — a replica whose table was altered out of step with the others.

[!WARNING] validate_schema: off preserves the historical behavior: nothing is checked, and a wrong column order writes garbage that ClickHouse may happily accept. Turn on at least names for any table you did not create in the same commit as the pipeline.