laminar_connectors/postgres/cdc/
types.rs

1//! `PostgreSQL` type OID to Arrow `DataType` mapping.
2//!
3//! Maps `PostgreSQL`'s internal type OIDs to Apache Arrow data types
4//! for zero-copy CDC event conversion. Also provides text-format
5//! value parsing for `pgoutput` protocol data.
6
7use arrow_schema::DataType;
8
9// ── Well-known PostgreSQL type OIDs ──
10
11/// `bool` — boolean
12pub const BOOL_OID: u32 = 16;
13/// `bytea` — variable-length binary string
14pub const BYTEA_OID: u32 = 17;
15/// `char` — single character (internal type)
16pub const CHAR_OID: u32 = 18;
17/// `int8` (bigint) — 8-byte signed integer
18pub const INT8_OID: u32 = 20;
19/// `int2` (smallint) — 2-byte signed integer
20pub const INT2_OID: u32 = 21;
21/// `int4` (integer) — 4-byte signed integer
22pub const INT4_OID: u32 = 23;
23/// `text` — variable-length text
24pub const TEXT_OID: u32 = 25;
25/// `oid` — object identifier (unsigned 4 bytes)
26pub const OID_OID: u32 = 26;
27/// `float4` (real) — single precision floating-point
28pub const FLOAT4_OID: u32 = 700;
29/// `float8` (double precision) — double precision floating-point
30pub const FLOAT8_OID: u32 = 701;
31/// `varchar` — variable-length character string
32pub const VARCHAR_OID: u32 = 1043;
33/// `date` — calendar date
34pub const DATE_OID: u32 = 1082;
35/// `time` — time of day (without timezone)
36pub const TIME_OID: u32 = 1083;
37/// `timestamp` — date and time (without timezone)
38pub const TIMESTAMP_OID: u32 = 1114;
39/// `timestamptz` — date and time with timezone
40pub const TIMESTAMPTZ_OID: u32 = 1184;
41/// `interval` — time interval
42pub const INTERVAL_OID: u32 = 1186;
43/// `numeric` — exact numeric with arbitrary precision
44pub const NUMERIC_OID: u32 = 1700;
45/// `uuid` — universally unique identifier
46pub const UUID_OID: u32 = 2950;
47/// `json` — JSON data
48pub const JSON_OID: u32 = 114;
49/// `jsonb` — binary JSON data
50pub const JSONB_OID: u32 = 3802;
51/// `xml` — XML data
52pub const XML_OID: u32 = 142;
53/// `inet` — IPv4/IPv6 host address
54pub const INET_OID: u32 = 869;
55/// `cidr` — IPv4/IPv6 network address
56pub const CIDR_OID: u32 = 650;
57/// `macaddr` — MAC address
58pub const MACADDR_OID: u32 = 829;
59/// `bpchar` — fixed-length character (char(n))
60pub const BPCHAR_OID: u32 = 1042;
61/// `name` — 63-byte internal name type
62pub const NAME_OID: u32 = 19;
63
64// ── Array type OIDs ──
65
66/// `bool[]`
67pub const BOOL_ARRAY_OID: u32 = 1000;
68/// `int2[]`
69pub const INT2_ARRAY_OID: u32 = 1005;
70/// `int4[]`
71pub const INT4_ARRAY_OID: u32 = 1007;
72/// `int8[]`
73pub const INT8_ARRAY_OID: u32 = 1016;
74/// `float4[]`
75pub const FLOAT4_ARRAY_OID: u32 = 1021;
76/// `float8[]`
77pub const FLOAT8_ARRAY_OID: u32 = 1022;
78/// `text[]`
79pub const TEXT_ARRAY_OID: u32 = 1009;
80/// `varchar[]`
81pub const VARCHAR_ARRAY_OID: u32 = 1015;
82
83/// A column descriptor from a `PostgreSQL` relation.
84#[derive(Debug, Clone, PartialEq, Eq)]
85pub struct PgColumn {
86    /// Column name.
87    pub name: String,
88
89    /// `PostgreSQL` type OID.
90    pub type_oid: u32,
91
92    /// Type modifier (e.g., precision for numeric, length for varchar).
93    /// -1 means no modifier.
94    pub type_modifier: i32,
95
96    /// Whether this column is part of the replica identity key.
97    pub is_key: bool,
98}
99
100impl PgColumn {
101    /// Creates a new column descriptor.
102    #[must_use]
103    pub fn new(name: String, type_oid: u32, type_modifier: i32, is_key: bool) -> Self {
104        Self {
105            name,
106            type_oid,
107            type_modifier,
108            is_key,
109        }
110    }
111
112    /// Returns the Arrow `DataType` for this column.
113    #[must_use]
114    pub fn arrow_type(&self) -> DataType {
115        pg_type_to_arrow(self.type_oid)
116    }
117}
118
119/// Maps a `PostgreSQL` type OID to an Arrow `DataType`.
120///
121/// Covers the most common `PostgreSQL` types. Unknown OIDs map to
122/// `DataType::Utf8` as a safe fallback (text representation).
123#[must_use]
124pub fn pg_type_to_arrow(oid: u32) -> DataType {
125    match oid {
126        // Boolean
127        BOOL_OID => DataType::Boolean,
128
129        // Integer types
130        INT2_OID => DataType::Int16,
131        INT4_OID | OID_OID => DataType::Int32,
132        INT8_OID => DataType::Int64,
133
134        // Floating point
135        FLOAT4_OID => DataType::Float32,
136        FLOAT8_OID => DataType::Float64,
137
138        // Binary
139        BYTEA_OID => DataType::Binary,
140
141        // Date/Time
142        DATE_OID => DataType::Date32,
143        TIME_OID => DataType::Time64(arrow_schema::TimeUnit::Microsecond),
144        TIMESTAMP_OID => DataType::Timestamp(arrow_schema::TimeUnit::Microsecond, None),
145        TIMESTAMPTZ_OID => {
146            DataType::Timestamp(arrow_schema::TimeUnit::Microsecond, Some("UTC".into()))
147        }
148        // Types without a lossless Arrow scalar mapping use their text representation.
149        _ => DataType::Utf8,
150    }
151}
152
153/// Returns a human-readable name for a `PostgreSQL` type OID.
154#[must_use]
155pub fn pg_type_name(oid: u32) -> &'static str {
156    match oid {
157        BOOL_OID => "bool",
158        BYTEA_OID => "bytea",
159        CHAR_OID => "char",
160        INT2_OID => "int2",
161        INT4_OID => "int4",
162        INT8_OID => "int8",
163        TEXT_OID => "text",
164        OID_OID => "oid",
165        FLOAT4_OID => "float4",
166        FLOAT8_OID => "float8",
167        VARCHAR_OID => "varchar",
168        BPCHAR_OID => "bpchar",
169        NAME_OID => "name",
170        DATE_OID => "date",
171        TIME_OID => "time",
172        TIMESTAMP_OID => "timestamp",
173        TIMESTAMPTZ_OID => "timestamptz",
174        INTERVAL_OID => "interval",
175        NUMERIC_OID => "numeric",
176        UUID_OID => "uuid",
177        JSON_OID => "json",
178        JSONB_OID => "jsonb",
179        XML_OID => "xml",
180        INET_OID => "inet",
181        CIDR_OID => "cidr",
182        MACADDR_OID => "macaddr",
183        _ => "unknown",
184    }
185}
186
187#[cfg(test)]
188mod tests {
189    use super::*;
190
191    #[test]
192    fn test_integer_type_mapping() {
193        assert_eq!(pg_type_to_arrow(INT2_OID), DataType::Int16);
194        assert_eq!(pg_type_to_arrow(INT4_OID), DataType::Int32);
195        assert_eq!(pg_type_to_arrow(INT8_OID), DataType::Int64);
196    }
197
198    #[test]
199    fn test_float_type_mapping() {
200        assert_eq!(pg_type_to_arrow(FLOAT4_OID), DataType::Float32);
201        assert_eq!(pg_type_to_arrow(FLOAT8_OID), DataType::Float64);
202    }
203
204    #[test]
205    fn test_text_type_mapping() {
206        assert_eq!(pg_type_to_arrow(TEXT_OID), DataType::Utf8);
207        assert_eq!(pg_type_to_arrow(VARCHAR_OID), DataType::Utf8);
208        assert_eq!(pg_type_to_arrow(BPCHAR_OID), DataType::Utf8);
209        assert_eq!(pg_type_to_arrow(NAME_OID), DataType::Utf8);
210    }
211
212    #[test]
213    fn test_bool_type_mapping() {
214        assert_eq!(pg_type_to_arrow(BOOL_OID), DataType::Boolean);
215    }
216
217    #[test]
218    fn test_timestamp_type_mapping() {
219        assert!(matches!(
220            pg_type_to_arrow(TIMESTAMP_OID),
221            DataType::Timestamp(arrow_schema::TimeUnit::Microsecond, None)
222        ));
223        assert!(matches!(
224            pg_type_to_arrow(TIMESTAMPTZ_OID),
225            DataType::Timestamp(arrow_schema::TimeUnit::Microsecond, Some(_))
226        ));
227    }
228
229    #[test]
230    fn test_date_time_mapping() {
231        assert_eq!(pg_type_to_arrow(DATE_OID), DataType::Date32);
232        assert!(matches!(
233            pg_type_to_arrow(TIME_OID),
234            DataType::Time64(arrow_schema::TimeUnit::Microsecond)
235        ));
236    }
237
238    #[test]
239    fn test_binary_type_mapping() {
240        assert_eq!(pg_type_to_arrow(BYTEA_OID), DataType::Binary);
241    }
242
243    #[test]
244    fn test_json_type_mapping() {
245        assert_eq!(pg_type_to_arrow(JSON_OID), DataType::Utf8);
246        assert_eq!(pg_type_to_arrow(JSONB_OID), DataType::Utf8);
247    }
248
249    #[test]
250    fn test_unknown_type_fallback() {
251        assert_eq!(pg_type_to_arrow(99999), DataType::Utf8);
252    }
253
254    #[test]
255    fn test_pg_column() {
256        let col = PgColumn::new("id".to_string(), INT8_OID, -1, true);
257        assert_eq!(col.name, "id");
258        assert_eq!(col.type_oid, INT8_OID);
259        assert!(col.is_key);
260        assert_eq!(col.arrow_type(), DataType::Int64);
261    }
262
263    #[test]
264    fn test_pg_type_name() {
265        assert_eq!(pg_type_name(INT4_OID), "int4");
266        assert_eq!(pg_type_name(TEXT_OID), "text");
267        assert_eq!(pg_type_name(BOOL_OID), "bool");
268        assert_eq!(pg_type_name(99999), "unknown");
269    }
270
271    #[test]
272    fn test_numeric_maps_to_utf8() {
273        // Numeric must be Utf8 to preserve arbitrary precision
274        assert_eq!(pg_type_to_arrow(NUMERIC_OID), DataType::Utf8);
275    }
276
277    #[test]
278    fn test_array_types_map_to_utf8() {
279        assert_eq!(pg_type_to_arrow(INT4_ARRAY_OID), DataType::Utf8);
280        assert_eq!(pg_type_to_arrow(TEXT_ARRAY_OID), DataType::Utf8);
281    }
282}
laminar_connectors/postgres/cdc/types.rs

laminar_connectors/postgres/cdc/
types.rs
