# faucet-source-file
The general local file source for [faucet-stream](https://github.com/faucet-hq/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.
```toml
[dependencies]
faucet-source-file = { version = "1.0.0", features = ["file-formats"] }
```
```yaml
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
| `data/orders.jsonl` | that file |
| `data/` | the directory's regular files; subdirectories too with `recursive: true` |
| `data/**/*.csv.gz` | every file the glob matches |
| `https://example.com/export.csv?sig=…` | one remote file, fetched with `reqwest` and streamed |
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:
| `.jsonl`, `.ndjson` | JSON Lines | streamed line by line |
| `.json` | JSON array (a lone object is one record) | whole file |
| `.csv` | CSV (dialect from `csv:`) | streamed row by row |
| `.xml` | XML (framing from `xml:`) | whole file |
| `.xlsx` | Excel (sheet from `excel:`) | whole file |
| `.parquet` | Parquet (projection from `parquet:`) | streamed by row group |
| `.avro` | Avro Object Container File (reader schema from `avro:`) | streamed by block |
| `.orc` | ORC (projection from `orc:`) | streamed by stripe |
| `.txt` | raw text: one record `{path, content}` per file | whole file |
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
| `delimiter` | `","` | One byte; `"\t"` for tabs. |
| `has_headers` | `true` | The first row names the fields; otherwise `column_0`, `column_1`, …. A repeated header name fails the read (it would drop a column). |
| `quote` | `"\""` | The quote character. |
| `flexible` | `false` | Accept rows with more or fewer fields than the header. Off, the first ragged row fails the read naming its line. |
| `null_values` | `[]` | Cell values read as `null` (e.g. `["", "NULL"]`). |
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
| csv `path`, `has_headers`, `delimiter`, `quote`, `flexible`, `null_values`, `batch_size`, `compression` | `path`, `csv.has_headers`, `csv.delimiter`, `csv.quote`, `csv.flexible`, `csv.null_values`, `batch_size`, `compression` |
| parquet `source: {type: local_path, path}` / `{type: glob, pattern}` | `path` (a file, directory or glob) |
| parquet `columns`, `batch_size`, `concurrency` | `parquet.columns`, `batch_size`, `concurrency` |
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 `RecordBatch`es 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](https://faucet-hq.github.io/faucet-stream/reference/value-fidelity.html)).
## Incremental mode
```yaml
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
```rust,no_run
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.