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    /// Whether the source follows the change data feed or reads one snapshot.
23    pub read_mode: DeltaReadMode,
24
25    /// Where a change-feed source begins on its first start.
26    pub starting_version: DeltaStartingVersion,
27
28    /// How often to poll for new versions (default: 1 second).
29    pub poll_interval: Duration,
30
31    /// Storage options (S3 credentials, Azure keys, etc.).
32    pub storage_options: HashMap<String, String>,
33
34    /// Option keys populated from the environment during parsing.
35    env_resolved_storage_keys: Vec<String>,
36
37    /// Catalog type for table discovery.
38    pub catalog_type: DeltaCatalogType,
39
40    /// Catalog database name (required for Glue).
41    pub catalog_database: Option<String>,
42
43    /// Catalog name (required for Unity).
44    pub catalog_name: Option<String>,
45
46    /// Catalog schema name (required for Unity).
47    pub catalog_schema: Option<String>,
48}
49
50impl Default for DeltaSourceConfig {
51    fn default() -> Self {
52        Self {
53            table_path: String::new(),
54            read_mode: DeltaReadMode::Changes,
55            starting_version: DeltaStartingVersion::Snapshot,
56            poll_interval: Duration::from_secs(1),
57            storage_options: HashMap::new(),
58            env_resolved_storage_keys: Vec::new(),
59            catalog_type: DeltaCatalogType::None,
60            catalog_database: None,
61            catalog_name: None,
62            catalog_schema: None,
63        }
64    }
65}
66
67impl DeltaSourceConfig {
68    /// Creates a minimal config for testing.
69    #[must_use]
70    pub fn new(table_path: &str) -> Self {
71        Self {
72            table_path: table_path.to_string(),
73            ..Default::default()
74        }
75    }
76
77    /// Parses a source config from a resolved [`ConnectorConfig`].
78    ///
79    /// # Required keys
80    ///
81    /// - `table.path` - Path to Delta Lake table
82    ///
83    /// # Errors
84    ///
85    /// Returns `ConnectorError::MissingConfig` if required keys are absent,
86    /// or `ConnectorError::ConfigurationError` on invalid values.
87    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
88        let mut cfg = Self {
89            table_path: config.require("table.path")?.to_string(),
90            ..Self::default()
91        };
92
93        if let Some(v) = config.get("read.mode") {
94            cfg.read_mode = match v.to_ascii_lowercase().as_str() {
95                "changes" => DeltaReadMode::Changes,
96                "snapshot" => DeltaReadMode::Snapshot,
97                _ => {
98                    return Err(ConnectorError::ConfigurationError(format!(
99                        "invalid read.mode: '{v}' (expected 'changes' or 'snapshot')"
100                    )))
101                }
102            };
103        }
104        if let Some(v) = config.get("starting.version") {
105            if cfg.read_mode == DeltaReadMode::Snapshot {
106                return Err(ConnectorError::ConfigurationError(
107                    "starting.version applies only to read.mode=changes; read.mode=snapshot \
108                     reads the latest version"
109                        .into(),
110                ));
111            }
112            cfg.starting_version = if v.eq_ignore_ascii_case("latest") {
113                DeltaStartingVersion::Latest
114            } else {
115                DeltaStartingVersion::Version(v.parse().map_err(|_| {
116                    ConnectorError::ConfigurationError(format!(
117                        "invalid starting.version: '{v}' (expected a non-negative version or \
118                         'latest')"
119                    ))
120                })?)
121            };
122        }
123        if let Some(v) = config.get("poll.interval.ms") {
124            let ms: u64 = v.parse().map_err(|_| {
125                ConnectorError::ConfigurationError(format!("invalid poll.interval.ms: '{v}'"))
126            })?;
127            cfg.poll_interval = Duration::from_millis(ms);
128        }
129        if config.get("cdf.enabled").is_some() {
130            return Err(ConnectorError::ConfigurationError(
131                "Delta source does not support cdf.enabled".into(),
132            ));
133        }
134        if config.get("partition.filter").is_some() {
135            return Err(ConnectorError::ConfigurationError(
136                "Delta source does not support partition.filter".into(),
137            ));
138        }
139        if config.get("schema.evolution.action").is_some() {
140            return Err(ConnectorError::ConfigurationError(
141                "Delta CDF source always fails on schema evolution; schema.evolution.action is unsupported"
142                    .into(),
143            ));
144        }
145
146        // ── Catalog configuration ──
147        cfg.catalog_type = DeltaCatalogType::from_config(config)?;
148        if let Some(v) = config.get("catalog.database") {
149            cfg.catalog_database = Some(v.to_string());
150        }
151        if let Some(v) = config.get("catalog.name") {
152            cfg.catalog_name = Some(v.to_string());
153        }
154        if let Some(v) = config.get("catalog.schema") {
155            cfg.catalog_schema = Some(v.to_string());
156        }
157        // Resolve storage credentials.
158        let explicit_storage = config.properties_with_prefix("storage.");
159        let resolved = StorageCredentialResolver::resolve(&cfg.table_path, &explicit_storage);
160        cfg.env_resolved_storage_keys = resolved.env_resolved_keys;
161        cfg.storage_options = resolved.options;
162
163        // Map LogStore configuration keys to delta-rs storage options.
164        if let Some(v) = config.get("storage.s3_locking_provider") {
165            cfg.storage_options
166                .insert("AWS_S3_LOCKING_PROVIDER".to_string(), v.to_string());
167        }
168        if let Some(v) = config.get("storage.dynamodb_table_name") {
169            cfg.storage_options
170                .insert("DELTA_DYNAMO_TABLE_NAME".to_string(), v.to_string());
171        }
172
173        cfg.validate()?;
174        Ok(cfg)
175    }
176
177    /// Returns configured options without values copied from the environment.
178    #[cfg(feature = "delta-lake")]
179    pub(crate) fn stable_storage_options(&self) -> HashMap<String, String> {
180        let mut options = self.storage_options.clone();
181        for key in &self.env_resolved_storage_keys {
182            options.remove(key);
183        }
184        options
185    }
186
187    /// Formats the storage options for safe logging with secrets redacted.
188    #[must_use]
189    pub fn display_storage_options(&self) -> String {
190        SecretMasker::display_map(&self.storage_options)
191    }
192
193    /// Validates the configuration for consistency.
194    ///
195    /// # Errors
196    ///
197    /// Returns `ConnectorError::ConfigurationError` on invalid combinations.
198    pub fn validate(&self) -> Result<(), ConnectorError> {
199        if self.table_path.is_empty() {
200            return Err(ConnectorError::missing_config("table.path"));
201        }
202        if self.read_mode == DeltaReadMode::Snapshot
203            && self.starting_version != DeltaStartingVersion::Snapshot
204        {
205            return Err(ConnectorError::ConfigurationError(
206                "starting.version applies only to read.mode=changes".into(),
207            ));
208        }
209        if self.poll_interval.is_zero() {
210            return Err(ConnectorError::ConfigurationError(
211                "poll.interval.ms must be > 0".into(),
212            ));
213        }
214
215        // Validate catalog-specific requirements.
216        match &self.catalog_type {
217            DeltaCatalogType::None => {}
218            DeltaCatalogType::Glue => {
219                if self.catalog_database.is_none() {
220                    return Err(ConnectorError::ConfigurationError(
221                        "Glue catalog requires 'catalog.database' to be set".into(),
222                    ));
223                }
224            }
225            DeltaCatalogType::Unity(unity) => {
226                unity.validate()?;
227                if self.catalog_name.is_none() || self.catalog_schema.is_none() {
228                    return Err(ConnectorError::ConfigurationError(
229                        "Unity catalog requires 'catalog.name' and 'catalog.schema' to be set"
230                            .into(),
231                    ));
232                }
233            }
234        }
235
236        // Validate cloud storage credentials for the detected provider.
237        let resolved = StorageCredentialResolver::resolve(&self.table_path, &self.storage_options);
238        let cloud_result = CloudConfigValidator::validate(&resolved);
239        if !cloud_result.is_valid() {
240            return Err(ConnectorError::ConfigurationError(
241                cloud_result.error_message(),
242            ));
243        }
244
245        Ok(())
246    }
247}
248
249/// What a Delta source reads.
250#[derive(Debug, Clone, Copy, PartialEq, Eq)]
251pub enum DeltaReadMode {
252    /// Follow the table's change data feed (`read.mode = 'changes'`, the default).
253    Changes,
254    /// Emit the latest snapshot as inserts and then finish (`read.mode = 'snapshot'`).
255    Snapshot,
256}
257
258/// Where a change-feed source begins on its first start. Recovery always resumes from the
259/// checkpointed cursor instead.
260#[derive(Debug, Clone, Copy, PartialEq, Eq)]
261pub enum DeltaStartingVersion {
262    /// Emit the snapshot at the version current at first start as inserts, then every change
263    /// committed after it (`starting.version` unset).
264    Snapshot,
265    /// Emit only changes committed after the version current at first start
266    /// (`starting.version = 'latest'`).
267    Latest,
268    /// Emit changes from this version, inclusive, without a snapshot. Rows that existed
269    /// before it reach consumers only through their later updates and deletes.
270    Version(u64),
271}
272
273#[cfg(test)]
274mod tests;
laminar_connectors/lakehouse/delta_source_config/mod.rs

laminar_connectors/lakehouse/delta_source_config/
mod.rs
