faucet_source_delta/
config.rs

1//! Delta Lake source configuration.
2
3use faucet_common_delta::DeltaConnection;
4use faucet_core::{DEFAULT_BATCH_SIZE, validate_batch_size};
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7
8fn default_batch_size() -> usize {
9    DEFAULT_BATCH_SIZE
10}
11
12/// Configuration for the Delta Lake source connector.
13///
14/// The `table_uri` / `credentials` / `storage_options` keys are flattened in
15/// from the shared [`DeltaConnection`]. Reads scan the table's active data
16/// files at the latest version (or a pinned `version` / `timestamp` for time
17/// travel) and yield each row as a JSON object.
18#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
19pub struct DeltaSourceConfig {
20    /// Table location + object-store credentials.
21    #[serde(flatten)]
22    pub connection: DeltaConnection,
23
24    /// Time travel: read the table as of this version. Mutually exclusive with
25    /// `timestamp`.
26    #[serde(default, skip_serializing_if = "Option::is_none")]
27    pub version: Option<u64>,
28
29    /// Time travel: read the table as of this RFC 3339 timestamp. Mutually
30    /// exclusive with `version`.
31    #[serde(default, skip_serializing_if = "Option::is_none")]
32    pub timestamp: Option<String>,
33
34    /// Projection pushdown: only read these columns. Empty = all columns.
35    #[serde(default)]
36    pub columns: Vec<String>,
37
38    /// Page size hint for [`stream_pages`](faucet_core::Source::stream_pages).
39    /// Governs how many rows accumulate before a `StreamPage` is emitted; the
40    /// underlying parquet row-group size still bounds each read. `0` is the
41    /// "no batching" sentinel — one page per file.
42    #[serde(default = "default_batch_size")]
43    pub batch_size: usize,
44}
45
46impl DeltaSourceConfig {
47    /// Build a source config for `table_uri` with defaults.
48    pub fn new(table_uri: impl Into<String>) -> Self {
49        Self {
50            connection: DeltaConnection {
51                table_uri: table_uri.into(),
52                credentials: Default::default(),
53                storage_options: Default::default(),
54            },
55            version: None,
56            timestamp: None,
57            columns: Vec::new(),
58            batch_size: DEFAULT_BATCH_SIZE,
59        }
60    }
61
62    /// Validate the config; returns a human-readable error string if invalid.
63    pub fn validate(&self) -> Result<(), String> {
64        if self.connection.table_uri.trim().is_empty() {
65            return Err("delta source: `table_uri` must not be empty".to_string());
66        }
67        if self.version.is_some() && self.timestamp.is_some() {
68            return Err(
69                "delta source: `version` and `timestamp` are mutually exclusive".to_string(),
70            );
71        }
72        validate_batch_size(self.batch_size)
73            .map_err(|e| format!("delta source: invalid batch_size: {e}"))?;
74        Ok(())
75    }
76}
77
78#[cfg(test)]
79mod tests {
80    use super::*;
81    use serde_json::json;
82
83    #[test]
84    fn new_has_sane_defaults() {
85        let c = DeltaSourceConfig::new("file:///tmp/t");
86        assert_eq!(c.connection.table_uri, "file:///tmp/t");
87        assert_eq!(c.batch_size, DEFAULT_BATCH_SIZE);
88        assert!(c.columns.is_empty());
89        c.validate().unwrap();
90    }
91
92    #[test]
93    fn rejects_empty_uri() {
94        let mut c = DeltaSourceConfig::new("");
95        assert!(c.validate().is_err());
96        c.connection.table_uri = "   ".into();
97        assert!(c.validate().is_err());
98    }
99
100    #[test]
101    fn rejects_version_and_timestamp_together() {
102        let mut c = DeltaSourceConfig::new("file:///tmp/t");
103        c.version = Some(3);
104        c.timestamp = Some("2026-01-01T00:00:00Z".into());
105        assert!(c.validate().is_err());
106    }
107
108    #[test]
109    fn rejects_oversized_batch() {
110        let mut c = DeltaSourceConfig::new("file:///tmp/t");
111        c.batch_size = faucet_core::MAX_BATCH_SIZE + 1;
112        assert!(c.validate().is_err());
113    }
114
115    #[test]
116    fn deserializes_flattened_shape() {
117        let v = json!({
118            "table_uri": "s3://b/t",
119            "credentials": { "type": "aws", "config": { "region": "us-east-1" } },
120            "columns": ["id", "name"],
121            "version": 5
122        });
123        let c: DeltaSourceConfig = serde_json::from_value(v).unwrap();
124        assert_eq!(c.connection.table_uri, "s3://b/t");
125        assert_eq!(c.columns, vec!["id", "name"]);
126        assert_eq!(c.version, Some(5));
127        c.validate().unwrap();
128    }
129}
faucet_source_delta/config.rs

faucet_source_delta/
config.rs
