spate_clickhouse/schema/
mod.rs

1//! Opt-in startup schema validation: fetch the target table's columns
2//! from every replica, fail fast (with a readable diff) when the
3//! configured columns cannot work, and hand the encoder a parsed,
4//! config-ordered schema for its first-record struct check.
5//!
6//! Two moments, two checks:
7//!
8//! - **Startup** (`validate`): the config's `columns` against
9//!   `system.columns` on every replica of every shard — missing columns,
10//!   non-insertable (MATERIALIZED/ALIAS) columns, inter-replica drift.
11//!   Config order differing from *table* order is deliberately not an
12//!   error: the `INSERT` column list maps by name, so only the struct
13//!   must agree with the config.
14//! - **First record** (`check_first_record`, driven by both encoders —
15//!   RowBinary's [`crate::ClickHouseEncoder::with_schema`] and the Native
16//!   encoder whenever its schema was fetched): the row struct's probed
17//!   field names, order, and — in `full` mode — type classes against the
18//!   configured columns, including wire-wrapper scale vs the column's
19//!   declared precision (`DateTime64Millis` into `DateTime64(6)` fails
20//!   here). This is where the positional wire contract is actually
21//!   enforced.
22
23pub(crate) mod probe;
24pub(crate) mod typeparse;
25
26use crate::config::SchemaValidation;
27use crate::writer::ClickHouseEndpoint;
28use probe::{FieldShape, Shape, compatible};
29use serde::Deserialize;
30use std::fmt::Write as _;
31use std::sync::Arc;
32use typeparse::ChType;
33
34/// Schema validation failed at sink startup.
35#[derive(Debug, thiserror::Error)]
36#[non_exhaustive]
37pub enum SchemaError {
38    /// A replica's schema could not be fetched. The user opted into
39    /// fail-fast validation, so connectivity problems fail startup too —
40    /// the message keeps "could not fetch" and "mismatch" distinguishable
41    /// at a glance.
42    #[error("sink.clickhouse: could not fetch schema for {table} from {url}: {reason}")]
43    Fetch {
44        /// The (possibly database-qualified) table.
45        table: String,
46        /// The replica URL.
47        url: String,
48        /// Human-readable cause.
49        reason: String,
50    },
51    /// The configuration does not match the live table (pre-formatted
52    /// multi-line diff).
53    #[error("{0}")]
54    Mismatch(String),
55}
56
57/// The validated, config-ordered expected schema. Produced by
58/// [`crate::ClickHouseSink::validate_schema`]; consumed by
59/// [`crate::ClickHouseEncoder::with_schema`] for the first-record check.
60/// Opaque: holds no `clickhouse` crate types.
61#[derive(Debug)]
62pub struct RowSchema {
63    pub(crate) mode: SchemaValidation,
64    pub(crate) table: String,
65    /// `(column name, parsed type, raw type string)` in **config** order.
66    pub(crate) columns: Vec<(String, ChType, String)>,
67}
68
69/// What `build()` captures for a later `validate_schema()` call.
70#[derive(Clone, Debug)]
71pub(crate) struct SchemaCheck {
72    pub(crate) mode: SchemaValidation,
73    pub(crate) database: Option<String>,
74    pub(crate) table: String,
75    pub(crate) columns: Vec<String>,
76}
77
78/// One row of `system.columns`, crate-private.
79#[derive(Clone, Debug, PartialEq, Eq, clickhouse::Row, Deserialize)]
80pub(crate) struct ColumnRow {
81    pub(crate) name: String,
82    #[serde(rename = "type")]
83    pub(crate) type_: String,
84    pub(crate) default_kind: String,
85}
86
87impl SchemaCheck {
88    /// `(database, bare table)` — a `db.table` qualification wins over the
89    /// config's `database` field, mirroring how the INSERT resolves.
90    fn target(&self) -> (Option<&str>, &str) {
91        match self.table.split_once('.') {
92            Some((db, tbl)) => (Some(db), tbl),
93            None => (self.database.as_deref(), &self.table),
94        }
95    }
96
97    fn display_table(&self) -> String {
98        match self.target() {
99            (Some(db), tbl) => format!("`{db}`.`{tbl}`"),
100            (None, tbl) => format!("`{tbl}`"),
101        }
102    }
103}
104
105/// Fetch one replica's view of the table.
106async fn fetch_columns(
107    endpoint: &ClickHouseEndpoint,
108    check: &SchemaCheck,
109) -> Result<Vec<ColumnRow>, SchemaError> {
110    let (db, table) = check.target();
111    let query = match db {
112        Some(db) => endpoint
113            .client()
114            .query(
115                "SELECT name, type, default_kind FROM system.columns \
116                 WHERE database = ? AND table = ? ORDER BY position",
117            )
118            .bind(db)
119            .bind(table),
120        None => endpoint
121            .client()
122            .query(
123                "SELECT name, type, default_kind FROM system.columns \
124                 WHERE database = currentDatabase() AND table = ? ORDER BY position",
125            )
126            .bind(table),
127    };
128    query
129        .fetch_all::<ColumnRow>()
130        .await
131        .map_err(|e| SchemaError::Fetch {
132            table: check.display_table(),
133            url: endpoint.url().to_string(),
134            reason: e.to_string(),
135        })
136}
137
138fn table_column_list(cols: &[ColumnRow]) -> String {
139    cols.iter()
140        .map(|c| {
141            if c.default_kind.is_empty() {
142                format!("{} {}", c.name, c.type_)
143            } else {
144                format!("{} {} {}", c.name, c.type_, c.default_kind)
145            }
146        })
147        .collect::<Vec<_>>()
148        .join(", ")
149}
150
151/// If a configured column's declared type is an `AggregateFunction(...)`,
152/// return an actionable error explaining the correct ingestion path.
153///
154/// The sink cannot write aggregate *states* directly: their wire form is the
155/// opaque, unframed, version-dependent internal serialization, and
156/// reproducing it client-side would couple us to one ClickHouse serialization
157/// version. A sink pointed straight at such a column is therefore a
158/// misconfiguration — the supported pattern is to INSERT raw rows into an
159/// `ENGINE = Null` landing table and let a `MATERIALIZED VIEW` build the
160/// states into the target.
161///
162/// Matched on the exact constructor name so `SimpleAggregateFunction(...)`,
163/// which stores the raw value and *is* directly insertable, is deliberately
164/// not flagged.
165fn aggregate_function_remedy(col: &str, type_: &str) -> Option<String> {
166    let ctor = type_.split_once('(').map(|(name, _)| name.trim())?;
167    if ctor != "AggregateFunction" {
168        return None;
169    }
170    Some(format!(
171        "configured column `{col}` has type `{type_}`: the sink cannot write \
172         aggregate states directly (their wire format is opaque and \
173         version-dependent). Insert raw rows into an `ENGINE = Null` landing \
174         table and let a `MATERIALIZED VIEW` compute the states \
175         (minState/maxState/sumMapState/...) into this table. Client-side \
176         aggregate-state serialization is not supported."
177    ))
178}
179
180/// Startup validation against every replica of every shard. `Ok(None)`
181/// when the mode is `Off`; `Ok(Some(schema))` for the encoder otherwise.
182pub(crate) async fn validate(
183    check: &SchemaCheck,
184    endpoints: &[Vec<ClickHouseEndpoint>],
185) -> Result<Option<Arc<RowSchema>>, SchemaError> {
186    if check.mode == SchemaValidation::Off {
187        return Ok(None);
188    }
189
190    // Every replica must agree — shard-local tables drift independently,
191    // and a broken replica should surface now, not at its first rotated
192    // write.
193    let mut reference: Option<(String, Vec<ColumnRow>)> = None;
194    for shard in endpoints {
195        for endpoint in shard {
196            let cols = fetch_columns(endpoint, check).await?;
197            if cols.is_empty() {
198                return Err(SchemaError::Mismatch(format!(
199                    "sink.clickhouse: table {} not found (or not visible to this user) \
200                     on replica {}",
201                    check.display_table(),
202                    endpoint.url()
203                )));
204            }
205            match &reference {
206                None => reference = Some((endpoint.url().to_string(), cols)),
207                Some((ref_url, ref_cols)) => {
208                    if *ref_cols != cols {
209                        return Err(SchemaError::Mismatch(format!(
210                            "sink.clickhouse: replicas disagree on the schema of {}:\n  \
211                             {ref_url}: {}\n  {}: {}",
212                            check.display_table(),
213                            table_column_list(ref_cols),
214                            endpoint.url(),
215                            table_column_list(&cols),
216                        )));
217                    }
218                }
219            }
220        }
221    }
222    let (ref_url, table_cols) =
223        reference.expect("config validation guarantees at least one replica");
224
225    let mut findings = Vec::new();
226    for col in &check.columns {
227        match table_cols.iter().find(|c| c.name == *col) {
228            None => findings.push(format!(
229                "configured column `{col}` does not exist in the table"
230            )),
231            Some(c) if c.default_kind == "MATERIALIZED" || c.default_kind == "ALIAS" => {
232                findings.push(format!(
233                    "configured column `{col}` is {} and cannot be inserted into",
234                    c.default_kind
235                ));
236            }
237            Some(c) => {
238                if let Some(msg) = aggregate_function_remedy(col, &c.type_) {
239                    findings.push(msg);
240                }
241            }
242        }
243    }
244    for c in &table_cols {
245        if !check.columns.contains(&c.name) && c.default_kind.is_empty() {
246            tracing::warn!(
247                table = %check.display_table(),
248                column = %c.name,
249                r#type = %c.type_,
250                "table column is not in the configured insert columns and has no DEFAULT; \
251                 the server will fill type-default values"
252            );
253        }
254    }
255
256    if !findings.is_empty() {
257        let mut msg = format!(
258            "sink.clickhouse: schema validation failed for {} (replica {ref_url}):\n",
259            check.display_table()
260        );
261        for f in &findings {
262            let _ = writeln!(msg, "  - {f}");
263        }
264        let _ = writeln!(
265            msg,
266            "  table columns:      {}",
267            table_column_list(&table_cols)
268        );
269        let _ = write!(msg, "  configured columns: {}", check.columns.join(", "));
270        return Err(SchemaError::Mismatch(msg));
271    }
272
273    let columns = check
274        .columns
275        .iter()
276        .map(|name| {
277            let c = table_cols
278                .iter()
279                .find(|c| c.name == *name)
280                .expect("checked above");
281            (name.clone(), typeparse::parse(&c.type_), c.type_.clone())
282        })
283        .collect();
284    Ok(Some(Arc::new(RowSchema {
285        mode: check.mode,
286        table: check.display_table(),
287        columns,
288    })))
289}
290
291/// The first-record struct check: field names and order against the
292/// configured columns (both modes), plus class-based type compatibility
293/// per position (`full` mode). Returns the pre-formatted diff on failure.
294pub(crate) fn check_first_record(schema: &RowSchema, fields: &[FieldShape]) -> Result<(), String> {
295    let mut findings = Vec::new();
296    if fields.len() != schema.columns.len() {
297        findings.push(format!(
298            "row struct serialized {} field(s) but {} column(s) are configured \
299             (a #[serde(skip)] attribute shortens rows silently)",
300            fields.len(),
301            schema.columns.len()
302        ));
303    }
304    for (i, (field, (col, ty, ty_str))) in fields.iter().zip(&schema.columns).enumerate() {
305        if field.name != col {
306            findings.push(format!(
307                "position {i}: struct field `{}` vs configured column `{col}`",
308                field.name
309            ));
310        } else if schema.mode == SchemaValidation::Full && !compatible(&field.shape, ty) {
311            findings.push(format!(
312                "position {i}: struct field `{}` ({}) is not compatible with `{col}` {ty_str}",
313                field.name,
314                shape_desc(&field.shape),
315            ));
316        }
317    }
318    if findings.is_empty() {
319        return Ok(());
320    }
321    let mut msg = format!(
322        "row struct does not match configured columns for {}:\n",
323        schema.table
324    );
325    for f in &findings {
326        let _ = writeln!(msg, "  - {f}");
327    }
328    let _ = writeln!(
329        msg,
330        "  struct fields (declaration order): {}",
331        fields.iter().map(|f| f.name).collect::<Vec<_>>().join(", ")
332    );
333    let _ = write!(
334        msg,
335        "  configured columns:                {}",
336        schema
337            .columns
338            .iter()
339            .map(|(name, ..)| name.as_str())
340            .collect::<Vec<_>>()
341            .join(", ")
342    );
343    Err(msg)
344}
345
346fn shape_desc(shape: &Shape) -> String {
347    match shape {
348        Shape::Bool => "bool".into(),
349        Shape::I8 => "i8".into(),
350        Shape::I16 => "i16".into(),
351        Shape::I32 => "i32".into(),
352        Shape::I64 => "i64".into(),
353        Shape::I128 => "i128".into(),
354        Shape::U8 => "u8".into(),
355        Shape::U16 => "u16".into(),
356        Shape::U32 => "u32".into(),
357        Shape::U64 => "u64".into(),
358        Shape::U128 => "u128".into(),
359        Shape::F32 => "f32".into(),
360        Shape::F64 => "f64".into(),
361        Shape::Str => "String".into(),
362        Shape::Bytes => "bytes".into(),
363        Shape::Option(inner) => format!("Option<{}>", shape_desc(inner)),
364        Shape::Seq(inner) => format!("Vec<{}>", shape_desc(inner)),
365        Shape::Map(k, v) => format!("Map<{}, {}>", shape_desc(k), shape_desc(v)),
366        Shape::Tuple(elems) => format!(
367            "({})",
368            elems.iter().map(shape_desc).collect::<Vec<_>>().join(", ")
369        ),
370        Shape::Struct(fields) => format!(
371            "struct {{ {} }}",
372            fields
373                .iter()
374                .map(|f| f.name.to_string())
375                .collect::<Vec<_>>()
376                .join(", ")
377        ),
378        Shape::Newtype(name, _) => (*name).to_string(),
379        Shape::Unknown => "_".into(),
380    }
381}
382
383#[cfg(test)]
384mod tests {
385    use super::*;
386    use probe::probe_row;
387    use serde::Serialize;
388
389    fn schema(mode: SchemaValidation, cols: &[(&str, &str)]) -> RowSchema {
390        RowSchema {
391            mode,
392            table: "`orders`".into(),
393            columns: cols
394                .iter()
395                .map(|(n, t)| ((*n).to_string(), typeparse::parse(t), (*t).to_string()))
396                .collect(),
397        }
398    }
399
400    #[derive(Serialize)]
401    struct Row {
402        id: u64,
403        amount: i64,
404        name: String,
405    }
406
407    fn fields() -> Vec<FieldShape> {
408        probe_row(&Row {
409            id: 1,
410            amount: 2,
411            name: "x".into(),
412        })
413        .unwrap()
414    }
415
416    #[test]
417    fn matching_struct_passes_both_modes() {
418        for mode in [SchemaValidation::Names, SchemaValidation::Full] {
419            let s = schema(
420                mode,
421                &[("id", "UInt64"), ("amount", "Int64"), ("name", "String")],
422            );
423            assert_eq!(check_first_record(&s, &fields()), Ok(()));
424        }
425    }
426
427    #[test]
428    fn field_order_mismatch_is_reported_per_position() {
429        let s = schema(
430            SchemaValidation::Names,
431            &[("id", "UInt64"), ("name", "String"), ("amount", "Int64")],
432        );
433        let err = check_first_record(&s, &fields()).unwrap_err();
434        assert!(err.contains("position 1: struct field `amount` vs configured column `name`"));
435        assert!(err.contains("position 2: struct field `name` vs configured column `amount`"));
436        assert!(err.contains("struct fields (declaration order): id, amount, name"));
437        assert!(err.contains("configured columns:                id, name, amount"));
438    }
439
440    #[test]
441    fn type_mismatches_only_fail_full_mode() {
442        let cols = [
443            ("id", "UInt64"),
444            ("amount", "DateTime"), // i64 field is not a u32 DateTime
445            ("name", "String"),
446        ];
447        let names = schema(SchemaValidation::Names, &cols);
448        assert_eq!(check_first_record(&names, &fields()), Ok(()));
449
450        let full = schema(SchemaValidation::Full, &cols);
451        let err = check_first_record(&full, &fields()).unwrap_err();
452        assert!(
453            err.contains("struct field `amount` (i64) is not compatible with `amount` DateTime"),
454            "{err}"
455        );
456    }
457
458    #[test]
459    fn field_count_mismatch_names_the_skip_footgun() {
460        let s = schema(SchemaValidation::Names, &[("id", "UInt64")]);
461        let err = check_first_record(&s, &fields()).unwrap_err();
462        assert!(err.contains("serialized 3 field(s) but 1 column(s)"));
463        assert!(err.contains("#[serde(skip)]"));
464    }
465
466    #[test]
467    fn aggregate_function_columns_are_flagged_with_the_null_mv_remedy() {
468        for ty in [
469            "AggregateFunction(min, DateTime)",
470            "AggregateFunction(max, DateTime)",
471            "AggregateFunction(sumMap, Map(String, UInt64))",
472        ] {
473            let msg = aggregate_function_remedy("agg", ty)
474                .unwrap_or_else(|| panic!("`{ty}` should be flagged"));
475            assert!(msg.contains(ty), "{msg}");
476            assert!(msg.contains("Null"), "{msg}");
477            assert!(msg.contains("MATERIALIZED VIEW"), "{msg}");
478        }
479    }
480
481    #[test]
482    fn directly_insertable_types_are_not_flagged() {
483        for ty in [
484            // SimpleAggregateFunction stores the raw value — insertable.
485            "SimpleAggregateFunction(sum, UInt64)",
486            "SimpleAggregateFunction(sumMap, Map(String, UInt64))",
487            "UInt64",
488            "Map(String, UInt64)",
489            "DateTime",
490        ] {
491            assert_eq!(aggregate_function_remedy("c", ty), None, "{ty}");
492        }
493    }
494}
spate_clickhouse/schema/mod.rs

spate_clickhouse/schema/
mod.rs
