feldera_types/transport/
postgres.rs

1use serde::{Deserialize, Serialize};
2use std::fmt::Display;
3use utoipa::ToSchema;
4
5/// PostgreSQL write mode.
6///
7/// Determines how the PostgreSQL output connector writes data to the target table.
8#[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema, Default)]
9pub enum PostgresWriteMode {
10    /// Materialized mode: perform direct INSERT, UPDATE, and DELETE operations on the table.
11    /// This is the default behavior and maintains the postgres table as a materialized snapshot of the output view.
12    #[default]
13    #[serde(rename = "materialized")]
14    Materialized,
15
16    /// CDC (Change Data Capture) mode: write all operations as INSERT operations
17    /// into a Postgres table that serves as an append-only event log.
18    /// In this mode, inserts, updates, and deletes are all represented as new rows
19    /// with metadata columns describing the operation type and timestamp.
20    #[serde(rename = "cdc")]
21    Cdc,
22}
23
24impl Display for PostgresWriteMode {
25    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
26        match self {
27            Self::Materialized => write!(f, "materialized"),
28            Self::Cdc => write!(f, "cdc"),
29        }
30    }
31}
32
33/// TLS/SSL configuration for PostgreSQL connectors.
34#[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema, Default)]
35pub struct PostgresTlsConfig {
36    /// A sequence of CA certificates in PEM format.
37    pub ssl_ca_pem: Option<String>,
38
39    /// Path to a file containing a sequence of CA certificates in PEM format.
40    pub ssl_ca_location: Option<String>,
41
42    /// The client certificate in PEM format.
43    pub ssl_client_pem: Option<String>,
44
45    /// Path to the client certificate.
46    pub ssl_client_location: Option<String>,
47
48    /// The client certificate key in PEM format.
49    pub ssl_client_key: Option<String>,
50
51    /// Path to the client certificate key.
52    pub ssl_client_key_location: Option<String>,
53
54    /// The path to the certificate chain file.
55    /// The file must contain a sequence of PEM-formatted certificates,
56    /// the first being the leaf certificate, and the remainder forming
57    /// the chain of certificates up to and including the trusted root certificate.
58    pub ssl_certificate_chain_location: Option<String>,
59
60    /// True to enable hostname verification when using TLS. True by default.
61    ///
62    /// When false, the certificate chain is still verified against the
63    /// trusted CA; only the requirement that the server name appears in the
64    /// certificate is lifted.
65    pub verify_hostname: Option<bool>,
66}
67
68impl PostgresTlsConfig {
69    pub fn has_tls(&self) -> bool {
70        self.ssl_ca_pem.is_some() || self.ssl_ca_location.is_some()
71    }
72}
73
74/// Postgres CDC input connector configuration.
75///
76/// Uses logical replication to capture ongoing changes from a Postgres database.
77/// Requires a pre-created publication and a user with REPLICATION privilege.
78/// Tables must have primary keys and `REPLICA IDENTITY FULL` is recommended
79/// for UPDATE/DELETE support.
80#[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema)]
81pub struct PostgresCdcReaderConfig {
82    /// Postgres connection URI. The user must have REPLICATION privilege.
83    /// See: <https://docs.rs/tokio-postgres/0.7.12/tokio_postgres/config/struct.Config.html>
84    pub uri: String,
85
86    /// Name of the pre-created Postgres publication to replicate from.
87    pub publication: String,
88
89    /// Postgres table to replicate (e.g. "public.orders").
90    /// Must be included in the publication.
91    pub source_table: String,
92
93    /// TLS/SSL configuration.
94    #[serde(flatten)]
95    #[schema(inline)]
96    pub tls: PostgresTlsConfig,
97}
98
99impl PostgresCdcReaderConfig {
100    pub fn validate(&self) -> Result<(), String> {
101        if self.publication.trim().is_empty() {
102            return Err("publication cannot be empty".to_string());
103        }
104
105        if self.source_table.trim().is_empty() {
106            return Err("source_table cannot be empty".to_string());
107        }
108
109        if self.tls.ssl_client_pem.is_some()
110            || self.tls.ssl_client_location.is_some()
111            || self.tls.ssl_client_key.is_some()
112            || self.tls.ssl_client_key_location.is_some()
113            || self.tls.ssl_certificate_chain_location.is_some()
114        {
115            return Err(
116                "client-certificate TLS options (ssl_client_pem, ssl_client_location, \
117                 ssl_client_key, ssl_client_key_location, ssl_certificate_chain_location) \
118                 are not supported by the Postgres CDC connector as the underlying etl crate \
119                 doesn't support client-certificate TLS yet. CA-based TLS via ssl_ca_pem \
120                 or ssl_ca_location is supported. Please file an issue if you require \
121                 client-certificate TLS support: https://github.com/feldera/feldera/issues/
122                 "
123                .to_string(),
124            );
125        }
126
127        if self.tls.verify_hostname == Some(false) {
128            return Err(
129                "disabling hostname verification is not supported by the Postgres CDC connector"
130                    .to_string(),
131            );
132        }
133
134        Ok(())
135    }
136}
137
138/// Postgres input connector configuration.
139#[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema)]
140pub struct PostgresReaderConfig {
141    /// Postgres URI.
142    /// See: <https://docs.rs/tokio-postgres/0.7.12/tokio_postgres/config/struct.Config.html>
143    pub uri: String,
144
145    /// Query that specifies what data to fetch from postgres.
146    pub query: String,
147
148    /// TLS/SSL configuration.
149    #[serde(flatten)]
150    #[schema(inline)]
151    pub tls: PostgresTlsConfig,
152}
153
154/// Postgres output connector configuration.
155#[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema)]
156pub struct PostgresWriterConfig {
157    /// Postgres URI.
158    /// See: <https://docs.rs/tokio-postgres/0.7.12/tokio_postgres/config/struct.Config.html>
159    pub uri: String,
160
161    /// The table to write the output to.
162    pub table: String,
163
164    /// Write mode for the connector.
165    ///
166    /// - `materialized` (default): Perform direct INSERT, UPDATE, and DELETE operations on the table.
167    /// - `cdc`: Write all operations as INSERT operations into an append-only event log
168    ///   with additional metadata columns describing the operation type and timestamp.
169    #[serde(default)]
170    #[schema(default = PostgresWriteMode::default)]
171    pub mode: PostgresWriteMode,
172
173    /// Name of the operation metadata column in CDC mode.
174    ///
175    /// Only used when `mode = "cdc"`. This column will contain:
176    /// - `"i"` for insert operations
177    /// - `"u"` for upsert operations
178    /// - `"d"` for delete operations
179    ///
180    /// Default: `"__feldera_op"`
181    #[serde(default = "default_cdc_op_column")]
182    #[schema(default = default_cdc_op_column)]
183    pub cdc_op_column: String,
184
185    /// Name of the timestamp metadata column in CDC mode.
186    ///
187    /// Only used when `mode = "cdc"`. This column will contain the timestamp
188    /// (in RFC 3339 format) when the batch of updates was output
189    /// by the pipeline.
190    ///
191    /// Default: `"__feldera_ts"`
192    #[serde(default = "default_cdc_ts_column")]
193    #[schema(default = default_cdc_ts_column)]
194    pub cdc_ts_column: String,
195
196    /// TLS/SSL configuration.
197    #[serde(flatten)]
198    #[schema(inline)]
199    pub tls: PostgresTlsConfig,
200
201    /// The maximum number of records in a single buffer.
202    pub max_records_in_buffer: Option<usize>,
203
204    /// The maximum buffer size in for a single operation.
205    /// Note that the buffers of `INSERT`, `UPDATE` and `DELETE` queries are
206    /// separate.
207    /// Default: 1 MiB
208    #[schema(default = default_max_buffer_size)]
209    #[serde(default = "default_max_buffer_size")]
210    pub max_buffer_size_bytes: usize,
211
212    /// Specifies how the connector handles conflicts when executing an `INSERT`
213    /// into a table with a primary key. By default, an existing row with the same
214    /// key is overwritten. Setting this flag to `true` preserves the existing row
215    /// and ignores the new insert.
216    ///
217    /// This setting does not affect `UPDATE` statements, which always replace the
218    /// value associated with the key.
219    ///
220    /// This setting is not supported when `mode = "cdc"`, since all operations
221    /// are performed as append-only `INSERT`s into the target table.
222    /// Any conflict in CDC mode will result in an error.
223    ///
224    /// Default: `false`
225    #[serde(default)]
226    pub on_conflict_do_nothing: bool,
227
228    /// The number of threads to use during encoding.
229    ///
230    /// Default: 1
231    #[serde(default = "default_writer_threads")]
232    #[schema(default = default_writer_threads)]
233    pub threads: usize,
234
235    /// The names of the extra columns in the Postgres table that are not part of the view schema.
236    ///
237    /// These connector can write user-defined values, configured using the `set_extra_columns` connector command,
238    /// to these columns.
239    #[serde(default, skip_serializing_if = "Vec::is_empty")]
240    pub extra_columns: Vec<String>,
241}
242
243fn default_max_buffer_size() -> usize {
244    usize::pow(2, 20)
245}
246
247fn default_writer_threads() -> usize {
248    1
249}
250
251fn default_cdc_op_column() -> String {
252    "__feldera_op".to_string()
253}
254
255fn default_cdc_ts_column() -> String {
256    "__feldera_ts".to_string()
257}
258
259impl PostgresWriterConfig {
260    pub fn validate(&self) -> Result<(), String> {
261        match self.mode {
262            PostgresWriteMode::Cdc => {
263                if self.cdc_op_column.trim().is_empty() {
264                    return Err("cdc_op_column cannot be empty in CDC mode".to_string());
265                }
266                if self.cdc_ts_column.trim().is_empty() {
267                    return Err("cdc_ts_column cannot be empty in CDC mode".to_string());
268                }
269
270                if !self.cdc_op_column.is_ascii() {
271                    return Err("cdc_op_column must contain only ASCII characters".to_string());
272                }
273
274                if !self.cdc_ts_column.is_ascii() {
275                    return Err("cdc_ts_column must contain only ASCII characters".to_string());
276                }
277
278                if self.on_conflict_do_nothing {
279                    return Err("on_conflict_do_nothing not supported in CDC mode since all operations are performed as append-only INSERTs into the target table".to_string());
280                }
281            }
282            PostgresWriteMode::Materialized => {
283                if self.cdc_ts_column != default_cdc_ts_column()
284                    && !self.cdc_ts_column.trim().is_empty()
285                {
286                    return Err(
287                        "cdc_ts_column must not be set when in MATERIALIZED mode".to_string()
288                    );
289                }
290                if self.cdc_op_column != default_cdc_op_column()
291                    && !self.cdc_op_column.trim().is_empty()
292                {
293                    return Err(
294                        "cdc_op_column must not be set when in MATERIALIZED mode".to_string()
295                    );
296                }
297            }
298        };
299
300        if self.threads == 0 {
301            return Err("threads must be at least 1".to_string());
302        }
303
304        Ok(())
305    }
306}
307
308#[cfg(test)]
309mod tests {
310    use super::*;
311
312    fn postgres_cdc_config(tls: PostgresTlsConfig) -> PostgresCdcReaderConfig {
313        PostgresCdcReaderConfig {
314            uri: "postgres://user:password@localhost:5432/database".to_string(),
315            publication: "publication".to_string(),
316            source_table: "public.table".to_string(),
317            tls,
318        }
319    }
320
321    #[test]
322    fn postgres_cdc_config_rejects_client_certificate_tls_options() {
323        let config = postgres_cdc_config(PostgresTlsConfig {
324            ssl_client_pem: Some("client".to_string()),
325            ..Default::default()
326        });
327
328        let err = config.validate().unwrap_err();
329        assert!(err.contains("client-certificate TLS options"));
330        assert!(err.contains("client-certificate TLS support"));
331        assert!(!err.contains("doesn't support TLS yet"));
332    }
333
334    #[test]
335    fn postgres_cdc_config_rejects_disabled_hostname_verification() {
336        let config = postgres_cdc_config(PostgresTlsConfig {
337            verify_hostname: Some(false),
338            ..Default::default()
339        });
340
341        let err = config.validate().unwrap_err();
342        assert!(err.contains("disabling hostname verification"));
343    }
344
345    #[test]
346    fn postgres_cdc_config_accepts_default_tls() {
347        let config = postgres_cdc_config(PostgresTlsConfig::default());
348
349        assert!(config.validate().is_ok());
350    }
351
352    #[test]
353    fn postgres_cdc_config_rejects_empty_publication() {
354        let mut config = postgres_cdc_config(PostgresTlsConfig::default());
355        config.publication = "   ".to_string();
356
357        let err = config.validate().unwrap_err();
358        assert!(err.contains("publication cannot be empty"));
359    }
360
361    #[test]
362    fn postgres_cdc_config_rejects_empty_source_table() {
363        let mut config = postgres_cdc_config(PostgresTlsConfig::default());
364        config.source_table = "\t".to_string();
365
366        let err = config.validate().unwrap_err();
367        assert!(err.contains("source_table cannot be empty"));
368    }
369}
feldera_types/transport/postgres.rs

feldera_types/transport/
postgres.rs
