feat(iceberg): Enable delete files processing in snapshot producer

crates/iceberg/src/spec/manifest/writer.rs

@@ -299,10 +299,6 @@ impl ManifestWriter {
     /// Add a delete manifest entry. This method will update following status of the entry:
     /// - Update the entry status to `Deleted`
     /// - Set the snapshot id to the current snapshot id
-    ///
-    /// # TODO
-    /// Remove this allow later
-    #[allow(dead_code)]
     pub(crate) fn add_delete_entry(&mut self, mut entry: ManifestEntry) -> Result<()> {
         self.check_data_file(&entry.data_file)?;
         entry.status = ManifestStatus::Deleted;
@@ -345,7 +341,7 @@ impl ManifestWriter {
         Ok(())
     }
 
-    /// Add an file as existing manifest entry. The original data and file sequence numbers, snapshot ID,
+    /// Add a file as existing manifest entry. The original data and file sequence numbers, snapshot ID,
     /// which were assigned at commit, must be preserved when adding an existing entry.
     pub fn add_existing_file(
         &mut self,

crates/iceberg/src/spec/snapshot.rs

@@ -36,7 +36,7 @@ pub const MAIN_BRANCH: &str = "main";
 
 /// Reference to [`Snapshot`].
 pub type SnapshotRef = Arc<Snapshot>;
-#[derive(Debug, Default, Serialize, Deserialize, PartialEq, Eq, Clone)]
+#[derive(Debug, Default, Serialize, Deserialize, PartialEq, Eq, Clone, Hash)]
 #[serde(rename_all = "lowercase")]
 /// The operation field is used by some operations, like snapshot expiration, to skip processing certain snapshots.
 pub enum Operation {

crates/iceberg/src/transaction/append.rs

@@ -84,13 +84,13 @@ impl FastAppendAction {
 #[async_trait]
 impl TransactionAction for FastAppendAction {
     async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> {
-        let snapshot_producer = SnapshotProducer::new(
-            table,
-            self.commit_uuid.unwrap_or_else(Uuid::now_v7),
-            self.key_metadata.clone(),
-            self.snapshot_properties.clone(),
-            self.added_data_files.clone(),
-        );
+        let snapshot_producer = SnapshotProducer::builder()
+            .with_table(table)
+            .with_commit_uuid(self.commit_uuid.unwrap_or_else(Uuid::now_v7))
+            .with_key_metadata(self.key_metadata.clone())
+            .with_snapshot_properties(self.snapshot_properties.clone())
+            .with_added_data_files(self.added_data_files.clone())
+            .build();
 
         // validate added files
         snapshot_producer.validate_added_data_files()?;
@@ -122,7 +122,7 @@ impl SnapshotProduceOperation for FastAppendOperation {
 
     async fn existing_manifest(
         &self,
-        snapshot_produce: &SnapshotProducer<'_>,
+        snapshot_produce: &mut SnapshotProducer<'_>,
     ) -> Result<Vec<ManifestFile>> {
         let Some(snapshot) = snapshot_produce.table.metadata().current_snapshot() else {
             return Ok(vec![]);