laminar_connectors/lakehouse/iceberg_config/
mod.rs

1//! Typed Apache Iceberg connector configuration.
2#![allow(clippy::disallowed_types)] // cold path: connector configuration
3
4#[cfg(feature = "iceberg-core")]
5mod identity;
6mod modes;
7mod registry;
8mod source;
9#[cfg(feature = "iceberg-core")]
10mod storage_diagnostics;
11mod table_definition;
12mod warehouse_location;
13
14use std::collections::HashMap;
15use std::fmt;
16use std::time::Duration;
17
18use crate::config::ConnectorConfig;
19use crate::connector::DeliveryGuarantee;
20use crate::error::ConnectorError;
21
22#[cfg(feature = "iceberg-core")]
23pub(crate) use identity::stable_catalog_identity;
24pub use modes::{
25    IcebergCatalogAuthType, IcebergCatalogType, IcebergReadBootstrap, IcebergReadMode,
26    IcebergSchemaEvolutionMode, IcebergStorageEncryption, IcebergStorageType,
27    IcebergWriteDistributionMode, IcebergWriteMode,
28};
29pub(crate) use registry::{sink_config_keys, source_config_keys};
30pub use source::IcebergSourceConfig;
31pub(crate) use table_definition::{
32    parse_parquet_compression, parse_table_fields, validate_distinct_names,
33    validate_persisted_properties, validate_table_definition,
34};
35pub use table_definition::{
36    IcebergNullOrder, IcebergPartitionField, IcebergSortDirection, IcebergSortField,
37    IcebergTransform,
38};
39#[cfg(feature = "iceberg-core")]
40pub(crate) use table_definition::{
41    PARQUET_COMPRESSION_PROPERTY, PARQUET_ROW_GROUP_SIZE_PROPERTY, TARGET_FILE_SIZE_PROPERTY,
42};
43
44const MIB: usize = 1024 * 1024;
45const MAX_ICEBERG_IO_TIMEOUT: Duration = Duration::from_secs(24 * 60 * 60);
46pub(crate) const ICEBERG_MAX_FILES_PER_CHECKPOINT: usize = 4_096;
47const MAX_CONFIG_COLLECTION_BYTES: usize = MIB;
48const MAX_CONFIG_PROPERTY_ENTRIES: usize = 256;
49
50pub(crate) fn validate_io_timeout(name: &str, value: Duration) -> Result<(), ConnectorError> {
51    if value.is_zero() {
52        return Err(ConnectorError::ConfigurationError(format!(
53            "{name} must be greater than zero"
54        )));
55    }
56    if value > MAX_ICEBERG_IO_TIMEOUT {
57        return Err(ConnectorError::ConfigurationError(format!(
58            "[LDB-ICEBERG-IO-TIMEOUT-LIMIT] {name} must not exceed {}s",
59            MAX_ICEBERG_IO_TIMEOUT.as_secs()
60        )));
61    }
62    Ok(())
63}
64
65/// Catalog connection settings shared by Iceberg sources and sinks.
66#[derive(Clone)]
67pub struct IcebergCatalogConfig {
68    /// Catalog implementation.
69    pub catalog_type: IcebergCatalogType,
70    /// Catalog endpoint.
71    pub catalog_uri: String,
72    /// Warehouse name or location.
73    pub warehouse: String,
74    /// Optional catalog namespace prefix.
75    pub prefix: Option<String>,
76    /// Authentication mechanism.
77    pub auth_type: IcebergCatalogAuthType,
78    /// `OAuth2` token endpoint.
79    pub oauth2_server_uri: Option<String>,
80    /// `OAuth2` client identifier.
81    pub oauth2_client_id: Option<String>,
82    /// `OAuth2` requested scope.
83    pub oauth2_scope: Option<String>,
84    /// Whether REST credential vending may be used.
85    pub access_delegation: bool,
86    /// Bound for catalog connection establishment.
87    pub connect_timeout: Duration,
88    /// Bound for an individual catalog request.
89    pub request_timeout: Duration,
90    /// End-to-end catalog commit bound.
91    pub commit_timeout: Duration,
92    /// Iceberg namespace.
93    pub namespace: String,
94    /// Table name within the namespace.
95    pub table_name: String,
96    /// Additional catalog properties. Values may contain secrets.
97    pub properties: HashMap<String, String>,
98}
99
100impl fmt::Debug for IcebergCatalogConfig {
101    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
102        let mut property_keys: Vec<&str> = self.properties.keys().map(String::as_str).collect();
103        property_keys.sort_unstable();
104        formatter
105            .debug_struct("IcebergCatalogConfig")
106            .field("catalog_type", &self.catalog_type)
107            .field("catalog_uri", &"<configured>")
108            .field("warehouse", &"<configured>")
109            .field("prefix", &self.prefix)
110            .field("auth_type", &self.auth_type)
111            .field(
112                "oauth2_server_uri",
113                &self.oauth2_server_uri.as_ref().map(|_| "<configured>"),
114            )
115            .field(
116                "oauth2_client_id",
117                &self.oauth2_client_id.as_ref().map(|_| "<configured>"),
118            )
119            .field("oauth2_scope", &self.oauth2_scope)
120            .field("access_delegation", &self.access_delegation)
121            .field("connect_timeout", &self.connect_timeout)
122            .field("request_timeout", &self.request_timeout)
123            .field("commit_timeout", &self.commit_timeout)
124            .field("namespace", &self.namespace)
125            .field("table_name", &self.table_name)
126            .field("property_keys", &property_keys)
127            .finish()
128    }
129}
130
131fn parse_catalog_auth_type(
132    config: &ConnectorConfig,
133    properties: &HashMap<String, String>,
134) -> Result<IcebergCatalogAuthType, ConnectorError> {
135    let auth_type = match config.get("catalog.auth.type") {
136        Some(value) => value.parse().map_err(ConnectorError::ConfigurationError)?,
137        None if properties.contains_key("token") => IcebergCatalogAuthType::Bearer,
138        None if properties.contains_key("credential") => IcebergCatalogAuthType::OAuth2,
139        None => IcebergCatalogAuthType::None,
140    };
141    match auth_type {
142        IcebergCatalogAuthType::None
143            if properties.contains_key("token") || properties.contains_key("credential") =>
144        {
145            return Err(ConnectorError::ConfigurationError(
146                "catalog.auth.type=none cannot configure catalog.property.token or catalog.property.credential"
147                    .into(),
148            ));
149        }
150        IcebergCatalogAuthType::Bearer if properties.get("token").is_none_or(String::is_empty) => {
151            return Err(ConnectorError::ConfigurationError(
152                "catalog.auth.type=bearer requires a non-empty catalog.property.token".into(),
153            ));
154        }
155        IcebergCatalogAuthType::Bearer if properties.contains_key("credential") => {
156            return Err(ConnectorError::ConfigurationError(
157                "catalog.auth.type=bearer cannot also configure catalog.property.credential".into(),
158            ));
159        }
160        IcebergCatalogAuthType::OAuth2
161            if properties.get("credential").is_none_or(String::is_empty) =>
162        {
163            return Err(ConnectorError::ConfigurationError(
164                "catalog.auth.type=oauth2 requires a non-empty resolved catalog.property.credential"
165                    .into(),
166            ));
167        }
168        IcebergCatalogAuthType::OAuth2 if properties.contains_key("token") => {
169            return Err(ConnectorError::ConfigurationError(
170                "catalog.auth.type=oauth2 cannot also configure catalog.property.token".into(),
171            ));
172        }
173        IcebergCatalogAuthType::None
174        | IcebergCatalogAuthType::Bearer
175        | IcebergCatalogAuthType::OAuth2 => {}
176    }
177
178    if auth_type != IcebergCatalogAuthType::OAuth2 {
179        let typed_option = [
180            "catalog.oauth2.server_uri",
181            "catalog.oauth2.client_id",
182            "catalog.oauth2.scope",
183        ]
184        .into_iter()
185        .find(|key| {
186            config
187                .get(key)
188                .is_some_and(|value| !value.trim().is_empty())
189        });
190        let legacy_option = ["oauth2-server-uri", "scope", "audience", "resource"]
191            .into_iter()
192            .find(|key| properties.contains_key(*key));
193        if let Some(option) = typed_option {
194            return Err(ConnectorError::ConfigurationError(format!(
195                "{option} requires catalog.auth.type=oauth2"
196            )));
197        }
198        if let Some(option) = legacy_option {
199            return Err(ConnectorError::ConfigurationError(format!(
200                "catalog.property.{option} requires catalog.auth.type=oauth2"
201            )));
202        }
203    }
204
205    Ok(auth_type)
206}
207
208impl IcebergCatalogConfig {
209    /// Parses catalog settings from resolved connector configuration.
210    ///
211    /// # Errors
212    ///
213    /// Returns an error for missing, malformed, or contradictory settings.
214    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
215        let catalog_type = parse_or_default(config, "catalog.type")?;
216        let catalog_uri = config.require("catalog.uri")?.trim().to_string();
217        let warehouse = warehouse_location::canonicalize(required_alias(
218            config,
219            "catalog.warehouse",
220            "warehouse",
221        )?)?;
222        let namespace = config.require("namespace")?.trim().to_string();
223        let table_name = config.require("table.name")?.trim().to_string();
224        for (name, value) in [
225            ("catalog.uri", catalog_uri.as_str()),
226            ("catalog.warehouse", warehouse.as_str()),
227            ("namespace", namespace.as_str()),
228            ("table.name", table_name.as_str()),
229        ] {
230            if value.is_empty() {
231                return Err(ConnectorError::ConfigurationError(format!(
232                    "{name} must not be empty"
233                )));
234            }
235        }
236        for (name, value) in [
237            ("catalog.uri", catalog_uri.as_str()),
238            ("catalog.warehouse", warehouse.as_str()),
239        ] {
240            if crate::security::value_contains_uri_secret(value, false) {
241                return Err(ConnectorError::ConfigurationError(format!(
242                    "{name} must not embed credentials; use resolved catalog or storage secret options"
243                )));
244            }
245        }
246        let properties = config.properties_with_prefix("catalog.property.");
247        validate_property_map_bounds("catalog.property.*", &properties)?;
248        let auth_type = parse_catalog_auth_type(config, &properties)?;
249
250        let oauth2_server_uri = optional_non_empty(config, "catalog.oauth2.server_uri");
251        if oauth2_server_uri
252            .as_deref()
253            .is_some_and(|value| crate::security::value_contains_uri_secret(value, false))
254        {
255            return Err(ConnectorError::ConfigurationError(
256                "catalog.oauth2.server_uri must not embed credentials".into(),
257            ));
258        }
259
260        Ok(Self {
261            catalog_type,
262            catalog_uri,
263            warehouse,
264            prefix: optional_non_empty(config, "catalog.prefix"),
265            auth_type,
266            oauth2_server_uri,
267            oauth2_client_id: optional_non_empty(config, "catalog.oauth2.client_id"),
268            oauth2_scope: optional_non_empty(config, "catalog.oauth2.scope"),
269            access_delegation: parse_bool(config, "catalog.access_delegation", false)?,
270            connect_timeout: parse_duration(
271                config,
272                "catalog.connect_timeout",
273                Duration::from_secs(10),
274            )?,
275            request_timeout: parse_duration(
276                config,
277                "catalog.request_timeout",
278                Duration::from_secs(30),
279            )?,
280            commit_timeout: parse_duration(
281                config,
282                "catalog.commit_timeout",
283                Duration::from_secs(120),
284            )?,
285            namespace,
286            table_name,
287            properties,
288        })
289    }
290}
291
292/// Storage settings kept separate from catalog authentication.
293#[derive(Clone)]
294pub struct IcebergStorageConfig {
295    /// Explicit storage implementation, or one inferred from the warehouse URI.
296    pub storage_type: Option<IcebergStorageType>,
297    /// Storage service endpoint override.
298    pub endpoint: Option<String>,
299    /// Cloud region.
300    pub region: Option<String>,
301    /// Whether S3 path-style addressing is required.
302    pub path_style: bool,
303    /// Bound for storage requests.
304    pub request_timeout: Duration,
305    /// Bound for storage connection establishment.
306    pub connect_timeout: Duration,
307    /// Server-side encryption mode.
308    pub encryption: IcebergStorageEncryption,
309    /// KMS key identifier, resolved without logging its value.
310    pub kms_key: Option<String>,
311    /// Additional storage properties. Values may contain secrets.
312    pub properties: HashMap<String, String>,
313}
314
315impl fmt::Debug for IcebergStorageConfig {
316    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
317        let mut property_keys: Vec<&str> = self.properties.keys().map(String::as_str).collect();
318        property_keys.sort_unstable();
319        formatter
320            .debug_struct("IcebergStorageConfig")
321            .field("storage_type", &self.storage_type)
322            .field("endpoint", &self.endpoint.as_ref().map(|_| "<configured>"))
323            .field("region", &self.region)
324            .field("path_style", &self.path_style)
325            .field("request_timeout", &self.request_timeout)
326            .field("connect_timeout", &self.connect_timeout)
327            .field("encryption", &self.encryption)
328            .field("kms_key", &self.kms_key.as_ref().map(|_| "<configured>"))
329            .field("property_keys", &property_keys)
330            .finish()
331    }
332}
333
334impl IcebergStorageConfig {
335    /// Parses storage settings from resolved connector configuration.
336    ///
337    /// # Errors
338    ///
339    /// Returns an error for malformed or contradictory settings.
340    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
341        let storage_type = config
342            .get("storage.type")
343            .map(str::parse)
344            .transpose()
345            .map_err(ConnectorError::ConfigurationError)?;
346        let encryption = parse_or_default(config, "storage.encryption")?;
347        let kms_key = optional_non_empty(config, "storage.kms_key");
348        if encryption == IcebergStorageEncryption::Kms && kms_key.is_none() {
349            return Err(ConnectorError::ConfigurationError(
350                "storage.kms_key is required when storage.encryption=kms".into(),
351            ));
352        }
353        if encryption != IcebergStorageEncryption::Kms && kms_key.is_some() {
354            return Err(ConnectorError::ConfigurationError(
355                "storage.kms_key requires storage.encryption=kms".into(),
356            ));
357        }
358        let path_style = parse_optional_bool(
359            "storage.path_style",
360            config
361                .get("storage.path_style")
362                .or_else(|| config.get("storage.property.s3.path-style-access"))
363                .or_else(|| config.get("catalog.property.s3.path-style-access")),
364            false,
365        )?;
366
367        let endpoint = optional_non_empty(config, "storage.endpoint");
368        if endpoint
369            .as_deref()
370            .is_some_and(|value| crate::security::value_contains_uri_secret(value, false))
371        {
372            return Err(ConnectorError::ConfigurationError(
373                "storage.endpoint must not embed credentials".into(),
374            ));
375        }
376        let properties = config.properties_with_prefix("storage.property.");
377        validate_property_map_bounds("storage.property.*", &properties)?;
378
379        Ok(Self {
380            storage_type,
381            endpoint,
382            region: optional_non_empty(config, "storage.region"),
383            path_style,
384            request_timeout: parse_duration(
385                config,
386                "storage.request_timeout",
387                Duration::from_secs(30),
388            )?,
389            connect_timeout: parse_duration(
390                config,
391                "storage.connect_timeout",
392                Duration::from_secs(10),
393            )?,
394            encryption,
395            kms_key,
396            properties,
397        })
398    }
399}
400
401/// Configuration for the Iceberg append sink.
402#[derive(Clone)]
403pub struct IcebergSinkConfig {
404    /// Catalog settings.
405    pub catalog: IcebergCatalogConfig,
406    /// Table storage settings.
407    pub storage: IcebergStorageConfig,
408    /// Requested mutation mode.
409    pub write_mode: IcebergWriteMode,
410    /// Parquet compression codec.
411    pub compression: String,
412    /// Target size used by Iceberg rolling writers.
413    pub target_file_size_bytes: usize,
414    /// Parquet row-group size.
415    pub parquet_row_group_size_bytes: usize,
416    /// Maximum rows accepted in one in-flight batch.
417    pub max_buffer_rows: usize,
418    /// Maximum Arrow bytes accepted in one in-flight batch.
419    pub max_buffer_bytes: usize,
420    /// Maximum simultaneously open partition writers.
421    pub max_open_partitions: usize,
422    /// Maximum data files produced by one checkpoint participant.
423    pub max_files_per_checkpoint: usize,
424    /// Maximum encoded participant descriptor size.
425    pub max_descriptor_bytes: usize,
426    /// Maximum age of an open data file.
427    pub max_flush_age: Duration,
428    /// Partition-writer distribution policy.
429    pub distribution_mode: IcebergWriteDistributionMode,
430    /// Iceberg identifier field names.
431    pub identifier_fields: Vec<String>,
432    /// Permitted table schema evolution.
433    pub schema_evolution_mode: IcebergSchemaEvolutionMode,
434    /// Pipeline delivery guarantee.
435    pub delivery_guarantee: DeliveryGuarantee,
436    /// Table ref to publish.
437    pub table_ref: String,
438    /// Auto-create the table if absent.
439    pub auto_create: bool,
440    /// Iceberg format version for auto-created tables.
441    pub format_version: u8,
442    /// Declarative partition spec for auto-create.
443    pub partition_spec: Vec<IcebergPartitionField>,
444    /// Declarative sort order for auto-create.
445    pub sort_order: Vec<IcebergSortField>,
446    /// Initial table properties for auto-create.
447    pub initial_table_properties: HashMap<String, String>,
448}
449
450impl fmt::Debug for IcebergSinkConfig {
451    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
452        formatter
453            .debug_struct("IcebergSinkConfig")
454            .field("catalog", &self.catalog)
455            .field("storage", &self.storage)
456            .field("write_mode", &self.write_mode)
457            .field("compression", &self.compression)
458            .field("target_file_size_bytes", &self.target_file_size_bytes)
459            .field(
460                "parquet_row_group_size_bytes",
461                &self.parquet_row_group_size_bytes,
462            )
463            .field("max_buffer_rows", &self.max_buffer_rows)
464            .field("max_buffer_bytes", &self.max_buffer_bytes)
465            .field("max_open_partitions", &self.max_open_partitions)
466            .field("max_files_per_checkpoint", &self.max_files_per_checkpoint)
467            .field("max_descriptor_bytes", &self.max_descriptor_bytes)
468            .field("max_flush_age", &self.max_flush_age)
469            .field("distribution_mode", &self.distribution_mode)
470            .field("identifier_fields", &self.identifier_fields)
471            .field("schema_evolution_mode", &self.schema_evolution_mode)
472            .field("delivery_guarantee", &self.delivery_guarantee)
473            .field("table_ref", &self.table_ref)
474            .field("auto_create", &self.auto_create)
475            .field("format_version", &self.format_version)
476            .field("partition_spec", &self.partition_spec)
477            .field("sort_order", &self.sort_order)
478            .field(
479                "initial_table_property_count",
480                &self.initial_table_properties.len(),
481            )
482            .finish()
483    }
484}
485
486impl IcebergSinkConfig {
487    /// Parses sink settings from resolved connector configuration.
488    ///
489    /// # Errors
490    ///
491    /// Returns an error for missing, malformed, or contradictory settings.
492    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
493        let catalog = IcebergCatalogConfig::from_config(config)?;
494        let storage = IcebergStorageConfig::from_config(config)?;
495        let delivery_guarantee = config
496            .get("delivery.guarantee")
497            .unwrap_or("at-least-once")
498            .parse()
499            .map_err(ConnectorError::ConfigurationError)?;
500        let compression = parse_parquet_compression(
501            optional_alias(config, "parquet.compression", "compression")?.unwrap_or("zstd"),
502        )?;
503        let target_file_size_bytes = parse_nonzero_alias(
504            config,
505            "target.file.size.bytes",
506            "target.file.size",
507            128 * MIB,
508        )?;
509        let parquet_row_group_size_bytes =
510            parse_nonzero(config, "parquet.row.group.size.bytes", 128 * MIB)?;
511        let format_version = config.get_parsed("format.version")?.unwrap_or(2);
512        if !matches!(format_version, 1..=3) {
513            return Err(ConnectorError::ConfigurationError(
514                "format.version must be 1, 2, or 3".into(),
515            ));
516        }
517        let auto_create = parse_bool(config, "auto.create", false)?;
518        let partition_spec = parse_table_fields(config, "partition.spec")?;
519        let sort_order = parse_table_fields(config, "sort.order")?;
520        validate_table_definition(&partition_spec, &sort_order)?;
521        let identifier_fields =
522            parse_comma_list(config.get("identifier.fields"), "identifier.fields", 128)?;
523        validate_distinct_names(&identifier_fields, "identifier.fields")?;
524        let initial_table_properties = config.properties_with_prefix("table.property.");
525        validate_persisted_properties(&initial_table_properties)?;
526        if !auto_create
527            && (config.get("format.version").is_some()
528                || !partition_spec.is_empty()
529                || !sort_order.is_empty()
530                || !initial_table_properties.is_empty())
531        {
532            return Err(ConnectorError::ConfigurationError(
533                "format.version, partition.spec, sort.order, and table.property.* require auto.create=true"
534                    .into(),
535            ));
536        }
537        if auto_create && format_version == 1 && !identifier_fields.is_empty() {
538            return Err(ConnectorError::ConfigurationError(
539                "identifier.fields require format.version=2 or 3 for an auto-created table".into(),
540            ));
541        }
542
543        let parsed = Self {
544            catalog,
545            storage,
546            write_mode: parse_or_default(config, "write.mode")?,
547            compression,
548            target_file_size_bytes,
549            parquet_row_group_size_bytes,
550            max_buffer_rows: parse_nonzero(config, "max.buffer.rows", 65_536)?,
551            max_buffer_bytes: parse_nonzero(config, "max.buffer.bytes", 64 * MIB)?,
552            max_open_partitions: parse_nonzero(config, "max.open.partitions", 64)?,
553            max_files_per_checkpoint: parse_nonzero(
554                config,
555                "max.files.per.checkpoint",
556                ICEBERG_MAX_FILES_PER_CHECKPOINT,
557            )?,
558            max_descriptor_bytes: parse_nonzero(
559                config,
560                "max.descriptor.bytes",
561                crate::connector::MAX_COORDINATED_COMMIT_PAYLOAD_BYTES,
562            )?,
563            max_flush_age: parse_duration(config, "max.flush.age", Duration::from_secs(300))?,
564            distribution_mode: parse_or_default(config, "write.distribution.mode")?,
565            identifier_fields,
566            schema_evolution_mode: parse_or_default(config, "schema.evolution.mode")?,
567            delivery_guarantee,
568            table_ref: config.get("table.ref").unwrap_or("main").trim().to_string(),
569            auto_create,
570            format_version,
571            partition_spec,
572            sort_order,
573            initial_table_properties,
574        };
575        parsed.validate_writer_limits()?;
576        Ok(parsed)
577    }
578
579    #[cfg(feature = "iceberg-core")]
580    pub(crate) fn validate_table_creation(&self) -> Result<(), ConnectorError> {
581        if !matches!(self.format_version, 1..=3) {
582            return Err(ConnectorError::ConfigurationError(
583                "format.version must be 1, 2, or 3".into(),
584            ));
585        }
586        parse_parquet_compression(&self.compression)?;
587        validate_distinct_names(&self.identifier_fields, "identifier.fields")?;
588        validate_table_definition(&self.partition_spec, &self.sort_order)?;
589        validate_persisted_properties(&self.initial_table_properties)?;
590        if self.format_version == 1 && !self.identifier_fields.is_empty() {
591            return Err(ConnectorError::ConfigurationError(
592                "identifier.fields require format.version=2 or 3 for an auto-created table".into(),
593            ));
594        }
595        Ok(())
596    }
597
598    pub(crate) fn validate_writer_limits(&self) -> Result<(), ConnectorError> {
599        validate_property_map_bounds("catalog.property.*", &self.catalog.properties)?;
600        validate_property_map_bounds("storage.property.*", &self.storage.properties)?;
601        for (name, value) in [
602            ("target.file.size.bytes", self.target_file_size_bytes),
603            (
604                "parquet.row.group.size.bytes",
605                self.parquet_row_group_size_bytes,
606            ),
607            ("max.buffer.rows", self.max_buffer_rows),
608            ("max.buffer.bytes", self.max_buffer_bytes),
609            ("max.open.partitions", self.max_open_partitions),
610            ("max.files.per.checkpoint", self.max_files_per_checkpoint),
611            ("max.descriptor.bytes", self.max_descriptor_bytes),
612        ] {
613            if value == 0 {
614                return Err(ConnectorError::ConfigurationError(format!(
615                    "{name} must be greater than zero"
616                )));
617            }
618        }
619        if self.max_flush_age.is_zero() {
620            return Err(ConnectorError::ConfigurationError(
621                "max.flush.age must be greater than zero".into(),
622            ));
623        }
624        for (name, value) in [
625            ("catalog.connect_timeout", self.catalog.connect_timeout),
626            ("catalog.request_timeout", self.catalog.request_timeout),
627            ("catalog.commit_timeout", self.catalog.commit_timeout),
628            ("storage.request_timeout", self.storage.request_timeout),
629            ("storage.connect_timeout", self.storage.connect_timeout),
630        ] {
631            validate_io_timeout(name, value)?;
632        }
633        if self.max_files_per_checkpoint > ICEBERG_MAX_FILES_PER_CHECKPOINT {
634            return Err(ConnectorError::ConfigurationError(format!(
635                "max.files.per.checkpoint must not exceed {ICEBERG_MAX_FILES_PER_CHECKPOINT}"
636            )));
637        }
638        if self.max_open_partitions > self.max_files_per_checkpoint {
639            return Err(ConnectorError::ConfigurationError(
640                "max.open.partitions must not exceed max.files.per.checkpoint".into(),
641            ));
642        }
643        if self.max_descriptor_bytes > crate::connector::MAX_COORDINATED_COMMIT_PAYLOAD_BYTES {
644            return Err(ConnectorError::ConfigurationError(format!(
645                "max.descriptor.bytes must not exceed {}",
646                crate::connector::MAX_COORDINATED_COMMIT_PAYLOAD_BYTES
647            )));
648        }
649        parse_parquet_compression(&self.compression)?;
650        Ok(())
651    }
652}
653
654fn parse_or_default<T>(config: &ConnectorConfig, key: &str) -> Result<T, ConnectorError>
655where
656    T: Default + std::str::FromStr<Err = String>,
657{
658    config
659        .get(key)
660        .map(str::parse)
661        .transpose()
662        .map(Option::unwrap_or_default)
663        .map_err(ConnectorError::ConfigurationError)
664}
665
666fn parse_bool(config: &ConnectorConfig, key: &str, default: bool) -> Result<bool, ConnectorError> {
667    parse_optional_bool(key, config.get(key), default)
668}
669
670fn parse_optional_bool(
671    key: &str,
672    value: Option<&str>,
673    default: bool,
674) -> Result<bool, ConnectorError> {
675    match value {
676        None => Ok(default),
677        Some(value) if value.eq_ignore_ascii_case("true") => Ok(true),
678        Some(value) if value.eq_ignore_ascii_case("false") => Ok(false),
679        Some(value) => Err(ConnectorError::ConfigurationError(format!(
680            "invalid {key}: '{value}'; expected true or false"
681        ))),
682    }
683}
684
685fn parse_nonzero(
686    config: &ConnectorConfig,
687    key: &str,
688    default: usize,
689) -> Result<usize, ConnectorError> {
690    config
691        .get(key)
692        .map_or(Ok(default), |value| parse_nonzero_value(key, value))
693}
694
695fn parse_nonzero_alias(
696    config: &ConnectorConfig,
697    key: &str,
698    alias: &str,
699    default: usize,
700) -> Result<usize, ConnectorError> {
701    optional_alias(config, key, alias)?.map_or(Ok(default), |value| parse_nonzero_value(key, value))
702}
703
704fn parse_nonzero_value(key: &str, value: &str) -> Result<usize, ConnectorError> {
705    let parsed = value
706        .parse::<usize>()
707        .map_err(|_| ConnectorError::ConfigurationError(format!("invalid {key}: '{value}'")))?;
708    if parsed == 0 {
709        return Err(ConnectorError::ConfigurationError(format!(
710            "{key} must be greater than zero"
711        )));
712    }
713    Ok(parsed)
714}
715
716fn parse_duration(
717    config: &ConnectorConfig,
718    key: &str,
719    default: Duration,
720) -> Result<Duration, ConnectorError> {
721    config
722        .get(key)
723        .map_or(Ok(default), |value| parse_duration_value(key, value))
724}
725
726fn parse_duration_value(key: &str, value: &str) -> Result<Duration, ConnectorError> {
727    let value = value.trim();
728    let (number, multiplier) = if let Some(number) = value.strip_suffix("ms") {
729        (number, 1_u64)
730    } else if let Some(number) = value.strip_suffix('s') {
731        (number, 1_000)
732    } else if let Some(number) = value.strip_suffix('m') {
733        (number, 60_000)
734    } else {
735        (value, 1)
736    };
737    let milliseconds = number.trim().parse::<u64>().map_err(|_| {
738        ConnectorError::ConfigurationError(format!(
739            "invalid {key}: '{value}'; expected an integer with ms, s, or m"
740        ))
741    })?;
742    if milliseconds == 0 {
743        return Err(ConnectorError::ConfigurationError(format!(
744            "{key} must be greater than zero"
745        )));
746    }
747    milliseconds
748        .checked_mul(multiplier)
749        .map(Duration::from_millis)
750        .ok_or_else(|| ConnectorError::ConfigurationError(format!("{key} is too large")))
751}
752
753fn optional_non_empty(config: &ConnectorConfig, key: &str) -> Option<String> {
754    config
755        .get(key)
756        .map(str::trim)
757        .filter(|value| !value.is_empty())
758        .map(str::to_string)
759}
760
761fn optional_alias<'a>(
762    config: &'a ConnectorConfig,
763    key: &str,
764    alias: &str,
765) -> Result<Option<&'a str>, ConnectorError> {
766    match (config.get(key), config.get(alias)) {
767        (Some(_), Some(_)) => Err(ConnectorError::ConfigurationError(format!(
768            "{key} and its legacy alias {alias} cannot both be set"
769        ))),
770        (value @ Some(_), None) | (None, value @ Some(_)) => Ok(value),
771        (None, None) => Ok(None),
772    }
773}
774
775fn required_alias<'a>(
776    config: &'a ConnectorConfig,
777    key: &str,
778    alias: &str,
779) -> Result<&'a str, ConnectorError> {
780    optional_alias(config, key, alias)?.ok_or_else(|| ConnectorError::missing_config(key))
781}
782
783fn parse_optional_alias<T>(
784    config: &ConnectorConfig,
785    key: &str,
786    alias: &str,
787) -> Result<Option<T>, ConnectorError>
788where
789    T: std::str::FromStr,
790    T::Err: fmt::Display,
791{
792    optional_alias(config, key, alias)?
793        .map(|value| {
794            value.parse().map_err(|error| {
795                ConnectorError::ConfigurationError(format!("invalid {key}: {error}"))
796            })
797        })
798        .transpose()
799}
800
801fn parse_comma_list(
802    value: Option<&str>,
803    key: &str,
804    max_entries: usize,
805) -> Result<Vec<String>, ConnectorError> {
806    let value = value.unwrap_or_default();
807    if value.len() > MAX_CONFIG_COLLECTION_BYTES {
808        return Err(ConnectorError::ConfigurationError(format!(
809            "{key} is {} bytes; the limit is {MAX_CONFIG_COLLECTION_BYTES}",
810            value.len()
811        )));
812    }
813    let values = value
814        .split(',')
815        .map(str::trim)
816        .filter(|item| !item.is_empty())
817        .map(str::to_string)
818        .take(max_entries.saturating_add(1))
819        .collect::<Vec<_>>();
820    if values.len() > max_entries {
821        return Err(ConnectorError::ConfigurationError(format!(
822            "{key} contains more than {max_entries} entries"
823        )));
824    }
825    Ok(values)
826}
827
828fn validate_property_map_bounds(
829    name: &str,
830    properties: &HashMap<String, String>,
831) -> Result<(), ConnectorError> {
832    if properties.len() > MAX_CONFIG_PROPERTY_ENTRIES {
833        return Err(ConnectorError::ConfigurationError(format!(
834            "[LDB-ICEBERG-CONFIG-PROPERTY-LIMIT] {name} contains {} entries; the limit is {MAX_CONFIG_PROPERTY_ENTRIES}",
835            properties.len()
836        )));
837    }
838    let bytes = properties.iter().try_fold(0_usize, |total, (key, value)| {
839        total.checked_add(key.len())?.checked_add(value.len())
840    });
841    if bytes.is_none_or(|bytes| bytes > MAX_CONFIG_COLLECTION_BYTES) {
842        return Err(ConnectorError::ConfigurationError(format!(
843            "[LDB-ICEBERG-CONFIG-PROPERTY-LIMIT] {name} exceeds the {MAX_CONFIG_COLLECTION_BYTES}-byte limit"
844        )));
845    }
846    if properties
847        .keys()
848        .any(|key| key.is_empty() || key.trim() != key)
849    {
850        return Err(ConnectorError::ConfigurationError(format!(
851            "{name} contains an empty or whitespace-padded property name"
852        )));
853    }
854    Ok(())
855}
856
857/// Validates an Arrow pipeline schema against an Iceberg-derived Arrow schema.
858///
859/// # Errors
860///
861/// Returns a schema mismatch for missing fields or unsafe type changes.
862pub fn validate_sink_schema(
863    pipeline: &arrow_schema::Schema,
864    table: &arrow_schema::Schema,
865) -> Result<(), ConnectorError> {
866    for field in pipeline.fields() {
867        let table_field = table.field_with_name(field.name()).map_err(|_| {
868            ConnectorError::SchemaMismatch(format!(
869                "pipeline field '{}' ({}) not found in Iceberg table schema",
870                field.name(),
871                field.data_type()
872            ))
873        })?;
874        if field.data_type() != table_field.data_type()
875            && !is_safe_iceberg_widening(field.data_type(), table_field.data_type())
876        {
877            return Err(ConnectorError::SchemaMismatch(format!(
878                "field '{}': pipeline type {} incompatible with table type {}",
879                field.name(),
880                field.data_type(),
881                table_field.data_type()
882            )));
883        }
884    }
885    Ok(())
886}
887
888pub(crate) fn is_safe_iceberg_widening(
889    from: &arrow_schema::DataType,
890    to: &arrow_schema::DataType,
891) -> bool {
892    use arrow_schema::DataType;
893    matches!(
894        (from, to),
895        (
896            DataType::Int8,
897            DataType::Int16 | DataType::Int32 | DataType::Int64
898        ) | (DataType::Int16, DataType::Int32 | DataType::Int64)
899            | (DataType::Int32, DataType::Int64)
900            | (DataType::Float32, DataType::Float64)
901            | (DataType::Utf8, DataType::LargeUtf8)
902    )
903}
904
905#[cfg(test)]
906mod tests;
laminar_connectors/lakehouse/iceberg_config/mod.rs

laminar_connectors/lakehouse/iceberg_config/
mod.rs
