laminar_connectors/postgres/cdc/decoder/
mod.rs

1//! `PostgreSQL` `pgoutput` logical replication protocol decoder.
2//!
3//! Implements a binary protocol parser for the `pgoutput` output plugin
4//! used by `PostgreSQL` logical replication (PG 10+). Parses WAL stream
5//! bytes into structured [`WalMessage`] variants.
6//!
7//! # Protocol Reference
8//!
9//! See `PostgreSQL` docs: "Logical Replication Message Formats"
10//! (<https://www.postgresql.org/docs/current/protocol-logicalrep-message-formats.html>)
11
12use super::lsn::Lsn;
13use super::types::PgColumn;
14use bytes::Bytes;
15
16/// Offset from `PostgreSQL` epoch (2000-01-01) to Unix epoch (1970-01-01)
17/// in microseconds.
18const PG_EPOCH_OFFSET_US: i64 = 946_684_800_000_000;
19const MAX_POSTGRES_COLUMNS: usize = 1_600;
20
21/// A decoded WAL message from the `pgoutput` protocol.
22#[derive(Debug, Clone, PartialEq)]
23pub enum WalMessage {
24    /// Transaction begin.
25    Begin(BeginMessage),
26    /// Transaction commit.
27    Commit(CommitMessage),
28    /// Relation (table) metadata.
29    Relation(RelationMessage),
30    /// Row inserted.
31    Insert(InsertMessage),
32    /// Row updated.
33    Update(UpdateMessage),
34    /// Row deleted.
35    Delete(DeleteMessage),
36    /// Table(s) truncated.
37    Truncate(TruncateMessage),
38    /// Origin information.
39    Origin(OriginMessage),
40    /// Custom type definition.
41    Type(TypeMessage),
42}
43
44/// Transaction begin message.
45#[derive(Debug, Clone, PartialEq, Eq)]
46pub struct BeginMessage {
47    /// LSN of the final record of the transaction.
48    pub final_lsn: Lsn,
49    /// Commit timestamp in milliseconds since Unix epoch.
50    pub commit_ts_ms: i64,
51    /// Transaction ID (XID).
52    pub xid: u32,
53}
54
55/// Transaction commit message.
56#[derive(Debug, Clone, PartialEq, Eq)]
57pub struct CommitMessage {
58    /// Flags (currently unused by `PostgreSQL`).
59    pub flags: u8,
60    /// LSN of the commit record.
61    pub commit_lsn: Lsn,
62    /// End LSN of the transaction.
63    pub end_lsn: Lsn,
64    /// Commit timestamp in milliseconds since Unix epoch.
65    pub commit_ts_ms: i64,
66}
67
68/// Relation (table schema) message.
69#[derive(Debug, Clone, PartialEq, Eq)]
70pub struct RelationMessage {
71    /// Relation OID.
72    pub relation_id: u32,
73    /// Schema (namespace) name.
74    pub namespace: String,
75    /// Table name.
76    pub name: String,
77    /// Replica identity setting: 'd', 'n', 'f', or 'i'.
78    pub replica_identity: u8,
79    /// Column descriptors.
80    pub columns: Vec<PgColumn>,
81}
82
83/// Row insert message.
84#[derive(Debug, Clone, PartialEq)]
85pub struct InsertMessage {
86    /// Relation OID of the target table.
87    pub relation_id: u32,
88    /// The new row data.
89    pub new_tuple: TupleData,
90}
91
92/// Old tuple representation identified by the pgoutput wire tag.
93#[derive(Debug, Clone, PartialEq)]
94pub enum OldTuple {
95    /// Replica-identity key (`K`); non-key column positions are unavailable.
96    Key(TupleData),
97    /// Old tuple (`O`) emitted for `REPLICA IDENTITY FULL`.
98    Full(TupleData),
99}
100
101/// Row update message.
102#[derive(Debug, Clone, PartialEq)]
103pub struct UpdateMessage {
104    /// Relation OID of the target table.
105    pub relation_id: u32,
106    /// Old identity/full-row data when pgoutput includes a `K` or `O` tuple.
107    pub old_tuple: Option<OldTuple>,
108    /// The new row data.
109    pub new_tuple: TupleData,
110}
111
112/// Row delete message.
113#[derive(Debug, Clone, PartialEq)]
114pub struct DeleteMessage {
115    /// Relation OID of the target table.
116    pub relation_id: u32,
117    /// The old identity or full-row tuple.
118    pub old_tuple: OldTuple,
119}
120
121/// Table truncate message.
122#[derive(Debug, Clone, PartialEq, Eq)]
123pub struct TruncateMessage {
124    /// Relation OIDs of truncated tables.
125    pub relation_ids: Vec<u32>,
126    /// Option flags: bit 0 = CASCADE, bit 1 = RESTART IDENTITY.
127    pub options: u8,
128}
129
130/// Origin message (for replication from a downstream).
131#[derive(Debug, Clone, PartialEq, Eq)]
132pub struct OriginMessage {
133    /// Origin LSN.
134    pub origin_lsn: Lsn,
135    /// Origin name.
136    pub name: String,
137}
138
139/// Custom type definition message.
140#[derive(Debug, Clone, PartialEq, Eq)]
141pub struct TypeMessage {
142    /// Type OID.
143    pub type_id: u32,
144    /// Schema (namespace) name.
145    pub namespace: String,
146    /// Type name.
147    pub name: String,
148}
149
150/// Tuple data containing column values.
151#[derive(Debug, Clone, PartialEq)]
152pub struct TupleData {
153    /// Column values in ordinal order.
154    pub columns: Vec<ColumnValue>,
155}
156
157/// A single column value in a tuple.
158#[derive(Debug, Clone, PartialEq)]
159pub enum ColumnValue {
160    /// NULL value.
161    Null,
162    /// Unchanged TOAST value (not sent by server).
163    Unchanged,
164    /// Text-format value.
165    Text(Bytes),
166}
167
168impl ColumnValue {
169    /// Returns the text value if present.
170    #[must_use]
171    pub fn as_text(&self) -> Option<&str> {
172        match self {
173            ColumnValue::Text(bytes) => std::str::from_utf8(bytes).ok(),
174            _ => None,
175        }
176    }
177
178    /// Returns `true` if the value is NULL.
179    #[must_use]
180    pub fn is_null(&self) -> bool {
181        matches!(self, ColumnValue::Null)
182    }
183}
184
185/// Errors from the `pgoutput` protocol decoder.
186#[derive(Debug, Clone, thiserror::Error)]
187pub enum DecoderError {
188    /// Not enough bytes to read the expected value.
189    #[error("unexpected end of data at offset {offset}, need {needed} bytes")]
190    UnexpectedEof {
191        /// Current position in the buffer.
192        offset: usize,
193        /// Number of bytes needed.
194        needed: usize,
195    },
196
197    /// Invalid message type byte.
198    #[error("unknown message type: 0x{0:02X}")]
199    UnknownMessageType(u8),
200
201    /// Invalid or corrupted data.
202    #[error("invalid data: {0}")]
203    InvalidData(String),
204
205    /// Invalid UTF-8 in a string field.
206    #[error("invalid UTF-8 at offset {0}")]
207    InvalidUtf8(usize),
208}
209
210/// A cursor for reading binary data from a byte buffer.
211struct Cursor {
212    data: Bytes,
213    pos: usize,
214}
215
216impl Cursor {
217    fn new(data: Bytes) -> Self {
218        Self { data, pos: 0 }
219    }
220
221    fn remaining(&self) -> usize {
222        self.data.len().saturating_sub(self.pos)
223    }
224
225    fn read_u8(&mut self) -> Result<u8, DecoderError> {
226        if self.pos >= self.data.len() {
227            return Err(DecoderError::UnexpectedEof {
228                offset: self.pos,
229                needed: 1,
230            });
231        }
232        let val = self.data[self.pos];
233        self.pos += 1;
234        Ok(val)
235    }
236
237    fn read_i16(&mut self) -> Result<i16, DecoderError> {
238        self.check_remaining(2)?;
239        let val = i16::from_be_bytes([self.data[self.pos], self.data[self.pos + 1]]);
240        self.pos += 2;
241        Ok(val)
242    }
243
244    fn read_i32(&mut self) -> Result<i32, DecoderError> {
245        self.check_remaining(4)?;
246        let bytes: [u8; 4] = self.data[self.pos..self.pos + 4].try_into().map_err(|_| {
247            DecoderError::UnexpectedEof {
248                offset: self.pos,
249                needed: 4,
250            }
251        })?;
252        let val = i32::from_be_bytes(bytes);
253        self.pos += 4;
254        Ok(val)
255    }
256
257    fn read_u32(&mut self) -> Result<u32, DecoderError> {
258        self.check_remaining(4)?;
259        let bytes: [u8; 4] = self.data[self.pos..self.pos + 4].try_into().map_err(|_| {
260            DecoderError::UnexpectedEof {
261                offset: self.pos,
262                needed: 4,
263            }
264        })?;
265        let val = u32::from_be_bytes(bytes);
266        self.pos += 4;
267        Ok(val)
268    }
269
270    fn read_i64(&mut self) -> Result<i64, DecoderError> {
271        self.check_remaining(8)?;
272        let bytes: [u8; 8] = self.data[self.pos..self.pos + 8].try_into().map_err(|_| {
273            DecoderError::UnexpectedEof {
274                offset: self.pos,
275                needed: 8,
276            }
277        })?;
278        let val = i64::from_be_bytes(bytes);
279        self.pos += 8;
280        Ok(val)
281    }
282
283    fn read_u64(&mut self) -> Result<u64, DecoderError> {
284        self.check_remaining(8)?;
285        let bytes: [u8; 8] = self.data[self.pos..self.pos + 8].try_into().map_err(|_| {
286            DecoderError::UnexpectedEof {
287                offset: self.pos,
288                needed: 8,
289            }
290        })?;
291        let val = u64::from_be_bytes(bytes);
292        self.pos += 8;
293        Ok(val)
294    }
295
296    /// Reads a null-terminated string.
297    fn read_cstring(&mut self) -> Result<String, DecoderError> {
298        let start = self.pos;
299        let nul_pos = self.data[self.pos..]
300            .iter()
301            .position(|&b| b == 0)
302            .ok_or(DecoderError::InvalidData("unterminated string".to_string()))?;
303
304        let s = std::str::from_utf8(&self.data[self.pos..self.pos + nul_pos])
305            .map_err(|_| DecoderError::InvalidUtf8(start))?;
306
307        self.pos += nul_pos + 1; // skip the NUL byte
308        Ok(s.to_string())
309    }
310
311    fn read_bytes(&mut self, len: usize) -> Result<Bytes, DecoderError> {
312        self.check_remaining(len)?;
313        let slice = self.data.slice(self.pos..self.pos + len);
314        self.pos += len;
315        Ok(slice)
316    }
317
318    fn check_remaining(&self, needed: usize) -> Result<(), DecoderError> {
319        if self.remaining() < needed {
320            return Err(DecoderError::UnexpectedEof {
321                offset: self.pos,
322                needed,
323            });
324        }
325        Ok(())
326    }
327}
328
329/// Converts a `PostgreSQL` timestamp (microseconds since 2000-01-01) to
330/// milliseconds since Unix epoch (1970-01-01).
331///
332/// # Errors
333///
334/// Returns [`DecoderError`] when converting to the Unix epoch overflows.
335pub(super) fn pg_timestamp_to_unix_ms(pg_us: i64) -> Result<i64, DecoderError> {
336    pg_us
337        .checked_add(PG_EPOCH_OFFSET_US)
338        .map(|unix_us| unix_us.div_euclid(1000))
339        .ok_or_else(|| DecoderError::InvalidData("PostgreSQL timestamp overflow".into()))
340}
341
342/// Decodes a single `pgoutput` WAL message from raw bytes.
343///
344/// # Errors
345///
346/// Returns [`DecoderError`] if the data is truncated, malformed, or
347/// contains an unknown message type.
348pub(super) fn decode_message(data: Bytes) -> Result<WalMessage, DecoderError> {
349    if data.is_empty() {
350        return Err(DecoderError::InvalidData("empty message".to_string()));
351    }
352
353    let mut cur = Cursor::new(data);
354    let msg_type = cur.read_u8()?;
355
356    let message = match msg_type {
357        b'B' => decode_begin(&mut cur),
358        b'C' => decode_commit(&mut cur),
359        b'R' => decode_relation(&mut cur),
360        b'I' => decode_insert(&mut cur),
361        b'U' => decode_update(&mut cur),
362        b'D' => decode_delete(&mut cur),
363        b'T' => decode_truncate(&mut cur),
364        b'O' => decode_origin(&mut cur),
365        b'Y' => decode_type(&mut cur),
366        _ => Err(DecoderError::UnknownMessageType(msg_type)),
367    }?;
368    if cur.remaining() != 0 {
369        return Err(DecoderError::InvalidData(format!(
370            "trailing bytes after pgoutput message: {}",
371            cur.remaining()
372        )));
373    }
374    Ok(message)
375}
376
377fn decode_begin(cur: &mut Cursor) -> Result<WalMessage, DecoderError> {
378    let final_lsn = Lsn::new(cur.read_u64()?);
379    let commit_ts_us = cur.read_i64()?;
380    let xid = cur.read_u32()?;
381    Ok(WalMessage::Begin(BeginMessage {
382        final_lsn,
383        commit_ts_ms: pg_timestamp_to_unix_ms(commit_ts_us)?,
384        xid,
385    }))
386}
387
388fn decode_commit(cur: &mut Cursor) -> Result<WalMessage, DecoderError> {
389    let flags = cur.read_u8()?;
390    if flags != 0 {
391        return Err(DecoderError::InvalidData(format!(
392            "unsupported COMMIT flags: 0x{flags:02X}"
393        )));
394    }
395    let commit_lsn = Lsn::new(cur.read_u64()?);
396    let end_lsn = Lsn::new(cur.read_u64()?);
397    let commit_ts_us = cur.read_i64()?;
398    Ok(WalMessage::Commit(CommitMessage {
399        flags,
400        commit_lsn,
401        end_lsn,
402        commit_ts_ms: pg_timestamp_to_unix_ms(commit_ts_us)?,
403    }))
404}
405
406fn decode_relation(cur: &mut Cursor) -> Result<WalMessage, DecoderError> {
407    let relation_id = cur.read_u32()?;
408    let namespace = cur.read_cstring()?;
409    let name = cur.read_cstring()?;
410    let replica_identity = cur.read_u8()?;
411    let n_cols_raw = cur.read_i16()?;
412    let n_cols = usize::try_from(n_cols_raw)
413        .map_err(|_| DecoderError::InvalidData(format!("negative column count: {n_cols_raw}")))?;
414    validate_column_count(n_cols)?;
415
416    let mut columns = Vec::with_capacity(n_cols);
417    for _ in 0..n_cols {
418        let flags = cur.read_u8()?;
419        let col_name = cur.read_cstring()?;
420        let type_oid = cur.read_u32()?;
421        let type_modifier = cur.read_i32()?;
422        columns.push(PgColumn::new(
423            col_name,
424            type_oid,
425            type_modifier,
426            flags & 1 != 0,
427        ));
428    }
429
430    Ok(WalMessage::Relation(RelationMessage {
431        relation_id,
432        namespace,
433        name,
434        replica_identity,
435        columns,
436    }))
437}
438
439fn decode_insert(cur: &mut Cursor) -> Result<WalMessage, DecoderError> {
440    let relation_id = cur.read_u32()?;
441    let tag = cur.read_u8()?;
442    if tag != b'N' {
443        return Err(DecoderError::InvalidData(format!(
444            "expected 'N' tag in INSERT, got 0x{tag:02X}"
445        )));
446    }
447    let new_tuple = decode_tuple_data(cur)?;
448    Ok(WalMessage::Insert(InsertMessage {
449        relation_id,
450        new_tuple,
451    }))
452}
453
454fn decode_update(cur: &mut Cursor) -> Result<WalMessage, DecoderError> {
455    let relation_id = cur.read_u32()?;
456    let tag = cur.read_u8()?;
457
458    let (old_tuple, new_tuple) = match tag {
459        // No old tuple, just new
460        b'N' => (None, decode_tuple_data(cur)?),
461        // Old replica-identity key or full tuple followed by new.
462        b'K' | b'O' => {
463            let old_data = decode_tuple_data(cur)?;
464            let old = if tag == b'K' {
465                OldTuple::Key(old_data)
466            } else {
467                OldTuple::Full(old_data)
468            };
469            let new_tag = cur.read_u8()?;
470            if new_tag != b'N' {
471                return Err(DecoderError::InvalidData(format!(
472                    "expected 'N' tag after old tuple in UPDATE, got 0x{new_tag:02X}"
473                )));
474            }
475            let new = decode_tuple_data(cur)?;
476            (Some(old), new)
477        }
478        _ => {
479            return Err(DecoderError::InvalidData(format!(
480                "unexpected tag in UPDATE: 0x{tag:02X}"
481            )));
482        }
483    };
484
485    Ok(WalMessage::Update(UpdateMessage {
486        relation_id,
487        old_tuple,
488        new_tuple,
489    }))
490}
491
492fn decode_delete(cur: &mut Cursor) -> Result<WalMessage, DecoderError> {
493    let relation_id = cur.read_u32()?;
494    let tag = cur.read_u8()?;
495    if tag != b'K' && tag != b'O' {
496        return Err(DecoderError::InvalidData(format!(
497            "expected 'K' or 'O' tag in DELETE, got 0x{tag:02X}"
498        )));
499    }
500    let old_data = decode_tuple_data(cur)?;
501    let old_tuple = if tag == b'K' {
502        OldTuple::Key(old_data)
503    } else {
504        OldTuple::Full(old_data)
505    };
506    Ok(WalMessage::Delete(DeleteMessage {
507        relation_id,
508        old_tuple,
509    }))
510}
511
512fn decode_truncate(cur: &mut Cursor) -> Result<WalMessage, DecoderError> {
513    let n_relations_raw = cur.read_u32()?;
514    let n_relations = usize::try_from(n_relations_raw).map_err(|_| {
515        DecoderError::InvalidData(format!(
516            "TRUNCATE relation count {n_relations_raw} does not fit this platform"
517        ))
518    })?;
519    let options = cur.read_u8()?;
520    let relation_bytes = n_relations
521        .checked_mul(std::mem::size_of::<u32>())
522        .ok_or_else(|| DecoderError::InvalidData("TRUNCATE relation count overflow".into()))?;
523    if relation_bytes != cur.remaining() {
524        return Err(DecoderError::InvalidData(format!(
525            "TRUNCATE relation count {n_relations} requires {relation_bytes} bytes, but {} remain",
526            cur.remaining()
527        )));
528    }
529    let mut relation_ids = Vec::with_capacity(n_relations);
530    for _ in 0..n_relations {
531        relation_ids.push(cur.read_u32()?);
532    }
533    Ok(WalMessage::Truncate(TruncateMessage {
534        relation_ids,
535        options,
536    }))
537}
538
539fn decode_origin(cur: &mut Cursor) -> Result<WalMessage, DecoderError> {
540    let origin_lsn = Lsn::new(cur.read_u64()?);
541    let name = cur.read_cstring()?;
542    Ok(WalMessage::Origin(OriginMessage { origin_lsn, name }))
543}
544
545fn decode_type(cur: &mut Cursor) -> Result<WalMessage, DecoderError> {
546    let type_id = cur.read_u32()?;
547    let namespace = cur.read_cstring()?;
548    let name = cur.read_cstring()?;
549    Ok(WalMessage::Type(TypeMessage {
550        type_id,
551        namespace,
552        name,
553    }))
554}
555
556fn decode_tuple_data(cur: &mut Cursor) -> Result<TupleData, DecoderError> {
557    let n_cols_raw = cur.read_i16()?;
558    let n_cols = usize::try_from(n_cols_raw)
559        .map_err(|_| DecoderError::InvalidData(format!("negative column count: {n_cols_raw}")))?;
560    validate_column_count(n_cols)?;
561    let mut columns = Vec::with_capacity(n_cols);
562
563    for _ in 0..n_cols {
564        let col_type = cur.read_u8()?;
565        match col_type {
566            b'n' => columns.push(ColumnValue::Null),
567            b'u' => columns.push(ColumnValue::Unchanged),
568            b't' => {
569                let len_raw = cur.read_i32()?;
570                let len = usize::try_from(len_raw).map_err(|_| {
571                    DecoderError::InvalidData(format!("negative text length: {len_raw}"))
572                })?;
573                let data = cur.read_bytes(len)?;
574                std::str::from_utf8(&data).map_err(|_| DecoderError::InvalidUtf8(cur.pos - len))?;
575                columns.push(ColumnValue::Text(data));
576            }
577            _ => {
578                return Err(DecoderError::InvalidData(format!(
579                    "unknown column type: 0x{col_type:02X}"
580                )));
581            }
582        }
583    }
584
585    Ok(TupleData { columns })
586}
587
588fn validate_column_count(count: usize) -> Result<(), DecoderError> {
589    if count > MAX_POSTGRES_COLUMNS {
590        return Err(DecoderError::InvalidData(format!(
591            "column count {count} exceeds PostgreSQL maximum {MAX_POSTGRES_COLUMNS}"
592        )));
593    }
594    Ok(())
595}
596
597#[cfg(test)]
598mod tests;
laminar_connectors/postgres/cdc/decoder/mod.rs

laminar_connectors/postgres/cdc/decoder/
mod.rs
