laminar_connectors/postgres/cdc/config/
mod.rs

1//! `PostgreSQL` CDC source connector configuration.
2//!
3//! Provides [`PostgresCdcConfig`] with all settings needed to connect to
4//! a `PostgreSQL` database and stream logical replication changes.
5
6use std::path::PathBuf;
7
8use crate::config::ConnectorConfig;
9use crate::error::ConnectorError;
10use crate::postgres::SslMode;
11
12const REMOVED_CONFIG_KEYS: &[&str] = &[
13    "backpressure.high.watermark",
14    "keepalive.interval.ms",
15    "max.buffered.events",
16    "max.poll.records",
17    "poll.timeout.ms",
18    "start.lsn",
19    "table.exclude",
20    "table.include",
21    "wal.sender.timeout.ms",
22];
23
24/// A slot name is at most 63 bytes; the claim suffix `_<16 hex digits>` takes 17 of them.
25const MAX_SLOT_PREFIX_BYTES: usize = 46;
26const DEFAULT_BUFFERED_BYTES: usize = 256 * 1024 * 1024;
27const MIN_BUFFERED_BYTES: usize = 1024 * 1024;
28const MAX_BUFFERED_BYTES: usize = 4 * 1024 * 1024 * 1024;
29const WORKING_SET_STAGES: usize = 3;
30/// Relation metadata may use at most this fraction of the decoded stage, leaving the rest for
31/// drainable transaction data.
32const RELATION_METADATA_DIVISOR: usize = 8;
33
34const ALLOWED_CONFIG_KEYS: &[&str] = &[
35    "_arrow_schema",
36    "_primary_key_columns",
37    "database",
38    "host",
39    "laminar.source.name",
40    "max.buffered.bytes",
41    "output.mode",
42    "password",
43    "port",
44    "publication",
45    "slot.name",
46    "snapshot.mode",
47    "ssl.ca.cert.path",
48    "ssl.mode",
49    "table",
50    "username",
51];
52
53/// How a source without a checkpoint establishes its starting position.
54#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
55pub enum SnapshotMode {
56    /// Copy the table at the new slot's consistent point, then stream every later change.
57    #[default]
58    Initial,
59    /// Stream only changes committed after the new slot's consistent point.
60    Never,
61}
62
63str_enum!(SnapshotMode, lowercase, ConnectorError, "unknown snapshot.mode",
64    Initial => "initial";
65    Never => "never"
66);
67
68/// Update model of the emitted rows.
69#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
70pub enum OutputMode {
71    /// Keyed current-row images and key-only deletes by the declared primary key.
72    #[default]
73    Upsert,
74    /// Full before/after images as a Z-set with a trailing `__weight` column.
75    Changelog,
76}
77
78str_enum!(OutputMode, lowercase, ConnectorError, "unknown output.mode",
79    Upsert => "upsert";
80    Changelog => "changelog"
81);
82
83/// A schema-qualified `PostgreSQL` table name, compared exactly as stored in the catalog.
84#[derive(Debug, Clone, Default, PartialEq, Eq, PartialOrd, Ord)]
85pub struct TableName {
86    /// Schema (namespace) name.
87    pub schema: String,
88    /// Table name.
89    pub name: String,
90}
91
92impl TableName {
93    /// Parse `schema.table`.
94    ///
95    /// # Errors
96    /// Returns a configuration error unless the value has exactly one dot separating two
97    /// nonempty names.
98    pub fn parse(value: &str) -> Result<Self, ConnectorError> {
99        let invalid = || {
100            ConnectorError::ConfigurationError(format!(
101                "table '{value}' must be schema-qualified as schema.table"
102            ))
103        };
104        let (schema, name) = value.split_once('.').ok_or_else(invalid)?;
105        if schema.is_empty() || name.is_empty() || name.contains('.') || value.contains('\0') {
106            return Err(invalid());
107        }
108        Ok(Self {
109            schema: schema.to_string(),
110            name: name.to_string(),
111        })
112    }
113}
114
115impl std::fmt::Display for TableName {
116    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
117        write!(f, "{}.{}", self.schema, self.name)
118    }
119}
120
121/// Configuration for the `PostgreSQL` CDC source connector.
122#[derive(Debug, Clone)]
123pub struct PostgresCdcConfig {
124    // ── Connection ──
125    /// `PostgreSQL` host address.
126    pub host: String,
127
128    /// `PostgreSQL` port.
129    pub port: u16,
130
131    /// Database name.
132    pub database: String,
133
134    /// Username for authentication.
135    pub username: String,
136
137    /// Password for authentication.
138    pub password: Option<String>,
139
140    /// SSL mode for the connection.
141    pub ssl_mode: SslMode,
142
143    /// Optional PEM file containing trusted CA certificates.
144    pub ssl_ca_cert_path: Option<PathBuf>,
145
146    // ── Replication ──
147    /// Prefix of the logical replication slots the source creates: each claim creates
148    /// `<slot_name>_<16 hex digits>` once a checkpoint commits the claim.
149    pub slot_name: String,
150
151    /// Name of the publication to subscribe to.
152    pub publication: String,
153
154    /// The one captured table; the publication must contain exactly this table.
155    pub table: TableName,
156
157    /// Fresh-start behavior when no checkpoint exists.
158    pub snapshot_mode: SnapshotMode,
159
160    /// Emitted update model.
161    pub output_mode: OutputMode,
162
163    /// Total connector-owned payload budget across raw WAL, decoded state, and Arrow construction.
164    pub max_buffered_bytes: usize,
165}
166
167impl Default for PostgresCdcConfig {
168    fn default() -> Self {
169        Self {
170            host: "localhost".to_string(),
171            port: 5432,
172            database: "postgres".to_string(),
173            username: "postgres".to_string(),
174            password: None,
175            ssl_mode: SslMode::VerifyFull,
176            ssl_ca_cert_path: None,
177            slot_name: "laminar_slot".to_string(),
178            publication: "laminar_pub".to_string(),
179            table: TableName::default(),
180            snapshot_mode: SnapshotMode::default(),
181            output_mode: OutputMode::default(),
182            max_buffered_bytes: DEFAULT_BUFFERED_BYTES,
183        }
184    }
185}
186
187impl PostgresCdcConfig {
188    /// Raw pgwire/frame ownership share of the private working-set limit.
189    #[must_use]
190    pub(crate) fn raw_wal_bytes(&self) -> usize {
191        self.max_buffered_bytes / WORKING_SET_STAGES
192    }
193
194    /// Decoded transaction ownership share of the private working-set limit.
195    #[must_use]
196    pub(crate) fn decoded_event_bytes(&self) -> usize {
197        self.max_buffered_bytes / WORKING_SET_STAGES
198    }
199
200    /// Persistent relation metadata limit, carved out of the decoded stage.
201    #[must_use]
202    pub(crate) fn relation_metadata_bytes(&self) -> usize {
203        self.decoded_event_bytes() / RELATION_METADATA_DIVISOR
204    }
205
206    /// Arrow construction share, including division remainder.
207    #[must_use]
208    pub(crate) fn arrow_build_bytes(&self) -> usize {
209        self.max_buffered_bytes
210            .saturating_sub(self.raw_wal_bytes())
211            .saturating_sub(self.decoded_event_bytes())
212    }
213
214    /// Creates a new config with required fields.
215    #[must_use]
216    pub fn new(host: &str, database: &str, slot_name: &str, publication: &str) -> Self {
217        Self {
218            host: host.to_string(),
219            database: database.to_string(),
220            slot_name: slot_name.to_string(),
221            publication: publication.to_string(),
222            ..Self::default()
223        }
224    }
225
226    /// Builds the typed control-plane connection configuration.
227    ///
228    /// Using setters keeps credentials and database names out of libpq-style
229    /// tokenization, so whitespace, quotes, and backslashes are data rather
230    /// than connection-string syntax.
231    pub(super) fn control_connection_config(
232        &self,
233        application_name: &str,
234    ) -> Result<tokio_postgres::Config, ConnectorError> {
235        self.validate_connection()?;
236
237        let mut config = tokio_postgres::Config::new();
238        config
239            .host(&self.host)
240            .port(self.port)
241            .dbname(&self.database)
242            .user(&self.username)
243            .application_name(application_name)
244            .options(super::typed_rows::SESSION_OPTIONS)
245            .ssl_mode(match self.ssl_mode {
246                SslMode::Disable => tokio_postgres::config::SslMode::Disable,
247                SslMode::VerifyFull => tokio_postgres::config::SslMode::Require,
248            })
249            .connect_timeout(super::postgres_io::CONNECT_TIMEOUT);
250        if let Some(password) = &self.password {
251            config.password(password);
252        }
253        Ok(config)
254    }
255
256    /// Parses configuration from a generic [`ConnectorConfig`].
257    ///
258    /// # Errors
259    ///
260    /// Returns `ConnectorError` if required keys are missing or values are
261    /// invalid.
262    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
263        Self::reject_removed_keys(config)?;
264        config.reject_unknown_properties(ALLOWED_CONFIG_KEYS, "PostgreSQL CDC")?;
265
266        let mut cfg = Self {
267            host: config.require("host")?.to_string(),
268            database: config.require("database")?.to_string(),
269            slot_name: config.require("slot.name")?.to_string(),
270            publication: config.require("publication")?.to_string(),
271            table: TableName::parse(config.require("table")?)?,
272            ssl_mode: config
273                .get_parsed::<SslMode>("ssl.mode")?
274                .unwrap_or_default(),
275            snapshot_mode: config
276                .get_parsed::<SnapshotMode>("snapshot.mode")?
277                .unwrap_or_default(),
278            output_mode: config
279                .get_parsed::<OutputMode>("output.mode")?
280                .unwrap_or_default(),
281            ..Self::default()
282        };
283
284        if let Some(port) = config.get("port") {
285            cfg.port = crate::config::parse_port(port)?;
286        }
287        if let Some(user) = config.get("username") {
288            cfg.username = user.to_string();
289        }
290        cfg.password = config.get("password").map(String::from);
291        cfg.ssl_ca_cert_path = config.get("ssl.ca.cert.path").map(PathBuf::from);
292        if let Some(max) = config.get_parsed::<usize>("max.buffered.bytes")? {
293            cfg.max_buffered_bytes = max;
294        }
295        cfg.validate()?;
296        Ok(cfg)
297    }
298
299    fn reject_removed_keys(config: &ConnectorConfig) -> Result<(), ConnectorError> {
300        if let Some(key) = REMOVED_CONFIG_KEYS
301            .iter()
302            .find(|key| config.get(key).is_some())
303        {
304            let reason = match *key {
305                "start.lsn" => {
306                    "recovery cursors are owned by a validated engine checkpoint or the durable slot"
307                }
308                "max.buffered.events" => {
309                    "resource ownership is bounded by max.buffered.bytes instead"
310                }
311                "table.include" | "table.exclude" => {
312                    "the source captures exactly one table; set 'table' to schema.table"
313                }
314                _ => "the connector did not execute it",
315            };
316            return Err(ConnectorError::ConfigurationError(format!(
317                "PostgreSQL CDC property '{key}' is not supported: {reason}"
318            )));
319        }
320        Ok(())
321    }
322
323    /// Validates the configuration.
324    ///
325    /// # Errors
326    ///
327    /// Returns `ConnectorError::ConfigurationError` for invalid settings.
328    pub fn validate(&self) -> Result<(), ConnectorError> {
329        self.validate_connection()?;
330        if self.table.schema.is_empty() || self.table.name.is_empty() {
331            return Err(ConnectorError::ConfigurationError(
332                "table must name the one captured table as schema.table".into(),
333            ));
334        }
335        if self.output_mode == OutputMode::Changelog && self.snapshot_mode == SnapshotMode::Never {
336            return Err(ConnectorError::ConfigurationError(
337                "output.mode=changelog requires snapshot.mode=initial: a retraction of a row \
338                 that was never emitted would corrupt downstream state"
339                    .into(),
340            ));
341        }
342        Ok(())
343    }
344
345    fn validate_connection(&self) -> Result<(), ConnectorError> {
346        crate::config::require_non_empty(&self.host, "host")?;
347        crate::config::require_non_empty(&self.database, "database")?;
348        crate::config::require_non_empty(&self.username, "username")?;
349        crate::config::require_non_empty(&self.slot_name, "slot.name")?;
350        crate::config::require_non_empty(&self.publication, "publication")?;
351        for (value, label) in [
352            (&self.host, "host"),
353            (&self.database, "database"),
354            (&self.username, "username"),
355            (&self.slot_name, "slot.name"),
356            (&self.publication, "publication"),
357        ] {
358            if value.contains('\0') {
359                return Err(ConnectorError::ConfigurationError(format!(
360                    "{label} must not contain NUL"
361                )));
362            }
363        }
364        if self.slot_name.len() > MAX_SLOT_PREFIX_BYTES
365            || !self
366                .slot_name
367                .bytes()
368                .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_')
369        {
370            return Err(ConnectorError::ConfigurationError(format!(
371                "slot.name is the prefix of the slots the source creates (<slot.name>_<16 hex \
372                 digits>): at most {MAX_SLOT_PREFIX_BYTES} bytes of lower-case ASCII letters, \
373                 digits, or underscore"
374            )));
375        }
376        if self.publication.len() > 63
377            || !self
378                .publication
379                .bytes()
380                .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_')
381            || self.publication.as_bytes()[0].is_ascii_digit()
382        {
383            return Err(ConnectorError::ConfigurationError(
384                "publication must start with a lower-case ASCII letter or underscore and contain only lower-case ASCII letters, digits, or underscore (63 bytes maximum)"
385                    .into(),
386            ));
387        }
388        if self
389            .password
390            .as_deref()
391            .is_some_and(|value| value.contains('\0'))
392        {
393            return Err(ConnectorError::ConfigurationError(
394                "password must not contain NUL".into(),
395            ));
396        }
397        if self.port == 0 {
398            return Err(ConnectorError::ConfigurationError(
399                "port must be > 0".to_string(),
400            ));
401        }
402        if !(MIN_BUFFERED_BYTES..=MAX_BUFFERED_BYTES).contains(&self.max_buffered_bytes) {
403            return Err(ConnectorError::ConfigurationError(format!(
404                "max.buffered.bytes must be between {MIN_BUFFERED_BYTES} and {MAX_BUFFERED_BYTES}"
405            )));
406        }
407        if self
408            .ssl_ca_cert_path
409            .as_ref()
410            .is_some_and(|path| path.as_os_str().is_empty())
411        {
412            return Err(ConnectorError::ConfigurationError(
413                "ssl.ca.cert.path must not be empty".to_string(),
414            ));
415        }
416        if self.ssl_mode == SslMode::Disable && self.ssl_ca_cert_path.is_some() {
417            return Err(ConnectorError::ConfigurationError(
418                "ssl.ca.cert.path requires ssl.mode=verify-full".to_string(),
419            ));
420        }
421        Ok(())
422    }
423}
424
425#[cfg(test)]
426mod tests;
laminar_connectors/postgres/cdc/config/mod.rs

laminar_connectors/postgres/cdc/config/
mod.rs
