diff --git a/pyiceberg/catalog/__init__.py b/pyiceberg/catalog/__init__.py index 8de113404c..7a5d8b2aaa 100644 --- a/pyiceberg/catalog/__init__.py +++ b/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} diff --git a/pyiceberg/cli/output.py b/pyiceberg/cli/output.py index 4a508d5343..ea78d1b384 100644 --- a/pyiceberg/cli/output.py +++ b/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): diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py index 88ca051015..b7ce2e0bbb 100644 --- a/pyiceberg/manifest.py +++ b/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,6 +1023,9 @@ 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. @@ -1025,7 +1033,7 @@ def _manifests(io: FileIO, manifest_list: str) -> tuple[ManifestFile, ...]: 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]: diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 303b3db135..d77efa0710 100644 --- a/pyiceberg/table/__init__.py +++ b/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: diff --git a/pyiceberg/table/inspect.py b/pyiceberg/table/inspect.py index e24e251fa4..9b6a5e778a 100644 --- a/pyiceberg/table/inspect.py +++ b/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} diff --git a/pyiceberg/table/snapshots.py b/pyiceberg/table/snapshots.py index 0450df2861..adcda2acd9 100644 --- a/pyiceberg/table/snapshots.py +++ b/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): diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index 8024e808b2..235a550f9c 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/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 = [] diff --git a/pyiceberg/table/update/validate.py b/pyiceberg/table/update/validate.py index 0545182bf0..b320b098cb 100644 --- a/pyiceberg/table/update/validate.py +++ b/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 ] ) diff --git a/tests/table/test_validate.py b/tests/table/test_validate.py index a19983fd66..1df081308e 100644 --- a/tests/table/test_validate.py +++ b/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: diff --git a/tests/utils/test_manifest.py b/tests/utils/test_manifest.py index 331146346e..7de936e2e4 100644 --- a/tests/utils/test_manifest.py +++ b/tests/utils/test_manifest.py @@ -16,6 +16,7 @@ # under the License. # pylint: disable=redefined-outer-name,arguments-renamed,fixme import importlib +import uuid from pathlib import Path from tempfile import TemporaryDirectory from typing import Any @@ -509,11 +510,12 @@ def test_read_manifest_v2(generated_manifest_file_file_v2: str) -> None: def test_read_manifest_cache(generated_manifest_file_file_v2: str) -> None: """Test that ManifestFile objects are cached and reused across multiple reads. - The cache now stores individual ManifestFile objects by their manifest_path, + The cache stores individual ManifestFile objects by (table_uuid, manifest_path), rather than caching entire manifest list tuples. This is more memory-efficient when multiple manifest lists share overlapping ManifestFile objects. """ io = load_file_io() + table_uuid = uuid.uuid4() snapshot = Snapshot( snapshot_id=25, @@ -525,8 +527,8 @@ def test_read_manifest_cache(generated_manifest_file_file_v2: str) -> None: ) # Access the manifests property multiple times - manifests_first_call = snapshot.manifests(io) - manifests_second_call = snapshot.manifests(io) + manifests_first_call = snapshot.manifests(io, table_uuid=table_uuid) + manifests_second_call = snapshot.manifests(io, table_uuid=table_uuid) # Ensure that the same manifest list content is returned assert manifests_first_call == manifests_second_call @@ -845,10 +847,11 @@ def test_manifest_cache_deduplicates_manifest_files() -> None: - ManifestList3: (ManifestFile1, ManifestFile2, ManifestFile3) With the old approach, ManifestFile1 was stored 3 times in the cache. - With the new approach, ManifestFile objects are cached individually by their - manifest_path, so ManifestFile1 is stored only once and reused. + With the new approach, ManifestFile objects are cached individually by + (table_uuid, manifest_path), so ManifestFile1 is stored only once and reused. """ io = PyArrowFileIO() + table_uuid = uuid.uuid4() with TemporaryDirectory() as tmp_dir: # Create three manifest files to simulate manifests created during appends @@ -974,9 +977,9 @@ def test_manifest_cache_deduplicates_manifest_files() -> None: list_writer.add_manifests([manifest_file1, manifest_file2, manifest_file3]) # Read all three manifest lists - manifests1 = _manifests(io, manifest_list1_path) - manifests2 = _manifests(io, manifest_list2_path) - manifests3 = _manifests(io, manifest_list3_path) + manifests1 = _manifests(io, manifest_list1_path, table_uuid) + manifests2 = _manifests(io, manifest_list2_path, table_uuid) + manifests3 = _manifests(io, manifest_list3_path, table_uuid) # Verify the manifest files have the expected paths assert len(manifests1) == 1 @@ -1008,6 +1011,7 @@ def test_manifest_cache_efficiency_with_many_overlapping_lists() -> None: manifest lists that increasingly overlap. """ io = PyArrowFileIO() + table_uuid = uuid.uuid4() with TemporaryDirectory() as tmp_dir: schema = Schema(NestedField(field_id=1, name="id", field_type=IntegerType(), required=True)) @@ -1063,7 +1067,7 @@ def test_manifest_cache_efficiency_with_many_overlapping_lists() -> None: # Read all manifest lists all_results = [] for path in manifest_list_paths: - result = _manifests(io, path) + result = _manifests(io, path, table_uuid) all_results.append(result) # With the old cache approach, we would have: @@ -1260,7 +1264,7 @@ def test_clear_manifest_cache() -> None: list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="clear", snapshot_id=1) # Populate the cache - _manifests(io, list_path) + _manifests(io, list_path, uuid.uuid4()) # Verify cache has entries assert len(manifest_module._manifest_cache) > 0, "Cache should have entries after reading manifests" @@ -1282,12 +1286,13 @@ def test_manifest_cache_can_be_disabled_with_size_zero(monkeypatch: pytest.Monke assert len(manifest_module._manifest_cache) == 0 io = PyArrowFileIO() + table_uuid = uuid.uuid4() with TemporaryDirectory() as tmp_dir: list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="disabled", snapshot_id=1) - manifests_first_call = manifest_module._manifests(io, list_path) - manifests_second_call = manifest_module._manifests(io, list_path) + manifests_first_call = manifest_module._manifests(io, list_path, table_uuid) + manifests_second_call = manifest_module._manifests(io, list_path, table_uuid) assert len(manifest_module._manifest_cache) == 0 assert manifests_first_call[0] is not manifests_second_call[0] @@ -1305,18 +1310,19 @@ def test_manifest_cache_respects_positive_env_size(monkeypatch: pytest.MonkeyPat assert manifest_module._manifest_cache.maxsize == 1 io = PyArrowFileIO() + table_uuid = uuid.uuid4() with TemporaryDirectory() as tmp_dir: first_list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="first", snapshot_id=1) second_list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="second", snapshot_id=2) - manifests_first_call = manifest_module._manifests(io, first_list_path) - manifests_second_call = manifest_module._manifests(io, first_list_path) + manifests_first_call = manifest_module._manifests(io, first_list_path, table_uuid) + manifests_second_call = manifest_module._manifests(io, first_list_path, table_uuid) assert manifests_first_call[0] is manifests_second_call[0] assert len(manifest_module._manifest_cache) == 1 - manifest_module._manifests(io, second_list_path) + manifest_module._manifests(io, second_list_path, table_uuid) assert len(manifest_module._manifest_cache) == 1 finally: @@ -1324,6 +1330,89 @@ def test_manifest_cache_respects_positive_env_size(monkeypatch: pytest.MonkeyPat importlib.reload(manifest_module) +def test_manifest_cache_scopes_entries_by_table_uuid() -> None: + """Test that get_or_cache never serves one table's cached content for another table's manifest_path.""" + manifest_path = "s3://bucket/metadata/manifest.avro" + table_a = uuid.uuid4() + table_b = uuid.uuid4() + + manifest_a = ManifestFile.from_args( + manifest_path=manifest_path, + manifest_length=1000, + partition_spec_id=0, + added_snapshot_id=1, + sequence_number=1, + existing_files_count=0, + ) + manifest_b = ManifestFile.from_args( + manifest_path=manifest_path, + manifest_length=1000, + partition_spec_id=0, + added_snapshot_id=2, + sequence_number=1, + existing_files_count=5, + ) + + cached_a = manifest_module._manifest_cache.get_or_cache(manifest_a, table_a) + assert cached_a is manifest_a + + cached_b = manifest_module._manifest_cache.get_or_cache(manifest_b, table_b) + assert cached_b is manifest_b + assert cached_b.added_snapshot_id == 2 + assert cached_b.existing_files_count == 5 + + reread_a = ManifestFile.from_args( + manifest_path=manifest_path, + manifest_length=1000, + partition_spec_id=0, + added_snapshot_id=1, + sequence_number=1, + existing_files_count=0, + ) + assert manifest_module._manifest_cache.get_or_cache(reread_a, table_a) is manifest_a + + +def test_manifest_cache_scoping_through_snapshot_manifests() -> None: + """Test that scoping holds end-to-end through Snapshot.manifests, not just get_or_cache.""" + io = PyArrowFileIO() + table_a = uuid.uuid4() + table_b = uuid.uuid4() + + with TemporaryDirectory() as tmp_dir: + list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="shared", snapshot_id=1) + snapshot = Snapshot( + snapshot_id=1, + timestamp_ms=1602638573590, + manifest_list=list_path, + summary=Summary(Operation.APPEND), + ) + + first = snapshot.manifests(io, table_uuid=table_a) + assert first[0].added_snapshot_id == 1 + + _create_test_manifest_list(manifest_module, io, tmp_dir, name="shared", snapshot_id=2) + + assert snapshot.manifests(io, table_uuid=table_b)[0].added_snapshot_id == 2 + + assert snapshot.manifests(io, table_uuid=table_a)[0] is first[0] + + +def test_manifest_cache_is_bypassed_without_table_uuid() -> None: + """Test that manifests read without a table_uuid are never cached or shared.""" + io = PyArrowFileIO() + + with TemporaryDirectory() as tmp_dir: + list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="unattributed", snapshot_id=1) + entries_before = len(manifest_module._manifest_cache) + + first = manifest_module._manifests(io, list_path) + second = manifest_module._manifests(io, list_path) + + assert len(manifest_module._manifest_cache) == entries_before, "An unattributed read must not populate the cache" + assert first[0] is not second[0], "An unattributed read must not reuse another read's instance" + assert first[0] == second[0] + + def test_manifest_cache_reads_size_from_configuration_file(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: """Test that manifest-cache-size can be loaded from .pyiceberg.yaml.""" config_dir = tmp_path / "config" @@ -1338,15 +1427,16 @@ def test_manifest_cache_reads_size_from_configuration_file(monkeypatch: pytest.M assert manifest_module._manifest_cache.maxsize == 2 io = PyArrowFileIO() + table_uuid = uuid.uuid4() with TemporaryDirectory() as tmp_dir: first_list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="first", snapshot_id=1) second_list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="second", snapshot_id=2) third_list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="third", snapshot_id=3) - manifest_module._manifests(io, first_list_path) - manifest_module._manifests(io, second_list_path) - manifest_module._manifests(io, third_list_path) + manifest_module._manifests(io, first_list_path, table_uuid) + manifest_module._manifests(io, second_list_path, table_uuid) + manifest_module._manifests(io, third_list_path, table_uuid) assert len(manifest_module._manifest_cache) == 2 finally: