faucet_source_csv/
config.rs

1//! CSV source configuration.
2
3use faucet_core::DEFAULT_BATCH_SIZE;
4use schemars::JsonSchema;
5use serde::{Deserialize, Serialize};
6
7/// Configuration for the CSV file source.
8#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
9pub struct CsvSourceConfig {
10    /// Path to the CSV file.
11    pub path: String,
12    /// Whether the file has a header row. Defaults to `true`.
13    #[serde(default = "default_true")]
14    pub has_headers: bool,
15    /// Field delimiter byte. Defaults to `b','`.
16    #[serde(default = "default_delimiter")]
17    pub delimiter: u8,
18    /// Quote character byte. Defaults to `b'"'`.
19    #[serde(default = "default_quote")]
20    pub quote: u8,
21    /// Whether to tolerate rows with a field count that differs from the
22    /// header (or from the first data row when header-less). Defaults to
23    /// `false` (strict).
24    ///
25    /// When `false` (the default), a row whose field count does not match the
26    /// expected width is a structural defect and aborts the run with a typed
27    /// [`FaucetError::Source`](faucet_core::FaucetError::Source) naming the
28    /// offending line — silently emitting an incomplete record would corrupt
29    /// downstream consumers. Set to `true` only when ragged rows are expected
30    /// and acceptable (short rows yield records missing the trailing columns;
31    /// long rows gain `column_N` keys).
32    #[serde(default)]
33    pub flexible: bool,
34    /// Records per emitted [`StreamPage`](faucet_core::StreamPage). Rows are
35    /// parsed line-by-line from a tokio `BufReader` and yielded whenever the
36    /// buffer reaches this size. Defaults to [`DEFAULT_BATCH_SIZE`].
37    ///
38    /// `batch_size = 0` is the "no batching" sentinel: the file is fully
39    /// drained and the entire result set is emitted in a single page. Useful
40    /// for small lookup tables or for sinks (e.g. SQL `COPY`, BigQuery load
41    /// jobs) that prefer one large request to many small ones.
42    #[serde(default = "default_batch_size")]
43    pub batch_size: usize,
44    /// Compression codec for the input file. Defaults to
45    /// [`CompressionConfig::Auto`](faucet_core::CompressionConfig::Auto) —
46    /// `.gz` / `.zst` suffix selects gzip / zstd. Requires the crate-local
47    /// `compression` feature.
48    #[cfg(feature = "compression")]
49    #[serde(default)]
50    pub compression: faucet_core::CompressionConfig,
51}
52
53fn default_true() -> bool {
54    true
55}
56
57fn default_delimiter() -> u8 {
58    b','
59}
60
61fn default_quote() -> u8 {
62    b'"'
63}
64
65fn default_batch_size() -> usize {
66    DEFAULT_BATCH_SIZE
67}
68
69impl CsvSourceConfig {
70    /// Create a new config with the required file path and sensible defaults.
71    pub fn new(path: impl Into<String>) -> Self {
72        Self {
73            path: path.into(),
74            has_headers: true,
75            delimiter: b',',
76            quote: b'"',
77            flexible: false,
78            batch_size: DEFAULT_BATCH_SIZE,
79            #[cfg(feature = "compression")]
80            compression: faucet_core::CompressionConfig::Auto,
81        }
82    }
83
84    /// Set whether the file has a header row.
85    pub fn has_headers(mut self, v: bool) -> Self {
86        self.has_headers = v;
87        self
88    }
89
90    /// Set the field delimiter byte.
91    pub fn delimiter(mut self, d: u8) -> Self {
92        self.delimiter = d;
93        self
94    }
95
96    /// Set the quote character byte.
97    pub fn quote(mut self, q: u8) -> Self {
98        self.quote = q;
99        self
100    }
101
102    /// Set whether to tolerate rows with an uneven field count.
103    ///
104    /// Defaults to `false` (strict): a ragged row aborts the run. Pass `true`
105    /// to opt in to lenient parsing where short/long rows are accepted.
106    pub fn flexible(mut self, v: bool) -> Self {
107        self.flexible = v;
108        self
109    }
110
111    /// Set the per-page row count for [`Source::stream_pages`](faucet_core::Source::stream_pages).
112    ///
113    /// Pass `0` to opt out of batching — the entire file is emitted in a
114    /// single [`StreamPage`](faucet_core::StreamPage).
115    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
116        self.batch_size = batch_size;
117        self
118    }
119
120    /// Set the compression codec. Available only with the `compression` feature.
121    #[cfg(feature = "compression")]
122    pub fn compression(mut self, c: faucet_core::CompressionConfig) -> Self {
123        self.compression = c;
124        self
125    }
126
127    /// Validate the config at load time so a bad config fails fast with a typed
128    /// `FaucetError::Config` instead of surfacing deep in a run: rejects an
129    /// out-of-range `batch_size` (`> MAX_BATCH_SIZE`) and an empty `path`.
130    ///
131    /// `faucet_core` is referenced by full path here (rather than imported) so
132    /// the field-doc links above keep their explicit targets and the committed
133    /// config JSON Schema stays byte-identical.
134    pub fn validate(&self) -> Result<(), faucet_core::FaucetError> {
135        if self.path.trim().is_empty() {
136            return Err(faucet_core::FaucetError::Config(
137                "CSV source requires a non-empty `path`".into(),
138            ));
139        }
140        faucet_core::validate_batch_size(self.batch_size)?;
141        Ok(())
142    }
143}
144
145#[cfg(test)]
146mod tests {
147    use super::*;
148
149    #[test]
150    fn default_config() {
151        let config = CsvSourceConfig::new("/tmp/data.csv");
152        assert_eq!(config.path, "/tmp/data.csv");
153        assert!(config.has_headers);
154        assert_eq!(config.delimiter, b',');
155        assert_eq!(config.quote, b'"');
156    }
157
158    #[test]
159    fn builder_methods() {
160        let config = CsvSourceConfig::new("/tmp/data.tsv")
161            .has_headers(false)
162            .delimiter(b'\t')
163            .quote(b'\'');
164        assert!(!config.has_headers);
165        assert_eq!(config.delimiter, b'\t');
166        assert_eq!(config.quote, b'\'');
167    }
168
169    #[test]
170    fn flexible_defaults_to_false_strict() {
171        let config = CsvSourceConfig::new("/tmp/data.csv");
172        assert!(!config.flexible);
173    }
174
175    #[test]
176    fn flexible_builder_and_serde_default() {
177        let config = CsvSourceConfig::new("/tmp/data.csv").flexible(true);
178        assert!(config.flexible);
179
180        // Absent in JSON => strict default.
181        let json = r#"{ "path": "/tmp/data.csv" }"#;
182        let parsed: CsvSourceConfig = serde_json::from_str(json).unwrap();
183        assert!(!parsed.flexible);
184
185        // Explicit value round-trips.
186        let json = r#"{ "path": "/tmp/data.csv", "flexible": true }"#;
187        let parsed: CsvSourceConfig = serde_json::from_str(json).unwrap();
188        assert!(parsed.flexible);
189    }
190
191    #[test]
192    fn batch_size_defaults_to_default_batch_size() {
193        let config = CsvSourceConfig::new("/tmp/data.csv");
194        assert_eq!(config.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
195    }
196
197    #[test]
198    fn with_batch_size_overrides_default() {
199        let config = CsvSourceConfig::new("/tmp/data.csv").with_batch_size(500);
200        assert_eq!(config.batch_size, 500);
201    }
202
203    #[test]
204    fn batch_size_zero_is_accepted_as_no_batching_sentinel() {
205        let config = CsvSourceConfig::new("/tmp/data.csv").with_batch_size(0);
206        assert_eq!(config.batch_size, 0);
207        assert!(faucet_core::validate_batch_size(config.batch_size).is_ok());
208    }
209
210    #[test]
211    fn batch_size_above_max_is_rejected_by_validate_batch_size() {
212        let config =
213            CsvSourceConfig::new("/tmp/data.csv").with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
214        assert!(faucet_core::validate_batch_size(config.batch_size).is_err());
215    }
216
217    #[test]
218    fn batch_size_deserializes_from_json() {
219        let json = r#"{
220            "path": "/tmp/data.csv",
221            "batch_size": 250
222        }"#;
223        let config: CsvSourceConfig = serde_json::from_str(json).unwrap();
224        assert_eq!(config.batch_size, 250);
225    }
226
227    #[test]
228    fn validate_accepts_valid_config() {
229        assert!(CsvSourceConfig::new("/tmp/data.csv").validate().is_ok());
230    }
231
232    #[test]
233    fn validate_rejects_oversized_batch_size() {
234        let config =
235            CsvSourceConfig::new("/tmp/data.csv").with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
236        assert!(matches!(
237            config.validate(),
238            Err(faucet_core::FaucetError::Config(_))
239        ));
240    }
241
242    #[test]
243    fn validate_rejects_empty_path() {
244        assert!(matches!(
245            CsvSourceConfig::new("   ").validate(),
246            Err(faucet_core::FaucetError::Config(_))
247        ));
248    }
249}
faucet_source_csv/config.rs

faucet_source_csv/
config.rs
