faucet_sink_s3/
config.rs

1//! S3 sink configuration.
2
3use faucet_core::DEFAULT_BATCH_SIZE;
4use schemars::JsonSchema;
5use serde::{Deserialize, Serialize};
6
7/// On-the-wire format of objects written by the S3 sink.
8#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
9#[serde(rename_all = "snake_case")]
10pub enum S3SinkFormat {
11    /// Newline-delimited JSON — one JSON record per line (the default).
12    #[default]
13    JsonLines,
14    /// Apache Parquet. Each written object is a complete, self-contained
15    /// Parquet file. Enables the **columnar** fast path
16    /// ([`Sink::write_batch_columnar`](faucet_core::Sink::write_batch_columnar))
17    /// so a `parquet`/`delta` → `s3(parquet)` chain never materializes
18    /// `serde_json::Value`. Requires the crate-local `arrow` feature
19    /// (RFC 0002 / #375).
20    #[cfg(feature = "arrow")]
21    Parquet,
22}
23
24/// Configuration for the S3 sink connector.
25#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
26pub struct S3SinkConfig {
27    /// S3 bucket name.
28    pub bucket: String,
29    /// Key prefix for written objects.
30    pub prefix: String,
31    /// Object format (default: `json_lines`). Set to `parquet` (with the
32    /// `arrow` feature) to write Parquet objects and enable the columnar
33    /// fast path.
34    #[serde(default)]
35    pub format: S3SinkFormat,
36    /// AWS region. `None` uses the SDK default.
37    pub region: Option<String>,
38    /// Custom endpoint URL for S3-compatible services (e.g. MinIO).
39    pub endpoint_url: Option<String>,
40    /// File extension for written objects (default: `.jsonl`).
41    pub file_extension: String,
42    /// Maximum records per file. `None` removes the per-file record cap — but
43    /// the sink still writes **one object per `write_batch` call** (i.e. one per
44    /// upstream page), and `batch_size` may chunk a call further; it does not
45    /// coalesce a streaming run into a single object.
46    pub max_records_per_file: Option<usize>,
47    /// Maximum number of concurrent file uploads (default: 10).
48    pub concurrency: usize,
49    /// Records per S3 object written by a single
50    /// [`Sink::write_batch`](faucet_core::Sink::write_batch) call. When a call
51    /// hands the sink `N` records with `batch_size = M > 0`, the sink writes
52    /// `ceil(N / M)` objects, each containing at most `M` records (the final
53    /// object holds the remainder). Defaults to [`DEFAULT_BATCH_SIZE`].
54    ///
55    /// `batch_size = 0` is the "no batching" sentinel: the sink writes
56    /// whatever upstream hands it without re-chunking (still honouring
57    /// `max_records_per_file` if set). Recommended for S3 — most callers
58    /// should leave this at `0` and let the source's `batch_size` drive
59    /// object sizing, because many tiny S3 objects are a well-known
60    /// anti-pattern (per-request overhead, slower downstream reads,
61    /// LIST/PUT cost).
62    ///
63    /// When both `batch_size > 0` and `max_records_per_file` are set, the
64    /// effective per-object cap is `min(batch_size, max_records_per_file)`.
65    #[serde(default = "default_batch_size")]
66    pub batch_size: usize,
67    /// Compression codec applied to each uploaded object body. Defaults to
68    /// [`CompressionConfig::Auto`](faucet_core::CompressionConfig::Auto) —
69    /// resolves against `file_extension` (so `.jsonl.gz` triggers gzip).
70    /// Requires the crate-local `compression` feature. Note: this sink does
71    /// **not** set the S3 `Content-Encoding` header, so consumers must
72    /// decompress explicitly.
73    #[cfg(feature = "compression")]
74    #[serde(default)]
75    pub compression: faucet_core::CompressionConfig,
76}
77
78fn default_batch_size() -> usize {
79    DEFAULT_BATCH_SIZE
80}
81
82impl S3SinkConfig {
83    /// Create a new config with the required bucket name and sensible defaults.
84    pub fn new(bucket: impl Into<String>) -> Self {
85        Self {
86            bucket: bucket.into(),
87            prefix: String::new(),
88            format: S3SinkFormat::default(),
89            region: None,
90            endpoint_url: None,
91            file_extension: ".jsonl".to_string(),
92            max_records_per_file: None,
93            concurrency: 10,
94            batch_size: DEFAULT_BATCH_SIZE,
95            #[cfg(feature = "compression")]
96            compression: faucet_core::CompressionConfig::Auto,
97        }
98    }
99
100    /// Set the key prefix for written objects.
101    pub fn prefix(mut self, prefix: impl Into<String>) -> Self {
102        self.prefix = prefix.into();
103        self
104    }
105
106    /// Set the object format (`json_lines` or, with the `arrow` feature,
107    /// `parquet`).
108    pub fn format(mut self, format: S3SinkFormat) -> Self {
109        self.format = format;
110        self
111    }
112
113    /// Set the AWS region.
114    pub fn region(mut self, region: impl Into<String>) -> Self {
115        self.region = Some(region.into());
116        self
117    }
118
119    /// Set a custom endpoint URL for S3-compatible services.
120    pub fn endpoint_url(mut self, url: impl Into<String>) -> Self {
121        self.endpoint_url = Some(url.into());
122        self
123    }
124
125    /// Set the file extension for written objects.
126    pub fn file_extension(mut self, ext: impl Into<String>) -> Self {
127        self.file_extension = ext.into();
128        self
129    }
130
131    /// Set the maximum number of records per file.
132    pub fn max_records_per_file(mut self, max: usize) -> Self {
133        self.max_records_per_file = Some(max);
134        self
135    }
136
137    /// Set the maximum number of concurrent file uploads.
138    pub fn concurrency(mut self, concurrency: usize) -> Self {
139        self.concurrency = concurrency;
140        self
141    }
142
143    /// Set the per-object record count for
144    /// [`Sink::write_batch`](faucet_core::Sink::write_batch).
145    ///
146    /// Pass `0` to opt out of write-side re-chunking — the sink writes
147    /// whatever upstream hands it as a single object (still honouring
148    /// `max_records_per_file` if set). `0` is the recommended value for S3
149    /// because writing many small objects is an anti-pattern.
150    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
151        self.batch_size = batch_size;
152        self
153    }
154
155    /// Set the compression codec. Available only with the `compression` feature.
156    #[cfg(feature = "compression")]
157    pub fn compression(mut self, c: faucet_core::CompressionConfig) -> Self {
158        self.compression = c;
159        self
160    }
161}
162
163#[cfg(test)]
164mod tests {
165    use super::*;
166
167    #[test]
168    fn default_config() {
169        let config = S3SinkConfig::new("my-bucket");
170        assert_eq!(config.bucket, "my-bucket");
171        assert_eq!(config.prefix, "");
172        assert!(config.region.is_none());
173        assert!(config.endpoint_url.is_none());
174        assert_eq!(config.file_extension, ".jsonl");
175        assert!(config.max_records_per_file.is_none());
176    }
177
178    #[test]
179    fn builder_methods() {
180        let config = S3SinkConfig::new("my-bucket")
181            .prefix("output/")
182            .region("eu-west-1")
183            .endpoint_url("http://localhost:9000")
184            .file_extension(".json")
185            .max_records_per_file(1000);
186
187        assert_eq!(config.bucket, "my-bucket");
188        assert_eq!(config.prefix, "output/");
189        assert_eq!(config.region.as_deref(), Some("eu-west-1"));
190        assert_eq!(
191            config.endpoint_url.as_deref(),
192            Some("http://localhost:9000")
193        );
194        assert_eq!(config.file_extension, ".json");
195        assert_eq!(config.max_records_per_file, Some(1000));
196    }
197
198    #[test]
199    fn batch_size_defaults_to_default_batch_size() {
200        let config = S3SinkConfig::new("my-bucket");
201        assert_eq!(config.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
202    }
203
204    #[test]
205    fn with_batch_size_overrides_default() {
206        let config = S3SinkConfig::new("my-bucket").with_batch_size(500);
207        assert_eq!(config.batch_size, 500);
208    }
209
210    #[test]
211    fn batch_size_zero_is_accepted_as_no_batching_sentinel() {
212        let config = S3SinkConfig::new("my-bucket").with_batch_size(0);
213        assert_eq!(config.batch_size, 0);
214        assert!(faucet_core::validate_batch_size(config.batch_size).is_ok());
215    }
216
217    #[test]
218    fn batch_size_above_max_is_rejected_by_validate_batch_size() {
219        let config =
220            S3SinkConfig::new("my-bucket").with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
221        assert!(faucet_core::validate_batch_size(config.batch_size).is_err());
222    }
223
224    #[test]
225    fn batch_size_deserializes_from_json() {
226        let json = r#"{
227            "bucket": "my-bucket",
228            "prefix": "",
229            "region": null,
230            "endpoint_url": null,
231            "file_extension": ".jsonl",
232            "max_records_per_file": null,
233            "concurrency": 10,
234            "batch_size": 250
235        }"#;
236        let config: S3SinkConfig = serde_json::from_str(json).unwrap();
237        assert_eq!(config.batch_size, 250);
238    }
239
240    #[test]
241    fn batch_size_defaults_when_omitted_from_json() {
242        let json = r#"{
243            "bucket": "my-bucket",
244            "prefix": "",
245            "region": null,
246            "endpoint_url": null,
247            "file_extension": ".jsonl",
248            "max_records_per_file": null,
249            "concurrency": 10
250        }"#;
251        let config: S3SinkConfig = serde_json::from_str(json).unwrap();
252        assert_eq!(config.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
253    }
254
255    #[cfg(feature = "compression")]
256    #[test]
257    fn compression_config_round_trips() {
258        let json = r#"{
259            "bucket": "b",
260            "prefix": "",
261            "region": null,
262            "endpoint_url": null,
263            "file_extension": ".jsonl.gz",
264            "max_records_per_file": null,
265            "concurrency": 1,
266            "batch_size": 0,
267            "compression": "gzip"
268        }"#;
269        let config: S3SinkConfig = serde_json::from_str(json).unwrap();
270        assert_eq!(config.compression, faucet_core::CompressionConfig::Gzip);
271    }
272
273    #[cfg(feature = "compression")]
274    #[test]
275    fn compression_default_is_auto() {
276        let cfg = S3SinkConfig::new("bucket");
277        assert_eq!(cfg.compression, faucet_core::CompressionConfig::Auto);
278    }
279}
faucet_sink_s3/config.rs

faucet_sink_s3/
config.rs
