fix(manifest): verify cached ManifestFile content before reusing it

pyiceberg/catalog/__init__.py

@@ -962,7 +962,7 @@ def purge_table(self, identifier: str | Identifier) -> None:
         manifest_lists_to_delete = set()
         manifests_to_delete: list[ManifestFile] = []
         for snapshot in metadata.snapshots:
-            manifests_to_delete += snapshot.manifests(io)
+            manifests_to_delete += snapshot.manifests(io, table_uuid=metadata.table_uuid)
             manifest_lists_to_delete.add(snapshot.manifest_list)
 
         manifest_paths_to_delete = {manifest.manifest_path for manifest in manifests_to_delete}

pyiceberg/cli/output.py

@@ -170,7 +170,7 @@ def files(self, table: Table, history: bool) -> None:
                 f"Snapshot {snapshot.snapshot_id}, schema {snapshot.schema_id}: {snapshot.manifest_list}"
             )
 
-            manifest_list = snapshot.manifests(io)
+            manifest_list = snapshot.manifests(io, table_uuid=table.metadata.table_uuid)
             for manifest in manifest_list:
                 manifest_tree = list_tree.add(f"Manifest: {manifest.manifest_path}")
                 for manifest_entry in manifest.fetch_manifest_entry(io, discard_deleted=False):

pyiceberg/manifest.py

@@ -18,6 +18,7 @@
 
 import math
 import threading
+import uuid
 from abc import ABC, abstractmethod
 from collections.abc import Callable, Iterator
 from copy import copy
@@ -937,16 +938,19 @@ def __hash__(self) -> int:
 
 
 class _ManifestCache:
-    """Process-wide ManifestFile cache keyed by manifest_path.
+    """Process-wide ManifestFile cache keyed by (table_uuid, manifest_path).
 
     Consecutive snapshots often reference the same manifests after append
     operations, so reusing ManifestFile instances avoids retaining duplicate
-    objects.
+    objects. The table_uuid is part of the key so that two tables can never
+    collide on a manifest_path, even if one happens to reuse the other's path.
+    A caller that does not supply a table_uuid cannot be attributed to a table,
+    so it bypasses the cache rather than sharing an unattributed entry.
     """
 
     DEFAULT_SIZE = 128
 
-    _cache: LRUCache[str, ManifestFile] | None
+    _cache: LRUCache[tuple[uuid.UUID, str], ManifestFile] | None
 
     def __init__(self) -> None:
         self.maxsize = self._load_configured_size()
@@ -969,16 +973,16 @@ def clear(self) -> None:
             if self._cache is not None:
                 self._cache.clear()
 
-    def get_or_cache(self, manifest_file: ManifestFile) -> ManifestFile:
-        if self._cache is None:
+    def get_or_cache(self, manifest_file: ManifestFile, table_uuid: uuid.UUID | None = None) -> ManifestFile:
+        if self._cache is None or table_uuid is None:
             return manifest_file
 
         with self._lock:
-            manifest_path = manifest_file.manifest_path
-            if manifest_path in self._cache:
-                return self._cache[manifest_path]
+            key = (table_uuid, manifest_file.manifest_path)
+            if key in self._cache:
+                return self._cache[key]
 
-            self._cache[manifest_path] = manifest_file
+            self._cache[key] = manifest_file
             return manifest_file
 
     def __len__(self) -> int:
@@ -998,11 +1002,12 @@ def clear_manifest_cache() -> None:
     _manifest_cache.clear()
 
 
-def _manifests(io: FileIO, manifest_list: str) -> tuple[ManifestFile, ...]:
+def _manifests(io: FileIO, manifest_list: str, table_uuid: uuid.UUID | None = None) -> tuple[ManifestFile, ...]:
     """Read manifests from a manifest list, reusing cached ManifestFile objects.
 
-    Caches individual ManifestFile objects by manifest_path. This is memory-efficient
-    because consecutive manifest lists typically share most of their manifests:
+    Caches individual ManifestFile objects by (table_uuid, manifest_path). This is
+    memory-efficient because consecutive manifest lists typically share most of
+    their manifests:
 
         ManifestList1: [ManifestFile1]
         ManifestList2: [ManifestFile1, ManifestFile2]
@@ -1018,14 +1023,17 @@ def _manifests(io: FileIO, manifest_list: str) -> tuple[ManifestFile, ...]:
     Args:
         io: FileIO instance for reading the manifest list.
         manifest_list: Path to the manifest list file.
+        table_uuid: UUID of the table this manifest list belongs to, used to scope
+            the cache so that two tables can never share a cached entry. When
+            omitted the manifests are returned uncached.
 
     Returns:
         A tuple of ManifestFile objects.
     """
     file = io.new_input(manifest_list)
     manifest_files = list(read_manifest_list(file))
 
-    return tuple(_manifest_cache.get_or_cache(manifest_file) for manifest_file in manifest_files)
+    return tuple(_manifest_cache.get_or_cache(manifest_file, table_uuid) for manifest_file in manifest_files)
 
 
 def read_manifest_list(input_file: InputFile) -> Iterator[ManifestFile]:

pyiceberg/table/__init__.py

@@ -995,13 +995,16 @@ def upsert(
 
     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()
+        table_metadata = self.table_metadata
+        snapshot = table_metadata.current_snapshot()
         if snapshot is None:
             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 = [
+            m for m in snapshot.manifests(io, table_uuid=table_metadata.table_uuid) if m.content == ManifestContent.DATA
+        ]
 
         def path_filter(data_file: DataFile) -> bool:
             return data_file.file_path in candidates
@@ -2425,7 +2428,9 @@ def _plan_manifest_entries(self) -> Iterator[list[ManifestEntry]]:
         if not snapshot:
             return iter([])
 
-        return self._manifest_planner.plan_manifest_entries(snapshot.manifests(self.io))
+        return self._manifest_planner.plan_manifest_entries(
+            snapshot.manifests(self.io, table_uuid=self.table_metadata.table_uuid)
+        )
 
     def _should_use_server_side_planning(self) -> bool:
         """Check if server-side scan planning should be used for this scan."""
@@ -2462,7 +2467,7 @@ def _plan_files_local(self) -> Iterable[FileScanTask]:
         snapshot = self.snapshot()
         if not snapshot:
             return []
-        return self._manifest_planner.plan_files(snapshot.manifests(self.io))
+        return self._manifest_planner.plan_files(snapshot.manifests(self.io, table_uuid=self.table_metadata.table_uuid))
 
     def plan_files(self) -> Iterable[FileScanTask]:
         """Plans the relevant files by filtering on the PartitionSpecs.
@@ -2654,7 +2659,7 @@ def plan_files(self) -> Iterable[FileScanTask]:
             {
                 manifest_file
                 for snapshot in append_snapshots
-                for manifest_file in snapshot.manifests(self.io)
+                for manifest_file in snapshot.manifests(self.io, table_uuid=self.table_metadata.table_uuid)
                 if manifest_file.content == ManifestContent.DATA and manifest_file.added_snapshot_id in append_snapshot_ids
             }
         )
@@ -2667,8 +2672,9 @@ def plan_files(self) -> Iterable[FileScanTask]:
             options=self.options,
         ).plan_files(
             manifests=manifests,
-            manifest_entry_filter=lambda manifest_entry: manifest_entry.snapshot_id in append_snapshot_ids
-            and manifest_entry.status == ManifestEntryStatus.ADDED,
+            manifest_entry_filter=lambda manifest_entry: (
+                manifest_entry.snapshot_id in append_snapshot_ids and manifest_entry.status == ManifestEntryStatus.ADDED
+            ),
         )
 
     def to_arrow(self) -> pa.Table:

pyiceberg/table/inspect.py

@@ -232,7 +232,7 @@ def _readable_metrics_struct(bound_type: PrimitiveType) -> pa.StructType:
 
         entries = []
         snapshot = self._get_snapshot(snapshot_id)
-        for manifest in snapshot.manifests(self.tbl.io):
+        for manifest in snapshot.manifests(self.tbl.io, table_uuid=self.tbl.metadata.table_uuid):
             for entry in manifest.fetch_manifest_entry(io=self.tbl.io, discard_deleted=False):
                 column_sizes = entry.data_file.column_sizes or {}
                 value_counts = entry.data_file.value_counts or {}
@@ -383,7 +383,9 @@ def partitions(
         )
 
         snapshot = self._get_snapshot(snapshot_id)
-        spec_ids = {manifest.partition_spec_id for manifest in snapshot.manifests(self.tbl.io)}
+        spec_ids = {
+            manifest.partition_spec_id for manifest in snapshot.manifests(self.tbl.io, table_uuid=self.tbl.metadata.table_uuid)
+        }
         partition_record = self.tbl.metadata.specs_struct(spec_ids=spec_ids)
         has_partitions = len(partition_record.fields) > 0
 
@@ -590,7 +592,7 @@ def _partition_summaries_to_rows(
         specs = self.tbl.metadata.specs()
         manifests = []
         if snapshot:
-            for manifest in snapshot.manifests(self.tbl.io):
+            for manifest in snapshot.manifests(self.tbl.io, table_uuid=self.tbl.metadata.table_uuid):
                 is_data_file = manifest.content == ManifestContent.DATA
                 is_delete_file = manifest.content == ManifestContent.DELETES
                 manifest_row = {
@@ -882,7 +884,8 @@ def _files(self, snapshot_id: int | None = None, data_file_filter: set[DataFileC
         executor = ExecutorFactory.get_or_create()
         results = list(
             executor.map(
-                lambda manifest_list: self._get_files_from_manifest(manifest_list, data_file_filter), snapshot.manifests(io)
+                lambda manifest_list: self._get_files_from_manifest(manifest_list, data_file_filter),
+                snapshot.manifests(io, table_uuid=self.tbl.metadata.table_uuid),
             )
         )
         return pa.concat_tables(results)
@@ -970,7 +973,9 @@ def _all_files(self, data_file_filter: set[DataFileContent] | None = None) -> pa
             return pa.Table.from_pylist([], schema=self._get_files_schema())
 
         executor = ExecutorFactory.get_or_create()
-        manifest_lists = executor.map(lambda snapshot: snapshot.manifests(self.tbl.io), snapshots)
+        manifest_lists = executor.map(
+            lambda snapshot: snapshot.manifests(self.tbl.io, table_uuid=self.tbl.metadata.table_uuid), snapshots
+        )
 
         unique_manifests = {(manifest.manifest_path, manifest) for manifest_list in manifest_lists for manifest in manifest_list}
 

pyiceberg/table/snapshots.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import time
+import uuid
 import warnings
 from collections import defaultdict
 from collections.abc import Iterable, Mapping
@@ -289,9 +290,17 @@ def __repr__(self) -> str:
         filtered_fields = [field for field in fields if field is not None]
         return f"Snapshot({', '.join(filtered_fields)})"
 
-    def manifests(self, io: FileIO) -> list[ManifestFile]:
-        """Return the manifests for the given snapshot."""
-        return list(_manifests(io, self.manifest_list))
+    def manifests(self, io: FileIO, table_uuid: uuid.UUID | None = None) -> list[ManifestFile]:
+        """Return the manifests for the given snapshot.
+
+        Args:
+            io: FileIO instance for reading the manifest list.
+            table_uuid: UUID of the table this snapshot belongs to, used to scope
+                the process-wide manifest cache so that two tables can never
+                share a cached entry. When omitted the manifests are returned
+                uncached.
+        """
+        return list(_manifests(io, self.manifest_list, table_uuid))
 
 
 class MetadataLogEntry(IcebergBaseModel):

pyiceberg/table/update/snapshot.py

@@ -608,7 +608,7 @@ def _copy_with_new_status(entry: ManifestEntry, status: ManifestEntryStatus) ->
         if parent_snapshot_id_for_delete_source is not None:
             snapshot = table_metadata.snapshot_by_id(parent_snapshot_id_for_delete_source)
             if snapshot:  # Ensure snapshot is found
-                for manifest_file in snapshot.manifests(io=self._io):
+                for manifest_file in snapshot.manifests(io=self._io, table_uuid=table_metadata.table_uuid):
                     if manifest_file.content == ManifestContent.DATA:
                         if not manifest_evaluators[manifest_file.partition_spec_id](manifest_file):
                             # If the manifest isn't relevant, we can just keep it in the manifest-list
@@ -699,12 +699,13 @@ def _existing_manifests(self) -> list[ManifestFile]:
         existing_manifests = []
 
         if self._parent_snapshot_id is not None:
-            previous_snapshot = self._transaction.table_metadata.snapshot_by_id(self._parent_snapshot_id)
+            table_metadata = self._transaction.table_metadata
+            previous_snapshot = table_metadata.snapshot_by_id(self._parent_snapshot_id)
 
             if previous_snapshot is None:
                 raise ValueError(f"Snapshot could not be found: {self._parent_snapshot_id}")
 
-            for manifest in previous_snapshot.manifests(io=self._io):
+            for manifest in previous_snapshot.manifests(io=self._io, table_uuid=table_metadata.table_uuid):
                 if manifest.has_added_files() or manifest.has_existing_files() or manifest.added_snapshot_id == self._snapshot_id:
                     existing_manifests.append(manifest)
 
@@ -782,8 +783,9 @@ def _existing_manifests(self) -> list[ManifestFile]:
         existing_files = []
 
         manifest_evaluators: dict[int, Callable[[ManifestFile], bool]] = KeyDefaultDict(self._build_manifest_evaluator)
-        if snapshot := self._transaction.table_metadata.snapshot_by_name(name=self._target_branch):
-            for manifest_file in snapshot.manifests(io=self._io):
+        table_metadata = self._transaction.table_metadata
+        if snapshot := table_metadata.snapshot_by_name(name=self._target_branch):
+            for manifest_file in snapshot.manifests(io=self._io, table_uuid=table_metadata.table_uuid):
                 # Manifest does not contain rows that match the files to delete partitions
                 if not manifest_evaluators[manifest_file.partition_spec_id](manifest_file):
                     existing_files.append(manifest_file)
@@ -831,7 +833,8 @@ def _deleted_entries(self) -> list[ManifestEntry]:
         which entries are affected.
         """
         if self._parent_snapshot_id is not None:
-            previous_snapshot = self._transaction.table_metadata.snapshot_by_id(self._parent_snapshot_id)
+            table_metadata = self._transaction.table_metadata
+            previous_snapshot = table_metadata.snapshot_by_id(self._parent_snapshot_id)
             if previous_snapshot is None:
                 # This should never happen since you cannot overwrite an empty table
                 raise ValueError(f"Could not find the previous snapshot: {self._parent_snapshot_id}")
@@ -855,7 +858,9 @@ def _get_entries(manifest: ManifestFile) -> list[ManifestEntry]:
                     if entry.data_file.content == DataFileContent.DATA and entry.data_file in self._deleted_data_files
                 ]
 
-            list_of_entries = executor.map(_get_entries, previous_snapshot.manifests(self._io))
+            list_of_entries = executor.map(
+                _get_entries, previous_snapshot.manifests(self._io, table_uuid=table_metadata.table_uuid)
+            )
             deleted_entries = list(itertools.chain(*list_of_entries))
         else:
             deleted_entries = []

pyiceberg/table/update/validate.py

@@ -87,7 +87,7 @@ def _validation_history(
         manifests_files.extend(
             [
                 manifest
-                for manifest in snapshot.manifests(table.io)
+                for manifest in snapshot.manifests(table.io, table_uuid=table.metadata.table_uuid)
                 if manifest.added_snapshot_id == snapshot.snapshot_id and manifest.content == manifest_content_filter
             ]
         )

tests/table/test_validate.py

@@ -15,6 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 # pylint:disable=redefined-outer-name,eval-used
+import uuid
 from typing import cast
 from unittest.mock import patch
 
@@ -77,7 +78,7 @@ def test_validation_history(table_v2_with_extensive_snapshots_and_manifests: tup
         ]
     )
 
-    def mock_read_manifest_side_effect(self: Snapshot, io: FileIO) -> list[ManifestFile]:
+    def mock_read_manifest_side_effect(self: Snapshot, io: FileIO, table_uuid: uuid.UUID | None = None) -> list[ManifestFile]:
         """Mock the manifests method to use the snapshot_id for lookup."""
         snapshot_id = self.snapshot_id
         if snapshot_id in mock_manifests:
@@ -133,7 +134,7 @@ def test_validation_history_fails_on_from_snapshot_not_matching_last_snapshot(
     oldest_snapshot = table.snapshots()[0]
     newest_snapshot = cast(Snapshot, table.current_snapshot())
 
-    def mock_read_manifest_side_effect(self: Snapshot, io: FileIO) -> list[ManifestFile]:
+    def mock_read_manifest_side_effect(self: Snapshot, io: FileIO, table_uuid: uuid.UUID | None = None) -> list[ManifestFile]:
         """Mock the manifests method to use the snapshot_id for lookup."""
         snapshot_id = self.snapshot_id
         if snapshot_id in mock_manifests:
@@ -162,7 +163,7 @@ def test_deleted_data_files(
     oldest_snapshot = table.snapshots()[0]
     newest_snapshot = cast(Snapshot, table.current_snapshot())
 
-    def mock_read_manifest_side_effect(self: Snapshot, io: FileIO) -> list[ManifestFile]:
+    def mock_read_manifest_side_effect(self: Snapshot, io: FileIO, table_uuid: uuid.UUID | None = None) -> list[ManifestFile]:
         """Mock the manifests method to use the snapshot_id for lookup."""
         snapshot_id = self.snapshot_id
         if snapshot_id in mock_manifests:
@@ -257,7 +258,7 @@ def test_validate_added_data_files_conflicting_count(
     boundary_snapshot = table.snapshots()[-(snapshot_history + 1)]
     newest_snapshot = cast(Snapshot, table.current_snapshot())
 
-    def mock_read_manifest_side_effect(self: Snapshot, io: FileIO) -> list[ManifestFile]:
+    def mock_read_manifest_side_effect(self: Snapshot, io: FileIO, table_uuid: uuid.UUID | None = None) -> list[ManifestFile]:
         """Mock the manifests method to use the snapshot_id for lookup."""
         snapshot_id = self.snapshot_id
         if snapshot_id in mock_manifests:
@@ -311,7 +312,7 @@ def test_validate_added_data_files_non_conflicting_count(
     oldest_snapshot = table.snapshots()[-snapshot_history]
     newest_snapshot = cast(Snapshot, table.current_snapshot())
 
-    def mock_read_manifest_side_effect(self: Snapshot, io: FileIO) -> list[ManifestFile]:
+    def mock_read_manifest_side_effect(self: Snapshot, io: FileIO, table_uuid: uuid.UUID | None = None) -> list[ManifestFile]:
         """Mock the manifests method to use the snapshot_id for lookup."""
         snapshot_id = self.snapshot_id
         if snapshot_id in mock_manifests:
@@ -384,7 +385,7 @@ def test_added_delete_files_non_conflicting_count(
     oldest_snapshot = table.snapshots()[-snapshot_history]
     newest_snapshot = cast(Snapshot, table.current_snapshot())
 
-    def mock_read_manifest_side_effect(self: Snapshot, io: FileIO) -> list[ManifestFile]:
+    def mock_read_manifest_side_effect(self: Snapshot, io: FileIO, table_uuid: uuid.UUID | None = None) -> list[ManifestFile]:
         """Mock the manifests method to use the snapshot_id for lookup."""
         snapshot_id = self.snapshot_id
         if snapshot_id in mock_manifests:
@@ -442,7 +443,7 @@ def test_added_delete_files_conflicting_count(
 
     mock_delete_file.spec_id = 0
 
-    def mock_read_manifest_side_effect(self: Snapshot, io: FileIO) -> list[ManifestFile]:
+    def mock_read_manifest_side_effect(self: Snapshot, io: FileIO, table_uuid: uuid.UUID | None = None) -> list[ManifestFile]:
         """Mock the manifests method to use the snapshot_id for lookup."""
         snapshot_id = self.snapshot_id
         if snapshot_id in mock_manifests: