laminar_connectors/postgres/
mod.rs

1//! PostgreSQL connector-specific configuration and implementations.
2
3#[cfg(feature = "postgres-cdc")]
4pub mod cdc;
5#[cfg(feature = "postgres-cdc")]
6pub mod lookup;
7#[cfg(feature = "postgres-cdc")]
8pub mod reference;
9#[cfg(any(feature = "postgres-cdc", feature = "postgres-sink"))]
10mod schema_metadata;
11#[cfg(feature = "postgres-sink")]
12pub mod sink;
13#[cfg(feature = "postgres-sink")]
14pub mod sink_config;
15#[cfg(feature = "postgres-sink")]
16pub mod sink_metrics;
17mod tls;
18#[cfg(feature = "postgres-sink")]
19pub mod types;
20
21pub(crate) use tls::make_rustls_connector;
22/// `PostgreSQL` connection security policy.
23pub use tls::SslMode;
24
25#[cfg(feature = "postgres-cdc")]
26pub use cdc::{register_postgres_cdc_source, Lsn, PostgresCdcConfig, PostgresCdcSource};
27#[cfg(feature = "postgres-cdc")]
28pub use lookup::{PostgresLookupSource, PostgresLookupSourceConfig};
29#[cfg(feature = "postgres-cdc")]
30pub use reference::PostgresReferenceTableSource;
31
32// Re-export primary sink types at module level.
33#[cfg(feature = "postgres-sink")]
34pub use sink::PostgresSink;
35#[cfg(feature = "postgres-sink")]
36pub use sink_config::{PostgresSinkConfig, WriteMode};
37#[cfg(feature = "postgres-sink")]
38pub use sink_metrics::PostgresSinkMetrics;
39
40#[cfg(feature = "postgres-cdc")]
41use std::future::Future;
42#[cfg(feature = "postgres-sink")]
43use std::sync::Arc;
44
45#[cfg(feature = "postgres-sink")]
46use crate::config::{ConfigKeySpec, ConnectorInfo};
47#[cfg(feature = "postgres-sink")]
48use crate::registry::ConnectorRegistry;
49
50/// Poll a driver future from a task whose lifetime is independent of its caller.
51/// Dropping the waiter detaches the task; it does not cancel the driver operation.
52#[cfg(feature = "postgres-cdc")]
53async fn await_owned_driver<T, E>(
54    future: impl Future<Output = Result<T, E>> + Send + 'static,
55    join_error: impl FnOnce(tokio::task::JoinError) -> E + Send + 'static,
56) -> Result<T, E>
57where
58    T: Send + 'static,
59    E: Send + 'static,
60{
61    match tokio::spawn(future).await {
62        Ok(result) => result,
63        Err(error) => Err(join_error(error)),
64    }
65}
66
67/// Registers the `PostgreSQL` sink connector with the given registry.
68///
69/// # Errors
70///
71/// Returns an error if the connector name is already registered or the registry is frozen.
72#[cfg(feature = "postgres-sink")]
73pub fn register_postgres_sink(
74    registry: &ConnectorRegistry,
75) -> Result<(), crate::error::ConnectorError> {
76    let info = ConnectorInfo {
77        schema_capabilities: crate::schema::resolution::SchemaCapabilities::metadata(
78            &[],
79            true,
80            crate::schema::resolution::SchemaPreparation::ExplicitTableCreation,
81        ),
82        name: "postgres-sink".to_string(),
83        display_name: "PostgreSQL Sink".to_string(),
84        version: env!("CARGO_PKG_VERSION").to_string(),
85        is_source: false,
86        is_sink: true,
87        config_keys: postgres_sink_config_keys(),
88    };
89
90    registry.register_sink(
91        "postgres-sink",
92        info,
93        Arc::new(|config, registry: Option<&Arc<prometheus::Registry>>| {
94            Ok(Box::new(PostgresSink::from_connector_config(
95                config,
96                registry.map(Arc::as_ref),
97            )?))
98        }),
99    )
100}
101
102#[cfg(feature = "postgres-sink")]
103fn postgres_sink_config_keys() -> Vec<ConfigKeySpec> {
104    vec![
105        ConfigKeySpec::required("hostname", "PostgreSQL server hostname"),
106        ConfigKeySpec::required("database", "Target database name"),
107        ConfigKeySpec::required("username", "Authentication username"),
108        ConfigKeySpec::required("table.name", "Target table name"),
109        ConfigKeySpec::optional("password", "Authentication password", ""),
110        ConfigKeySpec::optional("port", "PostgreSQL port", "5432"),
111        ConfigKeySpec::optional("schema.name", "Target schema name", "public"),
112        ConfigKeySpec::optional(
113            "write.mode",
114            "Write mode: 'append' (COPY BINARY) or 'upsert' (ON CONFLICT)",
115            "append",
116        ),
117        ConfigKeySpec::optional(
118            "primary.key",
119            "Comma-separated primary key columns (required for upsert mode)",
120            "",
121        ),
122        ConfigKeySpec::optional("flush.interval.ms", "Max time before flush (ms)", "250"),
123        ConfigKeySpec::optional("connect.timeout.ms", "Connection timeout (ms)", "10000"),
124        ConfigKeySpec::optional("statement.timeout.ms", "Statement timeout (ms)", "30000"),
125        ConfigKeySpec::optional(
126            "ssl.mode",
127            "Connection security: verify-full or explicit disable",
128            "verify-full",
129        ),
130        ConfigKeySpec::optional(
131            "ssl.ca.cert.path",
132            "PEM file with trusted CA certificates; defaults to webpki roots",
133            "",
134        ),
135        ConfigKeySpec::optional(
136            "auto.create.table",
137            "Create target table from Arrow schema if missing",
138            "false",
139        ),
140        ConfigKeySpec::optional(
141            "changelog.mode",
142            "Handle Z-set records (split INSERT/DELETE by _op)",
143            "false",
144        ),
145    ]
146}
147
148#[cfg(all(test, feature = "postgres-cdc"))]
149mod driver_tests;
laminar_connectors/postgres/mod.rs

laminar_connectors/postgres/
mod.rs
