blackmwk commented on code in PR #2955: URL: https://github.com/apache/iceberg-rust/pull/2955#discussion_r3718417931
########## crates/iceberg/src/spec/parsed_table_prop.rs: ########## @@ -0,0 +1,1181 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Typed access to Iceberg table properties. +//! +//! [`ParsedTableProperties`] exposes Iceberg's string-keyed table properties as typed public +//! fields. Its JSON representation is a flat object whose keys and values are strings. +//! +//! # Create from defaults +//! +//! Start with Iceberg's defaults and modify public fields directly: +//! +//! ``` +//! use iceberg::spec::{DataFileFormat, ParsedTableProperties}; +//! +//! let mut properties = ParsedTableProperties::default(); +//! properties.write_format_default = DataFileFormat::Orc; +//! properties.write_data_path = Some("s3://warehouse/table/data".to_string()); +//! +//! assert_eq!(properties.write_format_default, DataFileFormat::Orc); +//! ``` +//! +//! # Deserialize from JSON +//! +//! JSON property values must be strings, matching Iceberg's table property map: +//! +//! ``` +//! use iceberg::spec::{DataFileFormat, ParsedTableProperties}; +//! +//! # fn main() -> Result<(), serde_json::Error> { +//! let properties: ParsedTableProperties = serde_json::from_value(serde_json::json!({ +//! "commit.retry.num-retries": "8", +//! "write.format.default": "orc" +//! }))?; +//! +//! assert_eq!(properties.commit_retry_num_retries, 8); +//! assert_eq!(properties.write_format_default, DataFileFormat::Orc); +//! # Ok(()) +//! # } +//! ``` +//! +//! # Serialize to JSON +//! +//! Serialization converts the typed fields back into Iceberg property keys: +//! +//! ``` +//! use iceberg::spec::ParsedTableProperties; +//! +//! # fn main() -> Result<(), serde_json::Error> { +//! let mut properties = ParsedTableProperties::default(); +//! properties.commit_retry_num_retries = 8; +//! properties.write_data_path = Some("s3://warehouse/table/data".to_string()); +//! +//! let json = serde_json::to_value(&properties)?; +//! assert_eq!(json["commit.retry.num-retries"], "8"); +//! assert_eq!(json["write.data.path"], "s3://warehouse/table/data"); +//! # Ok(()) +//! # } +//! ``` + +use std::collections::HashMap; + +use iceberg_property_macro::Properties; +use serde_with::{DeserializeFromStr, SerializeDisplay}; + +use crate::compression::CompressionCodec; +use crate::error::{Error, ErrorKind, Result}; +use crate::spec::{DataFileFormat, NameMapping}; + +/// Parquet data page version 1. +pub const PARQUET_PAGE_VERSION_V1: &str = "v1"; + +/// Parquet data page version 2. +pub const PARQUET_PAGE_VERSION_V2: &str = "v2"; + +/// Distribution applied to rows before writing files. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum DistributionMode { + /// Do not redistribute rows. + None, + /// Hash-distribute rows by partition values. + Hash, + /// Range-distribute rows by partition or sort values. + Range, +} + +/// Granularity used when creating position delete files. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum DeleteGranularity { + /// Group deletes for each referenced data file separately. + File, + /// Group deletes for different data files within a partition. + Partition, +} + +/// Isolation level used by row-level operations. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum IsolationLevel { + /// Fail if concurrent changes may contain rows matching the operation. + Serializable, + /// Validate only against data visible in the operation's snapshot. + Snapshot, +} + +/// Strategy used to apply row-level changes. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum RowLevelOperationMode { + /// Replace affected data files immediately. + CopyOnWrite, + /// Write delete files and merge changes while reading. + MergeOnRead, +} + +/// Strips trailing slashes from a location, preserving a bare URI scheme root. +fn strip_trailing_slash(path: &str) -> &str { + let mut path = path; + while !path.ends_with("://") { + let Some(stripped) = path.strip_suffix('/') else { + break; + }; + path = stripped; + } + path +} + +fn parse_metadata_location(value: &str) -> Result<Option<String>> { + if value.is_empty() { + return Err(Error::new(ErrorKind::DataInvalid, "path must not be empty")); + } + + Ok(Some(strip_trailing_slash(value).to_string())) +} + +fn parse_compression_codec(value: &str) -> Result<CompressionCodec> { + serde_json::from_value(serde_json::Value::String(value.to_lowercase())).map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid compression codec: {value}"), + ) + }) +} + +fn parse_metadata_file_compression(value: &str) -> Result<CompressionCodec> { + if value.is_empty() { + return Ok(CompressionCodec::None); + } + + let codec: CompressionCodec = serde_json::from_value(serde_json::Value::String( + value.to_lowercase(), + )) + .map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported.", + CompressionCodec::None.name(), + CompressionCodec::gzip_default().name() + ), + ) + })?; + + match codec { + CompressionCodec::None | CompressionCodec::Gzip(_) => Ok(codec), + _ => Err(Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported for metadata files.", + CompressionCodec::None.name(), + CompressionCodec::gzip_default().name() + ), + )), + } +} + +fn serialize_compression_codec(codec: &CompressionCodec) -> String { + codec.name().to_string() +} + +fn parse_comma_separated_strings(value: &str) -> Result<Vec<String>> { + Ok(value + .split(',') + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_string) + .collect()) +} + +fn serialize_comma_separated_strings(values: &[String]) -> String { + values.join(",") +} + +fn parse_name_mapping(value: &str) -> Result<Option<NameMapping>> { + serde_json::from_str(value).map(Some).map_err(|error| { + Error::new(ErrorKind::DataInvalid, "Invalid name mapping").with_source(error) + }) +} + +fn serialize_name_mapping(mapping: &Option<NameMapping>) -> String { + serde_json::to_string(mapping.as_ref().expect("checked is_some above")) + .expect("serializing a NameMapping cannot fail") +} + +/// Typed Iceberg table properties organized into documented sections. +/// +/// Serde represents this struct as Iceberg's flat string-to-string property map. Property +/// definitions and descriptions are based on the pinned [Java TableProperties implementation] +/// and [Apache Iceberg configuration documentation]. +/// +/// [Java TableProperties implementation]: https://github.com/apache/iceberg/blob/d8c10a1608170f0ba83be740d6ab0b6a3757cb3e/core/src/main/java/org/apache/iceberg/TableProperties.java +/// [Apache Iceberg configuration documentation]: https://github.com/apache/iceberg/blob/d8c10a1608170f0ba83be740d6ab0b6a3757cb3e/docs/docs/configuration.md +#[derive(Clone, Debug, Properties)] +pub struct ParsedTableProperties { + // General properties. + #[key = "comment"] + #[default(None)] + #[doc = "Table-level description of the table's business meaning and usage context."] + pub comment: Option<String>, + + #[key = "identifier-fields.rely"] + #[default(false)] + #[doc = "Whether query engines may rely on identifier fields as a primary key for optimization; this is not enforced on writes."] + pub identifier_fields_rely: bool, + + // Commit properties. + #[key = "commit.retry.num-retries"] + #[default(4)] + #[doc = "Number of times to retry a commit before failing."] + pub commit_retry_num_retries: usize, + + #[key = "commit.retry.min-wait-ms"] + #[default(100)] + #[doc = "Minimum time in milliseconds to wait before retrying a commit."] + pub commit_retry_min_wait_ms: u64, + + #[key = "commit.retry.max-wait-ms"] + #[default(60 * 1000)] + #[doc = "Maximum time in milliseconds to wait before retrying a commit."] + pub commit_retry_max_wait_ms: u64, + + #[key = "commit.retry.total-timeout-ms"] + #[default(30 * 60 * 1000)] + #[doc = "Total commit retry timeout in milliseconds."] + pub commit_retry_total_timeout_ms: u64, + + #[key = "commit.status-check.num-retries"] + #[default(3)] + #[doc = "Number of times to check whether a commit succeeded after connectivity is lost."] + pub commit_status_check_num_retries: usize, + + #[key = "commit.status-check.min-wait-ms"] + #[default(1000)] + #[doc = "Minimum time in milliseconds to wait before retrying a commit status check."] + pub commit_status_check_min_wait_ms: u64, + + #[key = "commit.status-check.max-wait-ms"] + #[default(60 * 1000)] + #[doc = "Maximum time in milliseconds to wait before retrying a commit status check."] + pub commit_status_check_max_wait_ms: u64, + + #[key = "commit.status-check.total-timeout-ms"] + #[default(30 * 60 * 1000)] + #[doc = "Total timeout in milliseconds in which commit status checking must succeed."] + pub commit_status_check_total_timeout_ms: u64, + + // Manifest properties. + #[key = "commit.manifest.target-size-bytes"] + #[default(8 * 1024 * 1024)] + #[doc = "Target size in bytes when merging manifest files."] + pub commit_manifest_target_size_bytes: usize, + + #[key = "commit.manifest.min-count-to-merge"] + #[default(100)] + #[doc = "Minimum number of manifests to accumulate before merging."] + pub commit_manifest_min_count_to_merge: usize, + + #[key = "commit.manifest-merge.enabled"] + #[default(true)] + #[doc = "Whether manifests are automatically merged during writes."] + pub commit_manifest_merge_enabled: bool, + + #[key = "write.manifest.compression-codec"] + #[default(CompressionCodec::gzip_default())] + #[parse_with(parse_compression_codec)] Review Comment: This is incorrect, we should also consider compression level ########## crates/iceberg/src/spec/parsed_table_prop.rs: ########## @@ -0,0 +1,1181 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Typed access to Iceberg table properties. +//! +//! [`ParsedTableProperties`] exposes Iceberg's string-keyed table properties as typed public +//! fields. Its JSON representation is a flat object whose keys and values are strings. +//! +//! # Create from defaults +//! +//! Start with Iceberg's defaults and modify public fields directly: +//! +//! ``` +//! use iceberg::spec::{DataFileFormat, ParsedTableProperties}; +//! +//! let mut properties = ParsedTableProperties::default(); +//! properties.write_format_default = DataFileFormat::Orc; +//! properties.write_data_path = Some("s3://warehouse/table/data".to_string()); +//! +//! assert_eq!(properties.write_format_default, DataFileFormat::Orc); +//! ``` +//! +//! # Deserialize from JSON +//! +//! JSON property values must be strings, matching Iceberg's table property map: +//! +//! ``` +//! use iceberg::spec::{DataFileFormat, ParsedTableProperties}; +//! +//! # fn main() -> Result<(), serde_json::Error> { +//! let properties: ParsedTableProperties = serde_json::from_value(serde_json::json!({ +//! "commit.retry.num-retries": "8", +//! "write.format.default": "orc" +//! }))?; +//! +//! assert_eq!(properties.commit_retry_num_retries, 8); +//! assert_eq!(properties.write_format_default, DataFileFormat::Orc); +//! # Ok(()) +//! # } +//! ``` +//! +//! # Serialize to JSON +//! +//! Serialization converts the typed fields back into Iceberg property keys: +//! +//! ``` +//! use iceberg::spec::ParsedTableProperties; +//! +//! # fn main() -> Result<(), serde_json::Error> { +//! let mut properties = ParsedTableProperties::default(); +//! properties.commit_retry_num_retries = 8; +//! properties.write_data_path = Some("s3://warehouse/table/data".to_string()); +//! +//! let json = serde_json::to_value(&properties)?; +//! assert_eq!(json["commit.retry.num-retries"], "8"); +//! assert_eq!(json["write.data.path"], "s3://warehouse/table/data"); +//! # Ok(()) +//! # } +//! ``` + +use std::collections::HashMap; + +use iceberg_property_macro::Properties; +use serde_with::{DeserializeFromStr, SerializeDisplay}; + +use crate::compression::CompressionCodec; +use crate::error::{Error, ErrorKind, Result}; +use crate::spec::{DataFileFormat, NameMapping}; + +/// Parquet data page version 1. +pub const PARQUET_PAGE_VERSION_V1: &str = "v1"; + +/// Parquet data page version 2. +pub const PARQUET_PAGE_VERSION_V2: &str = "v2"; + +/// Distribution applied to rows before writing files. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum DistributionMode { + /// Do not redistribute rows. + None, + /// Hash-distribute rows by partition values. + Hash, + /// Range-distribute rows by partition or sort values. + Range, +} + +/// Granularity used when creating position delete files. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum DeleteGranularity { + /// Group deletes for each referenced data file separately. + File, + /// Group deletes for different data files within a partition. + Partition, +} + +/// Isolation level used by row-level operations. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum IsolationLevel { + /// Fail if concurrent changes may contain rows matching the operation. + Serializable, + /// Validate only against data visible in the operation's snapshot. + Snapshot, +} + +/// Strategy used to apply row-level changes. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum RowLevelOperationMode { + /// Replace affected data files immediately. + CopyOnWrite, + /// Write delete files and merge changes while reading. + MergeOnRead, +} + +/// Strips trailing slashes from a location, preserving a bare URI scheme root. +fn strip_trailing_slash(path: &str) -> &str { + let mut path = path; + while !path.ends_with("://") { + let Some(stripped) = path.strip_suffix('/') else { + break; + }; + path = stripped; + } + path +} + +fn parse_metadata_location(value: &str) -> Result<Option<String>> { + if value.is_empty() { + return Err(Error::new(ErrorKind::DataInvalid, "path must not be empty")); + } + + Ok(Some(strip_trailing_slash(value).to_string())) +} + +fn parse_compression_codec(value: &str) -> Result<CompressionCodec> { + serde_json::from_value(serde_json::Value::String(value.to_lowercase())).map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid compression codec: {value}"), + ) + }) +} + +fn parse_metadata_file_compression(value: &str) -> Result<CompressionCodec> { + if value.is_empty() { + return Ok(CompressionCodec::None); + } + + let codec: CompressionCodec = serde_json::from_value(serde_json::Value::String( + value.to_lowercase(), + )) + .map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported.", + CompressionCodec::None.name(), + CompressionCodec::gzip_default().name() + ), + ) + })?; + + match codec { + CompressionCodec::None | CompressionCodec::Gzip(_) => Ok(codec), + _ => Err(Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported for metadata files.", + CompressionCodec::None.name(), + CompressionCodec::gzip_default().name() + ), + )), + } +} + +fn serialize_compression_codec(codec: &CompressionCodec) -> String { + codec.name().to_string() +} + +fn parse_comma_separated_strings(value: &str) -> Result<Vec<String>> { + Ok(value + .split(',') + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_string) + .collect()) +} + +fn serialize_comma_separated_strings(values: &[String]) -> String { + values.join(",") +} + +fn parse_name_mapping(value: &str) -> Result<Option<NameMapping>> { + serde_json::from_str(value).map(Some).map_err(|error| { + Error::new(ErrorKind::DataInvalid, "Invalid name mapping").with_source(error) + }) +} + +fn serialize_name_mapping(mapping: &Option<NameMapping>) -> String { + serde_json::to_string(mapping.as_ref().expect("checked is_some above")) + .expect("serializing a NameMapping cannot fail") +} + +/// Typed Iceberg table properties organized into documented sections. +/// +/// Serde represents this struct as Iceberg's flat string-to-string property map. Property +/// definitions and descriptions are based on the pinned [Java TableProperties implementation] +/// and [Apache Iceberg configuration documentation]. +/// +/// [Java TableProperties implementation]: https://github.com/apache/iceberg/blob/d8c10a1608170f0ba83be740d6ab0b6a3757cb3e/core/src/main/java/org/apache/iceberg/TableProperties.java +/// [Apache Iceberg configuration documentation]: https://github.com/apache/iceberg/blob/d8c10a1608170f0ba83be740d6ab0b6a3757cb3e/docs/docs/configuration.md +#[derive(Clone, Debug, Properties)] +pub struct ParsedTableProperties { + // General properties. + #[key = "comment"] + #[default(None)] + #[doc = "Table-level description of the table's business meaning and usage context."] + pub comment: Option<String>, + + #[key = "identifier-fields.rely"] + #[default(false)] + #[doc = "Whether query engines may rely on identifier fields as a primary key for optimization; this is not enforced on writes."] + pub identifier_fields_rely: bool, + + // Commit properties. + #[key = "commit.retry.num-retries"] + #[default(4)] + #[doc = "Number of times to retry a commit before failing."] + pub commit_retry_num_retries: usize, + + #[key = "commit.retry.min-wait-ms"] + #[default(100)] + #[doc = "Minimum time in milliseconds to wait before retrying a commit."] + pub commit_retry_min_wait_ms: u64, + + #[key = "commit.retry.max-wait-ms"] + #[default(60 * 1000)] + #[doc = "Maximum time in milliseconds to wait before retrying a commit."] + pub commit_retry_max_wait_ms: u64, + + #[key = "commit.retry.total-timeout-ms"] + #[default(30 * 60 * 1000)] + #[doc = "Total commit retry timeout in milliseconds."] + pub commit_retry_total_timeout_ms: u64, + + #[key = "commit.status-check.num-retries"] + #[default(3)] + #[doc = "Number of times to check whether a commit succeeded after connectivity is lost."] + pub commit_status_check_num_retries: usize, + + #[key = "commit.status-check.min-wait-ms"] + #[default(1000)] + #[doc = "Minimum time in milliseconds to wait before retrying a commit status check."] + pub commit_status_check_min_wait_ms: u64, + + #[key = "commit.status-check.max-wait-ms"] + #[default(60 * 1000)] + #[doc = "Maximum time in milliseconds to wait before retrying a commit status check."] + pub commit_status_check_max_wait_ms: u64, + + #[key = "commit.status-check.total-timeout-ms"] + #[default(30 * 60 * 1000)] + #[doc = "Total timeout in milliseconds in which commit status checking must succeed."] + pub commit_status_check_total_timeout_ms: u64, + + // Manifest properties. + #[key = "commit.manifest.target-size-bytes"] + #[default(8 * 1024 * 1024)] + #[doc = "Target size in bytes when merging manifest files."] + pub commit_manifest_target_size_bytes: usize, + + #[key = "commit.manifest.min-count-to-merge"] + #[default(100)] + #[doc = "Minimum number of manifests to accumulate before merging."] + pub commit_manifest_min_count_to_merge: usize, + + #[key = "commit.manifest-merge.enabled"] + #[default(true)] + #[doc = "Whether manifests are automatically merged during writes."] + pub commit_manifest_merge_enabled: bool, + + #[key = "write.manifest.compression-codec"] + #[default(CompressionCodec::gzip_default())] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "Compression codec used for manifest files."] + pub write_manifest_compression_codec: CompressionCodec, + + #[key = "write.manifest-lists.enabled"] + #[default(true)] + #[doc = "Deprecated flag for writing manifest lists; manifest lists are always enabled."] + pub write_manifest_lists_enabled: bool, + + // Write properties. + #[key = "write.format.default"] + #[default(DataFileFormat::Parquet)] + #[doc = "Default data file format: Parquet, Avro, or ORC."] + pub write_format_default: DataFileFormat, + + #[key = "write.delete.format.default"] + #[default(DataFileFormat::Parquet)] + #[doc = "Default delete file format: Parquet, Avro, or ORC."] + pub write_delete_format_default: DataFileFormat, + + #[key = "write.target-file-size-bytes"] + #[default(512 * 1024 * 1024)] + #[doc = "Target size in bytes for generated data files."] + pub write_target_file_size_bytes: usize, + + #[key = "write.delete.target-file-size-bytes"] + #[default(64 * 1024 * 1024)] + #[doc = "Target size in bytes for generated delete files."] + pub write_delete_target_file_size_bytes: usize, + + #[key = "write.object-storage.enabled"] + #[default(false)] + #[doc = "Whether the object-storage location provider adds a hash component to file paths."] + pub write_object_storage_enabled: bool, + + #[key = "write.object-storage.partitioned-paths"] + #[default(true)] + #[doc = "Whether object-storage file paths include partition values."] + pub write_object_storage_partitioned_paths: bool, + + #[key = "write.object-storage.path"] + #[default(None)] + #[doc = "Deprecated base object-storage path; use write.data.path instead."] + pub write_object_storage_path: Option<String>, + + #[key = "write.location-provider.impl"] + #[default(None)] + #[doc = "Optional custom location provider implementation."] + pub write_location_provider_impl: Option<String>, + + #[key = "write.folder-storage.path"] + #[default(None)] + #[doc = "Deprecated base folder-storage path; use write.data.path instead."] + pub write_folder_storage_path: Option<String>, + + #[key = "write.data.path"] + #[default(None)] + #[doc = "Base location for data files written after this property is set."] + pub write_data_path: Option<String>, + + #[key = "write.wap.enabled"] + #[default(false)] + #[doc = "Whether write-audit-publish writes are enabled."] + pub write_wap_enabled: bool, + + #[key = "write.distribution-mode"] + #[default(None)] + #[doc = "Optional write distribution mode: none, hash, or range."] + pub write_distribution_mode: Option<DistributionMode>, + + #[key = "write.datafusion.fanout.enabled"] + #[default(true)] + #[doc = "Whether DataFusion uses a fanout writer for partitioned tables."] + pub write_datafusion_fanout_enabled: bool, + + // Parquet properties. + #[key = "write.parquet.row-group-size-bytes"] + #[default(128 * 1024 * 1024)] + #[doc = "Parquet row group size in bytes for data files."] + pub write_parquet_row_group_size_bytes: usize, + + #[key = "write.delete.parquet.row-group-size-bytes"] + #[default(128 * 1024 * 1024)] + #[doc = "Parquet row group size in bytes for delete files."] + pub write_delete_parquet_row_group_size_bytes: usize, + + #[key = "write.parquet.page-size-bytes"] + #[default(1024 * 1024)] + #[doc = "Parquet page size in bytes for data files."] + pub write_parquet_page_size_bytes: usize, + + #[key = "write.delete.parquet.page-size-bytes"] + #[default(1024 * 1024)] + #[doc = "Parquet page size in bytes for delete files."] + pub write_delete_parquet_page_size_bytes: usize, + + #[key = "write.parquet.page-version"] + #[default(PARQUET_PAGE_VERSION_V1)] + #[doc = "Parquet data page version for data files: v1 or v2."] + pub write_parquet_page_version: String, + + #[key = "write.delete.parquet.page-version"] + #[default(PARQUET_PAGE_VERSION_V1)] + #[doc = "Parquet data page version for delete files: v1 or v2."] + pub write_delete_parquet_page_version: String, + + #[key = "write.parquet.page-row-limit"] + #[default(20_000)] + #[doc = "Maximum number of rows per Parquet page in data files."] + pub write_parquet_page_row_limit: usize, + + #[key = "write.delete.parquet.page-row-limit"] + #[default(20_000)] + #[doc = "Maximum number of rows per Parquet page in delete files."] + pub write_delete_parquet_page_row_limit: usize, + + #[key = "write.parquet.dict-size-bytes"] + #[default(2 * 1024 * 1024)] + #[doc = "Parquet dictionary page size in bytes for data files."] + pub write_parquet_dict_size_bytes: usize, + + #[key = "write.delete.parquet.dict-size-bytes"] + #[default(2 * 1024 * 1024)] + #[doc = "Parquet dictionary page size in bytes for delete files."] + pub write_delete_parquet_dict_size_bytes: usize, + + #[key = "write.parquet.compression-codec"] + #[default(CompressionCodec::zstd_default())] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "Parquet compression codec used for data files."] + pub write_parquet_compression_codec: CompressionCodec, + + #[key = "write.delete.parquet.compression-codec"] + #[default(CompressionCodec::zstd_default())] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "Parquet compression codec used for delete files."] + pub write_delete_parquet_compression_codec: CompressionCodec, + + #[key = "write.parquet.shred-variants"] + #[default(false)] + #[doc = "Whether variant columns use shredded Parquet encoding for improved query performance."] + pub write_parquet_shred_variants: bool, + + #[key = "write.parquet.variant-inference-buffer-size"] + #[default(100)] + #[doc = "Number of rows buffered for schema inference when variant shredding is enabled."] + pub write_parquet_variant_inference_buffer_size: usize, + + #[key = "write.parquet.row-group-check-min-record-count"] + #[default(100)] + #[doc = "Minimum record count between Parquet data-file row group size checks."] + pub write_parquet_row_group_check_min_record_count: usize, + + #[key = "write.delete.parquet.row-group-check-min-record-count"] + #[default(100)] + #[doc = "Minimum record count between Parquet delete-file row group size checks."] + pub write_delete_parquet_row_group_check_min_record_count: usize, + + #[key = "write.parquet.row-group-check-max-record-count"] + #[default(10_000)] + #[doc = "Maximum record count between Parquet data-file row group size checks."] + pub write_parquet_row_group_check_max_record_count: usize, + + #[key = "write.delete.parquet.row-group-check-max-record-count"] + #[default(10_000)] + #[doc = "Maximum record count between Parquet delete-file row group size checks."] + pub write_delete_parquet_row_group_check_max_record_count: usize, + + #[key = "write.parquet.row-group-size-track-uncompressed"] + #[default(false)] + #[doc = "Whether uncompressed data size is tracked to enforce the Parquet row group target."] + pub write_parquet_row_group_size_track_uncompressed: bool, + + #[key = "write.parquet.bloom-filter-max-bytes"] + #[default(1024 * 1024)] + #[doc = "Maximum number of bytes for a Parquet bloom filter bitset."] + pub write_parquet_bloom_filter_max_bytes: usize, + + #[key = "write.parquet.bloom-filter-adaptive-enabled"] + #[default(false)] + #[doc = "Whether adaptive Parquet bloom filter sizing selects the smallest suitable filter."] + pub write_parquet_bloom_filter_adaptive_enabled: bool, + + #[prefix = "write.parquet.bloom-filter-fpp.column."] + #[default(HashMap::new())] + #[doc = "Per-column Parquet bloom filter false-positive probabilities, keyed by column name."] + pub write_parquet_bloom_filter_fpp_column: HashMap<String, f64>, + + #[prefix = "write.parquet.bloom-filter-ndv.column."] + #[default(HashMap::new())] + #[doc = "Per-column expected distinct-value counts for Parquet bloom filters."] + pub write_parquet_bloom_filter_ndv_column: HashMap<String, u64>, + + #[prefix = "write.parquet.bloom-filter-enabled.column."] + #[default(HashMap::new())] + #[doc = "Per-column flags controlling whether Parquet bloom filters are written."] + pub write_parquet_bloom_filter_enabled_column: HashMap<String, bool>, + + #[prefix = "write.parquet.stats-enabled.column."] + #[default(HashMap::new())] + #[doc = "Per-column flags controlling whether Parquet column statistics are collected."] + pub write_parquet_stats_enabled_column: HashMap<String, bool>, + + #[prefix = "write.parquet.dict-encoding-enabled.column."] + #[default(HashMap::new())] + #[doc = "Per-column flags controlling whether Parquet dictionary encoding is used."] + pub write_parquet_dict_encoding_enabled_column: HashMap<String, bool>, + + #[key = "write.parquet.content-defined-chunking.enabled"] + #[default(false)] + #[doc = "Whether Parquet content-defined chunking is enabled."] + pub write_parquet_content_defined_chunking_enabled: bool, + + #[key = "write.parquet.content-defined-chunking.min-chunk-size"] + #[default(256 * 1024)] + #[doc = "Minimum Parquet content-defined chunk size in bytes."] + pub write_parquet_content_defined_chunking_min_chunk_size: usize, + + #[key = "write.parquet.content-defined-chunking.max-chunk-size"] + #[default(1024 * 1024)] + #[doc = "Maximum Parquet content-defined chunk size in bytes."] + pub write_parquet_content_defined_chunking_max_chunk_size: usize, + + #[key = "write.parquet.content-defined-chunking.norm-level"] + #[default(0)] + #[doc = "Gearhash normalization level used by Parquet content-defined chunking."] + pub write_parquet_content_defined_chunking_norm_level: i32, + + // Avro properties. + #[key = "write.avro.compression-codec"] + #[default(CompressionCodec::gzip_default())] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "Avro compression codec used for data files."] + pub write_avro_compression_codec: CompressionCodec, + + #[key = "write.delete.avro.compression-codec"] + #[default(CompressionCodec::gzip_default())] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "Avro compression codec used for delete files."] + pub write_delete_avro_compression_codec: CompressionCodec, + + // ORC properties. + #[key = "write.orc.stripe-size-bytes"] + #[default(64 * 1024 * 1024)] + #[doc = "Default ORC stripe size in bytes for data files."] + pub write_orc_stripe_size_bytes: u64, + + #[key = "write.delete.orc.stripe-size-bytes"] + #[default(64 * 1024 * 1024)] + #[doc = "Default ORC stripe size in bytes for delete files."] + pub write_delete_orc_stripe_size_bytes: u64, + + #[key = "write.orc.bloom.filter.columns"] + #[default(Vec::new())] + #[parse_with(parse_comma_separated_strings)] + #[serialize_with(serialize_comma_separated_strings)] + #[doc = "Comma-separated column names for which ORC bloom filters are created."] + pub write_orc_bloom_filter_columns: Vec<String>, + + #[key = "write.orc.bloom.filter.fpp"] + #[default(0.05)] + #[doc = "False-positive probability for ORC bloom filters."] + pub write_orc_bloom_filter_fpp: f64, + + #[key = "write.orc.block-size-bytes"] + #[default(256 * 1024 * 1024)] + #[doc = "Default file-system block size in bytes for ORC data files."] + pub write_orc_block_size_bytes: u64, + + #[key = "write.delete.orc.block-size-bytes"] + #[default(256 * 1024 * 1024)] + #[doc = "Default file-system block size in bytes for ORC delete files."] + pub write_delete_orc_block_size_bytes: u64, + + #[key = "write.orc.vectorized.batch-size"] + #[default(1024)] + #[doc = "ORC vectorized write batch size for data files."] + pub write_orc_vectorized_batch_size: usize, + + #[key = "write.delete.orc.vectorized.batch-size"] + #[default(1024)] + #[doc = "ORC vectorized write batch size for delete files."] + pub write_delete_orc_vectorized_batch_size: usize, + + #[key = "write.orc.compression-codec"] + #[default(CompressionCodec::Zlib)] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "ORC compression codec used for data files."] + pub write_orc_compression_codec: CompressionCodec, + + #[key = "write.delete.orc.compression-codec"] + #[default(CompressionCodec::Zlib)] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "ORC compression codec used for delete files."] + pub write_delete_orc_compression_codec: CompressionCodec, + + #[key = "write.orc.compression-strategy"] + #[default("speed")] + #[doc = "ORC compression strategy for data files: speed or compression."] + pub write_orc_compression_strategy: String, + + #[key = "write.delete.orc.compression-strategy"] + #[default("speed")] + #[doc = "ORC compression strategy for delete files: speed or compression."] + pub write_delete_orc_compression_strategy: String, Review Comment: Same as https://github.com/apache/iceberg-rust/pull/2955/changes#r3718436541 ########## crates/iceberg/src/spec/parsed_table_prop.rs: ########## @@ -0,0 +1,1181 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Typed access to Iceberg table properties. +//! +//! [`ParsedTableProperties`] exposes Iceberg's string-keyed table properties as typed public +//! fields. Its JSON representation is a flat object whose keys and values are strings. +//! +//! # Create from defaults +//! +//! Start with Iceberg's defaults and modify public fields directly: +//! +//! ``` +//! use iceberg::spec::{DataFileFormat, ParsedTableProperties}; +//! +//! let mut properties = ParsedTableProperties::default(); +//! properties.write_format_default = DataFileFormat::Orc; +//! properties.write_data_path = Some("s3://warehouse/table/data".to_string()); +//! +//! assert_eq!(properties.write_format_default, DataFileFormat::Orc); +//! ``` +//! +//! # Deserialize from JSON +//! +//! JSON property values must be strings, matching Iceberg's table property map: +//! +//! ``` +//! use iceberg::spec::{DataFileFormat, ParsedTableProperties}; +//! +//! # fn main() -> Result<(), serde_json::Error> { +//! let properties: ParsedTableProperties = serde_json::from_value(serde_json::json!({ +//! "commit.retry.num-retries": "8", +//! "write.format.default": "orc" +//! }))?; +//! +//! assert_eq!(properties.commit_retry_num_retries, 8); +//! assert_eq!(properties.write_format_default, DataFileFormat::Orc); +//! # Ok(()) +//! # } +//! ``` +//! +//! # Serialize to JSON +//! +//! Serialization converts the typed fields back into Iceberg property keys: +//! +//! ``` +//! use iceberg::spec::ParsedTableProperties; +//! +//! # fn main() -> Result<(), serde_json::Error> { +//! let mut properties = ParsedTableProperties::default(); +//! properties.commit_retry_num_retries = 8; +//! properties.write_data_path = Some("s3://warehouse/table/data".to_string()); +//! +//! let json = serde_json::to_value(&properties)?; +//! assert_eq!(json["commit.retry.num-retries"], "8"); +//! assert_eq!(json["write.data.path"], "s3://warehouse/table/data"); +//! # Ok(()) +//! # } +//! ``` + +use std::collections::HashMap; + +use iceberg_property_macro::Properties; +use serde_with::{DeserializeFromStr, SerializeDisplay}; + +use crate::compression::CompressionCodec; +use crate::error::{Error, ErrorKind, Result}; +use crate::spec::{DataFileFormat, NameMapping}; + +/// Parquet data page version 1. +pub const PARQUET_PAGE_VERSION_V1: &str = "v1"; + +/// Parquet data page version 2. +pub const PARQUET_PAGE_VERSION_V2: &str = "v2"; + +/// Distribution applied to rows before writing files. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum DistributionMode { + /// Do not redistribute rows. + None, + /// Hash-distribute rows by partition values. + Hash, + /// Range-distribute rows by partition or sort values. + Range, +} + +/// Granularity used when creating position delete files. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum DeleteGranularity { + /// Group deletes for each referenced data file separately. + File, + /// Group deletes for different data files within a partition. + Partition, +} + +/// Isolation level used by row-level operations. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum IsolationLevel { + /// Fail if concurrent changes may contain rows matching the operation. + Serializable, + /// Validate only against data visible in the operation's snapshot. + Snapshot, +} + +/// Strategy used to apply row-level changes. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum RowLevelOperationMode { + /// Replace affected data files immediately. + CopyOnWrite, + /// Write delete files and merge changes while reading. + MergeOnRead, +} + +/// Strips trailing slashes from a location, preserving a bare URI scheme root. +fn strip_trailing_slash(path: &str) -> &str { + let mut path = path; + while !path.ends_with("://") { + let Some(stripped) = path.strip_suffix('/') else { + break; + }; + path = stripped; + } + path +} + +fn parse_metadata_location(value: &str) -> Result<Option<String>> { + if value.is_empty() { + return Err(Error::new(ErrorKind::DataInvalid, "path must not be empty")); + } + + Ok(Some(strip_trailing_slash(value).to_string())) +} + +fn parse_compression_codec(value: &str) -> Result<CompressionCodec> { + serde_json::from_value(serde_json::Value::String(value.to_lowercase())).map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid compression codec: {value}"), + ) + }) +} + +fn parse_metadata_file_compression(value: &str) -> Result<CompressionCodec> { + if value.is_empty() { + return Ok(CompressionCodec::None); + } + + let codec: CompressionCodec = serde_json::from_value(serde_json::Value::String( + value.to_lowercase(), + )) + .map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported.", + CompressionCodec::None.name(), + CompressionCodec::gzip_default().name() + ), + ) + })?; + + match codec { + CompressionCodec::None | CompressionCodec::Gzip(_) => Ok(codec), + _ => Err(Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported for metadata files.", + CompressionCodec::None.name(), + CompressionCodec::gzip_default().name() + ), + )), + } +} + +fn serialize_compression_codec(codec: &CompressionCodec) -> String { + codec.name().to_string() +} + +fn parse_comma_separated_strings(value: &str) -> Result<Vec<String>> { + Ok(value + .split(',') + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_string) + .collect()) +} + +fn serialize_comma_separated_strings(values: &[String]) -> String { + values.join(",") +} + +fn parse_name_mapping(value: &str) -> Result<Option<NameMapping>> { + serde_json::from_str(value).map(Some).map_err(|error| { + Error::new(ErrorKind::DataInvalid, "Invalid name mapping").with_source(error) + }) +} + +fn serialize_name_mapping(mapping: &Option<NameMapping>) -> String { + serde_json::to_string(mapping.as_ref().expect("checked is_some above")) + .expect("serializing a NameMapping cannot fail") +} + +/// Typed Iceberg table properties organized into documented sections. +/// +/// Serde represents this struct as Iceberg's flat string-to-string property map. Property +/// definitions and descriptions are based on the pinned [Java TableProperties implementation] +/// and [Apache Iceberg configuration documentation]. +/// +/// [Java TableProperties implementation]: https://github.com/apache/iceberg/blob/d8c10a1608170f0ba83be740d6ab0b6a3757cb3e/core/src/main/java/org/apache/iceberg/TableProperties.java +/// [Apache Iceberg configuration documentation]: https://github.com/apache/iceberg/blob/d8c10a1608170f0ba83be740d6ab0b6a3757cb3e/docs/docs/configuration.md +#[derive(Clone, Debug, Properties)] +pub struct ParsedTableProperties { + // General properties. + #[key = "comment"] + #[default(None)] + #[doc = "Table-level description of the table's business meaning and usage context."] + pub comment: Option<String>, + + #[key = "identifier-fields.rely"] + #[default(false)] + #[doc = "Whether query engines may rely on identifier fields as a primary key for optimization; this is not enforced on writes."] + pub identifier_fields_rely: bool, + + // Commit properties. + #[key = "commit.retry.num-retries"] + #[default(4)] + #[doc = "Number of times to retry a commit before failing."] + pub commit_retry_num_retries: usize, + + #[key = "commit.retry.min-wait-ms"] + #[default(100)] + #[doc = "Minimum time in milliseconds to wait before retrying a commit."] + pub commit_retry_min_wait_ms: u64, + + #[key = "commit.retry.max-wait-ms"] + #[default(60 * 1000)] + #[doc = "Maximum time in milliseconds to wait before retrying a commit."] + pub commit_retry_max_wait_ms: u64, + + #[key = "commit.retry.total-timeout-ms"] + #[default(30 * 60 * 1000)] + #[doc = "Total commit retry timeout in milliseconds."] + pub commit_retry_total_timeout_ms: u64, + + #[key = "commit.status-check.num-retries"] + #[default(3)] + #[doc = "Number of times to check whether a commit succeeded after connectivity is lost."] + pub commit_status_check_num_retries: usize, + + #[key = "commit.status-check.min-wait-ms"] + #[default(1000)] + #[doc = "Minimum time in milliseconds to wait before retrying a commit status check."] + pub commit_status_check_min_wait_ms: u64, + + #[key = "commit.status-check.max-wait-ms"] + #[default(60 * 1000)] + #[doc = "Maximum time in milliseconds to wait before retrying a commit status check."] + pub commit_status_check_max_wait_ms: u64, + + #[key = "commit.status-check.total-timeout-ms"] + #[default(30 * 60 * 1000)] + #[doc = "Total timeout in milliseconds in which commit status checking must succeed."] + pub commit_status_check_total_timeout_ms: u64, + + // Manifest properties. + #[key = "commit.manifest.target-size-bytes"] + #[default(8 * 1024 * 1024)] + #[doc = "Target size in bytes when merging manifest files."] + pub commit_manifest_target_size_bytes: usize, + + #[key = "commit.manifest.min-count-to-merge"] + #[default(100)] + #[doc = "Minimum number of manifests to accumulate before merging."] + pub commit_manifest_min_count_to_merge: usize, + + #[key = "commit.manifest-merge.enabled"] + #[default(true)] + #[doc = "Whether manifests are automatically merged during writes."] + pub commit_manifest_merge_enabled: bool, + + #[key = "write.manifest.compression-codec"] + #[default(CompressionCodec::gzip_default())] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "Compression codec used for manifest files."] + pub write_manifest_compression_codec: CompressionCodec, + + #[key = "write.manifest-lists.enabled"] + #[default(true)] + #[doc = "Deprecated flag for writing manifest lists; manifest lists are always enabled."] + pub write_manifest_lists_enabled: bool, + + // Write properties. + #[key = "write.format.default"] + #[default(DataFileFormat::Parquet)] + #[doc = "Default data file format: Parquet, Avro, or ORC."] + pub write_format_default: DataFileFormat, + + #[key = "write.delete.format.default"] + #[default(DataFileFormat::Parquet)] + #[doc = "Default delete file format: Parquet, Avro, or ORC."] + pub write_delete_format_default: DataFileFormat, + + #[key = "write.target-file-size-bytes"] + #[default(512 * 1024 * 1024)] + #[doc = "Target size in bytes for generated data files."] + pub write_target_file_size_bytes: usize, + + #[key = "write.delete.target-file-size-bytes"] + #[default(64 * 1024 * 1024)] + #[doc = "Target size in bytes for generated delete files."] + pub write_delete_target_file_size_bytes: usize, + + #[key = "write.object-storage.enabled"] + #[default(false)] + #[doc = "Whether the object-storage location provider adds a hash component to file paths."] + pub write_object_storage_enabled: bool, + + #[key = "write.object-storage.partitioned-paths"] + #[default(true)] + #[doc = "Whether object-storage file paths include partition values."] + pub write_object_storage_partitioned_paths: bool, + + #[key = "write.object-storage.path"] + #[default(None)] + #[doc = "Deprecated base object-storage path; use write.data.path instead."] + pub write_object_storage_path: Option<String>, + + #[key = "write.location-provider.impl"] + #[default(None)] + #[doc = "Optional custom location provider implementation."] + pub write_location_provider_impl: Option<String>, + + #[key = "write.folder-storage.path"] + #[default(None)] + #[doc = "Deprecated base folder-storage path; use write.data.path instead."] + pub write_folder_storage_path: Option<String>, + + #[key = "write.data.path"] + #[default(None)] + #[doc = "Base location for data files written after this property is set."] + pub write_data_path: Option<String>, + + #[key = "write.wap.enabled"] + #[default(false)] + #[doc = "Whether write-audit-publish writes are enabled."] + pub write_wap_enabled: bool, + + #[key = "write.distribution-mode"] + #[default(None)] + #[doc = "Optional write distribution mode: none, hash, or range."] + pub write_distribution_mode: Option<DistributionMode>, Review Comment: This should not be Option, the default mode should be DistributeMode::None ########## crates/iceberg/src/spec/parsed_table_prop.rs: ########## @@ -0,0 +1,1181 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Typed access to Iceberg table properties. +//! +//! [`ParsedTableProperties`] exposes Iceberg's string-keyed table properties as typed public +//! fields. Its JSON representation is a flat object whose keys and values are strings. +//! +//! # Create from defaults +//! +//! Start with Iceberg's defaults and modify public fields directly: +//! +//! ``` +//! use iceberg::spec::{DataFileFormat, ParsedTableProperties}; +//! +//! let mut properties = ParsedTableProperties::default(); +//! properties.write_format_default = DataFileFormat::Orc; +//! properties.write_data_path = Some("s3://warehouse/table/data".to_string()); +//! +//! assert_eq!(properties.write_format_default, DataFileFormat::Orc); +//! ``` +//! +//! # Deserialize from JSON +//! +//! JSON property values must be strings, matching Iceberg's table property map: +//! +//! ``` +//! use iceberg::spec::{DataFileFormat, ParsedTableProperties}; +//! +//! # fn main() -> Result<(), serde_json::Error> { +//! let properties: ParsedTableProperties = serde_json::from_value(serde_json::json!({ +//! "commit.retry.num-retries": "8", +//! "write.format.default": "orc" +//! }))?; +//! +//! assert_eq!(properties.commit_retry_num_retries, 8); +//! assert_eq!(properties.write_format_default, DataFileFormat::Orc); +//! # Ok(()) +//! # } +//! ``` +//! +//! # Serialize to JSON +//! +//! Serialization converts the typed fields back into Iceberg property keys: +//! +//! ``` +//! use iceberg::spec::ParsedTableProperties; +//! +//! # fn main() -> Result<(), serde_json::Error> { +//! let mut properties = ParsedTableProperties::default(); +//! properties.commit_retry_num_retries = 8; +//! properties.write_data_path = Some("s3://warehouse/table/data".to_string()); +//! +//! let json = serde_json::to_value(&properties)?; +//! assert_eq!(json["commit.retry.num-retries"], "8"); +//! assert_eq!(json["write.data.path"], "s3://warehouse/table/data"); +//! # Ok(()) +//! # } +//! ``` + +use std::collections::HashMap; + +use iceberg_property_macro::Properties; +use serde_with::{DeserializeFromStr, SerializeDisplay}; + +use crate::compression::CompressionCodec; +use crate::error::{Error, ErrorKind, Result}; +use crate::spec::{DataFileFormat, NameMapping}; + +/// Parquet data page version 1. +pub const PARQUET_PAGE_VERSION_V1: &str = "v1"; + +/// Parquet data page version 2. +pub const PARQUET_PAGE_VERSION_V2: &str = "v2"; + +/// Distribution applied to rows before writing files. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum DistributionMode { + /// Do not redistribute rows. + None, + /// Hash-distribute rows by partition values. + Hash, + /// Range-distribute rows by partition or sort values. + Range, +} + +/// Granularity used when creating position delete files. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum DeleteGranularity { + /// Group deletes for each referenced data file separately. + File, + /// Group deletes for different data files within a partition. + Partition, +} + +/// Isolation level used by row-level operations. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum IsolationLevel { + /// Fail if concurrent changes may contain rows matching the operation. + Serializable, + /// Validate only against data visible in the operation's snapshot. + Snapshot, +} + +/// Strategy used to apply row-level changes. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum RowLevelOperationMode { + /// Replace affected data files immediately. + CopyOnWrite, + /// Write delete files and merge changes while reading. + MergeOnRead, +} + +/// Strips trailing slashes from a location, preserving a bare URI scheme root. +fn strip_trailing_slash(path: &str) -> &str { + let mut path = path; + while !path.ends_with("://") { + let Some(stripped) = path.strip_suffix('/') else { + break; + }; + path = stripped; + } + path +} + +fn parse_metadata_location(value: &str) -> Result<Option<String>> { + if value.is_empty() { + return Err(Error::new(ErrorKind::DataInvalid, "path must not be empty")); + } + + Ok(Some(strip_trailing_slash(value).to_string())) +} + +fn parse_compression_codec(value: &str) -> Result<CompressionCodec> { + serde_json::from_value(serde_json::Value::String(value.to_lowercase())).map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid compression codec: {value}"), + ) + }) +} + +fn parse_metadata_file_compression(value: &str) -> Result<CompressionCodec> { + if value.is_empty() { + return Ok(CompressionCodec::None); + } + + let codec: CompressionCodec = serde_json::from_value(serde_json::Value::String( + value.to_lowercase(), + )) + .map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported.", + CompressionCodec::None.name(), + CompressionCodec::gzip_default().name() + ), + ) + })?; + + match codec { + CompressionCodec::None | CompressionCodec::Gzip(_) => Ok(codec), + _ => Err(Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported for metadata files.", + CompressionCodec::None.name(), + CompressionCodec::gzip_default().name() + ), + )), + } +} + +fn serialize_compression_codec(codec: &CompressionCodec) -> String { + codec.name().to_string() +} + +fn parse_comma_separated_strings(value: &str) -> Result<Vec<String>> { + Ok(value + .split(',') + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_string) + .collect()) +} + +fn serialize_comma_separated_strings(values: &[String]) -> String { + values.join(",") +} + +fn parse_name_mapping(value: &str) -> Result<Option<NameMapping>> { + serde_json::from_str(value).map(Some).map_err(|error| { + Error::new(ErrorKind::DataInvalid, "Invalid name mapping").with_source(error) + }) +} + +fn serialize_name_mapping(mapping: &Option<NameMapping>) -> String { + serde_json::to_string(mapping.as_ref().expect("checked is_some above")) + .expect("serializing a NameMapping cannot fail") +} + +/// Typed Iceberg table properties organized into documented sections. +/// +/// Serde represents this struct as Iceberg's flat string-to-string property map. Property +/// definitions and descriptions are based on the pinned [Java TableProperties implementation] +/// and [Apache Iceberg configuration documentation]. +/// +/// [Java TableProperties implementation]: https://github.com/apache/iceberg/blob/d8c10a1608170f0ba83be740d6ab0b6a3757cb3e/core/src/main/java/org/apache/iceberg/TableProperties.java +/// [Apache Iceberg configuration documentation]: https://github.com/apache/iceberg/blob/d8c10a1608170f0ba83be740d6ab0b6a3757cb3e/docs/docs/configuration.md +#[derive(Clone, Debug, Properties)] +pub struct ParsedTableProperties { + // General properties. + #[key = "comment"] + #[default(None)] + #[doc = "Table-level description of the table's business meaning and usage context."] + pub comment: Option<String>, + + #[key = "identifier-fields.rely"] + #[default(false)] + #[doc = "Whether query engines may rely on identifier fields as a primary key for optimization; this is not enforced on writes."] + pub identifier_fields_rely: bool, + + // Commit properties. + #[key = "commit.retry.num-retries"] + #[default(4)] + #[doc = "Number of times to retry a commit before failing."] + pub commit_retry_num_retries: usize, + + #[key = "commit.retry.min-wait-ms"] + #[default(100)] + #[doc = "Minimum time in milliseconds to wait before retrying a commit."] + pub commit_retry_min_wait_ms: u64, + + #[key = "commit.retry.max-wait-ms"] + #[default(60 * 1000)] + #[doc = "Maximum time in milliseconds to wait before retrying a commit."] + pub commit_retry_max_wait_ms: u64, + + #[key = "commit.retry.total-timeout-ms"] + #[default(30 * 60 * 1000)] + #[doc = "Total commit retry timeout in milliseconds."] + pub commit_retry_total_timeout_ms: u64, + + #[key = "commit.status-check.num-retries"] + #[default(3)] + #[doc = "Number of times to check whether a commit succeeded after connectivity is lost."] + pub commit_status_check_num_retries: usize, + + #[key = "commit.status-check.min-wait-ms"] + #[default(1000)] + #[doc = "Minimum time in milliseconds to wait before retrying a commit status check."] + pub commit_status_check_min_wait_ms: u64, + + #[key = "commit.status-check.max-wait-ms"] + #[default(60 * 1000)] + #[doc = "Maximum time in milliseconds to wait before retrying a commit status check."] + pub commit_status_check_max_wait_ms: u64, + + #[key = "commit.status-check.total-timeout-ms"] + #[default(30 * 60 * 1000)] + #[doc = "Total timeout in milliseconds in which commit status checking must succeed."] + pub commit_status_check_total_timeout_ms: u64, + + // Manifest properties. + #[key = "commit.manifest.target-size-bytes"] + #[default(8 * 1024 * 1024)] + #[doc = "Target size in bytes when merging manifest files."] + pub commit_manifest_target_size_bytes: usize, + + #[key = "commit.manifest.min-count-to-merge"] + #[default(100)] + #[doc = "Minimum number of manifests to accumulate before merging."] + pub commit_manifest_min_count_to_merge: usize, + + #[key = "commit.manifest-merge.enabled"] + #[default(true)] + #[doc = "Whether manifests are automatically merged during writes."] + pub commit_manifest_merge_enabled: bool, + + #[key = "write.manifest.compression-codec"] + #[default(CompressionCodec::gzip_default())] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "Compression codec used for manifest files."] + pub write_manifest_compression_codec: CompressionCodec, + + #[key = "write.manifest-lists.enabled"] + #[default(true)] + #[doc = "Deprecated flag for writing manifest lists; manifest lists are always enabled."] + pub write_manifest_lists_enabled: bool, + + // Write properties. + #[key = "write.format.default"] + #[default(DataFileFormat::Parquet)] + #[doc = "Default data file format: Parquet, Avro, or ORC."] + pub write_format_default: DataFileFormat, + + #[key = "write.delete.format.default"] + #[default(DataFileFormat::Parquet)] + #[doc = "Default delete file format: Parquet, Avro, or ORC."] + pub write_delete_format_default: DataFileFormat, + + #[key = "write.target-file-size-bytes"] + #[default(512 * 1024 * 1024)] + #[doc = "Target size in bytes for generated data files."] + pub write_target_file_size_bytes: usize, + + #[key = "write.delete.target-file-size-bytes"] + #[default(64 * 1024 * 1024)] + #[doc = "Target size in bytes for generated delete files."] + pub write_delete_target_file_size_bytes: usize, + + #[key = "write.object-storage.enabled"] + #[default(false)] + #[doc = "Whether the object-storage location provider adds a hash component to file paths."] + pub write_object_storage_enabled: bool, + + #[key = "write.object-storage.partitioned-paths"] + #[default(true)] + #[doc = "Whether object-storage file paths include partition values."] + pub write_object_storage_partitioned_paths: bool, + + #[key = "write.object-storage.path"] + #[default(None)] + #[doc = "Deprecated base object-storage path; use write.data.path instead."] + pub write_object_storage_path: Option<String>, + + #[key = "write.location-provider.impl"] + #[default(None)] + #[doc = "Optional custom location provider implementation."] + pub write_location_provider_impl: Option<String>, + + #[key = "write.folder-storage.path"] + #[default(None)] + #[doc = "Deprecated base folder-storage path; use write.data.path instead."] + pub write_folder_storage_path: Option<String>, + + #[key = "write.data.path"] + #[default(None)] + #[doc = "Base location for data files written after this property is set."] + pub write_data_path: Option<String>, + + #[key = "write.wap.enabled"] + #[default(false)] + #[doc = "Whether write-audit-publish writes are enabled."] + pub write_wap_enabled: bool, + + #[key = "write.distribution-mode"] + #[default(None)] + #[doc = "Optional write distribution mode: none, hash, or range."] + pub write_distribution_mode: Option<DistributionMode>, Review Comment: Apply to all compression codec logic ########## crates/iceberg/src/spec/parsed_table_prop.rs: ########## @@ -0,0 +1,1181 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Typed access to Iceberg table properties. +//! +//! [`ParsedTableProperties`] exposes Iceberg's string-keyed table properties as typed public +//! fields. Its JSON representation is a flat object whose keys and values are strings. +//! +//! # Create from defaults +//! +//! Start with Iceberg's defaults and modify public fields directly: +//! +//! ``` +//! use iceberg::spec::{DataFileFormat, ParsedTableProperties}; +//! +//! let mut properties = ParsedTableProperties::default(); +//! properties.write_format_default = DataFileFormat::Orc; +//! properties.write_data_path = Some("s3://warehouse/table/data".to_string()); +//! +//! assert_eq!(properties.write_format_default, DataFileFormat::Orc); +//! ``` +//! +//! # Deserialize from JSON +//! +//! JSON property values must be strings, matching Iceberg's table property map: +//! +//! ``` +//! use iceberg::spec::{DataFileFormat, ParsedTableProperties}; +//! +//! # fn main() -> Result<(), serde_json::Error> { +//! let properties: ParsedTableProperties = serde_json::from_value(serde_json::json!({ +//! "commit.retry.num-retries": "8", +//! "write.format.default": "orc" +//! }))?; +//! +//! assert_eq!(properties.commit_retry_num_retries, 8); +//! assert_eq!(properties.write_format_default, DataFileFormat::Orc); +//! # Ok(()) +//! # } +//! ``` +//! +//! # Serialize to JSON +//! +//! Serialization converts the typed fields back into Iceberg property keys: +//! +//! ``` +//! use iceberg::spec::ParsedTableProperties; +//! +//! # fn main() -> Result<(), serde_json::Error> { +//! let mut properties = ParsedTableProperties::default(); +//! properties.commit_retry_num_retries = 8; +//! properties.write_data_path = Some("s3://warehouse/table/data".to_string()); +//! +//! let json = serde_json::to_value(&properties)?; +//! assert_eq!(json["commit.retry.num-retries"], "8"); +//! assert_eq!(json["write.data.path"], "s3://warehouse/table/data"); +//! # Ok(()) +//! # } +//! ``` + +use std::collections::HashMap; + +use iceberg_property_macro::Properties; +use serde_with::{DeserializeFromStr, SerializeDisplay}; + +use crate::compression::CompressionCodec; +use crate::error::{Error, ErrorKind, Result}; +use crate::spec::{DataFileFormat, NameMapping}; + +/// Parquet data page version 1. +pub const PARQUET_PAGE_VERSION_V1: &str = "v1"; + +/// Parquet data page version 2. +pub const PARQUET_PAGE_VERSION_V2: &str = "v2"; + +/// Distribution applied to rows before writing files. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum DistributionMode { + /// Do not redistribute rows. + None, + /// Hash-distribute rows by partition values. + Hash, + /// Range-distribute rows by partition or sort values. + Range, +} + +/// Granularity used when creating position delete files. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum DeleteGranularity { + /// Group deletes for each referenced data file separately. + File, + /// Group deletes for different data files within a partition. + Partition, +} + +/// Isolation level used by row-level operations. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum IsolationLevel { + /// Fail if concurrent changes may contain rows matching the operation. + Serializable, + /// Validate only against data visible in the operation's snapshot. + Snapshot, +} + +/// Strategy used to apply row-level changes. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum RowLevelOperationMode { + /// Replace affected data files immediately. + CopyOnWrite, + /// Write delete files and merge changes while reading. + MergeOnRead, +} + +/// Strips trailing slashes from a location, preserving a bare URI scheme root. +fn strip_trailing_slash(path: &str) -> &str { + let mut path = path; + while !path.ends_with("://") { + let Some(stripped) = path.strip_suffix('/') else { + break; + }; + path = stripped; + } + path +} + +fn parse_metadata_location(value: &str) -> Result<Option<String>> { + if value.is_empty() { + return Err(Error::new(ErrorKind::DataInvalid, "path must not be empty")); + } + + Ok(Some(strip_trailing_slash(value).to_string())) +} + +fn parse_compression_codec(value: &str) -> Result<CompressionCodec> { + serde_json::from_value(serde_json::Value::String(value.to_lowercase())).map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid compression codec: {value}"), + ) + }) +} + +fn parse_metadata_file_compression(value: &str) -> Result<CompressionCodec> { + if value.is_empty() { + return Ok(CompressionCodec::None); + } + + let codec: CompressionCodec = serde_json::from_value(serde_json::Value::String( + value.to_lowercase(), + )) + .map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported.", + CompressionCodec::None.name(), + CompressionCodec::gzip_default().name() + ), + ) + })?; + + match codec { + CompressionCodec::None | CompressionCodec::Gzip(_) => Ok(codec), + _ => Err(Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported for metadata files.", + CompressionCodec::None.name(), + CompressionCodec::gzip_default().name() + ), + )), + } +} + +fn serialize_compression_codec(codec: &CompressionCodec) -> String { + codec.name().to_string() +} + +fn parse_comma_separated_strings(value: &str) -> Result<Vec<String>> { + Ok(value + .split(',') + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_string) + .collect()) +} + +fn serialize_comma_separated_strings(values: &[String]) -> String { + values.join(",") +} + +fn parse_name_mapping(value: &str) -> Result<Option<NameMapping>> { + serde_json::from_str(value).map(Some).map_err(|error| { + Error::new(ErrorKind::DataInvalid, "Invalid name mapping").with_source(error) + }) +} + +fn serialize_name_mapping(mapping: &Option<NameMapping>) -> String { + serde_json::to_string(mapping.as_ref().expect("checked is_some above")) + .expect("serializing a NameMapping cannot fail") +} + +/// Typed Iceberg table properties organized into documented sections. +/// +/// Serde represents this struct as Iceberg's flat string-to-string property map. Property +/// definitions and descriptions are based on the pinned [Java TableProperties implementation] +/// and [Apache Iceberg configuration documentation]. +/// +/// [Java TableProperties implementation]: https://github.com/apache/iceberg/blob/d8c10a1608170f0ba83be740d6ab0b6a3757cb3e/core/src/main/java/org/apache/iceberg/TableProperties.java +/// [Apache Iceberg configuration documentation]: https://github.com/apache/iceberg/blob/d8c10a1608170f0ba83be740d6ab0b6a3757cb3e/docs/docs/configuration.md +#[derive(Clone, Debug, Properties)] +pub struct ParsedTableProperties { + // General properties. + #[key = "comment"] + #[default(None)] + #[doc = "Table-level description of the table's business meaning and usage context."] + pub comment: Option<String>, + + #[key = "identifier-fields.rely"] + #[default(false)] + #[doc = "Whether query engines may rely on identifier fields as a primary key for optimization; this is not enforced on writes."] + pub identifier_fields_rely: bool, + + // Commit properties. + #[key = "commit.retry.num-retries"] + #[default(4)] + #[doc = "Number of times to retry a commit before failing."] + pub commit_retry_num_retries: usize, + + #[key = "commit.retry.min-wait-ms"] + #[default(100)] + #[doc = "Minimum time in milliseconds to wait before retrying a commit."] + pub commit_retry_min_wait_ms: u64, + + #[key = "commit.retry.max-wait-ms"] + #[default(60 * 1000)] + #[doc = "Maximum time in milliseconds to wait before retrying a commit."] + pub commit_retry_max_wait_ms: u64, + + #[key = "commit.retry.total-timeout-ms"] + #[default(30 * 60 * 1000)] + #[doc = "Total commit retry timeout in milliseconds."] + pub commit_retry_total_timeout_ms: u64, + + #[key = "commit.status-check.num-retries"] + #[default(3)] + #[doc = "Number of times to check whether a commit succeeded after connectivity is lost."] + pub commit_status_check_num_retries: usize, + + #[key = "commit.status-check.min-wait-ms"] + #[default(1000)] + #[doc = "Minimum time in milliseconds to wait before retrying a commit status check."] + pub commit_status_check_min_wait_ms: u64, + + #[key = "commit.status-check.max-wait-ms"] + #[default(60 * 1000)] + #[doc = "Maximum time in milliseconds to wait before retrying a commit status check."] + pub commit_status_check_max_wait_ms: u64, + + #[key = "commit.status-check.total-timeout-ms"] + #[default(30 * 60 * 1000)] + #[doc = "Total timeout in milliseconds in which commit status checking must succeed."] + pub commit_status_check_total_timeout_ms: u64, + + // Manifest properties. + #[key = "commit.manifest.target-size-bytes"] + #[default(8 * 1024 * 1024)] + #[doc = "Target size in bytes when merging manifest files."] + pub commit_manifest_target_size_bytes: usize, + + #[key = "commit.manifest.min-count-to-merge"] + #[default(100)] + #[doc = "Minimum number of manifests to accumulate before merging."] + pub commit_manifest_min_count_to_merge: usize, + + #[key = "commit.manifest-merge.enabled"] + #[default(true)] + #[doc = "Whether manifests are automatically merged during writes."] + pub commit_manifest_merge_enabled: bool, + + #[key = "write.manifest.compression-codec"] + #[default(CompressionCodec::gzip_default())] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "Compression codec used for manifest files."] + pub write_manifest_compression_codec: CompressionCodec, + + #[key = "write.manifest-lists.enabled"] + #[default(true)] + #[doc = "Deprecated flag for writing manifest lists; manifest lists are always enabled."] + pub write_manifest_lists_enabled: bool, + + // Write properties. + #[key = "write.format.default"] + #[default(DataFileFormat::Parquet)] + #[doc = "Default data file format: Parquet, Avro, or ORC."] + pub write_format_default: DataFileFormat, + + #[key = "write.delete.format.default"] + #[default(DataFileFormat::Parquet)] + #[doc = "Default delete file format: Parquet, Avro, or ORC."] + pub write_delete_format_default: DataFileFormat, + + #[key = "write.target-file-size-bytes"] + #[default(512 * 1024 * 1024)] + #[doc = "Target size in bytes for generated data files."] + pub write_target_file_size_bytes: usize, + + #[key = "write.delete.target-file-size-bytes"] + #[default(64 * 1024 * 1024)] + #[doc = "Target size in bytes for generated delete files."] + pub write_delete_target_file_size_bytes: usize, + + #[key = "write.object-storage.enabled"] + #[default(false)] + #[doc = "Whether the object-storage location provider adds a hash component to file paths."] + pub write_object_storage_enabled: bool, + + #[key = "write.object-storage.partitioned-paths"] + #[default(true)] + #[doc = "Whether object-storage file paths include partition values."] + pub write_object_storage_partitioned_paths: bool, + + #[key = "write.object-storage.path"] + #[default(None)] + #[doc = "Deprecated base object-storage path; use write.data.path instead."] + pub write_object_storage_path: Option<String>, + + #[key = "write.location-provider.impl"] + #[default(None)] + #[doc = "Optional custom location provider implementation."] + pub write_location_provider_impl: Option<String>, + + #[key = "write.folder-storage.path"] + #[default(None)] + #[doc = "Deprecated base folder-storage path; use write.data.path instead."] + pub write_folder_storage_path: Option<String>, + + #[key = "write.data.path"] + #[default(None)] + #[doc = "Base location for data files written after this property is set."] + pub write_data_path: Option<String>, + + #[key = "write.wap.enabled"] + #[default(false)] + #[doc = "Whether write-audit-publish writes are enabled."] + pub write_wap_enabled: bool, + + #[key = "write.distribution-mode"] + #[default(None)] + #[doc = "Optional write distribution mode: none, hash, or range."] + pub write_distribution_mode: Option<DistributionMode>, + + #[key = "write.datafusion.fanout.enabled"] + #[default(true)] + #[doc = "Whether DataFusion uses a fanout writer for partitioned tables."] + pub write_datafusion_fanout_enabled: bool, + + // Parquet properties. + #[key = "write.parquet.row-group-size-bytes"] + #[default(128 * 1024 * 1024)] + #[doc = "Parquet row group size in bytes for data files."] + pub write_parquet_row_group_size_bytes: usize, + + #[key = "write.delete.parquet.row-group-size-bytes"] + #[default(128 * 1024 * 1024)] + #[doc = "Parquet row group size in bytes for delete files."] + pub write_delete_parquet_row_group_size_bytes: usize, + + #[key = "write.parquet.page-size-bytes"] + #[default(1024 * 1024)] + #[doc = "Parquet page size in bytes for data files."] + pub write_parquet_page_size_bytes: usize, + + #[key = "write.delete.parquet.page-size-bytes"] + #[default(1024 * 1024)] + #[doc = "Parquet page size in bytes for delete files."] + pub write_delete_parquet_page_size_bytes: usize, + + #[key = "write.parquet.page-version"] + #[default(PARQUET_PAGE_VERSION_V1)] + #[doc = "Parquet data page version for data files: v1 or v2."] + pub write_parquet_page_version: String, + + #[key = "write.delete.parquet.page-version"] + #[default(PARQUET_PAGE_VERSION_V1)] + #[doc = "Parquet data page version for delete files: v1 or v2."] + pub write_delete_parquet_page_version: String, + + #[key = "write.parquet.page-row-limit"] + #[default(20_000)] + #[doc = "Maximum number of rows per Parquet page in data files."] + pub write_parquet_page_row_limit: usize, + + #[key = "write.delete.parquet.page-row-limit"] + #[default(20_000)] + #[doc = "Maximum number of rows per Parquet page in delete files."] + pub write_delete_parquet_page_row_limit: usize, + + #[key = "write.parquet.dict-size-bytes"] + #[default(2 * 1024 * 1024)] + #[doc = "Parquet dictionary page size in bytes for data files."] + pub write_parquet_dict_size_bytes: usize, + + #[key = "write.delete.parquet.dict-size-bytes"] + #[default(2 * 1024 * 1024)] + #[doc = "Parquet dictionary page size in bytes for delete files."] + pub write_delete_parquet_dict_size_bytes: usize, + + #[key = "write.parquet.compression-codec"] + #[default(CompressionCodec::zstd_default())] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "Parquet compression codec used for data files."] + pub write_parquet_compression_codec: CompressionCodec, + + #[key = "write.delete.parquet.compression-codec"] + #[default(CompressionCodec::zstd_default())] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "Parquet compression codec used for delete files."] + pub write_delete_parquet_compression_codec: CompressionCodec, + + #[key = "write.parquet.shred-variants"] + #[default(false)] + #[doc = "Whether variant columns use shredded Parquet encoding for improved query performance."] + pub write_parquet_shred_variants: bool, + + #[key = "write.parquet.variant-inference-buffer-size"] + #[default(100)] + #[doc = "Number of rows buffered for schema inference when variant shredding is enabled."] + pub write_parquet_variant_inference_buffer_size: usize, + + #[key = "write.parquet.row-group-check-min-record-count"] + #[default(100)] + #[doc = "Minimum record count between Parquet data-file row group size checks."] + pub write_parquet_row_group_check_min_record_count: usize, + + #[key = "write.delete.parquet.row-group-check-min-record-count"] + #[default(100)] + #[doc = "Minimum record count between Parquet delete-file row group size checks."] + pub write_delete_parquet_row_group_check_min_record_count: usize, + + #[key = "write.parquet.row-group-check-max-record-count"] + #[default(10_000)] + #[doc = "Maximum record count between Parquet data-file row group size checks."] + pub write_parquet_row_group_check_max_record_count: usize, + + #[key = "write.delete.parquet.row-group-check-max-record-count"] + #[default(10_000)] + #[doc = "Maximum record count between Parquet delete-file row group size checks."] + pub write_delete_parquet_row_group_check_max_record_count: usize, + + #[key = "write.parquet.row-group-size-track-uncompressed"] + #[default(false)] + #[doc = "Whether uncompressed data size is tracked to enforce the Parquet row group target."] + pub write_parquet_row_group_size_track_uncompressed: bool, + + #[key = "write.parquet.bloom-filter-max-bytes"] + #[default(1024 * 1024)] + #[doc = "Maximum number of bytes for a Parquet bloom filter bitset."] + pub write_parquet_bloom_filter_max_bytes: usize, + + #[key = "write.parquet.bloom-filter-adaptive-enabled"] + #[default(false)] + #[doc = "Whether adaptive Parquet bloom filter sizing selects the smallest suitable filter."] + pub write_parquet_bloom_filter_adaptive_enabled: bool, + + #[prefix = "write.parquet.bloom-filter-fpp.column."] + #[default(HashMap::new())] + #[doc = "Per-column Parquet bloom filter false-positive probabilities, keyed by column name."] + pub write_parquet_bloom_filter_fpp_column: HashMap<String, f64>, + + #[prefix = "write.parquet.bloom-filter-ndv.column."] + #[default(HashMap::new())] + #[doc = "Per-column expected distinct-value counts for Parquet bloom filters."] + pub write_parquet_bloom_filter_ndv_column: HashMap<String, u64>, + + #[prefix = "write.parquet.bloom-filter-enabled.column."] + #[default(HashMap::new())] + #[doc = "Per-column flags controlling whether Parquet bloom filters are written."] + pub write_parquet_bloom_filter_enabled_column: HashMap<String, bool>, + + #[prefix = "write.parquet.stats-enabled.column."] + #[default(HashMap::new())] + #[doc = "Per-column flags controlling whether Parquet column statistics are collected."] + pub write_parquet_stats_enabled_column: HashMap<String, bool>, + + #[prefix = "write.parquet.dict-encoding-enabled.column."] + #[default(HashMap::new())] + #[doc = "Per-column flags controlling whether Parquet dictionary encoding is used."] + pub write_parquet_dict_encoding_enabled_column: HashMap<String, bool>, + + #[key = "write.parquet.content-defined-chunking.enabled"] + #[default(false)] + #[doc = "Whether Parquet content-defined chunking is enabled."] + pub write_parquet_content_defined_chunking_enabled: bool, + + #[key = "write.parquet.content-defined-chunking.min-chunk-size"] + #[default(256 * 1024)] + #[doc = "Minimum Parquet content-defined chunk size in bytes."] + pub write_parquet_content_defined_chunking_min_chunk_size: usize, + + #[key = "write.parquet.content-defined-chunking.max-chunk-size"] + #[default(1024 * 1024)] + #[doc = "Maximum Parquet content-defined chunk size in bytes."] + pub write_parquet_content_defined_chunking_max_chunk_size: usize, + + #[key = "write.parquet.content-defined-chunking.norm-level"] + #[default(0)] + #[doc = "Gearhash normalization level used by Parquet content-defined chunking."] + pub write_parquet_content_defined_chunking_norm_level: i32, + + // Avro properties. + #[key = "write.avro.compression-codec"] + #[default(CompressionCodec::gzip_default())] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "Avro compression codec used for data files."] + pub write_avro_compression_codec: CompressionCodec, + + #[key = "write.delete.avro.compression-codec"] + #[default(CompressionCodec::gzip_default())] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "Avro compression codec used for delete files."] + pub write_delete_avro_compression_codec: CompressionCodec, + + // ORC properties. + #[key = "write.orc.stripe-size-bytes"] + #[default(64 * 1024 * 1024)] + #[doc = "Default ORC stripe size in bytes for data files."] + pub write_orc_stripe_size_bytes: u64, + + #[key = "write.delete.orc.stripe-size-bytes"] + #[default(64 * 1024 * 1024)] + #[doc = "Default ORC stripe size in bytes for delete files."] + pub write_delete_orc_stripe_size_bytes: u64, + + #[key = "write.orc.bloom.filter.columns"] + #[default(Vec::new())] + #[parse_with(parse_comma_separated_strings)] + #[serialize_with(serialize_comma_separated_strings)] + #[doc = "Comma-separated column names for which ORC bloom filters are created."] + pub write_orc_bloom_filter_columns: Vec<String>, + + #[key = "write.orc.bloom.filter.fpp"] + #[default(0.05)] + #[doc = "False-positive probability for ORC bloom filters."] + pub write_orc_bloom_filter_fpp: f64, + + #[key = "write.orc.block-size-bytes"] + #[default(256 * 1024 * 1024)] + #[doc = "Default file-system block size in bytes for ORC data files."] + pub write_orc_block_size_bytes: u64, + + #[key = "write.delete.orc.block-size-bytes"] + #[default(256 * 1024 * 1024)] + #[doc = "Default file-system block size in bytes for ORC delete files."] + pub write_delete_orc_block_size_bytes: u64, + + #[key = "write.orc.vectorized.batch-size"] + #[default(1024)] + #[doc = "ORC vectorized write batch size for data files."] + pub write_orc_vectorized_batch_size: usize, + + #[key = "write.delete.orc.vectorized.batch-size"] + #[default(1024)] + #[doc = "ORC vectorized write batch size for delete files."] + pub write_delete_orc_vectorized_batch_size: usize, + + #[key = "write.orc.compression-codec"] + #[default(CompressionCodec::Zlib)] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "ORC compression codec used for data files."] + pub write_orc_compression_codec: CompressionCodec, + + #[key = "write.delete.orc.compression-codec"] + #[default(CompressionCodec::Zlib)] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "ORC compression codec used for delete files."] + pub write_delete_orc_compression_codec: CompressionCodec, + + #[key = "write.orc.compression-strategy"] + #[default("speed")] + #[doc = "ORC compression strategy for data files: speed or compression."] + pub write_orc_compression_strategy: String, Review Comment: Create string constatns for speed and compression, and we should use constants here. ########## crates/iceberg/src/spec/parsed_table_prop.rs: ########## @@ -0,0 +1,1181 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Typed access to Iceberg table properties. +//! +//! [`ParsedTableProperties`] exposes Iceberg's string-keyed table properties as typed public +//! fields. Its JSON representation is a flat object whose keys and values are strings. +//! +//! # Create from defaults +//! +//! Start with Iceberg's defaults and modify public fields directly: +//! +//! ``` +//! use iceberg::spec::{DataFileFormat, ParsedTableProperties}; +//! +//! let mut properties = ParsedTableProperties::default(); +//! properties.write_format_default = DataFileFormat::Orc; +//! properties.write_data_path = Some("s3://warehouse/table/data".to_string()); +//! +//! assert_eq!(properties.write_format_default, DataFileFormat::Orc); +//! ``` +//! +//! # Deserialize from JSON +//! +//! JSON property values must be strings, matching Iceberg's table property map: +//! +//! ``` +//! use iceberg::spec::{DataFileFormat, ParsedTableProperties}; +//! +//! # fn main() -> Result<(), serde_json::Error> { +//! let properties: ParsedTableProperties = serde_json::from_value(serde_json::json!({ +//! "commit.retry.num-retries": "8", +//! "write.format.default": "orc" +//! }))?; +//! +//! assert_eq!(properties.commit_retry_num_retries, 8); +//! assert_eq!(properties.write_format_default, DataFileFormat::Orc); +//! # Ok(()) +//! # } +//! ``` +//! +//! # Serialize to JSON +//! +//! Serialization converts the typed fields back into Iceberg property keys: +//! +//! ``` +//! use iceberg::spec::ParsedTableProperties; +//! +//! # fn main() -> Result<(), serde_json::Error> { +//! let mut properties = ParsedTableProperties::default(); +//! properties.commit_retry_num_retries = 8; +//! properties.write_data_path = Some("s3://warehouse/table/data".to_string()); +//! +//! let json = serde_json::to_value(&properties)?; +//! assert_eq!(json["commit.retry.num-retries"], "8"); +//! assert_eq!(json["write.data.path"], "s3://warehouse/table/data"); +//! # Ok(()) +//! # } +//! ``` + +use std::collections::HashMap; + +use iceberg_property_macro::Properties; +use serde_with::{DeserializeFromStr, SerializeDisplay}; + +use crate::compression::CompressionCodec; +use crate::error::{Error, ErrorKind, Result}; +use crate::spec::{DataFileFormat, NameMapping}; + +/// Parquet data page version 1. +pub const PARQUET_PAGE_VERSION_V1: &str = "v1"; + +/// Parquet data page version 2. +pub const PARQUET_PAGE_VERSION_V2: &str = "v2"; + +/// Distribution applied to rows before writing files. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum DistributionMode { + /// Do not redistribute rows. + None, + /// Hash-distribute rows by partition values. + Hash, + /// Range-distribute rows by partition or sort values. + Range, +} + +/// Granularity used when creating position delete files. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum DeleteGranularity { + /// Group deletes for each referenced data file separately. + File, + /// Group deletes for different data files within a partition. + Partition, +} + +/// Isolation level used by row-level operations. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum IsolationLevel { + /// Fail if concurrent changes may contain rows matching the operation. + Serializable, + /// Validate only against data visible in the operation's snapshot. + Snapshot, +} + +/// Strategy used to apply row-level changes. +#[derive( + Debug, + PartialEq, + Eq, + Clone, + Copy, + SerializeDisplay, + DeserializeFromStr, + strum::Display, + strum::EnumString, +)] +#[strum(ascii_case_insensitive, serialize_all = "kebab-case")] +pub enum RowLevelOperationMode { + /// Replace affected data files immediately. + CopyOnWrite, + /// Write delete files and merge changes while reading. + MergeOnRead, +} + +/// Strips trailing slashes from a location, preserving a bare URI scheme root. +fn strip_trailing_slash(path: &str) -> &str { + let mut path = path; + while !path.ends_with("://") { + let Some(stripped) = path.strip_suffix('/') else { + break; + }; + path = stripped; + } + path +} + +fn parse_metadata_location(value: &str) -> Result<Option<String>> { + if value.is_empty() { + return Err(Error::new(ErrorKind::DataInvalid, "path must not be empty")); + } + + Ok(Some(strip_trailing_slash(value).to_string())) +} + +fn parse_compression_codec(value: &str) -> Result<CompressionCodec> { + serde_json::from_value(serde_json::Value::String(value.to_lowercase())).map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid compression codec: {value}"), + ) + }) +} + +fn parse_metadata_file_compression(value: &str) -> Result<CompressionCodec> { + if value.is_empty() { + return Ok(CompressionCodec::None); + } + + let codec: CompressionCodec = serde_json::from_value(serde_json::Value::String( + value.to_lowercase(), + )) + .map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported.", + CompressionCodec::None.name(), + CompressionCodec::gzip_default().name() + ), + ) + })?; + + match codec { + CompressionCodec::None | CompressionCodec::Gzip(_) => Ok(codec), + _ => Err(Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported for metadata files.", + CompressionCodec::None.name(), + CompressionCodec::gzip_default().name() + ), + )), + } +} + +fn serialize_compression_codec(codec: &CompressionCodec) -> String { + codec.name().to_string() +} + +fn parse_comma_separated_strings(value: &str) -> Result<Vec<String>> { + Ok(value + .split(',') + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_string) + .collect()) +} + +fn serialize_comma_separated_strings(values: &[String]) -> String { + values.join(",") +} + +fn parse_name_mapping(value: &str) -> Result<Option<NameMapping>> { + serde_json::from_str(value).map(Some).map_err(|error| { + Error::new(ErrorKind::DataInvalid, "Invalid name mapping").with_source(error) + }) +} + +fn serialize_name_mapping(mapping: &Option<NameMapping>) -> String { + serde_json::to_string(mapping.as_ref().expect("checked is_some above")) + .expect("serializing a NameMapping cannot fail") +} + +/// Typed Iceberg table properties organized into documented sections. +/// +/// Serde represents this struct as Iceberg's flat string-to-string property map. Property +/// definitions and descriptions are based on the pinned [Java TableProperties implementation] +/// and [Apache Iceberg configuration documentation]. +/// +/// [Java TableProperties implementation]: https://github.com/apache/iceberg/blob/d8c10a1608170f0ba83be740d6ab0b6a3757cb3e/core/src/main/java/org/apache/iceberg/TableProperties.java +/// [Apache Iceberg configuration documentation]: https://github.com/apache/iceberg/blob/d8c10a1608170f0ba83be740d6ab0b6a3757cb3e/docs/docs/configuration.md +#[derive(Clone, Debug, Properties)] +pub struct ParsedTableProperties { + // General properties. + #[key = "comment"] + #[default(None)] + #[doc = "Table-level description of the table's business meaning and usage context."] + pub comment: Option<String>, + + #[key = "identifier-fields.rely"] + #[default(false)] + #[doc = "Whether query engines may rely on identifier fields as a primary key for optimization; this is not enforced on writes."] + pub identifier_fields_rely: bool, + + // Commit properties. + #[key = "commit.retry.num-retries"] + #[default(4)] + #[doc = "Number of times to retry a commit before failing."] + pub commit_retry_num_retries: usize, + + #[key = "commit.retry.min-wait-ms"] + #[default(100)] + #[doc = "Minimum time in milliseconds to wait before retrying a commit."] + pub commit_retry_min_wait_ms: u64, + + #[key = "commit.retry.max-wait-ms"] + #[default(60 * 1000)] + #[doc = "Maximum time in milliseconds to wait before retrying a commit."] + pub commit_retry_max_wait_ms: u64, + + #[key = "commit.retry.total-timeout-ms"] + #[default(30 * 60 * 1000)] + #[doc = "Total commit retry timeout in milliseconds."] + pub commit_retry_total_timeout_ms: u64, + + #[key = "commit.status-check.num-retries"] + #[default(3)] + #[doc = "Number of times to check whether a commit succeeded after connectivity is lost."] + pub commit_status_check_num_retries: usize, + + #[key = "commit.status-check.min-wait-ms"] + #[default(1000)] + #[doc = "Minimum time in milliseconds to wait before retrying a commit status check."] + pub commit_status_check_min_wait_ms: u64, + + #[key = "commit.status-check.max-wait-ms"] + #[default(60 * 1000)] + #[doc = "Maximum time in milliseconds to wait before retrying a commit status check."] + pub commit_status_check_max_wait_ms: u64, + + #[key = "commit.status-check.total-timeout-ms"] + #[default(30 * 60 * 1000)] + #[doc = "Total timeout in milliseconds in which commit status checking must succeed."] + pub commit_status_check_total_timeout_ms: u64, + + // Manifest properties. + #[key = "commit.manifest.target-size-bytes"] + #[default(8 * 1024 * 1024)] + #[doc = "Target size in bytes when merging manifest files."] + pub commit_manifest_target_size_bytes: usize, + + #[key = "commit.manifest.min-count-to-merge"] + #[default(100)] + #[doc = "Minimum number of manifests to accumulate before merging."] + pub commit_manifest_min_count_to_merge: usize, + + #[key = "commit.manifest-merge.enabled"] + #[default(true)] + #[doc = "Whether manifests are automatically merged during writes."] + pub commit_manifest_merge_enabled: bool, + + #[key = "write.manifest.compression-codec"] + #[default(CompressionCodec::gzip_default())] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "Compression codec used for manifest files."] + pub write_manifest_compression_codec: CompressionCodec, + + #[key = "write.manifest-lists.enabled"] + #[default(true)] + #[doc = "Deprecated flag for writing manifest lists; manifest lists are always enabled."] + pub write_manifest_lists_enabled: bool, + + // Write properties. + #[key = "write.format.default"] + #[default(DataFileFormat::Parquet)] + #[doc = "Default data file format: Parquet, Avro, or ORC."] + pub write_format_default: DataFileFormat, + + #[key = "write.delete.format.default"] + #[default(DataFileFormat::Parquet)] + #[doc = "Default delete file format: Parquet, Avro, or ORC."] + pub write_delete_format_default: DataFileFormat, + + #[key = "write.target-file-size-bytes"] + #[default(512 * 1024 * 1024)] + #[doc = "Target size in bytes for generated data files."] + pub write_target_file_size_bytes: usize, + + #[key = "write.delete.target-file-size-bytes"] + #[default(64 * 1024 * 1024)] + #[doc = "Target size in bytes for generated delete files."] + pub write_delete_target_file_size_bytes: usize, + + #[key = "write.object-storage.enabled"] + #[default(false)] + #[doc = "Whether the object-storage location provider adds a hash component to file paths."] + pub write_object_storage_enabled: bool, + + #[key = "write.object-storage.partitioned-paths"] + #[default(true)] + #[doc = "Whether object-storage file paths include partition values."] + pub write_object_storage_partitioned_paths: bool, + + #[key = "write.object-storage.path"] + #[default(None)] + #[doc = "Deprecated base object-storage path; use write.data.path instead."] + pub write_object_storage_path: Option<String>, + + #[key = "write.location-provider.impl"] + #[default(None)] + #[doc = "Optional custom location provider implementation."] + pub write_location_provider_impl: Option<String>, + + #[key = "write.folder-storage.path"] + #[default(None)] + #[doc = "Deprecated base folder-storage path; use write.data.path instead."] + pub write_folder_storage_path: Option<String>, + + #[key = "write.data.path"] + #[default(None)] + #[doc = "Base location for data files written after this property is set."] + pub write_data_path: Option<String>, + + #[key = "write.wap.enabled"] + #[default(false)] + #[doc = "Whether write-audit-publish writes are enabled."] + pub write_wap_enabled: bool, + + #[key = "write.distribution-mode"] + #[default(None)] + #[doc = "Optional write distribution mode: none, hash, or range."] + pub write_distribution_mode: Option<DistributionMode>, + + #[key = "write.datafusion.fanout.enabled"] + #[default(true)] + #[doc = "Whether DataFusion uses a fanout writer for partitioned tables."] + pub write_datafusion_fanout_enabled: bool, + + // Parquet properties. + #[key = "write.parquet.row-group-size-bytes"] + #[default(128 * 1024 * 1024)] + #[doc = "Parquet row group size in bytes for data files."] + pub write_parquet_row_group_size_bytes: usize, + + #[key = "write.delete.parquet.row-group-size-bytes"] + #[default(128 * 1024 * 1024)] + #[doc = "Parquet row group size in bytes for delete files."] + pub write_delete_parquet_row_group_size_bytes: usize, + + #[key = "write.parquet.page-size-bytes"] + #[default(1024 * 1024)] + #[doc = "Parquet page size in bytes for data files."] + pub write_parquet_page_size_bytes: usize, + + #[key = "write.delete.parquet.page-size-bytes"] + #[default(1024 * 1024)] + #[doc = "Parquet page size in bytes for delete files."] + pub write_delete_parquet_page_size_bytes: usize, + + #[key = "write.parquet.page-version"] + #[default(PARQUET_PAGE_VERSION_V1)] + #[doc = "Parquet data page version for data files: v1 or v2."] + pub write_parquet_page_version: String, + + #[key = "write.delete.parquet.page-version"] + #[default(PARQUET_PAGE_VERSION_V1)] + #[doc = "Parquet data page version for delete files: v1 or v2."] + pub write_delete_parquet_page_version: String, + + #[key = "write.parquet.page-row-limit"] + #[default(20_000)] + #[doc = "Maximum number of rows per Parquet page in data files."] + pub write_parquet_page_row_limit: usize, + + #[key = "write.delete.parquet.page-row-limit"] + #[default(20_000)] + #[doc = "Maximum number of rows per Parquet page in delete files."] + pub write_delete_parquet_page_row_limit: usize, + + #[key = "write.parquet.dict-size-bytes"] + #[default(2 * 1024 * 1024)] + #[doc = "Parquet dictionary page size in bytes for data files."] + pub write_parquet_dict_size_bytes: usize, + + #[key = "write.delete.parquet.dict-size-bytes"] + #[default(2 * 1024 * 1024)] + #[doc = "Parquet dictionary page size in bytes for delete files."] + pub write_delete_parquet_dict_size_bytes: usize, + + #[key = "write.parquet.compression-codec"] + #[default(CompressionCodec::zstd_default())] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "Parquet compression codec used for data files."] + pub write_parquet_compression_codec: CompressionCodec, + + #[key = "write.delete.parquet.compression-codec"] + #[default(CompressionCodec::zstd_default())] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "Parquet compression codec used for delete files."] + pub write_delete_parquet_compression_codec: CompressionCodec, + + #[key = "write.parquet.shred-variants"] + #[default(false)] + #[doc = "Whether variant columns use shredded Parquet encoding for improved query performance."] + pub write_parquet_shred_variants: bool, + + #[key = "write.parquet.variant-inference-buffer-size"] + #[default(100)] + #[doc = "Number of rows buffered for schema inference when variant shredding is enabled."] + pub write_parquet_variant_inference_buffer_size: usize, + + #[key = "write.parquet.row-group-check-min-record-count"] + #[default(100)] + #[doc = "Minimum record count between Parquet data-file row group size checks."] + pub write_parquet_row_group_check_min_record_count: usize, + + #[key = "write.delete.parquet.row-group-check-min-record-count"] + #[default(100)] + #[doc = "Minimum record count between Parquet delete-file row group size checks."] + pub write_delete_parquet_row_group_check_min_record_count: usize, + + #[key = "write.parquet.row-group-check-max-record-count"] + #[default(10_000)] + #[doc = "Maximum record count between Parquet data-file row group size checks."] + pub write_parquet_row_group_check_max_record_count: usize, + + #[key = "write.delete.parquet.row-group-check-max-record-count"] + #[default(10_000)] + #[doc = "Maximum record count between Parquet delete-file row group size checks."] + pub write_delete_parquet_row_group_check_max_record_count: usize, + + #[key = "write.parquet.row-group-size-track-uncompressed"] + #[default(false)] + #[doc = "Whether uncompressed data size is tracked to enforce the Parquet row group target."] + pub write_parquet_row_group_size_track_uncompressed: bool, + + #[key = "write.parquet.bloom-filter-max-bytes"] + #[default(1024 * 1024)] + #[doc = "Maximum number of bytes for a Parquet bloom filter bitset."] + pub write_parquet_bloom_filter_max_bytes: usize, + + #[key = "write.parquet.bloom-filter-adaptive-enabled"] + #[default(false)] + #[doc = "Whether adaptive Parquet bloom filter sizing selects the smallest suitable filter."] + pub write_parquet_bloom_filter_adaptive_enabled: bool, + + #[prefix = "write.parquet.bloom-filter-fpp.column."] + #[default(HashMap::new())] + #[doc = "Per-column Parquet bloom filter false-positive probabilities, keyed by column name."] + pub write_parquet_bloom_filter_fpp_column: HashMap<String, f64>, + + #[prefix = "write.parquet.bloom-filter-ndv.column."] + #[default(HashMap::new())] + #[doc = "Per-column expected distinct-value counts for Parquet bloom filters."] + pub write_parquet_bloom_filter_ndv_column: HashMap<String, u64>, + + #[prefix = "write.parquet.bloom-filter-enabled.column."] + #[default(HashMap::new())] + #[doc = "Per-column flags controlling whether Parquet bloom filters are written."] + pub write_parquet_bloom_filter_enabled_column: HashMap<String, bool>, + + #[prefix = "write.parquet.stats-enabled.column."] + #[default(HashMap::new())] + #[doc = "Per-column flags controlling whether Parquet column statistics are collected."] + pub write_parquet_stats_enabled_column: HashMap<String, bool>, + + #[prefix = "write.parquet.dict-encoding-enabled.column."] + #[default(HashMap::new())] + #[doc = "Per-column flags controlling whether Parquet dictionary encoding is used."] + pub write_parquet_dict_encoding_enabled_column: HashMap<String, bool>, + + #[key = "write.parquet.content-defined-chunking.enabled"] + #[default(false)] + #[doc = "Whether Parquet content-defined chunking is enabled."] + pub write_parquet_content_defined_chunking_enabled: bool, + + #[key = "write.parquet.content-defined-chunking.min-chunk-size"] + #[default(256 * 1024)] + #[doc = "Minimum Parquet content-defined chunk size in bytes."] + pub write_parquet_content_defined_chunking_min_chunk_size: usize, + + #[key = "write.parquet.content-defined-chunking.max-chunk-size"] + #[default(1024 * 1024)] + #[doc = "Maximum Parquet content-defined chunk size in bytes."] + pub write_parquet_content_defined_chunking_max_chunk_size: usize, + + #[key = "write.parquet.content-defined-chunking.norm-level"] + #[default(0)] + #[doc = "Gearhash normalization level used by Parquet content-defined chunking."] + pub write_parquet_content_defined_chunking_norm_level: i32, + + // Avro properties. + #[key = "write.avro.compression-codec"] + #[default(CompressionCodec::gzip_default())] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "Avro compression codec used for data files."] + pub write_avro_compression_codec: CompressionCodec, + + #[key = "write.delete.avro.compression-codec"] + #[default(CompressionCodec::gzip_default())] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "Avro compression codec used for delete files."] + pub write_delete_avro_compression_codec: CompressionCodec, + + // ORC properties. + #[key = "write.orc.stripe-size-bytes"] + #[default(64 * 1024 * 1024)] + #[doc = "Default ORC stripe size in bytes for data files."] + pub write_orc_stripe_size_bytes: u64, + + #[key = "write.delete.orc.stripe-size-bytes"] + #[default(64 * 1024 * 1024)] + #[doc = "Default ORC stripe size in bytes for delete files."] + pub write_delete_orc_stripe_size_bytes: u64, + + #[key = "write.orc.bloom.filter.columns"] + #[default(Vec::new())] + #[parse_with(parse_comma_separated_strings)] + #[serialize_with(serialize_comma_separated_strings)] + #[doc = "Comma-separated column names for which ORC bloom filters are created."] + pub write_orc_bloom_filter_columns: Vec<String>, + + #[key = "write.orc.bloom.filter.fpp"] + #[default(0.05)] + #[doc = "False-positive probability for ORC bloom filters."] + pub write_orc_bloom_filter_fpp: f64, + + #[key = "write.orc.block-size-bytes"] + #[default(256 * 1024 * 1024)] + #[doc = "Default file-system block size in bytes for ORC data files."] + pub write_orc_block_size_bytes: u64, + + #[key = "write.delete.orc.block-size-bytes"] + #[default(256 * 1024 * 1024)] + #[doc = "Default file-system block size in bytes for ORC delete files."] + pub write_delete_orc_block_size_bytes: u64, + + #[key = "write.orc.vectorized.batch-size"] + #[default(1024)] + #[doc = "ORC vectorized write batch size for data files."] + pub write_orc_vectorized_batch_size: usize, + + #[key = "write.delete.orc.vectorized.batch-size"] + #[default(1024)] + #[doc = "ORC vectorized write batch size for delete files."] + pub write_delete_orc_vectorized_batch_size: usize, + + #[key = "write.orc.compression-codec"] + #[default(CompressionCodec::Zlib)] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "ORC compression codec used for data files."] + pub write_orc_compression_codec: CompressionCodec, + + #[key = "write.delete.orc.compression-codec"] + #[default(CompressionCodec::Zlib)] + #[parse_with(parse_compression_codec)] + #[serialize_with(serialize_compression_codec)] + #[doc = "ORC compression codec used for delete files."] + pub write_delete_orc_compression_codec: CompressionCodec, + + #[key = "write.orc.compression-strategy"] + #[default("speed")] + #[doc = "ORC compression strategy for data files: speed or compression."] + pub write_orc_compression_strategy: String, + + #[key = "write.delete.orc.compression-strategy"] + #[default("speed")] + #[doc = "ORC compression strategy for delete files: speed or compression."] + pub write_delete_orc_compression_strategy: String, + + // Read properties. + #[key = "read.split.target-size"] + #[default(128 * 1024 * 1024)] + #[doc = "Target size in bytes when combining data input splits."] + pub read_split_target_size: u64, + + #[key = "read.split.metadata-target-size"] + #[default(32 * 1024 * 1024)] + #[doc = "Target size in bytes when combining metadata input splits."] + pub read_split_metadata_target_size: u64, + + #[key = "read.split.planning-lookback"] + #[default(10)] + #[doc = "Number of bins considered when combining input splits."] + pub read_split_planning_lookback: usize, + + #[key = "read.split.open-file-cost"] + #[default(4 * 1024 * 1024)] + #[doc = "Estimated file-open cost used as a minimum weight when combining splits."] + pub read_split_open_file_cost: u64, + + #[key = "read.split.adaptive-size.enabled"] + #[default(true)] + #[doc = "Whether split size is adapted to the scan size."] + pub read_split_adaptive_size_enabled: bool, + + #[key = "read.parquet.vectorization.enabled"] + #[default(true)] + #[doc = "Whether Parquet vectorized reads are enabled."] + pub read_parquet_vectorization_enabled: bool, + + #[key = "read.parquet.vectorization.batch-size"] + #[default(5000)] + #[doc = "Batch size for Parquet vectorized reads."] + pub read_parquet_vectorization_batch_size: usize, + + #[key = "read.orc.vectorization.enabled"] + #[default(false)] + #[doc = "Whether ORC vectorized reads are enabled."] + pub read_orc_vectorization_enabled: bool, + + #[key = "read.orc.vectorization.batch-size"] + #[default(5000)] + #[doc = "Batch size for ORC vectorized reads."] + pub read_orc_vectorization_batch_size: usize, + + #[key = "read.data-planning-mode"] + #[default("auto")] + #[doc = "Planning mode used for data files."] + pub read_data_planning_mode: String, + + #[key = "read.delete-planning-mode"] + #[default("auto")] + #[doc = "Planning mode used for delete files."] + pub read_delete_planning_mode: String, + + // Metadata properties. + #[key = "write.metadata.path"] + #[default(None)] + #[parse_with(parse_metadata_location)] + #[doc = "Base location for metadata files written after this property is set, with trailing slashes removed."] + pub write_metadata_path: Option<String>, + + #[key = "write.summary.partition-limit"] + #[default(0)] + #[doc = "Maximum changed-partition count for including partition-level statistics in snapshot summaries."] + pub write_summary_partition_limit: u64, + + #[key = "write.metadata.compression-codec"] + #[default(CompressionCodec::None)] + #[parse_with(parse_metadata_file_compression)] + #[serialize_with(serialize_compression_codec)] + #[doc = "Compression codec for metadata JSON files: none or gzip."] + pub write_metadata_compression_codec: CompressionCodec, + + #[key = "write.metadata.previous-versions-max"] + #[default(100)] + #[doc = "Maximum number of previous metadata file versions to track."] + pub write_metadata_previous_versions_max: usize, + + #[key = "write.metadata.delete-after-commit.enabled"] + #[default(false)] + #[doc = "Whether the oldest tracked metadata file is deleted after each commit."] + pub write_metadata_delete_after_commit_enabled: bool, + + #[key = "write.metadata.metrics.max-inferred-column-defaults"] + #[default(100)] + #[doc = "Maximum number of columns that receive inferred metrics defaults."] + pub write_metadata_metrics_max_inferred_column_defaults: usize, + + #[prefix = "write.metadata.metrics.column."] + #[default(HashMap::new())] + #[doc = "Per-column metrics modes keyed by column name."] + pub write_metadata_metrics_column: HashMap<String, String>, + + #[key = "write.metadata.metrics.default"] + #[default("truncate(16)")] + #[doc = "Default metrics mode for table columns."] + pub write_metadata_metrics_default: String, + + #[key = "schema.name-mapping.default"] + #[default(None)] + #[parse_with(parse_name_mapping)] + #[serialize_with(serialize_name_mapping)] Review Comment: NameMapping already implement Ser/De, do we really need this? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
