Avoid removing snapshots and data files

Author: SergeiPatiakinEdb

iceberg-rust/src/table/transaction/mod.rs

@@ -23,8 +23,6 @@ use crate::{catalog::commit::CommitTable, error::Error, table::Table};
 
 use self::operation::Operation;
 
-use super::delete_all_table_files;
-
 pub(crate) mod append;
 pub(crate) mod operation;
 
@@ -333,25 +331,8 @@ impl<'table> TableTransaction<'table> {
     /// ```
     pub async fn commit(self) -> Result<(), Error> {
         let catalog = self.table.catalog();
-        let object_store = self.table.object_store();
         let identifier = self.table.identifier.clone();
 
-        // Save old metadata to be able to remove old data after a rewrite operation
-        let delete_data = if self.operations.values().any(|x| {
-            matches!(
-                x,
-                Operation::Replace {
-                    branch: _,
-                    files: _,
-                    additional_summary: _,
-                }
-            )
-        }) {
-            Some(self.table.metadata())
-        } else {
-            None
-        };
-
         // Execute the table operations
         let (mut requirements, mut updates) = (Vec::new(), Vec::new());
         for operation in self.operations.into_values() {
@@ -374,10 +355,6 @@ impl<'table> TableTransaction<'table> {
             })
             .await?;
 
-        if let Some(old_metadata) = delete_data {
-            delete_all_table_files(old_metadata, object_store).await?;
-        }
-
         *self.table = new_table;
         Ok(())
     }

iceberg-rust/src/table/transaction/operation.rs

@@ -521,18 +521,12 @@ impl Operation {
                     .build()
                     .map_err(iceberg_rust_spec::error::Error::from)?;
 
-                let old_snapshot_ids: Vec<i64> =
-                    table_metadata.snapshots.keys().map(Clone::clone).collect();
-
                 Ok((
                     old_snapshot.map(|x| TableRequirement::AssertRefSnapshotId {
                         r#ref: branch.clone().unwrap_or("main".to_owned()),
                         snapshot_id: *x.snapshot_id(),
                     }),
                     vec![
-                        TableUpdate::RemoveSnapshots {
-                            snapshot_ids: old_snapshot_ids,
-                        },
                         TableUpdate::AddSnapshot { snapshot },
                         TableUpdate::SetSnapshotRef {
                             ref_name: branch.unwrap_or("main".to_owned()),