laminar_connectors/postgres/cdc/
mod.rs

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

laminar_connectors/postgres/cdc/
mod.rs
