Port parquet codec to v0.54.0-exaforce

lib/codecs/Cargo.toml

@@ -31,6 +31,7 @@ memchr = { version = "2", default-features = false }
 metrics.workspace = true
 opentelemetry-proto = { path = "../opentelemetry-proto", optional = true }
 ordered-float.workspace = true
+parquet = { version = "39.0.0", default-features = false }
 pin-project.workspace = true
 prost.workspace = true
 prost-reflect.workspace = true

lib/codecs/src/encoding/config.rs

@@ -1,6 +1,6 @@
 use vector_config::configurable_component;
 
-use super::{Encoder, EncoderKind, Transformer};
+use super::{BatchSerializer, Encoder, EncoderKind, Transformer};
 use crate::encoding::{
     CharacterDelimitedEncoder, Framer, FramingConfig, LengthDelimitedEncoder,
     NewlineDelimitedEncoder, Serializer, SerializerConfig,
@@ -149,6 +149,12 @@ impl EncodingConfigWithFraming {
         let encoder = EncoderKind::Framed(Box::new(Encoder::<Framer>::new(framer, serializer)));
         Ok((self.transformer(), encoder))
     }
+
+    /// Build a `BatchSerializer` for this config, if the configured serializer is batched
+    /// (e.g. Parquet). Returns `None` for per-event serializers.
+    pub fn build_batched(&self) -> vector_common::Result<Option<BatchSerializer>> {
+        self.encoding.config().build_batched()
+    }
 }
 
 /// The way a sink processes outgoing events.

lib/codecs/src/encoding/encoder.rs

@@ -6,7 +6,7 @@ use vector_core::event::Event;
 #[cfg(feature = "arrow")]
 use crate::encoding::ArrowStreamSerializer;
 use crate::{
-    encoding::{Error, Framer, Serializer},
+    encoding::{Error, Framer, ParquetSerializer, Serializer},
     internal_events::{EncoderFramingError, EncoderSerializeError},
 };
 
@@ -16,6 +16,14 @@ pub enum BatchSerializer {
     /// Arrow IPC stream format serializer.
     #[cfg(feature = "arrow")]
     Arrow(ArrowStreamSerializer),
+    /// Apache Parquet file format serializer.
+    Parquet(ParquetSerializer),
+}
+
+impl From<ParquetSerializer> for BatchSerializer {
+    fn from(serializer: ParquetSerializer) -> Self {
+        Self::Parquet(serializer)
+    }
 }
 
 /// An encoder that encodes batches of events.
@@ -36,20 +44,19 @@ impl BatchEncoder {
     }
 
     /// Get the HTTP content type.
-    #[cfg(feature = "arrow")]
     pub const fn content_type(&self) -> &'static str {
         match &self.serializer {
+            #[cfg(feature = "arrow")]
             BatchSerializer::Arrow(_) => "application/vnd.apache.arrow.stream",
+            BatchSerializer::Parquet(_) => "application/vnd.apache.parquet",
         }
     }
 }
 
 impl tokio_util::codec::Encoder<Vec<Event>> for BatchEncoder {
     type Error = Error;
 
-    #[allow(unused_variables)]
     fn encode(&mut self, events: Vec<Event>, buffer: &mut BytesMut) -> Result<(), Self::Error> {
-        #[allow(unreachable_patterns)]
         match &mut self.serializer {
             #[cfg(feature = "arrow")]
             BatchSerializer::Arrow(serializer) => {
@@ -63,7 +70,9 @@ impl tokio_util::codec::Encoder<Vec<Event>> for BatchEncoder {
                     }
                 })
             }
-            _ => unreachable!("BatchSerializer cannot be constructed without encode()"),
+            BatchSerializer::Parquet(serializer) => serializer
+                .encode(events, buffer)
+                .map_err(Error::SerializingError),
         }
     }
 }
@@ -74,7 +83,6 @@ pub enum EncoderKind {
     /// Uses framing to encode individual events
     Framed(Box<Encoder<Framer>>),
     /// Encodes events in batches without framing
-    #[cfg(feature = "arrow")]
     Batch(BatchEncoder),
 }
 

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

@@ -16,6 +16,7 @@ mod native;
 mod native_json;
 #[cfg(feature = "opentelemetry")]
 mod otlp;
+mod parquet;
 mod protobuf;
 mod raw_message;
 #[cfg(feature = "syslog")]
@@ -39,6 +40,7 @@ pub use native::{NativeSerializer, NativeSerializerConfig};
 pub use native_json::{NativeJsonSerializer, NativeJsonSerializerConfig};
 #[cfg(feature = "opentelemetry")]
 pub use otlp::{OtlpSerializer, OtlpSerializerConfig};
+pub use parquet::{ParquetSerializer, ParquetSerializerConfig, ParquetSerializerOptions};
 pub use protobuf::{ProtobufSerializer, ProtobufSerializerConfig, ProtobufSerializerOptions};
 pub use raw_message::{RawMessageSerializer, RawMessageSerializerConfig};
 #[cfg(feature = "syslog")]