faucet-source-file

The general local file source for faucet-stream: read JSON Lines, JSON, CSV, Excel, XML, Parquet, Avro and ORC from a file, a directory, a glob, or a single http:// / https:// URL. It is the local counterpart of the S3, GCS and Azure Blob sources and uses the same shared format layer, so a file reads the same here as it does from a bucket.

faucet-source-csv stays as it was for existing configs. Use this crate for anything new, CSV included.

Experimental (PRINCIPLES.md §3): this block's shape may change in a minor release; any change is called out in the changelog.

[dependencies]
faucet-source-file = { version = "1.0.0", features = ["file-formats"] }

source:
  type: file
  config:
    path: ./exports            # a file, a directory, a glob, or an http(s) URL
    recursive: true
    format: auto               # resolved per file from the extension
    compression: auto          # .gz / .zst resolved per file
    incremental: { by: mtime } # re-runs read only new files (needs `state:`)
    stable_for_secs: 30        # skip files still being written

Paths

Every listing is sorted by path, so runs are deterministic. Symlinks are followed; a recursive walk lists each directory once, under its real path when it is also reachable without the link (so bookmarks, which key on the path, never switch alias). An unreadable file or directory fails the run with its path.

When the host scopes a reference policy with path roots over the run (as faucet serve does for a config it did not load itself), each listed file is resolved and must lie under a root, and it is then opened through directories held open without following links, so a link put in place after the check fails the read instead of being followed.

A directory or glob never reads a file sink's unfinished output: scratch files (*.faucet-tmp, *.faucet-tmp-body, -old, -seal, -prev, -upload-<id>) and anything inside the .faucet-overwrite-* swap directory of an overwrite run that has not committed are skipped. A path that names one file directly is read as given.

{key} placeholders in path are filled from the fetch context (a library caller's fetch_with_context), as the csv and parquet sources do.

Formats

format: auto (the default) picks a format per file from its extension, looking through a compression suffix, so export.csv.gz is gzip-compressed CSV:

With format: auto, a file whose extension names no format is skipped with a warning. strict: true fails the run on such a file instead. An explicit format: applies to every file whatever its name.

A directory mixing formats reads in one run. Avro files are resolved against the first Avro file's schema, or against avro.schema when you set it. A later file whose schema cannot be resolved against it fails with an error naming both files. Parquet and ORC files must share one schema, and a mismatch likewise names both files. For local Parquet files the schemas are compared from the footers before any row is read, so a mismatch in a late file never leaves earlier files half-delivered.

CSV

Values are strings; cast them with a cast transform.

Parquet

parquet.columns projects the read — the same parquet: block as the S3, GCS and Azure Blob sources: only those column chunks are decoded, on both the row and the columnar path, and a name a file does not have fails naming the columns it does have. An empty list is refused at load. Nulls are explicit: a null column is read as "key": null (the parquet source omitted the key).

Compression (compression: auto | gzip | zstd | none) resolves per file from its suffix. A compressed Parquet, Avro or ORC file is decompressed into memory before decoding, because those formats need random access or a whole stream.

Encryption

encryption: { key, previous_keys } (feature encryption) reads files the file or jsonl sink encrypted: a file sealed whole is decrypted and then decompressed; JSON Lines or raw text sealed line by line is read whole and then decrypted a line at a time, its header and trailer verified: a line dropped, duplicated, reordered or moved in from another file, or a file cut short anywhere but at an earlier flush's trailer, fails the read. A sealed file written before the header and trailer existed is read line by line with a warning that no whole-file check applies. A file or line that is not sealed fails the read rather than being trusted as plaintext.

Coming from the csv or parquet source

Golden tests (crates/interop-tests/tests/file_source_parity.rs) read the same fixtures through the old sources and the file source and compare the records.

Columnar path

With an explicit format: avro, orc or parquet, the source advertises the Arrow columnar path. A pipeline into a columnar sink (Parquet, Delta and others) then moves RecordBatches and never builds JSON rows. Avro logical types arrive typed: decimal as Decimal128, date as Date32, and timestamps as Timestamp. format: auto stays on the row path, because it cannot promise that every file decodes to Arrow.

When Arrow data becomes JSON rows (a row-path sink, or a transform or governance pass), decimals arrive as exact strings ("12.50") so no digits are lost, non-finite floats as a null counted in the run summary (non_finite), binary columns as base64, and timestamps at their unit's precision (the canonical renderings of value fidelity).

Incremental mode

incremental: { by: mtime }   # or: { by: name }

• by: mtime reads files modified after the newest file the previous run read. Ties are broken by path, so a second file with the same timestamp is still new. A rewritten file is read again.
• by: name reads files whose path sorts after the last one read. Use it for dated names that are never rewritten.
• Producers that preserve timestamps. by: mtime is a watermark: a file that lands after a newer file was read but keeps an older modification time — moved in with mv, copied with cp -p or rsync -t, extracted from an archive — sits below the watermark and is never read. The run logs a warning naming such files (it sees their inode change time move past the watermark). Have the producer touch files as it publishes them, or use by: name with names that sort in arrival order.

The bookmark advances after each file, and pages never span two files in this mode, so an interrupted run resumes at the next unread file. It needs a pipeline state: block; for library use the key is file:<hash of path>. Over HTTP, the Last-Modified header is the modification time. A server that sends none cannot be used with by: mtime or stable_for_secs, and the run says so.

stable_for_secs: N skips files modified in the last N seconds, leaving them for a later run once they settle. With incremental: {by: name} a run also stops before the first unsettled file in name order: reading a later name would move the bookmark past it and skip it for good, so the files after it wait for the same later run.

Formats read whole into memory (JSON arrays, XML, Excel, and Avro/ORC/Parquet when compressed, remote or encrypted) stop at max_object_bytes (default 2 GiB) once decompressed: a larger file — or a decompression bomb — fails with an error naming the file instead of exhausting memory. Streamed JSON Lines and CSV files have no whole-file limit, but one line may not exceed max_object_bytes: a file (or http(s):// body) with no newline fails the same way instead of being buffered whole.

max_decoded_bytes (default 64 × max_object_bytes, at most 16 GiB) bounds what a file may decode to where a format can expand past its input: an Avro file's records (per batch, since Avro streams) and an Excel workbook's parts and shared strings. A file past it fails with an error instead of exhausting memory.

HTTP

A URL path is one file. headers: are sent with every request, for example Authorization: "Bearer ${env:TOKEN}". Connection errors, 429 and 5xx responses are retried up to http_retries times (default 3) with exponential backoff that honours Retry-After. Any other status fails the run with the status and the response body. A connection must open within http_connect_timeout_secs (default 30) and the server must keep sending — headers or body — at least once every http_read_timeout_secs (default 300): a stall before the body starts is retried like a connection error, one mid-body fails the file, so a server that accepts and never answers cannot hang the run. Requests are counted as get / head round trips on the pipeline's recorder, and 429s, retries and rate-limit waits feed the run's throttling metrics.

Sharding and discovery

Local paths are shardable by a hash of the file path (shard: { count: N } in faucet serve cluster mode): each worker reads a disjoint set of files. faucet discover lists each readable file as a dataset of kind file, with a {path, format} config patch. No file is opened during discovery.

Features

JSON Lines, JSON arrays and raw text are always available. The other formats are behind features: file-format-csv, -xml, -excel, -avro, -orc and -parquet, plus arrow for the columnar path. The file-formats feature turns them all on. A build that lacks a format names the missing feature instead of misreading the file. encryption enables the encryption block.

Library

use faucet_core::Source;
use faucet_source_file::{FileSource, FileSourceConfig, IncrementalBy};

# async fn run() -> Result<(), faucet_core::FaucetError> {
let source = FileSource::new(
    FileSourceConfig::new("./exports").recursive(true).incremental(IncrementalBy::Mtime),
)?;
let records = source.fetch_all().await?;

// One call, every default:
let rows = faucet_source_file::read_records("./out/orders.jsonl").await?;
# Ok(()) }

Capabilities

Discovers one dataset per file; shardable except for an http(s):// path; columnar (with arrow) only for an explicit format: avro, orc or parquet. Not exactly-once. faucet list --json prints the declaration.

Timeouts, cancellation and retries

An http(s):// path is fetched through the pipeline's CallContext (#880): the run's cancel token stops an in-flight request and the backoff between retries, and any caller deadline ends both. http_retries, the connect / read timeouts and the get / head round-trip and throttle metrics are unchanged. Local paths make no backend calls.