From 61195cd9652bb269144d6f11a392d05095e08a61 Mon Sep 17 00:00:00 2001 From: benjamin-awd Date: Tue, 7 Jul 2026 17:55:26 +0800 Subject: [PATCH 1/2] enhancement(clickhouse sink): add Arrow IPC compression to arrow_stream encoding Adds a `compression` option to the clickhouse sink's `arrow_stream` batch encoding for block-level compression of the Arrow IPC record batch buffers (`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 requires ClickHouse 23.11 or newer; older servers accept the insert but cannot read compressed Arrow IPC, so the sink validates the server version at startup and rejects the configuration on older servers. The ArrowIpcCompression enum mirrors databricks_zerobus's Compression (variant names, From conversion, skip_serializing_if) so the two can be consolidated into a shared codecs type in a follow-up. --- Cargo.toml | 2 +- ...house_arrow_ipc_compression.enhancement.md | 3 + lib/codecs/Cargo.toml | 2 +- lib/codecs/src/encoding/format/arrow.rs | 127 +++++++++++++- lib/codecs/src/encoding/format/mod.rs | 4 +- lib/codecs/src/encoding/mod.rs | 4 +- src/sinks/clickhouse/arrow/mod.rs | 2 +- src/sinks/clickhouse/arrow/schema.rs | 136 +++++++++++++-- src/sinks/clickhouse/config.rs | 29 +++- src/sinks/clickhouse/integration_tests.rs | 162 ++++++++++++++++-- .../clickhouse/config/compose.yaml | 3 + tests/integration/clickhouse/config/test.yaml | 3 +- .../components/sinks/generated/clickhouse.cue | 34 ++++ 13 files changed, 463 insertions(+), 48 deletions(-) create mode 100644 changelog.d/25756_clickhouse_arrow_ipc_compression.enhancement.md 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..3fc4554271a6c --- /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. + +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..f9c886014ed0c 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,10 @@ 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. + /// /// [apache_arrow]: https://arrow.apache.org/ ArrowStream(ArrowStreamSerializerConfig), } @@ -311,6 +315,29 @@ impl ClickhouseConfig { let ClickhouseBatchEncoding::ArrowStream(arrow_config) = batch_encoding; let mut arrow_config = arrow_config.clone(); + if arrow_config.compression != ArrowIpcCompression::None + && self.compression != Compression::None + { + warn!( + message = "Both Arrow IPC buffer compression \ + (`batch_encoding.arrow_stream.compression`) and HTTP compression \ + (`compression`) are enabled; performance may be improved by enabling \ + only one.", + arrow_ipc_compression = ?arrow_config.compression, + http_compression = ?self.compression, + ); + } + + // 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..11de7f7d248ce 100644 --- a/website/cue/reference/components/sinks/generated/clickhouse.cue +++ b/website/cue/reference/components/sinks/generated/clickhouse.cue @@ -273,6 +273,10 @@ 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. + [apache_arrow]: https://arrow.apache.org/ """ required: true @@ -282,9 +286,39 @@ 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. + [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. At a comparable ratio this costs far less CPU + than an HTTP `gzip` `Content-Encoding`; configure one or the other, not both. + """ + required: false + type: string: { + default: "none" + enum: { + lz4_frame: """ + LZ4 frame compression. + + Fastest option with near-negligible CPU cost, at the expense of a larger payload than `zstd`. + """ + none: "No compression." + zstd: """ + Zstandard compression. + + Higher compression ratio than `lz4_frame`, typically at lower CPU cost than a general-purpose + HTTP `gzip` `Content-Encoding`. A reasonable default when egress bandwidth is metered. + """ + } + } + } } } compression: { From 18fe83d9736bf2dde97c2f70ef48da2b3e82a27a Mon Sep 17 00:00:00 2001 From: benjamin-awd Date: Tue, 7 Jul 2026 18:31:24 +0800 Subject: [PATCH 2/2] enhancement(clickhouse sink): disable HTTP compression when Arrow IPC compression is enabled The sink's top-level `compression` defaults to `gzip`, so enabling Arrow IPC buffer compression without also setting `compression: none` would gzip a payload that already contains compressed Arrow buffers, wasting CPU for negligible gain. When Arrow IPC compression is enabled, force the HTTP `compression` to `none` and log that it was disabled, so the IPC compression path does not double-compress by default. --- ...house_arrow_ipc_compression.enhancement.md | 2 +- src/sinks/clickhouse/config.rs | 44 ++++++++++++------- .../components/sinks/generated/clickhouse.cue | 24 ++++------ 3 files changed, 37 insertions(+), 33 deletions(-) diff --git a/changelog.d/25756_clickhouse_arrow_ipc_compression.enhancement.md b/changelog.d/25756_clickhouse_arrow_ipc_compression.enhancement.md index 3fc4554271a6c..bcb5256d93a18 100644 --- a/changelog.d/25756_clickhouse_arrow_ipc_compression.enhancement.md +++ b/changelog.d/25756_clickhouse_arrow_ipc_compression.enhancement.md @@ -1,3 +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. +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/src/sinks/clickhouse/config.rs b/src/sinks/clickhouse/config.rs index f9c886014ed0c..8ff96e6a56463 100644 --- a/src/sinks/clickhouse/config.rs +++ b/src/sinks/clickhouse/config.rs @@ -60,7 +60,8 @@ pub enum ClickhouseBatchEncoding { /// /// 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. + /// 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), @@ -222,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, }; @@ -255,7 +258,7 @@ impl SinkConfig for ClickhouseConfig { .await?; let request_builder = ClickhouseRequestBuilder { - compression: self.compression, + compression, encoder: (self.encoding.clone(), encoder_kind), }; @@ -283,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 @@ -315,19 +340,6 @@ impl ClickhouseConfig { let ClickhouseBatchEncoding::ArrowStream(arrow_config) = batch_encoding; let mut arrow_config = arrow_config.clone(); - if arrow_config.compression != ArrowIpcCompression::None - && self.compression != Compression::None - { - warn!( - message = "Both Arrow IPC buffer compression \ - (`batch_encoding.arrow_stream.compression`) and HTTP compression \ - (`compression`) are enabled; performance may be improved by enabling \ - only one.", - arrow_ipc_compression = ?arrow_config.compression, - http_compression = ?self.compression, - ); - } - // 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( diff --git a/website/cue/reference/components/sinks/generated/clickhouse.cue b/website/cue/reference/components/sinks/generated/clickhouse.cue index 11de7f7d248ce..8af368fef3a0f 100644 --- a/website/cue/reference/components/sinks/generated/clickhouse.cue +++ b/website/cue/reference/components/sinks/generated/clickhouse.cue @@ -275,7 +275,8 @@ generated: components: sinks: clickhouse: configuration: { 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. + 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/ """ @@ -288,7 +289,8 @@ generated: components: sinks: clickhouse: configuration: { 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. + 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/ """ @@ -297,25 +299,15 @@ generated: components: sinks: clickhouse: configuration: { description: """ Block-level compression applied to the Arrow IPC record batch buffers. - Compresses each buffer inside the IPC stream. At a comparable ratio this costs far less CPU - than an HTTP `gzip` `Content-Encoding`; configure one or the other, not both. + Compresses each buffer inside the IPC stream. """ required: false type: string: { default: "none" enum: { - lz4_frame: """ - LZ4 frame compression. - - Fastest option with near-negligible CPU cost, at the expense of a larger payload than `zstd`. - """ - none: "No compression." - zstd: """ - Zstandard compression. - - Higher compression ratio than `lz4_frame`, typically at lower CPU cost than a general-purpose - HTTP `gzip` `Content-Encoding`. A reasonable default when egress bandwidth is metered. - """ + lz4_frame: "LZ4 frame compression." + none: "No compression." + zstd: "Zstandard compression." } } }