feat(codecs): add wire_to_arrow batch encoder

changelog.d/wire_to_arrow_batch_codec.feature.md

@@ -0,0 +1,10 @@
+Added a `wire_to_arrow` batch encoding codec that decodes protobuf wire bytes from each
+event's `message` field directly into an Apache Arrow `RecordBatch`, pairing proto fields to
+Arrow columns by name. This bypasses the generic `ProtobufDeserializer -> Event ->
+ArrowStreamSerializer` chain for sinks that already carry raw proto bytes, avoiding the
+intermediate `DynamicMessage` / `LogEvent` representations. The codec is configured with a
+proto descriptor (`desc_file` + `message_type`) for the incoming bytes; the sink injects the
+output Arrow schema. Malformed rows are isolated (dropped and counted via the
+`wire_to_arrow_rows_dropped` metric) rather than failing the whole batch.
+
+authors: amandaLi7

lib/codecs/Cargo.toml

@@ -68,6 +68,7 @@ toml = { version = "0.9.8", optional = true }
 criterion.workspace = true
 futures.workspace = true
 indoc.workspace = true
+proptest.workspace = true
 tokio = { workspace = true, features = ["test-util"] }
 toml.workspace = true
 similar-asserts = "1.7.0"

lib/codecs/src/encoding/encoder.rs

@@ -4,7 +4,7 @@ use vector_common::internal_event::emit;
 use vector_core::event::Event;
 
 #[cfg(feature = "arrow")]
-use crate::encoding::ArrowStreamSerializer;
+use crate::encoding::{ArrowStreamSerializer, WireToArrowSerializer};
 #[cfg(feature = "parquet")]
 use crate::encoding::ParquetSerializer;
 use crate::{
@@ -31,6 +31,9 @@ pub enum BatchOutput {
 pub enum BatchSerializer {
     /// Arrow IPC stream format serializer.
     Arrow(ArrowStreamSerializer),
+    /// Wire-to-Arrow serializer: decodes proto wire bytes straight into an
+    /// Arrow `RecordBatch`.
+    WireToArrow(WireToArrowSerializer),
     /// Parquet format serializer.
     #[cfg(feature = "parquet")]
     Parquet(Box<ParquetSerializer>),
@@ -58,7 +61,9 @@ impl BatchEncoder {
     /// Get the HTTP content type.
     pub const fn content_type(&self) -> Option<&'static str> {
         match &self.serializer {
-            BatchSerializer::Arrow(_) => Some("application/vnd.apache.arrow.stream"),
+            BatchSerializer::Arrow(_) | BatchSerializer::WireToArrow(_) => {
+                Some("application/vnd.apache.arrow.stream")
+            }
             #[cfg(feature = "parquet")]
             BatchSerializer::Parquet(_) => Some("application/vnd.apache.parquet"),
         }
@@ -79,6 +84,12 @@ impl BatchEncoder {
                 })?;
                 Ok(BatchOutput::Arrow(record_batch))
             }
+            BatchSerializer::WireToArrow(serializer) => {
+                let record_batch = serializer
+                    .encode_to_record_batch(events)
+                    .map_err(|err| Error::SerializingError(Box::new(err)))?;
+                Ok(BatchOutput::Arrow(record_batch))
+            }
             #[cfg(feature = "parquet")]
             BatchSerializer::Parquet(_) => Err(Error::SerializingError(Box::from(
                 "Parquet serializer does not support encode_batch; use the tokio Encoder interface instead",
@@ -104,6 +115,9 @@ impl tokio_util::codec::Encoder<Vec<Event>> for BatchEncoder {
                     }
                 })
             }
+            BatchSerializer::WireToArrow(_) => Err(Error::SerializingError(Box::from(
+                "WireToArrow serializer does not support the streaming Encoder interface; use encode_batch() instead",
+            ))),
             #[cfg(feature = "parquet")]
             BatchSerializer::Parquet(serializer) => serializer
                 .encode(events, buffer)

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

@@ -23,6 +23,8 @@ mod raw_message;
 #[cfg(feature = "syslog")]
 mod syslog;
 mod text;
+#[cfg(feature = "arrow")]
+mod wire_to_arrow;
 
 use std::fmt::Debug;
 
@@ -35,6 +37,10 @@ pub use arrow::{
     ArrowEncodingError, ArrowStreamSerializer, ArrowStreamSerializerConfig, SchemaProvider,
     find_null_non_nullable_fields,
 };
+#[cfg(feature = "arrow")]
+pub use wire_to_arrow::{
+    WireToArrowEncoder, WireToArrowError, WireToArrowSerializer, WireToArrowSerializerConfig,
+};
 pub use avro::{AvroSerializer, AvroSerializerConfig, AvroSerializerOptions};
 pub use cef::{CefSerializer, CefSerializerConfig};
 use dyn_clone::DynClone;