faucet_sink_delta/
config.rs

1//! Delta Lake sink configuration.
2
3use faucet_common_delta::DeltaConnection;
4use faucet_core::{DEFAULT_BATCH_SIZE, validate_batch_size};
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7
8/// Default number of records sampled to infer the table schema on first write.
9pub const DEFAULT_SAMPLE_SIZE: usize = 100;
10
11fn default_true() -> bool {
12    true
13}
14
15fn default_sample_size() -> usize {
16    DEFAULT_SAMPLE_SIZE
17}
18
19fn default_batch_size() -> usize {
20    DEFAULT_BATCH_SIZE
21}
22
23/// Configuration for the Delta Lake sink connector (append-only in v1).
24///
25/// The `table_uri` / `credentials` / `storage_options` keys are flattened in
26/// from the shared [`DeltaConnection`]. On first write the sink opens the
27/// table (creating it from the inferred schema when `create_if_not_missing`),
28/// then appends one Delta commit per `flush()`.
29#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
30pub struct DeltaSinkConfig {
31    /// Table location + object-store credentials.
32    #[serde(flatten)]
33    pub connection: DeltaConnection,
34
35    /// Create the table (and its schema, from the inferred record shape) on the
36    /// first write when it does not already exist. When `false`, an absent
37    /// table fails the first write with a typed error.
38    #[serde(default = "default_true")]
39    pub create_if_not_missing: bool,
40
41    /// Partition columns, applied only when the table is created. Ignored when
42    /// appending to an existing table (its stored partitioning is used).
43    #[serde(default)]
44    pub partition_by: Vec<String>,
45
46    /// Records sampled to infer the Arrow/Delta schema when creating the table.
47    #[serde(default = "default_sample_size")]
48    pub schema_sample_size: usize,
49
50    /// Re-chunk size for [`Sink::write_batch`](faucet_core::Sink::write_batch).
51    /// Incoming pages larger than this are sliced into `batch_size`-row Arrow
52    /// batches before writing. `0` writes each page through as one batch.
53    #[serde(default = "default_batch_size")]
54    pub batch_size: usize,
55
56    /// Optional target data-file size (bytes) hint for the writer's rollover.
57    /// Advisory — delta-rs rolls files at buffer boundaries near this size.
58    #[serde(default, skip_serializing_if = "Option::is_none")]
59    pub target_file_size: Option<usize>,
60}
61
62impl DeltaSinkConfig {
63    /// Build a sink config for `table_uri` with defaults.
64    pub fn new(table_uri: impl Into<String>) -> Self {
65        Self {
66            connection: DeltaConnection {
67                table_uri: table_uri.into(),
68                credentials: Default::default(),
69                storage_options: Default::default(),
70            },
71            create_if_not_missing: true,
72            partition_by: Vec::new(),
73            schema_sample_size: DEFAULT_SAMPLE_SIZE,
74            batch_size: DEFAULT_BATCH_SIZE,
75            target_file_size: None,
76        }
77    }
78
79    /// Validate the config; returns a human-readable error string if invalid.
80    pub fn validate(&self) -> Result<(), String> {
81        if self.connection.table_uri.trim().is_empty() {
82            return Err("delta sink: `table_uri` must not be empty".to_string());
83        }
84        if self.schema_sample_size == 0 {
85            return Err("delta sink: `schema_sample_size` must be greater than 0".to_string());
86        }
87        if matches!(self.target_file_size, Some(0)) {
88            return Err(
89                "delta sink: `target_file_size` must be greater than 0 when set".to_string(),
90            );
91        }
92        validate_batch_size(self.batch_size)
93            .map_err(|e| format!("delta sink: invalid batch_size: {e}"))?;
94        Ok(())
95    }
96
97    /// Effective schema sample size (never zero after validation).
98    pub fn effective_sample_size(&self) -> usize {
99        self.schema_sample_size.max(1)
100    }
101}
102
103#[cfg(test)]
104mod tests {
105    use super::*;
106    use serde_json::json;
107
108    #[test]
109    fn new_has_sane_defaults() {
110        let c = DeltaSinkConfig::new("file:///tmp/t");
111        assert!(c.create_if_not_missing);
112        assert_eq!(c.schema_sample_size, DEFAULT_SAMPLE_SIZE);
113        assert_eq!(c.batch_size, DEFAULT_BATCH_SIZE);
114        assert!(c.partition_by.is_empty());
115        c.validate().unwrap();
116    }
117
118    #[test]
119    fn rejects_empty_uri_and_zero_sample() {
120        assert!(DeltaSinkConfig::new("").validate().is_err());
121        let mut c = DeltaSinkConfig::new("file:///tmp/t");
122        c.schema_sample_size = 0;
123        assert!(c.validate().is_err());
124    }
125
126    #[test]
127    fn rejects_zero_target_file_size() {
128        let mut c = DeltaSinkConfig::new("file:///tmp/t");
129        c.target_file_size = Some(0);
130        assert!(c.validate().is_err());
131    }
132
133    #[test]
134    fn rejects_oversized_batch() {
135        let mut c = DeltaSinkConfig::new("file:///tmp/t");
136        c.batch_size = faucet_core::MAX_BATCH_SIZE + 1;
137        assert!(c.validate().is_err());
138    }
139
140    #[test]
141    fn deserializes_minimal_applies_defaults() {
142        // Only `table_uri` → every `#[serde(default = …)]` fn runs.
143        let c: DeltaSinkConfig =
144            serde_json::from_value(json!({ "table_uri": "file:///t" })).unwrap();
145        assert!(c.create_if_not_missing);
146        assert_eq!(c.schema_sample_size, DEFAULT_SAMPLE_SIZE);
147        assert_eq!(c.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
148        assert_eq!(c.effective_sample_size(), DEFAULT_SAMPLE_SIZE);
149        assert!(c.target_file_size.is_none());
150        c.validate().unwrap();
151    }
152
153    #[test]
154    fn deserializes_flattened_shape_with_partitions() {
155        let v = json!({
156            "table_uri": "s3://b/t",
157            "partition_by": ["dt"],
158            "create_if_not_missing": false,
159            "storage_options": { "AWS_ALLOW_HTTP": "true" }
160        });
161        let c: DeltaSinkConfig = serde_json::from_value(v).unwrap();
162        assert_eq!(c.partition_by, vec!["dt"]);
163        assert!(!c.create_if_not_missing);
164        assert_eq!(
165            c.connection.merged_storage_options()["AWS_ALLOW_HTTP"],
166            "true"
167        );
168        c.validate().unwrap();
169    }
170}
faucet_sink_delta/config.rs

faucet_sink_delta/
config.rs
