hubcio commented on code in PR #2886: URL: https://github.com/apache/iggy/pull/2886#discussion_r3505167605
########## core/connectors/sinks/clickhouse_sink/src/client.rs: ########## @@ -0,0 +1,453 @@ +// 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( + KEY_HEADER, + HeaderValue::from_str(password) + .map_err(|e| Error::InitError(format!("Invalid password header value: {e}")))?, + ); + + 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 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 ch_type = parse_type(&row.r#type)?; + let has_default = matches!( + row.default_kind.as_deref(), + Some("DEFAULT") | Some("MATERIALIZED") | Some("ALIAS") Review Comment: `fetch_schema` marks `has_default = true` for `MATERIALIZED` and `ALIAS` columns and pushes every `system.columns` row into the schema `Vec`. but with `FORMAT RowBinaryWithDefaults` clickhouse expects a prefix byte only for columns in the implicit insert set - ordinary + `DEFAULT`. `MATERIALIZED`/`ALIAS` (and `EPHEMERAL`) are not insertable, so clickhouse expects zero bytes for them, not even the `0x01` flag. `serialize_row` still emits one prefix byte per column, so each such column injects a phantom byte and the whole row stream shifts by one - silent corruption, or a `Cannot read all data` reject, on any table that has a materialized/alias column. concretely for `(id UInt64, m UInt64 MATERIALIZED id*2, name String)` clickhouse expects 2 slots per row and the connector emits 3. fix: in `fetch_schema` keep only columns whose `default_kind` is `''` or `DEFAULT` (drop `MATERIALIZED`/`ALIAS`/`EPHEMERAL` from the `Vec` entirely), and set `has_default` true only for `DEFAULT`. just flipping `has_default` to false is not enough - the column would still get a `0x00` + value slot and desync the same way. the test table (`iggy_messages`) has no default/materialized/alias columns, so this path is currently untested and ships green. ########## core/connectors/sinks/clickhouse_sink/src/binary.rs: ########## @@ -0,0 +1,1600 @@ +// 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 + let is_null_or_absent = field_value.map(|v| v.is_null()).unwrap_or(true); + if is_null_or_absent && col.has_default { Review Comment: for a column that is both `Nullable` and has a `DEFAULT`, an explicit json `null` hits `is_null_or_absent && col.has_default` and emits `0x01` (use default) before ever reaching the `Nullable` arm. so an explicit null is stored as the default value instead of NULL. only diverges when the default is non-null, but then it's silent. gate the default shortcut on the field being absent (`field_value.is_none()`) so an explicit null falls through to the nullable path (`0x00` + null marker). ########## core/connectors/sinks/clickhouse_sink/src/binary.rs: ########## @@ -0,0 +1,1600 @@ +// 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 + 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 + })?; + // 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; + 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 + } + }; + buf.extend_from_slice(&scaled.to_le_bytes()); + } + + // ── Decimal ────────────────────────────────────────────────────────── + ChType::Decimal(precision, scale) => { + let int_val = coerce_decimal(value, *precision, *scale)?; + 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).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).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 + })?; + let mut entries: Vec<(&str, &OwnedValue)> = + obj.iter().map(|(k, v)| (k.as_ref(), v)).collect(); + // Simple string-key sort; serialized-key sorting would differ for + // non-string key types (e.g. Int32 "10" < "2" by bytes but not by string), + // but non-string-keyed maps are uncommon in practice. + entries.sort_unstable_by_key(|(k, _)| *k); + write_varint(entries.len() as u64, buf); + for (k, v) in entries { + match key_type.as_ref() { + ChType::String => write_string(k.as_bytes(), buf), + _ => { + let key_val = OwnedValue::String(k.to_owned()); + 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(crate) 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), Review Comment: `coerce_i64` does `*n as i64` for the `U64` case, which wraps values above `i64::MAX` to negative. simd_json parses json integers larger than `i64::MAX` as `U64`, so `{"id": 18446744073709551615}` into an `Int64` column silently stores `-1`. `Int8`/`Int16`/`Int32` go through `coerce_i64` first as well, so the wrap happens before their `try_from` and `-1` sails through. `coerce_u64` already handles the mirror case with `u64::try_from` - do the same here with `i64::try_from(*n)` and return `InvalidRecord` on overflow. -- 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]
