feat(transaction): Make Transaction own base_table

crates/catalog/rest/src/catalog.rs

@@ -2125,7 +2125,7 @@ mod tests {
                 .unwrap()
         };
 
-        let table = Transaction::new(&table1)
+        let table = Transaction::new(table1)
             .upgrade_table_version(FormatVersion::V2)
             .unwrap()
             .commit(&catalog)
@@ -2250,7 +2250,7 @@ mod tests {
                 .unwrap()
         };
 
-        let table_result = Transaction::new(&table1)
+        let table_result = Transaction::new(table1)
             .upgrade_table_version(FormatVersion::V2)
             .unwrap()
             .commit(&catalog)

crates/catalog/rest/tests/rest_catalog_test.rs

@@ -347,7 +347,7 @@ async fn test_update_table() {
     );
 
     // Update table by committing transaction
-    let table2 = Transaction::new(&table)
+    let table2 = Transaction::new(table)
         .set_properties(HashMap::from([("prop1".to_string(), "v1".to_string())]))
         .unwrap()
         .commit(&catalog)

crates/iceberg/src/transaction/append.rs

@@ -29,15 +29,15 @@ use crate::writer::file_writer::ParquetWriter;
 use crate::{Error, ErrorKind};
 
 /// FastAppendAction is a transaction action for fast append data files to the table.
-pub struct FastAppendAction<'a> {
-    snapshot_produce_action: SnapshotProduceAction<'a>,
+pub struct FastAppendAction {
+    snapshot_produce_action: SnapshotProduceAction,
     check_duplicate: bool,
 }
 
-impl<'a> FastAppendAction<'a> {
+impl FastAppendAction {
     #[allow(clippy::too_many_arguments)]
     pub(crate) fn new(
-        tx: Transaction<'a>,
+        tx: Transaction,
         snapshot_id: i64,
         commit_uuid: Uuid,
         key_metadata: Vec<u8>,
@@ -87,7 +87,7 @@ impl<'a> FastAppendAction<'a> {
     /// Specifically, schema compatibility checks and support for adding to partitioned tables
     /// have not yet been implemented.
     #[allow(dead_code)]
-    async fn add_parquet_files(mut self, file_path: Vec<String>) -> Result<Transaction<'a>> {
+    async fn add_parquet_files(mut self, file_path: Vec<String>) -> Result<Transaction> {
         if !self
             .snapshot_produce_action
             .tx
@@ -117,7 +117,7 @@ impl<'a> FastAppendAction<'a> {
     }
 
     /// Finished building the action and apply it to the transaction.
-    pub async fn apply(self) -> Result<Transaction<'a>> {
+    pub async fn apply(self) -> Result<Transaction> {
         // Checks duplicate files
         if self.check_duplicate {
             let new_files: HashSet<&str> = self
@@ -170,14 +170,14 @@ impl SnapshotProduceOperation for FastAppendOperation {
 
     async fn delete_entries(
         &self,
-        _snapshot_produce: &SnapshotProduceAction<'_>,
+        _snapshot_produce: &SnapshotProduceAction,
     ) -> Result<Vec<ManifestEntry>> {
         Ok(vec![])
     }
 
     async fn existing_manifest(
         &self,
-        snapshot_produce: &SnapshotProduceAction<'_>,
+        snapshot_produce: &SnapshotProduceAction,
     ) -> Result<Vec<ManifestFile>> {
         let Some(snapshot) = snapshot_produce
             .tx
@@ -219,7 +219,7 @@ mod tests {
     #[tokio::test]
     async fn test_empty_data_append_action() {
         let table = make_v2_minimal_table();
-        let tx = Transaction::new(&table);
+        let tx = Transaction::new(table);
         let mut action = tx.fast_append(None, vec![]).unwrap();
         action.add_data_files(vec![]).unwrap();
         assert!(action.apply().await.is_err());
@@ -228,7 +228,7 @@ mod tests {
     #[tokio::test]
     async fn test_set_snapshot_properties() {
         let table = make_v2_minimal_table();
-        let tx = Transaction::new(&table);
+        let tx = Transaction::new(table.clone());
         let mut action = tx.fast_append(None, vec![]).unwrap();
 
         let mut snapshot_properties = HashMap::new();
@@ -266,7 +266,7 @@ mod tests {
     #[tokio::test]
     async fn test_fast_append_action() {
         let table = make_v2_minimal_table();
-        let tx = Transaction::new(&table);
+        let tx = Transaction::new(table.clone());
         let mut action = tx.fast_append(None, vec![]).unwrap();
 
         // check add data file with incompatible partition value
@@ -352,7 +352,7 @@ mod tests {
     async fn test_add_duplicated_parquet_files_to_unpartitioned_table() {
         let mut fixture = TableTestFixture::new_unpartitioned();
         fixture.setup_unpartitioned_manifest_files().await;
-        let tx = crate::transaction::Transaction::new(&fixture.table);
+        let tx = crate::transaction::Transaction::new(fixture.table.clone());
 
         let file_paths = vec![
             format!("{}/1.parquet", &fixture.table_location),
@@ -372,7 +372,7 @@ mod tests {
 
         let file_paths = vec![format!("{}/2.parquet", &fixture.table_location)];
 
-        let tx = crate::transaction::Transaction::new(&fixture.table);
+        let tx = crate::transaction::Transaction::new(fixture.table.clone());
         let fast_append_action = tx.fast_append(None, vec![]).unwrap();
 
         // Attempt to add Parquet file which was deleted from table.

crates/iceberg/src/transaction/mod.rs

@@ -37,18 +37,18 @@ use crate::transaction::sort_order::ReplaceSortOrderAction;
 use crate::{Catalog, Error, ErrorKind, TableCommit, TableRequirement, TableUpdate};
 
 /// Table transaction.
-pub struct Transaction<'a> {
-    base_table: &'a Table,
+pub struct Transaction {
+    base_table: Table,
     current_table: Table,
     updates: Vec<TableUpdate>,
     requirements: Vec<TableRequirement>,
 }
 
-impl<'a> Transaction<'a> {
+impl Transaction {
     /// Creates a new transaction.
-    pub fn new(table: &'a Table) -> Self {
+    pub fn new(table: Table) -> Self {
         Self {
-            base_table: table,
+            base_table: table.clone(),
             current_table: table.clone(),
             updates: vec![],
             requirements: vec![],
@@ -155,7 +155,7 @@ impl<'a> Transaction<'a> {
         self,
         commit_uuid: Option<Uuid>,
         key_metadata: Vec<u8>,
-    ) -> Result<FastAppendAction<'a>> {
+    ) -> Result<FastAppendAction> {
         let snapshot_id = self.generate_unique_snapshot_id();
         FastAppendAction::new(
             self,
@@ -167,7 +167,7 @@ impl<'a> Transaction<'a> {
     }
 
     /// Creates replace sort order action.
-    pub fn replace_sort_order(self) -> ReplaceSortOrderAction<'a> {
+    pub fn replace_sort_order(self) -> ReplaceSortOrderAction {
         ReplaceSortOrderAction {
             tx: self,
             sort_fields: vec![],
@@ -273,7 +273,7 @@ mod tests {
     #[test]
     fn test_upgrade_table_version_v1_to_v2() {
         let table = make_v1_table();
-        let tx = Transaction::new(&table);
+        let tx = Transaction::new(table);
         let tx = tx.upgrade_table_version(FormatVersion::V2).unwrap();
 
         assert_eq!(
@@ -287,7 +287,7 @@ mod tests {
     #[test]
     fn test_upgrade_table_version_v2_to_v2() {
         let table = make_v2_table();
-        let tx = Transaction::new(&table);
+        let tx = Transaction::new(table);
         let tx = tx.upgrade_table_version(FormatVersion::V2).unwrap();
 
         assert!(
@@ -303,7 +303,7 @@ mod tests {
     #[test]
     fn test_downgrade_table_version() {
         let table = make_v2_table();
-        let tx = Transaction::new(&table);
+        let tx = Transaction::new(table);
         let tx = tx.upgrade_table_version(FormatVersion::V1);
 
         assert!(tx.is_err(), "Downgrade table version should fail!");
@@ -312,7 +312,7 @@ mod tests {
     #[test]
     fn test_set_table_property() {
         let table = make_v2_table();
-        let tx = Transaction::new(&table);
+        let tx = Transaction::new(table);
         let tx = tx
             .set_properties(HashMap::from([("a".to_string(), "b".to_string())]))
             .unwrap();
@@ -328,7 +328,7 @@ mod tests {
     #[test]
     fn test_remove_property() {
         let table = make_v2_table();
-        let tx = Transaction::new(&table);
+        let tx = Transaction::new(table);
         let tx = tx
             .remove_properties(vec!["a".to_string(), "b".to_string()])
             .unwrap();
@@ -344,7 +344,7 @@ mod tests {
     #[test]
     fn test_set_location() {
         let table = make_v2_table();
-        let tx = Transaction::new(&table);
+        let tx = Transaction::new(table);
         let tx = tx
             .set_location(String::from("s3://bucket/prefix/new_table"))
             .unwrap();
@@ -360,7 +360,7 @@ mod tests {
     #[tokio::test]
     async fn test_transaction_apply_upgrade() {
         let table = make_v1_table();
-        let tx = Transaction::new(&table);
+        let tx = Transaction::new(table);
         // Upgrade v1 to v1, do nothing.
         let tx = tx.upgrade_table_version(FormatVersion::V1).unwrap();
         // Upgrade v1 to v2, success.

crates/iceberg/src/transaction/snapshot.rs

@@ -59,8 +59,8 @@ pub(crate) trait ManifestProcess: Send + Sync {
     fn process_manifests(&self, manifests: Vec<ManifestFile>) -> Vec<ManifestFile>;
 }
 
-pub(crate) struct SnapshotProduceAction<'a> {
-    pub tx: Transaction<'a>,
+pub(crate) struct SnapshotProduceAction {
+    pub tx: Transaction,
     snapshot_id: i64,
     key_metadata: Vec<u8>,
     commit_uuid: Uuid,
@@ -72,9 +72,9 @@ pub(crate) struct SnapshotProduceAction<'a> {
     manifest_counter: RangeFrom<u64>,
 }
 
-impl<'a> SnapshotProduceAction<'a> {
+impl SnapshotProduceAction {
     pub(crate) fn new(
-        tx: Transaction<'a>,
+        tx: Transaction,
         snapshot_id: i64,
         key_metadata: Vec<u8>,
         commit_uuid: Uuid,
@@ -310,7 +310,7 @@ impl<'a> SnapshotProduceAction<'a> {
         mut self,
         snapshot_produce_operation: OP,
         process: MP,
-    ) -> Result<Transaction<'a>> {
+    ) -> Result<Transaction> {
         let new_manifests = self
             .manifest_file(&snapshot_produce_operation, &process)
             .await?;

crates/iceberg/src/transaction/sort_order.rs

@@ -21,12 +21,12 @@ use crate::transaction::Transaction;
 use crate::{Error, ErrorKind, TableRequirement, TableUpdate};
 
 /// Transaction action for replacing sort order.
-pub struct ReplaceSortOrderAction<'a> {
-    pub tx: Transaction<'a>,
+pub struct ReplaceSortOrderAction {
+    pub tx: Transaction,
     pub sort_fields: Vec<SortField>,
 }
 
-impl<'a> ReplaceSortOrderAction<'a> {
+impl ReplaceSortOrderAction {
     /// Adds a field for sorting in ascending order.
     pub fn asc(self, name: &str, null_order: NullOrder) -> Result<Self> {
         self.add_sort_field(name, SortDirection::Ascending, null_order)
@@ -38,7 +38,7 @@ impl<'a> ReplaceSortOrderAction<'a> {
     }
 
     /// Finished building the action and apply it to the transaction.
-    pub fn apply(mut self) -> Result<Transaction<'a>> {
+    pub fn apply(mut self) -> Result<Transaction> {
         let unbound_sort_order = SortOrder::builder()
             .with_fields(self.sort_fields)
             .build_unbound()?;
@@ -114,7 +114,7 @@ mod tests {
     #[test]
     fn test_replace_sort_order() {
         let table = make_v2_table();
-        let tx = Transaction::new(&table);
+        let tx = Transaction::new(table);
         let tx = tx.replace_sort_order().apply().unwrap();
 
         assert_eq!(

crates/integration_tests/tests/shared_tests/append_data_file_test.rs

@@ -111,7 +111,7 @@ async fn test_append_data_file() {
     assert_eq!(field_ids, vec![1, 2, 3]);
 
     // commit result
-    let tx = Transaction::new(&table);
+    let tx = Transaction::new(table);
     let mut append_action = tx.fast_append(None, vec![]).unwrap();
     append_action.add_data_files(data_file.clone()).unwrap();
     let tx = append_action.apply().await.unwrap();

crates/integration_tests/tests/shared_tests/append_partition_data_file_test.rs

@@ -119,7 +119,7 @@ async fn test_append_partition_data_file() {
     let data_file_valid = data_file_writer_valid.close().await.unwrap();
 
     // commit result
-    let tx = Transaction::new(&table);
+    let tx = Transaction::new(table);
     let mut append_action = tx.fast_append(None, vec![]).unwrap();
     append_action
         .add_data_files(data_file_valid.clone())
@@ -179,7 +179,7 @@ async fn test_schema_incompatible_partition_type(
     data_file_writer_invalid.write(batch.clone()).await.unwrap();
     let data_file_invalid = data_file_writer_invalid.close().await.unwrap();
 
-    let tx = Transaction::new(&table);
+    let tx = Transaction::new(table);
     let mut append_action = tx.fast_append(None, vec![]).unwrap();
     if append_action
         .add_data_files(data_file_invalid.clone())
@@ -219,7 +219,7 @@ async fn test_schema_incompatible_partition_fields(
     data_file_writer_invalid.write(batch.clone()).await.unwrap();
     let data_file_invalid = data_file_writer_invalid.close().await.unwrap();
 
-    let tx = Transaction::new(&table);
+    let tx = Transaction::new(table);
     let mut append_action = tx.fast_append(None, vec![]).unwrap();
     if append_action
         .add_data_files(data_file_invalid.clone())

crates/integration_tests/tests/shared_tests/conflict_commit_test.rs

@@ -89,12 +89,12 @@ async fn test_append_data_file_conflict() {
     let data_file = data_file_writer.close().await.unwrap();
 
     // start two transaction and commit one of them
-    let tx1 = Transaction::new(&table);
+    let tx1 = Transaction::new(table.clone());
     let mut append_action = tx1.fast_append(None, vec![]).unwrap();
     append_action.add_data_files(data_file.clone()).unwrap();
     let tx1 = append_action.apply().await.unwrap();
 
-    let tx2 = Transaction::new(&table);
+    let tx2 = Transaction::new(table.clone());
     let mut append_action = tx2.fast_append(None, vec![]).unwrap();
     append_action.add_data_files(data_file.clone()).unwrap();
     let tx2 = append_action.apply().await.unwrap();

crates/integration_tests/tests/shared_tests/scan_all_type.rs

@@ -308,7 +308,7 @@ async fn test_scan_all_type() {
     let data_file = data_file_writer.close().await.unwrap();
 
     // commit result
-    let tx = Transaction::new(&table);
+    let tx = Transaction::new(table);
     let mut append_action = tx.fast_append(None, vec![]).unwrap();
     append_action.add_data_files(data_file.clone()).unwrap();
     let tx = append_action.apply().await.unwrap();