feat: Introduce snapshot summary properties (#1336)

## Which issue does this PR close?

- Closes #1329

TLDR:
- I want to add customized metadata (with size controlled within 200B)
for each snapshot, summary properties is the perfect place for this
feature request

## What changes are included in this PR?

I add a new table update action for snapshot summary properties.

## Are these changes tested?

Yes, unit tests are added.

Author: dentiny

crates/iceberg/src/catalog/mod.rs

@@ -495,6 +495,12 @@ pub enum TableUpdate {
         /// Schema IDs to remove.
         schema_ids: Vec<i32>,
     },
+    /// Add snapshot summary properties.
+    #[serde(rename_all = "kebab-case")]
+    AddSnapshotSummaryProperties {
+        /// Additional properties to add.
+        properties: HashMap<String, String>,
+    },
 }
 
 impl TableUpdate {
@@ -539,6 +545,9 @@ impl TableUpdate {
                 Ok(builder.remove_partition_statistics(snapshot_id))
             }
             TableUpdate::RemoveSchemas { schema_ids } => builder.remove_schemas(&schema_ids),
+            TableUpdate::AddSnapshotSummaryProperties { properties } => {
+                builder.add_snapshot_summary_properties(properties)
+            }
         }
     }
 }
@@ -2098,4 +2107,24 @@ mod tests {
             },
         );
     }
+
+    #[test]
+    fn test_add_snapshot_summary_properties() {
+        let mut expected_properties = HashMap::new();
+        expected_properties.insert("prop-key".to_string(), "prop-value".to_string());
+
+        test_serde_json(
+            r#"
+                {
+                    "action": "add-snapshot-summary-properties",
+                    "properties": {
+                            "prop-key": "prop-value"
+                    }
+                }        
+            "#,
+            TableUpdate::AddSnapshotSummaryProperties {
+                properties: expected_properties,
+            },
+        );
+    }
 }

crates/iceberg/src/spec/snapshot.rs

@@ -214,6 +214,11 @@ impl Snapshot {
             snapshot_id: self.snapshot_id,
         }
     }
+
+    /// Add the given properties map to snapshot summary.
+    pub(crate) fn add_summary_properties(&mut self, props: HashMap<String, String>) {
+        self.summary.additional_properties.extend(props);
+    }
 }
 
 pub(super) mod _serde {

crates/iceberg/src/spec/table_metadata_builder.rs

@@ -1244,6 +1244,33 @@ impl TableMetadataBuilder {
 
         Ok(self)
     }
+
+    /// Add summary properties to the latest snapshot for the table metadata.
+    pub fn add_snapshot_summary_properties(
+        mut self,
+        properties: HashMap<String, String>,
+    ) -> Result<Self> {
+        if properties.is_empty() {
+            return Ok(self);
+        }
+
+        let snapshot_id = self.metadata.current_snapshot_id.unwrap();
+        let mut cur_snapshot = self
+            .metadata
+            .snapshots
+            .remove(&snapshot_id)
+            .unwrap()
+            .as_ref()
+            .clone();
+        cur_snapshot.add_summary_properties(properties.clone());
+        self.metadata
+            .snapshots
+            .insert(snapshot_id, Arc::new(cur_snapshot));
+        self.changes
+            .push(TableUpdate::AddSnapshotSummaryProperties { properties });
+
+        Ok(self)
+    }
 }
 
 impl From<TableMetadataBuildResult> for TableMetadata {
@@ -2496,4 +2523,51 @@ mod tests {
             };
         assert_eq!(remove_schema_ids, &[0]);
     }
+
+    #[test]
+    fn test_add_snapshot_summary_properties() {
+        let file = File::open(format!(
+            "{}/testdata/table_metadata/{}",
+            env!("CARGO_MANIFEST_DIR"),
+            "TableMetadataV2Valid.json"
+        ))
+        .unwrap();
+        let reader = BufReader::new(file);
+        let resp = serde_json::from_reader::<_, TableMetadata>(reader).unwrap();
+
+        let table = Table::builder()
+            .metadata(resp)
+            .metadata_location("s3://bucket/test/location/metadata/v1.json".to_string())
+            .identifier(TableIdent::from_strs(["ns1", "test1"]).unwrap())
+            .file_io(FileIOBuilder::new("memory").build().unwrap())
+            .build()
+            .unwrap();
+        assert!(
+            table
+                .metadata()
+                .current_snapshot()
+                .unwrap()
+                .summary()
+                .additional_properties
+                .is_empty()
+        );
+
+        let mut new_properties = HashMap::new();
+        new_properties.insert("prop-key".to_string(), "prop-value".to_string());
+
+        let mut meta_data_builder = table.metadata().clone().into_builder(None);
+        meta_data_builder = meta_data_builder
+            .add_snapshot_summary_properties(new_properties.clone())
+            .unwrap();
+        let build_result = meta_data_builder.build().unwrap();
+        assert_eq!(
+            build_result
+                .metadata
+                .current_snapshot()
+                .unwrap()
+                .summary()
+                .additional_properties,
+            new_properties
+        );
+    }
 }

crates/iceberg/src/transaction/mod.rs

@@ -128,6 +128,18 @@ impl<'a> Transaction<'a> {
         Ok(self)
     }
 
+    /// Add snapshot summary properties.
+    pub fn add_snapshot_summary_properties(
+        mut self,
+        props: HashMap<String, String>,
+    ) -> Result<Self> {
+        self.apply(
+            vec![TableUpdate::AddSnapshotSummaryProperties { properties: props }],
+            vec![],
+        )?;
+        Ok(self)
+    }
+
     fn generate_unique_snapshot_id(&self) -> i64 {
         let generate_random_id = || -> i64 {
             let (lhs, rhs) = Uuid::new_v4().as_u64_pair();