laminar_connectors/serde/
mod.rs

1//! Record (de)serialization: bytes ↔ Arrow `RecordBatch`. Format
2//! implementations in `json`, `csv`, `raw`, `debezium`.
3
4pub mod csv;
5pub mod debezium;
6pub mod json;
7pub mod raw;
8#[cfg(any(
9    test,
10    feature = "kafka",
11    feature = "nats",
12    feature = "rabbitmq",
13    feature = "ibmmq",
14    feature = "websocket",
15    feature = "files"
16))]
17pub(crate) mod schema_contract;
18
19use arrow_array::RecordBatch;
20use arrow_schema::SchemaRef;
21
22use crate::error::SerdeError;
23
24/// Supported serialization formats.
25#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
26pub enum Format {
27    /// JSON format.
28    Json,
29
30    /// CSV format.
31    Csv,
32
33    /// Raw bytes (no deserialization).
34    Raw,
35
36    /// Debezium CDC envelope format.
37    Debezium,
38
39    /// Apache Avro format (with optional Confluent Schema Registry).
40    Avro,
41
42    /// One complete `FpML` document per record, validated and projected by the `fpml`
43    /// feature's decoder (source-only).
44    Fpml,
45}
46
47impl Format {
48    /// Returns the format name as a string.
49    #[must_use]
50    pub fn as_str(&self) -> &'static str {
51        match self {
52            Format::Json => "json",
53            Format::Csv => "csv",
54            Format::Raw => "raw",
55            Format::Debezium => "debezium",
56            Format::Avro => "avro",
57            Format::Fpml => "fpml",
58        }
59    }
60
61    /// Source options this format reads in addition to the connector's own.
62    #[cfg(any(feature = "rabbitmq", feature = "ibmmq"))]
63    pub(crate) fn source_option_keys(self) -> &'static [&'static str] {
64        match self {
65            #[cfg(feature = "fpml")]
66            Format::Fpml => crate::fpml::message::OPTION_KEYS,
67            _ => &[],
68        }
69    }
70
71    /// Parses a format from a string.
72    ///
73    /// # Errors
74    ///
75    /// Returns `SerdeError::UnsupportedFormat` if the format is not recognized.
76    pub fn parse(s: &str) -> Result<Self, SerdeError> {
77        s.parse()
78    }
79}
80
81impl std::str::FromStr for Format {
82    type Err = SerdeError;
83
84    fn from_str(s: &str) -> Result<Self, Self::Err> {
85        match s.to_lowercase().as_str() {
86            "json" => Ok(Format::Json),
87            "csv" => Ok(Format::Csv),
88            "raw" | "bytes" => Ok(Format::Raw),
89            "debezium" | "debezium-json" => Ok(Format::Debezium),
90            "avro" | "confluent-avro" => Ok(Format::Avro),
91            "fpml" => Ok(Format::Fpml),
92            other => Err(SerdeError::UnsupportedFormat(other.to_string())),
93        }
94    }
95}
96
97impl std::fmt::Display for Format {
98    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
99        write!(f, "{}", self.as_str())
100    }
101}
102
103/// Trait for deserializing raw bytes into Arrow `RecordBatch`.
104///
105/// Implementations convert from external formats (JSON, CSV, Avro, etc.)
106/// into Arrow columnar format for processing.
107pub trait RecordDeserializer: Send + Sync {
108    /// Deserializes a single record from raw bytes.
109    ///
110    /// # Errors
111    ///
112    /// Returns `SerdeError` if the input cannot be parsed or doesn't
113    /// match the expected schema.
114    fn deserialize(&self, data: &[u8], schema: &SchemaRef) -> Result<RecordBatch, SerdeError>;
115
116    /// Deserializes a batch of records from raw bytes.
117    ///
118    /// The default implementation calls `deserialize` for each record
119    /// and concatenates the results. Implementations should override
120    /// this for better performance with batch-oriented formats.
121    ///
122    /// # Errors
123    ///
124    /// Returns `SerdeError` if any record cannot be parsed.
125    fn deserialize_batch(
126        &self,
127        records: &[&[u8]],
128        schema: &SchemaRef,
129    ) -> Result<RecordBatch, SerdeError> {
130        if records.is_empty() {
131            return Ok(RecordBatch::new_empty(schema.clone()));
132        }
133
134        let batches: Result<Vec<RecordBatch>, SerdeError> = records
135            .iter()
136            .map(|data| self.deserialize(data, schema))
137            .collect();
138        let batches = batches?;
139
140        arrow_select::concat::concat_batches(schema, &batches)
141            .map_err(|e| SerdeError::MalformedInput(format!("failed to concat batches: {e}")))
142    }
143
144    /// Returns the format this deserializer handles.
145    fn format(&self) -> Format;
146
147    /// Downcasts to concrete type (e.g., for Avro schema resolution).
148    fn as_any_mut(&mut self) -> Option<&mut dyn std::any::Any> {
149        None
150    }
151}
152
153/// Trait for serializing Arrow `RecordBatch` into raw bytes.
154///
155/// Implementations convert from Arrow columnar format to external
156/// formats for writing to sinks.
157pub trait RecordSerializer: Send + Sync {
158    /// Serializes a `RecordBatch` into a vector of byte records.
159    ///
160    /// Each element in the returned vector represents one serialized record.
161    ///
162    /// # Errors
163    ///
164    /// Returns `SerdeError` if serialization fails.
165    fn serialize(&self, batch: &RecordBatch) -> Result<Vec<Vec<u8>>, SerdeError>;
166
167    /// Serializes a `RecordBatch` into a single byte buffer.
168    ///
169    /// For formats that support batch encoding (e.g., JSON array, CSV),
170    /// this may be more efficient than serializing individual records.
171    ///
172    /// # Errors
173    ///
174    /// Returns `SerdeError` if serialization fails.
175    fn serialize_batch(&self, batch: &RecordBatch) -> Result<Vec<u8>, SerdeError> {
176        let records = self.serialize(batch)?;
177        let total_len: usize = records.iter().map(Vec::len).sum();
178        let mut buf = Vec::with_capacity(total_len);
179        for record in &records {
180            buf.extend_from_slice(record);
181        }
182        Ok(buf)
183    }
184
185    /// Returns the format this serializer produces.
186    fn format(&self) -> Format;
187}
188
189/// Creates a deserializer for the given format.
190///
191/// # Errors
192///
193/// Returns `SerdeError::UnsupportedFormat` if the format is not supported.
194pub fn create_deserializer(format: Format) -> Result<Box<dyn RecordDeserializer>, SerdeError> {
195    match format {
196        Format::Json => Ok(Box::new(json::JsonDeserializer::new())),
197        Format::Csv => Ok(Box::new(csv::CsvDeserializer::new())),
198        Format::Raw => Ok(Box::new(raw::RawBytesDeserializer::new())),
199        Format::Debezium => Ok(Box::new(debezium::DebeziumDeserializer::new())),
200        Format::Avro => Err(SerdeError::UnsupportedFormat(
201            "Avro requires a connector-provided deserializer".into(),
202        )),
203        Format::Fpml => Err(SerdeError::UnsupportedFormat(
204            "FpML requires a connector-provided decoder".into(),
205        )),
206    }
207}
208
209/// Creates a serializer for the given format.
210///
211/// # Errors
212///
213/// Returns `SerdeError::UnsupportedFormat` if the format is not supported.
214pub fn create_serializer(format: Format) -> Result<Box<dyn RecordSerializer>, SerdeError> {
215    match format {
216        Format::Json => Ok(Box::new(json::JsonSerializer::new())),
217        Format::Csv => Ok(Box::new(csv::CsvSerializer::new())),
218        Format::Raw => Ok(Box::new(raw::RawBytesSerializer::new())),
219        Format::Debezium => Err(SerdeError::UnsupportedFormat(
220            "Debezium is a deserialization-only format".into(),
221        )),
222        Format::Avro => Err(SerdeError::UnsupportedFormat(
223            "Avro requires a connector-provided serializer".into(),
224        )),
225        Format::Fpml => Err(SerdeError::UnsupportedFormat(
226            "FpML is a source-only format".into(),
227        )),
228    }
229}
230
231#[cfg(test)]
232mod tests;
laminar_connectors/serde/mod.rs

laminar_connectors/serde/
mod.rs
