hubcio commented on code in PR #2886: URL: https://github.com/apache/iggy/pull/2886#discussion_r3527037552
########## core/connectors/sinks/clickhouse_sink/src/binary.rs: ########## @@ -0,0 +1,1679 @@ +// 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()); Review Comment: uuid halves are swapped. clickhouse rowbinary stores a uuid as two little-endian u64 words with the high word first: reverse of uuid bytes 0-8, then reverse of bytes 8-16. this code writes `lo` then `hi`, which is a flat 16-byte reversal (uuid as u128-LE) - a different layout. verified against a live clickhouse 25.5: selecting `61f0c404-5cb3-11e7-907b-a6006ad3dba0` as rowbinary emits `e7 11 b3 5c 04 c4 f0 61 a0 db d3 6a 00 a6 7b 90` (high half first), and inserting this code's byte order stores `907ba600-6ad3-dba0-61f0-c4045cb311e7` - silently, no error. so every uuid written by the sink comes back corrupted. the rowbinary format docs have this exact worked example: https://clickhouse.com/docs/interfaces/formats/RowBinary - and clickhouse-rs serializes `as_u64_pair()` = (high, low) in that order. fix is swapping the two writes. the unit test at line 1097 locks the wrong bytes so it needs updating too, and a live uuid round-trip in the integration test would have caught this. apologies - my earlier comment on this gave the order backwards. ########## core/connectors/sinks/clickhouse_sink/src/binary.rs: ########## @@ -0,0 +1,1679 @@ +// 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); Review Comment: over-length strings are silently truncated to `n` bytes. clickhouse itself rejects these - inserting 'abcdef' into `FixedString(3)` throws `TOO_LARGE_STRING_SIZE` (verified on 25.5; the fixedstring docs say it throws "Too large value for FixedString(N)"). silently dropping bytes is data corruption from the user's perspective - return `Error::InvalidRecord` like the other out-of-range paths. the test at line 1061 locks the truncating behavior. ########## core/connectors/sinks/clickhouse_sink/README.md: ########## @@ -0,0 +1,194 @@ +# ClickHouse Sink Connector + +The ClickHouse sink connector consumes messages from Iggy topics and inserts them into ClickHouse tables. Supports three insert formats: `json_each_row` (default), `row_binary`, and `string` passthrough. + +## Features + +- **Multiple Insert Formats**: Insert as `JSONEachRow`, `RowBinaryWithDefaults`, or raw string passthrough (CSV/TSV/JSON) +- **Schema Validation**: In `row_binary` mode, the table schema is fetched and validated at startup +- **Automatic Retries**: Configurable retry count and delay for transient errors +- **Batch Processing**: Insert messages in configurable batches via the stream configuration + +## Configuration + +```toml +type = "sink" +key = "clickhouse" +enabled = true +version = 0 +name = "ClickHouse sink" +path = "target/release/libiggy_connector_clickhouse_sink" + +[[streams]] +stream = "example_stream" +topics = ["example_topic"] +schema = "json" +batch_length = 1000 +poll_interval = "5ms" +consumer_group = "clickhouse_sink_connector" + +[plugin_config] +url = "http://localhost:8123"; +database = "default" +username = "default" +password = "" +table = "events" +insert_format = "json_each_row" +timeout_seconds = 30 +max_retries = 3 +retry_delay = 1 # seconds +verbose_logging = false +``` + +## Configuration Options + +| Option | Type | Default | Description | +| ------ | ---- | ------- | ----------- | +| `url` | string | required | ClickHouse HTTP endpoint | +| `table` | string | required | Target table name | +| `database` | string | `"default"` | ClickHouse database | +| `username` | string | `"default"` | ClickHouse username | +| `password` | string | `""` | ClickHouse password | +| `insert_format` | string | `"json_each_row"` | Insert format: `json_each_row`, `row_binary`, or `string` | +| `string_format` | string | `"json_each_row"` | ClickHouse format for `string` mode: `json_each_row`, `csv`, or `tsv` | +| `timeout_seconds` | u64 | `30` | HTTP request timeout | +| `max_retries` | u32 | `3` | Max retry attempts on transient errors | +| `retry_delay` | u64 | `1` | Delay between retries, in seconds | +| `verbose_logging` | bool | `false` | Log inserts at info level instead of debug | + +> **TODO:** `database` and `table` values are interpolated directly into SQL. Currently only +> single quotes are escaped; backslashes pass through unchanged, which can misparse string +> literals if a value ends with `\`. A future improvement should validate both fields against a +> strict allowlist (`^[A-Za-z_][A-Za-z0-9_]*$`) at config load and escape backslashes in SQL +> string literals. Deferred because these sinks run in operator-controlled environments where +> config values are trusted. + +## Insert Formats + +### `json_each_row` (Default) + +Accepts messages with a `Payload::Json` payload. Each message is sent as a JSON object on its own line using ClickHouse's `JSONEachRow` format. ClickHouse handles type coercion from the JSON values to the column types, so the table can have any schema. + +```toml +[plugin_config] +url = "http://localhost:8123"; +table = "events" +insert_format = "json_each_row" +``` + +### `row_binary` + +Accepts messages with a `Payload::Json` payload. At startup the connector fetches the table schema from `system.columns` and validates that all column types are supported. Messages are then serialised to ClickHouse's `RowBinaryWithDefaults` binary format, which is more efficient than JSON for large volumes. Review Comment: worth stating the minimum server version: `RowBinaryWithDefaults` was added in clickhouse 23.7, older servers reject the format. ########## core/connectors/sinks/clickhouse_sink/README.md: ########## @@ -0,0 +1,194 @@ +# ClickHouse Sink Connector + +The ClickHouse sink connector consumes messages from Iggy topics and inserts them into ClickHouse tables. Supports three insert formats: `json_each_row` (default), `row_binary`, and `string` passthrough. + +## Features + +- **Multiple Insert Formats**: Insert as `JSONEachRow`, `RowBinaryWithDefaults`, or raw string passthrough (CSV/TSV/JSON) +- **Schema Validation**: In `row_binary` mode, the table schema is fetched and validated at startup +- **Automatic Retries**: Configurable retry count and delay for transient errors +- **Batch Processing**: Insert messages in configurable batches via the stream configuration + +## Configuration + +```toml +type = "sink" +key = "clickhouse" +enabled = true +version = 0 +name = "ClickHouse sink" +path = "target/release/libiggy_connector_clickhouse_sink" + +[[streams]] +stream = "example_stream" +topics = ["example_topic"] +schema = "json" +batch_length = 1000 +poll_interval = "5ms" +consumer_group = "clickhouse_sink_connector" + +[plugin_config] +url = "http://localhost:8123"; +database = "default" +username = "default" +password = "" +table = "events" +insert_format = "json_each_row" +timeout_seconds = 30 +max_retries = 3 +retry_delay = 1 # seconds +verbose_logging = false +``` + +## Configuration Options + +| Option | Type | Default | Description | +| ------ | ---- | ------- | ----------- | +| `url` | string | required | ClickHouse HTTP endpoint | +| `table` | string | required | Target table name | +| `database` | string | `"default"` | ClickHouse database | +| `username` | string | `"default"` | ClickHouse username | +| `password` | string | `""` | ClickHouse password | +| `insert_format` | string | `"json_each_row"` | Insert format: `json_each_row`, `row_binary`, or `string` | +| `string_format` | string | `"json_each_row"` | ClickHouse format for `string` mode: `json_each_row`, `csv`, or `tsv` | +| `timeout_seconds` | u64 | `30` | HTTP request timeout | +| `max_retries` | u32 | `3` | Max retry attempts on transient errors | +| `retry_delay` | u64 | `1` | Delay between retries, in seconds | +| `verbose_logging` | bool | `false` | Log inserts at info level instead of debug | + +> **TODO:** `database` and `table` values are interpolated directly into SQL. Currently only +> single quotes are escaped; backslashes pass through unchanged, which can misparse string +> literals if a value ends with `\`. A future improvement should validate both fields against a +> strict allowlist (`^[A-Za-z_][A-Za-z0-9_]*$`) at config load and escape backslashes in SQL +> string literals. Deferred because these sinks run in operator-controlled environments where +> config values are trusted. + +## Insert Formats + +### `json_each_row` (Default) + +Accepts messages with a `Payload::Json` payload. Each message is sent as a JSON object on its own line using ClickHouse's `JSONEachRow` format. ClickHouse handles type coercion from the JSON values to the column types, so the table can have any schema. + +```toml +[plugin_config] +url = "http://localhost:8123"; +table = "events" +insert_format = "json_each_row" +``` + +### `row_binary` + +Accepts messages with a `Payload::Json` payload. At startup the connector fetches the table schema from `system.columns` and validates that all column types are supported. Messages are then serialised to ClickHouse's `RowBinaryWithDefaults` binary format, which is more efficient than JSON for large volumes. + +The table must already exist. Columns with an ordinary `DEFAULT` expression can be omitted from the message — the connector emits a `0x01` prefix byte to signal that the default should be used. `MATERIALIZED`, `ALIAS`, and `EPHEMERAL` columns are not insertable and are dropped from the schema entirely. + +**Supported types:** all integer and float primitives, `String`, `FixedString(n)`, `Bool`/`Boolean`, `UUID`, `Date`, `Date32`, `DateTime`, `DateTime64(p)`, `Decimal` (precision 1-38; `Decimal256` is not supported), `IPv4`, `IPv6`, `Enum8`, `Enum16`, and the composites `Nullable(T)`, `Array(T)`, `Map(K, V)`, `Tuple(...)`. `LowCardinality(T)` is transparently unwrapped to its inner type `T` (RowBinary serialises it identically). Review Comment: "all integer and float primitives" overclaims - `Int128`/`UInt128`/`Int256`/`UInt256` are regular insertable clickhouse types, but `parse_type` only handles the 8-64 bit widths, so a table containing one fails `open()` with "Unrecognised type". either add them (fixed-width little-endian like the rest, 16/32 bytes) or list them as unsupported here. ########## core/integration/tests/connectors/fixtures/clickhouse/container.rs: ########## @@ -0,0 +1,297 @@ +// 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. + +use crate::connectors::fixtures; +use integration::harness::TestBinaryError; +use reqwest::Client as HttpClient; +use reqwest::header::{HeaderMap, HeaderValue}; +use serde::Deserialize; +use std::time::Duration; +use testcontainers_modules::testcontainers::core::{IntoContainerPort, WaitFor}; +use testcontainers_modules::testcontainers::runners::AsyncRunner; +use testcontainers_modules::testcontainers::{ContainerAsync, GenericImage, ImageExt}; +use tokio::time::sleep; +use tracing::info; + +const CLICKHOUSE_IMAGE: &str = "clickhouse/clickhouse-server"; +const CLICKHOUSE_TAG: &str = "25.1"; +const CLICKHOUSE_HTTP_PORT: u16 = 8123; + +pub const CLICKHOUSE_TEST_USER: &str = "default"; +pub const CLICKHOUSE_TEST_PASSWORD: &str = "iggy_test"; + +pub const HEALTH_CHECK_ATTEMPTS: usize = 60; +pub const HEALTH_CHECK_INTERVAL_MS: u64 = 500; +pub const DEFAULT_POLL_ATTEMPTS: usize = 100; +pub const DEFAULT_POLL_INTERVAL_MS: u64 = 100; + +pub const DEFAULT_TEST_STREAM: &str = "test_stream"; +pub const DEFAULT_TEST_TOPIC: &str = "test_topic"; +pub const DEFAULT_DATABASE: &str = "default"; +pub const DEFAULT_SINK_TABLE: &str = "iggy_messages"; + +pub const ENV_SINK_URL: &str = "IGGY_CONNECTORS_SINK_CLICKHOUSE_PLUGIN_CONFIG_URL"; +pub const ENV_SINK_DATABASE: &str = "IGGY_CONNECTORS_SINK_CLICKHOUSE_PLUGIN_CONFIG_DATABASE"; +pub const ENV_SINK_TABLE: &str = "IGGY_CONNECTORS_SINK_CLICKHOUSE_PLUGIN_CONFIG_TABLE"; +pub const ENV_SINK_USERNAME: &str = "IGGY_CONNECTORS_SINK_CLICKHOUSE_PLUGIN_CONFIG_USERNAME"; +pub const ENV_SINK_PASSWORD: &str = "IGGY_CONNECTORS_SINK_CLICKHOUSE_PLUGIN_CONFIG_PASSWORD"; +pub const ENV_SINK_INSERT_FORMAT: &str = + "IGGY_CONNECTORS_SINK_CLICKHOUSE_PLUGIN_CONFIG_INSERT_FORMAT"; +pub const ENV_SINK_STRING_FORMAT: &str = + "IGGY_CONNECTORS_SINK_CLICKHOUSE_PLUGIN_CONFIG_STRING_FORMAT"; +pub const ENV_SINK_STREAMS_0_STREAM: &str = "IGGY_CONNECTORS_SINK_CLICKHOUSE_STREAMS_0_STREAM"; +pub const ENV_SINK_STREAMS_0_TOPICS: &str = "IGGY_CONNECTORS_SINK_CLICKHOUSE_STREAMS_0_TOPICS"; +pub const ENV_SINK_STREAMS_0_SCHEMA: &str = "IGGY_CONNECTORS_SINK_CLICKHOUSE_STREAMS_0_SCHEMA"; +pub const ENV_SINK_STREAMS_0_CONSUMER_GROUP: &str = + "IGGY_CONNECTORS_SINK_CLICKHOUSE_STREAMS_0_CONSUMER_GROUP"; +pub const ENV_SINK_PATH: &str = "IGGY_CONNECTORS_SINK_CLICKHOUSE_PATH"; + +/// DDL for the test table — matches the `TestMessage` struct. +pub const CREATE_TABLE_SQL: &str = " + CREATE TABLE IF NOT EXISTS iggy_messages ( Review Comment: none of the interesting column kinds are covered end-to-end: no `DEFAULT`/`MATERIALIZED`/`ALIAS` column (so the default-kind filtering and prefix-byte logic never run against a real server), no `UUID` column (a live round-trip would have caught the byte-order bug immediately), and every test message sends all six fields so the `0x01` use-default path is never exercised. worth extending this table. ########## core/connectors/sinks/clickhouse_sink/src/client.rs: ########## @@ -0,0 +1,499 @@ +// 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}")))?, + ); + let mut key_value = HeaderValue::from_str(password) + .map_err(|e| Error::InitError(format!("Invalid password header value: {e}")))?; + key_value.set_sensitive(true); + auth_headers.insert(KEY_HEADER, key_value); + + let inner = reqwest::Client::builder() + .timeout(timeout) + .default_headers(auth_headers) + .build() + .map_err(|e| Error::InitError(format!("Failed to build HTTP client: {e}")))?; + + let insert_url = format!( + "{}/?database={}&date_time_input_format=best_effort", + base_url, + urlencoded(&database), + ); + let insert_query = format!( + "INSERT INTO `{}`.`{}` FORMAT {}", + escape_backtick(&database), + escape_backtick(&table), + format_name, + ); + + Ok(ClickHouseClient { + inner, + base_url, + database, + table, + format_name, + insert_url, + insert_query, + }) + } + + /// Send `SELECT 1` to verify the server is reachable. + pub async fn ping(&self) -> Result<(), Error> { + let url = format!("{}/ping", self.base_url); + let response = self + .inner + .get(&url) + .send() + .await + .map_err(|e| Error::InitError(format!("Ping failed: {e}")))?; + + if response.status().is_success() { + Ok(()) + } else { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + error!("ClickHouse ping returned HTTP {status}: {body}"); + Err(Error::InitError(format!( + "ClickHouse ping returned HTTP {status}: {body}" + ))) + } + } + + /// Fetch the column definitions for the configured table. + /// Returns columns ordered by their position in the table definition. + pub async fn fetch_schema(&self) -> Result<Vec<Column>, Error> { + let query = format!( + "SELECT name, type, default_kind FROM system.columns \ + WHERE database = '{}' AND table = '{}' \ + ORDER BY position \ + FORMAT JSONEachRow", + escape_single_quote(&self.database), + escape_single_quote(&self.table), + ); + + let body = self.run_query(&query).await?; + let columns = parse_schema_body(&body)?; + + if columns.is_empty() { + error!( + "Table '{}' not found or has no columns in database '{}'", + self.table, self.database + ); + return Err(Error::InitError(format!( + "Table '{}' not found in database '{}'", + self.table, self.database + ))); + } + + info!( + "Fetched schema for table '{}': {} columns", + self.table, + columns.len() + ); + Ok(columns) + } + + /// Insert `body` into `table` using the given ClickHouse FORMAT string. + /// + /// Retries up to `max_retries` times on transient errors (network errors, + /// HTTP 429, HTTP 5xx). Does not retry on HTTP 4xx (data errors). + /// + /// # At-least-once semantics + /// + /// Each retry resends the identical body with no `insert_deduplication_token`. + /// If the server committed the batch but the response was lost, the retry + /// produces duplicate rows. Callers must tolerate this or handle + /// deduplication at read time. See the README for details. + // TODO: accept `Bytes` instead of `Vec<u8>` so callers can build into a + // `BytesMut`, freeze it zero-copy, and reuse a thread-local buffer across + // batches. `insert` is `pub(crate)` so the change is fully contained. + pub async fn insert( + &self, + body: Vec<u8>, + max_retries: u32, + retry_delay: Duration, + ) -> Result<(), Error> { + if body.is_empty() { + debug!("insert called with empty body — skipping"); + return Ok(()); + } + + let body = Bytes::from(body); + let mut attempts = 0u32; + loop { + let result = self + .inner + .post(&self.insert_url) + .header(CONTENT_TYPE, "application/octet-stream") + .query(&[("query", &self.insert_query)]) + .body(body.clone()) + .send() + .await; + + match result { + Ok(response) => { + let status = response.status(); + if status.is_success() { + debug!( + "Inserted {} bytes into {}.{} FORMAT {}", + body.len(), + self.database, + self.table, + self.format_name + ); + return Ok(()); + } + + let body_text = response.text().await.unwrap_or_default(); + + if is_retryable_status(status) { + attempts += 1; + if attempts >= max_retries { + error!( + "Insert failed after {attempts} attempts (HTTP {status}): {body_text}" + ); + return Err(Error::CannotStoreData(format!( + "HTTP {status}: {body_text}" + ))); + } + warn!( + "Retryable HTTP {status} on attempt {attempts}/{max_retries}: {body_text}" + ); + tokio::time::sleep(jittered_backoff(retry_delay, attempts)).await; + } else { + // Non-retryable 4xx data error. PermanentHttpError keeps + // the runtime circuit breaker from tripping on bad data. + error!("ClickHouse insert error HTTP {status}: {body_text}"); + return Err(Error::PermanentHttpError(format!( + "HTTP {status}: {body_text}" + ))); + } + } + Err(e) => { + // Network / timeout error — retryable. + attempts += 1; + if attempts >= max_retries { + error!("Insert failed after {attempts} attempts: {e}"); + return Err(Error::CannotStoreData(format!( + "Network error after {attempts} attempts: {e}" + ))); + } + warn!("Network error on attempt {attempts}/{max_retries}: {e}. Retrying..."); + tokio::time::sleep(retry_delay * attempts).await; + } + } + } + } + + // ── Private helpers ─────────────────────────────────────────────────────── + + /// Run a read-only query and return the response body as a String. + async fn run_query(&self, query: &str) -> Result<String, Error> { + let url = format!("{}/?database={}", self.base_url, urlencoded(&self.database)); + let response = self + .inner + .post(&url) + .body(query.to_owned()) + .send() + .await + .map_err(|e| Error::InitError(format!("Query failed: {e}")))?; + + let status = response.status(); + let body = response + .text() + .await + .map_err(|e| Error::InitError(format!("Failed to read response: {e}")))?; + + if !status.is_success() { + error!("Query returned HTTP {status}: {body}"); + return Err(Error::InitError(format!("HTTP {status}: {body}"))); + } + Ok(body) + } +} + +// ─── Helper types ───────────────────────────────────────────────────────────── + +#[derive(Deserialize)] +struct SchemaRow { + name: String, + r#type: String, + default_kind: Option<String>, +} + +/// Parse a `system.columns` JSONEachRow response into insertable columns. +/// +/// MATERIALIZED / ALIAS / EPHEMERAL columns are not part of the +/// RowBinaryWithDefaults insert set: ClickHouse expects zero bytes for them, +/// not even the DEFAULT flag. Emitting a prefix byte would shift the whole row +/// stream by one, so they are dropped entirely. `has_default` is set only for +/// ordinary DEFAULT columns. +fn parse_schema_body(body: &str) -> Result<Vec<Column>, Error> { + let mut columns = Vec::new(); + for line in body.lines() { + let line = line.trim(); + if line.is_empty() { + continue; + } + let row: SchemaRow = serde_json::from_str(line).map_err(|e| { + error!("Failed to parse schema row '{line}': {e}"); + Error::InitError(format!("Schema parse error: {e}")) + })?; + + let default_kind = row.default_kind.as_deref().unwrap_or(""); + match default_kind { + "MATERIALIZED" | "ALIAS" | "EPHEMERAL" => continue, + _ => {} + } + + let ch_type = parse_type(&row.r#type)?; + columns.push(Column { + name: row.name, + ch_type, + has_default: default_kind == "DEFAULT", + }); + } + Ok(columns) +} + +fn is_retryable_status(status: StatusCode) -> bool { + matches!( + status, + StatusCode::TOO_MANY_REQUESTS | StatusCode::REQUEST_TIMEOUT + ) || status.is_server_error() +} + +/// Exponential backoff with full jitter. +/// +/// Cap grows as `base * 2^attempt`, clamped to 60 s. The actual sleep is +/// a uniform random value in `[0, cap]`, so concurrent instances spread +/// their retries instead of thundering back together. +fn jittered_backoff(base: Duration, attempt: u32) -> Duration { + const MAX: Duration = Duration::from_secs(60); + let cap = base.saturating_mul(2u32.saturating_pow(attempt)).min(MAX); + let cap_ms = cap.as_millis() as u64; + Duration::from_millis(rand::rng().random_range(0..=cap_ms)) +} + +fn escape_single_quote(s: &str) -> String { + s.replace('\'', "''") +} + +fn escape_backtick(s: &str) -> String { Review Comment: doubling quotes/backticks isn't enough - clickhouse also treats backslash as an escape char in both single-quoted literals and quoted identifiers (syntax docs: "the same rules that apply for escaping in quoted identifiers also apply for string literals"). a table name ending in `\` produces `WHERE table = 'foo\'` which fails with "Single quoted string is not closed", and the backtick variant fails with "Back quoted string is not closed" (both verified live on 25.5). escape backslashes first in both helpers: `s.replace('\\', "\\\\")` before the quote doubling. the test at line 383 asserts the trailing backslash is inert, which is the opposite of what the clickhouse lexer does. ########## core/connectors/sinks/clickhouse_sink/src/sink.rs: ########## @@ -0,0 +1,187 @@ +// 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. + +use crate::body::{build_json_body, build_row_binary_body, build_string_body}; +use crate::{ClickHouseSink, InsertFormat, client::ClickHouseClient}; +use async_trait::async_trait; +use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata}; +use tracing::{debug, error, info, warn}; + +#[async_trait] +impl Sink for ClickHouseSink { + // ─── open ──────────────────────────────────────────────────────────────── + + async fn open(&mut self) -> Result<(), Error> { + info!( + "Opening ClickHouse sink connector ID: {} → {}/{} (format: {:?})", + self.id, self.config.url, self.config.table, self.insert_format, + ); + + let client = ClickHouseClient::new( + self.config.url.clone(), + self.database().to_owned(), + self.config.table.clone(), + self.insert_format + .clickhouse_format_name(self.string_format) + .to_owned(), + self.username(), + self.password(), + self.timeout(), + )?; + + let max_retries = self.max_retries(); + let retry_delay = self.retry_delay; + + let mut attempts = 0u32; + loop { + match client.ping().await { + Ok(()) => break, + Err(e) => { + attempts += 1; + if attempts >= max_retries { + error!("Ping failed after {attempts} attempt(s): {e}"); + return Err(e); + } + let backoff = retry_delay * attempts; + warn!( + "Ping failed (attempt {attempts}/{max_retries}): {e}. Retrying in {backoff:?}…" + ); + tokio::time::sleep(backoff).await; + } + } + } + info!("ClickHouse sink ID: {} — ping OK", self.id); + + // For RowBinary mode, fetch and validate the table schema at startup. + // This fails fast if the table doesn't exist or contains unsupported types. + if self.insert_format == InsertFormat::RowBinary { + let mut attempts = 0u32; + let schema = loop { + match client.fetch_schema().await { + Ok(schema) => break schema, + Err(e) => { + attempts += 1; + if attempts >= max_retries { + error!("fetch_schema failed after {attempts} attempt(s): {e}"); + return Err(e); + } + let backoff = retry_delay * attempts; + warn!( + "fetch_schema failed (attempt {attempts}/{max_retries}): {e}. Retrying in {backoff:?}…" + ); + tokio::time::sleep(backoff).await; + } + } + }; + info!( + "ClickHouse sink ID: {} — loaded schema ({} columns) for table '{}'", + self.id, + schema.len(), + self.config.table + ); + self.table_schema = Some(schema); Review Comment: schema is fetched once at `open()` and never refreshed. an `ALTER TABLE` while the sink runs (add/drop/reorder a column) desyncs the rowbinary layout and rows land corrupted or as phantom rows, silently - same failure mode the default-kind filtering protects against. at minimum document the restriction; refreshing the schema when an insert fails would be more robust. ########## core/connectors/sinks/clickhouse_sink/src/lib.rs: ########## @@ -0,0 +1,298 @@ +// 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. + +use iggy_connector_sdk::{Error, sink_connector}; +use secrecy::{ExposeSecret, SecretString}; +use serde::Deserialize; +use std::time::Duration; +use tokio::sync::Mutex; + +mod binary; +mod body; +mod client; +mod schema; +mod sink; + +sink_connector!(ClickHouseSink); + +const DEFAULT_DATABASE: &str = "default"; +const DEFAULT_USERNAME: &str = "default"; +const DEFAULT_PASSWORD: &str = ""; +const DEFAULT_TIMEOUT_SECONDS: u64 = 30; +const DEFAULT_MAX_RETRIES: u32 = 3; +const DEFAULT_RETRY_DELAY_SECS: u64 = 1; + +#[derive(Debug, Clone, Deserialize)] +pub struct ClickHouseSinkConfig { + pub url: String, + pub database: Option<String>, + pub username: Option<String>, + pub password: Option<SecretString>, + pub table: String, + /// "json_each_row" (default), "row_binary", or "string" + pub insert_format: Option<InsertFormat>, + /// "json_each_row" (default), "csv", or "tsv" — only used when insert_format = "string" + pub string_format: Option<StringFormat>, + pub timeout_seconds: Option<u64>, + pub max_retries: Option<u32>, + /// Delay between retry attempts, in seconds. + pub retry_delay: Option<u64>, + pub verbose_logging: Option<bool>, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum InsertFormat { + #[default] + JsonEachRow, + RowBinary, + #[serde(rename = "string")] + StringPassthrough, +} + +impl InsertFormat { + pub fn clickhouse_format_name(&self, string_fmt: StringFormat) -> &'static str { + match self { + InsertFormat::JsonEachRow => "JSONEachRow", + InsertFormat::RowBinary => "RowBinaryWithDefaults", + InsertFormat::StringPassthrough => string_fmt.clickhouse_format_name(), + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum StringFormat { + #[default] + JsonEachRow, + Csv, + Tsv, +} + +impl StringFormat { + pub fn clickhouse_format_name(&self) -> &'static str { + match self { + StringFormat::JsonEachRow => "JSONEachRow", + StringFormat::Csv => "CSV", + StringFormat::Tsv => "TSV", + } + } + + /// All three formats use newline as a row delimiter. + pub fn requires_newline(&self) -> bool { Review Comment: matches all three variants of a three-variant enum - always true. either return `true` with a note that all current formats are newline-delimited, or drop the method and append the newline unconditionally in body.rs. ########## core/connectors/sinks/clickhouse_sink/src/binary.rs: ########## @@ -0,0 +1,1679 @@ +// 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 = u16::try_from(coerce_to_days(value)?).map_err(|_| { + error!("Value out of range for Date (1970-01-01..2149-06-06)"); + Error::InvalidRecord + })?; + buf.extend_from_slice(&days.to_le_bytes()); + } + ChType::Date32 => { + let days = i32::try_from(coerce_to_days(value)?).map_err(|_| { + error!("Value out of range for Date32"); + Error::InvalidRecord + })?; + buf.extend_from_slice(&days.to_le_bytes()); + } + ChType::DateTime => { + // Unix seconds as UInt32. Accept integer or RFC 3339 string. + let secs = u32::try_from(coerce_to_unix_seconds(value)?).map_err(|_| { + error!("Value out of range for DateTime (1970-01-01..2106-02-07)"); + Error::InvalidRecord + })?; + buf.extend_from_slice(&secs.to_le_bytes()); + } + ChType::DateTime64(precision) => { + // Unix time scaled by 10^precision as Int64. + // precision is validated to be 0-9 by the schema parser, so + // 10i64.pow(*precision as u32) cannot overflow i64::MAX. + let scale = 10i64.pow(*precision as u32); + let scaled = match value { + OwnedValue::Static(simd_json::StaticNode::I64(n)) => { + // Integer fast path: multiply in i64 to avoid f64 precision loss. + // At precision=9, a current-era timestamp (~1.7e9 s) * 1e9 = ~1.7e18, + // which exceeds 2^53 and would lose ~256 ns if routed through f64. + n.checked_mul(scale).ok_or_else(|| { + error!("DateTime64 overflow"); + Error::InvalidRecord + })? + } + OwnedValue::Static(simd_json::StaticNode::U64(n)) => i64::try_from(*n) + .ok() + .and_then(|n| n.checked_mul(scale)) + .ok_or_else(|| { + error!("DateTime64 overflow"); + Error::InvalidRecord + })?, + _ => { + // Float or string inputs: f64 path is acceptable since floats + // already carry sub-second fractions, and strings go through + // parse_datetime_string which returns fractional seconds. + let secs_f64 = coerce_to_unix_seconds_f64(value)?; + (secs_f64 * scale as f64).round() as i64 Review Comment: `(secs_f64 * scale as f64).round() as i64` saturates instead of erroring - rust float-to-int `as` clamps to `i64::MAX`. a far-future datetime string (the ymd parser accepts any i32 year) or a huge float silently becomes 2262-04-11 23:47:16.854775807 at precision 9 instead of failing the record. the integer path right above uses `checked_mul` + error, so the two paths disagree. clickhouse won't catch it either - rowbinary input does no range validation. bound-check the f64 before the cast. ########## core/connectors/runtime/example_config/connectors/clickhouse_sink.toml: ########## @@ -0,0 +1,56 @@ +# 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. + +type = "sink" +key = "clickhouse" +enabled = true +version = 0 +name = "ClickHouse sink" +path = "target/release/libiggy_connector_clickhouse_sink" +verbose = false + +[[streams]] +stream = "qw" Review Comment: `stream = "qw"` looks like a dev-scratch leftover - the crate's config.toml uses `example_stream`/`events` while this example uses `qw`/`users`. align the two. ########## core/connectors/sinks/clickhouse_sink/Cargo.toml: ########## @@ -0,0 +1,52 @@ +# 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. + +[package] +name = "iggy_connector_clickhouse_sink" +version = "0.1.0" +description = "Iggy ClickHouse sink connector for streaming messages into ClickHouse" +edition = "2024" +license = "Apache-2.0" +keywords = ["iggy", "messaging", "streaming", "clickhouse", "sink"] +categories = ["command-line-utilities", "database", "network-programming"] +homepage = "https://iggy.apache.org"; +documentation = "https://iggy.apache.org/docs"; +repository = "https://github.com/apache/iggy"; +readme = "../../README.md" + +[package.metadata.cargo-machete] +ignored = ["dashmap"] + +[lib] +crate-type = ["cdylib", "lib"] + +[dependencies] +async-trait = { workspace = true } +bytes = { workspace = true } +dashmap = { workspace = true } Review Comment: `dashmap` has zero uses in this crate - the machete suppression above masks the problem rather than fixing it. the ffi macro resolves `DashMap` through the sdk's re-export, so sibling sinks (doris, http, mongodb) build without declaring it. drop the dep and the suppression. ########## core/connectors/sinks/clickhouse_sink/src/client.rs: ########## @@ -0,0 +1,499 @@ +// 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}")))?, + ); + let mut key_value = HeaderValue::from_str(password) + .map_err(|e| Error::InitError(format!("Invalid password header value: {e}")))?; + key_value.set_sensitive(true); + auth_headers.insert(KEY_HEADER, key_value); + + let inner = reqwest::Client::builder() + .timeout(timeout) + .default_headers(auth_headers) + .build() + .map_err(|e| Error::InitError(format!("Failed to build HTTP client: {e}")))?; + + let insert_url = format!( + "{}/?database={}&date_time_input_format=best_effort", + base_url, + urlencoded(&database), + ); + let insert_query = format!( + "INSERT INTO `{}`.`{}` FORMAT {}", + escape_backtick(&database), + escape_backtick(&table), + format_name, + ); + + Ok(ClickHouseClient { + inner, + base_url, + database, + table, + format_name, + insert_url, + insert_query, + }) + } + + /// Send `SELECT 1` to verify the server is reachable. + pub async fn ping(&self) -> Result<(), Error> { + let url = format!("{}/ping", self.base_url); + let response = self + .inner + .get(&url) + .send() + .await + .map_err(|e| Error::InitError(format!("Ping failed: {e}")))?; + + if response.status().is_success() { + Ok(()) + } else { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + error!("ClickHouse ping returned HTTP {status}: {body}"); + Err(Error::InitError(format!( + "ClickHouse ping returned HTTP {status}: {body}" + ))) + } + } + + /// Fetch the column definitions for the configured table. + /// Returns columns ordered by their position in the table definition. + pub async fn fetch_schema(&self) -> Result<Vec<Column>, Error> { + let query = format!( + "SELECT name, type, default_kind FROM system.columns \ + WHERE database = '{}' AND table = '{}' \ + ORDER BY position \ + FORMAT JSONEachRow", + escape_single_quote(&self.database), + escape_single_quote(&self.table), + ); + + let body = self.run_query(&query).await?; + let columns = parse_schema_body(&body)?; + + if columns.is_empty() { + error!( + "Table '{}' not found or has no columns in database '{}'", + self.table, self.database + ); + return Err(Error::InitError(format!( + "Table '{}' not found in database '{}'", + self.table, self.database + ))); + } + + info!( + "Fetched schema for table '{}': {} columns", + self.table, + columns.len() + ); + Ok(columns) + } + + /// Insert `body` into `table` using the given ClickHouse FORMAT string. + /// + /// Retries up to `max_retries` times on transient errors (network errors, + /// HTTP 429, HTTP 5xx). Does not retry on HTTP 4xx (data errors). + /// + /// # At-least-once semantics + /// + /// Each retry resends the identical body with no `insert_deduplication_token`. + /// If the server committed the batch but the response was lost, the retry + /// produces duplicate rows. Callers must tolerate this or handle + /// deduplication at read time. See the README for details. + // TODO: accept `Bytes` instead of `Vec<u8>` so callers can build into a + // `BytesMut`, freeze it zero-copy, and reuse a thread-local buffer across + // batches. `insert` is `pub(crate)` so the change is fully contained. + pub async fn insert( + &self, + body: Vec<u8>, + max_retries: u32, + retry_delay: Duration, + ) -> Result<(), Error> { + if body.is_empty() { + debug!("insert called with empty body — skipping"); + return Ok(()); + } + + let body = Bytes::from(body); + let mut attempts = 0u32; + loop { + let result = self + .inner + .post(&self.insert_url) + .header(CONTENT_TYPE, "application/octet-stream") + .query(&[("query", &self.insert_query)]) + .body(body.clone()) + .send() + .await; + + match result { + Ok(response) => { + let status = response.status(); + if status.is_success() { + debug!( + "Inserted {} bytes into {}.{} FORMAT {}", + body.len(), + self.database, + self.table, + self.format_name + ); + return Ok(()); + } + + let body_text = response.text().await.unwrap_or_default(); + + if is_retryable_status(status) { + attempts += 1; + if attempts >= max_retries { + error!( + "Insert failed after {attempts} attempts (HTTP {status}): {body_text}" + ); + return Err(Error::CannotStoreData(format!( + "HTTP {status}: {body_text}" + ))); + } + warn!( + "Retryable HTTP {status} on attempt {attempts}/{max_retries}: {body_text}" + ); + tokio::time::sleep(jittered_backoff(retry_delay, attempts)).await; + } else { + // Non-retryable 4xx data error. PermanentHttpError keeps + // the runtime circuit breaker from tripping on bad data. + error!("ClickHouse insert error HTTP {status}: {body_text}"); + return Err(Error::PermanentHttpError(format!( + "HTTP {status}: {body_text}" + ))); + } + } + Err(e) => { + // Network / timeout error — retryable. + attempts += 1; + if attempts >= max_retries { + error!("Insert failed after {attempts} attempts: {e}"); + return Err(Error::CannotStoreData(format!( + "Network error after {attempts} attempts: {e}" + ))); + } + warn!("Network error on attempt {attempts}/{max_retries}: {e}. Retrying..."); + tokio::time::sleep(retry_delay * attempts).await; Review Comment: retryable http statuses get jittered exponential backoff (line 227) but network/timeout errors back off linearly - and a clickhouse outage surfaces as network errors, which is exactly the case where many sink instances retry in lockstep. same linear backoff in the `open()` ping and fetch_schema loops in sink.rs. the readme documents the split, but the network path is the one that needs jitter most. -- 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]
