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
128#[cfg(test)]
129mod tests {
130    use super::*;
131
132    #[test]
133    fn default_config() {
134        let config = CsvSourceConfig::new("/tmp/data.csv");
135        assert_eq!(config.path, "/tmp/data.csv");
136        assert!(config.has_headers);
137        assert_eq!(config.delimiter, b',');
138        assert_eq!(config.quote, b'"');
139    }
140
141    #[test]
142    fn builder_methods() {
143        let config = CsvSourceConfig::new("/tmp/data.tsv")
144            .has_headers(false)
145            .delimiter(b'\t')
146            .quote(b'\'');
147        assert!(!config.has_headers);
148        assert_eq!(config.delimiter, b'\t');
149        assert_eq!(config.quote, b'\'');
150    }
151
152    #[test]
153    fn flexible_defaults_to_false_strict() {
154        let config = CsvSourceConfig::new("/tmp/data.csv");
155        assert!(!config.flexible);
156    }
157
158    #[test]
159    fn flexible_builder_and_serde_default() {
160        let config = CsvSourceConfig::new("/tmp/data.csv").flexible(true);
161        assert!(config.flexible);
162
163        // Absent in JSON => strict default.
164        let json = r#"{ "path": "/tmp/data.csv" }"#;
165        let parsed: CsvSourceConfig = serde_json::from_str(json).unwrap();
166        assert!(!parsed.flexible);
167
168        // Explicit value round-trips.
169        let json = r#"{ "path": "/tmp/data.csv", "flexible": true }"#;
170        let parsed: CsvSourceConfig = serde_json::from_str(json).unwrap();
171        assert!(parsed.flexible);
172    }
173
174    #[test]
175    fn batch_size_defaults_to_default_batch_size() {
176        let config = CsvSourceConfig::new("/tmp/data.csv");
177        assert_eq!(config.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
178    }
179
180    #[test]
181    fn with_batch_size_overrides_default() {
182        let config = CsvSourceConfig::new("/tmp/data.csv").with_batch_size(500);
183        assert_eq!(config.batch_size, 500);
184    }
185
186    #[test]
187    fn batch_size_zero_is_accepted_as_no_batching_sentinel() {
188        let config = CsvSourceConfig::new("/tmp/data.csv").with_batch_size(0);
189        assert_eq!(config.batch_size, 0);
190        assert!(faucet_core::validate_batch_size(config.batch_size).is_ok());
191    }
192
193    #[test]
194    fn batch_size_above_max_is_rejected_by_validate_batch_size() {
195        let config =
196            CsvSourceConfig::new("/tmp/data.csv").with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
197        assert!(faucet_core::validate_batch_size(config.batch_size).is_err());
198    }
199
200    #[test]
201    fn batch_size_deserializes_from_json() {
202        let json = r#"{
203            "path": "/tmp/data.csv",
204            "batch_size": 250
205        }"#;
206        let config: CsvSourceConfig = serde_json::from_str(json).unwrap();
207        assert_eq!(config.batch_size, 250);
208    }
209}
faucet_source_csv/config.rs

faucet_source_csv/
config.rs
