hubcio commented on code in PR #2886: URL: https://github.com/apache/iggy/pull/2886#discussion_r3343313317
########## core/connectors/sinks/clickhouse_sink/src/binary.rs: ########## @@ -0,0 +1,1238 @@ +/* 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 crate::schema::{ChType, Column}; +use iggy_connector_sdk::Error; +use simd_json::OwnedValue; +use simd_json::prelude::{TypedScalarValue, ValueAsArray, ValueAsObject}; +use tracing::error; + +// ─── 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 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 + let is_null_or_absent = field_value.map(|v| v.is_null()).unwrap_or(true); + if is_null_or_absent && 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 + })?; + let bytes = hex::decode(hex_str).map_err(|_| { + error!("Cannot decode UUID hex: {s}"); + Error::InvalidRecord + })?; + // ClickHouse UUID layout: first 8 bytes reversed, second 8 bytes reversed + let mut uuid_buf = [0u8; 16]; + uuid_buf[..8].copy_from_slice(&bytes[..8]); + uuid_buf[8..].copy_from_slice(&bytes[8..]); + uuid_buf[..8].reverse(); + uuid_buf[8..].reverse(); + buf.extend_from_slice(&uuid_buf); + } + + // ── 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; + 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 secs_f64 = coerce_to_unix_seconds_f64(value)?; + let scale = 10i64.pow(*precision as u32) as f64; + let scaled = (secs_f64 * scale).round() as i64; + buf.extend_from_slice(&scaled.to_le_bytes()); + } + + // ── Decimal ────────────────────────────────────────────────────────── + ChType::Decimal(precision, scale) => { + let f = coerce_f64(value)?; + let scale_factor = 10f64.powi(*scale as i32); + let int_val = (f * scale_factor).round() as i128; + if *precision <= 9 { + buf.extend_from_slice( + &i32::try_from(int_val) + .map_err(|_| { + error!("Decimal value out of range for precision {precision}"); + Error::InvalidRecord + })? + .to_le_bytes(), + ); + } else if *precision <= 18 { + buf.extend_from_slice( + &i64::try_from(int_val) + .map_err(|_| { + error!("Decimal value out of range for precision {precision}"); + Error::InvalidRecord + })? + .to_le_bytes(), + ); + } else { + buf.extend_from_slice(&int_val.to_le_bytes()); + } + } + + // ── IP addresses ───────────────────────────────────────────────────── + ChType::IPv4 => { + let s = coerce_to_string(value)?; + let addr: std::net::Ipv4Addr = s.parse().map_err(|_| { + error!("Invalid IPv4 address: {s}"); + Error::InvalidRecord + })?; + buf.extend_from_slice(&u32::from(addr).to_le_bytes()); + } + ChType::IPv6 => { + let s = coerce_to_string(value)?; + let addr: std::net::Ipv6Addr = s.parse().map_err(|_| { + error!("Invalid IPv6 address: {s}"); + Error::InvalidRecord + })?; + buf.extend_from_slice(&addr.octets()); // big-endian + } + + // ── Enums ──────────────────────────────────────────────────────────── + ChType::Enum8(map) => { + let s = coerce_to_string(value)?; + let v = map.get(s.as_str()).ok_or_else(|| { + error!("Unknown Enum8 value: {s}"); + Error::InvalidRecord + })?; + buf.push(*v as u8); + } + ChType::Enum16(map) => { + let s = coerce_to_string(value)?; + let v = map.get(s.as_str()).ok_or_else(|| { + error!("Unknown Enum16 value: {s}"); + Error::InvalidRecord + })?; + buf.extend_from_slice(&v.to_le_bytes()); + } + + // ── Composites ─────────────────────────────────────────────────────── + ChType::Array(elem_type) => { + let arr = value.as_array().ok_or_else(|| { + error!("Expected JSON array for Array type, got: {value:?}"); + Error::InvalidRecord + })?; + write_varint(arr.len() as u64, buf); + for elem in arr { + serialize_value(elem, elem_type, buf)?; + } + } + ChType::Map(key_type, val_type) => { + let obj = value.as_object().ok_or_else(|| { + error!("Expected JSON object for Map type, got: {value:?}"); + Error::InvalidRecord + })?; + write_varint(obj.len() as u64, buf); + for (k, v) in obj { + // Map keys must be serialisable as the key type. JSON object + // keys are always strings, so we wrap them in OwnedValue::String. + let key_val = OwnedValue::String(k.clone()); + serialize_value(&key_val, key_type, buf)?; + serialize_value(v, val_type, buf)?; + } + } + ChType::Tuple(field_types) => { + // Tuples may arrive as JSON arrays (unnamed) or objects (named). + match value { + OwnedValue::Array(arr) => { + if arr.len() != field_types.len() { + error!( + "Tuple length mismatch: expected {}, got {}", + field_types.len(), + arr.len() + ); + return Err(Error::InvalidRecord); + } + for (elem, (_, ft)) in arr.iter().zip(field_types.iter()) { + serialize_value(elem, ft, buf)?; + } + } + OwnedValue::Object(obj) => { + // Named tuple: look up each field by name to guarantee + // correct ordering regardless of JSON key iteration order. + if obj.len() != field_types.len() { + error!( + "Tuple length mismatch: expected {}, got {}", + field_types.len(), + obj.len() + ); + return Err(Error::InvalidRecord); + } + for (name_opt, ft) in field_types.iter() { + let Some(name) = name_opt else { + error!("Cannot serialise unnamed tuple fields from a JSON object"); + return Err(Error::InvalidRecord); + }; + let Some(v) = obj.get(name.as_str()) else { + error!("Tuple field '{name}' not found in JSON object"); + return Err(Error::InvalidRecord); + }; + serialize_value(v, ft, buf)?; + } + } + other => { + error!("Expected JSON array or object for Tuple type, got: {other:?}"); + return Err(Error::InvalidRecord); + } + } + } + } + Ok(()) +} + +// ─── Low-level helpers ──────────────────────────────────────────────────────── + +/// Write a ClickHouse-style unsigned LEB128 varint (7 bits per byte, MSB = continuation). +pub fn write_varint(mut n: u64, buf: &mut Vec<u8>) { + loop { + let byte = (n & 0x7F) as u8; + n >>= 7; + if n == 0 { + buf.push(byte); + break; + } else { + buf.push(byte | 0x80); + } + } +} + +/// Write a string: varint length prefix + UTF-8 bytes. +fn write_string(bytes: &[u8], buf: &mut Vec<u8>) { + write_varint(bytes.len() as u64, buf); + buf.extend_from_slice(bytes); +} + +/// Write a zero / null value for a column that is absent and has no default. +/// Nullable columns get the null marker; non-nullable columns are an error. +fn write_zero_or_null(ch_type: &ChType, buf: &mut Vec<u8>, col_name: &str) -> Result<(), Error> { + match ch_type { + ChType::Nullable(_) => { + buf.push(0x01); // null + Ok(()) + } + _ => { + error!( + "Column '{col_name}' is non-nullable with no default, but is absent from the message" + ); + Err(Error::InvalidRecord) + } + } +} + +// ─── Value coercion helpers ─────────────────────────────────────────────────── + +fn coerce_i64(value: &OwnedValue) -> Result<i64, Error> { + match value { + OwnedValue::Static(simd_json::StaticNode::I64(n)) => Ok(*n), + OwnedValue::Static(simd_json::StaticNode::U64(n)) => Ok(*n as i64), + OwnedValue::Static(simd_json::StaticNode::F64(f)) => Ok(*f as i64), + OwnedValue::String(s) => s.parse::<i64>().map_err(|_| { + error!("Cannot parse '{s}' as integer"); + Error::InvalidRecord + }), + other => { + error!("Cannot coerce {other:?} to integer"); + Err(Error::InvalidRecord) + } + } +} + +fn coerce_u64(value: &OwnedValue) -> Result<u64, Error> { + match value { + OwnedValue::Static(simd_json::StaticNode::U64(n)) => Ok(*n), + OwnedValue::Static(simd_json::StaticNode::I64(n)) => u64::try_from(*n).map_err(|_| { + error!("Cannot coerce negative integer {n} to unsigned integer"); + Error::InvalidRecord + }), + OwnedValue::Static(simd_json::StaticNode::F64(f)) => Ok(*f as u64), + OwnedValue::String(s) => s.parse::<u64>().map_err(|_| { + error!("Cannot parse '{s}' as unsigned integer"); + Error::InvalidRecord + }), + other => { + error!("Cannot coerce {other:?} to unsigned integer"); + Err(Error::InvalidRecord) + } + } +} + +fn coerce_f64(value: &OwnedValue) -> Result<f64, Error> { + match value { + OwnedValue::Static(simd_json::StaticNode::F64(f)) => Ok(*f), + OwnedValue::Static(simd_json::StaticNode::I64(n)) => Ok(*n as f64), + OwnedValue::Static(simd_json::StaticNode::U64(n)) => Ok(*n as f64), + OwnedValue::String(s) => s.parse::<f64>().map_err(|_| { + error!("Cannot parse '{s}' as float"); + Error::InvalidRecord + }), + other => { + error!("Cannot coerce {other:?} to float"); + Err(Error::InvalidRecord) + } + } +} + +fn coerce_to_string(value: &OwnedValue) -> Result<String, Error> { + match value { + OwnedValue::String(s) => Ok(s.to_string()), + OwnedValue::Static(simd_json::StaticNode::I64(n)) => Ok(n.to_string()), + OwnedValue::Static(simd_json::StaticNode::U64(n)) => Ok(n.to_string()), + OwnedValue::Static(simd_json::StaticNode::F64(f)) => Ok(f.to_string()), + OwnedValue::Static(simd_json::StaticNode::Bool(b)) => Ok(b.to_string()), + other => { + error!("Cannot coerce {other:?} to string"); + Err(Error::InvalidRecord) + } + } +} + +/// Returns days since 1970-01-01. Accepts integer (days) or "YYYY-MM-DD" string. +fn coerce_to_days(value: &OwnedValue) -> Result<i64, Error> { + match value { + OwnedValue::Static(simd_json::StaticNode::I64(n)) => Ok(*n), + OwnedValue::Static(simd_json::StaticNode::U64(n)) => Ok(*n as i64), + OwnedValue::String(s) => { + // Parse "YYYY-MM-DD" manually + let parts: Vec<&str> = s.splitn(3, '-').collect(); + if parts.len() != 3 { + error!("Invalid date string: {s}"); + return Err(Error::InvalidRecord); + } + let (y, m, d) = parse_ymd(parts[0], parts[1], parts[2]).map_err(|_| { + error!("Cannot parse date: {s}"); + Error::InvalidRecord + })?; + Ok(ymd_to_days(y, m, d)) + } + other => { + error!("Cannot coerce {other:?} to Date"); + Err(Error::InvalidRecord) + } + } +} + +/// Returns Unix seconds as f64 (fractional seconds for sub-second precision). +/// Accepts integer, float, or RFC 3339 / ISO 8601 string. +fn coerce_to_unix_seconds_f64(value: &OwnedValue) -> Result<f64, Error> { + match value { + OwnedValue::Static(simd_json::StaticNode::I64(n)) => Ok(*n as f64), + OwnedValue::Static(simd_json::StaticNode::U64(n)) => Ok(*n as f64), + OwnedValue::Static(simd_json::StaticNode::F64(f)) => Ok(*f), + OwnedValue::String(s) => parse_datetime_string(s), + other => { + error!("Cannot coerce {other:?} to DateTime"); + Err(Error::InvalidRecord) + } + } +} + +/// Returns Unix seconds as i64 (truncates fractional seconds). +fn coerce_to_unix_seconds(value: &OwnedValue) -> Result<i64, Error> { + Ok(coerce_to_unix_seconds_f64(value)? as i64) +} + +/// Parse "YYYY-MM-DDThh:mm:ss[.frac][Z|±hh:mm]" into Unix seconds (f64). +/// This is a minimal parser sufficient for common ISO 8601 / RFC 3339 formats. +fn parse_datetime_string(s: &str) -> Result<f64, Error> { + // Split on 'T' or ' ' for date-time separator + let (date_part, time_part) = if let Some(idx) = s.find('T').or_else(|| s.find(' ')) { + (&s[..idx], &s[idx + 1..]) + } else { + // Date-only string — treat as midnight UTC + (s, "00:00:00") + }; + + let date_parts: Vec<&str> = date_part.splitn(3, '-').collect(); + if date_parts.len() != 3 { + error!("Cannot parse datetime string: {s}"); + return Err(Error::InvalidRecord); + } + let (y, m, d) = parse_ymd(date_parts[0], date_parts[1], date_parts[2]).map_err(|_| { + error!("Cannot parse date component of: {s}"); + Error::InvalidRecord + })?; + + // Strip timezone suffix and optional fractional seconds + let (time_no_tz, tz_offset_secs) = strip_timezone(time_part); + let frac_secs = parse_time(time_no_tz).map_err(|_| { + error!("Cannot parse time component of: {s}"); + Error::InvalidRecord + })?; + + let days = ymd_to_days(y, m, d) as f64; + Ok(days * 86400.0 + frac_secs - tz_offset_secs as f64) +} + +fn parse_ymd(y: &str, m: &str, d: &str) -> Result<(i32, u32, u32), ()> { + let year: i32 = y.trim().parse().map_err(|_| ())?; + let month: u32 = m.trim().parse().map_err(|_| ())?; + let day: u32 = d.trim().parse().map_err(|_| ())?; Review Comment: no problem. -- 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]
