laminar_connectors/lakehouse/delta_source_config/
mod.rs

1//! Delta Lake source config. Parsed from the resolved connector config via
2//! [`DeltaSourceConfig::from_config`].
3#![allow(clippy::disallowed_types)] // cold path: lakehouse configuration
4
5use std::collections::HashMap;
6use std::time::Duration;
7
8use crate::config::ConnectorConfig;
9use crate::error::ConnectorError;
10use crate::storage::{CloudConfigValidator, SecretMasker, StorageCredentialResolver};
11
12use super::delta_config::DeltaCatalogType;
13
14/// Configuration for the Delta Lake source connector.
15///
16/// Parsed from resolved source connector options or constructed programmatically.
17#[derive(Debug, Clone)]
18pub struct DeltaSourceConfig {
19    /// Path to the Delta Lake table (local, `s3://`, `az://`, `gs://`).
20    pub table_path: String,
21
22    /// First table version to read. `None` reads only versions committed after startup.
23    pub starting_version: Option<i64>,
24
25    /// How often to poll for new versions (default: 1 second).
26    pub poll_interval: Duration,
27
28    /// Storage options (S3 credentials, Azure keys, etc.).
29    pub storage_options: HashMap<String, String>,
30
31    /// Option keys populated from the environment during parsing.
32    env_resolved_storage_keys: Vec<String>,
33
34    /// Catalog type for table discovery.
35    pub catalog_type: DeltaCatalogType,
36
37    /// Catalog database name (required for Glue).
38    pub catalog_database: Option<String>,
39
40    /// Catalog name (required for Unity).
41    pub catalog_name: Option<String>,
42
43    /// Catalog schema name (required for Unity).
44    pub catalog_schema: Option<String>,
45}
46
47impl Default for DeltaSourceConfig {
48    fn default() -> Self {
49        Self {
50            table_path: String::new(),
51            starting_version: None,
52            poll_interval: Duration::from_secs(1),
53            storage_options: HashMap::new(),
54            env_resolved_storage_keys: Vec::new(),
55            catalog_type: DeltaCatalogType::None,
56            catalog_database: None,
57            catalog_name: None,
58            catalog_schema: None,
59        }
60    }
61}
62
63impl DeltaSourceConfig {
64    /// Creates a minimal config for testing.
65    #[must_use]
66    pub fn new(table_path: &str) -> Self {
67        Self {
68            table_path: table_path.to_string(),
69            ..Default::default()
70        }
71    }
72
73    /// Parses a source config from a resolved [`ConnectorConfig`].
74    ///
75    /// # Required keys
76    ///
77    /// - `table.path` - Path to Delta Lake table
78    ///
79    /// # Errors
80    ///
81    /// Returns `ConnectorError::MissingConfig` if required keys are absent,
82    /// or `ConnectorError::ConfigurationError` on invalid values.
83    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
84        let mut cfg = Self {
85            table_path: config.require("table.path")?.to_string(),
86            ..Self::default()
87        };
88
89        if let Some(v) = config.get("starting.version") {
90            cfg.starting_version = Some(v.parse().map_err(|_| {
91                ConnectorError::ConfigurationError(format!("invalid starting.version: '{v}'"))
92            })?);
93        }
94        if let Some(v) = config.get("poll.interval.ms") {
95            let ms: u64 = v.parse().map_err(|_| {
96                ConnectorError::ConfigurationError(format!("invalid poll.interval.ms: '{v}'"))
97            })?;
98            cfg.poll_interval = Duration::from_millis(ms);
99        }
100        if config.get("read.mode").is_some() {
101            return Err(ConnectorError::ConfigurationError(
102                "Delta source does not support read.mode".into(),
103            ));
104        }
105        if config.get("cdf.enabled").is_some() {
106            return Err(ConnectorError::ConfigurationError(
107                "Delta source does not support cdf.enabled".into(),
108            ));
109        }
110        if config.get("partition.filter").is_some() {
111            return Err(ConnectorError::ConfigurationError(
112                "Delta source does not support partition.filter".into(),
113            ));
114        }
115        if config.get("schema.evolution.action").is_some() {
116            return Err(ConnectorError::ConfigurationError(
117                "Delta CDF source always fails on schema evolution; schema.evolution.action is unsupported"
118                    .into(),
119            ));
120        }
121
122        // ── Catalog configuration ──
123        if let Some(v) = config.get("catalog.type") {
124            cfg.catalog_type = v.parse().map_err(|_| {
125                ConnectorError::ConfigurationError(format!(
126                    "invalid catalog.type: '{v}' (expected 'none', 'glue', or 'unity')"
127                ))
128            })?;
129        }
130        if let Some(v) = config.get("catalog.database") {
131            cfg.catalog_database = Some(v.to_string());
132        }
133        if let Some(v) = config.get("catalog.name") {
134            cfg.catalog_name = Some(v.to_string());
135        }
136        if let Some(v) = config.get("catalog.schema") {
137            cfg.catalog_schema = Some(v.to_string());
138        }
139        if let DeltaCatalogType::Unity {
140            ref mut workspace_url,
141            ref mut access_token,
142        } = cfg.catalog_type
143        {
144            if let Some(v) = config.get("catalog.workspace_url") {
145                *workspace_url = v.to_string();
146            }
147            if let Some(v) = config.get("catalog.access_token") {
148                *access_token = v.to_string();
149            }
150        }
151        // Resolve storage credentials.
152        let explicit_storage = config.properties_with_prefix("storage.");
153        let resolved = StorageCredentialResolver::resolve(&cfg.table_path, &explicit_storage);
154        cfg.env_resolved_storage_keys = resolved.env_resolved_keys;
155        cfg.storage_options = resolved.options;
156
157        // Map LogStore configuration keys to delta-rs storage options.
158        if let Some(v) = config.get("storage.s3_locking_provider") {
159            cfg.storage_options
160                .insert("AWS_S3_LOCKING_PROVIDER".to_string(), v.to_string());
161        }
162        if let Some(v) = config.get("storage.dynamodb_table_name") {
163            cfg.storage_options
164                .insert("DELTA_DYNAMO_TABLE_NAME".to_string(), v.to_string());
165        }
166
167        cfg.validate()?;
168        Ok(cfg)
169    }
170
171    /// Returns configured options without values copied from the environment.
172    #[cfg(feature = "delta-lake")]
173    pub(crate) fn stable_storage_options(&self) -> HashMap<String, String> {
174        let mut options = self.storage_options.clone();
175        for key in &self.env_resolved_storage_keys {
176            options.remove(key);
177        }
178        options
179    }
180
181    /// Formats the storage options for safe logging with secrets redacted.
182    #[must_use]
183    pub fn display_storage_options(&self) -> String {
184        SecretMasker::display_map(&self.storage_options)
185    }
186
187    /// Validates the configuration for consistency.
188    ///
189    /// # Errors
190    ///
191    /// Returns `ConnectorError::ConfigurationError` on invalid combinations.
192    pub fn validate(&self) -> Result<(), ConnectorError> {
193        if self.table_path.is_empty() {
194            return Err(ConnectorError::missing_config("table.path"));
195        }
196        if self.starting_version.is_some_and(|version| version < 0) {
197            return Err(ConnectorError::ConfigurationError(
198                "starting.version must be non-negative".into(),
199            ));
200        }
201
202        // Validate catalog-specific requirements.
203        match &self.catalog_type {
204            DeltaCatalogType::None => {}
205            DeltaCatalogType::Glue => {
206                if self.catalog_database.is_none() {
207                    return Err(ConnectorError::ConfigurationError(
208                        "Glue catalog requires 'catalog.database' to be set".into(),
209                    ));
210                }
211            }
212            DeltaCatalogType::Unity {
213                workspace_url,
214                access_token,
215            } => {
216                if workspace_url.is_empty() {
217                    return Err(ConnectorError::ConfigurationError(
218                        "Unity catalog requires 'catalog.workspace_url' to be set".into(),
219                    ));
220                }
221                if access_token.is_empty() {
222                    return Err(ConnectorError::ConfigurationError(
223                        "Unity catalog requires 'catalog.access_token' to be set".into(),
224                    ));
225                }
226                if self.catalog_name.is_none() {
227                    return Err(ConnectorError::ConfigurationError(
228                        "Unity catalog requires 'catalog.name' to be set".into(),
229                    ));
230                }
231                if self.catalog_schema.is_none() {
232                    return Err(ConnectorError::ConfigurationError(
233                        "Unity catalog requires 'catalog.schema' to be set".into(),
234                    ));
235                }
236            }
237        }
238
239        // Validate cloud storage credentials for the detected provider.
240        let resolved = StorageCredentialResolver::resolve(&self.table_path, &self.storage_options);
241        let cloud_result = CloudConfigValidator::validate(&resolved);
242        if !cloud_result.is_valid() {
243            return Err(ConnectorError::ConfigurationError(
244                cloud_result.error_message(),
245            ));
246        }
247
248        Ok(())
249    }
250}
251
252#[cfg(test)]
253mod tests;
laminar_connectors/lakehouse/delta_source_config/mod.rs

laminar_connectors/lakehouse/delta_source_config/
mod.rs
