faucet_source_mssql/
config.rs

1//! Configuration for the MSSQL query source.
2
3use faucet_common_mssql::MssqlConnectionConfig;
4use faucet_core::{DEFAULT_BATCH_SIZE, FaucetError, validate_batch_size};
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8
9fn default_max_connections() -> u32 {
10    10
11}
12fn default_batch_size() -> usize {
13    DEFAULT_BATCH_SIZE
14}
15fn default_statement_timeout_secs() -> u64 {
16    300
17}
18
19/// How the source replicates rows across runs.
20///
21/// Serializes as `{ type: full }` or
22/// `{ type: incremental, column: "...", initial_value: ... }`.
23#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema, Default, PartialEq)]
24#[serde(tag = "type", rename_all = "snake_case")]
25pub enum MssqlReplication {
26    /// Every run fetches the full result set (default).
27    #[default]
28    Full,
29    /// Only rows whose `column` is strictly greater than the stored bookmark
30    /// (or `initial_value` on the first run) are emitted.
31    ///
32    /// The bookmark is applied two ways: if the query contains the literal
33    /// token `@bookmark`, it is bound as a parameter so the server filters
34    /// (efficient); the source *also* filters client-side as a correctness
35    /// backstop. The new maximum of `column` is persisted on the final page.
36    Incremental {
37        /// Column whose value is the replication cursor (e.g. `updated_at`).
38        column: String,
39        /// Lower bound used on the first run, before any bookmark is stored.
40        initial_value: Value,
41    },
42}
43
44/// Configuration for [`MssqlSource`](crate::MssqlSource).
45#[derive(Clone, Serialize, Deserialize, JsonSchema)]
46pub struct MssqlSourceConfig {
47    /// Connection + TLS settings (`connection_url` or `connection_string`).
48    #[serde(flatten)]
49    pub connection: MssqlConnectionConfig,
50    /// SQL query to run. Use `@P1`, `@P2`, … for [`params`](Self::params), and
51    /// the literal `@bookmark` token to bind the incremental cursor server-side.
52    pub query: String,
53    /// Positional bind parameters for the query (`@P1`…`@Pn`). Defaults to empty.
54    #[serde(default)]
55    pub params: Vec<Value>,
56    /// Maximum pooled connections. Defaults to 10.
57    #[serde(default = "default_max_connections")]
58    pub max_connections: u32,
59    /// Records per emitted [`StreamPage`](faucet_core::StreamPage). `0` emits the
60    /// whole result set as a single page. Defaults to [`DEFAULT_BATCH_SIZE`].
61    #[serde(default = "default_batch_size")]
62    pub batch_size: usize,
63    /// Per-query timeout in seconds (`0` disables). Defaults to 300.
64    #[serde(default = "default_statement_timeout_secs")]
65    pub statement_timeout_secs: u64,
66    /// Replication mode. Defaults to [`MssqlReplication::Full`].
67    #[serde(default)]
68    pub replication: MssqlReplication,
69    /// Explicit state-store key for the bookmark. When unset, a key is derived
70    /// from the connection host and a query fingerprint.
71    #[serde(default, skip_serializing_if = "Option::is_none")]
72    pub state_key: Option<String>,
73}
74
75impl std::fmt::Debug for MssqlSourceConfig {
76    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
77        f.debug_struct("MssqlSourceConfig")
78            .field("connection", &"***")
79            .field("query", &self.query)
80            .field("params", &self.params)
81            .field("max_connections", &self.max_connections)
82            .field("batch_size", &self.batch_size)
83            .field("statement_timeout_secs", &self.statement_timeout_secs)
84            .field("replication", &self.replication)
85            .field("state_key", &self.state_key)
86            .finish()
87    }
88}
89
90impl MssqlSourceConfig {
91    /// Build a config from a connection URL and query, with defaults elsewhere.
92    pub fn new(connection_url: impl Into<String>, query: impl Into<String>) -> Self {
93        Self {
94            connection: MssqlConnectionConfig {
95                connection_url: Some(connection_url.into()),
96                ..Default::default()
97            },
98            query: query.into(),
99            params: Vec::new(),
100            max_connections: default_max_connections(),
101            batch_size: default_batch_size(),
102            statement_timeout_secs: default_statement_timeout_secs(),
103            replication: MssqlReplication::Full,
104            state_key: None,
105        }
106    }
107
108    /// Validate connection source, batch size, and replication settings.
109    pub fn validate(&self) -> Result<(), FaucetError> {
110        self.connection.validate()?;
111        validate_batch_size(self.batch_size)?;
112        if let MssqlReplication::Incremental { column, .. } = &self.replication
113            && column.trim().is_empty()
114        {
115            return Err(FaucetError::Config(
116                "MSSQL incremental replication requires a non-empty `column`".into(),
117            ));
118        }
119        if self.incremental_without_bookmark_pushdown() {
120            tracing::warn!(
121                "MSSQL incremental replication query has no `@bookmark` token: the \
122                 cursor is applied client-side only, so the server returns the ENTIRE \
123                 table on every run (correctness is preserved, but it is a full re-scan). \
124                 Add `@bookmark` to the WHERE clause to push the cursor down, e.g. \
125                 `... WHERE {column} > @bookmark`",
126                column = match &self.replication {
127                    MssqlReplication::Incremental { column, .. } => column.as_str(),
128                    _ => "<column>",
129                }
130            );
131        }
132        Ok(())
133    }
134
135    /// `true` when replication is `Incremental` but the query omits the
136    /// `@bookmark` token, so the cursor cannot be pushed down to the server and
137    /// every run re-scans the whole table (F53). Pure predicate so the
138    /// load-time warning's condition is unit-testable.
139    pub(crate) fn incremental_without_bookmark_pushdown(&self) -> bool {
140        matches!(self.replication, MssqlReplication::Incremental { .. })
141            && !self.query.contains("@bookmark")
142    }
143}
144
145#[cfg(test)]
146mod tests {
147    use super::*;
148    use serde_json::json;
149
150    #[test]
151    fn replication_full_is_default_and_round_trips() {
152        let r: MssqlReplication = serde_json::from_value(json!({"type": "full"})).unwrap();
153        assert_eq!(r, MssqlReplication::Full);
154        assert_eq!(MssqlReplication::default(), MssqlReplication::Full);
155    }
156
157    #[test]
158    fn replication_incremental_parses_column_and_initial_value() {
159        let r: MssqlReplication = serde_json::from_value(json!({
160            "type": "incremental",
161            "column": "updated_at",
162            "initial_value": "1970-01-01T00:00:00Z"
163        }))
164        .unwrap();
165        assert_eq!(
166            r,
167            MssqlReplication::Incremental {
168                column: "updated_at".into(),
169                initial_value: json!("1970-01-01T00:00:00Z"),
170            }
171        );
172    }
173
174    #[test]
175    fn config_flattens_connection_fields() {
176        let cfg: MssqlSourceConfig = serde_json::from_value(json!({
177            "connection_url": "mssql://sa:pw@host:1433/db",
178            "query": "SELECT 1",
179        }))
180        .unwrap();
181        assert_eq!(
182            cfg.connection.connection_url.as_deref(),
183            Some("mssql://sa:pw@host:1433/db")
184        );
185        assert_eq!(cfg.batch_size, DEFAULT_BATCH_SIZE);
186        assert_eq!(cfg.max_connections, 10);
187        assert_eq!(cfg.statement_timeout_secs, 300);
188    }
189
190    #[test]
191    fn validate_rejects_incremental_without_column() {
192        let cfg = MssqlSourceConfig {
193            replication: MssqlReplication::Incremental {
194                column: "  ".into(),
195                initial_value: json!(0),
196            },
197            ..MssqlSourceConfig::new("mssql://sa:pw@h/db", "SELECT 1")
198        };
199        assert!(cfg.validate().is_err());
200    }
201
202    #[test]
203    fn incremental_without_bookmark_pushdown_flags_missing_token() {
204        // F53: Incremental + query lacking `@bookmark` → full re-scan warning.
205        let missing = MssqlSourceConfig {
206            replication: MssqlReplication::Incremental {
207                column: "updated_at".into(),
208                initial_value: json!("1970-01-01"),
209            },
210            ..MssqlSourceConfig::new("mssql://sa:pw@h/db", "SELECT * FROM t")
211        };
212        assert!(missing.incremental_without_bookmark_pushdown());
213        // validate still succeeds (warn, not hard-error).
214        assert!(missing.validate().is_ok());
215
216        // With the token present, no warning.
217        let with_token = MssqlSourceConfig {
218            replication: MssqlReplication::Incremental {
219                column: "updated_at".into(),
220                initial_value: json!("1970-01-01"),
221            },
222            ..MssqlSourceConfig::new(
223                "mssql://sa:pw@h/db",
224                "SELECT * FROM t WHERE updated_at > @bookmark",
225            )
226        };
227        assert!(!with_token.incremental_without_bookmark_pushdown());
228
229        // Full replication never warns.
230        let full = MssqlSourceConfig::new("mssql://sa:pw@h/db", "SELECT * FROM t");
231        assert!(!full.incremental_without_bookmark_pushdown());
232    }
233
234    #[test]
235    fn validate_rejects_bad_batch_size() {
236        let cfg = MssqlSourceConfig {
237            batch_size: faucet_core::MAX_BATCH_SIZE + 1,
238            ..MssqlSourceConfig::new("mssql://sa:pw@h/db", "SELECT 1")
239        };
240        assert!(cfg.validate().is_err());
241    }
242
243    #[test]
244    fn debug_masks_connection() {
245        let cfg = MssqlSourceConfig::new("mssql://sa:secret@h/db", "SELECT 1");
246        let dbg = format!("{cfg:?}");
247        assert!(dbg.contains("***"));
248        assert!(!dbg.contains("secret"));
249    }
250}
faucet_source_mssql/config.rs

faucet_source_mssql/
config.rs
