laminar_connectors/lakehouse/delta_config/
mod.rs

1//! Delta Lake sink config. Parsed from the resolved connector config via
2//! [`DeltaLakeSinkConfig::from_config`].
3#![allow(clippy::disallowed_types)] // cold path: lakehouse configuration
4
5use std::collections::HashMap;
6use std::fmt;
7use std::str::FromStr;
8use std::time::Duration;
9
10use crate::connector::DeliveryGuarantee;
11
12use crate::config::ConnectorConfig;
13use crate::error::ConnectorError;
14use crate::storage::{CloudConfigValidator, SecretMasker, StorageCredentialResolver};
15
16/// Configuration for the Delta Lake sink connector.
17///
18/// Parsed from resolved sink connector options or constructed programmatically.
19#[derive(Debug, Clone)]
20pub struct DeltaLakeSinkConfig {
21    /// Path to the Delta Lake table (local, `s3://`, `az://`, `gs://`).
22    pub table_path: String,
23
24    /// Columns to partition by (e.g., `["trade_date", "hour"]`).
25    pub partition_columns: Vec<String>,
26
27    /// Target Parquet file size in bytes (default: 128 MB).
28    pub target_file_size: usize,
29
30    /// Maximum number of records to buffer before flushing to Parquet.
31    pub max_buffer_records: usize,
32
33    /// Maximum time to buffer records before flushing.
34    pub max_buffer_duration: Duration,
35
36    /// Whether to enable schema evolution (auto-merge new columns).
37    pub schema_evolution: bool,
38
39    /// Write mode: Append, Overwrite, or Upsert (CDC merge).
40    pub write_mode: DeltaWriteMode,
41
42    /// Key columns for upsert/merge operations (required for Upsert mode).
43    pub merge_key_columns: Vec<String>,
44
45    /// Storage options (S3 credentials, Azure keys, etc.).
46    pub storage_options: HashMap<String, String>,
47
48    /// Delivery guarantee: `AtLeastOnce` or `ExactlyOnce`.
49    pub delivery_guarantee: DeliveryGuarantee,
50
51    /// Catalog type for table discovery.
52    pub catalog_type: DeltaCatalogType,
53
54    /// Catalog database name (required for Glue).
55    pub catalog_database: Option<String>,
56
57    /// Catalog name (required for Unity).
58    pub catalog_name: Option<String>,
59
60    /// Catalog schema name (required for Unity).
61    pub catalog_schema: Option<String>,
62
63    /// Storage location for auto-created Unity Catalog external tables.
64    /// When set and the `uc://` table doesn't exist, the sink creates it
65    /// via the Unity Catalog REST API at this storage location.
66    pub catalog_storage_location: Option<String>,
67
68    /// End-to-end timeout for one Delta write, including table reopen and all
69    /// optimistic commit retries (default: 30s).
70    pub write_timeout: Duration,
71
72    /// Parquet writer properties (compression, bloom filters, statistics, etc.).
73    pub parquet: ParquetWriteConfig,
74}
75
76impl Default for DeltaLakeSinkConfig {
77    fn default() -> Self {
78        Self {
79            table_path: String::new(),
80            partition_columns: Vec::new(),
81            target_file_size: 128 * 1024 * 1024, // 128 MB
82            max_buffer_records: 100_000,
83            max_buffer_duration: Duration::from_secs(60),
84            schema_evolution: false,
85            write_mode: DeltaWriteMode::Append,
86            merge_key_columns: Vec::new(),
87            storage_options: HashMap::new(),
88            delivery_guarantee: DeliveryGuarantee::AtLeastOnce,
89            catalog_type: DeltaCatalogType::None,
90            catalog_database: None,
91            catalog_name: None,
92            catalog_schema: None,
93            catalog_storage_location: None,
94            write_timeout: Duration::from_secs(30),
95            parquet: ParquetWriteConfig::default(),
96        }
97    }
98}
99
100impl DeltaLakeSinkConfig {
101    /// Creates a minimal config for testing.
102    #[must_use]
103    pub fn new(table_path: &str) -> Self {
104        Self {
105            table_path: table_path.to_string(),
106            ..Default::default()
107        }
108    }
109
110    /// Parses a sink config from a resolved [`ConnectorConfig`].
111    ///
112    /// # Required keys
113    ///
114    /// - `table.path` - Path to Delta Lake table
115    ///
116    /// # Errors
117    ///
118    /// Returns `ConnectorError::MissingConfig` if required keys are absent,
119    /// or `ConnectorError::ConfigurationError` on invalid values.
120    #[allow(clippy::too_many_lines)]
121    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
122        let mut cfg = Self {
123            table_path: config.require("table.path")?.to_string(),
124            ..Self::default()
125        };
126
127        if let Some(v) = config.get("partition.columns") {
128            cfg.partition_columns = v
129                .split(',')
130                .map(|c| c.trim().to_string())
131                .filter(|c| !c.is_empty())
132                .collect();
133        }
134        if let Some(v) = config.get("target.file.size") {
135            cfg.target_file_size = v.parse().map_err(|_| {
136                ConnectorError::ConfigurationError(format!("invalid target.file.size: '{v}'"))
137            })?;
138        }
139        if let Some(v) = config.get("max.buffer.records") {
140            cfg.max_buffer_records = v.parse().map_err(|_| {
141                ConnectorError::ConfigurationError(format!("invalid max.buffer.records: '{v}'"))
142            })?;
143        }
144        if let Some(v) = config.get("max.buffer.duration.ms") {
145            let ms: u64 = v.parse().map_err(|_| {
146                ConnectorError::ConfigurationError(format!("invalid max.buffer.duration.ms: '{v}'"))
147            })?;
148            cfg.max_buffer_duration = Duration::from_millis(ms);
149        }
150        if let Some(v) = config.get("schema.evolution") {
151            cfg.schema_evolution = v.eq_ignore_ascii_case("true");
152        }
153        if let Some(v) = config.get("write.mode") {
154            cfg.write_mode = v.parse().map_err(|_| {
155                ConnectorError::ConfigurationError(format!(
156                    "invalid write.mode: '{v}' (expected 'append', 'overwrite', or 'upsert')"
157                ))
158            })?;
159        }
160        if let Some(v) = config.get("merge.key.columns") {
161            cfg.merge_key_columns = v
162                .split(',')
163                .map(|c| c.trim().to_string())
164                .filter(|c| !c.is_empty())
165                .collect();
166        }
167        if let Some(v) = config.get("delivery.guarantee") {
168            cfg.delivery_guarantee = v.parse().map_err(|_| {
169                ConnectorError::ConfigurationError(format!(
170                    "invalid delivery.guarantee: '{v}' \
171                     (expected 'exactly-once' or 'at-least-once')"
172                ))
173            })?;
174        }
175        if cfg.delivery_guarantee == DeliveryGuarantee::ExactlyOnce
176            && cfg.write_mode != DeltaWriteMode::Append
177        {
178            return Err(ConnectorError::ConfigurationError(
179                "Delta exactly-once is supported only for coordinated append mode; \
180                 upsert/overwrite do not expose a certified distributed committable"
181                    .into(),
182            ));
183        }
184
185        // ── Catalog configuration ──
186        if let Some(v) = config.get("catalog.type") {
187            cfg.catalog_type = v.parse().map_err(|_| {
188                ConnectorError::ConfigurationError(format!(
189                    "invalid catalog.type: '{v}' (expected 'none', 'glue', or 'unity')"
190                ))
191            })?;
192        }
193        if let Some(v) = config.get("catalog.database") {
194            cfg.catalog_database = Some(v.to_string());
195        }
196        if let Some(v) = config.get("catalog.name") {
197            cfg.catalog_name = Some(v.to_string());
198        }
199        if let Some(v) = config.get("catalog.schema") {
200            cfg.catalog_schema = Some(v.to_string());
201        }
202        // Unity-specific: populate workspace_url and access_token into the enum variant.
203        if let DeltaCatalogType::Unity {
204            ref mut workspace_url,
205            ref mut access_token,
206        } = cfg.catalog_type
207        {
208            if let Some(v) = config.get("catalog.workspace_url") {
209                *workspace_url = v.to_string();
210            }
211            if let Some(v) = config.get("catalog.access_token") {
212                *access_token = v.to_string();
213            }
214        }
215        if let Some(v) = config.get("catalog.storage.location") {
216            cfg.catalog_storage_location = Some(v.to_string());
217        }
218        if let Some(v) = config.get("write.timeout.ms") {
219            let ms: u64 = v.parse().map_err(|_| {
220                ConnectorError::ConfigurationError(format!("invalid write.timeout.ms: '{v}'"))
221            })?;
222            cfg.write_timeout = Duration::from_millis(ms);
223        }
224
225        // ── Parquet writer configuration ──
226        if let Some(v) = config.get("parquet.compression") {
227            cfg.parquet.compression = v.to_string();
228        }
229        if let Some(v) = config.get("parquet.compression.level") {
230            cfg.parquet.compression_level = v.parse().map_err(|_| {
231                ConnectorError::ConfigurationError(format!(
232                    "invalid parquet.compression.level: '{v}'"
233                ))
234            })?;
235        }
236        if let Some(v) = config.get("parquet.dictionary.enabled") {
237            cfg.parquet.dictionary_enabled = v.eq_ignore_ascii_case("true");
238        }
239        if let Some(v) = config.get("parquet.statistics") {
240            cfg.parquet.statistics = v.to_string();
241        }
242        if let Some(v) = config.get("parquet.bloom.filter.columns") {
243            cfg.parquet.bloom_filter_columns = v
244                .split(',')
245                .map(|c| c.trim().to_string())
246                .filter(|c| !c.is_empty())
247                .collect();
248        }
249        if let Some(v) = config.get("parquet.bloom.filter.fpp") {
250            cfg.parquet.bloom_filter_fpp = v.parse().map_err(|_| {
251                ConnectorError::ConfigurationError(format!(
252                    "invalid parquet.bloom.filter.fpp: '{v}'"
253                ))
254            })?;
255        }
256        if let Some(v) = config.get("parquet.bloom.filter.ndv") {
257            cfg.parquet.bloom_filter_ndv = v.parse().map_err(|_| {
258                ConnectorError::ConfigurationError(format!(
259                    "invalid parquet.bloom.filter.ndv: '{v}'"
260                ))
261            })?;
262        }
263        if let Some(v) = config.get("parquet.max.row.group.size") {
264            cfg.parquet.max_row_group_size = v.parse().map_err(|_| {
265                ConnectorError::ConfigurationError(format!(
266                    "invalid parquet.max.row.group.size: '{v}'"
267                ))
268            })?;
269        }
270
271        // Resolve storage credentials: explicit options + environment variable fallbacks.
272        let explicit_storage = config.properties_with_prefix("storage.");
273        let resolved = StorageCredentialResolver::resolve(&cfg.table_path, &explicit_storage);
274        cfg.storage_options = resolved.options;
275
276        // Map LogStore configuration keys to delta-rs storage options.
277        if let Some(v) = config.get("storage.s3_locking_provider") {
278            cfg.storage_options
279                .insert("AWS_S3_LOCKING_PROVIDER".to_string(), v.to_string());
280        }
281        if let Some(v) = config.get("storage.dynamodb_table_name") {
282            cfg.storage_options
283                .insert("DELTA_DYNAMO_TABLE_NAME".to_string(), v.to_string());
284        }
285
286        cfg.validate()?;
287        Ok(cfg)
288    }
289
290    /// Formats the storage options for safe logging with secrets redacted.
291    #[must_use]
292    pub fn display_storage_options(&self) -> String {
293        SecretMasker::display_map(&self.storage_options)
294    }
295
296    /// Validates the configuration for consistency.
297    ///
298    /// # Errors
299    ///
300    /// Returns `ConnectorError::ConfigurationError` on invalid combinations.
301    pub fn validate(&self) -> Result<(), ConnectorError> {
302        if self.table_path.is_empty() {
303            return Err(ConnectorError::missing_config("table.path"));
304        }
305        if self.write_mode == DeltaWriteMode::Upsert && self.merge_key_columns.is_empty() {
306            return Err(ConnectorError::ConfigurationError(
307                "upsert mode requires 'merge.key.columns' to be set".into(),
308            ));
309        }
310        if self.max_buffer_records == 0 {
311            return Err(ConnectorError::ConfigurationError(
312                "max.buffer.records must be > 0".into(),
313            ));
314        }
315        if self.target_file_size == 0 {
316            return Err(ConnectorError::ConfigurationError(
317                "target.file.size must be > 0".into(),
318            ));
319        }
320        if self.write_timeout < Duration::from_secs(5) {
321            return Err(ConnectorError::ConfigurationError(
322                "write.timeout.ms must be >= 5000 (5 seconds)".into(),
323            ));
324        }
325
326        match self.parquet.compression.to_lowercase().as_str() {
327            "zstd" | "snappy" | "lz4" | "gzip" | "none" | "uncompressed" => {}
328            other => {
329                return Err(ConnectorError::ConfigurationError(format!(
330                    "unknown parquet.compression: '{other}' \
331                     (expected 'zstd', 'snappy', 'lz4', 'gzip', or 'none')"
332                )));
333            }
334        }
335        match self.parquet.statistics.to_lowercase().as_str() {
336            "none" | "chunk" | "page" => {}
337            other => {
338                return Err(ConnectorError::ConfigurationError(format!(
339                    "unknown parquet.statistics: '{other}' (expected 'none', 'chunk', or 'page')"
340                )));
341            }
342        }
343        if self.parquet.bloom_filter_fpp <= 0.0 || self.parquet.bloom_filter_fpp >= 1.0 {
344            return Err(ConnectorError::ConfigurationError(
345                "parquet.bloom.filter.fpp must be in (0.0, 1.0)".into(),
346            ));
347        }
348        if self.parquet.max_row_group_size == 0 {
349            return Err(ConnectorError::ConfigurationError(
350                "parquet.max.row.group.size must be > 0".into(),
351            ));
352        }
353        // Eagerly validate that WriterProperties can be built so invalid
354        // codec/level combos are caught at config time, not first write.
355        #[cfg(feature = "delta-lake")]
356        {
357            self.parquet.to_writer_properties()?;
358        }
359        self.validate_catalog()?;
360
361        // Validate cloud storage credentials (skip when catalog resolves the path).
362        if self.catalog_type == DeltaCatalogType::None {
363            let resolved =
364                StorageCredentialResolver::resolve(&self.table_path, &self.storage_options);
365            let cloud_result = CloudConfigValidator::validate(&resolved);
366            if !cloud_result.is_valid() {
367                return Err(ConnectorError::ConfigurationError(
368                    cloud_result.error_message(),
369                ));
370            }
371        }
372
373        Ok(())
374    }
375
376    /// Validates catalog-specific requirements.
377    fn validate_catalog(&self) -> Result<(), ConnectorError> {
378        match &self.catalog_type {
379            DeltaCatalogType::None => {}
380            DeltaCatalogType::Glue => {
381                #[cfg(not(feature = "delta-lake-glue"))]
382                return Err(ConnectorError::ConfigurationError(
383                    "Glue catalog requires the 'delta-lake-glue' feature. \
384                     Build with: cargo build --features delta-lake-glue"
385                        .into(),
386                ));
387                #[cfg(feature = "delta-lake-glue")]
388                if self.catalog_database.is_none() {
389                    return Err(ConnectorError::ConfigurationError(
390                        "Glue catalog requires 'catalog.database' to be set".into(),
391                    ));
392                }
393            }
394            DeltaCatalogType::Unity {
395                workspace_url,
396                access_token,
397            } => {
398                #[cfg(not(feature = "delta-lake-unity"))]
399                {
400                    let _ = (workspace_url, access_token);
401                    return Err(ConnectorError::ConfigurationError(
402                        "Unity catalog requires the 'delta-lake-unity' feature. \
403                         Build with: cargo build --features delta-lake-unity"
404                            .into(),
405                    ));
406                }
407                #[cfg(feature = "delta-lake-unity")]
408                {
409                    if workspace_url.is_empty() {
410                        return Err(ConnectorError::ConfigurationError(
411                            "Unity catalog requires 'catalog.workspace_url' to be set".into(),
412                        ));
413                    }
414                    if access_token.is_empty() {
415                        return Err(ConnectorError::ConfigurationError(
416                            "Unity catalog requires 'catalog.access_token' to be set".into(),
417                        ));
418                    }
419                    if self.catalog_storage_location.is_some() {
420                        if self.catalog_name.is_none() {
421                            return Err(ConnectorError::ConfigurationError(
422                                "Unity catalog auto-create requires 'catalog.name' to be set"
423                                    .into(),
424                            ));
425                        }
426                        if self.catalog_schema.is_none() {
427                            return Err(ConnectorError::ConfigurationError(
428                                "Unity catalog auto-create requires 'catalog.schema' to be set"
429                                    .into(),
430                            ));
431                        }
432                    }
433                }
434            }
435        }
436        Ok(())
437    }
438}
439
440/// Delta Lake write mode.
441#[derive(Debug, Clone, Copy, PartialEq, Eq)]
442pub enum DeltaWriteMode {
443    /// Append-only: all records are inserts. Most efficient for immutable streams.
444    Append,
445    /// Overwrite: replace partition contents. Used for batch-style recomputation.
446    Overwrite,
447    /// Upsert/Merge: CDC-style insert/update/delete via MERGE statement.
448    /// Requires `merge_key_columns` to be set. Integrates with Z-sets.
449    Upsert,
450}
451
452impl FromStr for DeltaWriteMode {
453    type Err = String;
454
455    fn from_str(s: &str) -> Result<Self, Self::Err> {
456        match s.to_lowercase().as_str() {
457            "append" => Ok(Self::Append),
458            "overwrite" => Ok(Self::Overwrite),
459            "upsert" | "merge" => Ok(Self::Upsert),
460            other => Err(format!("unknown write mode: '{other}'")),
461        }
462    }
463}
464
465impl fmt::Display for DeltaWriteMode {
466    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
467        match self {
468            Self::Append => write!(f, "append"),
469            Self::Overwrite => write!(f, "overwrite"),
470            Self::Upsert => write!(f, "upsert"),
471        }
472    }
473}
474
475/// Delta Lake catalog type for table discovery.
476///
477/// Catalogs enable referencing tables by logical names instead of raw paths.
478#[derive(Debug, Clone, PartialEq, Eq, Default)]
479pub enum DeltaCatalogType {
480    /// No catalog — table path is a direct file or cloud URI.
481    #[default]
482    None,
483    /// AWS Glue Data Catalog.
484    Glue,
485    /// Databricks Unity Catalog.
486    Unity {
487        /// Databricks workspace URL (e.g., `https://xxx.cloud.databricks.com`).
488        workspace_url: String,
489        /// Databricks access token.
490        access_token: String,
491    },
492}
493
494impl FromStr for DeltaCatalogType {
495    type Err = String;
496
497    fn from_str(s: &str) -> Result<Self, Self::Err> {
498        match s.to_lowercase().as_str() {
499            "none" | "" => Ok(Self::None),
500            "glue" => Ok(Self::Glue),
501            "unity" => Ok(Self::Unity {
502                workspace_url: String::new(),
503                access_token: String::new(),
504            }),
505            other => Err(format!("unknown catalog type: '{other}'")),
506        }
507    }
508}
509
510impl fmt::Display for DeltaCatalogType {
511    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
512        match self {
513            Self::None => write!(f, "none"),
514            Self::Glue => write!(f, "glue"),
515            Self::Unity { .. } => write!(f, "unity"),
516        }
517    }
518}
519
520/// Configuration for Parquet writer properties (compression, dictionary
521/// encoding, statistics, bloom filters, row group sizing).
522#[derive(Debug, Clone)]
523pub struct ParquetWriteConfig {
524    /// Compression codec: `"zstd"`, `"snappy"`, `"lz4"`, `"gzip"`, or `"none"`.
525    pub compression: String,
526    /// Compression level (default: 1 — ZSTD L1 for hot writes).
527    pub compression_level: i32,
528    /// Whether to enable dictionary encoding (default: true).
529    pub dictionary_enabled: bool,
530    /// Statistics granularity: `"none"`, `"chunk"`, or `"page"` (default: `"page"`).
531    pub statistics: String,
532    /// Columns to build bloom filters for (default: empty).
533    pub bloom_filter_columns: Vec<String>,
534    /// Bloom filter false-positive probability (default: 0.01).
535    pub bloom_filter_fpp: f64,
536    /// Bloom filter expected number of distinct values (0 = parquet default).
537    pub bloom_filter_ndv: u64,
538    /// Maximum rows per row group (default: 1,000,000).
539    pub max_row_group_size: usize,
540}
541
542impl Default for ParquetWriteConfig {
543    fn default() -> Self {
544        Self {
545            compression: "zstd".to_string(),
546            compression_level: 1,
547            dictionary_enabled: true,
548            statistics: "page".to_string(),
549            bloom_filter_columns: Vec::new(),
550            bloom_filter_fpp: 0.01,
551            bloom_filter_ndv: 0,
552            max_row_group_size: 1_000_000,
553        }
554    }
555}
556
557#[cfg(feature = "delta-lake")]
558impl ParquetWriteConfig {
559    /// Builds `WriterProperties` for hot-path writes (uses `compression_level`).
560    ///
561    /// # Errors
562    ///
563    /// Returns `ConnectorError::ConfigurationError` on invalid codec/level.
564    pub fn to_writer_properties(
565        &self,
566    ) -> Result<deltalake::parquet::file::properties::WriterProperties, ConnectorError> {
567        self.build_properties(self.compression_level)
568    }
569
570    /// Shared builder: maps string codec → `Compression`, sets dictionary,
571    /// statistics, bloom filters, and row group size.
572    fn build_properties(
573        &self,
574        level: i32,
575    ) -> Result<deltalake::parquet::file::properties::WriterProperties, ConnectorError> {
576        use deltalake::parquet::basic::{Compression, GzipLevel, ZstdLevel};
577        use deltalake::parquet::file::properties::{EnabledStatistics, WriterProperties};
578        use deltalake::parquet::schema::types::ColumnPath;
579
580        let compression = match self.compression.to_lowercase().as_str() {
581            "zstd" => {
582                let zstd_level = ZstdLevel::try_new(level).map_err(|e| {
583                    ConnectorError::ConfigurationError(format!("invalid ZSTD level {level}: {e}"))
584                })?;
585                Compression::ZSTD(zstd_level)
586            }
587            "snappy" => Compression::SNAPPY,
588            "lz4" => Compression::LZ4_RAW,
589            "gzip" => {
590                let level_u32: u32 = level.try_into().map_err(|_| {
591                    ConnectorError::ConfigurationError(format!(
592                        "invalid GZIP level {level}: must be non-negative"
593                    ))
594                })?;
595                let gzip_level = GzipLevel::try_new(level_u32).map_err(|e| {
596                    ConnectorError::ConfigurationError(format!("invalid GZIP level {level}: {e}"))
597                })?;
598                Compression::GZIP(gzip_level)
599            }
600            "none" | "uncompressed" => Compression::UNCOMPRESSED,
601            other => {
602                return Err(ConnectorError::ConfigurationError(format!(
603                    "unknown parquet.compression: '{other}' \
604                     (expected 'zstd', 'snappy', 'lz4', 'gzip', or 'none')"
605                )));
606            }
607        };
608
609        let statistics = match self.statistics.to_lowercase().as_str() {
610            "none" => EnabledStatistics::None,
611            "chunk" => EnabledStatistics::Chunk,
612            "page" => EnabledStatistics::Page,
613            other => {
614                return Err(ConnectorError::ConfigurationError(format!(
615                    "unknown parquet.statistics: '{other}' (expected 'none', 'chunk', or 'page')"
616                )));
617            }
618        };
619
620        let mut builder = WriterProperties::builder()
621            .set_compression(compression)
622            .set_dictionary_enabled(self.dictionary_enabled)
623            .set_statistics_enabled(statistics)
624            .set_max_row_group_row_count(Some(self.max_row_group_size));
625
626        for col_name in &self.bloom_filter_columns {
627            let col_path = ColumnPath::from(col_name.as_str());
628            builder = builder
629                .set_column_bloom_filter_enabled(col_path.clone(), true)
630                .set_column_bloom_filter_fpp(col_path.clone(), self.bloom_filter_fpp);
631            if self.bloom_filter_ndv > 0 {
632                builder = builder.set_column_bloom_filter_ndv(col_path, self.bloom_filter_ndv);
633            }
634        }
635
636        Ok(builder.build())
637    }
638}
639
640#[cfg(test)]
641#[allow(clippy::field_reassign_with_default)]
642mod tests;
laminar_connectors/lakehouse/delta_config/mod.rs

laminar_connectors/lakehouse/delta_config/
mod.rs
