kevinjqliu commented on code in PR #2955:
URL: https://github.com/apache/iceberg-rust/pull/2955#discussion_r3725030472
##########
crates/iceberg/src/spec/table_metadata.rs:
##########
@@ -4346,8 +4346,7 @@ mod tests {
}
#[test]
- fn test_metadata_location_trims_trailing_slash() {
- // A configured path with a trailing slash must not yield a doubled
separator
Review Comment:
is this behavior change intentional?
previous parser
([`parse_location_property`](https://github.com/apache/iceberg-rust/pull/2955/changes#diff-b3937e8d5290e685464b83957bdce1758ebe4ce122e25b12edcffd76280582afL58))
had custom logic to strip out the slashes.
And this test verifies that behavior
##########
crates/iceberg/Cargo.toml:
##########
@@ -59,6 +59,7 @@ flate2 = { workspace = true }
fnv = { workspace = true }
form_urlencoded = { workspace = true }
futures = { workspace = true }
+iceberg-property-macro = { version = "0.10.0", path = "../property-macro" }
Review Comment:
i think we have to publish the new crate, otherwise the `iceberg` crate
cannot be packaged. 😭
```
cargo package -p iceberg --no-verify
```
##########
crates/iceberg/src/compression.rs:
##########
@@ -68,10 +75,113 @@ impl CompressionCodec {
pub fn name(&self) -> &'static str {
match self {
CompressionCodec::None => "none",
+ CompressionCodec::Brotli => "brotli",
CompressionCodec::Lz4 => "lz4",
+ CompressionCodec::Lzo => "lzo",
CompressionCodec::Zstd(_) => "zstd",
CompressionCodec::Gzip(_) => "gzip",
CompressionCodec::Snappy => "snappy",
+ CompressionCodec::Zlib => "zlib",
+ }
+ }
+
+ /// Parses a compression codec name used by an Iceberg table property.
+ pub(crate) fn parse_property(value: &str) -> Result<Self> {
+
serde_json::from_value(serde_json::Value::String(value.to_lowercase())).map_err(|_|
{
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("Invalid compression codec: {value}"),
+ )
+ })
+ }
+
+ /// Parses the metadata-file compression codec table property.
+ pub(crate) fn parse_metadata_property(value: &str) -> Result<Self> {
+ if value.is_empty() {
+ return Ok(Self::None);
+ }
+
+ let codec = Self::parse_property(value).map_err(|_| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "Invalid metadata compression codec: {value}. Only '{}'
and '{}' are supported.",
+ Self::None.name(),
+ Self::gzip_default().name()
+ ),
+ )
+ })?;
+
+ match codec {
+ Self::None | Self::Gzip(_) => Ok(codec),
+ _ => Err(Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "Invalid metadata compression codec: {value}. Only '{}'
and '{}' are supported for metadata files.",
+ Self::None.name(),
+ Self::gzip_default().name()
+ ),
+ )),
+ }
+ }
+
+ /// Returns the codec name used by an Iceberg table property.
+ pub(crate) fn property_value(&self) -> String {
+ self.name().to_string()
Review Comment:
This round trip changes a valid Parquet property into an invalid one:
```
write.parquet.compression-codec=uncompressed
→ CompressionCodec::None
→ write.parquet.compression-codec=none
```
Parquet expects `uncompressed`, not `none`.
Should serialization preserve the format-specific value?
##########
crates/iceberg/src/spec/table_props.rs:
##########
@@ -0,0 +1,1129 @@
+// 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.
+//!
+//! [`TableProperties`] 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, TableProperties};
+//!
+//! let mut properties = TableProperties::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, TableProperties};
+//!
+//! let properties: TableProperties =
serde_json::from_value(serde_json::json!({
+//! "commit.retry.num-retries": "8",
+//! "write.format.default": "orc"
+//! })).unwrap();
+//!
+//! assert_eq!(properties.commit_retry_num_retries, 8);
+//! assert_eq!(properties.write_format_default, DataFileFormat::Orc);
+//! ```
+//!
+//! # Serialize to JSON
+//!
+//! Serialization converts non-default typed fields back into Iceberg property
keys. Fields whose
+//! values match their defaults are omitted:
+//!
+//! ```
+//! use iceberg::spec::TableProperties;
+//!
+//! let mut properties = TableProperties::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).unwrap();
+//! assert_eq!(json["commit.retry.num-retries"], "8");
+//! assert_eq!(json["write.data.path"], "s3://warehouse/table/data");
+//! assert!(json.get("write.format.default").is_none());
+//! ```
+
+use std::collections::HashMap;
+
+use iceberg_property_macro::Properties;
+use serde_with::{DeserializeFromStr, SerializeDisplay};
+
+use crate::compression::CompressionCodec;
+use crate::error::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";
+
+/// ORC compression strategy that prioritizes speed.
+pub const ORC_COMPRESSION_STRATEGY_SPEED: &str = "speed";
+
+/// ORC compression strategy that prioritizes compression ratio.
+pub const ORC_COMPRESSION_STRATEGY_COMPRESSION: &str = "compression";
+
+/// 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,
+}
+
+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(",")
+}
+
+/// 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 TableProperties {
+ // 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"]
+ #[additional_key = "write.manifest.compression-level"]
+ #[default(CompressionCodec::gzip_default())]
+ #[parse_properties_with(CompressionCodec::parse_properties)]
+ #[write_properties_with(CompressionCodec::write_properties)]
+ #[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,
Review Comment:
The `DataFileFormat` enum is broader than this property’s valid domain.
For example, the parser currently accepts `write.format.default=puffin`,
although only Parquet, Avro, and ORC are allowed.
The same applies to `CompressionCodec`, which permits invalid format/codec
combinations. Consider property-specific validation.
##########
crates/iceberg/src/spec/table_props.rs:
##########
@@ -0,0 +1,1129 @@
+// 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.
+//!
+//! [`TableProperties`] 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, TableProperties};
+//!
+//! let mut properties = TableProperties::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, TableProperties};
+//!
+//! let properties: TableProperties =
serde_json::from_value(serde_json::json!({
+//! "commit.retry.num-retries": "8",
+//! "write.format.default": "orc"
+//! })).unwrap();
+//!
+//! assert_eq!(properties.commit_retry_num_retries, 8);
+//! assert_eq!(properties.write_format_default, DataFileFormat::Orc);
+//! ```
+//!
+//! # Serialize to JSON
+//!
+//! Serialization converts non-default typed fields back into Iceberg property
keys. Fields whose
+//! values match their defaults are omitted:
+//!
+//! ```
+//! use iceberg::spec::TableProperties;
+//!
+//! let mut properties = TableProperties::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).unwrap();
+//! assert_eq!(json["commit.retry.num-retries"], "8");
+//! assert_eq!(json["write.data.path"], "s3://warehouse/table/data");
+//! assert!(json.get("write.format.default").is_none());
+//! ```
+
+use std::collections::HashMap;
+
+use iceberg_property_macro::Properties;
+use serde_with::{DeserializeFromStr, SerializeDisplay};
+
+use crate::compression::CompressionCodec;
+use crate::error::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";
+
+/// ORC compression strategy that prioritizes speed.
+pub const ORC_COMPRESSION_STRATEGY_SPEED: &str = "speed";
+
+/// ORC compression strategy that prioritizes compression ratio.
+pub const ORC_COMPRESSION_STRATEGY_COMPRESSION: &str = "compression";
+
+/// 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,
+}
+
+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(",")
+}
+
+/// 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 TableProperties {
+ // 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"]
+ #[additional_key = "write.manifest.compression-level"]
+ #[default(CompressionCodec::gzip_default())]
+ #[parse_properties_with(CompressionCodec::parse_properties)]
+ #[write_properties_with(CompressionCodec::write_properties)]
+ #[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,
Review Comment:
`write.delete.format.default` should inherit `write.format.default` when
absent, not always default to Parquet.
With only `write.format.default=orc`, the PR returns
`write_delete_format_default == Parquet`
[upstream
documents](https://github.com/apache/iceberg/blob/d8c10a1608170f0ba83be740d6ab0b6a3757cb3e/docs/docs/configuration.md#L45)
the default as “data file format.”
##########
crates/iceberg/src/spec/name_mapping/mod.rs:
##########
@@ -44,6 +48,23 @@ impl NameMapping {
}
}
+impl FromStr for NameMapping {
Review Comment:
nit this NameMapping change seems unrelated to the PR
##########
crates/iceberg/src/spec/table_props.rs:
##########
@@ -0,0 +1,1129 @@
+// 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.
+//!
+//! [`TableProperties`] 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, TableProperties};
+//!
+//! let mut properties = TableProperties::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, TableProperties};
+//!
+//! let properties: TableProperties =
serde_json::from_value(serde_json::json!({
+//! "commit.retry.num-retries": "8",
+//! "write.format.default": "orc"
+//! })).unwrap();
+//!
+//! assert_eq!(properties.commit_retry_num_retries, 8);
+//! assert_eq!(properties.write_format_default, DataFileFormat::Orc);
+//! ```
+//!
+//! # Serialize to JSON
+//!
+//! Serialization converts non-default typed fields back into Iceberg property
keys. Fields whose
+//! values match their defaults are omitted:
+//!
+//! ```
+//! use iceberg::spec::TableProperties;
+//!
+//! let mut properties = TableProperties::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).unwrap();
+//! assert_eq!(json["commit.retry.num-retries"], "8");
+//! assert_eq!(json["write.data.path"], "s3://warehouse/table/data");
+//! assert!(json.get("write.format.default").is_none());
+//! ```
+
+use std::collections::HashMap;
+
+use iceberg_property_macro::Properties;
+use serde_with::{DeserializeFromStr, SerializeDisplay};
+
+use crate::compression::CompressionCodec;
+use crate::error::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";
+
+/// ORC compression strategy that prioritizes speed.
+pub const ORC_COMPRESSION_STRATEGY_SPEED: &str = "speed";
+
+/// ORC compression strategy that prioritizes compression ratio.
+pub const ORC_COMPRESSION_STRATEGY_COMPRESSION: &str = "compression";
+
+/// 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,
+}
+
+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(",")
+}
+
+/// 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 TableProperties {
+ // 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"]
+ #[additional_key = "write.manifest.compression-level"]
+ #[default(CompressionCodec::gzip_default())]
+ #[parse_properties_with(CompressionCodec::parse_properties)]
+ #[write_properties_with(CompressionCodec::write_properties)]
+ #[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(DistributionMode::None)]
Review Comment:
Iceberg leaves these [distribution properties
unset](https://github.com/apache/iceberg/blob/46f72f7a0b6ce7901be1db90ae708dccad2fd414/docs/docs/configuration.md#L78)
so engines can choose their defaults.
Using `DistributionMode::None` conflates absence with an explicit `"none"`
and imposes behavior that is not standardized. Consider
`Option<DistributionMode>` to preserve that distinction.
##########
crates/iceberg/src/spec/table_metadata.rs:
##########
@@ -384,13 +384,13 @@ impl TableMetadata {
///
/// Returns an error if the compression codec property has an invalid
value.
pub fn metadata_compression_codec(&self) -> Result<CompressionCodec> {
- parse_metadata_file_compression(&self.properties)
+ Ok(self.table_properties()?.write_metadata_compression_codec)
Review Comment:
`self.table_properties()?` parses every modeled property.
`parse_metadata_file_compression` only parsed what it needs.
An invalid unrelated value can therefore make this accessor fail. Parse only
the required property here, or make typed access lazy.
##########
crates/iceberg/src/spec/table_properties.rs:
##########
Review Comment:
This refactor changes existing behavior and drops several regression tests
from `table_properties.rs`.
Behavior lost or changed:
- `write.metadata.path` no longer rejects empty values or trims trailing
slashes.
- Every call to `table_properties()` eagerly parses the expanded set of
modeled properties, so an invalid unrelated property can make an operation
fail. Some previous paths, such as metadata compression, parsed only the
required key.
- Metadata compression now accepts `uncompressed`; the previous allowlist
was `""`, `none`, and `gzip`.
Missing regression coverage:
- Empty and trailing-slash metadata paths.
- Metadata-compression defaults, case-insensitive values, and invalid codecs.
- Invalid numeric and boolean values.
- CDC defaults, partial overrides, negative normalization levels, and
parsing errors.
Could we preserve the previous behavior and port these regression tests as
part of the refactor?
--
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]
