faucet_sink_csv/
config.rs

1//! CSV sink configuration.
2
3use faucet_core::DEFAULT_BATCH_SIZE;
4use schemars::JsonSchema;
5use serde::{Deserialize, Serialize};
6
7/// Configuration for the CSV file sink.
8#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
9pub struct CsvSinkConfig {
10    /// Path to the output CSV file.
11    pub path: String,
12    /// Field delimiter byte. Defaults to `b','`.
13    #[serde(default = "default_delimiter")]
14    pub delimiter: u8,
15    /// Whether to write a header row. Defaults to `true`.
16    #[serde(default = "default_true")]
17    pub write_headers: bool,
18    /// Whether to append to an existing file. Defaults to `false` (truncates).
19    #[serde(default)]
20    pub append: bool,
21    /// Records per upstream [`StreamPage`](faucet_core::StreamPage). The CSV
22    /// sink writes rows to disk one at a time inside a `spawn_blocking` task
23    /// fronted by a buffered `csv::Writer`, so this field has **no
24    /// behavioural impact** at the sink — it is exposed purely for config
25    /// parity across every sink in the workspace. Defaults to
26    /// [`DEFAULT_BATCH_SIZE`].
27    ///
28    /// `batch_size = 0` (the "no batching" sentinel) and any positive value
29    /// produce byte-for-byte identical output for this sink: each record is
30    /// serialised and written individually regardless of how upstream chunked
31    /// the page.
32    #[serde(default = "default_batch_size")]
33    pub batch_size: usize,
34    /// What to do when a record contains a field that is not part of the
35    /// frozen column set (the CSV header is fixed from the first batch and
36    /// cannot be rewritten without rewriting the whole file). A field that
37    /// first appears in a *later* batch — or in a later record once the header
38    /// is already written — cannot be added to the header, so its value is
39    /// dropped from the output. Defaults to [`OnUnknownField::Warn`].
40    #[serde(default)]
41    pub on_unknown_field: OnUnknownField,
42    /// Compression codec for the output file. Defaults to
43    /// [`CompressionConfig::Auto`](faucet_core::CompressionConfig::Auto) —
44    /// `.gz` / `.zst` suffix selects gzip / zstd. Requires the crate-local
45    /// `compression` feature.
46    #[cfg(feature = "compression")]
47    #[serde(default)]
48    pub compression: faucet_core::CompressionConfig,
49}
50
51/// Policy for a record key that is not in the sink's frozen column set.
52///
53/// The CSV header is fixed from the first non-empty batch and cannot be
54/// rewritten in place. A field that first appears after the header is written
55/// (in a later batch, or a later record of the first batch once the header is
56/// determined) therefore cannot become a column, so its value would be lost.
57#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
58#[serde(rename_all = "snake_case")]
59pub enum OnUnknownField {
60    /// Emit a one-shot `tracing::warn!` naming the dropped field(s) and keep
61    /// writing — the unknown field's value is dropped from the output. This is
62    /// the default: it preserves the historical streaming/append behaviour
63    /// while making the column-level loss visible instead of silent.
64    #[default]
65    Warn,
66    /// Abort the write with a [`FaucetError::Sink`](faucet_core::FaucetError::Sink)
67    /// the first time any record carries a field outside the frozen column set.
68    /// Opt into this when silent column-level data loss is unacceptable.
69    Error,
70}
71
72fn default_delimiter() -> u8 {
73    b','
74}
75
76fn default_true() -> bool {
77    true
78}
79
80fn default_batch_size() -> usize {
81    DEFAULT_BATCH_SIZE
82}
83
84impl CsvSinkConfig {
85    /// Create a new config with the required file path and sensible defaults.
86    pub fn new(path: impl Into<String>) -> Self {
87        Self {
88            path: path.into(),
89            delimiter: b',',
90            write_headers: true,
91            append: false,
92            batch_size: DEFAULT_BATCH_SIZE,
93            on_unknown_field: OnUnknownField::Warn,
94            #[cfg(feature = "compression")]
95            compression: faucet_core::CompressionConfig::Auto,
96        }
97    }
98
99    /// Set the field delimiter byte.
100    pub fn delimiter(mut self, d: u8) -> Self {
101        self.delimiter = d;
102        self
103    }
104
105    /// Set whether to write a header row.
106    pub fn write_headers(mut self, v: bool) -> Self {
107        self.write_headers = v;
108        self
109    }
110
111    /// Set whether to append to an existing file.
112    pub fn append(mut self, v: bool) -> Self {
113        self.append = v;
114        self
115    }
116
117    /// Set the per-page record count hint reported alongside other sink
118    /// configs.
119    ///
120    /// This sink writes per-record through a buffered writer, so the value is
121    /// observably a no-op: `0` (the "no batching" sentinel) and any positive
122    /// value produce the same on-disk output. Present for symmetry with sinks
123    /// whose `batch_size` does drive I/O sizing (e.g. SQL multi-row inserts,
124    /// BigQuery streaming inserts).
125    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
126        self.batch_size = batch_size;
127        self
128    }
129
130    /// Set the policy for record keys outside the frozen column set.
131    pub fn on_unknown_field(mut self, policy: OnUnknownField) -> Self {
132        self.on_unknown_field = policy;
133        self
134    }
135
136    /// Set the compression codec. Available only with the `compression` feature.
137    #[cfg(feature = "compression")]
138    pub fn compression(mut self, c: faucet_core::CompressionConfig) -> Self {
139        self.compression = c;
140        self
141    }
142}
143
144#[cfg(test)]
145mod tests {
146    use super::*;
147
148    #[test]
149    fn default_config() {
150        let config = CsvSinkConfig::new("/tmp/out.csv");
151        assert_eq!(config.path, "/tmp/out.csv");
152        assert_eq!(config.delimiter, b',');
153        assert!(config.write_headers);
154        assert!(!config.append);
155    }
156
157    #[test]
158    fn builder_methods() {
159        let config = CsvSinkConfig::new("/tmp/out.tsv")
160            .delimiter(b'\t')
161            .write_headers(false)
162            .append(true);
163        assert_eq!(config.delimiter, b'\t');
164        assert!(!config.write_headers);
165        assert!(config.append);
166    }
167
168    #[test]
169    fn batch_size_defaults_to_default_batch_size() {
170        let config = CsvSinkConfig::new("/tmp/out.csv");
171        assert_eq!(config.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
172    }
173
174    #[test]
175    fn with_batch_size_overrides_default() {
176        let config = CsvSinkConfig::new("/tmp/out.csv").with_batch_size(250);
177        assert_eq!(config.batch_size, 250);
178    }
179
180    #[test]
181    fn batch_size_zero_is_accepted_as_no_batching_sentinel() {
182        let config = CsvSinkConfig::new("/tmp/out.csv").with_batch_size(0);
183        assert_eq!(config.batch_size, 0);
184        assert!(faucet_core::validate_batch_size(config.batch_size).is_ok());
185    }
186
187    #[test]
188    fn batch_size_above_max_is_rejected_by_validate_batch_size() {
189        let config =
190            CsvSinkConfig::new("/tmp/out.csv").with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
191        assert!(faucet_core::validate_batch_size(config.batch_size).is_err());
192    }
193
194    #[test]
195    fn batch_size_deserializes_from_json() {
196        let json = r#"{
197            "path": "/tmp/out.csv",
198            "delimiter": 44,
199            "write_headers": true,
200            "append": false,
201            "batch_size": 500
202        }"#;
203        let config: CsvSinkConfig = serde_json::from_str(json).unwrap();
204        assert_eq!(config.batch_size, 500);
205    }
206
207    #[test]
208    fn batch_size_defaults_when_missing_in_json() {
209        let json = r#"{"path": "/tmp/out.csv"}"#;
210        let config: CsvSinkConfig = serde_json::from_str(json).unwrap();
211        assert_eq!(config.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
212    }
213
214    #[test]
215    fn on_unknown_field_defaults_to_warn() {
216        let config = CsvSinkConfig::new("/tmp/out.csv");
217        assert_eq!(config.on_unknown_field, OnUnknownField::Warn);
218        let json = r#"{"path": "/tmp/out.csv"}"#;
219        let config: CsvSinkConfig = serde_json::from_str(json).unwrap();
220        assert_eq!(config.on_unknown_field, OnUnknownField::Warn);
221    }
222
223    #[test]
224    fn on_unknown_field_deserializes_snake_case() {
225        let json = r#"{"path": "/tmp/out.csv", "on_unknown_field": "error"}"#;
226        let config: CsvSinkConfig = serde_json::from_str(json).unwrap();
227        assert_eq!(config.on_unknown_field, OnUnknownField::Error);
228    }
229
230    #[test]
231    fn on_unknown_field_builder_sets_policy() {
232        let config = CsvSinkConfig::new("/tmp/out.csv").on_unknown_field(OnUnknownField::Error);
233        assert_eq!(config.on_unknown_field, OnUnknownField::Error);
234    }
235}
faucet_sink_csv/config.rs

faucet_sink_csv/
config.rs
