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

feldera_types/transport/
postgres.rs
