pgsync_core/
event.rs

1//! The logical change model shared by decoders, transformers (Beta 2) and sinks.
2//!
3//! Values are kept in the wire representation (`Bytes`) so that forwarding a change
4//! does not allocate per column; conversion happens in the sink.
5
6use crate::lsn::Lsn;
7use bytes::Bytes;
8use std::sync::Arc;
9
10/// `pg_class.relreplident`.
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub enum ReplicaIdentity {
13    /// Primary key (`d`).
14    Default,
15    /// No identity (`n`): UPDATE/DELETE cannot be replicated.
16    Nothing,
17    /// All columns (`f`).
18    Full,
19    /// A unique index (`i`).
20    Index,
21}
22
23impl ReplicaIdentity {
24    pub fn from_char(c: u8) -> Option<Self> {
25        Some(match c {
26            b'd' => ReplicaIdentity::Default,
27            b'n' => ReplicaIdentity::Nothing,
28            b'f' => ReplicaIdentity::Full,
29            b'i' => ReplicaIdentity::Index,
30            _ => return None,
31        })
32    }
33}
34
35#[derive(Debug, Clone, PartialEq, Eq)]
36pub struct Column {
37    pub name: String,
38    pub type_oid: u32,
39    pub type_modifier: i32,
40    /// Part of the replica identity key.
41    pub is_key: bool,
42}
43
44/// Relation metadata as sent by pgoutput. Cached by the sink by `id`.
45#[derive(Debug, Clone, PartialEq, Eq)]
46pub struct Relation {
47    pub id: u32,
48    pub schema: String,
49    pub name: String,
50    pub replica_identity: ReplicaIdentity,
51    pub columns: Vec<Column>,
52}
53
54impl Relation {
55    pub fn key_columns(&self) -> impl Iterator<Item = (usize, &Column)> {
56        self.columns.iter().enumerate().filter(|(_, c)| c.is_key)
57    }
58
59    pub fn qualified_name(&self) -> String {
60        format!("{}.{}", self.schema, self.name)
61    }
62}
63
64#[derive(Debug, Clone, PartialEq, Eq)]
65pub enum Value {
66    Null,
67    /// TOASTed value that did not change; the sink must leave the stored value alone.
68    Unchanged,
69    /// Text representation (output function). The default for Beta 1.
70    Text(Bytes),
71    /// Binary send-function representation (`binary` option; unused in Beta 1).
72    Binary(Bytes),
73}
74
75impl Value {
76    pub fn byte_len(&self) -> usize {
77        match self {
78            Value::Null | Value::Unchanged => 0,
79            Value::Text(b) | Value::Binary(b) => b.len(),
80        }
81    }
82
83    pub fn text(s: &str) -> Value {
84        Value::Text(Bytes::copy_from_slice(s.as_bytes()))
85    }
86}
87
88#[derive(Debug, Clone, PartialEq, Eq, Default)]
89pub struct Tuple {
90    pub values: Vec<Value>,
91}
92
93impl Tuple {
94    pub fn byte_len(&self) -> usize {
95        self.values.iter().map(|v| v.byte_len() + 8).sum()
96    }
97}
98
99#[derive(Debug, Clone, PartialEq, Eq)]
100pub enum Op {
101    Insert {
102        relation: u32,
103        new: Tuple,
104    },
105    Update {
106        relation: u32,
107        /// Old key (`K`) when a key column changed.
108        key: Option<Tuple>,
109        /// Old full row (`O`) for REPLICA IDENTITY FULL.
110        old: Option<Tuple>,
111        new: Tuple,
112    },
113    Delete {
114        relation: u32,
115        key: Option<Tuple>,
116        old: Option<Tuple>,
117    },
118    Truncate {
119        relations: Vec<u32>,
120        cascade: bool,
121        restart_identity: bool,
122    },
123}
124
125impl Op {
126    pub fn relation_ids(&self) -> Vec<u32> {
127        match self {
128            Op::Insert { relation, .. }
129            | Op::Update { relation, .. }
130            | Op::Delete { relation, .. } => {
131                vec![*relation]
132            }
133            Op::Truncate { relations, .. } => relations.clone(),
134        }
135    }
136
137    pub fn byte_len(&self) -> usize {
138        let base = 32;
139        match self {
140            Op::Insert { new, .. } => base + new.byte_len(),
141            Op::Update { key, old, new, .. } => {
142                base + new.byte_len()
143                    + key.as_ref().map_or(0, Tuple::byte_len)
144                    + old.as_ref().map_or(0, Tuple::byte_len)
145            }
146            Op::Delete { key, old, .. } => {
147                base + key.as_ref().map_or(0, Tuple::byte_len)
148                    + old.as_ref().map_or(0, Tuple::byte_len)
149            }
150            Op::Truncate { relations, .. } => base + relations.len() * 4,
151        }
152    }
153
154    pub fn kind(&self) -> &'static str {
155        match self {
156            Op::Insert { .. } => "insert",
157            Op::Update { .. } => "update",
158            Op::Delete { .. } => "delete",
159            Op::Truncate { .. } => "truncate",
160        }
161    }
162}
163
164/// Ordered events produced by a replication source.
165#[derive(Debug, Clone, PartialEq, Eq)]
166pub enum SourceEvent {
167    Begin {
168        /// LSN of the transaction's commit record (equals `Commit::commit_lsn`).
169        final_lsn: Lsn,
170        /// PostgreSQL timestamp (microseconds since 2000-01-01).
171        commit_ts: i64,
172        xid: u32,
173    },
174    Relation(Arc<Relation>),
175    Op(Op),
176    Commit {
177        commit_lsn: Lsn,
178        /// LSN after the commit record; this is what gets acknowledged.
179        end_lsn: Lsn,
180        commit_ts: i64,
181    },
182    /// Server keepalive. Safe to acknowledge only between transactions.
183    Keepalive {
184        wal_end: Lsn,
185    },
186}
187
188impl SourceEvent {
189    /// Approximate memory footprint, used for inflight accounting.
190    pub fn byte_len(&self) -> usize {
191        match self {
192            SourceEvent::Begin { .. }
193            | SourceEvent::Commit { .. }
194            | SourceEvent::Keepalive { .. } => 32,
195            SourceEvent::Relation(r) => 64 + r.columns.len() * 48,
196            SourceEvent::Op(op) => op.byte_len(),
197        }
198    }
199}
200
201/// Converts a PostgreSQL timestamp (µs since 2000-01-01) to unix microseconds.
202pub fn pg_micros_to_unix_micros(pg: i64) -> i64 {
203    pg + 946_684_800_000_000
204}
205
206/// Current time as a PostgreSQL timestamp (µs since 2000-01-01).
207pub fn now_pg_micros() -> i64 {
208    let unix = std::time::SystemTime::now()
209        .duration_since(std::time::UNIX_EPOCH)
210        .map(|d| d.as_micros() as i64)
211        .unwrap_or(0);
212    unix - 946_684_800_000_000
213}
214
215#[cfg(test)]
216mod tests {
217    use super::*;
218
219    #[test]
220    fn byte_accounting_counts_payloads() {
221        let t = Tuple {
222            values: vec![Value::text("abc"), Value::Null, Value::Unchanged],
223        };
224        assert_eq!(t.byte_len(), 3 + 3 * 8);
225        let op = Op::Insert {
226            relation: 1,
227            new: t.clone(),
228        };
229        assert_eq!(op.byte_len(), 32 + t.byte_len());
230        assert_eq!(SourceEvent::Op(op).byte_len(), 32 + t.byte_len());
231        assert_eq!(SourceEvent::Keepalive { wal_end: Lsn(1) }.byte_len(), 32);
232    }
233
234    #[test]
235    fn epoch_conversion() {
236        // 2000-01-01T00:00:00Z == 946684800 unix seconds.
237        assert_eq!(pg_micros_to_unix_micros(0), 946_684_800_000_000);
238        assert!(now_pg_micros() > 0);
239    }
240
241    #[test]
242    fn replica_identity_chars() {
243        assert_eq!(
244            ReplicaIdentity::from_char(b'd'),
245            Some(ReplicaIdentity::Default)
246        );
247        assert_eq!(
248            ReplicaIdentity::from_char(b'f'),
249            Some(ReplicaIdentity::Full)
250        );
251        assert_eq!(ReplicaIdentity::from_char(b'x'), None);
252    }
253}
pgsync_core/event.rs

pgsync_core/
event.rs
