Make manifest cache size configurable (#2993)

<!--
Thanks for opening a pull request!
-->

<!-- In the case this PR will resolve an issue, please replace
${GITHUB_ISSUE_ID} below with the actual Github issue id. -->
<!-- Closes #${2952} -->

Closes #2952

# Rationale for this change
Through discussion in issue #2325, we realized that there was a memory
leak in the manifest cache. PR
#2951 fixed this memory
leak, but we decided that it would be best for developer experience if
users could configure the cache for their needs.

This includes the ability to configure the manifest cache size,
including setting it to `0` to disable caching.

## Are these changes tested?
Yes - unit tests are included to verify these changes

## Are there any user-facing changes?
Yes:
- `manifest-cache-size` can be used in `.pyiceberg.yaml` to configure
the size of the manifest cache
- `PYICEBERG_MANIFEST_CACHE_SIZE` can be used to configure the size of
the manifest cache

<!-- In the case of user-facing changes, please add the changelog label.
-->

---------

Co-authored-by: Kevin Liu <kevin.jq.liu@gmail.com>

Author: kris-gaudel

mkdocs/docs/configuration.md

@@ -48,6 +48,29 @@ The environment variable picked up by Iceberg starts with `PYICEBERG_` and then
 
 For example, `PYICEBERG_CATALOG__DEFAULT__S3__ACCESS_KEY_ID`, sets `s3.access-key-id` on the `default` catalog.
 
+## Manifest Caching
+
+PyIceberg caches `ManifestFile` objects locally and uses an LRU policy to bound the cache size. By default, up to `128`
+distinct manifest files are retained.
+
+You can tune the `manifest-cache-size` configuration in `.pyiceberg.yaml`:
+
+```yaml
+manifest-cache-size: 256
+```
+
+Permitted values: any non-negative integer. Set the value to `0` to disable manifest caching entirely.
+
+You can also set it with the `PYICEBERG_MANIFEST_CACHE_SIZE` environment variable:
+
+```sh
+export PYICEBERG_MANIFEST_CACHE_SIZE=256
+```
+
+The memory used by this cache depends on the size and number of distinct manifests your workload touches. Lower the value
+if you want a tighter memory bound, or call `clear_manifest_cache()` to proactively release cached manifest metadata in
+long-lived processes.
+
 ## Tables
 
 Iceberg tables support table properties to configure table behavior.

pyiceberg/manifest.py

@@ -51,6 +51,7 @@
     StringType,
     StructType,
 )
+from pyiceberg.utils.config import Config
 
 UNASSIGNED_SEQ = -1
 DEFAULT_BLOCK_SIZE = 67108864  # 64 * 1024 * 1024
@@ -891,17 +892,70 @@ def __hash__(self) -> int:
         return hash(self.manifest_path)
 
 
-# Global cache for ManifestFile objects, keyed by manifest_path.
-# This deduplicates ManifestFile objects across manifest lists, which commonly
-# share manifests after append operations.
-_manifest_cache: LRUCache[str, ManifestFile] = LRUCache(maxsize=128)
+class _ManifestCache:
+    """Process-wide ManifestFile cache keyed by manifest_path.
 
-# Lock for thread-safe cache access
-_manifest_cache_lock = threading.RLock()
+    Consecutive snapshots often reference the same manifests after append
+    operations, so reusing ManifestFile instances avoids retaining duplicate
+    objects.
+    """
+
+    DEFAULT_SIZE = 128
+
+    _cache: LRUCache[str, ManifestFile] | None
+
+    def __init__(self) -> None:
+        self.maxsize = self._load_configured_size()
+        self._cache = LRUCache(maxsize=self.maxsize) if self.maxsize > 0 else None
+        self._lock = threading.RLock()
+
+    @classmethod
+    def _load_configured_size(cls) -> int:
+        configured_size = Config().get_int("manifest-cache-size")
+        if configured_size is None:
+            return cls.DEFAULT_SIZE
+        if configured_size < 0:
+            raise ValueError(
+                f"manifest-cache-size should be a non-negative integer or left unset. Current value: {configured_size}"
+            )
+        return configured_size
+
+    def clear(self) -> None:
+        with self._lock:
+            if self._cache is not None:
+                self._cache.clear()
+
+    def get_or_cache(self, manifest_file: ManifestFile) -> ManifestFile:
+        if self._cache 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]
+
+            self._cache[manifest_path] = manifest_file
+            return manifest_file
+
+    def __len__(self) -> int:
+        with self._lock:
+            return len(self._cache) if self._cache is not None else 0
+
+
+_manifest_cache = _ManifestCache()
+
+
+def clear_manifest_cache() -> None:
+    """Clear cached ManifestFile objects.
+
+    This is primarily useful in long-lived or memory-sensitive processes that
+    want to release cached manifest metadata between bursts of table reads.
+    """
+    _manifest_cache.clear()
 
 
 def _manifests(io: FileIO, manifest_list: str) -> tuple[ManifestFile, ...]:
-    """Read manifests from a manifest list, deduplicating ManifestFile objects via cache.
+    """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:
@@ -927,17 +981,7 @@ def _manifests(io: FileIO, manifest_list: str) -> tuple[ManifestFile, ...]:
     file = io.new_input(manifest_list)
     manifest_files = list(read_manifest_list(file))
 
-    result = []
-    with _manifest_cache_lock:
-        for manifest_file in manifest_files:
-            manifest_path = manifest_file.manifest_path
-            if manifest_path in _manifest_cache:
-                result.append(_manifest_cache[manifest_path])
-            else:
-                _manifest_cache[manifest_path] = manifest_file
-                result.append(manifest_file)
-
-    return tuple(result)
+    return tuple(_manifest_cache.get_or_cache(manifest_file) for manifest_file in manifest_files)
 
 
 def read_manifest_list(input_file: InputFile) -> Iterator[ManifestFile]:

tests/benchmark/test_memory_benchmark.py

@@ -32,8 +32,9 @@
 import pyarrow as pa
 import pytest
 
+from pyiceberg import manifest as manifest_module
 from pyiceberg.catalog.memory import InMemoryCatalog
-from pyiceberg.manifest import _manifest_cache
+from pyiceberg.manifest import clear_manifest_cache
 
 
 def generate_test_dataframe() -> pa.Table:
@@ -64,16 +65,16 @@ def memory_catalog(tmp_path_factory: pytest.TempPathFactory) -> InMemoryCatalog:
 @pytest.fixture(autouse=True)
 def clear_caches() -> None:
     """Clear caches before each test."""
-    _manifest_cache.clear()
+    clear_manifest_cache()
     gc.collect()
 
 
 @pytest.mark.benchmark
 def test_manifest_cache_memory_growth(memory_catalog: InMemoryCatalog) -> None:
     """Benchmark memory growth of manifest cache during repeated appends.
 
-    This test reproduces the issue from GitHub #2325 where each append creates
-    a new manifest list entry in the cache, causing memory to grow.
+    This test reproduces the issue from GitHub #2325 where the old cache stored
+    each manifest list result, causing memory to grow.
 
     With the old caching strategy (tuple per manifest list), memory grew as O(N²).
     With the new strategy (individual ManifestFile objects), memory grows as O(N).
@@ -95,7 +96,7 @@ def test_manifest_cache_memory_growth(memory_catalog: InMemoryCatalog) -> None:
         # Sample memory at intervals
         if (i + 1) % 10 == 0:
             current, _ = tracemalloc.get_traced_memory()
-            cache_size = len(_manifest_cache)
+            cache_size = len(manifest_module._manifest_cache)
 
             memory_samples.append((i + 1, current, cache_size))
             print(f"  Iteration {i + 1}: Memory={current / 1024:.1f} KB, Cache entries={cache_size}")
@@ -150,13 +151,13 @@ def test_memory_after_gc_with_cache_cleared(memory_catalog: InMemoryCatalog) ->
 
     gc.collect()
     before_clear_memory, _ = tracemalloc.get_traced_memory()
-    cache_size_before = len(_manifest_cache)
+    cache_size_before = len(manifest_module._manifest_cache)
     print(f"  Memory before clear: {before_clear_memory / 1024:.1f} KB")
     print(f"  Cache size: {cache_size_before}")
 
     # Phase 2: Clear cache and GC
     print("\nPhase 2: Clearing cache and running GC...")
-    _manifest_cache.clear()
+    clear_manifest_cache()
     gc.collect()
     gc.collect()  # Multiple GC passes for thorough cleanup
 
@@ -192,6 +193,7 @@ def test_manifest_cache_deduplication_efficiency() -> None:
         ManifestEntry,
         ManifestEntryStatus,
         _manifests,
+        clear_manifest_cache,
         write_manifest,
         write_manifest_list,
     )
@@ -245,7 +247,7 @@ def test_manifest_cache_deduplication_efficiency() -> None:
         num_lists = 10
         print(f"Creating {num_lists} manifest lists with overlapping manifests...")
 
-        _manifest_cache.clear()
+        clear_manifest_cache()
 
         for i in range(num_lists):
             list_path = f"{tmp_dir}/manifest-list_{i}.avro"
@@ -265,7 +267,7 @@ def test_manifest_cache_deduplication_efficiency() -> None:
             _manifests(io, list_path)
 
         # Analyze cache efficiency
-        cache_entries = len(_manifest_cache)
+        cache_entries = len(manifest_module._manifest_cache)
         # List i contains manifests 0..i, so only the first num_lists manifests are actually used
         manifests_actually_used = num_lists