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

The process-wide cache was keyed on manifest_path alone, and
ManifestFile equality compares only that path, so the first read of a
path won and any later read of the same path returned the earlier
object. Two tables referencing one path therefore shared a single
ManifestFile, and since its partition summaries, counts and sequence
numbers drive scan pruning, a mismatched entry changed which files a
scan considered.

Compare the full record before reusing a cached instance, falling back
to Record's structural equality, and replace the entry when the
content differs. Genuine reuse across consecutive snapshots still hits
the cache.

Closes #3978

Author: ArulJerald

pyiceberg/manifest.py

@@ -941,7 +941,8 @@ class _ManifestCache:
 
     Consecutive snapshots often reference the same manifests after append
     operations, so reusing ManifestFile instances avoids retaining duplicate
-    objects.
+    objects. A path that matches but carries different content is detected,
+    so a collision does not result in stale data being served.
     """
 
     DEFAULT_SIZE = 128
@@ -975,8 +976,11 @@ def get_or_cache(self, manifest_file: ManifestFile) -> ManifestFile:
 
         with self._lock:
             manifest_path = manifest_file.manifest_path
-            if manifest_path in self._cache:
-                return self._cache[manifest_path]
+            cached = self._cache.get(manifest_path)
+            # ManifestFile equality compares manifest_path only, so fall back to the
+            # full record to guard against a path collision with different content.
+            if cached is not None and cached._data == manifest_file._data:
+                return cached
 
             self._cache[manifest_path] = manifest_file
             return manifest_file

tests/utils/test_manifest.py

@@ -1324,6 +1324,46 @@ def test_manifest_cache_respects_positive_env_size(monkeypatch: pytest.MonkeyPat
         importlib.reload(manifest_module)
 
 
+def test_manifest_cache_detects_path_collision_with_different_content() -> None:
+    """Test that get_or_cache does not serve stale content for a manifest_path collision."""
+    manifest_path = "s3://bucket/metadata/manifest.avro"
+    first = 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,
+    )
+    second = 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_first = manifest_module._manifest_cache.get_or_cache(first)
+    assert cached_first is first
+
+    cached_second = manifest_module._manifest_cache.get_or_cache(second)
+    assert cached_second is second
+    assert cached_second.added_snapshot_id == 2
+    assert cached_second.existing_files_count == 5
+
+    identical = 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_identical = manifest_module._manifest_cache.get_or_cache(identical)
+    assert cached_identical is second
+
+
 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"