refactor: parse table properties lazily

crates/iceberg/src/catalog/metadata_location.rs

@@ -15,14 +15,13 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::collections::HashMap;
 use std::fmt::Display;
 use std::str::FromStr;
 
 use uuid::Uuid;
 
 use crate::compression::CompressionCodec;
-use crate::spec::{TableMetadata, parse_metadata_file_compression};
+use crate::spec::TableMetadata;
 use crate::{Error, ErrorKind, Result};
 
 /// Default folder name for metadata files under the table location, used when the
@@ -43,12 +42,6 @@ pub struct MetadataLocation {
 }
 
 impl MetadataLocation {
-    /// Determines the compression codec from table properties.
-    /// Parse errors result in CompressionCodec::None.
-    fn compression_from_properties(properties: &HashMap<String, String>) -> CompressionCodec {
-        parse_metadata_file_compression(properties).unwrap_or(CompressionCodec::None)
-    }
-
     /// Creates a completely new metadata location starting at version 0, deriving the
     /// metadata directory and compression settings from the table metadata.
     /// Only used for creating a new table. For updates, see `with_next_version` and
@@ -58,7 +51,7 @@ impl MetadataLocation {
             location: metadata.metadata_location()?,
             version: 0,
             id: Uuid::new_v4(),
-            compression_codec: Self::compression_from_properties(metadata.properties()),
+            compression_codec: metadata.metadata_compression_codec()?,
         })
     }
 
@@ -80,7 +73,7 @@ impl MetadataLocation {
             location: new_metadata.metadata_location()?,
             version: self.version,
             id: self.id,
-            compression_codec: Self::compression_from_properties(new_metadata.properties()),
+            compression_codec: new_metadata.metadata_compression_codec()?,
         })
     }
 

crates/iceberg/src/catalog/utils.rs

@@ -61,7 +61,7 @@ pub async fn drop_table_data(table_info: &Table) -> Result<()> {
     }
 
     // Delete data files only if gc.enabled is true, to avoid corrupting shared tables
-    if metadata.table_properties()?.gc_enabled() {
+    if metadata.table_properties().gc_enabled()? {
         delete_data_files(io, &manifests_to_delete).await?;
     }
 

crates/iceberg/src/encryption/manager.rs

@@ -113,8 +113,9 @@ impl EncryptionManager {
             return Ok(None);
         }
 
-        let table_properties = metadata.table_properties()?;
-        let Some(table_key_id) = table_properties.encryption_key_id().as_deref() else {
+        let table_properties = metadata.table_properties();
+        let encryption_key_id = table_properties.encryption_key_id()?;
+        let Some(table_key_id) = encryption_key_id.as_deref() else {
             if kms_client.is_some() {
                 tracing::warn!(
                     "KeyManagementClient provided but table does not have encryption.key-id set"

crates/iceberg/src/spec/mod.rs

@@ -50,7 +50,6 @@ pub use sort::*;
 pub use statistic_file::*;
 pub use table_metadata::*;
 pub(crate) use table_metadata_builder::FIRST_FIELD_ID;
-pub(crate) use table_properties::parse_metadata_file_compression;
 pub use table_properties::*;
 pub use transform::*;
 pub(crate) use values::decimal_utils;

crates/iceberg/src/spec/table_metadata.rs

@@ -35,7 +35,7 @@ pub use super::table_metadata_builder::{TableMetadataBuildResult, TableMetadataB
 use super::{
     DEFAULT_PARTITION_SPEC_ID, PartitionSpecRef, PartitionStatisticsFile, SchemaId, SchemaRef,
     SnapshotRef, SnapshotRetention, SortOrder, SortOrderRef, StatisticsFile, StructType,
-    TableProperties, parse_metadata_file_compression,
+    TableProperties,
 };
 use crate::catalog::{METADATA_FOLDER_NAME, MetadataLocation};
 use crate::compression::CompressionCodec;
@@ -370,9 +370,8 @@ impl TableMetadata {
     /// to the `metadata` subdirectory under the table location.
     pub fn metadata_location(&self) -> Result<String> {
         Ok(self
-            .table_properties()?
-            .write_metadata_path()
-            .clone()
+            .table_properties()
+            .write_metadata_path()?
             .unwrap_or_else(|| format!("{}/{}", self.location(), METADATA_FOLDER_NAME)))
     }
 
@@ -385,14 +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)
+        self.table_properties().metadata_compression_codec()
     }
 
-    /// Returns typed table properties parsed from the raw properties map with defaults.
-    pub fn table_properties(&self) -> Result<TableProperties> {
-        TableProperties::try_from(&self.properties).map_err(|e| {
-            Error::new(ErrorKind::DataInvalid, "Invalid table properties").with_source(e)
-        })
+    /// Returns a typed view that parses each table property when its getter is called.
+    #[inline]
+    pub fn table_properties(&self) -> TableProperties<'_> {
+        TableProperties::new(&self.properties)
     }
 
     /// Return location of statistics files.
@@ -499,7 +497,7 @@ impl TableMetadata {
         let json_data = serde_json::to_vec(self)?;
 
         // Check if compression codec from properties matches the one in metadata_location
-        let codec = parse_metadata_file_compression(&self.properties)?;
+        let codec = self.table_properties().metadata_compression_codec()?;
 
         if codec != metadata_location.compression_codec() {
             return Err(Error::new(
@@ -4041,14 +4039,14 @@ mod tests {
         .unwrap()
         .metadata;
 
-        let props = metadata.table_properties().unwrap();
+        let props = metadata.table_properties();
 
         assert_eq!(
-            props.commit_num_retries(),
+            props.commit_num_retries().unwrap(),
             TableProperties::PROPERTY_COMMIT_NUM_RETRIES_DEFAULT
         );
         assert_eq!(
-            props.write_target_file_size_bytes(),
+            props.write_target_file_size_bytes().unwrap(),
             TableProperties::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT
         );
     }
@@ -4088,10 +4086,67 @@ mod tests {
         .unwrap()
         .metadata;
 
-        let props = metadata.table_properties().unwrap();
+        let props = metadata.table_properties();
+
+        assert_eq!(props.commit_num_retries().unwrap(), 10);
+        assert_eq!(props.write_target_file_size_bytes().unwrap(), 1024);
+    }
+
+    #[test]
+    fn test_deserialize_metadata_defers_invalid_table_property_errors() {
+        let invalid_retries = "not_a_number";
+        let invalid_codec = "unknown";
+        let target_file_size = "1024";
+
+        for file_name in [
+            "TableMetadataV1Valid.json",
+            "TableMetadataV2ValidMinimal.json",
+            "TableMetadataV3ValidMinimal.json",
+        ] {
+            let path = format!("testdata/table_metadata/{file_name}");
+            let mut json: serde_json::Value =
+                serde_json::from_str(&fs::read_to_string(path).unwrap()).unwrap();
+            json["properties"] = serde_json::json!({
+                (TableProperties::PROPERTY_COMMIT_NUM_RETRIES): invalid_retries,
+                (TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC): invalid_codec,
+                (TableProperties::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES): target_file_size,
+            });
+
+            let metadata: TableMetadata = serde_json::from_value(json).unwrap();
+            assert_eq!(
+                metadata
+                    .properties()
+                    .get(TableProperties::PROPERTY_COMMIT_NUM_RETRIES)
+                    .map(String::as_str),
+                Some(invalid_retries)
+            );
+
+            let table_properties = metadata.table_properties();
+            let error = table_properties.commit_num_retries().unwrap_err();
+            assert!(
+                error
+                    .message()
+                    .contains(TableProperties::PROPERTY_COMMIT_NUM_RETRIES)
+            );
+            assert_eq!(
+                table_properties.write_target_file_size_bytes().unwrap(),
+                1024
+            );
+            let error = table_properties.metadata_compression_codec().unwrap_err();
+            assert!(
+                format!("{error}").contains(TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC)
+            );
 
-        assert_eq!(props.commit_num_retries(), 10);
-        assert_eq!(props.write_target_file_size_bytes(), 1024);
+            let serialized = serde_json::to_value(metadata).unwrap();
+            assert_eq!(
+                serialized["properties"][TableProperties::PROPERTY_COMMIT_NUM_RETRIES],
+                invalid_retries
+            );
+            assert_eq!(
+                serialized["properties"][TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC],
+                invalid_codec
+            );
+        }
     }
 
     #[test]
@@ -4103,10 +4158,16 @@ mod tests {
             .build()
             .unwrap();
 
-        let properties = HashMap::from([(
-            "commit.retry.num-retries".to_string(),
-            "not_a_number".to_string(),
-        )]);
+        let properties = HashMap::from([
+            (
+                TableProperties::PROPERTY_COMMIT_NUM_RETRIES.to_string(),
+                "not_a_number".to_string(),
+            ),
+            (
+                TableProperties::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES.to_string(),
+                "1024".to_string(),
+            ),
+        ]);
 
         let metadata = TableMetadataBuilder::new(
             schema,
@@ -4121,9 +4182,17 @@ mod tests {
         .unwrap()
         .metadata;
 
-        let err = metadata.table_properties().unwrap_err();
+        let table_properties = metadata.table_properties();
+        let err = table_properties.commit_num_retries().unwrap_err();
         assert_eq!(err.kind(), ErrorKind::DataInvalid);
-        assert!(err.message().contains("Invalid table properties"));
+        assert!(
+            err.message()
+                .contains(TableProperties::PROPERTY_COMMIT_NUM_RETRIES)
+        );
+        assert_eq!(
+            table_properties.write_target_file_size_bytes().unwrap(),
+            1024
+        );
     }
 
     #[test]

crates/iceberg/src/spec/table_metadata_builder.rs

@@ -2095,6 +2095,32 @@ mod tests {
         assert_eq!(build_result.metadata.metadata_log.len(), 0);
     }
 
+    #[test]
+    fn test_table_properties_view_reflects_metadata_updates() {
+        let property = TableProperties::PROPERTY_COMMIT_NUM_RETRIES.to_string();
+        let metadata = builder_without_changes(FormatVersion::V2)
+            .set_properties(HashMap::from([(property.clone(), "7".to_string())]))
+            .unwrap()
+            .build()
+            .unwrap()
+            .metadata;
+
+        assert_eq!(metadata.table_properties().commit_num_retries().unwrap(), 7);
+
+        let metadata = metadata
+            .into_builder(None)
+            .remove_properties(&[property])
+            .unwrap()
+            .build()
+            .unwrap()
+            .metadata;
+
+        assert_eq!(
+            metadata.table_properties().commit_num_retries().unwrap(),
+            TableProperties::PROPERTY_COMMIT_NUM_RETRIES_DEFAULT
+        );
+    }
+
     #[test]
     fn test_no_metadata_log_entry_for_no_previous_location() {
         // Used for first commit after stage-creation of tables