feat(transaction): Implement TransactionAction for updata_loc, update_props, and upgrade_format (#1433)

## Which issue does this PR close?

Related Issues:
- #1382 [EPIC]
- #1386 
- #1387 
- #1388 
- #1389

## What changes are included in this PR?


- Added implementations of `TransactionAction`
  - `UpdateLocationAction`
  - `UpdatePropertiesAction`
  - `UpgradeFormatVersionAction`
- Added `as_any` to `TransactionAction` trait
- Added `do_commit` in `Transaction` to commit applied actions

We will also need to implement `TransactionAction` for
`FastAppendAction` and `ReplaceSortOrderAction`, and I'm planning to do
that in a separate PR.

## Are these changes tested?
Added unit tests


---------

Co-authored-by: Renjie Liu <liurenjie2008@gmail.com>

Author: CTTY

Cargo.lock

@@ -50,6 +50,7 @@ arrow-ord = { version = "55" }
 arrow-schema = { version = "55" }
 arrow-select = { version = "55" }
 arrow-string = { version = "55" }
+as-any = "0.3.2"
 async-std = "1.12"
 async-trait = "0.1.88"
 aws-config = "1.6.1"

Cargo.toml

@@ -802,7 +802,7 @@ mod tests {
         SnapshotLog, SortDirection, SortField, SortOrder, Summary, Transform, Type,
         UnboundPartitionField, UnboundPartitionSpec,
     };
-    use iceberg::transaction::Transaction;
+    use iceberg::transaction::{ApplyTransactionAction, Transaction};
     use mockito::{Mock, Server, ServerGuard};
     use serde_json::json;
     use uuid::uuid;
@@ -2093,6 +2093,17 @@ mod tests {
 
         let config_mock = create_config_mock(&mut server).await;
 
+        let load_table_mock = server
+            .mock("GET", "/v1/namespaces/ns1/tables/test1")
+            .with_status(200)
+            .with_body_from_file(format!(
+                "{}/testdata/{}",
+                env!("CARGO_MANIFEST_DIR"),
+                "load_table_response.json"
+            ))
+            .create_async()
+            .await;
+
         let update_table_mock = server
             .mock("POST", "/v1/namespaces/ns1/tables/test1")
             .with_status(200)
@@ -2125,8 +2136,11 @@ mod tests {
                 .unwrap()
         };
 
-        let table = Transaction::new(&table1)
-            .upgrade_table_version(FormatVersion::V2)
+        let tx = Transaction::new(&table1);
+        let table = tx
+            .upgrade_table_version()
+            .set_format_version(FormatVersion::V2)
+            .apply(tx)
             .unwrap()
             .commit(&catalog)
             .await
@@ -2204,6 +2218,7 @@ mod tests {
 
         config_mock.assert_async().await;
         update_table_mock.assert_async().await;
+        load_table_mock.assert_async().await
     }
 
     #[tokio::test]
@@ -2212,6 +2227,17 @@ mod tests {
 
         let config_mock = create_config_mock(&mut server).await;
 
+        let load_table_mock = server
+            .mock("GET", "/v1/namespaces/ns1/tables/test1")
+            .with_status(200)
+            .with_body_from_file(format!(
+                "{}/testdata/{}",
+                env!("CARGO_MANIFEST_DIR"),
+                "load_table_response.json"
+            ))
+            .create_async()
+            .await;
+
         let update_table_mock = server
             .mock("POST", "/v1/namespaces/ns1/tables/test1")
             .with_status(404)
@@ -2250,8 +2276,11 @@ mod tests {
                 .unwrap()
         };
 
-        let table_result = Transaction::new(&table1)
-            .upgrade_table_version(FormatVersion::V2)
+        let tx = Transaction::new(&table1);
+        let table_result = tx
+            .upgrade_table_version()
+            .set_format_version(FormatVersion::V2)
+            .apply(tx)
             .unwrap()
             .commit(&catalog)
             .await;
@@ -2267,5 +2296,6 @@ mod tests {
 
         config_mock.assert_async().await;
         update_table_mock.assert_async().await;
+        load_table_mock.assert_async().await;
     }
 }

crates/catalog/rest/src/catalog.rs

@@ -23,7 +23,7 @@ use std::sync::RwLock;
 
 use ctor::{ctor, dtor};
 use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-use iceberg::transaction::Transaction;
+use iceberg::transaction::{ApplyTransactionAction, Transaction};
 use iceberg::{Catalog, Namespace, NamespaceIdent, TableCreation, TableIdent};
 use iceberg_catalog_rest::{RestCatalog, RestCatalogConfig};
 use iceberg_test_utils::docker::DockerCompose;
@@ -346,9 +346,12 @@ async fn test_update_table() {
         &TableIdent::new(ns.name().clone(), "t1".to_string())
     );
 
+    let tx = Transaction::new(&table);
     // Update table by committing transaction
-    let table2 = Transaction::new(&table)
-        .set_properties(HashMap::from([("prop1".to_string(), "v1".to_string())]))
+    let table2 = tx
+        .update_table_properties()
+        .set("prop1".to_string(), "v1".to_string())
+        .apply(tx)
         .unwrap()
         .commit(&catalog)
         .await

crates/catalog/rest/tests/rest_catalog_test.rs

@@ -54,6 +54,7 @@ arrow-ord = { workspace = true }
 arrow-schema = { workspace = true }
 arrow-select = { workspace = true }
 arrow-string = { workspace = true }
+as-any = { workspace = true }
 async-std = { workspace = true, optional = true, features = ["attributes"] }
 async-trait = { workspace = true }
 base64 = { workspace = true }

crates/iceberg/Cargo.toml

@@ -15,26 +15,26 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#![allow(dead_code)]
 use std::mem::take;
 use std::sync::Arc;
 
+use as_any::AsAny;
 use async_trait::async_trait;
 
 use crate::table::Table;
 use crate::transaction::Transaction;
 use crate::{Result, TableRequirement, TableUpdate};
 
 /// A boxed, thread-safe reference to a `TransactionAction`.
-pub type BoxedTransactionAction = Arc<dyn TransactionAction>;
+pub(crate) type BoxedTransactionAction = Arc<dyn TransactionAction>;
 
 /// A trait representing an atomic action that can be part of a transaction.
 ///
 /// Implementors of this trait define how a specific action is committed to a table.
 /// Each action is responsible for generating the updates and requirements needed
 /// to modify the table metadata.
 #[async_trait]
-pub(crate) trait TransactionAction: Sync + Send {
+pub(crate) trait TransactionAction: AsAny + Sync + Send {
     /// Commits this action against the provided table and returns the resulting updates.
     /// NOTE: This function is intended for internal use only and should not be called directly by users.
     ///
@@ -108,6 +108,7 @@ mod tests {
     use std::str::FromStr;
     use std::sync::Arc;
 
+    use as_any::Downcast;
     use async_trait::async_trait;
     use uuid::Uuid;
 
@@ -158,9 +159,12 @@ mod tests {
         let tx = Transaction::new(&table);
 
         let updated_tx = action.apply(tx).unwrap();
-
         // There should be one action in the transaction now
         assert_eq!(updated_tx.actions.len(), 1);
+
+        (*updated_tx.actions[0])
+            .downcast_ref::<TestAction>()
+            .expect("TestAction was not applied to Transaction!");
     }
 
     #[test]