laminar_connectors/lakehouse/iceberg_config/
table_definition.rs

1use std::collections::{HashMap, HashSet};
2use std::fmt;
3use std::str::FromStr;
4
5use serde::Deserialize;
6
7use crate::config::ConnectorConfig;
8use crate::error::ConnectorError;
9
10use super::validate_property_map_bounds;
11
12pub(crate) const TARGET_FILE_SIZE_PROPERTY: &str = "write.target-file-size-bytes";
13pub(crate) const PARQUET_ROW_GROUP_SIZE_PROPERTY: &str = "write.parquet.row-group-size-bytes";
14pub(crate) const PARQUET_COMPRESSION_PROPERTY: &str = "write.parquet.compression-codec";
15const MAX_TABLE_DEFINITION_FIELDS: usize = 128;
16const MAX_TABLE_DEFINITION_BYTES: usize = 1024 * 1024;
17
18/// Iceberg partition or sort transform used for table creation.
19#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
20pub enum IcebergTransform {
21    /// Preserve the source value.
22    Identity,
23    /// Hash into a fixed number of buckets.
24    Bucket(u32),
25    /// Truncate values to a fixed width.
26    Truncate(u32),
27    /// Extract the calendar year.
28    Year,
29    /// Extract the calendar month.
30    Month,
31    /// Extract the calendar day.
32    Day,
33    /// Extract the calendar hour.
34    Hour,
35    /// Always produce null.
36    Void,
37}
38
39impl FromStr for IcebergTransform {
40    type Err = String;
41
42    fn from_str(value: &str) -> Result<Self, Self::Err> {
43        let value = value.trim().to_ascii_lowercase();
44        match value.as_str() {
45            "identity" => Ok(Self::Identity),
46            "year" => Ok(Self::Year),
47            "month" => Ok(Self::Month),
48            "day" => Ok(Self::Day),
49            "hour" => Ok(Self::Hour),
50            "void" => Ok(Self::Void),
51            _ => parameterized_transform(&value, "bucket", Self::Bucket)
52                .or_else(|| parameterized_transform(&value, "truncate", Self::Truncate))
53                .ok_or_else(|| format!("unsupported Iceberg transform '{value}'")),
54        }
55    }
56}
57
58impl<'de> Deserialize<'de> for IcebergTransform {
59    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
60    where
61        D: serde::Deserializer<'de>,
62    {
63        String::deserialize(deserializer)?
64            .parse()
65            .map_err(serde::de::Error::custom)
66    }
67}
68
69impl fmt::Display for IcebergTransform {
70    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
71        match self {
72            Self::Identity => formatter.write_str("identity"),
73            Self::Bucket(count) => write!(formatter, "bucket[{count}]"),
74            Self::Truncate(width) => write!(formatter, "truncate[{width}]"),
75            Self::Year => formatter.write_str("year"),
76            Self::Month => formatter.write_str("month"),
77            Self::Day => formatter.write_str("day"),
78            Self::Hour => formatter.write_str("hour"),
79            Self::Void => formatter.write_str("void"),
80        }
81    }
82}
83
84fn parameterized_transform(
85    value: &str,
86    name: &str,
87    constructor: impl FnOnce(u32) -> IcebergTransform,
88) -> Option<IcebergTransform> {
89    let parameter = value
90        .strip_prefix(name)?
91        .strip_prefix('[')?
92        .strip_suffix(']')?;
93    let parameter = parameter.parse::<u32>().ok().filter(|value| *value > 0)?;
94    Some(constructor(parameter))
95}
96
97/// One partition field in an auto-created Iceberg table.
98#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
99#[serde(deny_unknown_fields)]
100pub struct IcebergPartitionField {
101    /// Source field name, including a dotted nested path when needed.
102    pub source: String,
103    /// Stable partition field name stored in Iceberg metadata.
104    pub name: String,
105    /// Transform applied to the source field.
106    pub transform: IcebergTransform,
107}
108
109/// Sort direction used by an auto-created Iceberg table.
110#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
111#[serde(rename_all = "lowercase")]
112pub enum IcebergSortDirection {
113    /// Ascending order.
114    Asc,
115    /// Descending order.
116    Desc,
117}
118
119/// Null placement used by an auto-created Iceberg table.
120#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
121#[serde(rename_all = "kebab-case")]
122pub enum IcebergNullOrder {
123    /// Nulls precede non-null values.
124    NullsFirst,
125    /// Nulls follow non-null values.
126    NullsLast,
127}
128
129/// One sort field in an auto-created Iceberg table.
130#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
131#[serde(deny_unknown_fields)]
132pub struct IcebergSortField {
133    /// Source field name, including a dotted nested path when needed.
134    pub source: String,
135    /// Transform applied before ordering.
136    pub transform: IcebergTransform,
137    /// Sort direction.
138    pub direction: IcebergSortDirection,
139    /// Placement of null values.
140    pub null_order: IcebergNullOrder,
141}
142
143pub(crate) fn parse_parquet_compression(value: &str) -> Result<String, ConnectorError> {
144    match value.trim().to_ascii_lowercase().as_str() {
145        "zstd" => Ok("zstd".into()),
146        "snappy" => Ok("snappy".into()),
147        "lz4" => Ok("lz4".into()),
148        "none" | "uncompressed" => Ok("uncompressed".into()),
149        other => Err(ConnectorError::ConfigurationError(format!(
150            "unsupported parquet.compression '{other}'; expected zstd, snappy, lz4, or uncompressed"
151        ))),
152    }
153}
154
155pub(crate) fn parse_table_fields<T>(
156    config: &ConnectorConfig,
157    key: &str,
158) -> Result<Vec<T>, ConnectorError>
159where
160    T: serde::de::DeserializeOwned,
161{
162    let Some(value) = config.get(key) else {
163        return Ok(Vec::new());
164    };
165    if value.trim().is_empty() {
166        return Ok(Vec::new());
167    }
168    if value.len() > MAX_TABLE_DEFINITION_BYTES {
169        return Err(ConnectorError::ConfigurationError(format!(
170            "{key} is {} bytes; the limit is {MAX_TABLE_DEFINITION_BYTES}",
171            value.len()
172        )));
173    }
174    let fields = serde_json::from_str::<Vec<T>>(value).map_err(|error| {
175        ConnectorError::ConfigurationError(format!("invalid {key} JSON: {error}"))
176    })?;
177    if fields.len() > MAX_TABLE_DEFINITION_FIELDS {
178        return Err(ConnectorError::ConfigurationError(format!(
179            "{key} contains {} fields; the limit is {MAX_TABLE_DEFINITION_FIELDS}",
180            fields.len()
181        )));
182    }
183    Ok(fields)
184}
185
186pub(crate) fn validate_distinct_names(values: &[String], key: &str) -> Result<(), ConnectorError> {
187    if values.len() > MAX_TABLE_DEFINITION_FIELDS {
188        return Err(ConnectorError::ConfigurationError(format!(
189            "{key} contains {} fields; the limit is {MAX_TABLE_DEFINITION_FIELDS}",
190            values.len()
191        )));
192    }
193    let mut names = HashSet::with_capacity(values.len());
194    for value in values {
195        if value.is_empty() || value.trim() != value || !names.insert(value) {
196            return Err(ConnectorError::ConfigurationError(format!(
197                "{key} contains an empty, whitespace-padded, or duplicate field name"
198            )));
199        }
200    }
201    Ok(())
202}
203
204pub(crate) fn validate_table_definition(
205    partition_spec: &[IcebergPartitionField],
206    sort_order: &[IcebergSortField],
207) -> Result<(), ConnectorError> {
208    let mut partition_names = HashSet::with_capacity(partition_spec.len());
209    let mut partition_sources = HashSet::with_capacity(partition_spec.len());
210    for field in partition_spec {
211        if field.source.trim() != field.source
212            || field.name.trim() != field.name
213            || field.source.is_empty()
214            || field.name.is_empty()
215            || !partition_names.insert(field.name.as_str())
216            || !partition_sources.insert((field.source.as_str(), field.transform))
217        {
218            return Err(ConnectorError::ConfigurationError(
219                "partition.spec contains an empty, whitespace-padded, or duplicate field".into(),
220            ));
221        }
222    }
223    let mut sort_sources = HashSet::with_capacity(sort_order.len());
224    for field in sort_order {
225        if field.source.trim() != field.source
226            || field.source.is_empty()
227            || !sort_sources.insert((field.source.as_str(), field.transform))
228        {
229            return Err(ConnectorError::ConfigurationError(
230                "sort.order contains an empty, whitespace-padded, or duplicate field".into(),
231            ));
232        }
233    }
234    Ok(())
235}
236
237pub(crate) fn validate_persisted_properties(
238    properties: &HashMap<String, String>,
239) -> Result<(), ConnectorError> {
240    validate_property_map_bounds("table.property.*", properties)?;
241    if let Some(key) = properties.keys().find(|key| {
242        matches!(
243            key.as_str(),
244            TARGET_FILE_SIZE_PROPERTY
245                | PARQUET_ROW_GROUP_SIZE_PROPERTY
246                | PARQUET_COMPRESSION_PROPERTY
247        )
248    }) {
249        return Err(ConnectorError::ConfigurationError(format!(
250            "table.property.{key} duplicates a typed writer option"
251        )));
252    }
253    if let Some(key) = properties.keys().find(|key| {
254        let normalized = key.to_ascii_lowercase().replace(['.', '-'], "_");
255        crate::security::is_secret_option_key(key)
256            || normalized.contains("access_key")
257            || normalized.contains("service_account")
258    }) {
259        return Err(ConnectorError::ConfigurationError(format!(
260            "table.property.{key} cannot persist credential material in Iceberg metadata"
261        )));
262    }
263    if let Some((key, _)) = properties
264        .iter()
265        .find(|(_, value)| crate::security::value_contains_uri_secret(value, false))
266    {
267        return Err(ConnectorError::ConfigurationError(format!(
268            "table.property.{key} contains credentials in a URI and cannot be persisted"
269        )));
270    }
271    Ok(())
272}
273
274#[cfg(test)]
275mod tests {
276    use super::*;
277
278    #[test]
279    fn parameterized_transforms_require_positive_values() {
280        assert_eq!("bucket[32]".parse(), Ok(IcebergTransform::Bucket(32)));
281        assert_eq!("truncate[8]".parse(), Ok(IcebergTransform::Truncate(8)));
282        assert!("bucket[0]".parse::<IcebergTransform>().is_err());
283        assert!("bucket(16)".parse::<IcebergTransform>().is_err());
284    }
285
286    #[test]
287    fn metadata_identifiers_are_not_confused_with_credential_material() {
288        let properties = HashMap::from([
289            ("tenant_id".to_string(), "tenant-reference".to_string()),
290            ("client_id".to_string(), "client-reference".to_string()),
291            ("profile".to_string(), "analytics".to_string()),
292        ]);
293        validate_persisted_properties(&properties).unwrap();
294    }
295
296    #[test]
297    fn access_keys_remain_forbidden_in_table_metadata() {
298        for key in ["access-key-id", "aws_access_key_id", "secret_access_key"] {
299            let properties = HashMap::from([(key.to_string(), "credential".to_string())]);
300            assert!(validate_persisted_properties(&properties).is_err(), "{key}");
301        }
302    }
303
304    #[test]
305    fn service_account_material_remains_forbidden_in_table_metadata() {
306        for key in [
307            "service_account",
308            "service-account-json",
309            "gcs.service_account_path",
310        ] {
311            let properties =
312                HashMap::from([(key.to_string(), "private-credential-value".to_string())]);
313            let error = validate_persisted_properties(&properties).unwrap_err();
314            assert!(error.to_string().contains(key), "{key}: {error}");
315            assert!(!error.to_string().contains("private-credential-value"));
316        }
317    }
318}
laminar_connectors/lakehouse/iceberg_config/table_definition.rs

laminar_connectors/lakehouse/iceberg_config/
table_definition.rs
