faucet_sink_gcs/
config.rs

1//! GCS sink configuration.
2
3use faucet_common_gcs::GcsCredentials;
4use faucet_core::DEFAULT_BATCH_SIZE;
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7
8/// On-the-wire format of objects written by the GCS sink.
9#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
10#[serde(rename_all = "snake_case")]
11pub enum GcsSinkFormat {
12    /// Newline-delimited JSON — one JSON record per line (the default).
13    #[default]
14    JsonLines,
15    /// Apache Parquet. Each written object is a complete, self-contained
16    /// Parquet file. Enables the **columnar** fast path
17    /// ([`Sink::write_batch_columnar`](faucet_core::Sink::write_batch_columnar))
18    /// so a `parquet`/`delta` → `gcs(parquet)` chain never materializes
19    /// `serde_json::Value`. Requires the crate-local `arrow` feature
20    /// (RFC 0002 / #375).
21    #[cfg(feature = "arrow")]
22    Parquet,
23}
24
25/// Configuration for the GCS sink connector.
26#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
27pub struct GcsSinkConfig {
28    /// GCS bucket name.
29    pub bucket: String,
30    /// Object-name prefix for written files.
31    pub prefix: String,
32    /// Object format (default: `json_lines`). Set to `parquet` (with the
33    /// `arrow` feature) to write Parquet objects and enable the columnar
34    /// fast path.
35    #[serde(default)]
36    pub format: GcsSinkFormat,
37    /// Credential source.
38    #[serde(default)]
39    pub auth: GcsCredentials,
40    /// File extension for written objects (default `.jsonl`).
41    #[serde(default = "default_file_extension")]
42    pub file_extension: String,
43    /// Hard cap on records per uploaded object. `None` means a single
44    /// object per `write_batch` call (still subject to `batch_size`).
45    pub max_records_per_file: Option<usize>,
46    /// Maximum number of concurrent uploads (default 10).
47    #[serde(default = "default_concurrency")]
48    pub concurrency: usize,
49    /// Records per uploaded object from a single `write_batch` call.
50    /// `batch_size = 0` writes whatever upstream hands the sink as one
51    /// object. Recommended value for GCS is `0` — many tiny objects is
52    /// a well-known anti-pattern.
53    #[serde(default = "default_batch_size")]
54    pub batch_size: usize,
55    /// Optional storage-host override (integration-test escape hatch).
56    pub storage_host: Option<String>,
57    /// Compression codec applied to each uploaded object body. Defaults to
58    /// [`CompressionConfig::Auto`](faucet_core::CompressionConfig::Auto) —
59    /// resolves against `file_extension` (so `.jsonl.gz` triggers gzip).
60    /// Requires the crate-local `compression` feature. Note: this sink does
61    /// **not** set the GCS `Content-Encoding` metadata, so consumers must
62    /// decompress explicitly.
63    #[cfg(feature = "compression")]
64    #[serde(default)]
65    pub compression: faucet_core::CompressionConfig,
66}
67
68fn default_file_extension() -> String {
69    ".jsonl".to_string()
70}
71fn default_batch_size() -> usize {
72    DEFAULT_BATCH_SIZE
73}
74fn default_concurrency() -> usize {
75    10
76}
77
78impl GcsSinkConfig {
79    pub fn new(bucket: impl Into<String>) -> Self {
80        Self {
81            bucket: bucket.into(),
82            prefix: String::new(),
83            format: GcsSinkFormat::default(),
84            auth: GcsCredentials::default(),
85            file_extension: default_file_extension(),
86            max_records_per_file: None,
87            concurrency: default_concurrency(),
88            batch_size: default_batch_size(),
89            storage_host: None,
90            #[cfg(feature = "compression")]
91            compression: faucet_core::CompressionConfig::Auto,
92        }
93    }
94
95    pub fn prefix(mut self, p: impl Into<String>) -> Self {
96        self.prefix = p.into();
97        self
98    }
99    /// Set the object format (`json_lines` or, with the `arrow` feature,
100    /// `parquet`).
101    pub fn format(mut self, format: GcsSinkFormat) -> Self {
102        self.format = format;
103        self
104    }
105    pub fn auth(mut self, c: GcsCredentials) -> Self {
106        self.auth = c;
107        self
108    }
109    pub fn file_extension(mut self, ext: impl Into<String>) -> Self {
110        self.file_extension = ext.into();
111        self
112    }
113    pub fn max_records_per_file(mut self, n: usize) -> Self {
114        self.max_records_per_file = Some(n);
115        self
116    }
117    pub fn concurrency(mut self, n: usize) -> Self {
118        self.concurrency = n;
119        self
120    }
121    pub fn with_batch_size(mut self, n: usize) -> Self {
122        self.batch_size = n;
123        self
124    }
125    pub fn storage_host(mut self, h: impl Into<String>) -> Self {
126        self.storage_host = Some(h.into());
127        self
128    }
129
130    /// Set the compression codec. Available only with the `compression` feature.
131    #[cfg(feature = "compression")]
132    pub fn compression(mut self, c: faucet_core::CompressionConfig) -> Self {
133        self.compression = c;
134        self
135    }
136}
137
138#[cfg(test)]
139mod tests {
140    use super::*;
141
142    #[test]
143    fn defaults() {
144        let c = GcsSinkConfig::new("b");
145        assert_eq!(c.bucket, "b");
146        assert_eq!(c.prefix, "");
147        assert!(matches!(c.auth, GcsCredentials::ApplicationDefault));
148        assert_eq!(c.file_extension, ".jsonl");
149        assert!(c.max_records_per_file.is_none());
150        assert_eq!(c.concurrency, 10);
151        assert_eq!(c.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
152        assert!(c.storage_host.is_none());
153    }
154
155    #[test]
156    fn builder_methods() {
157        let c = GcsSinkConfig::new("b")
158            .prefix("out/")
159            .file_extension(".ndjson")
160            .max_records_per_file(500)
161            .concurrency(4)
162            .with_batch_size(0)
163            .storage_host("http://localhost:4443");
164        assert_eq!(c.prefix, "out/");
165        assert_eq!(c.file_extension, ".ndjson");
166        assert_eq!(c.max_records_per_file, Some(500));
167        assert_eq!(c.concurrency, 4);
168        assert_eq!(c.batch_size, 0);
169        assert_eq!(c.storage_host.as_deref(), Some("http://localhost:4443"));
170    }
171
172    #[test]
173    fn batch_size_sentinel_accepted_and_above_max_rejected() {
174        assert!(faucet_core::validate_batch_size(0).is_ok());
175        assert!(faucet_core::validate_batch_size(faucet_core::MAX_BATCH_SIZE + 1).is_err());
176    }
177
178    #[test]
179    fn batch_size_defaults_when_omitted_from_json() {
180        let json = r#"{
181            "bucket": "b",
182            "prefix": "p/",
183            "max_records_per_file": null,
184            "concurrency": 10,
185            "storage_host": null
186        }"#;
187        let c: GcsSinkConfig = serde_json::from_str(json).unwrap();
188        assert_eq!(c.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
189    }
190
191    #[cfg(feature = "compression")]
192    #[test]
193    fn compression_default_is_auto() {
194        let cfg = GcsSinkConfig::new("bucket");
195        assert_eq!(cfg.compression, faucet_core::CompressionConfig::Auto);
196    }
197
198    #[cfg(feature = "compression")]
199    #[test]
200    fn compression_config_round_trips() {
201        let json = r#"{
202            "bucket": "b",
203            "prefix": "",
204            "file_extension": ".jsonl.gz",
205            "max_records_per_file": null,
206            "concurrency": 1,
207            "batch_size": 0,
208            "storage_host": null,
209            "compression": "gzip"
210        }"#;
211        let cfg: GcsSinkConfig = serde_json::from_str(json).unwrap();
212        assert_eq!(cfg.compression, faucet_core::CompressionConfig::Gzip);
213    }
214
215    #[cfg(feature = "arrow")]
216    #[test]
217    fn format_defaults_json_lines_and_builder_sets_parquet() {
218        assert_eq!(GcsSinkConfig::new("b").format, GcsSinkFormat::JsonLines);
219        let cfg = GcsSinkConfig::new("b").format(GcsSinkFormat::Parquet);
220        assert_eq!(cfg.format, GcsSinkFormat::Parquet);
221    }
222}
faucet_sink_gcs/config.rs

faucet_sink_gcs/
config.rs
