laminar_connectors/serde/debezium/
mod.rs

1//! Debezium CDC envelope format deserialization.
2//!
3//! Parses Debezium JSON change events into Arrow `RecordBatch`.
4//!
5//! ## Debezium Envelope Format
6//!
7//! ```json
8//! {
9//!   "before": { ... },       // null for inserts
10//!   "after":  { ... },       // null for deletes
11//!   "source": { ... },       // source metadata
12//!   "op": "c|u|d|r",         // operation: create, update, delete, read (snapshot)
13//!   "ts_ms": 1234567890      // timestamp
14//! }
15//! ```
16//!
17//! The deserializer extracts the `after` payload for inserts/updates
18//! and the `before` payload for deletes, adding `__op` and `__ts_ms`
19//! metadata columns.
20
21use std::sync::Arc;
22
23use arrow_array::builder::{Int64Builder, StringBuilder};
24use arrow_array::{ArrayRef, RecordBatch};
25use arrow_schema::{DataType, Field, Schema, SchemaRef};
26use serde_json::Value;
27
28use super::json::JsonDeserializer;
29use super::{Format, RecordDeserializer};
30use crate::error::SerdeError;
31
32/// Debezium operation types.
33#[derive(Debug, Clone, Copy, PartialEq, Eq)]
34pub enum DebeziumOp {
35    /// Create (insert).
36    Create,
37    /// Update.
38    Update,
39    /// Delete.
40    Delete,
41    /// Read (snapshot).
42    Read,
43}
44
45impl DebeziumOp {
46    /// Parses an operation from the Debezium `op` field.
47    fn from_str(s: &str) -> Result<Self, SerdeError> {
48        match s {
49            "c" => Ok(DebeziumOp::Create),
50            "u" => Ok(DebeziumOp::Update),
51            "d" => Ok(DebeziumOp::Delete),
52            "r" => Ok(DebeziumOp::Read),
53            other => Err(SerdeError::MalformedInput(format!(
54                "unknown Debezium op: {other}"
55            ))),
56        }
57    }
58
59    /// Returns the operation as a string.
60    #[must_use]
61    pub fn as_str(&self) -> &'static str {
62        match self {
63            DebeziumOp::Create => "c",
64            DebeziumOp::Update => "u",
65            DebeziumOp::Delete => "d",
66            DebeziumOp::Read => "r",
67        }
68    }
69}
70
71/// Debezium CDC envelope deserializer.
72///
73/// Extracts the data payload from a Debezium change event and converts
74/// it to an Arrow `RecordBatch`. Two metadata columns are appended:
75/// - `__op`: The operation type (c/u/d/r)
76/// - `__ts_ms`: The event timestamp in milliseconds
77///
78/// The provided schema should describe the data columns only (without
79/// `__op` and `__ts_ms`); these are added automatically.
80#[derive(Debug, Clone)]
81pub struct DebeziumDeserializer {
82    json_deser: JsonDeserializer,
83}
84
85impl DebeziumDeserializer {
86    /// Creates a new Debezium deserializer.
87    #[must_use]
88    pub fn new() -> Self {
89        Self {
90            json_deser: JsonDeserializer::new(),
91        }
92    }
93}
94
95impl Default for DebeziumDeserializer {
96    fn default() -> Self {
97        Self::new()
98    }
99}
100
101fn output_schema(schema: &SchemaRef) -> SchemaRef {
102    let mut fields: Vec<Arc<Field>> = schema.fields().iter().cloned().collect();
103    fields.push(Arc::new(Field::new("__op", DataType::Utf8, false)));
104    fields.push(Arc::new(Field::new("__ts_ms", DataType::Int64, false)));
105    Arc::new(Schema::new(fields))
106}
107
108impl RecordDeserializer for DebeziumDeserializer {
109    fn deserialize(&self, data: &[u8], schema: &SchemaRef) -> Result<RecordBatch, SerdeError> {
110        let envelope: Value = serde_json::from_slice(data)?;
111        let obj = envelope
112            .as_object()
113            .ok_or_else(|| SerdeError::MalformedInput("expected JSON object".into()))?;
114
115        // Extract operation
116        let op_str = obj
117            .get("op")
118            .and_then(Value::as_str)
119            .ok_or_else(|| SerdeError::MissingField("op".into()))?;
120        let op = DebeziumOp::from_str(op_str)?;
121
122        // Extract timestamp
123        let ts_ms = obj.get("ts_ms").and_then(Value::as_i64).unwrap_or(0);
124
125        // Extract payload: `after` for create/update/read, `before` for delete
126        let payload = match op {
127            DebeziumOp::Create | DebeziumOp::Update | DebeziumOp::Read => obj
128                .get("after")
129                .ok_or_else(|| SerdeError::MissingField("after".into()))?,
130            DebeziumOp::Delete => obj
131                .get("before")
132                .ok_or_else(|| SerdeError::MissingField("before".into()))?,
133        };
134
135        if payload.is_null() {
136            return Err(SerdeError::MalformedInput(format!(
137                "payload is null for op={op_str}"
138            )));
139        }
140
141        // Deserialize the payload directly from the parsed Value (avoids double parse)
142        let data_batch = self.json_deser.deserialize_value(payload, schema)?;
143
144        // Append __op and __ts_ms columns
145        let mut columns: Vec<ArrayRef> = data_batch.columns().to_vec();
146
147        let mut op_builder = StringBuilder::with_capacity(1, 1);
148        op_builder.append_value(op.as_str());
149        columns.push(Arc::new(op_builder.finish()));
150
151        let mut ts_builder = Int64Builder::with_capacity(1);
152        ts_builder.append_value(ts_ms);
153        columns.push(Arc::new(ts_builder.finish()));
154
155        RecordBatch::try_new(output_schema(schema), columns)
156            .map_err(|e| SerdeError::MalformedInput(format!("failed to create RecordBatch: {e}")))
157    }
158
159    fn deserialize_batch(
160        &self,
161        records: &[&[u8]],
162        schema: &SchemaRef,
163    ) -> Result<RecordBatch, SerdeError> {
164        if records.is_empty() {
165            return Ok(RecordBatch::new_empty(output_schema(schema)));
166        }
167        let batches = records
168            .iter()
169            .map(|record| self.deserialize(record, schema))
170            .collect::<Result<Vec<_>, _>>()?;
171        let schema = batches[0].schema();
172        arrow_select::concat::concat_batches(&schema, &batches)
173            .map_err(|error| SerdeError::MalformedInput(format!("failed to concat batch: {error}")))
174    }
175
176    fn format(&self) -> Format {
177        Format::Debezium
178    }
179}
180
181#[cfg(test)]
182mod tests;
laminar_connectors/serde/debezium/mod.rs

laminar_connectors/serde/debezium/
mod.rs
