laminar_connectors/lakehouse/iceberg_config/
modes.rs

1use std::fmt;
2use std::str::FromStr;
3
4macro_rules! string_enum {
5    (
6        $(#[$meta:meta])*
7        pub enum $name:ident {
8            $($(#[$variant_meta:meta])* $variant:ident => [$canonical:literal $(, $alias:literal)*]),+ $(,)?
9        }
10        default $default:ident;
11        error $error:literal;
12    ) => {
13        $(#[$meta])*
14        #[derive(Debug, Clone, Copy, PartialEq, Eq)]
15        pub enum $name {
16            $($(#[$variant_meta])* $variant),+
17        }
18
19        impl Default for $name {
20            fn default() -> Self {
21                Self::$default
22            }
23        }
24
25        impl FromStr for $name {
26            type Err = String;
27
28            fn from_str(value: &str) -> Result<Self, Self::Err> {
29                match value.trim().to_ascii_lowercase().as_str() {
30                    $($canonical $(| $alias)* => Ok(Self::$variant),)+
31                    other => Err(format!(concat!($error, ": '{}'"), other)),
32                }
33            }
34        }
35
36        impl fmt::Display for $name {
37            fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
38                match self {
39                    $(Self::$variant => formatter.write_str($canonical),)+
40                }
41            }
42        }
43    };
44}
45
46string_enum! {
47    /// Read semantics exposed by the Iceberg source.
48    pub enum IcebergReadMode {
49        /// Read one immutable snapshot and then complete.
50        Snapshot => ["snapshot"],
51        /// Emit a bootstrap snapshot and then append-only snapshot additions.
52        Append => ["append", "incremental"],
53        /// Emit row-level changes when upstream delete semantics are available.
54        Changelog => ["changelog", "cdc"]
55    }
56    default Snapshot;
57    error "invalid Iceberg read.mode";
58}
59
60string_enum! {
61    /// Bootstrap policy for append reads.
62    pub enum IcebergReadBootstrap {
63        /// Emit the selected starting snapshot before following later appends.
64        Initial => ["initial", "snapshot", "current"],
65        /// Start after the selected snapshot without emitting its rows.
66        None => ["none", "skip"]
67    }
68    default Initial;
69    error "invalid Iceberg read.bootstrap";
70}
71
72string_enum! {
73    /// Mutation semantics requested from the Iceberg sink.
74    pub enum IcebergWriteMode {
75        /// Add new data files with an atomic Iceberg `FastAppend`.
76        Append => ["append"],
77        /// Write data and delete files with an atomic `RowDelta`.
78        MergeOnRead => ["merge-on-read", "merge_on_read", "mor"],
79        /// Atomically replace affected data files.
80        CopyOnWrite => ["copy-on-write", "copy_on_write", "cow"]
81    }
82    default Append;
83    error "invalid Iceberg write.mode";
84}
85
86string_enum! {
87    /// How input rows are distributed across partition writers.
88    pub enum IcebergWriteDistributionMode {
89        /// Input is grouped by partition and never returns to a closed partition.
90        Clustered => ["clustered"],
91        /// Interleaved partitions use a bounded set of active writers.
92        Fanout => ["fanout", "hash"]
93    }
94    default Fanout;
95    error "invalid Iceberg write.distribution.mode";
96}
97
98string_enum! {
99    /// Permitted schema changes while a sink is running.
100    pub enum IcebergSchemaEvolutionMode {
101        /// Require the schema bound at open for every checkpoint.
102        Strict => ["strict", "none", "disabled"],
103        /// Permit safe Iceberg promotions and nullable additions.
104        Safe => ["safe", "additive"]
105    }
106    default Strict;
107    error "invalid Iceberg schema.evolution.mode";
108}
109
110string_enum! {
111    /// Iceberg catalog implementation requested by configuration.
112    pub enum IcebergCatalogType {
113        /// Iceberg REST catalog.
114        Rest => ["rest"],
115        /// AWS Glue catalog.
116        Glue => ["glue"],
117        /// Hive Metastore catalog.
118        Hms => ["hms", "hive"],
119        /// Amazon S3 Tables catalog.
120        S3Tables => ["s3tables", "s3-tables"],
121        /// SQL-backed Iceberg catalog.
122        Sql => ["sql"]
123    }
124    default Rest;
125    error "invalid Iceberg catalog.type";
126}
127
128string_enum! {
129    /// Catalog authentication mechanism.
130    pub enum IcebergCatalogAuthType {
131        /// No connector-managed authentication.
132        None => ["none"],
133        /// Static bearer token resolved by the existing secret layer.
134        Bearer => ["bearer", "token"],
135        /// `OAuth2` client credentials with refresh.
136        OAuth2 => ["oauth2", "oauth"]
137    }
138    default None;
139    error "invalid Iceberg catalog.auth.type";
140}
141
142string_enum! {
143    /// Storage backend used for Iceberg table data and metadata.
144    pub enum IcebergStorageType {
145        /// Amazon S3 or an S3-compatible object store.
146        S3 => ["s3", "s3a"],
147        /// Google Cloud Storage.
148        Gcs => ["gcs", "gs"],
149        /// Azure Data Lake Storage Gen2.
150        Azure => ["azure", "azdls", "adls", "blob", "abfs", "abfss", "wasb", "wasbs"],
151        /// Local filesystem storage.
152        Fs => ["fs", "file", "filesystem"]
153    }
154    default S3;
155    error "invalid Iceberg storage.type";
156}
157
158string_enum! {
159    /// Server-side encryption requested for object storage.
160    pub enum IcebergStorageEncryption {
161        /// Use the storage service default.
162        None => ["none"],
163        /// Use service-managed encryption keys.
164        Sse => ["sse", "sse-s3"],
165        /// Use a configured key-management-service key.
166        Kms => ["kms", "sse-kms"]
167    }
168    default None;
169    error "invalid Iceberg storage.encryption";
170}
171
172#[cfg(test)]
173mod tests {
174    use super::*;
175
176    #[test]
177    fn aliases_parse_to_typed_modes() {
178        assert_eq!("incremental".parse(), Ok(IcebergReadMode::Append));
179        assert_eq!("mor".parse(), Ok(IcebergWriteMode::MergeOnRead));
180        assert_eq!("abfss".parse(), Ok(IcebergStorageType::Azure));
181        assert!("overwrite".parse::<IcebergWriteMode>().is_err());
182    }
183}
laminar_connectors/lakehouse/iceberg_config/modes.rs

laminar_connectors/lakehouse/iceberg_config/
modes.rs
