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

faucet_sink_mysql/
config.rs
