atharvalade commented on code in PR #2886: URL: https://github.com/apache/iggy/pull/2886#discussion_r3522765234
########## core/connectors/sinks/clickhouse_sink/src/client.rs: ########## @@ -0,0 +1,500 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Thin `reqwest`-based HTTP client for the ClickHouse HTTP interface. +//! +//! ClickHouse exposes its HTTP API at `http://host:port/`. Queries are sent +//! either as a URL query parameter (`?query=...`) or in the request body. +//! Authentication uses the `X-ClickHouse-User` and `X-ClickHouse-Key` headers. +//! +//! Insert format: +//! POST /?database={db}&query=INSERT+INTO+{table}+FORMAT+{fmt} +//! Body: row data in the chosen format + +use crate::schema::{Column, parse_type}; +use bytes::Bytes; +use iggy_connector_sdk::Error; +use rand::RngExt; +use reqwest::StatusCode; +use reqwest::header::{CONTENT_TYPE, HeaderMap, HeaderValue}; +use serde::Deserialize; +use std::time::Duration; +use tracing::{debug, error, info, warn}; + +const USER_HEADER: &str = "X-ClickHouse-User"; +const KEY_HEADER: &str = "X-ClickHouse-Key"; + +/// Thin wrapper around `reqwest::Client` pre-configured for a ClickHouse +/// endpoint. +#[derive(Debug)] +pub(crate) struct ClickHouseClient { + inner: reqwest::Client, + base_url: String, + database: String, + table: String, + format_name: String, + insert_url: String, + insert_query: String, +} + +impl ClickHouseClient { + /// Build a new client. + pub fn new( + base_url: String, + database: String, + table: String, + format_name: String, + username: &str, + password: &str, + timeout: Duration, + ) -> Result<Self, Error> { + let mut auth_headers = HeaderMap::new(); + auth_headers.insert( + USER_HEADER, + HeaderValue::from_str(username) + .map_err(|e| Error::InitError(format!("Invalid username header value: {e}")))?, + ); + auth_headers.insert( Review Comment: The config uses `SecretString` now, but the password ends up in a plain header and reqwest's Debug prints `default_headers`. I checked, format!("{:?}") on the client shows the password verbatim, and both structs derive `Debug`. Calling `set_sensitive(true)` on the `HeaderValue` would fix it. ########## core/connectors/sinks/clickhouse_sink/src/binary.rs: ########## @@ -0,0 +1,1647 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! RowBinary / RowBinaryWithDefaults byte serialization. +//! +//! Follows the ClickHouse binary format specification: +//! <https://clickhouse.com/docs/en/interfaces/formats#rowbinary> +//! +//! Key layout rules: +//! - All integers are **little-endian**. +//! - Strings are prefixed with an **unsigned LEB128 varint** length. +//! - `Nullable(T)`: 1-byte null marker (`0x01` = null, `0x00` = not null) +//! followed by T bytes when not null. +//! - `RowBinaryWithDefaults`: each top-level column is preceded by a 1-byte +//! flag (`0x01` = use server DEFAULT, `0x00` = value follows). + +use std::borrow::Cow; + +use crate::schema::{ChType, Column}; +use iggy_connector_sdk::Error; +use simd_json::OwnedValue; +use simd_json::prelude::{TypedScalarValue, ValueAsArray, ValueAsObject}; +use tracing::{error, warn}; + +// ─── Public API ────────────────────────────────────────────────────────────── + +/// Serialise one message (a JSON object) as a RowBinaryWithDefaults row. +/// +/// Columns are written in schema order. When a column is absent from the JSON +/// object and `has_default` is true the DEFAULT prefix byte (`0x01`) is +/// written and the column value is skipped. When a column is absent but has no +/// default and is not Nullable this is an error. +pub(crate) fn serialize_row( + value: &OwnedValue, + columns: &[Column], + buf: &mut Vec<u8>, +) -> Result<(), Error> { + let obj = value.as_object().ok_or_else(|| { + error!("RowBinary: message payload is not a JSON object"); + Error::InvalidRecord + })?; + + for col in columns { + let field_value = obj.get(col.name.as_str()); + + // RowBinaryWithDefaults prefix byte. Only an absent field defers to the + // server DEFAULT; an explicit JSON null must fall through to the + // Nullable path so it is stored as NULL rather than the default value. + if field_value.is_none() && col.has_default { + buf.push(0x01); // use DEFAULT + continue; + } + buf.push(0x00); // value follows + + match field_value { + Some(v) if !v.is_null() => serialize_value(v, &col.ch_type, buf)?, + _ => { + // Field is absent or null — write zero value if Nullable, + // otherwise error. + write_zero_or_null(&col.ch_type, buf, &col.name)?; + } + } + } + Ok(()) +} + +// ─── Core recursive serializer ──────────────────────────────────────────────── + +pub(crate) fn serialize_value( + value: &OwnedValue, + ch_type: &ChType, + buf: &mut Vec<u8>, +) -> Result<(), Error> { + match ch_type { + // ── Nullable ───────────────────────────────────────────────────────── + ChType::Nullable(inner) => { + if value.is_null() { + buf.push(0x01); // null + } else { + buf.push(0x00); // not null + serialize_value(value, inner, buf)?; + } + } + + // ── String ─────────────────────────────────────────────────────────── + ChType::String => { + let s = coerce_to_string(value)?; + write_string(s.as_bytes(), buf); + } + ChType::FixedString(n) => { + let s = coerce_to_string(value)?; + let bytes = s.as_bytes(); + // Pad or truncate to exactly n bytes + let copy_len = bytes.len().min(*n); + buf.extend_from_slice(&bytes[..copy_len]); + buf.resize(buf.len() + (n - copy_len), 0u8); + } + + // ── Integers ───────────────────────────────────────────────────────── + ChType::Int8 => buf.push(i8::try_from(coerce_i64(value)?).map_err(|_| { + error!("Value out of range for Int8"); + Error::InvalidRecord + })? as u8), + ChType::Int16 => buf.extend_from_slice( + &i16::try_from(coerce_i64(value)?) + .map_err(|_| { + error!("Value out of range for Int16"); + Error::InvalidRecord + })? + .to_le_bytes(), + ), + ChType::Int32 => buf.extend_from_slice( + &i32::try_from(coerce_i64(value)?) + .map_err(|_| { + error!("Value out of range for Int32"); + Error::InvalidRecord + })? + .to_le_bytes(), + ), + ChType::Int64 => buf.extend_from_slice(&coerce_i64(value)?.to_le_bytes()), + ChType::UInt8 => buf.push(u8::try_from(coerce_u64(value)?).map_err(|_| { + error!("Value out of range for UInt8"); + Error::InvalidRecord + })?), + ChType::UInt16 => buf.extend_from_slice( + &u16::try_from(coerce_u64(value)?) + .map_err(|_| { + error!("Value out of range for UInt16"); + Error::InvalidRecord + })? + .to_le_bytes(), + ), + ChType::UInt32 => buf.extend_from_slice( + &u32::try_from(coerce_u64(value)?) + .map_err(|_| { + error!("Value out of range for UInt32"); + Error::InvalidRecord + })? + .to_le_bytes(), + ), + ChType::UInt64 => buf.extend_from_slice(&coerce_u64(value)?.to_le_bytes()), + + // ── Floats ─────────────────────────────────────────────────────────── + ChType::Float32 => { + let f = coerce_f64(value)? as f32; + buf.extend_from_slice(&f.to_le_bytes()); + } + ChType::Float64 => { + let f = coerce_f64(value)?; + buf.extend_from_slice(&f.to_le_bytes()); + } + + // ── Boolean ────────────────────────────────────────────────────────── + ChType::Boolean => { + let b = match value { + OwnedValue::Static(simd_json::StaticNode::Bool(b)) => *b, + OwnedValue::Static(simd_json::StaticNode::I64(n)) => *n != 0, + OwnedValue::Static(simd_json::StaticNode::U64(n)) => *n != 0, + _ => { + error!("Cannot convert to Boolean: {value:?}"); + return Err(Error::InvalidRecord); + } + }; + buf.push(b as u8); + } + + // ── UUID ───────────────────────────────────────────────────────────── + // ClickHouse stores UUID as two little-endian 64-bit words. + // Input: standard hyphenated UUID string "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" + ChType::Uuid => { + let s = coerce_to_string(value)?; + let b = s.as_bytes(); + // Validate hyphenated UUID format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx + if b.len() != 36 || b[8] != b'-' || b[13] != b'-' || b[18] != b'-' || b[23] != b'-' { + error!("Invalid UUID string (expected xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx): {s}"); + return Err(Error::InvalidRecord); + } + // Copy hex segments into a stack buffer, skipping the 4 dashes + let mut hex = [0u8; 32]; + let segments: [&[u8]; 5] = [&b[0..8], &b[9..13], &b[14..18], &b[19..23], &b[24..36]]; + let mut i = 0; + for seg in segments { + hex[i..i + seg.len()].copy_from_slice(seg); + i += seg.len(); + } + let hex_str = std::str::from_utf8(&hex).map_err(|_| { + error!("Cannot decode UUID hex: {s}"); + Error::InvalidRecord + })?; + // ClickHouse UUID: two LE UInt64 words, low word (bytes 8-15) first. + // hex_str[..16] is the high u64, hex_str[16..] the low u64 (both big-endian). + let hi = u64::from_str_radix(&hex_str[..16], 16).map_err(|_| { + error!("Cannot decode UUID hex: {s}"); + Error::InvalidRecord + })?; + let lo = u64::from_str_radix(&hex_str[16..], 16).map_err(|_| { + error!("Cannot decode UUID hex: {s}"); + Error::InvalidRecord + })?; + buf.extend_from_slice(&lo.to_le_bytes()); + buf.extend_from_slice(&hi.to_le_bytes()); + } + + // ── Date types ─────────────────────────────────────────────────────── + ChType::Date => { + // Days since 1970-01-01 as UInt16. Accept integer or "YYYY-MM-DD". + let days = coerce_to_days(value)? as u16; + buf.extend_from_slice(&days.to_le_bytes()); + } + ChType::Date32 => { + let days = coerce_to_days(value)? as i32; + buf.extend_from_slice(&days.to_le_bytes()); + } + ChType::DateTime => { + // Unix seconds as UInt32. Accept integer or RFC 3339 string. + let secs = coerce_to_unix_seconds(value)? as u32; Review Comment: Same problem here, `as u32` wraps pre-1970 timestamps into garbage. "1969-12-31T00:00:00Z" stores 4294880896. I think these two casts need the same `try_from` treatment the integers already have. ########## core/connectors/sinks/clickhouse_sink/src/binary.rs: ########## @@ -0,0 +1,1647 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! RowBinary / RowBinaryWithDefaults byte serialization. +//! +//! Follows the ClickHouse binary format specification: +//! <https://clickhouse.com/docs/en/interfaces/formats#rowbinary> +//! +//! Key layout rules: +//! - All integers are **little-endian**. +//! - Strings are prefixed with an **unsigned LEB128 varint** length. +//! - `Nullable(T)`: 1-byte null marker (`0x01` = null, `0x00` = not null) +//! followed by T bytes when not null. +//! - `RowBinaryWithDefaults`: each top-level column is preceded by a 1-byte +//! flag (`0x01` = use server DEFAULT, `0x00` = value follows). + +use std::borrow::Cow; + +use crate::schema::{ChType, Column}; +use iggy_connector_sdk::Error; +use simd_json::OwnedValue; +use simd_json::prelude::{TypedScalarValue, ValueAsArray, ValueAsObject}; +use tracing::{error, warn}; + +// ─── Public API ────────────────────────────────────────────────────────────── + +/// Serialise one message (a JSON object) as a RowBinaryWithDefaults row. +/// +/// Columns are written in schema order. When a column is absent from the JSON +/// object and `has_default` is true the DEFAULT prefix byte (`0x01`) is +/// written and the column value is skipped. When a column is absent but has no +/// default and is not Nullable this is an error. +pub(crate) fn serialize_row( + value: &OwnedValue, + columns: &[Column], + buf: &mut Vec<u8>, +) -> Result<(), Error> { + let obj = value.as_object().ok_or_else(|| { + error!("RowBinary: message payload is not a JSON object"); + Error::InvalidRecord + })?; + + for col in columns { + let field_value = obj.get(col.name.as_str()); + + // RowBinaryWithDefaults prefix byte. Only an absent field defers to the + // server DEFAULT; an explicit JSON null must fall through to the + // Nullable path so it is stored as NULL rather than the default value. + if field_value.is_none() && col.has_default { + buf.push(0x01); // use DEFAULT + continue; + } + buf.push(0x00); // value follows + + match field_value { + Some(v) if !v.is_null() => serialize_value(v, &col.ch_type, buf)?, + _ => { + // Field is absent or null — write zero value if Nullable, + // otherwise error. + write_zero_or_null(&col.ch_type, buf, &col.name)?; + } + } + } + Ok(()) +} + +// ─── Core recursive serializer ──────────────────────────────────────────────── + +pub(crate) fn serialize_value( + value: &OwnedValue, + ch_type: &ChType, + buf: &mut Vec<u8>, +) -> Result<(), Error> { + match ch_type { + // ── Nullable ───────────────────────────────────────────────────────── + ChType::Nullable(inner) => { + if value.is_null() { + buf.push(0x01); // null + } else { + buf.push(0x00); // not null + serialize_value(value, inner, buf)?; + } + } + + // ── String ─────────────────────────────────────────────────────────── + ChType::String => { + let s = coerce_to_string(value)?; + write_string(s.as_bytes(), buf); + } + ChType::FixedString(n) => { + let s = coerce_to_string(value)?; + let bytes = s.as_bytes(); + // Pad or truncate to exactly n bytes + let copy_len = bytes.len().min(*n); + buf.extend_from_slice(&bytes[..copy_len]); + buf.resize(buf.len() + (n - copy_len), 0u8); + } + + // ── Integers ───────────────────────────────────────────────────────── + ChType::Int8 => buf.push(i8::try_from(coerce_i64(value)?).map_err(|_| { + error!("Value out of range for Int8"); + Error::InvalidRecord + })? as u8), + ChType::Int16 => buf.extend_from_slice( + &i16::try_from(coerce_i64(value)?) + .map_err(|_| { + error!("Value out of range for Int16"); + Error::InvalidRecord + })? + .to_le_bytes(), + ), + ChType::Int32 => buf.extend_from_slice( + &i32::try_from(coerce_i64(value)?) + .map_err(|_| { + error!("Value out of range for Int32"); + Error::InvalidRecord + })? + .to_le_bytes(), + ), + ChType::Int64 => buf.extend_from_slice(&coerce_i64(value)?.to_le_bytes()), + ChType::UInt8 => buf.push(u8::try_from(coerce_u64(value)?).map_err(|_| { + error!("Value out of range for UInt8"); + Error::InvalidRecord + })?), + ChType::UInt16 => buf.extend_from_slice( + &u16::try_from(coerce_u64(value)?) + .map_err(|_| { + error!("Value out of range for UInt16"); + Error::InvalidRecord + })? + .to_le_bytes(), + ), + ChType::UInt32 => buf.extend_from_slice( + &u32::try_from(coerce_u64(value)?) + .map_err(|_| { + error!("Value out of range for UInt32"); + Error::InvalidRecord + })? + .to_le_bytes(), + ), + ChType::UInt64 => buf.extend_from_slice(&coerce_u64(value)?.to_le_bytes()), + + // ── Floats ─────────────────────────────────────────────────────────── + ChType::Float32 => { + let f = coerce_f64(value)? as f32; + buf.extend_from_slice(&f.to_le_bytes()); + } + ChType::Float64 => { + let f = coerce_f64(value)?; + buf.extend_from_slice(&f.to_le_bytes()); + } + + // ── Boolean ────────────────────────────────────────────────────────── + ChType::Boolean => { + let b = match value { + OwnedValue::Static(simd_json::StaticNode::Bool(b)) => *b, + OwnedValue::Static(simd_json::StaticNode::I64(n)) => *n != 0, + OwnedValue::Static(simd_json::StaticNode::U64(n)) => *n != 0, + _ => { + error!("Cannot convert to Boolean: {value:?}"); + return Err(Error::InvalidRecord); + } + }; + buf.push(b as u8); + } + + // ── UUID ───────────────────────────────────────────────────────────── + // ClickHouse stores UUID as two little-endian 64-bit words. + // Input: standard hyphenated UUID string "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" + ChType::Uuid => { + let s = coerce_to_string(value)?; + let b = s.as_bytes(); + // Validate hyphenated UUID format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx + if b.len() != 36 || b[8] != b'-' || b[13] != b'-' || b[18] != b'-' || b[23] != b'-' { + error!("Invalid UUID string (expected xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx): {s}"); + return Err(Error::InvalidRecord); + } + // Copy hex segments into a stack buffer, skipping the 4 dashes + let mut hex = [0u8; 32]; + let segments: [&[u8]; 5] = [&b[0..8], &b[9..13], &b[14..18], &b[19..23], &b[24..36]]; + let mut i = 0; + for seg in segments { + hex[i..i + seg.len()].copy_from_slice(seg); + i += seg.len(); + } + let hex_str = std::str::from_utf8(&hex).map_err(|_| { + error!("Cannot decode UUID hex: {s}"); + Error::InvalidRecord + })?; + // ClickHouse UUID: two LE UInt64 words, low word (bytes 8-15) first. + // hex_str[..16] is the high u64, hex_str[16..] the low u64 (both big-endian). + let hi = u64::from_str_radix(&hex_str[..16], 16).map_err(|_| { + error!("Cannot decode UUID hex: {s}"); + Error::InvalidRecord + })?; + let lo = u64::from_str_radix(&hex_str[16..], 16).map_err(|_| { + error!("Cannot decode UUID hex: {s}"); + Error::InvalidRecord + })?; + buf.extend_from_slice(&lo.to_le_bytes()); + buf.extend_from_slice(&hi.to_le_bytes()); + } + + // ── Date types ─────────────────────────────────────────────────────── + ChType::Date => { + // Days since 1970-01-01 as UInt16. Accept integer or "YYYY-MM-DD". + let days = coerce_to_days(value)? as u16; Review Comment: This `as u16` silently wraps for dates outside Date's range. I tried "2200-01-01" and it stored day 18470, which reads back as 2020-07-27, and "1969-12-31" became 65535. Shouldn't this be a bounds check like the int columns got? ########## core/connectors/sinks/clickhouse_sink/src/schema.rs: ########## @@ -0,0 +1,778 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! ClickHouse column schema model and type string parser used by RowBinary mode. +//! +//! +//! # Type string grammar +//! +//! Grammar followed by ClickHouse type strings (as returned by `SELECT type FROM system.columns`): +//! ```text +//! type ::= composite | parameterised | primitive +//! +//! composite ::= "Nullable(" type ")" +//! | "Array(" type ")" +//! | "Map(" type ", " type ")" +//! | "Tuple(" tuple_fields ")" +//! +//! tuple_fields ::= type ("," type)* -- unnamed fields +//! | field type ("," field type)* -- named fields +//! +//! field ::= ident | "`" [^`]* "`" +//! +//! parameterised ::= "FixedString(" n ")" +//! | "DateTime64(" precision ["," tz] ")" +//! | "DateTime(" tz ")" +//! | "Decimal(" precision "," scale ")" +//! | "Decimal32(" scale ")" +//! | "Decimal64(" scale ")" +//! | "Decimal128(" scale ")" +//! | "Enum8(" enum_pairs ")" +//! | "Enum16(" enum_pairs ")" +//! +//! enum_pairs ::= "'" name "' = " int ("," "'" name "' = " int)* +//! +//! primitive ::= "String" | "Int8" | "Int16" | "Int32" | "Int64" +//! | "UInt8" | "UInt16" | "UInt32" | "UInt64" +//! | "Float32" | "Float64" | "Bool" | "Boolean" +//! | "UUID" | "Date" | "Date32" | "DateTime" +//! | "IPv4" | "IPv6" +//! ``` +//! +//! +//! ## Example +//! +//! ```text +//! Nullable(Map(String, Array(Tuple(id Int32, ts DateTime64(3, 'UTC'))))) +//! ``` +//! +//! Parses into the AST: +//! +//! ```text +//! Nullable +//! └── Map +//! ├── key: String +//! └── value: Array +//! └── Tuple +//! ├── [0] Int32 +//! └── [1] DateTime64(3) +//! ``` + +use iggy_connector_sdk::Error; +use std::collections::HashMap; +use tracing::error; + +/// A single ClickHouse table column. +#[derive(Debug, Clone)] +pub struct Column { + pub name: String, + pub ch_type: ChType, + /// True only for columns with an ordinary DEFAULT expression, meaning + /// RowBinaryWithDefaults can skip them with a 0x01 prefix byte. + /// MATERIALIZED / ALIAS / EPHEMERAL columns are not insertable and are + /// excluded from the schema entirely rather than flagged here. + pub has_default: bool, +} + +/// Supported ClickHouse column types. +/// +/// Unsupported types (Variant, the new JSON column type, geo types) cause +/// `parse_type` to return an error, which in turn makes `open()` fail rather +/// than silently producing corrupt data. `LowCardinality(T)` is transparently +/// unwrapped to `T` since RowBinary serialises it identically. +#[derive(Debug, Clone, PartialEq)] +pub enum ChType { + // ── Primitives ────────────────────────────────────────────────────────── + String, + Int8, + Int16, + Int32, + Int64, + UInt8, + UInt16, + UInt32, + UInt64, + Float32, + Float64, + Boolean, + Uuid, + /// Days since 1970-01-01 stored as UInt16. + Date, + /// Days since 1970-01-01 stored as Int32. + Date32, + /// Unix seconds stored as UInt32. Optional timezone suffix is ignored for + /// serialisation purposes. + DateTime, + /// Unix time scaled by 10^precision stored as Int64. + DateTime64(u8), + /// Fixed-width byte string padded with zeros. + FixedString(usize), + /// Decimal(precision, scale). Serialised as Int32 / Int64 / Int128. + Decimal(u8, u8), + /// IPv4 address — UInt32, 4 bytes little-endian. + IPv4, + /// IPv6 address — 16 bytes, big-endian. + IPv6, + /// Enum8: maps string → i8. Values parsed from the type definition. + Enum8(HashMap<String, i8>), + /// Enum16: maps string → i16. Values parsed from the type definition. + Enum16(HashMap<String, i16>), + + // ── Composites (recursive) ─────────────────────────────────────────────── + Nullable(Box<ChType>), + Array(Box<ChType>), + /// Map(key_type, value_type) + Map(Box<ChType>, Box<ChType>), + /// Tuple of ordered fields. Each field carries an optional name (named tuples). + Tuple(Vec<(Option<String>, ChType)>), +} + +// ─── Public entry point ────────────────────────────────────────────────────── + +/// Parse a ClickHouse type string (as returned by `system.columns`) into a +/// `ChType`. Returns `Err(Error::InitError(...))` for unsupported or +/// unrecognised types. +pub fn parse_type(s: &str) -> Result<ChType, Error> { + parse_type_inner(s.trim()) +} + +// ─── Recursive descent parser ──────────────────────────────────────────────── + +fn parse_type_inner(s: &str) -> Result<ChType, Error> { + // Strip a single pair of outer parentheses if the entire string is wrapped. + // This shouldn't be needed for well-formed ClickHouse type strings, but is + // a defensive measure. + let s = s.trim(); + // e.g. "Nullable(Int32)" + if let Some(inner) = strip_wrapper(s, "Nullable") { + return Ok(ChType::Nullable(Box::new(parse_type_inner(inner)?))); + } + // e.g. "LowCardinality(String)" — serialised identically to the inner type + // in RowBinary format, so just unwrap it. + if let Some(inner) = strip_wrapper(s, "LowCardinality") { + return parse_type_inner(inner); + } + // e.g. "Array(String)" + if let Some(inner) = strip_wrapper(s, "Array") { + return Ok(ChType::Array(Box::new(parse_type_inner(inner)?))); + } + // e.g. "Map(String, Int64)" + if let Some(inner) = strip_wrapper(s, "Map") { + let (k, v) = split_two_args(inner)?; + return Ok(ChType::Map( + Box::new(parse_type_inner(k)?), + Box::new(parse_type_inner(v)?), + )); + } + // e.g. "Tuple(Int32, String)" or "Tuple(id Int32, name String)" + if let Some(inner) = strip_wrapper(s, "Tuple") { + let parts = split_args(inner)?; + let fields: Result<Vec<(Option<String>, ChType)>, Error> = parts + .iter() + .map(|p| { + let p = p.trim(); + if let Some((name, rest)) = strip_named_tuple_field(p) + && let Ok(ch_type) = parse_type_inner(rest) + { + return Ok((Some(name.to_string()), ch_type)); + } + Ok((None, parse_type_inner(p)?)) + }) + .collect(); + return Ok(ChType::Tuple(fields?)); + } + // e.g. "Enum8('a' = 1, 'b' = 2)" + if let Some(inner) = strip_wrapper(s, "Enum8") { + let map = parse_enum_values_i8(inner)?; + return Ok(ChType::Enum8(map)); + } + // e.g. "Enum16('a' = 1, 'b' = 2)" + if let Some(inner) = strip_wrapper(s, "Enum16") { + let map = parse_enum_values_i16(inner)?; + return Ok(ChType::Enum16(map)); + } + // e.g. "FixedString(16)" + if let Some(inner) = strip_wrapper(s, "FixedString") { + let n: usize = inner + .trim() + .parse() + .map_err(|_| init_err(format!("Invalid FixedString length: {inner}")))?; + return Ok(ChType::FixedString(n)); + } + // e.g. "DateTime64(3)" or "DateTime64(3, 'UTC')" + if let Some(inner) = strip_wrapper(s, "DateTime64") { + // DateTime64(precision) or DateTime64(precision, 'timezone') + let precision_str = inner.split(',').next().unwrap_or(inner).trim(); + let precision: u8 = precision_str + .parse() + .map_err(|_| init_err(format!("Invalid DateTime64 precision: {precision_str}")))?; + if precision > 9 { + return Err(init_err(format!( + "DateTime64 precision must be 0-9, got {precision}" + ))); + } + return Ok(ChType::DateTime64(precision)); + } + // e.g. "DateTime('UTC')" + if let Some(inner) = strip_wrapper(s, "DateTime") { + // DateTime('timezone') — timezone is ignored for serialisation. + let _ = inner; + return Ok(ChType::DateTime); + } + // e.g. "Decimal(18, 4)" + if let Some(inner) = strip_wrapper(s, "Decimal") { Review Comment: ClickHouse allows Decimal precision up to 76 (Decimal256, 32 bytes on the wire) but this parses it fine and binary.rs then writes only 16 bytes, shifting every column after it. I verified with Decimal(76, 2), it serializes 16 bytes without error.I think it would be better to reject precision > 38 here so open() fails fast like it does for Variant -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
