feat: include delete files for fast append action

Author: jdockerty

crates/iceberg/src/transaction/append.rs

@@ -110,6 +110,13 @@ impl<'a> FastAppendAction<'a> {
 
     /// Finished building the action and apply it to the transaction.
     pub async fn apply(self) -> Result<Transaction<'a>> {
+        if self.snapshot_produce_action.added_data_files.is_empty() {
+            return Err(Error::new(
+                ErrorKind::DataInvalid,
+                "cannot apply no data files to transaction",
+            ));
+        }
+
         // Checks duplicate files
         if self.check_duplicate {
             let new_files: HashSet<&str> = self
@@ -247,6 +254,8 @@ mod tests {
             .unwrap();
         assert!(action.add_data_files(vec![data_file.clone()]).is_err());
 
+        let tx = Transaction::new(&table);
+        let mut action = tx.fast_append(None, vec![]).unwrap();
         let data_file = DataFileBuilder::default()
             .content(DataContentType::Data)
             .file_path("test/3.parquet".to_string())

crates/iceberg/src/transaction/snapshot.rs

@@ -24,10 +24,11 @@ use uuid::Uuid;
 use crate::error::Result;
 use crate::io::OutputFile;
 use crate::spec::{
-    DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestEntry, ManifestFile,
-    ManifestListWriter, ManifestWriterBuilder, Operation, PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT,
-    PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT_DEFAULT, Snapshot, SnapshotReference, SnapshotRetention,
-    SnapshotSummaryCollector, Struct, StructType, Summary, update_snapshot_summaries,
+    DataContentType, DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType,
+    ManifestEntry, ManifestFile, ManifestListWriter, ManifestWriterBuilder, Operation,
+    PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT, PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT_DEFAULT,
+    Snapshot, SnapshotReference, SnapshotRetention, SnapshotSummaryCollector, Struct, StructType,
+    Summary, update_snapshot_summaries,
 };
 use crate::transaction::Transaction;
 use crate::{Error, ErrorKind, TableRequirement, TableUpdate};
@@ -66,6 +67,7 @@ pub(crate) struct SnapshotProduceAction<'a> {
     commit_uuid: Uuid,
     snapshot_properties: HashMap<String, String>,
     pub added_data_files: Vec<DataFile>,
+    pub added_delete_files: Vec<DataFile>,
     // A counter used to generate unique manifest file names.
     // It starts from 0 and increments for each new manifest file.
     // Note: This counter is limited to the range of (0..u64::MAX).
@@ -86,6 +88,7 @@ impl<'a> SnapshotProduceAction<'a> {
             commit_uuid,
             snapshot_properties,
             added_data_files: vec![],
+            added_delete_files: vec![],
             manifest_counter: (0..),
             key_metadata,
         })
@@ -129,12 +132,14 @@ impl<'a> SnapshotProduceAction<'a> {
     ) -> Result<&mut Self> {
         let data_files: Vec<DataFile> = data_files.into_iter().collect();
         for data_file in &data_files {
-            if data_file.content_type() != crate::spec::DataContentType::Data {
-                return Err(Error::new(
-                    ErrorKind::DataInvalid,
-                    "Only data content type is allowed for fast append",
-                ));
+            match data_file.content_type() {
+                crate::spec::DataContentType::Data => self.added_data_files.push(data_file.clone()),
+                crate::spec::DataContentType::PositionDeletes
+                | crate::spec::DataContentType::EqualityDeletes => {
+                    self.added_delete_files.push(data_file.clone())
+                }
             }
+
             // Check if the data file partition spec id matches the table default partition spec id.
             if self.tx.current_table.metadata().default_partition_spec_id()
                 != data_file.partition_spec_id
@@ -149,7 +154,6 @@ impl<'a> SnapshotProduceAction<'a> {
                 self.tx.current_table.metadata().default_partition_type(),
             )?;
         }
-        self.added_data_files.extend(data_files);
         Ok(self)
     }
 
@@ -169,15 +173,41 @@ impl<'a> SnapshotProduceAction<'a> {
     }
 
     // Write manifest file for added data files and return the ManifestFile for ManifestList.
-    async fn write_added_manifest(&mut self) -> Result<ManifestFile> {
-        let added_data_files = std::mem::take(&mut self.added_data_files);
+    async fn write_added_manifest(
+        &mut self,
+        added_data_files: Vec<DataFile>,
+    ) -> Result<ManifestFile> {
         if added_data_files.is_empty() {
             return Err(Error::new(
                 ErrorKind::PreconditionFailed,
                 "No added data files found when write a manifest file",
             ));
         }
 
+        let file_count = added_data_files.len();
+        let manifest_content_type = {
+            let mut data_num = 0;
+            let mut delete_num = 0;
+            for f in &added_data_files {
+                match f.content_type() {
+                    DataContentType::Data => data_num += 1,
+                    DataContentType::PositionDeletes | DataContentType::EqualityDeletes => {
+                        delete_num += 1
+                    }
+                }
+            }
+            if data_num == file_count {
+                ManifestContentType::Data
+            } else if delete_num == file_count {
+                ManifestContentType::Deletes
+            } else {
+                return Err(Error::new(
+                    ErrorKind::DataInvalid,
+                    "added DataFile for a ManifestFile should be same type (Data or Delete)",
+                ));
+            }
+        };
+
         let snapshot_id = self.snapshot_id;
         let format_version = self.tx.current_table.metadata().format_version();
         let manifest_entries = added_data_files.into_iter().map(|data_file| {
@@ -207,8 +237,10 @@ impl<'a> SnapshotProduceAction<'a> {
             );
             if self.tx.current_table.metadata().format_version() == FormatVersion::V1 {
                 builder.build_v1()
-            } else {
+            } else if manifest_content_type == ManifestContentType::Data {
                 builder.build_v2_data()
+            } else {
+                builder.build_v2_deletes()
             }
         };
         for entry in manifest_entries {
@@ -222,14 +254,21 @@ impl<'a> SnapshotProduceAction<'a> {
         snapshot_produce_operation: &OP,
         manifest_process: &MP,
     ) -> Result<Vec<ManifestFile>> {
-        let added_manifest = self.write_added_manifest().await?;
-        let existing_manifests = snapshot_produce_operation.existing_manifest(self).await?;
-        // # TODO
-        // Support process delete entries.
-
-        let mut manifest_files = vec![added_manifest];
-        manifest_files.extend(existing_manifests);
-        let manifest_files = manifest_process.process_manifests(manifest_files);
+        let mut existing_manifests = snapshot_produce_operation.existing_manifest(self).await?;
+
+        if !self.added_data_files.is_empty() {
+            let added_data_files = std::mem::take(&mut self.added_data_files);
+            let added_manifest = self.write_added_manifest(added_data_files).await?;
+            existing_manifests.push(added_manifest);
+        }
+
+        if !self.added_delete_files.is_empty() {
+            let added_delete_files = std::mem::take(&mut self.added_delete_files);
+            let added_manifest = self.write_added_manifest(added_delete_files).await?;
+            existing_manifests.push(added_manifest);
+        }
+
+        let manifest_files = manifest_process.process_manifests(existing_manifests);
         Ok(manifest_files)
     }