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

laminar_connectors/postgres/
mod.rs
