diff --git a/Cargo.toml b/Cargo.toml index 391ab9c4394ba..4dd4f56c97ff3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -943,7 +943,7 @@ sinks-azure_logs_ingestion = ["dep:azure_core", "dep:azure_identity", "dep:azure sinks-azure_monitor_logs = [] sinks-blackhole = [] sinks-chronicle = [] -sinks-clickhouse = ["dep:nom", "dep:rust_decimal", "codecs-arrow"] +sinks-clickhouse = ["dep:nom", "dep:rust_decimal", "codecs-arrow", "arrow/ipc_compression"] sinks-console = [] sinks-databend = ["dep:databend-client"] sinks-databricks-zerobus = ["dep:databricks-zerobus-ingest-sdk", "codecs-arrow", "arrow/ipc_compression"] diff --git a/changelog.d/25756_clickhouse_arrow_ipc_compression.enhancement.md b/changelog.d/25756_clickhouse_arrow_ipc_compression.enhancement.md new file mode 100644 index 0000000000000..bcb5256d93a18 --- /dev/null +++ b/changelog.d/25756_clickhouse_arrow_ipc_compression.enhancement.md @@ -0,0 +1,3 @@ +The `clickhouse` sink's `arrow_stream` batch encoding now supports block-level compression of the Arrow IPC buffers via a `compression` option (`none`, `lz4_frame`, or `zstd`, defaulting to `none`). Compression is applied inside the IPC stream so each column buffer is compressed independently, using substantially less CPU than an equivalent whole-payload HTTP `gzip` `Content-Encoding` at a comparable ratio. Enabling `compression` (`lz4_frame` or `zstd`) requires ClickHouse 23.11 or newer; the sink validates the server version at startup and rejects the configuration on older servers. When Arrow IPC compression is enabled, the sink also disables the top-level HTTP `compression` (which defaults to `gzip`) to avoid double-compressing the payload. + +authors: benjamin-awd diff --git a/lib/codecs/Cargo.toml b/lib/codecs/Cargo.toml index cb9bad72b67d5..003fa188be5cc 100644 --- a/lib/codecs/Cargo.toml +++ b/lib/codecs/Cargo.toml @@ -14,7 +14,7 @@ path = "tests/bin/generate-avro-fixtures.rs" [dependencies] apache-avro = { version = "0.20.0", default-features = false } -arrow = { version = "58.2.0", default-features = false, features = ["ipc", "json"], optional = true } +arrow = { version = "58.2.0", default-features = false, features = ["ipc", "ipc_compression", "json"], optional = true } parquet = { version = "58.2.0", default-features = false, features = [ "arrow", "snap", diff --git a/lib/codecs/src/encoding/format/arrow.rs b/lib/codecs/src/encoding/format/arrow.rs index 0c67317b7f926..87233b56b02b1 100644 --- a/lib/codecs/src/encoding/format/arrow.rs +++ b/lib/codecs/src/encoding/format/arrow.rs @@ -7,7 +7,10 @@ use arrow::{ datatypes::{DataType, Field, Fields, Schema, SchemaRef}, error::ArrowError, - ipc::writer::StreamWriter, + ipc::{ + CompressionType, + writer::{IpcWriteOptions, StreamWriter}, + }, json::reader::ReaderBuilder, record_batch::RecordBatch, }; @@ -29,6 +32,35 @@ pub trait SchemaProvider: Send + Sync + std::fmt::Debug { async fn get_schema(&self) -> Result; } +/// Block-level compression applied to Arrow IPC record batch buffers. +/// +/// This is columnar compression applied inside the Arrow IPC stream itself, so each +/// buffer is compressed independently and can be decompressed per column by the reader. +#[configurable_component] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum ArrowIpcCompression { + /// No compression. + #[default] + None, + + /// LZ4 frame compression. + Lz4Frame, + + /// Zstandard compression. + Zstd, +} + +impl From for Option { + fn from(value: ArrowIpcCompression) -> Self { + match value { + ArrowIpcCompression::None => None, + ArrowIpcCompression::Lz4Frame => Some(CompressionType::LZ4_FRAME), + ArrowIpcCompression::Zstd => Some(CompressionType::ZSTD), + } + } +} + /// Configuration for Arrow IPC stream serialization #[configurable_component] #[derive(Clone, Default)] @@ -49,6 +81,13 @@ pub struct ArrowStreamSerializerConfig { #[serde(default)] #[configurable(derived)] pub allow_nullable_fields: bool, + + /// Block-level compression applied to the Arrow IPC record batch buffers. + /// + /// Compresses each buffer inside the IPC stream. + #[serde(default, skip_serializing_if = "vector_core::serde::is_default")] + #[configurable(derived)] + pub compression: ArrowIpcCompression, } impl std::fmt::Debug for ArrowStreamSerializerConfig { @@ -62,6 +101,7 @@ impl std::fmt::Debug for ArrowStreamSerializerConfig { .map(|s| format!("{} fields", s.fields().len())), ) .field("allow_nullable_fields", &self.allow_nullable_fields) + .field("compression", &self.compression) .finish() } } @@ -72,6 +112,7 @@ impl ArrowStreamSerializerConfig { Self { schema: Some(schema), allow_nullable_fields: false, + compression: ArrowIpcCompression::None, } } @@ -90,6 +131,7 @@ impl ArrowStreamSerializerConfig { #[derive(Clone, Debug)] pub struct ArrowStreamSerializer { schema: SchemaRef, + compression: ArrowIpcCompression, } impl ArrowStreamSerializer { @@ -108,6 +150,7 @@ impl ArrowStreamSerializer { /// Create a new ArrowStreamSerializer with the given configuration pub fn new(config: ArrowStreamSerializerConfig) -> Result { + let compression = config.compression; let schema = config.schema.ok_or(ArrowEncodingError::MissingSchema)?; // If allow_nullable_fields is enabled, transform the schema once here @@ -126,6 +169,7 @@ impl ArrowStreamSerializer { Ok(Self { schema: SchemaRef::new(schema), + compression, }) } } @@ -138,7 +182,8 @@ impl tokio_util::codec::Encoder> for ArrowStreamSerializer { return Err(ArrowEncodingError::NoEvents); } - let bytes = encode_events_to_arrow_ipc_stream(&events, self.schema.clone())?; + let bytes = + encode_events_to_arrow_ipc_stream(&events, self.schema.clone(), self.compression)?; buffer.extend_from_slice(&bytes); Ok(()) @@ -212,6 +257,7 @@ pub enum ArrowEncodingError { pub fn encode_events_to_arrow_ipc_stream( events: &[Event], schema: SchemaRef, + compression: ArrowIpcCompression, ) -> Result { if events.is_empty() { return Err(ArrowEncodingError::NoEvents); @@ -225,9 +271,14 @@ pub fn encode_events_to_arrow_ipc_stream( let record_batch = build_record_batch(schema, &json_values)?; + let write_options = IpcWriteOptions::default() + .try_with_compression(compression.into()) + .context(IpcWriteSnafu)?; + let mut buffer = BytesMut::new().writer(); let mut writer = - StreamWriter::try_new(&mut buffer, record_batch.schema_ref()).context(IpcWriteSnafu)?; + StreamWriter::try_new_with_options(&mut buffer, record_batch.schema_ref(), write_options) + .context(IpcWriteSnafu)?; writer.write(&record_batch).context(IpcWriteSnafu)?; writer.finish().context(IpcWriteSnafu)?; @@ -390,7 +441,8 @@ mod tests { events: Vec, schema: SchemaRef, ) -> Result> { - let bytes = encode_events_to_arrow_ipc_stream(&events, schema.clone())?; + let bytes = + encode_events_to_arrow_ipc_stream(&events, schema.clone(), ArrowIpcCompression::None)?; let cursor = Cursor::new(bytes); let mut reader = StreamReader::try_new(cursor, None)?; Ok(reader.next().unwrap()?) @@ -408,6 +460,61 @@ mod tests { Event::Log(log) } + mod compression { + use super::*; + + fn encode_with(compression: ArrowIpcCompression) -> (Bytes, SchemaRef) { + let schema = SchemaRef::new(Schema::new(vec![ + Field::new("level", DataType::Utf8, false), + Field::new("n", DataType::Int64, false), + ])); + // Highly repetitive data so compression has an obvious effect. + let events: Vec = (0..512) + .map(|i| { + create_event(vec![ + ("level", Value::from("INFO")), + ("n", Value::from(i % 4)), + ]) + }) + .collect(); + let bytes = encode_events_to_arrow_ipc_stream(&events, schema.clone(), compression) + .expect("encode should succeed"); + (bytes, schema) + } + + fn decode(bytes: &Bytes) -> RecordBatch { + let mut reader = StreamReader::try_new(Cursor::new(bytes.clone()), None) + .expect("compressed IPC stream should be readable"); + reader.next().unwrap().expect("record batch should decode") + } + + #[test] + fn compressed_streams_round_trip_and_shrink() { + let (uncompressed, _) = encode_with(ArrowIpcCompression::None); + let (lz4, _) = encode_with(ArrowIpcCompression::Lz4Frame); + let (zstd, _) = encode_with(ArrowIpcCompression::Zstd); + + // All variants must decode back to the same record batch. + let base = decode(&uncompressed); + assert_eq!(base, decode(&lz4), "lz4 stream must round-trip"); + assert_eq!(base, decode(&zstd), "zstd stream must round-trip"); + + // Compression must actually shrink this highly-repetitive payload. + assert!( + lz4.len() < uncompressed.len(), + "lz4 ({}) should be smaller than uncompressed ({})", + lz4.len(), + uncompressed.len() + ); + assert!( + zstd.len() < uncompressed.len(), + "zstd ({}) should be smaller than uncompressed ({})", + zstd.len(), + uncompressed.len() + ); + } + } + mod comprehensive { use super::*; @@ -608,7 +715,8 @@ mod tests { fn test_encode_empty_events() { let schema = Schema::new(vec![Field::new("message", DataType::Utf8, true)]).into(); let events: Vec = vec![]; - let result = encode_events_to_arrow_ipc_stream(&events, schema); + let result = + encode_events_to_arrow_ipc_stream(&events, schema, ArrowIpcCompression::None); assert!(matches!(result.unwrap_err(), ArrowEncodingError::NoEvents)); } @@ -623,7 +731,8 @@ mod tests { )]) .into(); - let result = encode_events_to_arrow_ipc_stream(&events, schema); + let result = + encode_events_to_arrow_ipc_stream(&events, schema, ArrowIpcCompression::None); assert!(result.is_err()); } } @@ -885,7 +994,8 @@ mod tests { // Event is missing "required_field" entirely let event = create_event(vec![("optional_field", "hello")]); - let result = encode_events_to_arrow_ipc_stream(&[event], schema); + let result = + encode_events_to_arrow_ipc_stream(&[event], schema, ArrowIpcCompression::None); let err = result.unwrap_err().to_string(); assert!( err.contains("required_field"), @@ -908,7 +1018,8 @@ mod tests { // Event has "id" but "name" is null let event = create_event(vec![("id", Value::Integer(1))]); - let result = encode_events_to_arrow_ipc_stream(&[event], schema); + let result = + encode_events_to_arrow_ipc_stream(&[event], schema, ArrowIpcCompression::None); let err = result.unwrap_err().to_string(); assert!( err.contains("name"), diff --git a/lib/codecs/src/encoding/format/mod.rs b/lib/codecs/src/encoding/format/mod.rs index f760ea5c8dd83..7c7d2117047d9 100644 --- a/lib/codecs/src/encoding/format/mod.rs +++ b/lib/codecs/src/encoding/format/mod.rs @@ -32,8 +32,8 @@ pub use self::parquet::{ }; #[cfg(feature = "arrow")] pub use arrow::{ - ArrowEncodingError, ArrowStreamSerializer, ArrowStreamSerializerConfig, SchemaProvider, - find_null_non_nullable_fields, + ArrowEncodingError, ArrowIpcCompression, ArrowStreamSerializer, ArrowStreamSerializerConfig, + SchemaProvider, find_null_non_nullable_fields, }; pub use avro::{AvroSerializer, AvroSerializerConfig, AvroSerializerOptions}; pub use cef::{CefSerializer, CefSerializerConfig}; diff --git a/lib/codecs/src/encoding/mod.rs b/lib/codecs/src/encoding/mod.rs index 361a62a301533..c3d049351ca86 100644 --- a/lib/codecs/src/encoding/mod.rs +++ b/lib/codecs/src/encoding/mod.rs @@ -15,8 +15,8 @@ pub use encoder::{BatchEncoder, BatchOutput, BatchSerializer}; pub use encoder::{Encoder, EncoderKind}; #[cfg(feature = "arrow")] pub use format::{ - ArrowEncodingError, ArrowStreamSerializer, ArrowStreamSerializerConfig, SchemaProvider, - find_null_non_nullable_fields, + ArrowEncodingError, ArrowIpcCompression, ArrowStreamSerializer, ArrowStreamSerializerConfig, + SchemaProvider, find_null_non_nullable_fields, }; pub use format::{ AvroSerializer, AvroSerializerConfig, AvroSerializerOptions, CefSerializer, diff --git a/src/sinks/clickhouse/arrow/mod.rs b/src/sinks/clickhouse/arrow/mod.rs index 9ca393833c0d0..7ecb2af697980 100644 --- a/src/sinks/clickhouse/arrow/mod.rs +++ b/src/sinks/clickhouse/arrow/mod.rs @@ -3,4 +3,4 @@ pub mod parser; pub mod schema; -pub use schema::ClickHouseSchemaProvider; +pub use schema::{ClickHouseSchemaProvider, ensure_arrow_compression_supported}; diff --git a/src/sinks/clickhouse/arrow/schema.rs b/src/sinks/clickhouse/arrow/schema.rs index 9f6341471b353..36764f9ef8584 100644 --- a/src/sinks/clickhouse/arrow/schema.rs +++ b/src/sinks/clickhouse/arrow/schema.rs @@ -4,10 +4,12 @@ use std::str::FromStr; use arrow::datatypes::{Field, Schema}; use async_trait::async_trait; +use bytes::Bytes; use http::{Request, StatusCode}; use hyper::Body; use serde::Deserialize; use url::form_urlencoded; +use vector_lib::codecs::encoding::ArrowIpcCompression; use vector_lib::codecs::encoding::format::{ArrowEncodingError, SchemaProvider}; use crate::http::{Auth, HttpClient}; @@ -18,6 +20,10 @@ use super::parser::ClickHouseType; const COLUMN_KIND_REGULAR: &str = ""; const COLUMN_KIND_DEFAULT: &str = "DEFAULT"; +/// Minimum ClickHouse `(major, minor)` that reliably decodes compressed Arrow IPC buffers +/// (see ). +pub const MIN_ARROW_IPC_COMPRESSION_VERSION: (u32, u32) = (23, 11); + #[derive(Debug, Deserialize)] struct ColumnInfo { name: String, @@ -40,6 +46,37 @@ impl TryFrom for Field { } } +/// Sends a GET query to the ClickHouse HTTP interface and returns the response body bytes. +/// +/// `query_string` is the already URL-encoded query string (everything after `?`). Applies auth if +/// present and errors on any non-`200` status, using `context` to describe the failed operation. +async fn get_query_bytes( + client: &HttpClient, + endpoint: &str, + query_string: &str, + auth: Option<&Auth>, + context: &str, +) -> crate::Result { + let uri = format!("{endpoint}?{query_string}"); + let mut request = Request::get(&uri) + .body(Body::empty()) + .map_err(|e| format!("Failed to build request: {e}"))?; + + if let Some(auth) = auth { + auth.apply(&mut request); + } + + let response = client.send(request).await?; + + if response.status() != StatusCode::OK { + return Err(format!("{context}: HTTP {}", response.status()).into()); + } + + Ok(http_body::Body::collect(response.into_body()) + .await? + .to_bytes()) +} + /// Fetches the schema for a ClickHouse table and converts it to an Arrow schema. pub async fn fetch_table_schema( client: &HttpClient, @@ -57,37 +94,85 @@ pub async fn fetch_table_schema( FORMAT JSONEachRow" ); - // Build URI with query and parameters let query_string = form_urlencoded::Serializer::new(String::new()) .append_pair("query", &query) .append_pair("param_db", database) .append_pair("param_tbl", table) .finish(); - let uri = format!("{endpoint}?{query_string}"); - let mut request = Request::get(&uri) - .body(Body::empty()) - .map_err(|e| format!("Failed to build request: {e}"))?; - if let Some(auth) = auth { - auth.apply(&mut request); - } + let body_bytes = get_query_bytes( + client, + endpoint, + &query_string, + auth, + "Failed to fetch schema from ClickHouse", + ) + .await?; - let response = client.send(request).await?; + // Pass bytes directly instead of converting to a UTF-8 String first + parse_schema_from_response(&body_bytes) +} - if response.status() != StatusCode::OK { +/// Returns an error if `compression` is requested but the ClickHouse server is too old to decode +/// compressed Arrow IPC. Older servers accept the insert but fail to read it, so this is checked +/// once at startup to fail fast rather than silently dropping data at runtime. +pub async fn ensure_arrow_compression_supported( + client: &HttpClient, + endpoint: &str, + auth: Option<&Auth>, + compression: ArrowIpcCompression, +) -> crate::Result<()> { + if compression == ArrowIpcCompression::None { + return Ok(()); + } + + let (major, minor) = fetch_server_version(client, endpoint, auth).await?; + let (min_major, min_minor) = MIN_ARROW_IPC_COMPRESSION_VERSION; + if (major, minor) < (min_major, min_minor) { return Err(format!( - "Failed to fetch schema from ClickHouse: HTTP {}", - response.status() + "Arrow IPC compression requires ClickHouse >= {min_major}.{min_minor}, but the server \ + reported {major}.{minor}. Set the batch encoding 'compression' to 'none' or upgrade \ + ClickHouse." ) .into()); } - let body_bytes = http_body::Body::collect(response.into_body()) - .await? - .to_bytes(); + Ok(()) +} - // Pass bytes directly instead of converting to a UTF-8 String first - parse_schema_from_response(&body_bytes) +/// Fetches the ClickHouse server version, returning it as a `(major, minor)` tuple. +async fn fetch_server_version( + client: &HttpClient, + endpoint: &str, + auth: Option<&Auth>, +) -> crate::Result<(u32, u32)> { + let query_string = form_urlencoded::Serializer::new(String::new()) + .append_pair("query", "SELECT version() FORMAT TabSeparated") + .finish(); + + let body_bytes = get_query_bytes( + client, + endpoint, + &query_string, + auth, + "Failed to query ClickHouse version", + ) + .await?; + + parse_version(std::str::from_utf8(&body_bytes)?) +} + +/// Parses a ClickHouse version string (e.g. `24.3.1.2823`) into a `(major, minor)` tuple. +pub(crate) fn parse_version(raw: &str) -> crate::Result<(u32, u32)> { + let trimmed = raw.trim(); + let mut parts = trimmed.split('.'); + match ( + parts.next().and_then(|s| s.parse::().ok()), + parts.next().and_then(|s| s.parse::().ok()), + ) { + (Some(major), Some(minor)) => Ok((major, minor)), + _ => Err(format!("Could not parse ClickHouse version from '{trimmed}'").into()), + } } /// Parses the JSONEachRow response from ClickHouse and builds an Arrow schema. @@ -156,6 +241,23 @@ mod tests { use arrow::datatypes::{DataType, TimeUnit}; use indoc::indoc; + #[test] + fn test_parse_version() { + assert_eq!(parse_version("24.3.1.2823").unwrap(), (24, 3)); + assert_eq!(parse_version("23.8.16.16\n").unwrap(), (23, 8)); + assert_eq!(parse_version("25.12.11.4").unwrap(), (25, 12)); + assert!(parse_version("not-a-version").is_err()); + assert!(parse_version("").is_err()); + } + + #[test] + fn test_min_version_boundary() { + // 23.10 must be rejected, 23.11 (the confirmed floor) must be accepted. + assert!((23, 10) < MIN_ARROW_IPC_COMPRESSION_VERSION); + assert!((23, 11) >= MIN_ARROW_IPC_COMPRESSION_VERSION); + assert!((24, 1) >= MIN_ARROW_IPC_COMPRESSION_VERSION); + } + #[test] fn test_parse_schema() { let response = indoc! {r#" diff --git a/src/sinks/clickhouse/config.rs b/src/sinks/clickhouse/config.rs index 53050dca831ea..8ff96e6a56463 100644 --- a/src/sinks/clickhouse/config.rs +++ b/src/sinks/clickhouse/config.rs @@ -4,8 +4,8 @@ use std::fmt; use http::{Request, StatusCode, Uri}; use hyper::Body; -use vector_lib::codecs::encoding::ArrowStreamSerializerConfig; use vector_lib::codecs::encoding::format::SchemaProvider; +use vector_lib::codecs::encoding::{ArrowIpcCompression, ArrowStreamSerializerConfig}; use super::{ request_builder::ClickhouseRequestBuilder, @@ -58,6 +58,11 @@ pub enum ClickhouseBatchEncoding { /// This is the streaming variant of the Arrow IPC format, which writes /// a continuous stream of record batches. /// + /// Note: enabling the `compression` option (any value other than `none`) requires ClickHouse + /// 23.11 or newer. Older servers accept the insert but cannot read compressed Arrow IPC, so the + /// sink rejects that combination at startup. When it is enabled, the sink also disables the + /// top-level HTTP `compression` to avoid double-compressing the payload. + /// /// [apache_arrow]: https://arrow.apache.org/ ArrowStream(ArrowStreamSerializerConfig), } @@ -218,13 +223,15 @@ impl SinkConfig for ClickhouseConfig { let client = HttpClient::new(tls_settings, &cx.proxy)?; + let compression = self.effective_http_compression(); + let clickhouse_service_request_builder = ClickhouseServiceRequestBuilder { auth: auth.clone(), endpoint: endpoint.clone(), skip_unknown_fields: self.skip_unknown_fields, date_time_best_effort: self.date_time_best_effort, insert_random_shard: self.insert_random_shard, - compression: self.compression, + compression, query_settings: self.query_settings, }; @@ -251,7 +258,7 @@ impl SinkConfig for ClickhouseConfig { .await?; let request_builder = ClickhouseRequestBuilder { - compression: self.compression, + compression, encoder: (self.encoding.clone(), encoder_kind), }; @@ -279,6 +286,28 @@ impl SinkConfig for ClickhouseConfig { } impl ClickhouseConfig { + /// Resolves the HTTP `compression` to apply to requests. + /// + /// When the `arrow_stream` batch encoding already compresses the IPC buffers, HTTP compression + /// is forced off to avoid double-compressing the payload (the sink defaults to `gzip`). + fn effective_http_compression(&self) -> Compression { + let ipc_compression = match &self.batch_encoding { + Some(ClickhouseBatchEncoding::ArrowStream(config)) => config.compression, + None => ArrowIpcCompression::None, + }; + + if ipc_compression != ArrowIpcCompression::None && self.compression != Compression::None { + warn!( + message = "Arrow IPC buffer compression is enabled; disabling redundant HTTP \ + compression to avoid double-compressing the payload.", + disabled_http_compression = ?self.compression, + ); + return Compression::None; + } + + self.compression + } + /// Resolves the encoding strategy (format + encoder) based on configuration. /// /// This method determines the appropriate ClickHouse format and Vector encoder @@ -311,6 +340,16 @@ impl ClickhouseConfig { let ClickhouseBatchEncoding::ArrowStream(arrow_config) = batch_encoding; let mut arrow_config = arrow_config.clone(); + // Older ClickHouse servers accept a compressed Arrow IPC insert but fail to read it, + // silently dropping the batch. Fail fast at startup instead. + super::arrow::ensure_arrow_compression_supported( + client, + &endpoint.to_string(), + auth, + arrow_config.compression, + ) + .await?; + self.resolve_arrow_schema( client, endpoint.to_string(), diff --git a/src/sinks/clickhouse/integration_tests.rs b/src/sinks/clickhouse/integration_tests.rs index a1405870ecd0e..8f79826930a9b 100644 --- a/src/sinks/clickhouse/integration_tests.rs +++ b/src/sinks/clickhouse/integration_tests.rs @@ -18,7 +18,7 @@ use serde::Deserialize; use serde_json::Value; use tokio::time::{Duration, timeout}; use vector_lib::{ - codecs::encoding::ArrowStreamSerializerConfig, + codecs::encoding::{ArrowIpcCompression, ArrowStreamSerializerConfig}, event::{BatchNotifier, BatchStatus, BatchStatusReceiver, Event, LogEvent}, lookup::PathPrefix, }; @@ -28,7 +28,10 @@ use crate::{ codecs::{TimestampFormat, Transformer}, config::{SinkConfig, SinkContext, log_schema}, sinks::{ - clickhouse::config::{ClickhouseBatchEncoding, ClickhouseConfig}, + clickhouse::{ + arrow::schema::parse_version, + config::{ClickhouseBatchEncoding, ClickhouseConfig}, + }, util::{BatchConfig, Compression, TowerRequestConfig}, }, test_util::{ @@ -42,6 +45,16 @@ fn clickhouse_address() -> String { std::env::var("CLICKHOUSE_ADDRESS").unwrap_or_else(|_| "http://localhost:8123".into()) } +/// Asserts a value from a ClickHouse JSON response holds an integer. ClickHouse renders Int64 as a +/// quoted string on older versions and as an unquoted JSON number on newer ones, so accept either. +fn assert_is_integer(value: Option<&Value>, field: &str) { + let value = value.unwrap_or_else(|| panic!("{field} column should be present")); + assert!( + value.as_i64().is_some() || value.as_str().and_then(|s| s.parse::().ok()).is_some(), + "{field} should be an integer, got: {value:?}" + ); +} + #[tokio::test] async fn insert_events() { trace_init(); @@ -452,6 +465,19 @@ impl ClickhouseClient { } } + /// Returns the server's `(major, minor)` version, e.g. `24.3.1.2823` -> `(24, 3)`. + async fn server_version(&self) -> (u32, u32) { + let response = self + .client + .post(&self.host) + .body("SELECT version()") + .send() + .await + .unwrap(); + let text = response.text().await.unwrap(); + parse_version(&text).unwrap() + } + async fn select_all(&self, table: &str) -> QueryResponse { let response = self .client @@ -542,15 +568,128 @@ async fn insert_events_arrow_format() { for row in output.data.iter() { assert!(row.get("host").and_then(|v| v.as_str()).is_some()); assert!(row.get("message").and_then(|v| v.as_str()).is_some()); - assert!( - row.get("count") - .and_then(|v| v.as_str()) - .and_then(|s| s.parse::().ok()) - .is_some() - ); + assert_is_integer(row.get("count"), "count"); } } +// Verifies ClickHouse can decode an Arrow IPC stream whose record batch buffers were compressed +// with the given codec. The HTTP `Content-Encoding` is left off (`Compression::None`) so this +// exercises the IPC block compression in isolation. +async fn assert_arrow_ipc_compression_round_trips(compression: ArrowIpcCompression) { + trace_init(); + + let table = random_table_name(); + let host = clickhouse_address(); + + let client = ClickhouseClient::new(host.clone()); + + // Compressed Arrow IPC (any codec) needs ClickHouse >= 23.11; older servers reject typical + // small batches with CANNOT_READ_ALL_DATA. The sink guards against this at build time. + if client.server_version().await < (23, 11) { + return; + } + + let mut batch = BatchConfig::default(); + batch.max_events = Some(5); + + let config = ClickhouseConfig { + endpoint: host.parse().unwrap(), + table: table.clone().try_into().unwrap(), + compression: Compression::None, + format: crate::sinks::clickhouse::config::Format::ArrowStream, + batch_encoding: Some(ClickhouseBatchEncoding::ArrowStream( + ArrowStreamSerializerConfig { + compression, + ..Default::default() + }, + )), + batch, + request: TowerRequestConfig { + retry_attempts: 1, + ..Default::default() + }, + ..Default::default() + }; + + client + .create_table( + &table, + "host String, timestamp DateTime64(3), message String, count Int64", + ) + .await; + + let (sink, _hc) = config.build(SinkContext::default()).await.unwrap(); + + let mut events: Vec = Vec::new(); + for i in 0..5 { + let mut event = LogEvent::from(format!("log message {}", i)); + event.insert("host", format!("host{}.example.com", i)); + event.insert("count", i as i64); + events.push(event.into()); + } + + run_and_assert_sink_compliance(sink, stream::iter(events), &SINK_TAGS).await; + + let output = client.select_all(&table).await; + assert_eq!( + 5, output.rows, + "ClickHouse should decode all rows from the {compression:?}-compressed Arrow IPC stream" + ); + + for row in output.data.iter() { + assert!(row.get("host").and_then(|v| v.as_str()).is_some()); + assert!(row.get("message").and_then(|v| v.as_str()).is_some()); + assert_is_integer(row.get("count"), "count"); + } +} + +#[tokio::test] +async fn insert_events_arrow_format_zstd_compression() { + assert_arrow_ipc_compression_round_trips(ArrowIpcCompression::Zstd).await; +} + +#[tokio::test] +async fn insert_events_arrow_format_lz4_compression() { + assert_arrow_ipc_compression_round_trips(ArrowIpcCompression::Lz4Frame).await; +} + +// On ClickHouse older than 23.11, the sink must reject Arrow IPC compression at build time rather +// than let it silently drop data at runtime. Trivially skips on newer servers. +#[tokio::test] +async fn arrow_compression_rejected_on_old_clickhouse() { + trace_init(); + + let host = clickhouse_address(); + let client = ClickhouseClient::new(host.clone()); + if client.server_version().await >= (23, 11) { + return; + } + + let mut batch = BatchConfig::default(); + batch.max_events = Some(5); + + let config = ClickhouseConfig { + endpoint: host.parse().unwrap(), + table: random_table_name().try_into().unwrap(), + compression: Compression::None, + format: crate::sinks::clickhouse::config::Format::ArrowStream, + batch_encoding: Some(ClickhouseBatchEncoding::ArrowStream( + ArrowStreamSerializerConfig { + compression: ArrowIpcCompression::Zstd, + ..Default::default() + }, + )), + batch, + ..Default::default() + }; + + let result = config.build(SinkContext::default()).await; + assert!( + result.is_err(), + "sink should reject Arrow IPC compression on ClickHouse older than 23.11" + ); +} + #[tokio::test] async fn insert_events_arrow_with_schema_fetching() { trace_init(); @@ -618,12 +757,7 @@ async fn insert_events_arrow_with_schema_fetching() { assert!(row.get("timestamp").is_some()); // Check custom fields have correct types - assert!( - row.get("id") - .and_then(|v| v.as_str()) - .and_then(|s| s.parse::().ok()) - .is_some() - ); + assert_is_integer(row.get("id"), "id"); assert!(row.get("name").and_then(|v| v.as_str()).is_some()); assert!(row.get("score").and_then(|v| v.as_f64()).is_some()); assert!(row.get("active").and_then(|v| v.as_bool()).is_some()); diff --git a/tests/integration/clickhouse/config/compose.yaml b/tests/integration/clickhouse/config/compose.yaml index e861c92c736f0..9ad40768f992c 100644 --- a/tests/integration/clickhouse/config/compose.yaml +++ b/tests/integration/clickhouse/config/compose.yaml @@ -3,6 +3,9 @@ version: "3" services: clickhouse: image: docker.io/clickhouse/clickhouse-server:${CONFIG_VERSION} + environment: + # ClickHouse >= 25 requires a password for the default user unless user setup is skipped. + CLICKHOUSE_SKIP_USER_SETUP: "1" networks: default: diff --git a/tests/integration/clickhouse/config/test.yaml b/tests/integration/clickhouse/config/test.yaml index 3f88eedf5a654..bb0edccd5900e 100644 --- a/tests/integration/clickhouse/config/test.yaml +++ b/tests/integration/clickhouse/config/test.yaml @@ -7,7 +7,8 @@ env: CLICKHOUSE_ADDRESS: http://clickhouse:8123 matrix: - version: ["23"] + # 23 covers the uncompressed paths; 25 exercises compressed Arrow IPC. + version: ["23", "25"] # changes to these files/paths will invoke the integration test in CI # expressions are evaluated using https://github.com/micromatch/picomatch diff --git a/website/cue/reference/components/sinks/generated/clickhouse.cue b/website/cue/reference/components/sinks/generated/clickhouse.cue index e07014f38a308..8af368fef3a0f 100644 --- a/website/cue/reference/components/sinks/generated/clickhouse.cue +++ b/website/cue/reference/components/sinks/generated/clickhouse.cue @@ -273,6 +273,11 @@ generated: components: sinks: clickhouse: configuration: { This is the streaming variant of the Arrow IPC format, which writes a continuous stream of record batches. + Note: enabling the `compression` option (any value other than `none`) requires ClickHouse + 23.11 or newer. Older servers accept the insert but cannot read compressed Arrow IPC, so the + sink rejects that combination at startup. When it is enabled, the sink also disables the + top-level HTTP `compression` to avoid double-compressing the payload. + [apache_arrow]: https://arrow.apache.org/ """ required: true @@ -282,9 +287,30 @@ generated: components: sinks: clickhouse: configuration: { This is the streaming variant of the Arrow IPC format, which writes a continuous stream of record batches. + Note: enabling the `compression` option (any value other than `none`) requires ClickHouse + 23.11 or newer. Older servers accept the insert but cannot read compressed Arrow IPC, so the + sink rejects that combination at startup. When it is enabled, the sink also disables the + top-level HTTP `compression` to avoid double-compressing the payload. + [apache_arrow]: https://arrow.apache.org/ """ } + compression: { + description: """ + Block-level compression applied to the Arrow IPC record batch buffers. + + Compresses each buffer inside the IPC stream. + """ + required: false + type: string: { + default: "none" + enum: { + lz4_frame: "LZ4 frame compression." + none: "No compression." + zstd: "Zstandard compression." + } + } + } } } compression: {