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

feldera_types/transport/
postgres.rs
