Merge branch 'main' into ctty/tx-commit

Author: CTTY

Cargo.lock

@@ -158,12 +158,18 @@ impl HttpClient {
                 .map(|(k, v)| (k.as_str(), v.as_str())),
         );
 
-        let auth_req = self
+        let mut auth_req = self
             .request(Method::POST, &self.token_endpoint)
             .form(&params)
             .build()?;
+        // extra headers add content-type application/json header it's necessary to override it with proper type
+        // note that form call doesn't add content-type header if already present
+        auth_req.headers_mut().insert(
+            http::header::CONTENT_TYPE,
+            http::HeaderValue::from_static("application/x-www-form-urlencoded"),
+        );
         let auth_url = auth_req.url().clone();
-        let auth_resp = self.execute(auth_req).await?;
+        let auth_resp = self.client.execute(auth_req).await?;
 
         let auth_res: TokenResponse = if auth_resp.status() == StatusCode::OK {
             let text = auth_resp

crates/catalog/rest/src/client.rs

@@ -72,6 +72,16 @@ impl<'a> FastAppendAction<'a> {
         Ok(self)
     }
 
+    /// Set snapshot summary properties.
+    pub fn set_snapshot_properties(
+        &mut self,
+        snapshot_properties: HashMap<String, String>,
+    ) -> Result<&mut Self> {
+        self.snapshot_produce_action
+            .set_snapshot_properties(snapshot_properties)?;
+        Ok(self)
+    }
+
     /// Adds existing parquet files
     ///
     /// Note: This API is not yet fully supported in version 0.5.x.  
@@ -211,6 +221,8 @@ impl SnapshotProduceOperation for FastAppendOperation {
 
 #[cfg(test)]
 mod tests {
+    use std::collections::HashMap;
+
     use crate::scan::tests::TableTestFixture;
     use crate::spec::{
         DataContentType, DataFileBuilder, DataFileFormat, Literal, MAIN_BRANCH, Struct,
@@ -228,6 +240,44 @@ mod tests {
         assert!(action.apply().await.is_err());
     }
 
+    #[tokio::test]
+    async fn test_set_snapshot_properties() {
+        let table = make_v2_minimal_table();
+        let tx = Transaction::new(&table);
+        let mut action = tx.fast_append(None, vec![]).unwrap();
+
+        let mut snapshot_properties = HashMap::new();
+        snapshot_properties.insert("key".to_string(), "val".to_string());
+        action.set_snapshot_properties(snapshot_properties).unwrap();
+        let data_file = DataFileBuilder::default()
+            .content(DataContentType::Data)
+            .file_path("test/1.parquet".to_string())
+            .file_format(DataFileFormat::Parquet)
+            .file_size_in_bytes(100)
+            .record_count(1)
+            .partition_spec_id(table.metadata().default_partition_spec_id())
+            .partition(Struct::from_iter([Some(Literal::long(300))]))
+            .build()
+            .unwrap();
+        action.add_data_files(vec![data_file]).unwrap();
+        let tx = action.apply().await.unwrap();
+
+        // Check customized properties is contained in snapshot summary properties.
+        let new_snapshot = if let TableUpdate::AddSnapshot { snapshot } = &tx.updates[0] {
+            snapshot
+        } else {
+            unreachable!()
+        };
+        assert_eq!(
+            new_snapshot
+                .summary()
+                .additional_properties
+                .get("key")
+                .unwrap(),
+            "val"
+        );
+    }
+
     #[tokio::test]
     async fn test_fast_append_action() {
         let table = make_v2_minimal_table();

crates/iceberg/src/transaction/append.rs

@@ -122,6 +122,15 @@ impl<'a> SnapshotProduceAction<'a> {
         Ok(())
     }
 
+    /// Set snapshot summary properties.
+    pub fn set_snapshot_properties(
+        &mut self,
+        snapshot_properties: HashMap<String, String>,
+    ) -> Result<&mut Self> {
+        self.snapshot_properties = snapshot_properties;
+        Ok(self)
+    }
+
     /// Add data files to the snapshot.
     pub fn add_data_files(
         &mut self,