faucet_source_gcs/
config.rs

1//! GCS source configuration.
2
3use faucet_common_gcs::GcsCredentials;
4use faucet_core::DEFAULT_BATCH_SIZE;
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7
8/// Format of files stored in GCS.
9#[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema)]
10#[serde(rename_all = "snake_case")]
11pub enum GcsFileFormat {
12    /// Each line in the file is a separate JSON record.
13    #[default]
14    JsonLines,
15    /// The entire file is a JSON array of records.
16    JsonArray,
17    /// Each file becomes a single record with `"key"` and `"content"` fields.
18    RawText,
19    /// Apache Parquet objects. Decoded via the Arrow Parquet reader; the
20    /// resulting `RecordBatch`es feed both the row path (converted to JSON)
21    /// and the **columnar** fast path
22    /// ([`Source::stream_batches`](faucet_core::Source::stream_batches)) so a
23    /// `gcs(parquet) → parquet`/`delta` chain never materializes
24    /// `serde_json::Value`. Requires the crate-local `arrow` feature
25    /// (RFC 0002 / #375).
26    #[cfg(feature = "arrow")]
27    Parquet,
28}
29
30/// Configuration for the GCS source connector.
31#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
32pub struct GcsSourceConfig {
33    /// GCS bucket name.
34    pub bucket: String,
35    /// Object name prefix filter. Ignored when `object_keys` is set.
36    pub prefix: Option<String>,
37    /// Explicit object names. When set, listing is skipped and `prefix`
38    /// is ignored.
39    pub object_keys: Option<Vec<String>>,
40    /// Credential source.
41    #[serde(default)]
42    pub auth: GcsCredentials,
43    /// File format.
44    #[serde(default)]
45    pub file_format: GcsFileFormat,
46    /// Hard cap on the number of objects read (after listing).
47    pub max_objects: Option<usize>,
48    /// Maximum concurrent object reads (default: 10).
49    #[serde(default = "default_concurrency")]
50    pub concurrency: usize,
51    /// Records per emitted `StreamPage`. See "Streaming and batching"
52    /// in the README. `batch_size = 0` is the "no batching" sentinel and
53    /// emits one page per object.
54    #[serde(default = "default_batch_size")]
55    pub batch_size: usize,
56    /// Verify each object's byte length against the `size` GCS reports for it,
57    /// failing the read with [`FaucetError::Source`](faucet_core::FaucetError::Source)
58    /// on a short (truncated) or over-long transfer (#161). Cheap (a byte
59    /// counter over the body that is read anyway) and defaults to `true`.
60    /// The check is automatically skipped for an object served with a
61    /// non-empty `Content-Encoding` (GCS may decompressively transcode it on
62    /// read, so the received byte count would not match the stored `size`).
63    #[serde(default = "default_true")]
64    pub verify_length: bool,
65    /// Verify each object's body against the CRC32C (or MD5) checksum GCS
66    /// reports for it (#161). Stronger than the length check but costs a hash
67    /// over the full body, so it defaults to `false`. Skipped for an object
68    /// with no usable checksum or one served with a non-empty
69    /// `Content-Encoding` (the stored checksum covers the stored bytes, which
70    /// transcoding would not return).
71    #[serde(default)]
72    pub verify_checksum: bool,
73    /// Optional storage-host override (e.g. `http://localhost:4443` for
74    /// fake-gcs-server). Production users should leave this unset.
75    pub storage_host: Option<String>,
76    /// Compression codec applied to each downloaded object. Defaults to
77    /// [`CompressionConfig::Auto`](faucet_core::CompressionConfig::Auto) —
78    /// the codec is resolved per-object-key, so a single source can read a
79    /// mix of compressed and uncompressed objects. Requires the
80    /// crate-local `compression` feature.
81    #[cfg(feature = "compression")]
82    #[serde(default)]
83    pub compression: faucet_core::CompressionConfig,
84}
85
86fn default_batch_size() -> usize {
87    DEFAULT_BATCH_SIZE
88}
89fn default_concurrency() -> usize {
90    10
91}
92fn default_true() -> bool {
93    true
94}
95
96impl GcsSourceConfig {
97    /// Create a new config with the required bucket name and sensible defaults.
98    pub fn new(bucket: impl Into<String>) -> Self {
99        Self {
100            bucket: bucket.into(),
101            prefix: None,
102            object_keys: None,
103            auth: GcsCredentials::default(),
104            file_format: GcsFileFormat::default(),
105            max_objects: None,
106            concurrency: default_concurrency(),
107            batch_size: default_batch_size(),
108            verify_length: true,
109            verify_checksum: false,
110            storage_host: None,
111            #[cfg(feature = "compression")]
112            compression: faucet_core::CompressionConfig::default(),
113        }
114    }
115
116    pub fn prefix(mut self, prefix: impl Into<String>) -> Self {
117        self.prefix = Some(prefix.into());
118        self
119    }
120
121    pub fn object_keys(mut self, keys: Vec<String>) -> Self {
122        self.object_keys = Some(keys);
123        self
124    }
125
126    pub fn auth(mut self, creds: GcsCredentials) -> Self {
127        self.auth = creds;
128        self
129    }
130
131    pub fn file_format(mut self, format: GcsFileFormat) -> Self {
132        self.file_format = format;
133        self
134    }
135
136    pub fn max_objects(mut self, max: usize) -> Self {
137        self.max_objects = Some(max);
138        self
139    }
140
141    pub fn concurrency(mut self, concurrency: usize) -> Self {
142        self.concurrency = concurrency;
143        self
144    }
145
146    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
147        self.batch_size = batch_size;
148        self
149    }
150
151    pub fn storage_host(mut self, host: impl Into<String>) -> Self {
152        self.storage_host = Some(host.into());
153        self
154    }
155
156    /// Enable or disable the per-object length verification (default `true`).
157    /// See [`verify_length`](Self::verify_length).
158    pub fn verify_length(mut self, verify: bool) -> Self {
159        self.verify_length = verify;
160        self
161    }
162
163    /// Enable or disable per-object checksum verification (default `false`).
164    /// See [`verify_checksum`](Self::verify_checksum).
165    pub fn verify_checksum(mut self, verify: bool) -> Self {
166        self.verify_checksum = verify;
167        self
168    }
169
170    /// Set the compression codec. Available only with the `compression` feature.
171    #[cfg(feature = "compression")]
172    pub fn compression(mut self, c: faucet_core::CompressionConfig) -> Self {
173        self.compression = c;
174        self
175    }
176
177    /// Validate the config at load time so a bad config fails fast with a typed
178    /// `FaucetError::Config` instead of surfacing deep in a run: rejects an
179    /// out-of-range `batch_size` (`> MAX_BATCH_SIZE`) and an empty `bucket`.
180    ///
181    /// `faucet_core` is referenced by full path here (rather than imported) so
182    /// the field-doc links above keep their explicit targets and the committed
183    /// config JSON Schema stays byte-identical.
184    pub fn validate(&self) -> Result<(), faucet_core::FaucetError> {
185        if self.bucket.trim().is_empty() {
186            return Err(faucet_core::FaucetError::Config(
187                "GCS source requires a non-empty `bucket`".into(),
188            ));
189        }
190        faucet_core::validate_batch_size(self.batch_size)?;
191        Ok(())
192    }
193}
194
195#[cfg(test)]
196mod tests {
197    use super::*;
198
199    #[test]
200    fn default_config() {
201        let config = GcsSourceConfig::new("my-bucket");
202        assert_eq!(config.bucket, "my-bucket");
203        assert!(config.prefix.is_none());
204        assert!(config.object_keys.is_none());
205        assert!(matches!(config.auth, GcsCredentials::ApplicationDefault));
206        assert!(matches!(config.file_format, GcsFileFormat::JsonLines));
207        assert!(config.max_objects.is_none());
208        assert_eq!(config.concurrency, 10);
209        assert_eq!(config.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
210        assert!(config.storage_host.is_none());
211    }
212
213    #[test]
214    fn builder_methods() {
215        let config = GcsSourceConfig::new("my-bucket")
216            .prefix("data/")
217            .file_format(GcsFileFormat::JsonArray)
218            .max_objects(5)
219            .concurrency(20)
220            .with_batch_size(250)
221            .storage_host("http://localhost:4443");
222
223        assert_eq!(config.bucket, "my-bucket");
224        assert_eq!(config.prefix.as_deref(), Some("data/"));
225        assert!(matches!(config.file_format, GcsFileFormat::JsonArray));
226        assert_eq!(config.max_objects, Some(5));
227        assert_eq!(config.concurrency, 20);
228        assert_eq!(config.batch_size, 250);
229        assert_eq!(
230            config.storage_host.as_deref(),
231            Some("http://localhost:4443")
232        );
233    }
234
235    #[test]
236    fn file_format_default_is_json_lines() {
237        assert!(matches!(GcsFileFormat::default(), GcsFileFormat::JsonLines));
238    }
239
240    #[test]
241    fn batch_size_zero_is_accepted_as_no_batching_sentinel() {
242        let config = GcsSourceConfig::new("b").with_batch_size(0);
243        assert_eq!(config.batch_size, 0);
244        assert!(faucet_core::validate_batch_size(config.batch_size).is_ok());
245    }
246
247    #[test]
248    fn batch_size_above_max_is_rejected() {
249        let config = GcsSourceConfig::new("b").with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
250        assert!(faucet_core::validate_batch_size(config.batch_size).is_err());
251    }
252
253    #[cfg(feature = "compression")]
254    #[test]
255    fn compression_default_is_auto() {
256        let cfg = GcsSourceConfig::new("bucket");
257        assert_eq!(cfg.compression, faucet_core::CompressionConfig::Auto);
258    }
259
260    #[test]
261    fn verify_defaults_length_on_checksum_off() {
262        let cfg = GcsSourceConfig::new("b");
263        assert!(cfg.verify_length);
264        assert!(!cfg.verify_checksum);
265    }
266
267    #[test]
268    fn verify_builders_override() {
269        let cfg = GcsSourceConfig::new("b")
270            .verify_length(false)
271            .verify_checksum(true);
272        assert!(!cfg.verify_length);
273        assert!(cfg.verify_checksum);
274    }
275
276    #[test]
277    fn verify_fields_default_when_absent_from_json() {
278        let json = r#"{
279            "bucket": "my-bucket",
280            "prefix": null,
281            "object_keys": null,
282            "file_format": "json_lines",
283            "max_objects": null,
284            "concurrency": 10,
285            "storage_host": null
286        }"#;
287        let config: GcsSourceConfig = serde_json::from_str(json).unwrap();
288        assert!(config.verify_length);
289        assert!(!config.verify_checksum);
290    }
291
292    #[test]
293    fn batch_size_defaults_when_omitted_from_json() {
294        let json = r#"{
295            "bucket": "my-bucket",
296            "prefix": null,
297            "object_keys": null,
298            "file_format": "json_lines",
299            "max_objects": null,
300            "concurrency": 10,
301            "storage_host": null
302        }"#;
303        let config: GcsSourceConfig = serde_json::from_str(json).unwrap();
304        assert_eq!(config.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
305    }
306
307    #[test]
308    fn validate_accepts_valid_config() {
309        assert!(GcsSourceConfig::new("my-bucket").validate().is_ok());
310    }
311
312    #[test]
313    fn validate_rejects_oversized_batch_size() {
314        let config =
315            GcsSourceConfig::new("my-bucket").with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
316        assert!(matches!(
317            config.validate(),
318            Err(faucet_core::FaucetError::Config(_))
319        ));
320    }
321
322    #[test]
323    fn validate_rejects_empty_bucket() {
324        assert!(matches!(
325            GcsSourceConfig::new("   ").validate(),
326            Err(faucet_core::FaucetError::Config(_))
327        ));
328    }
329}
faucet_source_gcs/config.rs

faucet_source_gcs/
config.rs
