laminar_connectors/serde/

mod.rs

//! Record (de)serialization: bytes ↔ Arrow `RecordBatch`. Format
//! implementations in `json`, `csv`, `raw`, `debezium`.

pub mod csv;
pub mod debezium;
pub mod json;
pub mod raw;

use arrow_array::RecordBatch;
use arrow_schema::SchemaRef;

use crate::error::SerdeError;

/// Supported serialization formats.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Format {
    /// JSON format.
    Json,

    /// CSV format.
    Csv,

    /// Raw bytes (no deserialization).
    Raw,

    /// Debezium CDC envelope format.
    Debezium,

    /// Apache Avro format (with optional Confluent Schema Registry).
    Avro,
}

impl Format {
    /// Returns the format name as a string.
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Format::Json => "json",
            Format::Csv => "csv",
            Format::Raw => "raw",
            Format::Debezium => "debezium",
            Format::Avro => "avro",
        }
    }

    /// Parses a format from a string.
    ///
    /// # Errors
    ///
    /// Returns `SerdeError::UnsupportedFormat` if the format is not recognized.
    pub fn parse(s: &str) -> Result<Self, SerdeError> {
        s.parse()
    }
}

impl std::str::FromStr for Format {
    type Err = SerdeError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s.to_lowercase().as_str() {
            "json" => Ok(Format::Json),
            "csv" => Ok(Format::Csv),
            "raw" | "bytes" => Ok(Format::Raw),
            "debezium" | "debezium-json" => Ok(Format::Debezium),
            "avro" | "confluent-avro" => Ok(Format::Avro),
            other => Err(SerdeError::UnsupportedFormat(other.to_string())),
        }
    }
}

impl std::fmt::Display for Format {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.as_str())
    }
}

/// Trait for deserializing raw bytes into Arrow `RecordBatch`.
///
/// Implementations convert from external formats (JSON, CSV, Avro, etc.)
/// into Arrow columnar format for processing.
pub trait RecordDeserializer: Send + Sync {
    /// Deserializes a single record from raw bytes.
    ///
    /// # Errors
    ///
    /// Returns `SerdeError` if the input cannot be parsed or doesn't
    /// match the expected schema.
    fn deserialize(&self, data: &[u8], schema: &SchemaRef) -> Result<RecordBatch, SerdeError>;

    /// Deserializes a batch of records from raw bytes.
    ///
    /// The default implementation calls `deserialize` for each record
    /// and concatenates the results. Implementations should override
    /// this for better performance with batch-oriented formats.
    ///
    /// # Errors
    ///
    /// Returns `SerdeError` if any record cannot be parsed.
    fn deserialize_batch(
        &self,
        records: &[&[u8]],
        schema: &SchemaRef,
    ) -> Result<RecordBatch, SerdeError> {
        if records.is_empty() {
            return Ok(RecordBatch::new_empty(schema.clone()));
        }

        let batches: Result<Vec<RecordBatch>, SerdeError> = records
            .iter()
            .map(|data| self.deserialize(data, schema))
            .collect();
        let batches = batches?;

        arrow_select::concat::concat_batches(schema, &batches)
            .map_err(|e| SerdeError::MalformedInput(format!("failed to concat batches: {e}")))
    }

    /// Returns the format this deserializer handles.
    fn format(&self) -> Format;

    /// Downcasts to concrete type (e.g., for Avro schema resolution).
    fn as_any_mut(&mut self) -> Option<&mut dyn std::any::Any> {
        None
    }
}

/// Trait for serializing Arrow `RecordBatch` into raw bytes.
///
/// Implementations convert from Arrow columnar format to external
/// formats for writing to sinks.
pub trait RecordSerializer: Send + Sync {
    /// Serializes a `RecordBatch` into a vector of byte records.
    ///
    /// Each element in the returned vector represents one serialized record.
    ///
    /// # Errors
    ///
    /// Returns `SerdeError` if serialization fails.
    fn serialize(&self, batch: &RecordBatch) -> Result<Vec<Vec<u8>>, SerdeError>;

    /// Serializes a `RecordBatch` into a single byte buffer.
    ///
    /// For formats that support batch encoding (e.g., JSON array, CSV),
    /// this may be more efficient than serializing individual records.
    ///
    /// # Errors
    ///
    /// Returns `SerdeError` if serialization fails.
    fn serialize_batch(&self, batch: &RecordBatch) -> Result<Vec<u8>, SerdeError> {
        let records = self.serialize(batch)?;
        let total_len: usize = records.iter().map(Vec::len).sum();
        let mut buf = Vec::with_capacity(total_len);
        for record in &records {
            buf.extend_from_slice(record);
        }
        Ok(buf)
    }

    /// Returns the format this serializer produces.
    fn format(&self) -> Format;
}

/// Creates a deserializer for the given format.
///
/// # Errors
///
/// Returns `SerdeError::UnsupportedFormat` if the format is not supported.
pub fn create_deserializer(format: Format) -> Result<Box<dyn RecordDeserializer>, SerdeError> {
    match format {
        Format::Json => Ok(Box::new(json::JsonDeserializer::new())),
        Format::Csv => Ok(Box::new(csv::CsvDeserializer::new())),
        Format::Raw => Ok(Box::new(raw::RawBytesDeserializer::new())),
        Format::Debezium => Ok(Box::new(debezium::DebeziumDeserializer::new())),
        Format::Avro => Err(SerdeError::UnsupportedFormat(
            "Avro requires a connector-provided deserializer".into(),
        )),
    }
}

/// Creates a serializer for the given format.
///
/// # Errors
///
/// Returns `SerdeError::UnsupportedFormat` if the format is not supported.
pub fn create_serializer(format: Format) -> Result<Box<dyn RecordSerializer>, SerdeError> {
    match format {
        Format::Json => Ok(Box::new(json::JsonSerializer::new())),
        Format::Csv => Ok(Box::new(csv::CsvSerializer::new())),
        Format::Raw => Ok(Box::new(raw::RawBytesSerializer::new())),
        Format::Debezium => Err(SerdeError::UnsupportedFormat(
            "Debezium is a deserialization-only format".into(),
        )),
        Format::Avro => Err(SerdeError::UnsupportedFormat(
            "Avro requires a connector-provided serializer".into(),
        )),
    }
}

#[cfg(test)]
mod tests;