benjamin-awd · benjamin-awd · Dec 22, 2025 · Dec 29, 2025 · Dec 29, 2025 · Jan 5, 2026
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -270,7 +270,7 @@ aws-types = { version = "1.3.8", default-features = false, optional = true }
 # The sts crate is needed despite not being referred to anywhere in the code because we need to set the
 # `behavior-version-latest` feature. Without this we get a runtime panic when `auth.assume_role` authentication
 # is configured.
-aws-sdk-sts = { version = "1.73.0", default-features = false, features = ["behavior-version-latest", "rt-tokio"], optional = true }
+aws-sdk-sts = { version = "1.95.0", default-features = false, features = ["behavior-version-latest", "rt-tokio"], optional = true }
 
 # The `aws-sdk-sts` crate is needed despite not being referred to anywhere in the code because we need to set the
 # `behavior-version-latest` feature. Without this we get a runtime panic when `auth.assume_role` authentication is configured.
@@ -297,6 +297,9 @@ azure_core_for_storage = { package = "azure_core", version = "0.21.0", default-f
 # OpenDAL
 opendal = { version = "0.54", default-features = false, features = ["services-webhdfs"], optional = true }
 
+# Delta Lake
+deltalake = { version = "0.30.0", default-features = false, features = ["datafusion", "gcs", "s3", "rustls"], optional = true }
+
 # Tower
 tower = { version = "0.5.2", default-features = false, features = ["buffer", "limit", "retry", "timeout", "util", "balance", "discover"] }
 tower-http = { version = "0.4.4", default-features = false, features = ["compression-full", "decompression-gzip", "trace"] }
@@ -338,10 +341,13 @@ greptimedb-ingester = { git = "https://github.com/GreptimeTeam/greptimedb-ingest
 
 # External libs
 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 }
+# Pinned to 0.4.19 to use xz2 instead of liblzma (avoids conflict with datafusion's xz2)
+# See: https://github.com/apache/datafusion/issues/15342
+async-compression = { version = "=0.4.19", 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 }
+# Updated to 57.1.0 for compatibility with deltalake 0.30.0
+arrow = { version = "57.1.0", default-features = false, features = ["ipc"], optional = true }
+arrow-schema = { version = "57.1.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 }
@@ -493,7 +499,6 @@ nix = { git = "https://github.com/vectordotdev/nix.git", branch = "memfd/gnu/mus
 # The `heim` crates depend on `ntapi` 0.3.7 on Windows, but that version has an
 # unaligned access bug fixed in the following revision.
 ntapi = { git = "https://github.com/MSxDOS/ntapi.git", rev = "24fc1e47677fc9f6e38e5f154e6011dc9b270da6" }
-
 [features]
 # Default features for *-unknown-linux-gnu and *-apple-darwin
 default = ["api", "api-client", "enrichment-tables", "sinks", "sources", "sources-dnstap", "transforms", "unix", "rdkafka?/gssapi-vendored", "secrets"]
@@ -601,6 +606,7 @@ sources-logs = [
   "sources-aws_s3",
   "sources-aws_sqs",
   "sources-datadog_agent",
+  "sources-delta_lake_cdf",
   "sources-demo_logs",
   "sources-docker_logs",
   "sources-exec",
@@ -653,6 +659,7 @@ sources-aws_kinesis_firehose = ["dep:base64"]
 sources-aws_s3 = ["aws-core", "dep:aws-sdk-sqs", "dep:aws-sdk-s3", "dep:async-compression", "sources-aws_sqs", "tokio-util/io"]
 sources-aws_sqs = ["aws-core", "dep:aws-sdk-sqs"]
 sources-datadog_agent = ["sources-utils-http-encoding", "protobuf-build", "dep:prost"]
+sources-delta_lake_cdf = ["dep:deltalake", "codecs-arrow"]
 sources-demo_logs = ["dep:fakedata"]
 sources-dnstap = ["sources-utils-net-tcp", "dep:base64", "dep:hickory-proto", "dep:dnsmsg-parser", "dep:dnstap-parser", "protobuf-build", "dep:prost"]
 sources-docker_logs = ["docker"]
@@ -789,6 +796,7 @@ sinks-logs = [
   "sinks-clickhouse",
   "sinks-console",
   "sinks-databend",
+  "sinks-delta_lake",
   "sinks-datadog_events",
   "sinks-datadog_logs",
   "sinks-datadog_traces",
@@ -855,6 +863,7 @@ sinks-chronicle = []
 sinks-clickhouse = ["dep:rust_decimal", "codecs-arrow"]
 sinks-console = []
 sinks-databend = ["dep:databend-client"]
+sinks-delta_lake = ["dep:deltalake", "codecs-arrow", "dep:opendal", "dep:arc-swap"]
 sinks-datadog_events = []
 sinks-datadog_logs = []
 sinks-datadog_metrics = ["protobuf-build", "dep:prost", "dep:prost-reflect"]
@@ -906,6 +915,7 @@ all-integration-tests = [
   "chronicle-integration-tests",
   "clickhouse-integration-tests",
   "databend-integration-tests",
+  "delta-lake-integration-tests",
   "datadog-agent-integration-tests",
   "datadog-logs-integration-tests",
   "datadog-metrics-integration-tests",
@@ -972,6 +982,8 @@ azure-blob-integration-tests = ["sinks-azure_blob"]
 chronicle-integration-tests = ["sinks-gcp"]
 clickhouse-integration-tests = ["sinks-clickhouse"]
 databend-integration-tests = ["sinks-databend"]
+delta-lake-integration-tests = ["sinks-delta_lake", "dep:deltalake"]
+delta-lake-cdf-integration-tests = ["sources-delta_lake_cdf", "dep:deltalake"]
 datadog-agent-integration-tests = ["sources-datadog_agent"]
 datadog-logs-integration-tests = ["sinks-datadog_logs"]
 datadog-metrics-integration-tests = ["sinks-datadog_metrics", "dep:prost"]

diff --git 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 = "56.2.0", default-features = false, features = ["ipc"] }
+arrow = { version = "57.1.0", default-features = false, features = ["ipc"] }
 async-trait.workspace = true
 bytes.workspace = true
 chrono.workspace = true

diff --git a/lib/codecs/src/encoding/format/arrow/builder.rs b/lib/codecs/src/encoding/format/arrow/builder.rs
@@ -391,7 +391,7 @@ fn build_array_for_field(events: &[Event], field: &Field) -> Result<ArrayRef, Ar
 }
 
 /// Builds an Arrow RecordBatch from events
-pub(crate) fn build_record_batch(
+pub fn build_record_batch(
     schema: SchemaRef,
     events: &[Event],
 ) -> Result<RecordBatch, ArrowEncodingError> {

diff --git a/lib/codecs/src/encoding/format/arrow/mod.rs b/lib/codecs/src/encoding/format/arrow/mod.rs
@@ -20,7 +20,7 @@ use snafu::Snafu;
 use std::sync::Arc;
 use vector_config::configurable_component;
 
-use builder::build_record_batch;
+pub use builder::build_record_batch;
 
 /// Provides Arrow schema for encoding.
 ///
@@ -123,6 +123,11 @@ impl ArrowStreamSerializer {
             schema: SchemaRef::new(schema),
         })
     }
+
+    /// Get a reference to the Arrow schema used by this serializer
+    pub fn schema(&self) -> &SchemaRef {
+        &self.schema
+    }
 }
 
 impl tokio_util::codec::Encoder<Vec<vector_core::event::Event>> for ArrowStreamSerializer {
@@ -234,8 +239,11 @@ pub fn encode_events_to_arrow_ipc_stream(
     Ok(buffer.into_inner().freeze())
 }
 
-/// Recursively makes a Field and all its nested fields nullable
-fn make_field_nullable(field: &arrow::datatypes::Field) -> arrow::datatypes::Field {
+/// Recursively makes a Field and all its nested fields nullable.
+///
+/// This is useful for schema evolution scenarios where new fields need to be nullable,
+/// or when events may have missing fields that should be represented as null.
+pub fn make_field_nullable(field: &arrow::datatypes::Field) -> arrow::datatypes::Field {
     let new_data_type = match field.data_type() {
         DataType::List(inner_field) => DataType::List(make_field_nullable(inner_field).into()),
         DataType::Struct(fields) => {

diff --git a/lib/codecs/src/encoding/format/mod.rs b/lib/codecs/src/encoding/format/mod.rs
@@ -25,6 +25,7 @@ use std::fmt::Debug;
 #[cfg(feature = "arrow")]
 pub use arrow::{
     ArrowEncodingError, ArrowStreamSerializer, ArrowStreamSerializerConfig, SchemaProvider,
+    build_record_batch, make_field_nullable,
 };
 pub use avro::{AvroSerializer, AvroSerializerConfig, AvroSerializerOptions};
 pub use cef::{CefSerializer, CefSerializerConfig};

diff --git a/lib/file-source-common/Cargo.toml b/lib/file-source-common/Cargo.toml
@@ -20,7 +20,7 @@ serde_json = { version = "1.0.143", default-features = false }
 bstr = { version = "1.12", default-features = false }
 bytes = { version = "1.10.1", default-features = false, features = ["serde"] }
 dashmap = { version = "6.1", default-features = false }
-async-compression = { version = "0.4.27", features = ["tokio", "gzip"] }
+async-compression = { version = "=0.4.19", features = ["tokio", "gzip"] }
 vector-common = { path = "../vector-common", default-features = false }
 vector-config = { path = "../vector-config", default-features = false }
 tokio = { workspace = true, features = ["full"] }

diff --git a/lib/file-source/Cargo.toml b/lib/file-source/Cargo.toml
@@ -21,7 +21,7 @@ futures = { version = "0.3.31", default-features = false, features = ["executor"
 futures-util.workspace = true
 vector-common = { path = "../vector-common", default-features = false }
 file-source-common = { path = "../file-source-common" }
-async-compression = { version = "0.4.27", features = ["tokio", "gzip"] }
+async-compression = { version = "=0.4.19", features = ["tokio", "gzip"] }
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["full"] }