enhancement(clickhouse sink): Add ArrowStream format

Cargo.toml

@@ -341,6 +341,7 @@ arc-swap = { version = "1.7", default-features = false, optional = true }
 async-compression = { version = "0.4.27", default-features = false, features = ["tokio", "gzip", "zstd"], optional = true }
 apache-avro = { version = "0.16.0", default-features = false, optional = true }
 arrow = { version = "56.2.0", default-features = false, features = ["ipc"], optional = true }
+arrow-schema = { version = "56.2.0", default-features = false, optional = true }
 axum = { version = "0.6.20", default-features = false }
 base64 = { workspace = true, optional = true }
 bloomy = { version = "1.2.0", default-features = false, optional = true }
@@ -406,7 +407,7 @@ redis = { version = "0.32.4", default-features = false, features = ["connection-
 regex.workspace = true
 roaring = { version = "0.11.2", default-features = false, features = ["std"], optional = true }
 rumqttc = { version = "0.24.0", default-features = false, features = ["use-rustls"], optional = true }
-rust_decimal = { workspace = true, optional = true }
+rust_decimal = { version = "1.37.0", default-features = false, features = ["std"], optional = true }
 seahash = { version = "4.1.0", default-features = false }
 smallvec = { version = "1", default-features = false, features = ["union", "serde"] }
 snap = { version = "1.1.1", default-features = false }
@@ -583,7 +584,7 @@ enrichment-tables-mmdb = ["dep:maxminddb"]
 enrichment-tables-memory = ["dep:evmap", "dep:evmap-derive", "dep:thread_local"]
 
 # Codecs
-codecs-arrow = ["vector-lib/arrow"]
+codecs-arrow = ["dep:arrow", "dep:arrow-schema", "vector-lib/arrow"]
 codecs-opentelemetry = ["vector-lib/opentelemetry"]
 codecs-syslog = ["vector-lib/syslog"]
 
@@ -851,7 +852,7 @@ sinks-azure_blob = ["dep:azure_core", "dep:azure_identity", "dep:azure_storage",
 sinks-azure_monitor_logs = []
 sinks-blackhole = []
 sinks-chronicle = []
-sinks-clickhouse = []
+sinks-clickhouse = ["dep:arrow", "dep:arrow-schema", "dep:rust_decimal", "codecs-arrow"]
 sinks-console = []
 sinks-databend = ["dep:databend-client"]
 sinks-datadog_events = []

changelog.d/24074_clickhouse_arrow_format.enhancement.md

@@ -0,0 +1,3 @@
+The `clickhouse` sink now supports the `arrow_stream` format option, enabling high-performance binary data transfer using Apache Arrow IPC. This provides significantly better performance and smaller payload sizes compared to JSON-based formats.
+
+authors: benjamin-awd

lib/codecs/Cargo.toml

@@ -15,6 +15,7 @@ path = "tests/bin/generate-avro-fixtures.rs"
 [dependencies]
 apache-avro = { version = "0.20.0", default-features = false }
 arrow = { version = "56.2.0", default-features = false, features = ["ipc"] }
+async-trait.workspace = true
 bytes.workspace = true
 chrono.workspace = true
 rust_decimal = { version = "1.37", default-features = false, features = ["std"] }

lib/codecs/src/encoding/format/arrow.rs

@@ -16,6 +16,7 @@ use arrow::{
     ipc::writer::StreamWriter,
     record_batch::RecordBatch,
 };
+use async_trait::async_trait;
 use bytes::{BufMut, Bytes, BytesMut};
 use chrono::{DateTime, Utc};
 use rust_decimal::Decimal;
@@ -25,6 +26,15 @@ use vector_config::configurable_component;
 
 use vector_core::event::{Event, Value};
 
+/// Provides Arrow schema for encoding.
+///
+/// Sinks can implement this trait to provide custom schema fetching logic.
+#[async_trait]
+pub trait SchemaProvider: Send + Sync + std::fmt::Debug {
+    /// Get the Arrow schema for encoding events.
+    async fn get_schema(&self) -> Result<Arc<Schema>, ArrowEncodingError>;
+}
+
 /// Configuration for Arrow IPC stream serialization
 #[configurable_component]
 #[derive(Clone, Default)]
@@ -45,6 +55,10 @@ pub struct ArrowStreamSerializerConfig {
     #[serde(default)]
     #[configurable(metadata(docs::examples = true))]
     pub allow_nullable_fields: bool,
+
+    /// Schema provider for lazy schema loading.
+    #[serde(skip)]
+    schema_provider: Option<Arc<dyn SchemaProvider>>,
 }
 
 impl std::fmt::Debug for ArrowStreamSerializerConfig {
@@ -58,6 +72,10 @@ impl std::fmt::Debug for ArrowStreamSerializerConfig {
                     .map(|s| format!("{} fields", s.fields().len())),
             )
             .field("allow_nullable_fields", &self.allow_nullable_fields)
+            .field(
+                "schema_provider",
+                &self.schema_provider.as_ref().map(|_| "<provider>"),
+            )
             .finish()
     }
 }
@@ -68,6 +86,38 @@ impl ArrowStreamSerializerConfig {
         Self {
             schema: Some(schema),
             allow_nullable_fields: false,
+            schema_provider: None,
+        }
+    }
+
+    /// Create a new ArrowStreamSerializerConfig with a schema provider
+    pub fn with_provider(provider: Arc<dyn SchemaProvider>) -> Self {
+        Self {
+            schema: None,
+            schema_provider: Some(provider),
+            allow_nullable_fields: false,
+        }
+    }
+
+    /// Get the schema provider if one was configured
+    pub fn provider(&self) -> Option<&Arc<dyn SchemaProvider>> {
+        self.schema_provider.as_ref()
+    }
+
+    /// Resolve the schema from the provider if present.
+    pub async fn resolve(&mut self) -> Result<(), ArrowEncodingError> {
+        // If schema already exists, nothing to do
+        if self.schema.is_some() {
+            return Ok(());
+        }
+
+        // Fetch from provider if available
+        if let Some(provider) = &self.schema_provider {
+            let schema = provider.get_schema().await?;
+            self.schema = Some(schema);
+            Ok(())
+        } else {
+            Err(ArrowEncodingError::NoSchemaProvided)
         }
     }
 
@@ -154,6 +204,13 @@ pub enum ArrowEncodingError {
     #[snafu(display("Schema must be provided before encoding"))]
     NoSchemaProvided,
 
+    /// Failed to fetch schema from provider
+    #[snafu(display("Failed to fetch schema from provider: {}", message))]
+    SchemaFetchError {
+        /// Error message from the provider
+        message: String,
+    },
+
     /// Unsupported Arrow data type for field
     #[snafu(display(
         "Unsupported Arrow data type for field '{}': {:?}",

lib/codecs/src/encoding/format/mod.rs

@@ -23,7 +23,9 @@ mod text;
 use std::fmt::Debug;
 
 #[cfg(feature = "arrow")]
-pub use arrow::{ArrowEncodingError, ArrowStreamSerializer, ArrowStreamSerializerConfig};
+pub use arrow::{
+    ArrowEncodingError, ArrowStreamSerializer, ArrowStreamSerializerConfig, SchemaProvider,
+};
 pub use avro::{AvroSerializer, AvroSerializerConfig, AvroSerializerOptions};
 pub use cef::{CefSerializer, CefSerializerConfig};
 use dyn_clone::DynClone;

lib/codecs/src/encoding/mod.rs

@@ -7,7 +7,9 @@ pub mod framing;
 pub mod serializer;
 pub use chunking::{Chunker, Chunking, GelfChunker};
 #[cfg(feature = "arrow")]
-pub use format::{ArrowEncodingError, ArrowStreamSerializer, ArrowStreamSerializerConfig};
+pub use format::{
+    ArrowEncodingError, ArrowStreamSerializer, ArrowStreamSerializerConfig, SchemaProvider,
+};
 pub use format::{
     AvroSerializer, AvroSerializerConfig, AvroSerializerOptions, CefSerializer,
     CefSerializerConfig, CsvSerializer, CsvSerializerConfig, GelfSerializer, GelfSerializerConfig,