faucet_common_file/
lib.rs

1#![cfg_attr(docsrs, feature(doc_cfg))]
2//! Config types shared by the local file source and sink, and the shared
3//! file-writing layer ([`write`](mod@write)) every file-writing sink builds on.
4//!
5//! **Experimental** (PRINCIPLES.md §3): this block's shape may change in a
6//! minor release; any change is called out in the changelog.
7
8use faucet_core::compression::{Compression, CompressionConfig};
9use faucet_core::{FaucetError, FileFormat};
10use schemars::JsonSchema;
11use serde::{Deserialize, Serialize};
12
13pub mod capabilities;
14#[cfg(feature = "encryption")]
15pub mod sealed_lines;
16pub mod write;
17
18#[doc(hidden)]
19pub mod __private {
20    pub use faucet_core::async_trait;
21    pub use faucet_core::check::{CheckContext, CheckReport};
22    pub use faucet_core::forward_sink;
23    pub use faucet_core::io::CallContext;
24    pub use faucet_core::observability::RoundtripRecorder;
25    pub use faucet_core::{
26        BatchAtomicity, Capabilities, FaucetError, LocalOutput, Sink, WriteMode,
27    };
28    pub use serde_json::Value;
29}
30
31/// Which format a file is in: resolved from its extension (`auto`) or fixed.
32#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
33#[serde(rename_all = "snake_case")]
34pub enum FileFormatChoice {
35    /// Resolve per file from its extension, looking through a compression
36    /// suffix (`.csv.gz` is CSV): `.jsonl`/`.ndjson`, `.json`, `.csv`, `.xml`,
37    /// `.xlsx`, `.parquet`, `.avro`, `.orc`, `.txt`.
38    #[default]
39    Auto,
40    /// One JSON value per line.
41    JsonLines,
42    /// A JSON array (a lone object is one record).
43    JsonArray,
44    /// Delimited text; dialect from `csv`.
45    Csv,
46    /// XML; record framing from `xml`.
47    Xml,
48    /// An Excel workbook; sheet from `excel`.
49    Xlsx,
50    /// Apache Parquet.
51    Parquet,
52    /// Unparsed text, one record per file.
53    RawText,
54    /// Apache Avro Object Container File; schema and codec from `avro`.
55    Avro,
56    /// Apache ORC (read-only); projection from `orc`.
57    Orc,
58}
59
60impl FileFormatChoice {
61    /// The explicit format, or `None` for `auto`.
62    pub fn explicit(self) -> Option<FileFormat> {
63        Some(match self {
64            Self::Auto => return None,
65            Self::JsonLines => FileFormat::JsonLines,
66            Self::JsonArray => FileFormat::JsonArray,
67            Self::Csv => FileFormat::Csv,
68            Self::Xml => FileFormat::Xml,
69            Self::Xlsx => FileFormat::Xlsx,
70            Self::Parquet => FileFormat::Parquet,
71            Self::RawText => FileFormat::RawText,
72            Self::Avro => FileFormat::Avro,
73            Self::Orc => FileFormat::Orc,
74        })
75    }
76
77    /// The format of the file called `name` (a path or a URL's last segment).
78    ///
79    /// `Ok(None)` when `auto` finds no format and `strict` is off — the caller
80    /// skips the file; with `strict` it is an error naming the file.
81    pub fn resolve(self, name: &str, strict: bool) -> Result<Option<FileFormat>, FaucetError> {
82        if let Some(f) = self.explicit() {
83            return Ok(Some(f));
84        }
85        match FileFormat::from_path(name) {
86            Some(f) => Ok(Some(f)),
87            None if strict => Err(FaucetError::Config(format!(
88                "'{name}' has no recognised extension and `strict` is on; set `format`"
89            ))),
90            None => Ok(None),
91        }
92    }
93}
94
95/// Extensions that name a format a file sink can write, for error messages.
96pub const WRITABLE_EXTENSIONS: &str = ".jsonl .json .csv .xml .xlsx .avro .parquet .txt";
97
98impl FileFormatChoice {
99    /// The format a file called `name` is **written** in: the explicit format,
100    /// or the one its extension names. Refuses a name with no recognised
101    /// extension and ORC, which has no writer.
102    pub fn resolve_writable(self, name: &str) -> Result<FileFormat, FaucetError> {
103        let format = self.resolve(name, false)?.ok_or_else(|| {
104            FaucetError::Config(format!(
105                "'{name}' has no extension naming a writable format — use one of \
106                 {WRITABLE_EXTENSIONS} (optionally + .gz/.zst) or set `format`"
107            ))
108        })?;
109        if !format.is_writable() && format != FileFormat::Parquet {
110            return Err(FaucetError::Config(format!(
111                "`{}` is read-only — there is no {} writer; write Parquet for a columnar output",
112                format.as_str(),
113                format.as_str().to_ascii_uppercase()
114            )));
115        }
116        Ok(format)
117    }
118}
119
120/// The compression codec for the file at `path` (a local path or a URL),
121/// resolved from its suffix under `auto`.
122pub fn resolve_compression(config: CompressionConfig, path: &str) -> Compression {
123    config.resolve(resolution_name(path))
124}
125
126/// Formats that compress internally (Parquet column chunks, Avro blocks, the
127/// xlsx zip container), so file-level compression does not apply on write.
128pub fn compresses_internally(format: FileFormat) -> bool {
129    matches!(
130        format,
131        FileFormat::Parquet | FileFormat::Avro | FileFormat::Xlsx
132    )
133}
134
135/// Formats a writer can add records to without rewriting the file.
136pub fn appendable(format: FileFormat) -> bool {
137    matches!(
138        format,
139        FileFormat::JsonLines | FileFormat::Csv | FileFormat::RawText
140    )
141}
142
143/// Whether `path` names a directory (ends in a separator) rather than a file.
144pub fn is_directory_path(path: &str) -> bool {
145    path.ends_with('/') || path.ends_with('\\')
146}
147
148/// Refuse an empty or blank `path`, naming the connector.
149pub fn require_path(connector: &str, path: &str) -> Result<(), FaucetError> {
150    if path.trim().is_empty() {
151        return Err(FaucetError::Config(format!("{connector}: `path` is empty")));
152    }
153    Ok(())
154}
155
156/// Prefix a [`FaucetError::Config`] message with `connector`; any other
157/// error passes through unchanged.
158pub fn config_context(connector: &str, e: FaucetError) -> FaucetError {
159    match e {
160        FaucetError::Config(m) => FaucetError::Config(format!("{connector}: {m}")),
161        other => other,
162    }
163}
164
165/// Whether `path` is an `http://` or `https://` URL.
166pub fn is_http_path(path: &str) -> bool {
167    let p = path.trim_start().to_ascii_lowercase();
168    p.starts_with("http://") || p.starts_with("https://")
169}
170
171/// The last path segment of a URL, without its query or fragment — the part
172/// that names the file for format and compression resolution.
173pub fn url_file_name(url: &str) -> &str {
174    let no_query = url.split(['?', '#']).next().unwrap_or(url);
175    no_query.rsplit('/').next().unwrap_or(no_query)
176}
177
178/// The name format and compression are resolved from: the URL's file name
179/// for an `http(s)://` path, the path itself otherwise.
180pub fn resolution_name(path: &str) -> &str {
181    if is_http_path(path) {
182        url_file_name(path)
183    } else {
184        path
185    }
186}
187
188/// Which listed objects an object-store source reads.
189///
190/// A folder marker (a zero-byte key ending in `/`) is never data. Without an
191/// `include` glob, a key with a path segment below the prefix that starts
192/// with `_` or `.` — `_SUCCESS`, `_temporary/…`, `.crc` files — is skipped
193/// too; with one, exactly the keys the glob matches are read.
194#[derive(Debug, Clone, Default)]
195pub struct ObjectFilter {
196    include: Option<glob::Pattern>,
197}
198
199impl ObjectFilter {
200    /// Compile the optional `include` glob (matched against the whole key).
201    pub fn new(include: Option<&str>) -> Result<Self, FaucetError> {
202        let include = include
203            .map(|g| {
204                glob::Pattern::new(g)
205                    .map_err(|e| FaucetError::Config(format!("`include` glob {g:?}: {e}")))
206            })
207            .transpose()?;
208        Ok(Self { include })
209    }
210
211    /// Whether to read `key` (`size` in bytes, when the listing reports it)
212    /// listed under `prefix`.
213    pub fn keep(&self, key: &str, size: Option<u64>, prefix: &str) -> bool {
214        if (key.ends_with('/') || key.ends_with("_$folder$")) && size.unwrap_or(0) == 0 {
215            return false;
216        }
217        match &self.include {
218            Some(p) => p.matches_with(
219                key,
220                glob::MatchOptions {
221                    case_sensitive: true,
222                    require_literal_separator: false,
223                    require_literal_leading_dot: false,
224                },
225            ),
226            None => !is_hidden_object(key, prefix),
227        }
228    }
229}
230
231/// Whether a path segment of `key` below `prefix` starts with `_` or `.`.
232/// Only what follows the whole prefix is judged, so `prefix: logs/_inbound`
233/// still reads `logs/_inbound-2026.jsonl` and `logs/_inbound/part-0.jsonl`.
234pub fn is_hidden_object(key: &str, prefix: &str) -> bool {
235    let hidden = |s: &str| s.starts_with('_') || s.starts_with('.');
236    let Some(rest) = key.strip_prefix(prefix) else {
237        return key.split('/').any(hidden);
238    };
239    let mut segments = rest.split('/');
240    if !prefix.is_empty() && !prefix.ends_with('/') {
241        // The rest of a segment the prefix names in part is the caller's choice.
242        segments.next();
243    }
244    segments.any(hidden)
245}
246
247#[cfg(test)]
248mod tests {
249    use super::*;
250
251    #[test]
252    fn markers_and_hidden_objects_are_skipped_unless_included() {
253        let f = ObjectFilter::default();
254        assert!(f.keep("out/part-0001.parquet", Some(10), "out/"));
255        assert!(!f.keep("out/_SUCCESS", Some(0), "out/"));
256        assert!(!f.keep("out/day=1_$folder$", Some(0), "out/"));
257        assert!(f.keep("out/day=1_$folder$", Some(3), "out/"));
258        assert!(!f.keep("out/.part-0001.parquet.crc", Some(8), "out/"));
259        assert!(!f.keep("out/_temporary/0/part-1.json", Some(8), "out/"));
260        assert!(!f.keep("out/sub/", Some(0), "out/"));
261        assert!(!f.keep("out/sub/", None, "out/"));
262        assert!(
263            f.keep("_raw/data.json", Some(4), "_raw/"),
264            "the prefix itself is not judged"
265        );
266        assert!(f.keep("_raw/part-1.json", Some(4), "_raw/part"));
267        assert!(!f.keep("data/_x.json", Some(4), ""));
268        assert!(f.keep("exports/_daily-2026-10-01.jsonl", Some(4), "exports/_daily"));
269        assert!(f.keep("exports/_daily/part-0.jsonl", Some(4), "exports/_daily"));
270        assert!(!f.keep("exports/_daily/_tmp/x.jsonl", Some(4), "exports/_daily"));
271        assert!(!f.keep("out/pa_x/_SUCCESS", Some(4), "out/pa"));
272        assert!(!f.keep("elsewhere/_x.json", Some(4), "out/"));
273        let inc = ObjectFilter::new(Some("out/_*")).unwrap();
274        assert!(inc.keep("out/_metadata", Some(4), "out/"));
275        assert!(!inc.keep("out/part-1.parquet", Some(4), "out/"));
276        assert!(ObjectFilter::new(Some("[")).is_err());
277    }
278
279    #[test]
280    fn every_choice_maps_onto_the_shared_vocabulary() {
281        let all = [
282            (FileFormatChoice::JsonLines, FileFormat::JsonLines),
283            (FileFormatChoice::JsonArray, FileFormat::JsonArray),
284            (FileFormatChoice::Csv, FileFormat::Csv),
285            (FileFormatChoice::Xml, FileFormat::Xml),
286            (FileFormatChoice::Xlsx, FileFormat::Xlsx),
287            (FileFormatChoice::Parquet, FileFormat::Parquet),
288            (FileFormatChoice::RawText, FileFormat::RawText),
289            (FileFormatChoice::Avro, FileFormat::Avro),
290            (FileFormatChoice::Orc, FileFormat::Orc),
291        ];
292        for (c, f) in all {
293            assert_eq!(c.explicit(), Some(f));
294            assert_eq!(c.resolve("x.unknown", true).unwrap(), Some(f));
295        }
296        assert_eq!(FileFormatChoice::default().explicit(), None);
297    }
298
299    #[test]
300    fn auto_resolves_by_extension_and_strict_refuses_unknowns() {
301        let auto = FileFormatChoice::Auto;
302        assert_eq!(
303            auto.resolve("a/b.csv.gz", false).unwrap(),
304            Some(FileFormat::Csv)
305        );
306        assert_eq!(auto.resolve("notes.md", false).unwrap(), None);
307        let err = auto.resolve("notes.md", true).unwrap_err();
308        assert!(err.to_string().contains("notes.md"), "{err}");
309    }
310
311    #[test]
312    fn urls_and_paths() {
313        assert!(is_http_path("HTTPS://h/x") && is_http_path("http://h"));
314        assert!(!is_http_path("/tmp/x") && !is_http_path("ftp://h"));
315        assert_eq!(url_file_name("https://h/a/b.csv.gz?sig=1#x"), "b.csv.gz");
316        assert_eq!(url_file_name("https://h/"), "");
317        assert_eq!(resolution_name("https://h/a.avro?x=1"), "a.avro");
318        assert_eq!(resolution_name("dir/a.avro"), "dir/a.avro");
319    }
320
321    #[test]
322    fn writable_resolution_refuses_unknown_extensions_and_orc() {
323        let auto = FileFormatChoice::Auto;
324        assert_eq!(auto.resolve_writable("a.csv.gz").unwrap(), FileFormat::Csv);
325        assert_eq!(
326            auto.resolve_writable("a.parquet").unwrap(),
327            FileFormat::Parquet
328        );
329        let e = auto.resolve_writable("a.dat").unwrap_err().to_string();
330        assert!(e.contains("a.dat") && e.contains(".jsonl"), "{e}");
331        let e = auto.resolve_writable("a.orc").unwrap_err().to_string();
332        assert!(e.contains("read-only"), "{e}");
333        let e = FileFormatChoice::Orc
334            .resolve_writable("x")
335            .unwrap_err()
336            .to_string();
337        assert!(e.contains("ORC"), "{e}");
338        assert_eq!(
339            FileFormatChoice::Xml.resolve_writable("a.dat").unwrap(),
340            FileFormat::Xml
341        );
342    }
343
344    #[test]
345    fn compression_and_format_properties() {
346        assert_eq!(
347            resolve_compression(CompressionConfig::Auto, "a.jsonl.gz"),
348            Compression::Gzip
349        );
350        assert_eq!(
351            resolve_compression(CompressionConfig::Auto, "https://h/a.csv.zst?x=1"),
352            Compression::Zstd
353        );
354        assert_eq!(
355            resolve_compression(CompressionConfig::Auto, "a.csv"),
356            Compression::None
357        );
358        assert_eq!(
359            resolve_compression(CompressionConfig::Gzip, "a.csv"),
360            Compression::Gzip
361        );
362        for f in [FileFormat::Parquet, FileFormat::Avro, FileFormat::Xlsx] {
363            assert!(compresses_internally(f) && !appendable(f), "{f:?}");
364        }
365        for f in [FileFormat::JsonLines, FileFormat::Csv, FileFormat::RawText] {
366            assert!(appendable(f) && !compresses_internally(f), "{f:?}");
367        }
368        assert!(!appendable(FileFormat::JsonArray) && !compresses_internally(FileFormat::Xml));
369    }
370
371    #[test]
372    fn paths_directories_and_emptiness() {
373        assert!(is_directory_path("out/") && is_directory_path("out\\"));
374        assert!(!is_directory_path("out/a.jsonl"));
375        assert!(require_path("file sink", "a").is_ok());
376        let e = require_path("file sink", "  ").unwrap_err().to_string();
377        assert!(e.contains("file sink") && e.contains("empty"), "{e}");
378    }
379
380    #[test]
381    fn serde_names_are_snake_case() {
382        let c: FileFormatChoice = serde_json::from_str("\"json_lines\"").unwrap();
383        assert_eq!(c, FileFormatChoice::JsonLines);
384        assert_eq!(
385            serde_json::to_string(&FileFormatChoice::Auto).unwrap(),
386            "\"auto\""
387        );
388    }
389
390    #[test]
391    fn config_context_prefixes_only_config_errors() {
392        assert_eq!(
393            config_context("S3 sink", FaucetError::Config("bad".into())).to_string(),
394            FaucetError::Config("S3 sink: bad".into()).to_string()
395        );
396        assert!(matches!(
397            config_context("S3 sink", FaucetError::Sink("io".into())),
398            FaucetError::Sink(m) if m == "io"
399        ));
400    }
401}
faucet_common_file/lib.rs

faucet_common_file/
lib.rs
