faucet_source_file/
config.rs

1//! File source configuration.
2
3use faucet_core::{
4    AvroOptions, CompressionConfig, CsvOptions, DEFAULT_BATCH_SIZE, ExcelOptions, FaucetError,
5    FormatOptions, OrcOptions, XmlOptions,
6};
7use schemars::JsonSchema;
8use serde::{Deserialize, Serialize};
9use std::collections::BTreeMap;
10
11/// The format choice — `auto` by extension, per file, or one fixed format.
12/// Shared with the file sink through `faucet-common-file`.
13pub use faucet_common_file::FileFormatChoice as FileSourceFormat;
14
15/// What makes a file "new" in incremental mode.
16#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
17#[serde(rename_all = "snake_case")]
18pub enum IncrementalBy {
19    /// Modification time: a run reads files modified after the newest one the
20    /// previous run read (ties broken by path), so a rewritten file is read
21    /// again. Over HTTP the `Last-Modified` header is the modification time.
22    Mtime,
23    /// Name: a run reads files whose path sorts after the last one read —
24    /// the right choice for dated names (`export-2026-09-27.csv`) that are
25    /// never rewritten.
26    Name,
27}
28
29/// Incremental mode: re-runs pick up only new files.
30#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
31#[serde(deny_unknown_fields)]
32pub struct IncrementalSpec {
33    /// What makes a file new.
34    pub by: IncrementalBy,
35}
36
37fn default_concurrency() -> usize {
38    4
39}
40
41fn default_batch_size() -> usize {
42    DEFAULT_BATCH_SIZE
43}
44
45fn default_http_retries() -> u32 {
46    3
47}
48
49fn default_max_object_bytes() -> u64 {
50    faucet_core::file_format::DEFAULT_MAX_OBJECT_BYTES
51}
52
53fn default_http_connect_timeout_secs() -> u64 {
54    30
55}
56
57fn default_http_read_timeout_secs() -> u64 {
58    300
59}
60
61/// Configuration for the local (and `http(s)://`) file source.
62///
63/// **Experimental** (PRINCIPLES.md §3): this block's shape may change in a
64/// minor release; any change is called out in the changelog.
65#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
66#[serde(deny_unknown_fields)]
67pub struct FileSourceConfig {
68    /// A file, a directory, a glob (`data/**/*.jsonl`), or one `http://` /
69    /// `https://` URL. A directory lists its regular files (descending into
70    /// subdirectories with `recursive: true`); every listing is sorted by path.
71    #[schemars(extend("x-faucet-resource" = "path"))]
72    pub path: String,
73    /// File format, `auto` (by extension, per file) by default. Under `auto`
74    /// a file with no recognised extension is skipped with a warning, or
75    /// refused when `strict: true`. JSON Lines, Avro, ORC and Parquet stream;
76    /// JSON arrays, CSV, XML and Excel are read whole per file.
77    #[serde(default)]
78    pub format: FileSourceFormat,
79    /// Compression codec. `auto` (default) resolves per file from its suffix
80    /// (`.gz`, `.zst`).
81    #[serde(default)]
82    pub compression: CompressionConfig,
83    /// Descend into subdirectories when `path` is a directory.
84    #[serde(default)]
85    pub recursive: bool,
86    /// Files opened ahead of the one being decoded (ordered prefetch). `0`
87    /// reads serially.
88    #[serde(default = "default_concurrency")]
89    pub concurrency: usize,
90    /// Records per emitted page. `0` emits one page per file.
91    #[serde(default = "default_batch_size")]
92    pub batch_size: usize,
93    /// Read only files that are new since the previous run. Needs a
94    /// pipeline `state:` block; the bookmark is kept under `file:<path>`.
95    #[serde(default, skip_serializing_if = "Option::is_none")]
96    pub incremental: Option<IncrementalSpec>,
97    /// Skip files modified less than this many seconds ago — files still
98    /// being written. They are read by a later run once they settle.
99    #[serde(default, skip_serializing_if = "Option::is_none")]
100    pub stable_for_secs: Option<u64>,
101    /// Refuse (instead of skipping with a warning) a file whose extension
102    /// names no format under `format: auto`.
103    #[serde(default)]
104    pub strict: bool,
105    /// Largest file, once decompressed, that a format read whole (JSON array,
106    /// XML, Excel, compressed or remote Avro/ORC/Parquet, encrypted files) may
107    /// reach; a larger one fails with an error instead of exhausting memory.
108    /// Default 2 GiB.
109    #[serde(default = "default_max_object_bytes")]
110    pub max_object_bytes: u64,
111    /// Most bytes one object may decode to in memory where a format can
112    /// expand past its input: an Avro file's records (per batch where the file
113    /// streams), an Excel workbook's parts and shared strings. A file past it
114    /// fails with an error instead of exhausting memory. Default 64 ×
115    /// `max_object_bytes`, at most 16 GiB.
116    #[serde(default, skip_serializing_if = "Option::is_none")]
117    pub max_decoded_bytes: Option<u64>,
118    /// Read at most this many files (after sorting and filtering).
119    #[serde(default, skip_serializing_if = "Option::is_none")]
120    pub max_files: Option<usize>,
121    /// Request headers for an `http(s)://` path (e.g. `Authorization`).
122    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
123    pub headers: BTreeMap<String, String>,
124    /// Retries for a failed HTTP request (connection error, 429, 5xx), with
125    /// exponential backoff that honours `Retry-After`.
126    #[serde(default = "default_http_retries")]
127    pub http_retries: u32,
128    /// Seconds to wait for an HTTP connection to open. A timeout is retried.
129    #[serde(default = "default_http_connect_timeout_secs")]
130    pub http_connect_timeout_secs: u64,
131    /// Seconds an HTTP response may go without sending a byte — waiting for
132    /// the headers or mid-body. A timeout before the body starts is retried;
133    /// one mid-body fails the file.
134    #[serde(default = "default_http_read_timeout_secs")]
135    pub http_read_timeout_secs: u64,
136    /// CSV dialect, used for CSV files.
137    #[serde(default)]
138    pub csv: CsvOptions,
139    /// Worksheet selection, used for Excel files.
140    #[serde(default)]
141    pub excel: ExcelOptions,
142    /// Record framing, used for XML files.
143    #[serde(default)]
144    pub xml: XmlOptions,
145    /// Reader schema, used for Avro files.
146    #[serde(default)]
147    pub avro: AvroOptions,
148    /// Column projection, used for ORC files.
149    #[serde(default)]
150    pub orc: OrcOptions,
151    /// Column projection, used for Parquet files.
152    #[serde(default)]
153    pub parquet: ParquetReadOptions,
154    /// Decrypt files sealed by the file or jsonl sink's `encryption` block
155    /// (AES-256-GCM). An encrypted file is read into memory whole: JSON Lines
156    /// and raw text written line by line are then opened one line at a time,
157    /// every other file as one sealed body. A file that is not sealed fails
158    /// the read.
159    #[cfg(feature = "encryption")]
160    #[serde(default, skip_serializing_if = "Option::is_none")]
161    pub encryption: Option<faucet_core::EncryptionSpec>,
162}
163
164/// Parquet read options: the same `parquet:` block as the object-store
165/// sources.
166pub use faucet_core::ParquetReadOptions;
167
168impl FileSourceConfig {
169    /// A config reading `path` with every default.
170    pub fn new(path: impl Into<String>) -> Self {
171        Self {
172            path: path.into(),
173            format: FileSourceFormat::Auto,
174            compression: CompressionConfig::Auto,
175            recursive: false,
176            concurrency: default_concurrency(),
177            batch_size: DEFAULT_BATCH_SIZE,
178            incremental: None,
179            stable_for_secs: None,
180            strict: false,
181            max_files: None,
182            max_object_bytes: default_max_object_bytes(),
183            max_decoded_bytes: None,
184            headers: BTreeMap::new(),
185            http_retries: default_http_retries(),
186            http_connect_timeout_secs: default_http_connect_timeout_secs(),
187            http_read_timeout_secs: default_http_read_timeout_secs(),
188            csv: CsvOptions::default(),
189            excel: ExcelOptions::default(),
190            xml: XmlOptions::default(),
191            avro: AvroOptions::default(),
192            orc: OrcOptions::default(),
193            parquet: ParquetReadOptions::default(),
194            #[cfg(feature = "encryption")]
195            encryption: None,
196        }
197    }
198
199    /// Set the format.
200    pub fn format(mut self, format: FileSourceFormat) -> Self {
201        self.format = format;
202        self
203    }
204
205    /// Descend into subdirectories.
206    pub fn recursive(mut self, recursive: bool) -> Self {
207        self.recursive = recursive;
208        self
209    }
210
211    /// Set the records per page.
212    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
213        self.batch_size = batch_size;
214        self
215    }
216
217    /// Turn on incremental mode.
218    pub fn incremental(mut self, by: IncrementalBy) -> Self {
219        self.incremental = Some(IncrementalSpec { by });
220        self
221    }
222
223    /// Whether `path` is an `http(s)://` URL.
224    pub fn is_http(&self) -> bool {
225        faucet_common_file::is_http_path(&self.path)
226    }
227
228    /// Project Parquet files to `columns`.
229    pub fn parquet_columns<I, S>(mut self, columns: I) -> Self
230    where
231        I: IntoIterator<Item = S>,
232        S: Into<String>,
233    {
234        self.parquet.columns = Some(columns.into_iter().map(Into::into).collect());
235        self
236    }
237
238    /// Whether an `encryption` block is set.
239    pub fn encryption_set(&self) -> bool {
240        #[cfg(feature = "encryption")]
241        return self.encryption.is_some();
242        #[cfg(not(feature = "encryption"))]
243        false
244    }
245
246    /// `max_decoded_bytes`, or its default.
247    #[cfg_attr(
248        not(any(feature = "file-format-avro", feature = "file-format-excel")),
249        allow(dead_code)
250    )]
251    pub(crate) fn decoded_bytes_limit(&self) -> u64 {
252        faucet_core::file_format::budget::effective_max_decoded_bytes(
253            self.max_decoded_bytes,
254            self.max_object_bytes,
255        )
256    }
257
258    /// The per-format option blocks. CSV is strict about ragged rows unless
259    /// `csv.flexible` says otherwise.
260    pub fn format_options(&self) -> FormatOptions {
261        let mut csv = self.csv.clone();
262        csv.flexible.get_or_insert(false);
263        FormatOptions {
264            csv,
265            excel: self.excel.clone(),
266            xml: self.xml.clone(),
267            avro: self.avro.clone(),
268            orc: self.orc.clone(),
269        }
270    }
271
272    /// Load-time checks: batch size bounds, a non-empty path, a parseable
273    /// Avro reader schema, and header names/values HTTP can carry.
274    pub fn validate(&self) -> Result<(), FaucetError> {
275        faucet_core::validate_batch_size(self.batch_size)?;
276        faucet_common_file::require_path("file source", &self.path)?;
277        if !self.headers.is_empty() && !self.is_http() {
278            return Err(FaucetError::Config(
279                "file source: `headers` apply only to an http(s):// path".into(),
280            ));
281        }
282        if self.max_object_bytes == 0 {
283            return Err(FaucetError::Config(
284                "file source: `max_object_bytes` must be at least 1".into(),
285            ));
286        }
287        if self.http_connect_timeout_secs == 0 || self.http_read_timeout_secs == 0 {
288            return Err(FaucetError::Config(
289                "file source: `http_connect_timeout_secs` and `http_read_timeout_secs` must be \
290                 at least 1"
291                    .into(),
292            ));
293        }
294        for (k, v) in &self.headers {
295            reqwest::header::HeaderName::from_bytes(k.as_bytes())
296                .map_err(|e| FaucetError::Config(format!("file source: header name {k:?}: {e}")))?;
297            reqwest::header::HeaderValue::from_str(v)
298                .map_err(|e| FaucetError::Config(format!("file source: header {k:?}: {e}")))?;
299        }
300        #[cfg(feature = "file-format-avro")]
301        self.avro.parsed_schema()?;
302        self.csv.validate()?;
303        self.parquet.validate()?;
304        #[cfg(feature = "encryption")]
305        if let Some(spec) = &self.encryption {
306            faucet_core::CompiledEncryption::compile(spec)?;
307        }
308        Ok(())
309    }
310}
311
312#[cfg(test)]
313mod tests {
314    use super::*;
315
316    #[test]
317    fn decoded_bytes_default_to_a_multiple_of_the_object_limit() {
318        let mut c = FileSourceConfig::new("x.avro");
319        c.max_object_bytes = 1024;
320        assert_eq!(c.decoded_bytes_limit(), 64 * 1024);
321        c.max_object_bytes = faucet_core::file_format::DEFAULT_MAX_OBJECT_BYTES;
322        assert_eq!(
323            c.decoded_bytes_limit(),
324            faucet_core::file_format::budget::MAX_DEFAULT_DECODED_BYTES
325        );
326        c.max_decoded_bytes = Some(10);
327        assert_eq!(c.decoded_bytes_limit(), 10);
328    }
329    use serde_json::json;
330
331    #[test]
332    fn defaults_and_builders() {
333        let c = FileSourceConfig::new("data/")
334            .format(FileSourceFormat::Csv)
335            .recursive(true)
336            .with_batch_size(10)
337            .incremental(IncrementalBy::Name);
338        assert_eq!(c.format.explicit(), Some(faucet_core::FileFormat::Csv));
339        assert!(c.recursive);
340        assert_eq!(c.batch_size, 10);
341        assert_eq!(c.incremental.unwrap().by, IncrementalBy::Name);
342        assert!(!c.is_http());
343        assert!(FileSourceConfig::new("HTTPS://x/y.csv").is_http());
344        c.validate().expect("valid");
345    }
346
347    #[test]
348    fn deserializes_with_option_blocks_and_refuses_unknown_keys() {
349        let c: FileSourceConfig = serde_json::from_value(json!({
350            "path": "in/*.csv",
351            "csv": {"delimiter": ";"},
352            "orc": {"columns": ["a"]},
353            "incremental": {"by": "mtime"},
354            "stable_for_secs": 5
355        }))
356        .expect("config");
357        assert_eq!(c.format_options().csv.delimiter, ";");
358        assert_eq!(c.format_options().orc.columns, Some(vec!["a".to_string()]));
359        assert_eq!(c.incremental.unwrap().by, IncrementalBy::Mtime);
360        assert!(
361            serde_json::from_value::<FileSourceConfig>(json!({"path": "x", "nope": 1})).is_err()
362        );
363    }
364
365    #[test]
366    fn validation_refuses_bad_configs() {
367        assert!(FileSourceConfig::new(" ").validate().is_err());
368        assert!(
369            FileSourceConfig::new("x")
370                .with_batch_size(usize::MAX)
371                .validate()
372                .is_err()
373        );
374        let mut c = FileSourceConfig::new("local.csv");
375        c.headers.insert("Authorization".into(), "x".into());
376        assert!(c.validate().is_err(), "headers need an http path");
377        let mut c = FileSourceConfig::new("https://h/x.csv");
378        c.headers.insert("bad header".into(), "x".into());
379        assert!(c.validate().is_err());
380        let mut c = FileSourceConfig::new("https://h/x.csv");
381        c.headers.insert("X-Ok".into(), "line\nbreak".into());
382        assert!(c.validate().is_err());
383        #[cfg(feature = "file-format-avro")]
384        {
385            let mut c = FileSourceConfig::new("x.avro");
386            c.avro.schema = Some(json!({"type": "nope"}));
387            assert!(c.validate().is_err());
388        }
389    }
390}
faucet_source_file/config.rs

faucet_source_file/
config.rs
