faucet_sink_postgres/
config.rs

1//! PostgreSQL sink configuration.
2
3use faucet_core::DEFAULT_BATCH_SIZE;
4use schemars::JsonSchema;
5use serde::{Deserialize, Serialize};
6
7/// How to map JSON records to table columns.
8#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
9#[serde(rename_all = "snake_case")]
10pub enum PostgresColumnMapping {
11    /// Insert each record as a single `jsonb` column. The column name
12    /// defaults to `"data"` but can be overridden.
13    Jsonb { column: String },
14    /// Map top-level JSON keys directly to table columns.
15    /// Only keys that match existing columns are inserted; extra keys are ignored.
16    AutoMap,
17}
18
19impl Default for PostgresColumnMapping {
20    fn default() -> Self {
21        Self::Jsonb {
22            column: "data".into(),
23        }
24    }
25}
26
27/// Configuration for the PostgreSQL sink.
28#[derive(Clone, Serialize, Deserialize, JsonSchema)]
29pub struct PostgresSinkConfig {
30    /// PostgreSQL connection URL (e.g. `postgres://user:pass@host/db`).
31    pub connection_url: String,
32    /// Target table name.
33    pub table_name: String,
34    /// Optional schema (namespace) qualifying [`table_name`](Self::table_name).
35    ///
36    /// When set, both the AutoMap column-discovery probe and the `INSERT`
37    /// target `schema.table_name` explicitly. When unset (the default), the
38    /// table resolves against the connection's `search_path`, and column
39    /// discovery is scoped to whichever schema the `INSERT` actually resolves
40    /// to — so a same-named table in another schema no longer pollutes the
41    /// AutoMap column set (#146 M13).
42    #[serde(default)]
43    pub schema: Option<String>,
44    /// How to map JSON records to columns.
45    pub column_mapping: PostgresColumnMapping,
46    /// Maximum rows per multi-row `INSERT` statement. Defaults to
47    /// [`DEFAULT_BATCH_SIZE`].
48    ///
49    /// When the upstream `StreamPage` carries more records than `batch_size`,
50    /// the sink slices the page into `batch_size`-row chunks and issues one
51    /// multi-row `INSERT` per chunk. When `batch_size = 0`, the entire slice
52    /// is sent in a single `INSERT` — useful when the source already chunks
53    /// to a Postgres-friendly size.
54    ///
55    /// `batch_size = 0` is the "no batching" sentinel: the entire upstream
56    /// page is forwarded in one statement, subject to Postgres' natural
57    /// per-statement bind-parameter limit of 65 535. AutoMap mode binds one
58    /// parameter per column per row, so the safe ceiling is roughly
59    /// `65_535 / num_columns` rows per call; JSONB mode binds a single
60    /// array parameter and has no such ceiling. Keep the default unless the
61    /// upstream page size is already tuned for Postgres.
62    ///
63    /// **Recommended value: ~1000** — Postgres' multi-row `INSERT` sweet
64    /// spot. Larger chunks rarely add throughput and risk hitting the
65    /// 65 535-parameter ceiling in AutoMap mode.
66    #[serde(default = "default_batch_size")]
67    pub batch_size: usize,
68    /// Maximum number of connections in the pool. Defaults to 5.
69    pub max_connections: u32,
70    /// Write mode, key columns, and optional delete marker. `write_mode`
71    /// defaults to `append`. Upsert/delete require `column_mapping: auto_map`
72    /// and a UNIQUE/PRIMARY KEY constraint on `key`.
73    #[serde(flatten)]
74    pub write: faucet_core::WriteSpec,
75}
76
77fn default_batch_size() -> usize {
78    DEFAULT_BATCH_SIZE
79}
80
81impl std::fmt::Debug for PostgresSinkConfig {
82    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
83        f.debug_struct("PostgresSinkConfig")
84            .field("connection_url", &"***")
85            .field("table_name", &self.table_name)
86            .field("schema", &self.schema)
87            .field("column_mapping", &self.column_mapping)
88            .field("batch_size", &self.batch_size)
89            .field("max_connections", &self.max_connections)
90            .finish()
91    }
92}
93
94impl PostgresSinkConfig {
95    /// Create a new config with required fields and sensible defaults.
96    pub fn new(connection_url: impl Into<String>, table_name: impl Into<String>) -> Self {
97        Self {
98            connection_url: connection_url.into(),
99            table_name: table_name.into(),
100            schema: None,
101            column_mapping: PostgresColumnMapping::default(),
102            batch_size: DEFAULT_BATCH_SIZE,
103            max_connections: 5,
104            write: faucet_core::WriteSpec::default(),
105        }
106    }
107
108    /// Set the schema (namespace) that qualifies the table. When unset, the
109    /// table resolves against the connection's `search_path`.
110    pub fn with_schema(mut self, schema: impl Into<String>) -> Self {
111        self.schema = Some(schema.into());
112        self
113    }
114
115    /// Set the column mapping strategy.
116    pub fn column_mapping(mut self, mapping: PostgresColumnMapping) -> Self {
117        self.column_mapping = mapping;
118        self
119    }
120
121    /// Set the per-statement row count for multi-row `INSERT`.
122    ///
123    /// Pass `0` to opt out of re-chunking — the sink forwards each upstream
124    /// [`StreamPage`](faucet_core::StreamPage) as a single `INSERT`
125    /// statement. Postgres' multi-row `INSERT` sweet spot is ~1000 rows.
126    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
127        self.batch_size = batch_size;
128        self
129    }
130
131    /// Set the maximum number of connections in the pool.
132    pub fn max_connections(mut self, n: u32) -> Self {
133        self.max_connections = n;
134        self
135    }
136}
137
138#[cfg(test)]
139mod tests {
140    use super::*;
141
142    #[test]
143    fn default_config() {
144        let config = PostgresSinkConfig::new("postgres://localhost/test", "events");
145        assert_eq!(config.table_name, "events");
146        assert_eq!(config.batch_size, DEFAULT_BATCH_SIZE);
147        assert!(matches!(
148            config.column_mapping,
149            PostgresColumnMapping::Jsonb { ref column } if column == "data"
150        ));
151    }
152
153    #[test]
154    fn builder_methods() {
155        let config = PostgresSinkConfig::new("postgres://localhost/test", "events")
156            .column_mapping(PostgresColumnMapping::AutoMap)
157            .with_batch_size(100);
158        assert_eq!(config.batch_size, 100);
159        assert!(matches!(
160            config.column_mapping,
161            PostgresColumnMapping::AutoMap
162        ));
163    }
164
165    #[test]
166    fn jsonb_custom_column() {
167        let config = PostgresSinkConfig::new("postgres://localhost/test", "events").column_mapping(
168            PostgresColumnMapping::Jsonb {
169                column: "payload".into(),
170            },
171        );
172        assert!(matches!(
173            config.column_mapping,
174            PostgresColumnMapping::Jsonb { ref column } if column == "payload"
175        ));
176    }
177
178    #[test]
179    fn with_batch_size_overrides_default() {
180        let config =
181            PostgresSinkConfig::new("postgres://localhost/test", "events").with_batch_size(250);
182        assert_eq!(config.batch_size, 250);
183    }
184
185    #[test]
186    fn batch_size_zero_is_accepted_as_no_batching_sentinel() {
187        let config =
188            PostgresSinkConfig::new("postgres://localhost/test", "events").with_batch_size(0);
189        assert_eq!(config.batch_size, 0);
190        assert!(faucet_core::validate_batch_size(config.batch_size).is_ok());
191    }
192
193    #[test]
194    fn batch_size_above_max_is_rejected_by_validate_batch_size() {
195        let config = PostgresSinkConfig::new("postgres://localhost/test", "events")
196            .with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
197        assert!(faucet_core::validate_batch_size(config.batch_size).is_err());
198    }
199
200    #[test]
201    fn batch_size_deserializes_from_json() {
202        let json = r#"{
203            "connection_url": "postgres://localhost/test",
204            "table_name": "events",
205            "column_mapping": {"jsonb": {"column": "data"}},
206            "batch_size": 250,
207            "max_connections": 5
208        }"#;
209        let config: PostgresSinkConfig = serde_json::from_str(json).unwrap();
210        assert_eq!(config.batch_size, 250);
211    }
212
213    #[test]
214    fn batch_size_defaults_when_absent_in_json() {
215        let json = r#"{
216            "connection_url": "postgres://localhost/test",
217            "table_name": "events",
218            "column_mapping": {"jsonb": {"column": "data"}},
219            "max_connections": 5
220        }"#;
221        let config: PostgresSinkConfig = serde_json::from_str(json).unwrap();
222        assert_eq!(config.batch_size, DEFAULT_BATCH_SIZE);
223    }
224
225    #[test]
226    fn config_builder_chaining() {
227        let config = PostgresSinkConfig::new("postgres://localhost/test", "events")
228            .with_batch_size(100)
229            .with_batch_size(250);
230        assert_eq!(config.batch_size, 250);
231    }
232}
faucet_sink_postgres/config.rs

faucet_sink_postgres/
config.rs
