laminar_connectors/postgres/cdc/schema/
mod.rs

1//! `PostgreSQL` relation layout announced by `pgoutput` Relation messages.
2
3use crate::error::ConnectorError;
4
5use super::types::PgColumn;
6
7/// Layout of a `PostgreSQL` relation (table).
8#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
9pub struct RelationInfo {
10    /// The relation OID from `pgoutput`.
11    pub relation_id: u32,
12
13    /// Schema (namespace) name.
14    pub namespace: String,
15
16    /// Table name.
17    pub name: String,
18
19    /// Replica identity setting: 'd' (default), 'n' (nothing),
20    /// 'f' (full), 'i' (index).
21    pub replica_identity: char,
22
23    /// Column descriptors in ordinal order.
24    pub columns: Vec<PgColumn>,
25}
26
27impl RelationInfo {
28    /// Heap bytes retained by this layout.
29    pub(crate) fn retained_bytes(&self) -> Result<usize, ConnectorError> {
30        let overflow = || ConnectorError::ReadError("PostgreSQL CDC relation size overflow".into());
31        let mut bytes = self
32            .columns
33            .capacity()
34            .checked_mul(std::mem::size_of::<PgColumn>())
35            .and_then(|bytes| bytes.checked_add(self.namespace.capacity()))
36            .and_then(|bytes| bytes.checked_add(self.name.capacity()))
37            .ok_or_else(overflow)?;
38        for column in &self.columns {
39            bytes = bytes
40                .checked_add(column.name.capacity())
41                .ok_or_else(overflow)?;
42        }
43        Ok(bytes)
44    }
45}
46
47#[cfg(test)]
48mod tests {
49    use super::*;
50    use crate::postgres::cdc::types::{INT8_OID, TEXT_OID};
51
52    #[test]
53    fn layout_reports_retained_bytes() {
54        let relation = RelationInfo {
55            relation_id: 16_384,
56            namespace: "public".into(),
57            name: "users".into(),
58            replica_identity: 'f',
59            columns: vec![
60                PgColumn::new("id".into(), INT8_OID, -1, true),
61                PgColumn::new("name".into(), TEXT_OID, -1, true),
62            ],
63        };
64        assert!(
65            relation.retained_bytes().unwrap()
66                >= 2 * std::mem::size_of::<PgColumn>() + "publicusersidname".len()
67        );
68    }
69}
laminar_connectors/postgres/cdc/schema/mod.rs

laminar_connectors/postgres/cdc/schema/
mod.rs
