diff --git a/Cargo.lock b/Cargo.lock index f1d00378f7..d6b897a8cf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3788,6 +3788,7 @@ dependencies = [ "fnv", "form_urlencoded", "futures", + "iceberg-property-macro", "iceberg_test_utils", "itertools 0.13.0", "minijinja", diff --git a/Cargo.toml b/Cargo.toml index aec84da8eb..32667b20c7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -108,6 +108,7 @@ iceberg-catalog-rest = { version = "0.10.0", path = "./crates/catalog/rest" } iceberg-catalog-s3tables = { version = "0.10.0", path = "./crates/catalog/s3tables" } iceberg-catalog-sql = { version = "0.10.0", path = "./crates/catalog/sql" } iceberg-datafusion = { version = "0.10.0", path = "./crates/integrations/datafusion" } +iceberg-property-macro = { version = "0.10.0", path = "./crates/property-macro" } iceberg-storage-opendal = { version = "0.10.0", path = "./crates/storage/opendal" } indicatif = "0.18" itertools = "0.13" diff --git a/crates/iceberg/Cargo.toml b/crates/iceberg/Cargo.toml index 9219c728a5..56e11b1804 100644 --- a/crates/iceberg/Cargo.toml +++ b/crates/iceberg/Cargo.toml @@ -60,6 +60,7 @@ flate2 = { workspace = true } fnv = { workspace = true } form_urlencoded = { workspace = true } futures = { workspace = true } +iceberg-property-macro = { workspace = true } itertools = { workspace = true } moka = { version = "0.12.10", features = ["future"] } murmur3 = { workspace = true } diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 9ef584678e..77b736b976 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -2889,9 +2889,39 @@ pub const iceberg::spec::TableProperties::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES: pub const iceberg::spec::TableProperties::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT: usize pub const iceberg::spec::TableProperties::RESERVED_PROPERTIES: [&str; 9] pub fn iceberg::spec::TableProperties::data_encryption_key_size(&self) -> iceberg::Result +impl iceberg::spec::TableProperties +pub fn iceberg::spec::TableProperties::cdc_enabled(&self) -> bool +pub fn iceberg::spec::TableProperties::cdc_max_chunk_size(&self) -> usize +pub fn iceberg::spec::TableProperties::cdc_min_chunk_size(&self) -> usize +pub fn iceberg::spec::TableProperties::cdc_norm_level(&self) -> i32 +pub fn iceberg::spec::TableProperties::commit_max_retry_wait_ms(&self) -> u64 +pub fn iceberg::spec::TableProperties::commit_min_retry_wait_ms(&self) -> u64 +pub fn iceberg::spec::TableProperties::commit_num_retries(&self) -> usize +pub fn iceberg::spec::TableProperties::commit_total_retry_timeout_ms(&self) -> u64 +pub fn iceberg::spec::TableProperties::encryption_data_key_length(&self) -> usize +pub fn iceberg::spec::TableProperties::encryption_key_id(&self) -> &core::option::Option +pub fn iceberg::spec::TableProperties::from_properties(properties: &std::collections::hash::map::HashMap) -> iceberg::Result +pub fn iceberg::spec::TableProperties::gc_enabled(&self) -> bool +pub fn iceberg::spec::TableProperties::max_ref_age_ms(&self) -> i64 +pub fn iceberg::spec::TableProperties::max_snapshot_age_ms(&self) -> i64 +pub fn iceberg::spec::TableProperties::metadata_compression_codec(&self) -> &iceberg::compression::CompressionCodec +pub fn iceberg::spec::TableProperties::min_snapshots_to_keep(&self) -> usize +pub fn iceberg::spec::TableProperties::parquet_compression_codec(&self) -> &iceberg::compression::CompressionCodec +pub fn iceberg::spec::TableProperties::parquet_dict_size_bytes(&self) -> usize +pub fn iceberg::spec::TableProperties::parquet_page_row_limit(&self) -> usize +pub fn iceberg::spec::TableProperties::parquet_page_size_bytes(&self) -> usize +pub fn iceberg::spec::TableProperties::parquet_row_group_size_bytes(&self) -> usize +pub fn iceberg::spec::TableProperties::write_data_location(&self) -> &core::option::Option +pub fn iceberg::spec::TableProperties::write_datafusion_fanout_enabled(&self) -> bool +pub fn iceberg::spec::TableProperties::write_folder_storage_location(&self) -> &core::option::Option +pub fn iceberg::spec::TableProperties::write_format_default(&self) -> &alloc::string::String +pub fn iceberg::spec::TableProperties::write_metadata_path(&self) -> &core::option::Option +pub fn iceberg::spec::TableProperties::write_object_storage_location(&self) -> &core::option::Option +pub fn iceberg::spec::TableProperties::write_object_storage_partitioned_paths(&self) -> bool +pub fn iceberg::spec::TableProperties::write_target_file_size_bytes(&self) -> usize impl core::convert::TryFrom<&std::collections::hash::map::HashMap> for iceberg::spec::TableProperties pub type iceberg::spec::TableProperties::Error = iceberg::Error -pub fn iceberg::spec::TableProperties::try_from(props: &std::collections::hash::map::HashMap) -> iceberg::Result +pub fn iceberg::spec::TableProperties::try_from(properties: &std::collections::hash::map::HashMap) -> iceberg::Result impl core::fmt::Debug for iceberg::spec::TableProperties pub fn iceberg::spec::TableProperties::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result pub struct iceberg::spec::UnboundPartitionField diff --git a/crates/iceberg/src/spec/table_properties.rs b/crates/iceberg/src/spec/table_properties.rs index 06dbe366cc..0a9393f12d 100644 --- a/crates/iceberg/src/spec/table_properties.rs +++ b/crates/iceberg/src/spec/table_properties.rs @@ -16,71 +16,20 @@ // under the License. use std::collections::HashMap; -use std::fmt::Display; -use std::str::FromStr; + +use iceberg_property_macro::Properties; use crate::compression::CompressionCodec; use crate::encryption::AesKeySize; use crate::error::{Error, ErrorKind, Result}; use crate::util::location::strip_trailing_slash; -fn parse_property( - properties: &HashMap, - key: &str, - default: T, -) -> Result -where - ::Err: Display, -{ - properties.get(key).map_or(Ok(default), |value| { - value.parse::().map_err(|e| { - Error::new( - ErrorKind::DataInvalid, - format!("Invalid value for {key}: {e}"), - ) - }) - }) -} - -/// Parse an optional property, returning `None` when the key is absent and an -/// error when the value is present but fails to parse. -fn parse_optional_property( - properties: &HashMap, - key: &str, -) -> Result> -where - ::Err: Display, -{ - properties - .get(key) - .map(|value| { - value.parse::().map_err(|e| { - Error::new( - ErrorKind::DataInvalid, - format!("Invalid value for {key}: {e}"), - ) - }) - }) - .transpose() -} - -fn parse_location_property( - properties: &HashMap, - key: &str, -) -> Result> { - properties - .get(key) - .map(|path| { - if path.is_empty() { - return Err(Error::new( - ErrorKind::DataInvalid, - format!("Invalid value for {key}: path must not be empty"), - )); - } +fn parse_location_property(path: &str) -> Result { + if path.is_empty() { + return Err(Error::new(ErrorKind::DataInvalid, "path must not be empty")); + } - Ok(strip_trailing_slash(path).to_string()) - }) - .transpose() + Ok(strip_trailing_slash(path).to_string()) } /// Parse compression codec for metadata files from table properties. @@ -103,6 +52,10 @@ pub(crate) fn parse_metadata_file_compression( .map(|s| s.as_str()) .unwrap_or(TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC_DEFAULT); + parse_metadata_compression(value) +} + +fn parse_metadata_compression(value: &str) -> Result { // Handle empty string as None if value.is_empty() { return Ok(CompressionCodec::None); @@ -143,27 +96,40 @@ pub(crate) fn parse_metadata_file_compression( /// Parse the Parquet data-file compression codec (`write.parquet.compression-codec`) /// and fold in the compression level (`write.parquet.compression-level`) for the /// codecs that accept one (`zstd`, `gzip`, `brotli`). -fn parse_parquet_compression(properties: &HashMap) -> Result { - let value = properties - .get(TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC) - .map(|s| s.as_str()) - .unwrap_or(TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC_DEFAULT); - - let codec: CompressionCodec = - serde_json::from_value(serde_json::Value::String(value.to_lowercase())).map_err(|_| { - Error::new( - ErrorKind::DataInvalid, - format!( - "Invalid Parquet compression codec: {value}. Supported codecs: \ - uncompressed, snappy, gzip, lzo, brotli, lz4, lz4_raw, zstd" - ), - ) - })?; +fn parse_parquet_compression( + properties: &HashMap, + codec_key: &str, + additional_keys: &[&str], + default: CompressionCodec, +) -> Result { + let level_key = additional_keys[0]; + let codec = properties + .get(codec_key) + .map(|value| { + serde_json::from_value(serde_json::Value::String(value.to_lowercase())).map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid Parquet compression codec: {value}. Supported codecs: \ + uncompressed, snappy, gzip, lzo, brotli, lz4, lz4_raw, zstd" + ), + ) + }) + }) + .transpose()? + .unwrap_or(default); - let level: Option = parse_optional_property( - properties, - TableProperties::PROPERTY_PARQUET_COMPRESSION_LEVEL, - )?; + let level = properties + .get(level_key) + .map(|value| { + value.parse::().map_err(|error| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid value for {level_key}: {error}"), + ) + }) + }) + .transpose()?; Ok(match (codec, level) { (CompressionCodec::Zstd(_), Some(level)) => CompressionCodec::Zstd(level), @@ -173,94 +139,224 @@ fn parse_parquet_compression(properties: &HashMap) -> Result, - key: &str, - default: bool, -) -> Result { - properties.get(key).map_or(Ok(default), |value| { - value.to_lowercase().parse::().map_err(|e| { - Error::new( - ErrorKind::DataInvalid, - format!("Invalid value for {key}: {e}"), - ) - }) - }) -} - /// TableProperties that contains the properties of a table. -#[derive(Debug)] +#[derive(Debug, Properties)] pub struct TableProperties { /// The number of times to retry a commit. + #[property( + key = Self::PROPERTY_COMMIT_NUM_RETRIES, + default = Self::PROPERTY_COMMIT_NUM_RETRIES_DEFAULT, + getter + )] pub commit_num_retries: usize, /// The minimum wait time between retries. + #[property( + key = Self::PROPERTY_COMMIT_MIN_RETRY_WAIT_MS, + default = Self::PROPERTY_COMMIT_MIN_RETRY_WAIT_MS_DEFAULT, + getter + )] pub commit_min_retry_wait_ms: u64, /// The maximum wait time between retries. + #[property( + key = Self::PROPERTY_COMMIT_MAX_RETRY_WAIT_MS, + default = Self::PROPERTY_COMMIT_MAX_RETRY_WAIT_MS_DEFAULT, + getter + )] pub commit_max_retry_wait_ms: u64, /// The total timeout for commit retries. + #[property( + key = Self::PROPERTY_COMMIT_TOTAL_RETRY_TIME_MS, + default = Self::PROPERTY_COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT, + getter + )] pub commit_total_retry_timeout_ms: u64, /// The default format for files. + #[property( + key = Self::PROPERTY_DEFAULT_FILE_FORMAT, + default = Self::PROPERTY_DEFAULT_FILE_FORMAT_DEFAULT, + getter + )] pub write_format_default: String, /// The target file size for files. + #[property( + key = Self::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES, + default = Self::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT, + getter + )] pub write_target_file_size_bytes: usize, /// Base directory for metadata files (manifests, manifest lists), with any /// trailing slash trimmed. `None` if `write.metadata.path` is not set. + #[property( + key = Self::PROPERTY_WRITE_METADATA_PATH, + default = None, + parse_with = parse_location_property, + getter + )] pub write_metadata_path: Option, /// Compression codec for metadata files (JSON) + #[property( + key = Self::PROPERTY_METADATA_COMPRESSION_CODEC, + default = CompressionCodec::None, + parse_with = parse_metadata_compression, + getter + )] pub metadata_compression_codec: CompressionCodec, /// Whether to use `FanoutWriter` for partitioned tables. + #[property( + key = Self::PROPERTY_DATAFUSION_WRITE_FANOUT_ENABLED, + default = Self::PROPERTY_DATAFUSION_WRITE_FANOUT_ENABLED_DEFAULT, + getter + )] pub write_datafusion_fanout_enabled: bool, /// Whether garbage collection is enabled on drop. /// When `false`, data files will not be deleted when a table is dropped. + #[property( + key = Self::PROPERTY_GC_ENABLED, + default = Self::PROPERTY_GC_ENABLED_DEFAULT, + getter + )] pub gc_enabled: bool, /// Default maximum age of a snapshot to keep when expiring snapshots. + #[property( + key = Self::PROPERTY_MAX_SNAPSHOT_AGE_MS, + default = Self::PROPERTY_MAX_SNAPSHOT_AGE_MS_DEFAULT, + getter + )] pub max_snapshot_age_ms: i64, /// Default minimum number of snapshots to keep per branch when expiring snapshots. + #[property( + key = Self::PROPERTY_MIN_SNAPSHOTS_TO_KEEP, + default = Self::PROPERTY_MIN_SNAPSHOTS_TO_KEEP_DEFAULT, + getter + )] pub min_snapshots_to_keep: usize, /// Default maximum age of a snapshot reference to keep when expiring snapshots. + #[property( + key = Self::PROPERTY_MAX_REF_AGE_MS, + default = Self::PROPERTY_MAX_REF_AGE_MS_DEFAULT, + getter + )] pub max_ref_age_ms: i64, /// Whether content-defined chunking is enabled. /// `true` only when `write.parquet.content-defined-chunking.enabled = "true"`. + #[property( + key = Self::PROPERTY_PARQUET_CDC_ENABLED, + default = Self::PROPERTY_PARQUET_CDC_ENABLED_DEFAULT, + getter + )] pub cdc_enabled: bool, /// Content-defined chunking minimum chunk size in bytes. + #[property( + key = Self::PROPERTY_PARQUET_CDC_MIN_CHUNK_SIZE, + default = Self::PROPERTY_PARQUET_CDC_MIN_CHUNK_SIZE_DEFAULT, + getter + )] pub cdc_min_chunk_size: usize, /// Content-defined chunking maximum chunk size in bytes. + #[property( + key = Self::PROPERTY_PARQUET_CDC_MAX_CHUNK_SIZE, + default = Self::PROPERTY_PARQUET_CDC_MAX_CHUNK_SIZE_DEFAULT, + getter + )] pub cdc_max_chunk_size: usize, /// Content-defined chunking normalization level (gearhash bit adjustment). + #[property( + key = Self::PROPERTY_PARQUET_CDC_NORM_LEVEL, + default = Self::PROPERTY_PARQUET_CDC_NORM_LEVEL_DEFAULT, + getter + )] pub cdc_norm_level: i32, /// Parquet compression codec for data files, with the resolved compression /// level folded in (from `write.parquet.compression-level`, or the codec's /// default when unset). + #[property( + key = Self::PROPERTY_PARQUET_COMPRESSION_CODEC, + additional_keys = [Self::PROPERTY_PARQUET_COMPRESSION_LEVEL], + default = CompressionCodec::zstd_default(), + parse_properties_with = parse_parquet_compression, + getter + )] pub parquet_compression_codec: CompressionCodec, /// Approximate maximum Parquet row group size in bytes. + #[property( + key = Self::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES, + default = Self::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT, + getter + )] pub parquet_row_group_size_bytes: usize, /// Approximate maximum Parquet data page size in bytes. + #[property( + key = Self::PROPERTY_PARQUET_PAGE_SIZE_BYTES, + default = Self::PROPERTY_PARQUET_PAGE_SIZE_BYTES_DEFAULT, + getter + )] pub parquet_page_size_bytes: usize, /// Maximum number of rows per Parquet data page. + #[property( + key = Self::PROPERTY_PARQUET_PAGE_ROW_LIMIT, + default = Self::PROPERTY_PARQUET_PAGE_ROW_LIMIT_DEFAULT, + getter + )] pub parquet_page_row_limit: usize, /// Approximate maximum Parquet dictionary page size in bytes. + #[property( + key = Self::PROPERTY_PARQUET_DICT_SIZE_BYTES, + default = Self::PROPERTY_PARQUET_DICT_SIZE_BYTES_DEFAULT, + getter + )] pub parquet_dict_size_bytes: usize, /// The master key id used to encrypt this table's manifest list and data /// files. `None` if `encryption.key-id` is not set. + #[property( + key = Self::PROPERTY_ENCRYPTION_KEY_ID, + default = None, + getter + )] pub encryption_key_id: Option, /// The encryption data encryption key length in bytes. + #[property( + key = Self::PROPERTY_ENCRYPTION_DATA_KEY_LENGTH, + default = Self::PROPERTY_ENCRYPTION_DATA_KEY_LENGTH_DEFAULT, + getter + )] pub encryption_data_key_length: usize, - /// Base directory for data files + /// Base directory for data files, with any trailing slash trimmed. + #[property( + key = Self::PROPERTY_WRITE_DATA_LOCATION, + default = None, + parse_with = parse_location_property, + getter + )] pub write_data_location: Option, - /// Deprecated table property for data file write location. + /// Deprecated table property for data file write location, with any trailing slash trimmed. /// /// Property will be removed at a later date. /// Superseded by [write_data_location]. + #[property( + key = Self::PROPERTY_WRITE_FOLDER_STORAGE_LOCATION, + default = None, + parse_with = parse_location_property, + getter + )] pub write_folder_storage_location: Option, - /// Deprecated table property for data file write location for object storage location generator. + /// Deprecated table property for data file write location for object storage location generator, + /// with any trailing slash trimmed. /// /// Property will be removed at a later date. /// Superseded by [write_data_location]. + #[property( + key = Self::PROPERTY_WRITE_OBJECT_STORAGE_LOCATION, + default = None, + parse_with = parse_location_property, + getter + )] pub write_object_storage_location: Option, /// Whether partition values are included in object storage paths. + #[property( + key = Self::PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS, + default = Self::PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS_DEFAULT, + getter + )] pub write_object_storage_partitioned_paths: bool, } @@ -465,135 +561,10 @@ impl TableProperties { } impl TryFrom<&HashMap> for TableProperties { - // parse by entry key or use default value type Error = Error; - fn try_from(props: &HashMap) -> Result { - Ok(TableProperties { - commit_num_retries: parse_property( - props, - TableProperties::PROPERTY_COMMIT_NUM_RETRIES, - TableProperties::PROPERTY_COMMIT_NUM_RETRIES_DEFAULT, - )?, - commit_min_retry_wait_ms: parse_property( - props, - TableProperties::PROPERTY_COMMIT_MIN_RETRY_WAIT_MS, - TableProperties::PROPERTY_COMMIT_MIN_RETRY_WAIT_MS_DEFAULT, - )?, - commit_max_retry_wait_ms: parse_property( - props, - TableProperties::PROPERTY_COMMIT_MAX_RETRY_WAIT_MS, - TableProperties::PROPERTY_COMMIT_MAX_RETRY_WAIT_MS_DEFAULT, - )?, - commit_total_retry_timeout_ms: parse_property( - props, - TableProperties::PROPERTY_COMMIT_TOTAL_RETRY_TIME_MS, - TableProperties::PROPERTY_COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT, - )?, - write_format_default: parse_property( - props, - TableProperties::PROPERTY_DEFAULT_FILE_FORMAT, - TableProperties::PROPERTY_DEFAULT_FILE_FORMAT_DEFAULT.to_string(), - )?, - write_target_file_size_bytes: parse_property( - props, - TableProperties::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES, - TableProperties::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT, - )?, - write_metadata_path: parse_location_property( - props, - TableProperties::PROPERTY_WRITE_METADATA_PATH, - )?, - metadata_compression_codec: parse_metadata_file_compression(props)?, - write_datafusion_fanout_enabled: parse_property_bool( - props, - TableProperties::PROPERTY_DATAFUSION_WRITE_FANOUT_ENABLED, - TableProperties::PROPERTY_DATAFUSION_WRITE_FANOUT_ENABLED_DEFAULT, - )?, - gc_enabled: parse_property_bool( - props, - TableProperties::PROPERTY_GC_ENABLED, - TableProperties::PROPERTY_GC_ENABLED_DEFAULT, - )?, - max_snapshot_age_ms: parse_property( - props, - TableProperties::PROPERTY_MAX_SNAPSHOT_AGE_MS, - TableProperties::PROPERTY_MAX_SNAPSHOT_AGE_MS_DEFAULT, - )?, - min_snapshots_to_keep: parse_property( - props, - TableProperties::PROPERTY_MIN_SNAPSHOTS_TO_KEEP, - TableProperties::PROPERTY_MIN_SNAPSHOTS_TO_KEEP_DEFAULT, - )?, - max_ref_age_ms: parse_property( - props, - TableProperties::PROPERTY_MAX_REF_AGE_MS, - TableProperties::PROPERTY_MAX_REF_AGE_MS_DEFAULT, - )?, - cdc_enabled: parse_property_bool( - props, - TableProperties::PROPERTY_PARQUET_CDC_ENABLED, - TableProperties::PROPERTY_PARQUET_CDC_ENABLED_DEFAULT, - )?, - cdc_min_chunk_size: parse_property( - props, - TableProperties::PROPERTY_PARQUET_CDC_MIN_CHUNK_SIZE, - TableProperties::PROPERTY_PARQUET_CDC_MIN_CHUNK_SIZE_DEFAULT, - )?, - cdc_max_chunk_size: parse_property( - props, - TableProperties::PROPERTY_PARQUET_CDC_MAX_CHUNK_SIZE, - TableProperties::PROPERTY_PARQUET_CDC_MAX_CHUNK_SIZE_DEFAULT, - )?, - cdc_norm_level: parse_property( - props, - TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL, - TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL_DEFAULT, - )?, - parquet_compression_codec: parse_parquet_compression(props)?, - parquet_row_group_size_bytes: parse_property( - props, - TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES, - TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT, - )?, - parquet_page_size_bytes: parse_property( - props, - TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES, - TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES_DEFAULT, - )?, - parquet_page_row_limit: parse_property( - props, - TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT, - TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT_DEFAULT, - )?, - parquet_dict_size_bytes: parse_property( - props, - TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES, - TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES_DEFAULT, - )?, - encryption_key_id: props - .get(TableProperties::PROPERTY_ENCRYPTION_KEY_ID) - .cloned(), - encryption_data_key_length: parse_property( - props, - TableProperties::PROPERTY_ENCRYPTION_DATA_KEY_LENGTH, - TableProperties::PROPERTY_ENCRYPTION_DATA_KEY_LENGTH_DEFAULT, - )?, - write_data_location: props - .get(TableProperties::PROPERTY_WRITE_DATA_LOCATION) - .cloned(), - write_folder_storage_location: props - .get(TableProperties::PROPERTY_WRITE_FOLDER_STORAGE_LOCATION) - .cloned(), - write_object_storage_location: props - .get(TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_LOCATION) - .cloned(), - write_object_storage_partitioned_paths: parse_property_bool( - props, - TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS, - TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS_DEFAULT, - )?, - }) + fn try_from(properties: &HashMap) -> Result { + Self::from_properties(properties) } } @@ -672,33 +643,50 @@ mod tests { } #[test] - fn test_table_properties_write_metadata_path() { - // Test unset + fn test_table_properties_location_paths() { + // Test unset. let table_properties = TableProperties::try_from(&HashMap::new()).unwrap(); assert_eq!(table_properties.write_metadata_path, None); - - // Test empty path is invalid - let props = HashMap::from([( - TableProperties::PROPERTY_WRITE_METADATA_PATH.to_string(), - String::new(), - )]); - let error = TableProperties::try_from(&props).unwrap_err(); - assert_eq!(error.kind(), ErrorKind::DataInvalid); - assert!( - error - .message() - .contains(TableProperties::PROPERTY_WRITE_METADATA_PATH) - ); - - let props = HashMap::from([( - TableProperties::PROPERTY_WRITE_METADATA_PATH.to_string(), - "s3://other-bucket/custom-meta/".to_string(), - )]); - let table_properties = TableProperties::try_from(&props).unwrap(); - assert_eq!( - table_properties.write_metadata_path.as_deref(), - Some("s3://other-bucket/custom-meta") - ); + assert_eq!(table_properties.write_data_location, None); + assert_eq!(table_properties.write_folder_storage_location, None); + assert_eq!(table_properties.write_object_storage_location, None); + + for key in [ + TableProperties::PROPERTY_WRITE_METADATA_PATH, + TableProperties::PROPERTY_WRITE_DATA_LOCATION, + TableProperties::PROPERTY_WRITE_FOLDER_STORAGE_LOCATION, + TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_LOCATION, + ] { + // Test empty paths are invalid and retain the property key as error context. + let error = + TableProperties::try_from(&HashMap::from([(key.to_string(), String::new())])) + .unwrap_err(); + assert_eq!(error.kind(), ErrorKind::DataInvalid); + assert!(format!("{error}").contains(key)); + + // Test all supported location properties share trailing-slash normalization. + let table_properties = TableProperties::try_from(&HashMap::from([( + key.to_string(), + "s3://other-bucket/custom-path/".to_string(), + )])) + .unwrap(); + let parsed = match key { + TableProperties::PROPERTY_WRITE_METADATA_PATH => { + table_properties.write_metadata_path.as_deref() + } + TableProperties::PROPERTY_WRITE_DATA_LOCATION => { + table_properties.write_data_location.as_deref() + } + TableProperties::PROPERTY_WRITE_FOLDER_STORAGE_LOCATION => { + table_properties.write_folder_storage_location.as_deref() + } + TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_LOCATION => { + table_properties.write_object_storage_location.as_deref() + } + _ => unreachable!(), + }; + assert_eq!(parsed, Some("s3://other-bucket/custom-path")); + } } #[test] diff --git a/crates/property-macro/Cargo.toml b/crates/property-macro/Cargo.toml index 700b2ecf44..cc2e4edd6b 100644 --- a/crates/property-macro/Cargo.toml +++ b/crates/property-macro/Cargo.toml @@ -19,7 +19,7 @@ edition = { workspace = true } homepage = { workspace = true } name = "iceberg-property-macro" -publish = false +publish = true readme = "README.md" rust-version = { workspace = true } version = { workspace = true } diff --git a/crates/property-macro/public-api.txt b/crates/property-macro/public-api.txt new file mode 100644 index 0000000000..8edb5d0952 --- /dev/null +++ b/crates/property-macro/public-api.txt @@ -0,0 +1,2 @@ +pub mod iceberg_property_macro +pub proc macro iceberg_property_macro::#[derive(Properties)] diff --git a/crates/property-macro/src/properties.rs b/crates/property-macro/src/properties.rs index 03820311fb..b4fcf2d63e 100644 --- a/crates/property-macro/src/properties.rs +++ b/crates/property-macro/src/properties.rs @@ -128,6 +128,7 @@ pub(crate) fn expand_properties(input: DeriveInput) -> syn::Result impl #impl_generics #struct_name #type_generics #where_clause { #(#accessors)* + /// Parses this typed property set from a flat string-to-string map. pub fn from_properties( properties: &::std::collections::HashMap< ::std::string::String,