laminar_connectors/lakehouse/
iceberg_config.rs

1//! Apache Iceberg connector configuration.
2//!
3//! [`IcebergSinkConfig`] and [`IcebergSourceConfig`] encapsulate settings for
4//! writing to and reading from Iceberg tables, parsed from SQL `WITH (...)`
5//! clauses via their respective `from_config` methods.
6#![allow(clippy::disallowed_types)] // cold path: lakehouse configuration
7
8use std::collections::HashMap;
9use std::fmt;
10use std::str::FromStr;
11use std::time::Duration;
12
13use crate::config::ConnectorConfig;
14use crate::error::ConnectorError;
15
16/// Iceberg catalog type.
17#[derive(Debug, Clone, PartialEq, Eq, Default)]
18pub enum IcebergCatalogType {
19    /// REST catalog (Polaris, Nessie, Unity, Glue adapter).
20    #[default]
21    Rest,
22}
23
24impl FromStr for IcebergCatalogType {
25    type Err = String;
26
27    fn from_str(s: &str) -> Result<Self, Self::Err> {
28        match s.to_lowercase().as_str() {
29            "rest" => Ok(Self::Rest),
30            other => Err(format!("unsupported iceberg catalog type: '{other}'")),
31        }
32    }
33}
34
35impl fmt::Display for IcebergCatalogType {
36    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
37        match self {
38            Self::Rest => write!(f, "rest"),
39        }
40    }
41}
42
43/// Shared catalog connection settings for both source and sink.
44#[derive(Debug, Clone)]
45pub struct IcebergCatalogConfig {
46    /// Catalog type (currently only REST).
47    pub catalog_type: IcebergCatalogType,
48    /// REST catalog URI (e.g., `http://polaris:8181`).
49    pub catalog_uri: String,
50    /// Warehouse URL (Hadoop-style: `s3://bucket/wh`) or name (REST catalogs).
51    pub warehouse: String,
52    /// Explicit storage backend. Required when `warehouse` is a name.
53    pub storage_type: Option<String>,
54    /// Iceberg namespace (e.g., `prod` or `prod.analytics`).
55    pub namespace: String,
56    /// Table name within the namespace.
57    pub table_name: String,
58    /// Additional catalog properties (credentials, endpoints, etc.).
59    pub properties: HashMap<String, String>,
60}
61
62impl IcebergCatalogConfig {
63    /// Parses shared catalog settings from a [`ConnectorConfig`].
64    ///
65    /// # Errors
66    ///
67    /// Returns `ConnectorError::MissingConfig` if required keys are absent.
68    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
69        let catalog_type = if let Some(v) = config.get("catalog.type") {
70            v.parse()
71                .map_err(|e: String| ConnectorError::ConfigurationError(e))?
72        } else {
73            IcebergCatalogType::default()
74        };
75
76        let catalog_uri = config.require("catalog.uri")?.to_string();
77        let warehouse = config.require("warehouse")?.to_string();
78        let storage_type = config
79            .get("storage.type")
80            .map(str::trim)
81            .filter(|storage| !storage.is_empty())
82            .map(str::to_ascii_lowercase);
83        if let Some(storage) = storage_type.as_deref() {
84            if !matches!(storage, "s3" | "s3a" | "fs") {
85                return Err(ConnectorError::ConfigurationError(format!(
86                    "unsupported Iceberg storage.type '{storage}'; expected s3 | s3a | fs"
87                )));
88            }
89        }
90        if let Some((scheme, _)) = warehouse.split_once("://") {
91            let scheme = scheme.to_ascii_lowercase();
92            if !matches!(scheme.as_str(), "s3" | "s3a" | "file") {
93                return Err(ConnectorError::ConfigurationError(format!(
94                    "unsupported Iceberg warehouse scheme '{scheme}'; expected s3 | s3a | file"
95                )));
96            }
97        }
98        let namespace = config.require("namespace")?.to_string();
99        let table_name = config.require("table.name")?.to_string();
100
101        let properties = config.properties_with_prefix("catalog.property.");
102
103        Ok(Self {
104            catalog_type,
105            catalog_uri,
106            warehouse,
107            storage_type,
108            namespace,
109            table_name,
110            properties,
111        })
112    }
113}
114
115// ── Sink Configuration ──
116
117/// Configuration for the Iceberg sink connector.
118#[derive(Debug, Clone)]
119pub struct IcebergSinkConfig {
120    /// Shared catalog connection settings.
121    pub catalog: IcebergCatalogConfig,
122    /// Parquet compression codec (default: zstd).
123    pub compression: String,
124    /// Auto-create table if it doesn't exist.
125    pub auto_create: bool,
126}
127
128impl IcebergSinkConfig {
129    /// Parses a sink config from a [`ConnectorConfig`] (SQL WITH clause).
130    ///
131    /// # Errors
132    ///
133    /// Returns `ConnectorError` on missing or invalid values.
134    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
135        let catalog = IcebergCatalogConfig::from_config(config)?;
136
137        let compression = config.get("compression").unwrap_or("zstd").to_string();
138
139        let auto_create = config
140            .get("auto.create")
141            .is_some_and(|v| v.eq_ignore_ascii_case("true"));
142
143        Ok(Self {
144            catalog,
145            compression,
146            auto_create,
147        })
148    }
149}
150
151// ── Source Configuration ──
152
153/// Configuration for the Iceberg source connector (lookup/reference table).
154#[derive(Debug, Clone)]
155pub struct IcebergSourceConfig {
156    /// Shared catalog connection settings.
157    pub catalog: IcebergCatalogConfig,
158    /// How often to poll for new snapshots (default: 60s).
159    pub poll_interval: Duration,
160    /// Pin to a specific snapshot ID (no polling if set).
161    pub snapshot_id: Option<i64>,
162    /// Columns to select (empty = all columns).
163    pub select_columns: Vec<String>,
164}
165
166impl IcebergSourceConfig {
167    /// Parses a source config from a [`ConnectorConfig`] (SQL WITH clause).
168    ///
169    /// # Errors
170    ///
171    /// Returns `ConnectorError` on missing or invalid values.
172    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
173        let catalog = IcebergCatalogConfig::from_config(config)?;
174
175        let poll_interval = config
176            .get_parsed::<u64>("poll.interval.ms")?
177            .map_or(Duration::from_secs(60), Duration::from_millis);
178
179        let snapshot_id = config.get_parsed::<i64>("snapshot.id")?;
180
181        let select_columns = config
182            .get("select.columns")
183            .unwrap_or("")
184            .split(',')
185            .map(|s| s.trim().to_string())
186            .filter(|s| !s.is_empty())
187            .collect();
188
189        Ok(Self {
190            catalog,
191            poll_interval,
192            snapshot_id,
193            select_columns,
194        })
195    }
196}
197
198/// Checks if `from` can be safely widened to `to` without data loss.
199fn is_safe_widening(from: &arrow_schema::DataType, to: &arrow_schema::DataType) -> bool {
200    use arrow_schema::DataType;
201    matches!(
202        (from, to),
203        (
204            DataType::Int8,
205            DataType::Int16 | DataType::Int32 | DataType::Int64
206        ) | (DataType::Int16, DataType::Int32 | DataType::Int64)
207            | (DataType::Int32, DataType::Int64)
208            | (DataType::Float32, DataType::Float64)
209            | (DataType::Utf8, DataType::LargeUtf8)
210    )
211}
212
213/// Validates that a pipeline's output schema is compatible with an Iceberg
214/// table's Arrow schema.
215///
216/// Every field in `pipeline` must exist in `table` with a matching or
217/// safely-widenable type. Extra columns in `table` are acceptable (Iceberg
218/// fills them with nulls).
219///
220/// # Errors
221///
222/// Returns `ConnectorError::SchemaMismatch` on incompatible fields.
223pub fn validate_sink_schema(
224    pipeline: &arrow_schema::Schema,
225    table: &arrow_schema::Schema,
226) -> Result<(), ConnectorError> {
227    for field in pipeline.fields() {
228        match table.field_with_name(field.name()) {
229            Ok(table_field) => {
230                if field.data_type() != table_field.data_type()
231                    && !is_safe_widening(field.data_type(), table_field.data_type())
232                {
233                    return Err(ConnectorError::SchemaMismatch(format!(
234                        "field '{}': pipeline type {} incompatible with table type {}",
235                        field.name(),
236                        field.data_type(),
237                        table_field.data_type(),
238                    )));
239                }
240            }
241            Err(_) => {
242                return Err(ConnectorError::SchemaMismatch(format!(
243                    "pipeline field '{}' ({}) not found in Iceberg table schema",
244                    field.name(),
245                    field.data_type(),
246                )));
247            }
248        }
249    }
250    Ok(())
251}
252
253#[cfg(test)]
254mod tests {
255    use super::*;
256
257    #[test]
258    fn test_catalog_type_parse() {
259        assert_eq!(
260            "rest".parse::<IcebergCatalogType>().unwrap(),
261            IcebergCatalogType::Rest
262        );
263        assert!("unknown".parse::<IcebergCatalogType>().is_err());
264    }
265
266    #[test]
267    fn test_sink_config_from_config() {
268        let mut config = ConnectorConfig::new("iceberg");
269        config.set("catalog.uri", "http://localhost:8181");
270        config.set("warehouse", "s3://bucket/wh");
271        config.set("namespace", "prod");
272        config.set("table.name", "events");
273        config.set("compression", "snappy");
274
275        let cfg = IcebergSinkConfig::from_config(&config).unwrap();
276        assert_eq!(cfg.catalog.catalog_uri, "http://localhost:8181");
277        assert_eq!(cfg.catalog.warehouse, "s3://bucket/wh");
278        assert_eq!(cfg.catalog.namespace, "prod");
279        assert_eq!(cfg.catalog.table_name, "events");
280        assert_eq!(cfg.compression, "snappy");
281    }
282
283    #[test]
284    fn test_source_config_from_config() {
285        let mut config = ConnectorConfig::new("iceberg");
286        config.set("catalog.uri", "http://localhost:8181");
287        config.set("warehouse", "s3://bucket/wh");
288        config.set("namespace", "prod");
289        config.set("table.name", "dim_customers");
290        config.set("poll.interval.ms", "30000");
291        config.set("snapshot.id", "42");
292
293        let cfg = IcebergSourceConfig::from_config(&config).unwrap();
294        assert_eq!(cfg.poll_interval, Duration::from_secs(30));
295        assert_eq!(cfg.snapshot_id, Some(42));
296    }
297
298    #[test]
299    fn test_missing_required_field() {
300        let config = ConnectorConfig::new("iceberg");
301        assert!(IcebergSinkConfig::from_config(&config).is_err());
302    }
303
304    #[test]
305    fn test_defaults() {
306        let mut config = ConnectorConfig::new("iceberg");
307        config.set("catalog.uri", "http://localhost:8181");
308        config.set("warehouse", "s3://bucket/wh");
309        config.set("namespace", "prod");
310        config.set("table.name", "events");
311
312        let cfg = IcebergSinkConfig::from_config(&config).unwrap();
313        assert_eq!(cfg.compression, "zstd");
314        assert!(!cfg.auto_create);
315    }
316
317    #[test]
318    fn test_unwired_storage_backends_are_rejected() {
319        let mut config = ConnectorConfig::new("iceberg");
320        config.set("catalog.uri", "http://localhost:8181");
321        config.set("warehouse", "demo");
322        config.set("storage.type", "gcs");
323        config.set("namespace", "prod");
324        config.set("table.name", "events");
325        assert!(IcebergSinkConfig::from_config(&config).is_err());
326
327        config.set("storage.type", "s3");
328        config.set("warehouse", "abfs://container/warehouse");
329        assert!(IcebergSinkConfig::from_config(&config).is_err());
330    }
331
332    // ── Schema validation tests ──
333
334    use arrow_schema::{DataType, Field, Schema};
335
336    fn schema(fields: Vec<(&str, DataType)>) -> Schema {
337        Schema::new(
338            fields
339                .into_iter()
340                .map(|(n, t)| Field::new(n, t, true))
341                .collect::<Vec<_>>(),
342        )
343    }
344
345    #[test]
346    fn test_validate_matching_schemas() {
347        let s = schema(vec![("id", DataType::Int64), ("name", DataType::Utf8)]);
348        assert!(validate_sink_schema(&s, &s).is_ok());
349    }
350
351    #[test]
352    fn test_validate_missing_field() {
353        let pipeline = schema(vec![("id", DataType::Int64), ("extra", DataType::Utf8)]);
354        let table = schema(vec![("id", DataType::Int64)]);
355        let err = validate_sink_schema(&pipeline, &table).unwrap_err();
356        assert!(err.to_string().contains("extra"));
357    }
358
359    #[test]
360    fn test_validate_type_mismatch() {
361        let pipeline = schema(vec![("id", DataType::Int64)]);
362        let table = schema(vec![("id", DataType::Utf8)]);
363        let err = validate_sink_schema(&pipeline, &table).unwrap_err();
364        assert!(err.to_string().contains("incompatible"));
365    }
366
367    #[test]
368    fn test_validate_extra_table_columns_ok() {
369        let pipeline = schema(vec![("id", DataType::Int64)]);
370        let table = schema(vec![("id", DataType::Int64), ("extra", DataType::Utf8)]);
371        assert!(validate_sink_schema(&pipeline, &table).is_ok());
372    }
373
374    #[test]
375    fn test_validate_safe_widening() {
376        let pipeline = schema(vec![("n", DataType::Int32), ("f", DataType::Float32)]);
377        let table = schema(vec![("n", DataType::Int64), ("f", DataType::Float64)]);
378        assert!(validate_sink_schema(&pipeline, &table).is_ok());
379    }
380}
laminar_connectors/lakehouse/iceberg_config.rs

laminar_connectors/lakehouse/
iceberg_config.rs
