datum_cdc/
event.rs

1use serde::{Deserialize, Serialize};
2
3use crate::PgLsn;
4
5/// Metadata identifying the PostgreSQL replication source that produced an
6/// event.
7#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
8pub struct SourceMetadata {
9    pub database: String,
10    pub slot: String,
11    pub publication: String,
12}
13
14/// PostgreSQL replica identity for a relation.
15#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
16pub enum ReplicaIdentity {
17    Default,
18    Nothing,
19    Full,
20    Index,
21    Unknown(u8),
22}
23
24impl ReplicaIdentity {
25    #[must_use]
26    pub fn from_wire(value: u8) -> Self {
27        match value {
28            b'd' => Self::Default,
29            b'n' => Self::Nothing,
30            b'f' => Self::Full,
31            b'i' => Self::Index,
32            other => Self::Unknown(other),
33        }
34    }
35}
36
37/// Column metadata as sent by a pgoutput `Relation` message.
38#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
39pub struct ColumnMetadata {
40    pub name: String,
41    pub type_oid: u32,
42    pub type_modifier: i32,
43    pub key: bool,
44}
45
46/// Relation metadata cached by relation OID.
47#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
48pub struct RelationMetadata {
49    pub oid: u32,
50    pub schema: String,
51    pub table: String,
52    pub replica_identity: ReplicaIdentity,
53    pub columns: Vec<ColumnMetadata>,
54    pub revision: u64,
55}
56
57impl RelationMetadata {
58    #[must_use]
59    pub fn column_index(&self, name: &str) -> Option<usize> {
60        self.columns.iter().position(|column| column.name == name)
61    }
62}
63
64/// A text-mode tuple value from pgoutput.
65#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
66pub enum ColumnValue {
67    Null,
68    ToastUnchanged,
69    Text(String),
70    Binary(Vec<u8>),
71}
72
73impl ColumnValue {
74    #[must_use]
75    pub fn as_text(&self) -> Option<&str> {
76        match self {
77            Self::Text(value) => Some(value),
78            Self::Null | Self::ToastUnchanged | Self::Binary(_) => None,
79        }
80    }
81}
82
83/// Row tuple values in relation-column order.
84#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
85pub struct RowData {
86    pub values: Vec<ColumnValue>,
87}
88
89impl RowData {
90    #[must_use]
91    pub fn get<'a>(
92        &'a self,
93        relation: &'a RelationMetadata,
94        column: &str,
95    ) -> Option<&'a ColumnValue> {
96        relation
97            .column_index(column)
98            .and_then(|index| self.values.get(index))
99    }
100
101    #[must_use]
102    pub fn get_text<'a>(&'a self, relation: &'a RelationMetadata, column: &str) -> Option<&'a str> {
103        self.get(relation, column).and_then(ColumnValue::as_text)
104    }
105}
106
107/// CDC operation kind.
108#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
109pub enum ChangeOperation {
110    Insert,
111    Update,
112    Delete,
113    Truncate,
114}
115
116impl ChangeOperation {
117    #[must_use]
118    pub const fn debezium_letter(self) -> &'static str {
119        match self {
120            Self::Insert => "c",
121            Self::Update => "u",
122            Self::Delete => "d",
123            Self::Truncate => "t",
124        }
125    }
126}
127
128/// Options attached to a pgoutput truncate message.
129#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
130pub struct TruncateOptions {
131    pub cascade: bool,
132    pub restart_identity: bool,
133}
134
135/// Stable offset for downstream checkpointing and de-duplication.
136#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
137pub struct CdcOffset {
138    pub slot: String,
139    pub tx_end_lsn: PgLsn,
140    pub commit_lsn: PgLsn,
141    pub xid: u32,
142    pub event_index: u32,
143    pub event_count: u32,
144}
145
146/// Transaction metadata attached after the commit boundary is known.
147#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
148pub struct TransactionMeta {
149    pub xid: u32,
150    /// Microseconds since PostgreSQL's epoch (`2000-01-01 00:00:00 UTC`).
151    pub commit_time_micros: i64,
152    pub event_index: u32,
153    pub event_count: u32,
154}
155
156/// One row/table change from PostgreSQL logical replication.
157#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
158pub struct ChangeEvent {
159    pub source: SourceMetadata,
160    pub schema: String,
161    pub table: String,
162    pub op: ChangeOperation,
163    pub before: Option<RowData>,
164    pub after: Option<RowData>,
165    pub truncate: Option<TruncateOptions>,
166    pub lsn: CdcOffset,
167    pub tx: TransactionMeta,
168    pub relation: RelationMetadata,
169}
datum_cdc/event.rs

datum_cdc/
event.rs
