faucet_core/file_format/
mod.rs

1//! One file-format vocabulary for every file and object-store connector (#604).
2//!
3//! S3, GCS, Azure Blob and SFTP all move *files*, and the files they move are
4//! CSV exports, JSON dumps, XML feeds and Excel workbooks as often as they are
5//! JSON Lines. Before this module each connector carried its own small format
6//! enum — `JsonLines | JsonArray | RawText | Parquet` on the source side, and
7//! only `JsonLines | Parquet` on the sink side — so what you could read depended
8//! on which bucket it was in, and what you could write was a strict subset of
9//! what you could read. The gap was filled by pre- and post-processing outside
10//! the pipeline, which is exactly the work a movement engine exists to absorb.
11//!
12//! The parsers themselves are not new: the REST source has decoded
13//! `json | csv | xlsx | xml` since #497/#515. This module is where they move so
14//! that every connector — including a third-party one, which depends only on
15//! `faucet-core` — gets the same set, with the same option names and the same
16//! edge-case behaviour.
17//!
18//! # Shape
19//!
20//! ```yaml
21//! source:
22//!   type: s3
23//!   config:
24//!     bucket: exports
25//!     prefix: daily/
26//!     format: csv                  # FileFormat
27//!     compression: auto            # faucet_core::compression
28//!     csv: { has_headers: true, delimiter: "," }
29//! sink:
30//!   type: s3
31//!   config:
32//!     bucket: reports
33//!     format: xlsx
34//!     compression: gzip
35//! ```
36//!
37//! # What is deliberately not here
38//!
39//! **Parquet.** It is columnar, self-describing, and already has a dedicated
40//! Arrow read/write path per connector that must not be routed through a
41//! `Vec<Value>`. [`FileFormat::Parquet`] therefore names the format for config
42//! purposes but [`decode`]/[`encode`] refuse it, so a caller cannot silently
43//! lose the columnar fast path by going through the generic helper.
44//!
45//! **Compression.** [`crate::compression`] already owns codec selection,
46//! extension-based `auto` resolution, and the mismatch warning. Format and
47//! codec compose; neither needs to know about the other.
48
49use crate::error::FaucetError;
50use schemars::JsonSchema;
51use serde::{Deserialize, Serialize};
52use serde_json::Value;
53
54#[cfg(feature = "file-format-avro")]
55pub mod avro;
56pub mod container;
57#[cfg(feature = "file-format-csv")]
58pub mod csv;
59#[cfg(feature = "file-format-excel")]
60pub mod excel;
61pub mod json;
62#[cfg(feature = "file-format-orc")]
63pub mod orc;
64pub mod parquet_io;
65#[cfg(feature = "file-format-xml")]
66pub mod xml;
67
68pub use container::{ContainerDecoder, FileInput};
69
70/// How the bytes of one object are laid out.
71///
72/// The variants are the union of what the file connectors could previously read
73/// *or* write, so adopting this enum never removes a format from a connector
74/// that had it.
75#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
76#[serde(rename_all = "snake_case")]
77pub enum FileFormat {
78    /// One JSON value per line (NDJSON). The default everywhere, because it is
79    /// the only format that both streams and round-trips arbitrary JSON.
80    #[default]
81    JsonLines,
82    /// A single JSON array holding every record.
83    JsonArray,
84    /// Delimited text. See [`CsvOptions`].
85    Csv,
86    /// XML. See [`XmlOptions`].
87    Xml,
88    /// An Excel workbook (`.xlsx`). See [`ExcelOptions`].
89    Xlsx,
90    /// Apache Parquet — named for config, handled by each connector's own
91    /// Arrow path, never by [`decode`]/[`encode`].
92    Parquet,
93    /// Unparsed text: one record per object, `{"text": "<whole body>"}`.
94    RawText,
95    /// Apache Avro Object Container File. See [`AvroOptions`]. Read and
96    /// write; decodes to records or, with `arrow`, straight to Arrow batches.
97    Avro,
98    /// Apache ORC. See [`OrcOptions`]. **Read-only** — see the `orc` module
99    /// for why there is no writer.
100    Orc,
101}
102
103impl FileFormat {
104    /// The conventional file extension, used to name objects a sink creates
105    /// when the user did not pick one.
106    pub fn extension(self) -> &'static str {
107        match self {
108            Self::JsonLines => ".jsonl",
109            Self::JsonArray => ".json",
110            Self::Csv => ".csv",
111            Self::Xml => ".xml",
112            Self::Xlsx => ".xlsx",
113            Self::Parquet => ".parquet",
114            Self::RawText => ".txt",
115            Self::Avro => ".avro",
116            Self::Orc => ".orc",
117        }
118    }
119
120    /// Whether the whole object must be in memory to produce the first record.
121    ///
122    /// Callers use this to decide between a streaming read and a buffered one.
123    /// It is a property of the format, not of the source: a JSON array has no
124    /// record boundary until it is parsed, a zip-container workbook has its
125    /// directory at the end, and an XML document is a tree.
126    pub fn requires_whole_object(self) -> bool {
127        matches!(self, Self::JsonArray | Self::Xml | Self::Xlsx | Self::Orc)
128    }
129
130    /// Whether the format is a self-describing binary container decoded by
131    /// [`ContainerDecoder`] (Avro, ORC) rather than by [`decode`] per object.
132    pub fn is_container(self) -> bool {
133        matches!(self, Self::Avro | Self::Orc)
134    }
135
136    /// Whether a sink can write this format through [`encode`].
137    pub fn is_writable(self) -> bool {
138        !matches!(self, Self::Orc | Self::Parquet)
139    }
140
141    /// The format a file name's extension names, looking through a trailing
142    /// compression suffix (`.gz`, `.gzip`, `.zst`, `.zstd`), so
143    /// `export.csv.gz` is CSV. `None` for an extension no format claims.
144    ///
145    /// `.json` is a JSON array (a lone object is one record); `.jsonl` and
146    /// `.ndjson` are JSON Lines; `.tsv` is not claimed, because its delimiter
147    /// is an option rather than a format.
148    pub fn from_path(path: &str) -> Option<Self> {
149        let name = path
150            .rsplit(['/', '\\'])
151            .next()
152            .unwrap_or(path)
153            .to_ascii_lowercase();
154        let stem = ["gz", "gzip", "zst", "zstd"]
155            .iter()
156            .find_map(|c| name.strip_suffix(&format!(".{c}")))
157            .unwrap_or(&name);
158        let ext = stem.rsplit_once('.')?.1;
159        Some(match ext {
160            "jsonl" | "ndjson" => Self::JsonLines,
161            "json" => Self::JsonArray,
162            "csv" => Self::Csv,
163            "xml" => Self::Xml,
164            "xlsx" => Self::Xlsx,
165            "parquet" => Self::Parquet,
166            "txt" => Self::RawText,
167            "avro" => Self::Avro,
168            "orc" => Self::Orc,
169            _ => return None,
170        })
171    }
172
173    /// The name used in config and error messages.
174    pub fn as_str(self) -> &'static str {
175        match self {
176            Self::JsonLines => "json_lines",
177            Self::JsonArray => "json_array",
178            Self::Csv => "csv",
179            Self::Xml => "xml",
180            Self::Xlsx => "xlsx",
181            Self::Parquet => "parquet",
182            Self::RawText => "raw_text",
183            Self::Avro => "avro",
184            Self::Orc => "orc",
185        }
186    }
187}
188
189fn default_true() -> bool {
190    true
191}
192
193fn default_delimiter() -> String {
194    ",".into()
195}
196
197/// CSV dialect.
198#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
199#[serde(deny_unknown_fields)]
200#[schemars(extend("x-faucet-aliases" = ["write_headers"]))]
201pub struct CsvOptions {
202    /// Field separator. A single character; `"\t"` is accepted for tabs.
203    #[serde(default = "default_delimiter")]
204    pub delimiter: String,
205    /// Whether the first row names the fields. When false, fields are named
206    /// `column_0`, `column_1`, … — the same fallback the REST source and the
207    /// `csv` connector already use. On write it decides whether a header row
208    /// is written. `write_headers` is a deprecated name for it.
209    #[serde(default = "default_true", alias = "write_headers")]
210    pub has_headers: bool,
211    /// Quote character. A single byte; default `"`.
212    #[serde(default = "default_quote")]
213    pub quote: String,
214    /// Whether rows may have a different number of fields than the header.
215    /// `false` fails the read on the first ragged row, naming its line; `true`
216    /// keeps the fields the row has. Unset, each connector picks: the `file`
217    /// source is strict, the object-store sources and the REST source are
218    /// lenient.
219    #[serde(default, skip_serializing_if = "Option::is_none")]
220    pub flexible: Option<bool>,
221    /// Cell values read as `null` instead of a string (e.g. `["", "NULL"]`).
222    #[serde(default, skip_serializing_if = "Vec::is_empty")]
223    pub null_values: Vec<String>,
224    /// What a streaming CSV writer does with a field the header does not
225    /// have. See [`CsvUnknownField`]. Whole-object encoders always write the
226    /// union of every record's fields, so it only affects the `file` sink.
227    #[serde(default)]
228    pub on_unknown_field: CsvUnknownField,
229}
230
231/// What a CSV writer does with a field that is not in the header.
232#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
233#[serde(rename_all = "snake_case")]
234pub enum CsvUnknownField {
235    /// Add it as a new column; earlier rows get an empty cell for it.
236    #[default]
237    Widen,
238    /// Fix the header from the first page and drop the field, logging a
239    /// warning once per field.
240    Warn,
241    /// Fix the header from the first page and fail the write naming the field.
242    Error,
243}
244
245fn default_quote() -> String {
246    "\"".into()
247}
248
249impl Default for CsvOptions {
250    fn default() -> Self {
251        Self {
252            delimiter: default_delimiter(),
253            has_headers: true,
254            quote: default_quote(),
255            flexible: None,
256            null_values: Vec::new(),
257            on_unknown_field: CsvUnknownField::Widen,
258        }
259    }
260}
261
262impl CsvOptions {
263    /// The delimiter as the single byte the CSV readers want.
264    ///
265    /// Rejected rather than truncated: a multi-byte delimiter silently becoming
266    /// its first byte would split every row in the wrong place and produce
267    /// plausible-looking garbage.
268    pub fn delimiter_byte(&self) -> Result<u8, FaucetError> {
269        single_byte("delimiter", &self.delimiter)
270    }
271
272    /// The quote character as a single byte.
273    pub fn quote_byte(&self) -> Result<u8, FaucetError> {
274        single_byte("quote", &self.quote)
275    }
276
277    /// Whether ragged rows are accepted, with `default` when unset.
278    pub fn flexible_or(&self, default: bool) -> bool {
279        self.flexible.unwrap_or(default)
280    }
281
282    /// Check the single-byte fields, so a bad dialect fails at load time.
283    pub fn validate(&self) -> Result<(), FaucetError> {
284        self.delimiter_byte()?;
285        self.quote_byte()?;
286        Ok(())
287    }
288}
289
290fn single_byte(name: &str, value: &str) -> Result<u8, FaucetError> {
291    let d = match value {
292        "\\t" => "\t",
293        other => other,
294    };
295    match d.as_bytes() {
296        [b] => Ok(*b),
297        _ => Err(FaucetError::Config(format!(
298            "csv.{name} must be exactly one byte, got {value:?}"
299        ))),
300    }
301}
302
303/// Excel worksheet selection.
304#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
305#[serde(deny_unknown_fields)]
306pub struct ExcelOptions {
307    /// Worksheet name, or an index as a string. Default: the first sheet.
308    #[serde(default, skip_serializing_if = "Option::is_none")]
309    pub sheet: Option<String>,
310    /// 0-based row supplying the field names.
311    #[serde(default)]
312    pub header_row: usize,
313}
314
315pub(crate) fn default_record_element() -> String {
316    "record".into()
317}
318
319fn default_root_element() -> String {
320    "records".into()
321}
322
323/// XML record framing.
324///
325/// XML has no canonical record boundary, so one must be declared. On read,
326/// `record_element` names the repeated element; without it the decoder falls
327/// back to "the document's root children", which is right for the common
328/// `<rows><row/>…</rows>` shape and wrong often enough that naming the element
329/// is recommended in the docs.
330#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
331#[serde(deny_unknown_fields)]
332pub struct XmlOptions {
333    /// Element that delimits one record. Read: select these. Write: wrap each
334    /// record in one.
335    #[serde(default = "default_record_element")]
336    pub record_element: String,
337    /// Document element wrapping the records. Write only.
338    #[serde(default = "default_root_element")]
339    pub root_element: String,
340}
341
342impl Default for XmlOptions {
343    fn default() -> Self {
344        Self {
345            record_element: default_record_element(),
346            root_element: default_root_element(),
347        }
348    }
349}
350
351/// Block compression for written files.
352#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
353#[serde(rename_all = "snake_case")]
354pub enum AvroCodec {
355    /// Uncompressed blocks — the Avro default and the most portable.
356    #[default]
357    Null,
358    /// Raw deflate (RFC 1951), readable by every Avro implementation.
359    Deflate,
360    /// Snappy with the per-block CRC-32 the spec requires.
361    Snappy,
362    /// Zstandard.
363    Zstd,
364}
365
366/// Avro read/write options.
367#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
368#[serde(deny_unknown_fields)]
369pub struct AvroOptions {
370    /// An Avro schema (the JSON form: an object, an array for a union, or a
371    /// primitive name). On a **source** it is the *reader* schema every file
372    /// is resolved against — projection, aliases, defaults for added fields.
373    /// On a **sink** it is the *writer* schema; without it the schema is
374    /// inferred from the records.
375    #[serde(default, skip_serializing_if = "Option::is_none")]
376    pub schema: Option<Value>,
377    /// Block codec for written files: `null` (default), `deflate`, `snappy`,
378    /// `zstd`. Reading detects the codec from the file header.
379    #[serde(default)]
380    pub codec: AvroCodec,
381}
382
383/// ORC read options.
384#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
385#[serde(deny_unknown_fields)]
386pub struct OrcOptions {
387    /// Top-level columns to read, in file order. Default: every column. A
388    /// name the file does not have is an error, not an empty column.
389    #[serde(default, skip_serializing_if = "Option::is_none")]
390    pub columns: Option<Vec<String>>,
391}
392
393/// The per-format option blocks, flattened into a connector's config so they
394/// appear at its top level (`csv: {...}`, `excel: {...}`, `xml: {...}`).
395///
396/// Every block is defaulted, so a connector that adopts this struct adds no
397/// required field and the change stays a minor bump under the project's
398/// [API-evolution policy](https://github.com/faucet-hq/faucet-stream).
399#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
400pub struct FormatOptions {
401    /// CSV dialect, used when `format: csv`.
402    #[serde(default)]
403    pub csv: CsvOptions,
404    /// Worksheet selection, used when `format: xlsx`.
405    #[serde(default)]
406    pub excel: ExcelOptions,
407    /// Record framing, used when `format: xml`.
408    #[serde(default)]
409    pub xml: XmlOptions,
410    /// Reader / writer schema and codec, used when `format: avro`.
411    #[serde(default)]
412    pub avro: AvroOptions,
413    /// Column projection, used when `format: orc`.
414    #[serde(default)]
415    pub orc: OrcOptions,
416}
417
418/// Decode one object's bytes into records.
419///
420/// Async because the CSV reader is (`csv-async`), and because keeping one
421/// signature for every format is worth more than saving an `.await` on the
422/// three that are synchronous.
423pub async fn decode(
424    bytes: &[u8],
425    format: FileFormat,
426    opts: &FormatOptions,
427) -> Result<Vec<Value>, FaucetError> {
428    match format {
429        FileFormat::JsonLines => json::decode_lines(bytes),
430        FileFormat::JsonArray => json::decode_array(bytes),
431        FileFormat::RawText => json::decode_raw_text(bytes),
432        FileFormat::Csv => decode_csv(bytes, opts).await,
433        FileFormat::Xml => decode_xml(bytes, opts),
434        FileFormat::Xlsx => decode_xlsx(bytes, opts),
435        FileFormat::Parquet => Err(columnar_refusal("decode")),
436        FileFormat::Avro => decode_avro(bytes, opts),
437        FileFormat::Orc => decode_orc(bytes, opts),
438    }
439}
440
441/// [`decode`] for an owned body, run on a blocking thread.
442///
443/// The XML / Excel / Avro / ORC / JSON-array / CSV decoders are synchronous and
444/// CPU-bound; inline on an async worker a large object stalls heartbeats,
445/// cancellation and every other task on that worker (CORE-34). The body is
446/// moved rather than copied (ORC reads it in place).
447pub async fn decode_owned(
448    bytes: Vec<u8>,
449    format: FileFormat,
450    opts: &FormatOptions,
451) -> Result<Vec<Value>, FaucetError> {
452    let opts = opts.clone();
453    tokio::task::spawn_blocking(move || match format {
454        #[cfg(feature = "file-format-orc")]
455        FileFormat::Orc => orc::decode_owned(bytes, &opts.orc),
456        other => futures::executor::block_on(decode(&bytes, other, &opts)),
457    })
458    .await
459    .map_err(|e| FaucetError::Source(format!("file_format: decode task failed: {e}")))?
460}
461
462/// Encode records into one object's bytes.
463pub fn encode(
464    records: &[Value],
465    format: FileFormat,
466    opts: &FormatOptions,
467) -> Result<Vec<u8>, FaucetError> {
468    match format {
469        FileFormat::JsonLines => json::encode_lines(records),
470        FileFormat::JsonArray => json::encode_array(records),
471        FileFormat::RawText => json::encode_raw_text(records),
472        FileFormat::Csv => encode_csv(records, opts),
473        FileFormat::Xml => encode_xml(records, opts),
474        FileFormat::Xlsx => encode_xlsx(records, opts),
475        FileFormat::Parquet => Err(columnar_refusal("encode")),
476        FileFormat::Avro => encode_avro(records, opts),
477        FileFormat::Orc => Err(FaucetError::Config(
478            "`format: orc` is read-only — there is no ORC writer; write Parquet for a columnar \
479             output"
480                .into(),
481        )),
482    }
483}
484
485/// Parquet never goes through the generic helper.
486///
487/// Routing it here would work and would silently cost the Arrow fast path and
488/// the columnar memory profile, so it is refused rather than accommodated.
489fn columnar_refusal(verb: &str) -> FaucetError {
490    FaucetError::Config(format!(
491        "file_format::{verb}: `parquet` is columnar and is handled by each connector's Arrow \
492         path, not the generic record helper"
493    ))
494}
495
496/// Default ceiling on one file's size once decompressed, for formats read
497/// whole into memory (2 GiB).
498pub const DEFAULT_MAX_OBJECT_BYTES: u64 = 2 * 1024 * 1024 * 1024;
499
500/// Read `reader` to the end, failing once it yields more than `max` bytes — so
501/// a decompression bomb fails the file instead of exhausting memory. `what`
502/// names the file in the error.
503pub async fn read_to_end_capped<R>(reader: R, max: u64, what: &str) -> Result<Vec<u8>, FaucetError>
504where
505    R: tokio::io::AsyncRead + Unpin,
506{
507    use tokio::io::AsyncReadExt as _;
508    let mut buf = Vec::new();
509    reader
510        .take(max.saturating_add(1))
511        .read_to_end(&mut buf)
512        .await
513        .map_err(|e| FaucetError::Source(format!("read error for '{what}': {e}")))?;
514    check_object_size(buf.len() as u64, max, what)?;
515    Ok(buf)
516}
517
518/// [`read_to_end_capped`] for a UTF-8 text body.
519pub async fn read_to_string_capped<R>(
520    reader: R,
521    max: u64,
522    what: &str,
523) -> Result<String, FaucetError>
524where
525    R: tokio::io::AsyncRead + Unpin,
526{
527    let bytes = read_to_end_capped(reader, max, what).await?;
528    String::from_utf8(bytes)
529        .map_err(|e| FaucetError::Source(format!("'{what}' is not valid UTF-8: {e}")))
530}
531
532/// Wrap `reader` so that no line read through it may exceed `max` bytes: the
533/// read fails (an `InvalidData` I/O error naming the cap) once more than `max`
534/// bytes arrive without a `\n`.
535///
536/// For streamed, newline-delimited formats — JSON Lines, CSV — whose readers
537/// buffer one line (or row) whole: without it, a body with no newline (a
538/// single-line dump, a binary file, a misbehaving endpoint) is accumulated in
539/// memory until it ends. The count is over the bytes as they arrive, so read
540/// buffering does not affect it. A CSV field quoting newlines is counted line
541/// by line.
542pub fn line_capped<R>(reader: R, max: u64) -> LineCapped<R> {
543    LineCapped {
544        inner: reader,
545        since_newline: 0,
546        max,
547    }
548}
549
550/// A reader whose lines are bounded (see [`line_capped`]).
551#[derive(Debug)]
552pub struct LineCapped<R> {
553    inner: R,
554    since_newline: u64,
555    max: u64,
556}
557
558impl<R: tokio::io::AsyncRead + Unpin> tokio::io::AsyncRead for LineCapped<R> {
559    fn poll_read(
560        mut self: std::pin::Pin<&mut Self>,
561        cx: &mut std::task::Context<'_>,
562        buf: &mut tokio::io::ReadBuf<'_>,
563    ) -> std::task::Poll<std::io::Result<()>> {
564        let before = buf.filled().len();
565        let polled = std::pin::Pin::new(&mut self.inner).poll_read(cx, buf);
566        if let std::task::Poll::Ready(Ok(())) = &polled {
567            let fresh = &buf.filled()[before..];
568            self.since_newline = match fresh.iter().rposition(|b| *b == b'\n') {
569                Some(i) => (fresh.len() - i - 1) as u64,
570                None => self.since_newline.saturating_add(fresh.len() as u64),
571            };
572            if self.since_newline > self.max {
573                return std::task::Poll::Ready(Err(std::io::Error::new(
574                    std::io::ErrorKind::InvalidData,
575                    format!(
576                        "a line is longer than `max_object_bytes` ({} bytes) — is the file \
577                         missing its newlines? raise `max_object_bytes` to read it",
578                        self.max
579                    ),
580                )));
581            }
582        }
583        polled
584    }
585}
586
587/// The error for a file past `max_object_bytes`.
588pub fn check_object_size(len: u64, max: u64, what: &str) -> Result<(), FaucetError> {
589    if len > max {
590        return Err(FaucetError::Source(format!(
591            "'{what}' is larger than `max_object_bytes` ({max} bytes) once decompressed; \
592             raise `max_object_bytes` to read it"
593        )));
594    }
595    Ok(())
596}
597
598/// The error a build without the feature reports.
599///
600/// Named rather than mis-parsed: decoding an Excel workbook as CSV produces
601/// records, which is the worst possible outcome.
602#[cfg_attr(
603    all(
604        feature = "file-format-csv",
605        feature = "file-format-xml",
606        feature = "file-format-excel",
607        feature = "file-format-avro",
608        feature = "file-format-orc"
609    ),
610    allow(dead_code)
611)]
612fn missing_feature(format: FileFormat, feature: &str) -> FaucetError {
613    FaucetError::Config(format!(
614        "`format: {}` requires the `{feature}` build feature — rebuild with \
615         `--features {feature}`",
616        format.as_str()
617    ))
618}
619
620#[cfg(feature = "file-format-csv")]
621async fn decode_csv(bytes: &[u8], opts: &FormatOptions) -> Result<Vec<Value>, FaucetError> {
622    csv::decode_with(bytes, &opts.csv, true).await
623}
624
625#[cfg(not(feature = "file-format-csv"))]
626async fn decode_csv(_: &[u8], _: &FormatOptions) -> Result<Vec<Value>, FaucetError> {
627    Err(missing_feature(FileFormat::Csv, "file-format-csv"))
628}
629
630#[cfg(feature = "file-format-csv")]
631fn encode_csv(records: &[Value], opts: &FormatOptions) -> Result<Vec<u8>, FaucetError> {
632    csv::encode_with(records, &opts.csv)
633}
634
635#[cfg(not(feature = "file-format-csv"))]
636fn encode_csv(_: &[Value], _: &FormatOptions) -> Result<Vec<u8>, FaucetError> {
637    Err(missing_feature(FileFormat::Csv, "file-format-csv"))
638}
639
640#[cfg(feature = "file-format-xml")]
641fn decode_xml(bytes: &[u8], opts: &FormatOptions) -> Result<Vec<Value>, FaucetError> {
642    xml::decode(bytes, &opts.xml.record_element)
643}
644
645#[cfg(not(feature = "file-format-xml"))]
646fn decode_xml(_: &[u8], _: &FormatOptions) -> Result<Vec<Value>, FaucetError> {
647    Err(missing_feature(FileFormat::Xml, "file-format-xml"))
648}
649
650#[cfg(feature = "file-format-xml")]
651fn encode_xml(records: &[Value], opts: &FormatOptions) -> Result<Vec<u8>, FaucetError> {
652    xml::encode(records, &opts.xml.root_element, &opts.xml.record_element)
653}
654
655#[cfg(not(feature = "file-format-xml"))]
656fn encode_xml(_: &[Value], _: &FormatOptions) -> Result<Vec<u8>, FaucetError> {
657    Err(missing_feature(FileFormat::Xml, "file-format-xml"))
658}
659
660#[cfg(feature = "file-format-excel")]
661fn decode_xlsx(bytes: &[u8], opts: &FormatOptions) -> Result<Vec<Value>, FaucetError> {
662    excel::decode(bytes, opts.excel.sheet.as_deref(), opts.excel.header_row)
663}
664
665#[cfg(not(feature = "file-format-excel"))]
666fn decode_xlsx(_: &[u8], _: &FormatOptions) -> Result<Vec<Value>, FaucetError> {
667    Err(missing_feature(FileFormat::Xlsx, "file-format-excel"))
668}
669
670#[cfg(feature = "file-format-excel")]
671fn encode_xlsx(records: &[Value], opts: &FormatOptions) -> Result<Vec<u8>, FaucetError> {
672    excel::encode(records, opts.excel.sheet.as_deref())
673}
674
675#[cfg(not(feature = "file-format-excel"))]
676fn encode_xlsx(_: &[Value], _: &FormatOptions) -> Result<Vec<u8>, FaucetError> {
677    Err(missing_feature(FileFormat::Xlsx, "file-format-excel"))
678}
679
680#[cfg(feature = "file-format-avro")]
681fn decode_avro(bytes: &[u8], opts: &FormatOptions) -> Result<Vec<Value>, FaucetError> {
682    avro::decode(bytes, &opts.avro)
683}
684
685#[cfg(not(feature = "file-format-avro"))]
686fn decode_avro(_: &[u8], _: &FormatOptions) -> Result<Vec<Value>, FaucetError> {
687    Err(missing_feature(FileFormat::Avro, "file-format-avro"))
688}
689
690#[cfg(feature = "file-format-avro")]
691fn encode_avro(records: &[Value], opts: &FormatOptions) -> Result<Vec<u8>, FaucetError> {
692    avro::encode(records, &opts.avro)
693}
694
695#[cfg(not(feature = "file-format-avro"))]
696fn encode_avro(_: &[Value], _: &FormatOptions) -> Result<Vec<u8>, FaucetError> {
697    Err(missing_feature(FileFormat::Avro, "file-format-avro"))
698}
699
700#[cfg(feature = "file-format-orc")]
701fn decode_orc(bytes: &[u8], opts: &FormatOptions) -> Result<Vec<Value>, FaucetError> {
702    orc::decode(bytes, &opts.orc)
703}
704
705#[cfg(not(feature = "file-format-orc"))]
706fn decode_orc(_: &[u8], _: &FormatOptions) -> Result<Vec<Value>, FaucetError> {
707    Err(missing_feature(FileFormat::Orc, "file-format-orc"))
708}
709
710/// Field names across `records`, in first-seen order.
711///
712/// A union rather than the first record's keys, so a later record carrying an
713/// extra field widens the file instead of losing the field silently.
714///
715/// "First-seen" orders *records* deterministically; within one record the order
716/// is whatever `serde_json::Map` iterates, which is insertion order only when
717/// the `preserve_order` feature is unified into the build and sorted otherwise.
718/// Column order is therefore a build-time property, not a guarantee — do not
719/// write a test that pins it.
720pub fn header_union(records: &[Value]) -> Vec<String> {
721    let mut seen = std::collections::HashSet::new();
722    let mut out = Vec::new();
723    for r in records {
724        if let Value::Object(map) = r {
725            for k in map.keys() {
726                if seen.insert(k.clone()) {
727                    out.push(k.clone());
728                }
729            }
730        }
731    }
732    out
733}
734
735/// Refuse a record a tabular encoder cannot place in a row: every cell is
736/// keyed by a field name, so a scalar or array would be written as an empty
737/// row and counted as written.
738#[cfg(any(feature = "file-format-csv", feature = "file-format-excel"))]
739pub(crate) fn require_objects(records: &[Value], format: &str) -> Result<(), FaucetError> {
740    match records.iter().position(|r| !r.is_object()) {
741        None => Ok(()),
742        Some(i) => Err(FaucetError::Sink(format!(
743            "{format}: record {i} is not a JSON object; a row needs named fields"
744        ))),
745    }
746}
747
748/// One cell's text for the tabular formats.
749///
750/// A nested object or array is re-serialized as JSON rather than dropped or
751/// `Debug`-printed: a spreadsheet cell cannot hold structure, and the round
752/// trip back through `json_parse` is at least lossless.
753pub fn cell_text(v: &Value) -> String {
754    match v {
755        Value::Null => String::new(),
756        Value::String(s) => s.clone(),
757        Value::Bool(b) => b.to_string(),
758        Value::Number(n) => n.to_string(),
759        other => crate::util::canonical_json(other),
760    }
761}
762
763#[cfg(test)]
764mod capped_read_tests {
765    use super::*;
766
767    #[tokio::test]
768    async fn a_body_past_the_cap_fails_and_one_at_the_cap_reads() {
769        let body = [b'x'; 64];
770        let ok = read_to_end_capped(&body[..], 64, "f").await.unwrap();
771        assert_eq!(ok.len(), 64);
772        let err = read_to_end_capped(&body[..], 63, "big.json.gz")
773            .await
774            .unwrap_err();
775        assert!(err.to_string().contains("big.json.gz"), "{err}");
776        assert!(err.to_string().contains("max_object_bytes"), "{err}");
777        assert_eq!(
778            read_to_string_capped(&b"hi"[..], 8, "t").await.unwrap(),
779            "hi"
780        );
781        assert!(
782            read_to_string_capped(&[0xff, 0xfe][..], 8, "t")
783                .await
784                .is_err()
785        );
786        assert!(check_object_size(5, 4, "x").is_err());
787    }
788}
789
790#[cfg(test)]
791mod tests {
792    use super::*;
793    use serde_json::json;
794
795    #[tokio::test]
796    async fn line_capped_bounds_one_line_not_the_stream() {
797        use tokio::io::AsyncBufReadExt as _;
798        let body = b"{\"a\":1}\n{\"a\":2}\n{\"a\":3}\n".repeat(100);
799        let mut lines = tokio::io::BufReader::new(line_capped(&body[..], 64)).lines();
800        let mut n = 0;
801        while let Some(_line) = lines.next_line().await.unwrap() {
802            n += 1;
803        }
804        assert_eq!(
805            n, 300,
806            "the whole stream is far past the cap, each record is not"
807        );
808
809        let long = vec![b'x'; 1_000];
810        let mut lines = tokio::io::BufReader::new(line_capped(&long[..], 64)).lines();
811        let err = lines.next_line().await.unwrap_err();
812        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
813        assert!(err.to_string().contains("max_object_bytes"), "{err}");
814    }
815
816    #[tokio::test]
817    async fn decode_owned_matches_decode_and_runs_off_the_runtime() {
818        let body = b"[{\"a\":1},{\"a\":2}]".to_vec();
819        let opts = FormatOptions::default();
820        assert_eq!(
821            decode_owned(body.clone(), FileFormat::JsonArray, &opts)
822                .await
823                .unwrap(),
824            decode(&body, FileFormat::JsonArray, &opts).await.unwrap()
825        );
826        assert!(
827            decode_owned(Vec::new(), FileFormat::Parquet, &opts)
828                .await
829                .is_err()
830        );
831    }
832
833    #[cfg(feature = "file-format-orc")]
834    #[tokio::test]
835    async fn decode_owned_reads_orc_in_place() {
836        const FIXTURE: &[u8] = include_bytes!("../../tests/fixtures/orc/people.orc");
837        let opts = FormatOptions::default();
838        assert_eq!(
839            decode_owned(FIXTURE.to_vec(), FileFormat::Orc, &opts)
840                .await
841                .unwrap(),
842            decode(FIXTURE, FileFormat::Orc, &opts).await.unwrap()
843        );
844    }
845
846    /// Every variant's extension and wire name, so adding a format without
847    /// giving it both is a test failure rather than a `.txt` file called
848    /// "raw_text".
849    #[test]
850    fn every_variant_has_an_extension_and_a_wire_name() {
851        let all = [
852            (FileFormat::JsonLines, ".jsonl", "json_lines"),
853            (FileFormat::JsonArray, ".json", "json_array"),
854            (FileFormat::Csv, ".csv", "csv"),
855            (FileFormat::Xml, ".xml", "xml"),
856            (FileFormat::Xlsx, ".xlsx", "xlsx"),
857            (FileFormat::Parquet, ".parquet", "parquet"),
858            (FileFormat::RawText, ".txt", "raw_text"),
859            (FileFormat::Avro, ".avro", "avro"),
860            (FileFormat::Orc, ".orc", "orc"),
861        ];
862        for (f, ext, name) in all {
863            assert_eq!(f.extension(), ext, "{f:?}");
864            assert_eq!(f.as_str(), name, "{f:?}");
865        }
866    }
867
868    /// `raw_text` is source-only in the connectors, but the shared helper
869    /// still round-trips it — the sink side is what `encode_raw_text` exists
870    /// for, and routing it through the top-level dispatch is how a connector
871    /// reaches it.
872    #[tokio::test]
873    async fn raw_text_round_trips_through_the_top_level_dispatch() {
874        let opts = FormatOptions::default();
875        let recs = decode(b"hello", FileFormat::RawText, &opts)
876            .await
877            .expect("decode");
878        assert_eq!(recs, vec![json!({"text": "hello"})]);
879        let bytes = encode(&recs, FileFormat::RawText, &opts).expect("encode");
880        assert_eq!(bytes, b"hello\n");
881    }
882
883    /// The refusal a build without the feature emits — it must name the
884    /// format and the feature, because mis-parsing an Excel blob as CSV is
885    /// the outcome this exists to prevent.
886    #[test]
887    fn a_missing_feature_is_named_not_guessed() {
888        let err = missing_feature(FileFormat::Xlsx, "file-format-excel");
889        let msg = err.to_string();
890        assert!(msg.contains("xlsx"), "{msg}");
891        assert!(msg.contains("file-format-excel"), "{msg}");
892    }
893
894    #[test]
895    fn extensions_and_names_are_stable() {
896        assert_eq!(FileFormat::default(), FileFormat::JsonLines);
897        assert_eq!(FileFormat::Csv.extension(), ".csv");
898        assert_eq!(FileFormat::Xlsx.as_str(), "xlsx");
899    }
900
901    #[test]
902    fn whole_object_formats_are_the_ones_without_a_record_boundary() {
903        for f in [
904            FileFormat::JsonArray,
905            FileFormat::Xml,
906            FileFormat::Xlsx,
907            FileFormat::Orc,
908        ] {
909            assert!(f.requires_whole_object(), "{f:?}");
910        }
911        for f in [
912            FileFormat::JsonLines,
913            FileFormat::Csv,
914            FileFormat::RawText,
915            FileFormat::Parquet,
916            FileFormat::Avro,
917        ] {
918            assert!(!f.requires_whole_object(), "{f:?}");
919        }
920    }
921
922    #[test]
923    fn extensions_resolve_through_compression_suffixes() {
924        let cases = [
925            ("a.jsonl", Some(FileFormat::JsonLines)),
926            ("dir/b.NDJSON", Some(FileFormat::JsonLines)),
927            ("c.json.gz", Some(FileFormat::JsonArray)),
928            ("d.csv.gz", Some(FileFormat::Csv)),
929            ("e.xml.zst", Some(FileFormat::Xml)),
930            ("f.xlsx", Some(FileFormat::Xlsx)),
931            ("g.parquet", Some(FileFormat::Parquet)),
932            ("h.txt", Some(FileFormat::RawText)),
933            ("i.avro", Some(FileFormat::Avro)),
934            ("j.orc", Some(FileFormat::Orc)),
935            ("k.csv.gzip", Some(FileFormat::Csv)),
936            ("l.jsonl.zstd", Some(FileFormat::JsonLines)),
937            ("m.tsv", None),
938            ("noext", None),
939            ("n.gz", None),
940        ];
941        for (path, want) in cases {
942            assert_eq!(FileFormat::from_path(path), want, "{path}");
943        }
944    }
945
946    #[test]
947    fn container_and_writable_classification() {
948        assert!(FileFormat::Avro.is_container() && FileFormat::Orc.is_container());
949        assert!(!FileFormat::Csv.is_container());
950        assert!(FileFormat::Avro.is_writable());
951        assert!(!FileFormat::Orc.is_writable() && !FileFormat::Parquet.is_writable());
952    }
953
954    #[tokio::test]
955    async fn orc_is_refused_for_writing() {
956        let err = encode(&[], FileFormat::Orc, &FormatOptions::default()).expect_err("orc");
957        assert!(err.to_string().contains("read-only"), "{err}");
958    }
959
960    #[tokio::test]
961    async fn parquet_is_refused_on_both_directions() {
962        // Going through the generic helper would work and would silently cost
963        // the Arrow path, so it must be an error, not a fallback.
964        let err = decode(b"", FileFormat::Parquet, &FormatOptions::default())
965            .await
966            .expect_err("decode refuses parquet");
967        assert!(err.to_string().contains("columnar"), "{err}");
968        let err = encode(&[], FileFormat::Parquet, &FormatOptions::default())
969            .expect_err("encode refuses parquet");
970        assert!(err.to_string().contains("columnar"), "{err}");
971    }
972
973    #[test]
974    fn delimiter_must_be_one_byte() {
975        let mut o = CsvOptions::default();
976        assert_eq!(o.delimiter_byte().expect("default"), b',');
977        o.delimiter = "\\t".into();
978        assert_eq!(o.delimiter_byte().expect("tab"), b'\t');
979        o.delimiter = "||".into();
980        let err = o.delimiter_byte().expect_err("multi-byte");
981        assert!(err.to_string().contains("exactly one byte"), "{err}");
982        // A multi-byte character is rejected too — truncating it to its first
983        // UTF-8 byte would split rows at a byte that never appears alone.
984        o.delimiter = "§".into();
985        assert!(o.delimiter_byte().is_err());
986    }
987
988    #[test]
989    fn header_union_widens_across_records_and_ignores_non_objects() {
990        // Records are visited in order, so a key only this record has lands
991        // after the earlier ones. Intra-record order is a build-time property
992        // of `serde_json::Map`, so it is deliberately not asserted.
993        let recs = vec![json!({"a": 1}), json!({"b": 2}), json!("not an object")];
994        assert_eq!(header_union(&recs), vec!["a", "b"]);
995        // A key seen twice appears once.
996        assert_eq!(header_union(&[json!({"a": 1}), json!({"a": 2})]), vec!["a"]);
997        assert!(header_union(&[json!([1, 2])]).is_empty());
998    }
999
1000    #[test]
1001    fn cells_render_scalars_plainly_and_structure_as_json() {
1002        assert_eq!(cell_text(&Value::Null), "");
1003        assert_eq!(cell_text(&json!("x")), "x");
1004        assert_eq!(cell_text(&json!(3)), "3");
1005        assert_eq!(cell_text(&json!(true)), "true");
1006        assert_eq!(cell_text(&json!({"k": 1})), r#"{"k":1}"#);
1007    }
1008}
faucet_core/file_format/mod.rs

faucet_core/file_format/
mod.rs
