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

laminar_connectors/lakehouse/delta_config/
mod.rs
