laminar_connectors/lakehouse/iceberg_config/
source.rs

1use std::time::Duration;
2
3use crate::config::ConnectorConfig;
4use crate::error::ConnectorError;
5
6use super::{
7    optional_alias, optional_non_empty, parse_comma_list, parse_duration_value, parse_nonzero,
8    parse_nonzero_value, parse_optional_alias, parse_or_default, validate_io_timeout,
9    validate_property_map_bounds, IcebergCatalogConfig, IcebergReadBootstrap, IcebergReadMode,
10    IcebergStorageConfig, MIB,
11};
12
13const MAX_SCAN_CHANNEL_CAPACITY: usize = 64;
14const MAX_SCAN_CONCURRENCY: usize = 64;
15
16/// Configuration for bounded snapshot and append reads.
17#[derive(Debug, Clone)]
18pub struct IcebergSourceConfig {
19    /// Catalog settings.
20    pub catalog: IcebergCatalogConfig,
21    /// Table storage settings.
22    pub storage: IcebergStorageConfig,
23    /// Requested read semantics.
24    pub read_mode: IcebergReadMode,
25    /// Append bootstrap policy.
26    pub bootstrap: IcebergReadBootstrap,
27    /// How often append mode refreshes table metadata.
28    pub poll_interval: Duration,
29    /// Selected start or bounded snapshot.
30    pub snapshot_id: Option<i64>,
31    /// Named Iceberg table ref.
32    pub table_ref: String,
33    /// Projected columns; empty selects all columns.
34    pub select_columns: Vec<String>,
35    /// Serialized Iceberg predicate pushed into scan planning.
36    pub filter: Option<String>,
37    /// Maximum lineage snapshots processed by one poll.
38    pub max_snapshots_per_poll: usize,
39    /// Maximum files exposed by scan planning.
40    pub max_planned_files: usize,
41    /// Maximum encoded manifest-list size.
42    pub max_manifest_list_bytes: usize,
43    /// Maximum encoded size of one manifest.
44    pub max_manifest_bytes: usize,
45    /// Maximum manifests referenced by one snapshot.
46    pub max_manifests_per_snapshot: usize,
47    /// Capacity of the scan-to-ingestion batch channel.
48    pub scan_channel_capacity: usize,
49    /// Maximum concurrent manifest and data-file reads.
50    pub scan_concurrency: usize,
51}
52
53impl IcebergSourceConfig {
54    /// Parses source settings from resolved connector configuration.
55    ///
56    /// `snapshot` mode without an explicit ID selects the current snapshot at
57    /// open and remains pinned to it. `append` mode uses that same selection as
58    /// its lineage root.
59    ///
60    /// # Errors
61    ///
62    /// Returns an error for missing, malformed, or contradictory settings.
63    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
64        let read_mode = parse_or_default(config, "read.mode")?;
65        if config.get("read.bootstrap").is_some() && read_mode != IcebergReadMode::Append {
66            return Err(ConnectorError::ConfigurationError(
67                "read.bootstrap is only valid when read.mode=append".into(),
68            ));
69        }
70        let poll_interval = match optional_alias(config, "poll.interval", "poll.interval.ms")? {
71            Some(value) if config.get("poll.interval").is_some() => {
72                parse_duration_value("poll.interval", value)?
73            }
74            Some(value) => Duration::from_millis(
75                u64::try_from(parse_nonzero_value("poll.interval.ms", value)?).map_err(|_| {
76                    ConnectorError::ConfigurationError("poll.interval.ms is too large".into())
77                })?,
78            ),
79            None => Duration::from_secs(60),
80        };
81
82        let parsed = Self {
83            catalog: IcebergCatalogConfig::from_config(config)?,
84            storage: IcebergStorageConfig::from_config(config)?,
85            read_mode,
86            bootstrap: parse_or_default(config, "read.bootstrap")?,
87            poll_interval,
88            snapshot_id: parse_optional_alias(config, "start.snapshot.id", "snapshot.id")?,
89            table_ref: config.get("table.ref").unwrap_or("main").trim().to_string(),
90            select_columns: parse_comma_list(
91                optional_alias(config, "projection", "select.columns")?,
92                "projection",
93                1_024,
94            )?,
95            filter: optional_non_empty(config, "filter"),
96            max_snapshots_per_poll: parse_nonzero(config, "read.max.snapshots.per.poll", 1_024)?,
97            max_planned_files: parse_nonzero(config, "read.max.planned.files", 65_536)?,
98            max_manifest_list_bytes: parse_nonzero(
99                config,
100                "read.max.manifest.list.bytes",
101                64 * MIB,
102            )?,
103            max_manifest_bytes: parse_nonzero(config, "read.max.manifest.bytes", 64 * MIB)?,
104            max_manifests_per_snapshot: parse_nonzero(
105                config,
106                "read.max.manifests.per.snapshot",
107                65_536,
108            )?,
109            scan_channel_capacity: parse_nonzero(config, "read.channel.capacity", 2)?,
110            scan_concurrency: parse_nonzero(config, "read.scan.concurrency", 4)?,
111        };
112        parsed.validate_read_limits()?;
113        Ok(parsed)
114    }
115
116    /// Validates programmatically constructed source bounds before external I/O.
117    ///
118    /// # Errors
119    ///
120    /// Returns a configuration error when a resource or timeout bound is zero.
121    pub fn validate_read_limits(&self) -> Result<(), ConnectorError> {
122        validate_property_map_bounds("catalog.property.*", &self.catalog.properties)?;
123        validate_property_map_bounds("storage.property.*", &self.storage.properties)?;
124        for (name, value) in [
125            ("read.max.snapshots.per.poll", self.max_snapshots_per_poll),
126            ("read.max.planned.files", self.max_planned_files),
127            ("read.max.manifest.list.bytes", self.max_manifest_list_bytes),
128            ("read.max.manifest.bytes", self.max_manifest_bytes),
129            (
130                "read.max.manifests.per.snapshot",
131                self.max_manifests_per_snapshot,
132            ),
133            ("read.channel.capacity", self.scan_channel_capacity),
134            ("read.scan.concurrency", self.scan_concurrency),
135        ] {
136            if value == 0 {
137                return Err(ConnectorError::ConfigurationError(format!(
138                    "{name} must be greater than zero"
139                )));
140            }
141        }
142        if self.poll_interval.is_zero() {
143            return Err(ConnectorError::ConfigurationError(
144                "poll.interval must be greater than zero".into(),
145            ));
146        }
147        for (name, value) in [
148            ("catalog.connect_timeout", self.catalog.connect_timeout),
149            ("catalog.request_timeout", self.catalog.request_timeout),
150            ("storage.request_timeout", self.storage.request_timeout),
151            ("storage.connect_timeout", self.storage.connect_timeout),
152        ] {
153            validate_io_timeout(name, value)?;
154        }
155        if self.scan_channel_capacity > MAX_SCAN_CHANNEL_CAPACITY {
156            return Err(ConnectorError::ConfigurationError(format!(
157                "[LDB-ICEBERG-SCAN-CHANNEL-LIMIT] read.channel.capacity must not exceed {MAX_SCAN_CHANNEL_CAPACITY}"
158            )));
159        }
160        if self.scan_concurrency > MAX_SCAN_CONCURRENCY {
161            return Err(ConnectorError::ConfigurationError(format!(
162                "[LDB-ICEBERG-SCAN-CONCURRENCY-LIMIT] read.scan.concurrency must not exceed {MAX_SCAN_CONCURRENCY}"
163            )));
164        }
165        if self.table_ref.is_empty() {
166            return Err(ConnectorError::ConfigurationError(
167                "table.ref must not be empty".into(),
168            ));
169        }
170        Ok(())
171    }
172}
laminar_connectors/lakehouse/iceberg_config/source.rs

laminar_connectors/lakehouse/iceberg_config/
source.rs
