laminar_connectors/postgres/cdc/
mod.rs

1//! `PostgreSQL` CDC source connector.
2
3pub mod changelog;
4pub mod config;
5pub mod decoder;
6pub mod lsn;
7pub mod metrics;
8pub mod postgres_io;
9pub mod schema;
10pub mod source;
11pub mod types;
12
13// Re-export primary types at module level.
14pub use crate::postgres::SslMode;
15pub use config::PostgresCdcConfig;
16pub use lsn::Lsn;
17pub use source::PostgresCdcSource;
18
19use std::sync::Arc;
20
21use crate::config::{ConfigKeySpec, ConnectorInfo};
22use crate::registry::ConnectorRegistry;
23
24/// Registers the `PostgreSQL` CDC source connector with the given registry.
25///
26/// # Errors
27///
28/// Returns an error if the connector name is already registered or the registry is frozen.
29pub fn register_postgres_cdc_source(
30    registry: &ConnectorRegistry,
31) -> Result<(), crate::error::ConnectorError> {
32    let info = ConnectorInfo {
33        name: "postgres-cdc".to_string(),
34        display_name: "PostgreSQL CDC Source".to_string(),
35        version: env!("CARGO_PKG_VERSION").to_string(),
36        is_source: true,
37        is_sink: false,
38        config_keys: postgres_cdc_config_keys(),
39    };
40
41    registry.register_source(
42        "postgres-cdc",
43        info,
44        Arc::new(|registry: Option<&Arc<prometheus::Registry>>| {
45            Ok(Box::new(PostgresCdcSource::new(
46                PostgresCdcConfig::default(),
47                registry.map(Arc::as_ref),
48            )))
49        }),
50    )?;
51
52    // Register standalone finite snapshots (no replication slot required).
53    let pg_info = ConnectorInfo {
54        name: "postgres".to_string(),
55        display_name: "PostgreSQL Lookup Source".to_string(),
56        version: env!("CARGO_PKG_VERSION").to_string(),
57        is_source: true,
58        is_sink: false,
59        config_keys: postgres_lookup_config_keys(),
60    };
61    registry.register_table_source(
62        "postgres",
63        pg_info.clone(),
64        Arc::new(|config, declared_schema| {
65            Ok(Box::new(
66                crate::postgres::reference::PostgresReferenceTableSource::new(
67                    config.clone(),
68                    declared_schema,
69                ),
70            ))
71        }),
72    )?;
73
74    // On-demand (partial cache mode) lookup source: pooled + WHERE pk = ANY($1).
75    registry.register_lookup_source("postgres", pg_info, Arc::new(PostgresLookupFactory))
76}
77
78fn postgres_lookup_config_keys() -> Vec<ConfigKeySpec> {
79    vec![
80        ConfigKeySpec::required("table", "Qualified PostgreSQL table name"),
81        ConfigKeySpec::optional(
82            "connection",
83            "libpq connection string (alternative to individual connection properties)",
84            "",
85        ),
86        ConfigKeySpec::optional("connection_string", "Alias for connection", ""),
87        ConfigKeySpec::optional("host", "PostgreSQL host", "localhost"),
88        ConfigKeySpec::optional("port", "PostgreSQL port", "5432"),
89        ConfigKeySpec::optional("database", "Database name (alternative to dbname)", ""),
90        ConfigKeySpec::optional("dbname", "Alias for database", ""),
91        ConfigKeySpec::optional("user", "Database user (alternative to username)", ""),
92        ConfigKeySpec::optional("username", "Alias for user", ""),
93        ConfigKeySpec::optional("password", "Database password", ""),
94        ConfigKeySpec::optional("options", "PostgreSQL command-line options", ""),
95        ConfigKeySpec::optional("pool_size", "On-demand lookup connection-pool size", "4"),
96        ConfigKeySpec::optional(
97            "ssl.mode",
98            "Connection security: verify-full or explicit disable",
99            "verify-full",
100        ),
101        ConfigKeySpec::optional(
102            "ssl.ca.cert.path",
103            "PEM file with trusted CA certificates; defaults to webpki roots",
104            "",
105        ),
106    ]
107}
108
109struct PostgresLookupFactory;
110
111#[async_trait::async_trait]
112impl crate::registry::LookupSourceFactory for PostgresLookupFactory {
113    async fn build(
114        &self,
115        config: crate::config::ConnectorConfig,
116        _declared_schema: Option<arrow_schema::SchemaRef>,
117    ) -> Result<Arc<dyn laminar_core::lookup::source::LookupSourceDyn>, crate::error::ConnectorError>
118    {
119        use crate::postgres::lookup::{PostgresLookupSource, PostgresLookupSourceConfig};
120
121        let pk_columns: Vec<String> = config
122            .get("_primary_key_columns")
123            .unwrap_or("")
124            .split(',')
125            .map(|s| s.trim().to_string())
126            .filter(|s| !s.is_empty())
127            .collect();
128        if pk_columns.is_empty() {
129            return Err(crate::error::ConnectorError::ConfigurationError(
130                "postgres lookup source requires primary key columns".into(),
131            ));
132        }
133
134        let table = config
135            .get("table")
136            .ok_or_else(|| {
137                crate::error::ConnectorError::ConfigurationError(
138                    "postgres lookup source requires a 'table' property".into(),
139                )
140            })?
141            .to_string();
142
143        let pool_size = if let Some(s) = config.get("pool_size") {
144            s.parse::<usize>().map_err(|e| {
145                crate::error::ConnectorError::ConfigurationError(format!(
146                    "invalid 'pool_size' value '{s}': {e}"
147                ))
148            })?
149        } else {
150            4
151        };
152
153        let lookup_config = PostgresLookupSourceConfig {
154            properties: config.properties().clone(),
155            table,
156            primary_key_columns: pk_columns,
157            pool_size,
158        };
159
160        let source = PostgresLookupSource::open(lookup_config).await?;
161        Ok(Arc::new(source) as Arc<dyn laminar_core::lookup::source::LookupSourceDyn>)
162    }
163}
164
165fn postgres_cdc_config_keys() -> Vec<ConfigKeySpec> {
166    vec![
167        ConfigKeySpec::required("host", "PostgreSQL host address"),
168        ConfigKeySpec::required("database", "Database name"),
169        ConfigKeySpec::required("slot.name", "Logical replication slot name"),
170        ConfigKeySpec::required("publication", "Publication name"),
171        ConfigKeySpec::optional("port", "PostgreSQL port", "5432"),
172        ConfigKeySpec::optional("username", "Connection username", "postgres"),
173        ConfigKeySpec::optional("password", "Connection password", ""),
174        ConfigKeySpec::optional(
175            "ssl.mode",
176            "Connection security: verify-full or explicit disable",
177            "verify-full",
178        ),
179        ConfigKeySpec::optional(
180            "ssl.ca.cert.path",
181            "PEM file with trusted CA certificates; defaults to webpki roots",
182            "",
183        ),
184        ConfigKeySpec::optional(
185            "table.include",
186            "Comma-separated schema-qualified tables to include",
187            "",
188        ),
189        ConfigKeySpec::optional(
190            "table.exclude",
191            "Comma-separated schema-qualified tables to exclude",
192            "",
193        ),
194        ConfigKeySpec::optional(
195            "max.buffered.bytes",
196            "Total connector-owned payload budget in bytes",
197            "268435456",
198        ),
199    ]
200}
laminar_connectors/postgres/cdc/mod.rs

laminar_connectors/postgres/cdc/
mod.rs
