fix(iceberg): add_files correctly check duplicates (#1395)

## Which issue does this PR close?

- Closes #1394.

## What changes are included in this PR?


- compare duplicates by loading manifest files and taking `file_path`
from it
- use direct calls instead of `scan`

## Are these changes tested?

- work for me local experiments
- fixed existing tests
- added a new test to showcase behavior


Co-authored-by: Roman Shanin <rshanin@bhft.com>

Author: Erigara

crates/iceberg/src/transaction/append.rs

@@ -17,8 +17,6 @@
 
 use std::collections::{HashMap, HashSet};
 
-use arrow_array::StringArray;
-use futures::TryStreamExt;
 use uuid::Uuid;
 
 use crate::error::Result;
@@ -129,32 +127,19 @@ impl<'a> FastAppendAction<'a> {
                 .map(|df| df.file_path.as_str())
                 .collect();
 
-            let mut manifest_stream = self
-                .snapshot_produce_action
-                .tx
-                .current_table
-                .inspect()
-                .manifests()
-                .scan()
-                .await?;
             let mut referenced_files = Vec::new();
-
-            while let Some(batch) = manifest_stream.try_next().await? {
-                let file_path_array = batch
-                    .column(1)
-                    .as_any()
-                    .downcast_ref::<StringArray>()
-                    .ok_or_else(|| {
-                        Error::new(
-                            ErrorKind::DataInvalid,
-                            "Failed to downcast file_path column to StringArray",
-                        )
-                    })?;
-
-                for i in 0..batch.num_rows() {
-                    let file_path = file_path_array.value(i);
-                    if new_files.contains(file_path) {
-                        referenced_files.push(file_path.to_string());
+            let table = &self.snapshot_produce_action.tx.current_table;
+            if let Some(current_snapshot) = table.metadata().current_snapshot() {
+                let manifest_list = current_snapshot
+                    .load_manifest_list(table.file_io(), &table.metadata_ref())
+                    .await?;
+                for manifest_list_entry in manifest_list.entries() {
+                    let manifest = manifest_list_entry.load_manifest(table.file_io()).await?;
+                    for entry in manifest.entries() {
+                        let file_path = entry.file_path();
+                        if new_files.contains(file_path) && entry.is_alive() {
+                            referenced_files.push(file_path.to_string());
+                        }
                     }
                 }
             }
@@ -364,81 +349,39 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_add_existing_parquet_files_to_unpartitioned_table() {
+    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 file_paths = vec![
             format!("{}/1.parquet", &fixture.table_location),
-            format!("{}/2.parquet", &fixture.table_location),
             format!("{}/3.parquet", &fixture.table_location),
         ];
 
         let fast_append_action = tx.fast_append(None, vec![]).unwrap();
 
-        // Attempt to add the existing Parquet files with fast append.
-        let new_tx = fast_append_action
-            .add_parquet_files(file_paths.clone())
-            .await
-            .expect("Adding existing Parquet files should succeed");
-
-        let mut found_add_snapshot = false;
-        let mut found_set_snapshot_ref = false;
-        for update in new_tx.updates.iter() {
-            match update {
-                TableUpdate::AddSnapshot { .. } => {
-                    found_add_snapshot = true;
-                }
-                TableUpdate::SetSnapshotRef {
-                    ref_name,
-                    reference,
-                } => {
-                    found_set_snapshot_ref = true;
-                    assert_eq!(ref_name, MAIN_BRANCH);
-                    assert!(reference.snapshot_id > 0);
-                }
-                _ => {}
-            }
-        }
-        assert!(found_add_snapshot);
-        assert!(found_set_snapshot_ref);
-
-        let new_snapshot = if let TableUpdate::AddSnapshot { snapshot } = &new_tx.updates[0] {
-            snapshot
-        } else {
-            panic!("Expected the first update to be an AddSnapshot update");
-        };
-
-        let manifest_list = new_snapshot
-            .load_manifest_list(fixture.table.file_io(), fixture.table.metadata())
-            .await
-            .expect("Failed to load manifest list");
-
-        assert_eq!(
-            manifest_list.entries().len(),
-            2,
-            "Expected 2 manifest list entries, got {}",
-            manifest_list.entries().len()
+        // Attempt to add duplicated Parquet files with fast append.
+        assert!(
+            fast_append_action
+                .add_parquet_files(file_paths.clone())
+                .await
+                .is_err(),
+            "file already in table"
         );
 
-        // Load the manifest from the manifest list
-        let manifest = manifest_list.entries()[0]
-            .load_manifest(fixture.table.file_io())
-            .await
-            .expect("Failed to load manifest");
+        let file_paths = vec![format!("{}/2.parquet", &fixture.table_location)];
 
-        // Check that the manifest contains three entries.
-        assert_eq!(manifest.entries().len(), 3);
+        let tx = crate::transaction::Transaction::new(&fixture.table);
+        let fast_append_action = tx.fast_append(None, vec![]).unwrap();
 
-        // Verify each file path appears in manifest.
-        let manifest_paths: Vec<String> = manifest
-            .entries()
-            .iter()
-            .map(|entry| entry.data_file().file_path.clone())
-            .collect();
-        for path in file_paths {
-            assert!(manifest_paths.contains(&path));
-        }
+        // Attempt to add Parquet file which was deleted from table.
+        assert!(
+            fast_append_action
+                .add_parquet_files(file_paths.clone())
+                .await
+                .is_ok(),
+            "file not in table"
+        );
     }
 }

crates/integration_tests/tests/shared_tests/append_data_file_test.rs

@@ -129,25 +129,4 @@ async fn test_append_data_file() {
     let batches: Vec<_> = batch_stream.try_collect().await.unwrap();
     assert_eq!(batches.len(), 1);
     assert_eq!(batches[0], batch);
-
-    // commit result again
-    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();
-    let table = tx.commit(&rest_catalog).await.unwrap();
-
-    // check result again
-    let batch_stream = table
-        .scan()
-        .select_all()
-        .build()
-        .unwrap()
-        .to_arrow()
-        .await
-        .unwrap();
-    let batches: Vec<_> = batch_stream.try_collect().await.unwrap();
-    assert_eq!(batches.len(), 2);
-    assert_eq!(batches[0], batch);
-    assert_eq!(batches[1], batch);
 }