faucet_sink_mysql/
config.rs

1//! MySQL sink configuration.
2
3use faucet_core::{DEFAULT_BATCH_SIZE, WriteSpec};
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 MysqlColumnMapping {
11    /// Insert each record as a single JSON column. The column name
12    /// defaults to `"data"` but can be overridden.
13    Json { 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 MysqlColumnMapping {
20    fn default() -> Self {
21        Self::Json {
22            column: "data".into(),
23        }
24    }
25}
26
27/// Configuration for the MySQL sink.
28#[derive(Clone, Serialize, Deserialize, JsonSchema)]
29pub struct MysqlSinkConfig {
30    /// MySQL connection URL (e.g. `mysql://user:pass@host/db`).
31    pub connection_url: String,
32    /// Target table name.
33    pub table_name: String,
34    /// How to map JSON records to columns. Defaults to a single JSON column.
35    #[serde(default)]
36    pub column_mapping: MysqlColumnMapping,
37    /// Maximum rows per multi-row `INSERT` statement. Defaults to
38    /// [`DEFAULT_BATCH_SIZE`].
39    ///
40    /// When the upstream `StreamPage` carries more records than `batch_size`,
41    /// the sink slices the page into `batch_size`-row chunks and issues one
42    /// multi-row `INSERT INTO ... VALUES (...), (...), ...` statement per
43    /// chunk. When `batch_size = 0`, the entire upstream page is forwarded
44    /// in a single multi-row `INSERT` — useful when the source already
45    /// chunks to a size tuned for MySQL.
46    ///
47    /// `batch_size = 0` is the "no batching" sentinel: the full upstream
48    /// page is forwarded as one `INSERT`, subject to MySQL's
49    /// `max_allowed_packet` limit (default 64MB). Keep the default unless
50    /// the upstream `StreamPage` size is already tuned for MySQL.
51    #[serde(default = "default_batch_size")]
52    pub batch_size: usize,
53    /// Maximum number of connections in the pool. Defaults to 5.
54    ///
55    /// Bounded on purpose: a finite pool keeps a wide fan-out from exhausting
56    /// the server's own connection limit. There is no "unlimited" setting.
57    #[serde(default = "default_max_connections")]
58    pub max_connections: u32,
59    /// Write mode: `append` (default), `upsert`, or `delete`.
60    ///
61    /// `upsert` and `delete` require `column_mapping: auto_map` (key columns
62    /// must be real table columns, not packed inside a JSON blob) and a
63    /// non-empty `key` list. The table must already have a PRIMARY or UNIQUE
64    /// index on the key columns; MySQL's `ON DUPLICATE KEY UPDATE` uses that
65    /// index to detect conflicts.
66    #[serde(flatten)]
67    pub write: WriteSpec,
68}
69
70fn default_batch_size() -> usize {
71    DEFAULT_BATCH_SIZE
72}
73
74fn default_max_connections() -> u32 {
75    5
76}
77
78impl std::fmt::Debug for MysqlSinkConfig {
79    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
80        f.debug_struct("MysqlSinkConfig")
81            .field("connection_url", &"***")
82            .field("table_name", &self.table_name)
83            .field("column_mapping", &self.column_mapping)
84            .field("batch_size", &self.batch_size)
85            .field("max_connections", &self.max_connections)
86            .finish()
87    }
88}
89
90impl MysqlSinkConfig {
91    /// Create a new config with required fields and sensible defaults.
92    pub fn new(connection_url: impl Into<String>, table_name: impl Into<String>) -> Self {
93        Self {
94            connection_url: connection_url.into(),
95            table_name: table_name.into(),
96            column_mapping: MysqlColumnMapping::default(),
97            batch_size: DEFAULT_BATCH_SIZE,
98            max_connections: 5,
99            write: WriteSpec::default(),
100        }
101    }
102
103    /// Set the column mapping strategy.
104    pub fn column_mapping(mut self, mapping: MysqlColumnMapping) -> Self {
105        self.column_mapping = mapping;
106        self
107    }
108
109    /// Set the per-statement row count for the multi-row `INSERT`.
110    ///
111    /// Pass `0` to opt out of re-chunking — the sink forwards each upstream
112    /// [`StreamPage`](faucet_core::StreamPage) as a single multi-row
113    /// `INSERT`. MySQL's multi-row insert sweet spot is ~1000 rows.
114    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
115        self.batch_size = batch_size;
116        self
117    }
118
119    /// Set the maximum number of connections in the pool.
120    pub fn max_connections(mut self, n: u32) -> Self {
121        self.max_connections = n;
122        self
123    }
124}
125
126#[cfg(test)]
127mod tests {
128    use super::*;
129
130    #[test]
131    fn default_config() {
132        let config = MysqlSinkConfig::new("mysql://localhost/test", "events");
133        assert_eq!(config.table_name, "events");
134        assert_eq!(config.batch_size, DEFAULT_BATCH_SIZE);
135        assert!(matches!(
136            config.column_mapping,
137            MysqlColumnMapping::Json { ref column } if column == "data"
138        ));
139    }
140
141    #[test]
142    fn builder_methods() {
143        let config = MysqlSinkConfig::new("mysql://localhost/test", "events")
144            .column_mapping(MysqlColumnMapping::AutoMap)
145            .with_batch_size(100);
146        assert_eq!(config.batch_size, 100);
147        assert!(matches!(config.column_mapping, MysqlColumnMapping::AutoMap));
148    }
149
150    #[test]
151    fn with_batch_size_overrides_default() {
152        let config = MysqlSinkConfig::new("mysql://localhost/test", "events").with_batch_size(250);
153        assert_eq!(config.batch_size, 250);
154    }
155
156    #[test]
157    fn json_custom_column() {
158        let config = MysqlSinkConfig::new("mysql://localhost/test", "events").column_mapping(
159            MysqlColumnMapping::Json {
160                column: "payload".into(),
161            },
162        );
163        assert!(matches!(
164            config.column_mapping,
165            MysqlColumnMapping::Json { ref column } if column == "payload"
166        ));
167    }
168
169    #[test]
170    fn debug_masks_connection_url() {
171        let config = MysqlSinkConfig::new("mysql://secret:pass@host/db", "events");
172        let debug = format!("{config:?}");
173        assert!(debug.contains("***"));
174        assert!(!debug.contains("secret"));
175        assert!(!debug.contains("pass"));
176    }
177
178    #[test]
179    fn batch_size_zero_is_accepted_as_no_batching_sentinel() {
180        let config = MysqlSinkConfig::new("mysql://localhost/test", "events").with_batch_size(0);
181        assert_eq!(config.batch_size, 0);
182        assert!(faucet_core::validate_batch_size(config.batch_size).is_ok());
183    }
184
185    #[test]
186    fn batch_size_above_max_is_rejected_by_validate_batch_size() {
187        let config = MysqlSinkConfig::new("mysql://localhost/test", "events")
188            .with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
189        assert!(faucet_core::validate_batch_size(config.batch_size).is_err());
190    }
191
192    #[test]
193    fn batch_size_deserializes_from_json() {
194        let json = r#"{
195            "connection_url": "mysql://localhost/test",
196            "table_name": "events",
197            "column_mapping": {"json": {"column": "data"}},
198            "batch_size": 250,
199            "max_connections": 5
200        }"#;
201        let config: MysqlSinkConfig = serde_json::from_str(json).unwrap();
202        assert_eq!(config.batch_size, 250);
203    }
204
205    #[test]
206    fn batch_size_defaults_when_absent_in_json() {
207        let json = r#"{
208            "connection_url": "mysql://localhost/test",
209            "table_name": "events",
210            "column_mapping": {"json": {"column": "data"}},
211            "max_connections": 5
212        }"#;
213        let config: MysqlSinkConfig = serde_json::from_str(json).unwrap();
214        assert_eq!(config.batch_size, DEFAULT_BATCH_SIZE);
215    }
216
217    #[test]
218    fn with_batch_size_chaining() {
219        let config = MysqlSinkConfig::new("mysql://localhost/test", "events")
220            .with_batch_size(100)
221            .with_batch_size(2_000);
222        assert_eq!(config.batch_size, 2_000);
223    }
224
225    #[test]
226    fn max_connections_and_column_mapping_default_when_absent_in_json() {
227        // Only the two genuinely-required fields are supplied; pool size and
228        // column mapping must fall back to their documented defaults.
229        let json = r#"{
230            "connection_url": "mysql://localhost/test",
231            "table_name": "events"
232        }"#;
233        let config: MysqlSinkConfig = serde_json::from_str(json).unwrap();
234        assert_eq!(config.max_connections, 5);
235        assert!(matches!(
236            config.column_mapping,
237            MysqlColumnMapping::Json { .. }
238        ));
239        assert_eq!(config.batch_size, DEFAULT_BATCH_SIZE);
240    }
241}
faucet_sink_mysql/config.rs

faucet_sink_mysql/
config.rs
