faucet_sink_mssql/
config.rs

1//! Configuration for the MSSQL sink.
2
3use faucet_common_mssql::MssqlConnectionConfig;
4use faucet_core::{FaucetError, validate_batch_size};
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7
8fn default_batch_size() -> usize {
9    500
10}
11fn default_max_connections() -> u32 {
12    5
13}
14fn default_statement_timeout_secs() -> u64 {
15    300
16}
17fn default_true() -> bool {
18    true
19}
20
21/// What to do with record keys that don't match a table column (`auto_columns`).
22#[derive(Clone, Copy, Debug, Serialize, Deserialize, JsonSchema, Default, PartialEq, Eq)]
23#[serde(rename_all = "snake_case")]
24pub enum OnUnknownField {
25    /// Log a one-shot warning and drop the unknown keys (default).
26    #[default]
27    Warn,
28    /// Silently drop the unknown keys.
29    Drop,
30    /// Fail the write with [`FaucetError::Sink`].
31    Error,
32}
33
34/// How records map onto table columns.
35///
36/// Serializes as `{ type: json_column, column: "data" }` (default) or
37/// `{ type: auto_columns, on_unknown_field: warn }`.
38#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema, PartialEq)]
39#[serde(tag = "type", rename_all = "snake_case")]
40pub enum MssqlColumnMapping {
41    /// Map top-level JSON keys to same-named table columns. IDENTITY columns are
42    /// skipped (the server generates them).
43    AutoColumns {
44        #[serde(default)]
45        on_unknown_field: OnUnknownField,
46    },
47    /// Serialize each record to a JSON string inserted into a single
48    /// `NVARCHAR(MAX)` (or native `JSON`) column.
49    JsonColumn { column: String },
50}
51
52impl Default for MssqlColumnMapping {
53    fn default() -> Self {
54        Self::JsonColumn {
55            column: "data".into(),
56        }
57    }
58}
59
60/// Configuration for [`MssqlSink`](crate::MssqlSink).
61#[derive(Clone, Serialize, Deserialize, JsonSchema)]
62pub struct MssqlSinkConfig {
63    /// Connection + TLS settings (`connection_url` or `connection_string`).
64    #[serde(flatten)]
65    pub connection: MssqlConnectionConfig,
66    /// Target table, optionally schema-qualified (e.g. `dbo.events`).
67    pub table: String,
68    /// How records map onto columns. Defaults to a single `data` JSON column.
69    #[serde(default)]
70    pub column_mapping: MssqlColumnMapping,
71    /// Rows per multi-row `INSERT`. Auto-split further so `rows * columns` stays
72    /// within MSSQL's 2100-parameter limit. `0` sends the whole page (still
73    /// param-split). Defaults to 500.
74    #[serde(default = "default_batch_size")]
75    pub batch_size: usize,
76    /// Maximum pooled connections. Defaults to 5.
77    #[serde(default = "default_max_connections")]
78    pub max_connections: u32,
79    /// Wrap each batch's INSERTs in `BEGIN TRAN` / `COMMIT TRAN`. Defaults to true.
80    #[serde(default = "default_true")]
81    pub transaction_per_batch: bool,
82    /// On a batch failure, retry row-by-row to isolate the offender so good rows
83    /// still land and only the bad row is DLQ-routed. When false, one bad row
84    /// fails the whole batch (fewer round-trips). Defaults to true.
85    #[serde(default = "default_true")]
86    pub isolate_row_failures: bool,
87    /// Per-statement timeout in seconds (`0` disables). Defaults to 300.
88    #[serde(default = "default_statement_timeout_secs")]
89    pub statement_timeout_secs: u64,
90    /// In `json_column` mode only, create the table if absent as
91    /// `(id BIGINT IDENTITY PRIMARY KEY, <column> NVARCHAR(MAX))`. Rejected with
92    /// `auto_columns` (schema inference is unsafe for MSSQL types). Defaults to false.
93    #[serde(default)]
94    pub create_table: bool,
95    /// Write mode (`append` / `upsert` / `delete`) + `key` / `delete_marker`.
96    /// `upsert` and `delete` require `column_mapping: auto_columns` (the key
97    /// columns must be real table columns, not buried inside a JSON column).
98    #[serde(flatten)]
99    pub write: faucet_core::WriteSpec,
100}
101
102impl std::fmt::Debug for MssqlSinkConfig {
103    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
104        f.debug_struct("MssqlSinkConfig")
105            .field("connection", &"***")
106            .field("table", &self.table)
107            .field("column_mapping", &self.column_mapping)
108            .field("batch_size", &self.batch_size)
109            .field("max_connections", &self.max_connections)
110            .field("transaction_per_batch", &self.transaction_per_batch)
111            .field("isolate_row_failures", &self.isolate_row_failures)
112            .field("statement_timeout_secs", &self.statement_timeout_secs)
113            .field("create_table", &self.create_table)
114            .field("write", &self.write)
115            .finish()
116    }
117}
118
119impl MssqlSinkConfig {
120    /// Build a config from a connection URL and table, with defaults elsewhere.
121    pub fn new(connection_url: impl Into<String>, table: impl Into<String>) -> Self {
122        Self {
123            connection: MssqlConnectionConfig {
124                connection_url: Some(connection_url.into()),
125                ..Default::default()
126            },
127            table: table.into(),
128            column_mapping: MssqlColumnMapping::default(),
129            batch_size: default_batch_size(),
130            max_connections: default_max_connections(),
131            transaction_per_batch: true,
132            isolate_row_failures: true,
133            statement_timeout_secs: default_statement_timeout_secs(),
134            create_table: false,
135            write: faucet_core::WriteSpec::default(),
136        }
137    }
138
139    /// Validate connection source, batch size, table, and mode combination.
140    pub fn validate(&self) -> Result<(), FaucetError> {
141        self.connection.validate()?;
142        validate_batch_size(self.batch_size)?;
143        if self.table.trim().is_empty() {
144            return Err(FaucetError::Config("MSSQL sink requires a `table`".into()));
145        }
146        if self.create_table
147            && matches!(self.column_mapping, MssqlColumnMapping::AutoColumns { .. })
148        {
149            return Err(FaucetError::Config(
150                "MSSQL sink `create_table` is only supported with `json_column` mode \
151                 (schema inference for auto_columns is unsafe — create the table first)"
152                    .into(),
153            ));
154        }
155        Ok(())
156    }
157}
158
159#[cfg(test)]
160mod tests {
161    use super::*;
162    use serde_json::json;
163
164    #[test]
165    fn json_column_is_default() {
166        assert_eq!(
167            MssqlColumnMapping::default(),
168            MssqlColumnMapping::JsonColumn {
169                column: "data".into()
170            }
171        );
172    }
173
174    #[test]
175    fn column_mapping_round_trips() {
176        let auto: MssqlColumnMapping =
177            serde_json::from_value(json!({"type": "auto_columns", "on_unknown_field": "error"}))
178                .unwrap();
179        assert_eq!(
180            auto,
181            MssqlColumnMapping::AutoColumns {
182                on_unknown_field: OnUnknownField::Error
183            }
184        );
185
186        let jc: MssqlColumnMapping =
187            serde_json::from_value(json!({"type": "json_column", "column": "payload"})).unwrap();
188        assert_eq!(
189            jc,
190            MssqlColumnMapping::JsonColumn {
191                column: "payload".into()
192            }
193        );
194    }
195
196    #[test]
197    fn auto_columns_defaults_unknown_field_to_warn() {
198        let auto: MssqlColumnMapping =
199            serde_json::from_value(json!({"type": "auto_columns"})).unwrap();
200        assert_eq!(
201            auto,
202            MssqlColumnMapping::AutoColumns {
203                on_unknown_field: OnUnknownField::Warn
204            }
205        );
206    }
207
208    #[test]
209    fn config_defaults() {
210        let cfg: MssqlSinkConfig = serde_json::from_value(json!({
211            "connection_url": "mssql://sa:pw@h/db",
212            "table": "dbo.events",
213        }))
214        .unwrap();
215        assert_eq!(cfg.batch_size, 500);
216        assert_eq!(cfg.max_connections, 5);
217        assert!(cfg.transaction_per_batch);
218        assert!(cfg.isolate_row_failures);
219        assert_eq!(cfg.statement_timeout_secs, 300);
220        assert!(!cfg.create_table);
221    }
222
223    #[test]
224    fn validate_rejects_auto_columns_with_create_table() {
225        let cfg = MssqlSinkConfig {
226            column_mapping: MssqlColumnMapping::AutoColumns {
227                on_unknown_field: OnUnknownField::Warn,
228            },
229            create_table: true,
230            ..MssqlSinkConfig::new("mssql://sa:pw@h/db", "dbo.events")
231        };
232        assert!(cfg.validate().is_err());
233    }
234
235    #[test]
236    fn validate_rejects_empty_table() {
237        let cfg = MssqlSinkConfig::new("mssql://sa:pw@h/db", "  ");
238        assert!(cfg.validate().is_err());
239    }
240
241    #[test]
242    fn debug_masks_connection() {
243        let cfg = MssqlSinkConfig::new("mssql://sa:secret@h/db", "dbo.t");
244        let dbg = format!("{cfg:?}");
245        assert!(dbg.contains("***"));
246        assert!(!dbg.contains("secret"));
247    }
248}
faucet_sink_mssql/config.rs

faucet_sink_mssql/
config.rs
