laminar_connectors/postgres/sink_config/
mod.rs

1//! `PostgreSQL` sink connector configuration.
2//!
3//! [`PostgresSinkConfig`] encapsulates all settings for writing Arrow
4//! `RecordBatch` data to `PostgreSQL`, parsed from resolved connector options.
5
6use std::path::PathBuf;
7use std::time::Duration;
8
9use crate::config::ConnectorConfig;
10use crate::error::ConnectorError;
11use crate::postgres::SslMode;
12
13const MAX_POSTGRES_IDENTIFIER_BYTES: usize = 63;
14const MAX_POSTGRES_STATEMENT_TIMEOUT_MS: u128 = 2_147_483_647;
15
16const REMOVED_CONFIG_KEYS: &[&str] = &["batch.size", "pool.size"];
17
18const ALLOWED_CONFIG_KEYS: &[&str] = &[
19    "_arrow_schema",
20    "auto.create.table",
21    "changelog.mode",
22    "connect.timeout.ms",
23    "database",
24    "delivery.guarantee",
25    "flush.interval.ms",
26    "hostname",
27    "password",
28    "port",
29    "primary.key",
30    "schema.name",
31    "ssl.ca.cert.path",
32    "ssl.mode",
33    "statement.timeout.ms",
34    "table.name",
35    "username",
36    "write.mode",
37];
38
39/// Configuration for the `PostgreSQL` sink connector.
40///
41/// Parsed from resolved sink connector options via [`from_config`](Self::from_config).
42#[derive(Debug, Clone)]
43pub struct PostgresSinkConfig {
44    /// `PostgreSQL` hostname.
45    pub hostname: String,
46
47    /// `PostgreSQL` port (default: 5432).
48    pub port: u16,
49
50    /// Database name.
51    pub database: String,
52
53    /// Username for authentication.
54    pub username: String,
55
56    /// Password for authentication.
57    pub password: String,
58
59    /// Target schema name (default: `"public"`).
60    pub schema_name: String,
61
62    /// Target table name.
63    pub table_name: String,
64
65    /// Write mode: append (COPY BINARY) or upsert (ON CONFLICT).
66    pub write_mode: WriteMode,
67
68    /// Primary key columns (required for upsert mode).
69    pub primary_key_columns: Vec<String>,
70
71    /// Maximum time to buffer before flushing.
72    pub flush_interval: Duration,
73
74    /// Connection timeout.
75    pub connect_timeout: Duration,
76
77    /// SSL mode for connections.
78    pub ssl_mode: SslMode,
79
80    /// Optional PEM file containing trusted CA certificates.
81    pub ssl_ca_cert_path: Option<PathBuf>,
82
83    /// Whether to create the target table if it doesn't exist.
84    pub auto_create_table: bool,
85
86    /// Whether to handle changelog/retraction records.
87    pub changelog_mode: bool,
88
89    /// Per-query statement timeout (default: 30s).
90    pub statement_timeout: Duration,
91}
92
93impl Default for PostgresSinkConfig {
94    fn default() -> Self {
95        Self {
96            hostname: "localhost".to_string(),
97            port: 5432,
98            database: String::new(),
99            username: "postgres".to_string(),
100            password: String::new(),
101            schema_name: "public".to_string(),
102            table_name: String::new(),
103            write_mode: WriteMode::Append,
104            primary_key_columns: Vec::new(),
105            flush_interval: Duration::from_millis(250),
106            connect_timeout: Duration::from_secs(10),
107            ssl_mode: SslMode::VerifyFull,
108            ssl_ca_cert_path: None,
109            auto_create_table: false,
110            changelog_mode: false,
111            statement_timeout: Duration::from_secs(30),
112        }
113    }
114}
115
116impl PostgresSinkConfig {
117    /// Creates a minimal config for testing.
118    #[must_use]
119    pub fn new(hostname: &str, database: &str, table_name: &str) -> Self {
120        Self {
121            hostname: hostname.to_string(),
122            database: database.to_string(),
123            table_name: table_name.to_string(),
124            ..Default::default()
125        }
126    }
127
128    /// Parses a sink config from a resolved [`ConnectorConfig`].
129    ///
130    /// # Required keys
131    ///
132    /// - `hostname` - `PostgreSQL` server hostname
133    /// - `database` - Target database name
134    /// - `username` - Authentication username
135    /// - `table.name` - Target table name
136    ///
137    /// # Errors
138    ///
139    /// Returns `ConnectorError::MissingConfig` if required keys are absent,
140    /// or `ConnectorError::ConfigurationError` on invalid values.
141    #[allow(clippy::field_reassign_with_default)]
142    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
143        Self::reject_removed_keys(config)?;
144        config.reject_unknown_properties(ALLOWED_CONFIG_KEYS, "PostgreSQL sink")?;
145
146        let mut cfg = Self::default();
147
148        cfg.hostname = config.require("hostname")?.to_string();
149        cfg.database = config.require("database")?.to_string();
150        cfg.username = config.require("username")?.to_string();
151        cfg.table_name = config.require("table.name")?.to_string();
152
153        if let Some(v) = config.get("password") {
154            cfg.password = v.to_string();
155        }
156        if let Some(v) = config.get("port") {
157            cfg.port = crate::config::parse_port(v)?;
158        }
159        if let Some(v) = config.get("schema.name") {
160            cfg.schema_name = v.to_string();
161        }
162        if let Some(v) = config.get("write.mode") {
163            cfg.write_mode = v.parse().map_err(|_| {
164                ConnectorError::ConfigurationError(format!(
165                    "invalid write.mode: '{v}' (expected 'append' or 'upsert')"
166                ))
167            })?;
168        }
169        if let Some(v) = config.get("primary.key") {
170            if v.trim().is_empty() {
171                cfg.primary_key_columns.clear();
172            } else {
173                let columns: Vec<_> = v.split(',').map(str::trim).collect();
174                if columns.iter().any(|column| column.is_empty()) {
175                    return Err(ConnectorError::ConfigurationError(
176                        "primary.key contains an empty column name".into(),
177                    ));
178                }
179                cfg.primary_key_columns = columns.into_iter().map(str::to_string).collect();
180            }
181        }
182        if let Some(v) = config.get("flush.interval.ms") {
183            let ms: u64 = v.parse().map_err(|_| {
184                ConnectorError::ConfigurationError(format!("invalid flush.interval.ms: '{v}'"))
185            })?;
186            cfg.flush_interval = Duration::from_millis(ms);
187        }
188        if let Some(v) = config.get("connect.timeout.ms") {
189            let ms: u64 = v.parse().map_err(|_| {
190                ConnectorError::ConfigurationError(format!("invalid connect.timeout.ms: '{v}'"))
191            })?;
192            cfg.connect_timeout = Duration::from_millis(ms);
193        }
194        if let Some(v) = config.get("ssl.mode") {
195            cfg.ssl_mode = v.parse().map_err(|_| {
196                ConnectorError::ConfigurationError(format!(
197                    "invalid ssl.mode: '{v}' (expected 'disable' or 'verify-full')"
198                ))
199            })?;
200        }
201        cfg.ssl_ca_cert_path = config.get("ssl.ca.cert.path").map(PathBuf::from);
202        if let Some(v) = config.get("auto.create.table") {
203            cfg.auto_create_table = v.eq_ignore_ascii_case("true");
204        }
205        if let Some(v) = config.get("changelog.mode") {
206            cfg.changelog_mode = v.eq_ignore_ascii_case("true");
207        }
208        if let Some(v) = config.get("statement.timeout.ms") {
209            let ms: u64 = v.parse().map_err(|_| {
210                ConnectorError::ConfigurationError(format!("invalid statement.timeout.ms: '{v}'"))
211            })?;
212            cfg.statement_timeout = Duration::from_millis(ms);
213        }
214        cfg.validate()?;
215        Ok(cfg)
216    }
217
218    fn reject_removed_keys(config: &ConnectorConfig) -> Result<(), ConnectorError> {
219        if let Some(key) = REMOVED_CONFIG_KEYS
220            .iter()
221            .find(|key| config.get(key).is_some())
222        {
223            return Err(ConnectorError::ConfigurationError(format!(
224                "PostgreSQL sink property '{key}' is not supported: batching is owned by the runtime and retained-byte limit"
225            )));
226        }
227        Ok(())
228    }
229
230    /// Validates the configuration for consistency.
231    ///
232    /// # Errors
233    ///
234    /// Returns `ConnectorError::ConfigurationError` on invalid combinations.
235    pub fn validate(&self) -> Result<(), ConnectorError> {
236        crate::config::require_non_empty(&self.hostname, "hostname")?;
237        crate::config::require_non_empty(&self.database, "database")?;
238        crate::config::require_non_empty(&self.username, "username")?;
239        crate::config::require_non_empty(&self.schema_name, "schema.name")?;
240        crate::config::require_non_empty(&self.table_name, "table.name")?;
241        validate_sql_identifier(&self.schema_name, "schema.name")?;
242        validate_sql_identifier(&self.table_name, "table.name")?;
243        let mut primary_keys = std::collections::HashSet::new();
244        for column in &self.primary_key_columns {
245            validate_sql_identifier(column, "primary.key column")?;
246            if !primary_keys.insert(column) {
247                return Err(ConnectorError::ConfigurationError(format!(
248                    "primary.key contains duplicate column '{column}'"
249                )));
250            }
251        }
252        if self.port == 0 {
253            return Err(ConnectorError::ConfigurationError(
254                "port must be > 0".into(),
255            ));
256        }
257        if self
258            .ssl_ca_cert_path
259            .as_ref()
260            .is_some_and(|path| path.as_os_str().is_empty())
261        {
262            return Err(ConnectorError::ConfigurationError(
263                "ssl.ca.cert.path must not be empty".into(),
264            ));
265        }
266        if self.ssl_mode == SslMode::Disable && self.ssl_ca_cert_path.is_some() {
267            return Err(ConnectorError::ConfigurationError(
268                "ssl.ca.cert.path requires ssl.mode=verify-full".into(),
269            ));
270        }
271        if self.write_mode == WriteMode::Upsert && self.primary_key_columns.is_empty() {
272            return Err(ConnectorError::ConfigurationError(
273                "upsert mode requires 'primary.key' to be set".into(),
274            ));
275        }
276        if self.changelog_mode && self.write_mode != WriteMode::Upsert {
277            return Err(ConnectorError::ConfigurationError(
278                "changelog mode requires write.mode = 'upsert'".into(),
279            ));
280        }
281        if self.flush_interval.is_zero() {
282            return Err(ConnectorError::ConfigurationError(
283                "flush.interval.ms must be > 0".into(),
284            ));
285        }
286        if self.connect_timeout.is_zero() {
287            return Err(ConnectorError::ConfigurationError(
288                "connect.timeout.ms must be > 0".into(),
289            ));
290        }
291        if self.statement_timeout < Duration::from_secs(1) {
292            return Err(ConnectorError::ConfigurationError(
293                "statement.timeout.ms must be >= 1000 (1 second)".into(),
294            ));
295        }
296        if self.statement_timeout.as_millis() > MAX_POSTGRES_STATEMENT_TIMEOUT_MS {
297            return Err(ConnectorError::ConfigurationError(format!(
298                "statement.timeout.ms must be <= {MAX_POSTGRES_STATEMENT_TIMEOUT_MS}"
299            )));
300        }
301        if !self
302            .statement_timeout
303            .subsec_nanos()
304            .is_multiple_of(1_000_000)
305        {
306            return Err(ConnectorError::ConfigurationError(
307                "statement timeout must be an integer number of milliseconds".into(),
308            ));
309        }
310        Ok(())
311    }
312
313    /// `PostgreSQL` startup option applied to every connection created by the pool.
314    #[must_use]
315    pub(super) fn statement_timeout_startup_option(&self) -> String {
316        format!(
317            "-c statement_timeout={}",
318            self.statement_timeout.as_millis()
319        )
320    }
321
322    /// Returns the safely quoted fully qualified table name (`"schema"."table"`).
323    #[must_use]
324    pub fn qualified_table_name(&self) -> String {
325        format!(
326            "{}.{}",
327            quote_sql_identifier(&self.schema_name),
328            quote_sql_identifier(&self.table_name)
329        )
330    }
331}
332
333/// Validates one `PostgreSQL` identifier segment before it can reach generated SQL.
334pub(super) fn validate_sql_identifier(identifier: &str, label: &str) -> Result<(), ConnectorError> {
335    if identifier.is_empty() {
336        return Err(ConnectorError::ConfigurationError(format!(
337            "{label} must not be empty"
338        )));
339    }
340    if identifier.contains('\0') {
341        return Err(ConnectorError::ConfigurationError(format!(
342            "{label} must not contain NUL"
343        )));
344    }
345    if identifier.len() > MAX_POSTGRES_IDENTIFIER_BYTES {
346        return Err(ConnectorError::ConfigurationError(format!(
347            "{label} exceeds PostgreSQL's {MAX_POSTGRES_IDENTIFIER_BYTES}-byte identifier limit"
348        )));
349    }
350    Ok(())
351}
352
353/// Quotes one already-validated identifier segment. `PostgreSQL` escapes embedded quotes by
354/// doubling them; dots remain literal characters within the segment.
355pub(super) fn quote_sql_identifier(identifier: &str) -> String {
356    format!("\"{}\"", identifier.replace('"', "\"\""))
357}
358
359/// Write mode for the `PostgreSQL` sink.
360#[derive(Debug, Clone, Copy, PartialEq, Eq)]
361pub enum WriteMode {
362    /// Append-only: uses COPY BINARY for maximum throughput.
363    /// No deduplication — every record is inserted.
364    Append,
365    /// Upsert: `INSERT ... ON CONFLICT DO UPDATE`.
366    /// Requires primary key columns. Deduplicates on key.
367    Upsert,
368}
369
370str_enum!(WriteMode, lowercase_nodash, String, "unknown write mode",
371    Append => "append", "copy";
372    Upsert => "upsert", "insert"
373);
374
375#[cfg(test)]
376mod tests;
laminar_connectors/postgres/sink_config/mod.rs

laminar_connectors/postgres/sink_config/
mod.rs
