Check add_files duplicates against the target branch

mkdocs/docs/api.md

@@ -1093,7 +1093,7 @@ Expert Iceberg users may choose to commit existing parquet files to the Iceberg
     Because `add_files` commits the existing parquet files to the Iceberg Table as any other data file, destructive maintenance operations like expiring snapshots will remove them.
 
 !!! warning "Check Duplicate Files"
-    The `check_duplicate_files` parameter determines whether the method validates that the specified `file_paths` do not already exist in the Iceberg table. When set to True (the default), the method performs a validation against the table’s current data files to prevent accidental duplication, helping to maintain data consistency by ensuring the same file is not added multiple times. While this check is important for data integrity, it can introduce performance overhead for tables with a large number of files. Setting check_duplicate_files=False can improve performance but increases the risk of duplicate files, which may lead to data inconsistencies or table corruption. It is strongly recommended to keep this parameter enabled unless duplicate file handling is strictly enforced elsewhere.
+    The `check_duplicate_files` parameter determines whether the method validates that the specified `file_paths` do not already exist in the Iceberg table. When set to True (the default), the method performs a validation against the table’s current data files and the data files of the target branch to prevent accidental duplication, helping to maintain data consistency by ensuring the same file is not added multiple times. While this check is important for data integrity, it can introduce performance overhead for tables with a large number of files. Setting check_duplicate_files=False can improve performance but increases the risk of duplicate files, which may lead to data inconsistencies or table corruption. It is strongly recommended to keep this parameter enabled unless duplicate file handling is strictly enforced elsewhere.
 
 <!-- prettier-ignore-end -->
 

pyiceberg/table/__init__.py

@@ -1009,15 +1009,26 @@ def upsert(
 
         return UpsertResult(rows_updated=update_row_cnt, rows_inserted=insert_row_cnt)
 
-    def _find_referenced_data_files(self, file_paths: list[str]) -> list[str]:
-        """Return file_paths already referenced by data files in the current snapshot."""
-        snapshot = self.table_metadata.current_snapshot()
-        if snapshot is None:
+    def _find_referenced_data_files(self, file_paths: list[str], branch: str | None) -> list[str]:
+        """Return file_paths already referenced by data files in the current snapshot or the target branch."""
+        # Check both: the current snapshot is always checked, and the target branch is what the append builds on.
+        # A snapshot or manifest shared by both is scanned once.
+        snapshots = {
+            snapshot.snapshot_id: snapshot
+            for snapshot in (self.table_metadata.current_snapshot(), self.table_metadata.snapshot_by_name(branch))
+            if snapshot is not None
+        }
+        if not snapshots:
             return []
 
         candidates = set(file_paths)
         io = self._table.io
-        data_manifests = [m for m in snapshot.manifests(io) if m.content == ManifestContent.DATA]
+        data_manifests = {
+            manifest.manifest_path: manifest
+            for snapshot in snapshots.values()
+            for manifest in snapshot.manifests(io)
+            if manifest.content == ManifestContent.DATA
+        }
 
         def path_filter(data_file: DataFile) -> bool:
             return data_file.file_path in candidates
@@ -1026,7 +1037,7 @@ def path_filter(data_file: DataFile) -> bool:
         entries = chain.from_iterable(
             executor.map(
                 lambda args: _open_manifest(*args),
-                [(io, manifest, path_filter, lambda _: True) for manifest in data_manifests],
+                [(io, manifest, path_filter, lambda _: True) for manifest in data_manifests.values()],
             )
         )
         return [entry.data_file.file_path for entry in entries]
@@ -1053,7 +1064,7 @@ def add_files(
             raise ValueError("File paths must be unique")
 
         if check_duplicate_files:
-            referenced_files = self._find_referenced_data_files(file_paths)
+            referenced_files = self._find_referenced_data_files(file_paths, branch)
             if referenced_files:
                 raise ValueError(f"Cannot add files that are already referenced by table, files: {', '.join(referenced_files)}")
 

tests/integration/test_add_files.py

@@ -22,6 +22,7 @@
 import threading
 from collections.abc import Iterator
 from datetime import date
+from pathlib import Path
 from unittest import mock
 
 import pyarrow as pa
@@ -1040,3 +1041,76 @@ def test_add_files_to_branch(spark: SparkSession, session_catalog: Catalog, form
 
     for col in branch_df.columns:
         assert branch_df.filter(branch_df[col].isNotNull()).count() == 6, "Expected all 6 rows to be non-null"
+
+
+def _create_table_with_branch(catalog: Catalog, branch: str) -> Table:
+    catalog.create_namespace("default")
+    tbl = catalog.create_table("default.test_add_files_branch_duplicate", schema=TABLE_SCHEMA)
+    tbl.append(ARROW_TABLE)
+    assert tbl.metadata.current_snapshot_id is not None
+    tbl.manage_snapshots().create_branch(snapshot_id=tbl.metadata.current_snapshot_id, branch_name=branch).commit()
+    return tbl
+
+
+@pytest.mark.integration
+def test_add_files_to_branch_rejects_file_already_referenced_by_branch(catalog: Catalog, tmp_path: Path) -> None:
+    branch = "branch1"
+    tbl = _create_table_with_branch(catalog, branch)
+    file_path = f"file://{tmp_path}/duplicate.parquet"
+    _write_parquet(tbl.io, file_path, ARROW_SCHEMA, ARROW_TABLE)
+
+    tbl.add_files(file_paths=[file_path], branch=branch)
+    with pytest.raises(ValueError, match="already referenced by table"):
+        tbl.add_files(file_paths=[file_path], branch=branch)
+
+    # the branch holds the row from main and the row from the file, each exactly once
+    branch_rows = tbl.scan(snapshot_id=tbl.metadata.refs[branch].snapshot_id).to_arrow().to_pylist()
+    assert branch_rows == ARROW_TABLE.to_pylist() * 2
+
+
+@pytest.mark.integration
+def test_add_files_to_branch_rejects_file_added_earlier_in_transaction(catalog: Catalog, tmp_path: Path) -> None:
+    branch = "branch1"
+    tbl = _create_table_with_branch(catalog, branch)
+    file_path = f"file://{tmp_path}/duplicate.parquet"
+    _write_parquet(tbl.io, file_path, ARROW_SCHEMA, ARROW_TABLE)
+
+    with tbl.transaction() as tx:
+        tx.add_files(file_paths=[file_path], branch=branch)
+        with pytest.raises(ValueError, match="already referenced by table"):
+            tx.add_files(file_paths=[file_path], branch=branch)
+
+    branch_rows = tbl.scan(snapshot_id=tbl.metadata.refs[branch].snapshot_id).to_arrow().to_pylist()
+    assert branch_rows == ARROW_TABLE.to_pylist() * 2
+
+
+@pytest.mark.integration
+def test_add_files_to_branch_accepts_distinct_file_with_same_rows(catalog: Catalog, tmp_path: Path) -> None:
+    branch = "branch1"
+    tbl = _create_table_with_branch(catalog, branch)
+    first_path = f"file://{tmp_path}/first.parquet"
+    second_path = f"file://{tmp_path}/second.parquet"
+    _write_parquet(tbl.io, first_path, ARROW_SCHEMA, ARROW_TABLE)
+    _write_parquet(tbl.io, second_path, ARROW_SCHEMA, ARROW_TABLE)
+
+    tbl.add_files(file_paths=[first_path], branch=branch)
+    tbl.add_files(file_paths=[second_path], branch=branch)
+
+    branch_rows = tbl.scan(snapshot_id=tbl.metadata.refs[branch].snapshot_id).to_arrow().to_pylist()
+    assert branch_rows == ARROW_TABLE.to_pylist() * 3
+
+
+@pytest.mark.integration
+def test_add_files_to_branch_rejects_file_referenced_only_by_main(catalog: Catalog, tmp_path: Path) -> None:
+    branch = "branch1"
+    tbl = _create_table_with_branch(catalog, branch)
+    file_path = f"file://{tmp_path}/main-only.parquet"
+    _write_parquet(tbl.io, file_path, ARROW_SCHEMA, ARROW_TABLE)
+
+    # main references the file after this call, the branch does not
+    tbl.add_files(file_paths=[file_path])
+    with pytest.raises(ValueError, match="already referenced by table"):
+        tbl.add_files(file_paths=[file_path], branch=branch)
+
+    branch_rows = tbl.scan(snapshot_id=tbl.metadata.refs[branch].snapshot_id).to_arrow().to_pylist()
+    assert branch_rows == ARROW_TABLE.to_pylist()