From 4d6ea75c00d65f448e3b880fe372ca9727f7beaf Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Sun, 19 Jul 2026 16:23:31 +0800
Subject: [PATCH 1/9] [python] Cache Parquet metadata across reads

---
 .../pypaimon/common/options/core_options.py   |  24 +++
 .../read/reader/format_pyarrow_reader.py      | 123 +++++++++++++-
 .../tests/parquet_metadata_cache_test.py      | 156 ++++++++++++++++++
 3 files changed, 300 insertions(+), 3 deletions(-)
 create mode 100644 paimon-python/pypaimon/tests/parquet_metadata_cache_test.py

diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py
index 4d2e6a7bb8bc..386ce2796f77 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -264,6 +264,24 @@ class CoreOptions:
         .with_description("Specify the message format of data files.")
     )
 
+    PARQUET_METADATA_CACHE_ENABLED: ConfigOption[bool] = (
+        ConfigOptions.key("parquet.metadata-cache-enabled")
+        .boolean_type()
+        .default_value(False)
+        .with_description(
+            "Cache immutable Parquet file metadata across reads in the current process."
+        )
+    )
+
+    PARQUET_METADATA_CACHE_SIZE: ConfigOption[int] = (
+        ConfigOptions.key("parquet.metadata-cache-size")
+        .int_type()
+        .default_value(256)
+        .with_description(
+            "Maximum number of Parquet metadata entries cached in the current process."
+        )
+    )
+
     FILE_COMPRESSION: ConfigOption[str] = (
         ConfigOptions.key("file.compression")
         .string_type()
@@ -1023,6 +1041,12 @@ def manifest_merge_min_count(self, default=None):
     def file_format(self, default=None):
         return self.options.get(CoreOptions.FILE_FORMAT, default)
 
+    def parquet_metadata_cache_enabled(self) -> bool:
+        return self.options.get(CoreOptions.PARQUET_METADATA_CACHE_ENABLED, False)
+
+    def parquet_metadata_cache_size(self) -> int:
+        return self.options.get(CoreOptions.PARQUET_METADATA_CACHE_SIZE, 256)
+
     def file_compression(self, default=None):
         return self.options.get(CoreOptions.FILE_COMPRESSION, default)
 
diff --git a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
index efaf8e0546b1..202898a7ff4c 100644
--- a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
@@ -15,7 +15,11 @@
 # specific language governing permissions and limitations
 # under the License.
 
-from typing import Any, Dict, List, Optional
+import os
+import threading
+from collections import OrderedDict
+from concurrent.futures import Future
+from typing import Any, Callable, Dict, List, Optional, Tuple
 
 import pyarrow as pa
 import pyarrow.dataset as ds
@@ -34,6 +38,105 @@
 from pypaimon.table.special_fields import SpecialFields
 
 
+class _ParquetDatasetCache:
+    def __init__(self, max_entries: int):
+        self.max_entries = max_entries
+        self._entries = OrderedDict()
+        self._loads = {}
+        self._lock = threading.Lock()
+
+    def get_or_load(self, key: Tuple[int, str], loader: Callable[[], Any]):
+        with self._lock:
+            dataset = self._entries.get(key)
+            if dataset is not None:
+                self._entries.move_to_end(key)
+                return dataset
+
+            future = self._loads.get(key)
+            if future is None:
+                future = Future()
+                self._loads[key] = future
+                should_load = True
+            else:
+                should_load = False
+
+        if not should_load:
+            return future.result()
+
+        try:
+            dataset = loader()
+        except BaseException as exception:
+            with self._lock:
+                self._loads.pop(key, None)
+            future.set_exception(exception)
+            raise
+
+        with self._lock:
+            self._entries[key] = dataset
+            self._entries.move_to_end(key)
+            while len(self._entries) > self.max_entries:
+                self._entries.popitem(last=False)
+            self._loads.pop(key, None)
+        future.set_result(dataset)
+        return dataset
+
+    def resize(self, max_entries: int):
+        with self._lock:
+            self.max_entries = max_entries
+            while len(self._entries) > self.max_entries:
+                self._entries.popitem(last=False)
+
+
+_PARQUET_DATASET_CACHE = None
+_PARQUET_DATASET_CACHE_LOCK = threading.Lock()
+_PARQUET_DATASET_CACHE_PID = os.getpid()
+
+
+def _ensure_parquet_dataset_cache_process():
+    global _PARQUET_DATASET_CACHE
+    global _PARQUET_DATASET_CACHE_LOCK
+    global _PARQUET_DATASET_CACHE_PID
+    current_pid = os.getpid()
+    if current_pid != _PARQUET_DATASET_CACHE_PID:
+        _PARQUET_DATASET_CACHE = None
+        _PARQUET_DATASET_CACHE_LOCK = threading.Lock()
+        _PARQUET_DATASET_CACHE_PID = current_pid
+
+
+def _global_parquet_dataset_cache(max_entries: int) -> _ParquetDatasetCache:
+    global _PARQUET_DATASET_CACHE
+    _ensure_parquet_dataset_cache_process()
+    with _PARQUET_DATASET_CACHE_LOCK:
+        if _PARQUET_DATASET_CACHE is None:
+            _PARQUET_DATASET_CACHE = _ParquetDatasetCache(max_entries)
+        elif _PARQUET_DATASET_CACHE.max_entries != max_entries:
+            _PARQUET_DATASET_CACHE.resize(max_entries)
+        return _PARQUET_DATASET_CACHE
+
+
+def _reset_parquet_dataset_cache():
+    global _PARQUET_DATASET_CACHE
+    _ensure_parquet_dataset_cache_process()
+    with _PARQUET_DATASET_CACHE_LOCK:
+        _PARQUET_DATASET_CACHE = None
+
+
+def _parquet_dataset(file_io: FileIO, file_path: str, cache_enabled: bool,
+                     cache_size: int):
+    file_path_for_pyarrow = file_io.to_filesystem_path(file_path)
+    filesystem = file_io.filesystem
+
+    def load():
+        return ds.dataset(
+            file_path_for_pyarrow, format='parquet', filesystem=filesystem)
+
+    if not cache_enabled or cache_size <= 0:
+        return load()
+
+    key = (id(filesystem), file_path_for_pyarrow)
+    return _global_parquet_dataset_cache(cache_size).get_or_load(key, load)
+
+
 class FormatPyArrowReader(RecordBatchReader):
     """
     A Format Reader that reads record batch from a Parquet or ORC file using PyArrow,
@@ -49,8 +152,22 @@ def __init__(self, file_io: FileIO, file_format: str, file_path: str,
                  push_down_predicate: Any, batch_size: int = 1024,
                  options: CoreOptions = None,
                  nested_name_paths: Optional[List[List[str]]] = None):
-        file_path_for_pyarrow = file_io.to_filesystem_path(file_path)
-        self.dataset = ds.dataset(file_path_for_pyarrow, format=file_format, filesystem=file_io.filesystem)
+        if file_format == 'parquet':
+            cache_enabled = (
+                options.parquet_metadata_cache_enabled()
+                if options is not None else False
+            )
+            cache_size = (
+                options.parquet_metadata_cache_size()
+                if options is not None else 256
+            )
+            self.dataset = _parquet_dataset(
+                file_io, file_path, cache_enabled, cache_size)
+        else:
+            file_path_for_pyarrow = file_io.to_filesystem_path(file_path)
+            self.dataset = ds.dataset(
+                file_path_for_pyarrow, format=file_format,
+                filesystem=file_io.filesystem)
         self._file_format = file_format
         self.read_fields = read_fields
         self._read_field_names = [f.name for f in read_fields]
diff --git a/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py b/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
new file mode 100644
index 000000000000..ba6fe596b378
--- /dev/null
+++ b/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
@@ -0,0 +1,156 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+import tempfile
+import time
+import unittest
+from concurrent.futures import ThreadPoolExecutor
+from unittest.mock import patch
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+
+from pypaimon.common.options import Options
+from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.filesystem.local_file_io import LocalFileIO
+from pypaimon.read.reader import format_pyarrow_reader as reader_module
+from pypaimon.read.reader.format_pyarrow_reader import FormatPyArrowReader
+from pypaimon.schema.data_types import AtomicType, DataField
+
+
+class ParquetMetadataCacheTest(unittest.TestCase):
+    def setUp(self):
+        reader_module._reset_parquet_dataset_cache()
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.file_io = LocalFileIO(self.temp_dir.name, Options({}))
+        self.paths = []
+        for index in range(3):
+            path = os.path.join(self.temp_dir.name, "data-{}.parquet".format(index))
+            pq.write_table(
+                pa.table({"value": list(range(index * 10, index * 10 + 10))}),
+                path,
+                row_group_size=2,
+            )
+            self.paths.append(path)
+
+    def tearDown(self):
+        reader_module._reset_parquet_dataset_cache()
+        self.temp_dir.cleanup()
+
+    @staticmethod
+    def _options(enabled, size=256):
+        return CoreOptions(Options({
+            "parquet.metadata-cache-enabled": enabled,
+            "parquet.metadata-cache-size": size,
+        }))
+
+    def _read(self, path, options):
+        reader = FormatPyArrowReader(
+            self.file_io,
+            "parquet",
+            path,
+            [DataField(0, "value", AtomicType("BIGINT"))],
+            None,
+            options=options,
+        )
+        values = []
+        try:
+            while True:
+                batch = reader.read_arrow_batch()
+                if batch is None:
+                    return values
+                values.extend(batch.column(0).to_pylist())
+        finally:
+            reader.close()
+
+    def test_disabled_by_default(self):
+        options = CoreOptions(Options({}))
+        self.assertFalse(options.parquet_metadata_cache_enabled())
+        self.assertEqual(256, options.parquet_metadata_cache_size())
+
+        original = reader_module.ds.dataset
+        with patch.object(reader_module.ds, "dataset", wraps=original) as dataset:
+            self._read(self.paths[0], options)
+            self._read(self.paths[0], options)
+        self.assertEqual(2, dataset.call_count)
+
+    def test_reuses_dataset(self):
+        options = self._options(True)
+        original = reader_module.ds.dataset
+        with patch.object(reader_module.ds, "dataset", wraps=original) as dataset:
+            first = self._read(self.paths[0], options)
+            second = self._read(self.paths[0], options)
+
+        self.assertEqual(list(range(10)), first)
+        self.assertEqual(first, second)
+        self.assertEqual(1, dataset.call_count)
+
+    def test_evicts_least_recently_used_entry(self):
+        options = self._options(True, size=2)
+        original = reader_module.ds.dataset
+        with patch.object(reader_module.ds, "dataset", wraps=original) as dataset:
+            for path in self.paths:
+                self._read(path, options)
+            self._read(self.paths[0], options)
+        self.assertEqual(4, dataset.call_count)
+
+    def test_resizes_existing_cache(self):
+        original = reader_module.ds.dataset
+        with patch.object(reader_module.ds, "dataset", wraps=original) as dataset:
+            for path in self.paths:
+                self._read(path, self._options(True, size=3))
+            self._read(self.paths[2], self._options(True, size=1))
+            self._read(self.paths[0], self._options(True, size=1))
+        self.assertEqual(4, dataset.call_count)
+
+    def test_does_not_share_across_filesystems(self):
+        other_file_io = LocalFileIO(self.temp_dir.name, Options({}))
+        original = reader_module.ds.dataset
+        with patch.object(reader_module.ds, "dataset", wraps=original) as dataset:
+            reader_module._parquet_dataset(
+                self.file_io, self.paths[0], True, 256)
+            reader_module._parquet_dataset(
+                other_file_io, self.paths[0], True, 256)
+        self.assertEqual(2, dataset.call_count)
+
+    def test_resets_after_process_change(self):
+        parent_cache = reader_module._global_parquet_dataset_cache(256)
+        with patch.object(reader_module.os, "getpid", return_value=os.getpid() + 1):
+            child_cache = reader_module._global_parquet_dataset_cache(256)
+        self.assertIsNot(parent_cache, child_cache)
+
+    def test_coalesces_concurrent_loads(self):
+        original = reader_module.ds.dataset
+
+        def delayed_dataset(*args, **kwargs):
+            time.sleep(0.05)
+            return original(*args, **kwargs)
+
+        with patch.object(
+                reader_module.ds, "dataset", side_effect=delayed_dataset) as dataset:
+            with ThreadPoolExecutor(max_workers=8) as executor:
+                results = list(executor.map(
+                    lambda _: self._read(self.paths[0], self._options(True)),
+                    range(8),
+                ))
+
+        self.assertEqual(1, dataset.call_count)
+        self.assertTrue(all(value == list(range(10)) for value in results))
+
+
+if __name__ == "__main__":
+    unittest.main()

From 8e52cb2823d3d066fe42fe2ff3e2611a3864c7ce Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Sun, 19 Jul 2026 16:39:38 +0800
Subject: [PATCH 2/9] [python] Verify Parquet metadata cache IO savings

---
 .../read/reader/format_pyarrow_reader.py      |  1 +
 .../tests/parquet_metadata_cache_test.py      | 74 ++++++++++++++++++-
 2 files changed, 73 insertions(+), 2 deletions(-)

diff --git a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
index 202898a7ff4c..62f69e4a6e0c 100644
--- a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
@@ -133,6 +133,7 @@ def load():
     if not cache_enabled or cache_size <= 0:
         return load()
 
+    # FileIO keeps this instance stable, and the cached Dataset retains it.
     key = (id(filesystem), file_path_for_pyarrow)
     return _global_parquet_dataset_cache(cache_size).get_or_load(key, load)
 
diff --git a/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py b/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
index ba6fe596b378..7da26f080da0 100644
--- a/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
+++ b/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
@@ -22,7 +22,9 @@
 from unittest.mock import patch
 
 import pyarrow as pa
+import pyarrow.fs as pafs
 import pyarrow.parquet as pq
+from fsspec.implementations.local import LocalFileSystem as FsspecLocalFileSystem
 
 from pypaimon.common.options import Options
 from pypaimon.common.options.core_options import CoreOptions
@@ -32,6 +34,45 @@
 from pypaimon.schema.data_types import AtomicType, DataField
 
 
+class _CountingInputFile:
+    def __init__(self, wrapped, file_system):
+        self._wrapped = wrapped
+        self._file_system = file_system
+
+    def read(self, size=-1):
+        offset = self._wrapped.tell()
+        data = self._wrapped.read(size)
+        self._file_system.reads.append((offset, len(data)))
+        return data
+
+    def readinto(self, buffer):
+        offset = self._wrapped.tell()
+        size = self._wrapped.readinto(buffer)
+        self._file_system.reads.append((offset, size))
+        return size
+
+    def __getattr__(self, name):
+        return getattr(self._wrapped, name)
+
+
+class _CountingLocalFileSystem(FsspecLocalFileSystem):
+    def __init__(self):
+        super().__init__()
+        self.opens = 0
+        self.reads = []
+
+    def _open(self, path, mode="rb", **kwargs):
+        wrapped = super()._open(path, mode=mode, **kwargs)
+        if "r" not in mode:
+            return wrapped
+        self.opens += 1
+        return _CountingInputFile(wrapped, self)
+
+    def reset_counts(self):
+        self.opens = 0
+        self.reads = []
+
+
 class ParquetMetadataCacheTest(unittest.TestCase):
     def setUp(self):
         reader_module._reset_parquet_dataset_cache()
@@ -58,9 +99,9 @@ def _options(enabled, size=256):
             "parquet.metadata-cache-size": size,
         }))
 
-    def _read(self, path, options):
+    def _read(self, path, options, file_io=None):
         reader = FormatPyArrowReader(
-            self.file_io,
+            file_io or self.file_io,
             "parquet",
             path,
             [DataField(0, "value", AtomicType("BIGINT"))],
@@ -99,6 +140,35 @@ def test_reuses_dataset(self):
         self.assertEqual(first, second)
         self.assertEqual(1, dataset.call_count)
 
+    def test_repeated_scan_skips_footer_io(self):
+        path = os.path.join(self.temp_dir.name, "footer-io.parquet")
+        pq.write_table(
+            pa.table({
+                "value": list(range(10000)),
+                "payload": ["x" * 100] * 10000,
+            }),
+            path,
+            row_group_size=100,
+            compression="none",
+        )
+        counting = _CountingLocalFileSystem()
+        file_io = LocalFileIO(self.temp_dir.name, Options({}))
+        file_io.filesystem = pafs.PyFileSystem(pafs.FSSpecHandler(counting))
+
+        uncached = self._read(path, self._options(False), file_io)
+        uncached_opens = counting.opens
+        uncached_reads = len(counting.reads)
+
+        counting.reset_counts()
+        reader_module._reset_parquet_dataset_cache()
+        self._read(path, self._options(True), file_io)
+        counting.reset_counts()
+        cached = self._read(path, self._options(True), file_io)
+
+        self.assertEqual(uncached, cached)
+        self.assertLess(counting.opens, uncached_opens)
+        self.assertLess(len(counting.reads), uncached_reads)
+
     def test_evicts_least_recently_used_entry(self):
         options = self._options(True, size=2)
         original = reader_module.ds.dataset

From ce981248f61f09da35b8d57eb31efd51e6be5a53 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Sun, 19 Jul 2026 17:12:03 +0800
Subject: [PATCH 3/9] [python] Harden Parquet metadata cache isolation

---
 .../pypaimon/common/options/core_options.py   |  3 +-
 .../read/reader/format_pyarrow_reader.py      | 53 +++++++++------
 .../tests/parquet_metadata_cache_test.py      | 67 +++++++++++++++++--
 3 files changed, 98 insertions(+), 25 deletions(-)

diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py
index 386ce2796f77..c548f99dd05b 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -278,7 +278,8 @@ class CoreOptions:
         .int_type()
         .default_value(256)
         .with_description(
-            "Maximum number of Parquet metadata entries cached in the current process."
+            "Maximum number of Parquet metadata entries cached per FileIO in the current "
+            "process. When readers share a FileIO, the largest requested value is used."
         )
     )
 
diff --git a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
index 62f69e4a6e0c..9ae3df54d687 100644
--- a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
@@ -17,6 +17,7 @@
 
 import os
 import threading
+import weakref
 from collections import OrderedDict
 from concurrent.futures import Future
 from typing import Any, Callable, Dict, List, Optional, Tuple
@@ -38,6 +39,20 @@
 from pypaimon.table.special_fields import SpecialFields
 
 
+class _FilesystemIdentity:
+    def __init__(self, filesystem):
+        self.filesystem = filesystem
+
+    def __hash__(self):
+        return id(self.filesystem)
+
+    def __eq__(self, other):
+        return (
+            isinstance(other, _FilesystemIdentity)
+            and self.filesystem is other.filesystem
+        )
+
+
 class _ParquetDatasetCache:
     def __init__(self, max_entries: int):
         self.max_entries = max_entries
@@ -45,7 +60,7 @@ def __init__(self, max_entries: int):
         self._loads = {}
         self._lock = threading.Lock()
 
-    def get_or_load(self, key: Tuple[int, str], loader: Callable[[], Any]):
+    def get_or_load(self, key: Tuple[Any, str], loader: Callable[[], Any]):
         with self._lock:
             dataset = self._entries.get(key)
             if dataset is not None:
@@ -80,45 +95,44 @@ def get_or_load(self, key: Tuple[int, str], loader: Callable[[], Any]):
         future.set_result(dataset)
         return dataset
 
-    def resize(self, max_entries: int):
+    def ensure_capacity(self, max_entries: int):
         with self._lock:
-            self.max_entries = max_entries
-            while len(self._entries) > self.max_entries:
-                self._entries.popitem(last=False)
+            self.max_entries = max(self.max_entries, max_entries)
 
 
-_PARQUET_DATASET_CACHE = None
+_PARQUET_DATASET_CACHES = weakref.WeakKeyDictionary()
 _PARQUET_DATASET_CACHE_LOCK = threading.Lock()
 _PARQUET_DATASET_CACHE_PID = os.getpid()
 
 
 def _ensure_parquet_dataset_cache_process():
-    global _PARQUET_DATASET_CACHE
+    global _PARQUET_DATASET_CACHES
     global _PARQUET_DATASET_CACHE_LOCK
     global _PARQUET_DATASET_CACHE_PID
     current_pid = os.getpid()
     if current_pid != _PARQUET_DATASET_CACHE_PID:
-        _PARQUET_DATASET_CACHE = None
+        _PARQUET_DATASET_CACHES = weakref.WeakKeyDictionary()
         _PARQUET_DATASET_CACHE_LOCK = threading.Lock()
         _PARQUET_DATASET_CACHE_PID = current_pid
 
 
-def _global_parquet_dataset_cache(max_entries: int) -> _ParquetDatasetCache:
-    global _PARQUET_DATASET_CACHE
+def _parquet_dataset_cache(file_io: FileIO, max_entries: int) -> _ParquetDatasetCache:
     _ensure_parquet_dataset_cache_process()
     with _PARQUET_DATASET_CACHE_LOCK:
-        if _PARQUET_DATASET_CACHE is None:
-            _PARQUET_DATASET_CACHE = _ParquetDatasetCache(max_entries)
-        elif _PARQUET_DATASET_CACHE.max_entries != max_entries:
-            _PARQUET_DATASET_CACHE.resize(max_entries)
-        return _PARQUET_DATASET_CACHE
+        cache = _PARQUET_DATASET_CACHES.get(file_io)
+        if cache is None:
+            cache = _ParquetDatasetCache(max_entries)
+            _PARQUET_DATASET_CACHES[file_io] = cache
+        else:
+            cache.ensure_capacity(max_entries)
+        return cache
 
 
 def _reset_parquet_dataset_cache():
-    global _PARQUET_DATASET_CACHE
+    global _PARQUET_DATASET_CACHES
     _ensure_parquet_dataset_cache_process()
     with _PARQUET_DATASET_CACHE_LOCK:
-        _PARQUET_DATASET_CACHE = None
+        _PARQUET_DATASET_CACHES = weakref.WeakKeyDictionary()
 
 
 def _parquet_dataset(file_io: FileIO, file_path: str, cache_enabled: bool,
@@ -133,9 +147,8 @@ def load():
     if not cache_enabled or cache_size <= 0:
         return load()
 
-    # FileIO keeps this instance stable, and the cached Dataset retains it.
-    key = (id(filesystem), file_path_for_pyarrow)
-    return _global_parquet_dataset_cache(cache_size).get_or_load(key, load)
+    key = (_FilesystemIdentity(filesystem), file_path_for_pyarrow)
+    return _parquet_dataset_cache(file_io, cache_size).get_or_load(key, load)
 
 
 class FormatPyArrowReader(RecordBatchReader):
diff --git a/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py b/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
index 7da26f080da0..7a770b182e40 100644
--- a/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
+++ b/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
@@ -14,10 +14,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import gc
 import os
 import tempfile
 import time
 import unittest
+import weakref
 from concurrent.futures import ThreadPoolExecutor
 from unittest.mock import patch
 
@@ -178,14 +180,26 @@ def test_evicts_least_recently_used_entry(self):
             self._read(self.paths[0], options)
         self.assertEqual(4, dataset.call_count)
 
-    def test_resizes_existing_cache(self):
+    def test_cache_capacity_does_not_shrink(self):
         original = reader_module.ds.dataset
         with patch.object(reader_module.ds, "dataset", wraps=original) as dataset:
             for path in self.paths:
                 self._read(path, self._options(True, size=3))
             self._read(self.paths[2], self._options(True, size=1))
             self._read(self.paths[0], self._options(True, size=1))
-        self.assertEqual(4, dataset.call_count)
+        self.assertEqual(3, dataset.call_count)
+        self.assertEqual(
+            3, reader_module._parquet_dataset_cache(self.file_io, 1).max_entries)
+
+    def test_cache_capacity_is_isolated_by_file_io(self):
+        other_file_io = LocalFileIO(self.temp_dir.name, Options({}))
+
+        first_cache = reader_module._parquet_dataset_cache(self.file_io, 3)
+        second_cache = reader_module._parquet_dataset_cache(other_file_io, 1)
+
+        self.assertIsNot(first_cache, second_cache)
+        self.assertEqual(3, first_cache.max_entries)
+        self.assertEqual(1, second_cache.max_entries)
 
     def test_does_not_share_across_filesystems(self):
         other_file_io = LocalFileIO(self.temp_dir.name, Options({}))
@@ -197,10 +211,55 @@ def test_does_not_share_across_filesystems(self):
                 other_file_io, self.paths[0], True, 256)
         self.assertEqual(2, dataset.call_count)
 
+    def test_cache_key_retains_filesystem_wrapper(self):
+        root = pafs.LocalFileSystem()
+        filesystem = pafs.SubTreeFileSystem(self.temp_dir.name, root)
+        filesystem_ref = weakref.ref(filesystem)
+        file_io = LocalFileIO(self.temp_dir.name, Options({}))
+        file_io.filesystem = filesystem
+
+        reader_module._parquet_dataset(
+            file_io, os.path.basename(self.paths[0]), True, 256)
+        file_io.filesystem = root
+        del filesystem
+        gc.collect()
+
+        self.assertIsNotNone(filesystem_ref())
+
+    def test_filesystem_hash_collision_does_not_share_dataset(self):
+        first_dir = tempfile.TemporaryDirectory()
+        second_dir = tempfile.TemporaryDirectory()
+        try:
+            file_name = "same.parquet"
+            pq.write_table(
+                pa.table({"value": [1]}), os.path.join(first_dir.name, file_name))
+            pq.write_table(
+                pa.table({"value": [2]}), os.path.join(second_dir.name, file_name))
+            file_io = LocalFileIO(first_dir.name, Options({}))
+            first_filesystem = pafs.SubTreeFileSystem(
+                first_dir.name, pafs.LocalFileSystem())
+            second_filesystem = pafs.SubTreeFileSystem(
+                second_dir.name, pafs.LocalFileSystem())
+
+            with patch.object(
+                    reader_module._FilesystemIdentity, "__hash__", return_value=1):
+                file_io.filesystem = first_filesystem
+                first = reader_module._parquet_dataset(
+                    file_io, file_name, True, 256).to_table()
+                file_io.filesystem = second_filesystem
+                second = reader_module._parquet_dataset(
+                    file_io, file_name, True, 256).to_table()
+
+            self.assertEqual([1], first.column("value").to_pylist())
+            self.assertEqual([2], second.column("value").to_pylist())
+        finally:
+            first_dir.cleanup()
+            second_dir.cleanup()
+
     def test_resets_after_process_change(self):
-        parent_cache = reader_module._global_parquet_dataset_cache(256)
+        parent_cache = reader_module._parquet_dataset_cache(self.file_io, 256)
         with patch.object(reader_module.os, "getpid", return_value=os.getpid() + 1):
-            child_cache = reader_module._global_parquet_dataset_cache(256)
+            child_cache = reader_module._parquet_dataset_cache(self.file_io, 256)
         self.assertIsNot(parent_cache, child_cache)
 
     def test_coalesces_concurrent_loads(self):

From 275c6cc9a9f8d54bb7a6c01481495b6b5ad2877b Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Mon, 20 Jul 2026 14:26:53 +0800
Subject: [PATCH 4/9] [python] Document Parquet metadata cache options

---
 docs/docs/pypaimon/pytorch.md                 | 23 +++++++++++++++++++
 .../pypaimon/common/options/core_options.py   | 15 +++++++-----
 .../read/reader/format_pyarrow_reader.py      | 13 ++++++-----
 .../tests/parquet_metadata_cache_test.py      | 14 +++++------
 4 files changed, 46 insertions(+), 19 deletions(-)

diff --git a/docs/docs/pypaimon/pytorch.md b/docs/docs/pypaimon/pytorch.md
index 0f7f7bbdef6d..9c87d635773d 100644
--- a/docs/docs/pypaimon/pytorch.md
+++ b/docs/docs/pypaimon/pytorch.md
@@ -59,6 +59,29 @@ when it is false, it will read the full amount of data into memory.
 
 **`prefetch_concurrency`** (default: 1): When streaming is true, number of threads used for parallel prefetch within each DataLoader worker. Set to a value greater than 1 to partition splits across threads and increase read throughput. Has no effect when streaming is false.
 
+## Parquet Metadata Cache
+
+Long-lived DataLoader workers may read the same immutable Parquet files across
+multiple epochs. Enable the process-local metadata cache to reuse PyArrow
+Dataset and footer-derived metadata such as schema and row-group statistics:
+
+```python
+table = table.copy({
+    "parquet.metadata-cache-enabled": "true",
+    "parquet.metadata-cache-max-entries": "256",
+})
+read_builder = table.new_read_builder()
+```
+
+The cache is disabled by default and is isolated by process and `FileIO`.
+Workers benefit when they remain alive and reuse the same table, for example
+with `DataLoader(..., persistent_workers=True)`. Separately deserialized
+`FileIO` instances do not share entries. The limit counts cached entries, not
+retained bytes.
+
+Paimon-managed data files are immutable, so caching by file path is safe. Do
+not enable this cache for paths that may be overwritten in place.
+
 ## Shuffle
 
 PyPaimon supports streaming shuffle for PyTorch `IterableDataset`. The shuffle
diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py
index c548f99dd05b..d851f0467d94 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -269,17 +269,20 @@ class CoreOptions:
         .boolean_type()
         .default_value(False)
         .with_description(
-            "Cache immutable Parquet file metadata across reads in the current process."
+            "Cache PyArrow Dataset and footer-derived metadata for immutable Paimon "
+            "Parquet data files across reads in the current process. Do not enable this "
+            "for paths that may be overwritten in place."
         )
     )
 
-    PARQUET_METADATA_CACHE_SIZE: ConfigOption[int] = (
-        ConfigOptions.key("parquet.metadata-cache-size")
+    PARQUET_METADATA_CACHE_MAX_ENTRIES: ConfigOption[int] = (
+        ConfigOptions.key("parquet.metadata-cache-max-entries")
         .int_type()
         .default_value(256)
         .with_description(
             "Maximum number of Parquet metadata entries cached per FileIO in the current "
-            "process. When readers share a FileIO, the largest requested value is used."
+            "process. This limit counts entries, not retained bytes. When readers share a "
+            "FileIO, the largest requested value is used."
         )
     )
 
@@ -1045,8 +1048,8 @@ def file_format(self, default=None):
     def parquet_metadata_cache_enabled(self) -> bool:
         return self.options.get(CoreOptions.PARQUET_METADATA_CACHE_ENABLED, False)
 
-    def parquet_metadata_cache_size(self) -> int:
-        return self.options.get(CoreOptions.PARQUET_METADATA_CACHE_SIZE, 256)
+    def parquet_metadata_cache_max_entries(self) -> int:
+        return self.options.get(CoreOptions.PARQUET_METADATA_CACHE_MAX_ENTRIES, 256)
 
     def file_compression(self, default=None):
         return self.options.get(CoreOptions.FILE_COMPRESSION, default)
diff --git a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
index 9ae3df54d687..e57d4f8d67a1 100644
--- a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
@@ -136,7 +136,7 @@ def _reset_parquet_dataset_cache():
 
 
 def _parquet_dataset(file_io: FileIO, file_path: str, cache_enabled: bool,
-                     cache_size: int):
+                     cache_max_entries: int):
     file_path_for_pyarrow = file_io.to_filesystem_path(file_path)
     filesystem = file_io.filesystem
 
@@ -144,11 +144,12 @@ def load():
         return ds.dataset(
             file_path_for_pyarrow, format='parquet', filesystem=filesystem)
 
-    if not cache_enabled or cache_size <= 0:
+    if not cache_enabled or cache_max_entries <= 0:
         return load()
 
     key = (_FilesystemIdentity(filesystem), file_path_for_pyarrow)
-    return _parquet_dataset_cache(file_io, cache_size).get_or_load(key, load)
+    return _parquet_dataset_cache(
+        file_io, cache_max_entries).get_or_load(key, load)
 
 
 class FormatPyArrowReader(RecordBatchReader):
@@ -171,12 +172,12 @@ def __init__(self, file_io: FileIO, file_format: str, file_path: str,
                 options.parquet_metadata_cache_enabled()
                 if options is not None else False
             )
-            cache_size = (
-                options.parquet_metadata_cache_size()
+            cache_max_entries = (
+                options.parquet_metadata_cache_max_entries()
                 if options is not None else 256
             )
             self.dataset = _parquet_dataset(
-                file_io, file_path, cache_enabled, cache_size)
+                file_io, file_path, cache_enabled, cache_max_entries)
         else:
             file_path_for_pyarrow = file_io.to_filesystem_path(file_path)
             self.dataset = ds.dataset(
diff --git a/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py b/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
index 7a770b182e40..44eef290a88d 100644
--- a/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
+++ b/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
@@ -95,10 +95,10 @@ def tearDown(self):
         self.temp_dir.cleanup()
 
     @staticmethod
-    def _options(enabled, size=256):
+    def _options(enabled, max_entries=256):
         return CoreOptions(Options({
             "parquet.metadata-cache-enabled": enabled,
-            "parquet.metadata-cache-size": size,
+            "parquet.metadata-cache-max-entries": max_entries,
         }))
 
     def _read(self, path, options, file_io=None):
@@ -123,7 +123,7 @@ def _read(self, path, options, file_io=None):
     def test_disabled_by_default(self):
         options = CoreOptions(Options({}))
         self.assertFalse(options.parquet_metadata_cache_enabled())
-        self.assertEqual(256, options.parquet_metadata_cache_size())
+        self.assertEqual(256, options.parquet_metadata_cache_max_entries())
 
         original = reader_module.ds.dataset
         with patch.object(reader_module.ds, "dataset", wraps=original) as dataset:
@@ -172,7 +172,7 @@ def test_repeated_scan_skips_footer_io(self):
         self.assertLess(len(counting.reads), uncached_reads)
 
     def test_evicts_least_recently_used_entry(self):
-        options = self._options(True, size=2)
+        options = self._options(True, max_entries=2)
         original = reader_module.ds.dataset
         with patch.object(reader_module.ds, "dataset", wraps=original) as dataset:
             for path in self.paths:
@@ -184,9 +184,9 @@ def test_cache_capacity_does_not_shrink(self):
         original = reader_module.ds.dataset
         with patch.object(reader_module.ds, "dataset", wraps=original) as dataset:
             for path in self.paths:
-                self._read(path, self._options(True, size=3))
-            self._read(self.paths[2], self._options(True, size=1))
-            self._read(self.paths[0], self._options(True, size=1))
+                self._read(path, self._options(True, max_entries=3))
+            self._read(self.paths[2], self._options(True, max_entries=1))
+            self._read(self.paths[0], self._options(True, max_entries=1))
         self.assertEqual(3, dataset.call_count)
         self.assertEqual(
             3, reader_module._parquet_dataset_cache(self.file_io, 1).max_entries)

From 0057495838db2cca08f93bea953bbaa553fbc764 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Mon, 20 Jul 2026 14:30:39 +0800
Subject: [PATCH 5/9] [docs] Simplify Parquet metadata cache guidance

---
 docs/docs/pypaimon/pytorch.md | 16 +++++-----------
 1 file changed, 5 insertions(+), 11 deletions(-)

diff --git a/docs/docs/pypaimon/pytorch.md b/docs/docs/pypaimon/pytorch.md
index 9c87d635773d..d82b00b93f59 100644
--- a/docs/docs/pypaimon/pytorch.md
+++ b/docs/docs/pypaimon/pytorch.md
@@ -61,9 +61,7 @@ when it is false, it will read the full amount of data into memory.
 
 ## Parquet Metadata Cache
 
-Long-lived DataLoader workers may read the same immutable Parquet files across
-multiple epochs. Enable the process-local metadata cache to reuse PyArrow
-Dataset and footer-derived metadata such as schema and row-group statistics:
+For repeated Parquet reads in long-lived workers, enable metadata reuse with:
 
 ```python
 table = table.copy({
@@ -73,14 +71,10 @@ table = table.copy({
 read_builder = table.new_read_builder()
 ```
 
-The cache is disabled by default and is isolated by process and `FileIO`.
-Workers benefit when they remain alive and reuse the same table, for example
-with `DataLoader(..., persistent_workers=True)`. Separately deserialized
-`FileIO` instances do not share entries. The limit counts cached entries, not
-retained bytes.
-
-Paimon-managed data files are immutable, so caching by file path is safe. Do
-not enable this cache for paths that may be overwritten in place.
+The cache is disabled by default and is local to each process and `FileIO`.
+It benefits workers reused with `DataLoader(..., persistent_workers=True)`.
+The limit counts entries, not bytes. Only use it with immutable files, such as
+Paimon-managed data files.
 
 ## Shuffle
 

From 018de22497145e046dfb74a437cfb2ecb6cd5b52 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Wed, 22 Jul 2026 16:41:15 +0800
Subject: [PATCH 6/9] [python] Bound file-format metadata cache by size

---
 docs/docs/pypaimon/pytorch.md                 |  17 +-
 .../pypaimon/common/options/core_options.py   |  36 ++--
 .../read/reader/format_pyarrow_reader.py      | 134 +++++++------
 .../tests/parquet_metadata_cache_test.py      | 180 +++++++++++++-----
 4 files changed, 229 insertions(+), 138 deletions(-)

diff --git a/docs/docs/pypaimon/pytorch.md b/docs/docs/pypaimon/pytorch.md
index d82b00b93f59..ef14a7f86bf4 100644
--- a/docs/docs/pypaimon/pytorch.md
+++ b/docs/docs/pypaimon/pytorch.md
@@ -59,22 +59,23 @@ when it is false, it will read the full amount of data into memory.
 
 **`prefetch_concurrency`** (default: 1): When streaming is true, number of threads used for parallel prefetch within each DataLoader worker. Set to a value greater than 1 to partition splits across threads and increase read throughput. Has no effect when streaming is false.
 
-## Parquet Metadata Cache
+## File Format Metadata Cache
 
-For repeated Parquet reads in long-lived workers, enable metadata reuse with:
+Reusable PyArrow Dataset metadata is cached across reads. Configure its estimated
+size limit with:
 
 ```python
 table = table.copy({
-    "parquet.metadata-cache-enabled": "true",
-    "parquet.metadata-cache-max-entries": "256",
+    "file-format.metadata-cache.max-size": "10 mb",
 })
 read_builder = table.new_read_builder()
 ```
 
-The cache is disabled by default and is local to each process and `FileIO`.
-It benefits workers reused with `DataLoader(..., persistent_workers=True)`.
-The limit counts entries, not bytes. Only use it with immutable files, such as
-Paimon-managed data files.
+The default limit is 10 MB; set it to `0 b` to disable the cache. The cache is
+local to each process and `FileIO`, and benefits workers reused with
+`DataLoader(..., persistent_workers=True)`. The limit estimates retained
+metadata size, so actual memory usage may be higher. The cache assumes immutable
+Paimon data files.
 
 ## Shuffle
 
diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py
index d851f0467d94..67e3ee26fd97 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -264,25 +264,15 @@ class CoreOptions:
         .with_description("Specify the message format of data files.")
     )
 
-    PARQUET_METADATA_CACHE_ENABLED: ConfigOption[bool] = (
-        ConfigOptions.key("parquet.metadata-cache-enabled")
-        .boolean_type()
-        .default_value(False)
-        .with_description(
-            "Cache PyArrow Dataset and footer-derived metadata for immutable Paimon "
-            "Parquet data files across reads in the current process. Do not enable this "
-            "for paths that may be overwritten in place."
-        )
-    )
-
-    PARQUET_METADATA_CACHE_MAX_ENTRIES: ConfigOption[int] = (
-        ConfigOptions.key("parquet.metadata-cache-max-entries")
-        .int_type()
-        .default_value(256)
+    FILE_FORMAT_METADATA_CACHE_MAX_SIZE: ConfigOption[MemorySize] = (
+        ConfigOptions.key("file-format.metadata-cache.max-size")
+        .memory_type()
+        .default_value(MemorySize.of_mebi_bytes(10))
         .with_description(
-            "Maximum number of Parquet metadata entries cached per FileIO in the current "
-            "process. This limit counts entries, not retained bytes. When readers share a "
-            "FileIO, the largest requested value is used."
+            "Maximum estimated size of reusable PyArrow Dataset metadata cached per "
+            "FileIO in the current process. Actual retained memory may be higher. "
+            "The cache assumes immutable Paimon data files. Set to 0 to disable it. "
+            "When readers share a FileIO, the largest requested value is used."
         )
     )
 
@@ -1045,11 +1035,11 @@ def manifest_merge_min_count(self, default=None):
     def file_format(self, default=None):
         return self.options.get(CoreOptions.FILE_FORMAT, default)
 
-    def parquet_metadata_cache_enabled(self) -> bool:
-        return self.options.get(CoreOptions.PARQUET_METADATA_CACHE_ENABLED, False)
-
-    def parquet_metadata_cache_max_entries(self) -> int:
-        return self.options.get(CoreOptions.PARQUET_METADATA_CACHE_MAX_ENTRIES, 256)
+    def file_format_metadata_cache_max_size(self) -> MemorySize:
+        return self.options.get(
+            CoreOptions.FILE_FORMAT_METADATA_CACHE_MAX_SIZE,
+            MemorySize.of_mebi_bytes(10),
+        )
 
     def file_compression(self, default=None):
         return self.options.get(CoreOptions.FILE_COMPRESSION, default)
diff --git a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
index e57d4f8d67a1..a57e7f3ba488 100644
--- a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
@@ -53,19 +53,21 @@ def __eq__(self, other):
         )
 
 
-class _ParquetDatasetCache:
-    def __init__(self, max_entries: int):
-        self.max_entries = max_entries
+class _FileFormatDatasetCache:
+    def __init__(self, max_size: int):
+        self.max_size = max_size
+        self.estimated_size = 0
         self._entries = OrderedDict()
         self._loads = {}
         self._lock = threading.Lock()
 
-    def get_or_load(self, key: Tuple[Any, str], loader: Callable[[], Any]):
+    def get_or_load(self, key: Tuple[Any, str, str], loader: Callable[[], Any],
+                    size_estimator: Callable[[Any], Optional[int]]):
         with self._lock:
-            dataset = self._entries.get(key)
-            if dataset is not None:
+            entry = self._entries.get(key)
+            if entry is not None:
                 self._entries.move_to_end(key)
-                return dataset
+                return entry[0]
 
             future = self._loads.get(key)
             if future is None:
@@ -80,6 +82,7 @@ def get_or_load(self, key: Tuple[Any, str], loader: Callable[[], Any]):
 
         try:
             dataset = loader()
+            estimated_size = size_estimator(dataset)
         except BaseException as exception:
             with self._lock:
                 self._loads.pop(key, None)
@@ -87,69 +90,92 @@ def get_or_load(self, key: Tuple[Any, str], loader: Callable[[], Any]):
             raise
 
         with self._lock:
-            self._entries[key] = dataset
-            self._entries.move_to_end(key)
-            while len(self._entries) > self.max_entries:
-                self._entries.popitem(last=False)
+            if estimated_size is not None:
+                estimated_size = max(1, estimated_size)
+                self._entries[key] = (dataset, estimated_size)
+                self.estimated_size += estimated_size
+                self._entries.move_to_end(key)
+                while self.estimated_size > self.max_size:
+                    _, (_, evicted_size) = self._entries.popitem(last=False)
+                    self.estimated_size -= evicted_size
             self._loads.pop(key, None)
         future.set_result(dataset)
         return dataset
 
-    def ensure_capacity(self, max_entries: int):
+    def ensure_capacity(self, max_size: int):
         with self._lock:
-            self.max_entries = max(self.max_entries, max_entries)
+            self.max_size = max(self.max_size, max_size)
 
 
-_PARQUET_DATASET_CACHES = weakref.WeakKeyDictionary()
-_PARQUET_DATASET_CACHE_LOCK = threading.Lock()
-_PARQUET_DATASET_CACHE_PID = os.getpid()
+_FILE_FORMAT_DATASET_CACHES = weakref.WeakKeyDictionary()
+_FILE_FORMAT_DATASET_CACHE_LOCK = threading.Lock()
+_FILE_FORMAT_DATASET_CACHE_PID = os.getpid()
 
 
-def _ensure_parquet_dataset_cache_process():
-    global _PARQUET_DATASET_CACHES
-    global _PARQUET_DATASET_CACHE_LOCK
-    global _PARQUET_DATASET_CACHE_PID
+def _ensure_file_format_dataset_cache_process():
+    global _FILE_FORMAT_DATASET_CACHES
+    global _FILE_FORMAT_DATASET_CACHE_LOCK
+    global _FILE_FORMAT_DATASET_CACHE_PID
     current_pid = os.getpid()
-    if current_pid != _PARQUET_DATASET_CACHE_PID:
-        _PARQUET_DATASET_CACHES = weakref.WeakKeyDictionary()
-        _PARQUET_DATASET_CACHE_LOCK = threading.Lock()
-        _PARQUET_DATASET_CACHE_PID = current_pid
+    if current_pid != _FILE_FORMAT_DATASET_CACHE_PID:
+        _FILE_FORMAT_DATASET_CACHES = weakref.WeakKeyDictionary()
+        _FILE_FORMAT_DATASET_CACHE_LOCK = threading.Lock()
+        _FILE_FORMAT_DATASET_CACHE_PID = current_pid
 
 
-def _parquet_dataset_cache(file_io: FileIO, max_entries: int) -> _ParquetDatasetCache:
-    _ensure_parquet_dataset_cache_process()
-    with _PARQUET_DATASET_CACHE_LOCK:
-        cache = _PARQUET_DATASET_CACHES.get(file_io)
+def _file_format_dataset_cache(file_io: FileIO,
+                               max_size: int) -> _FileFormatDatasetCache:
+    _ensure_file_format_dataset_cache_process()
+    with _FILE_FORMAT_DATASET_CACHE_LOCK:
+        cache = _FILE_FORMAT_DATASET_CACHES.get(file_io)
         if cache is None:
-            cache = _ParquetDatasetCache(max_entries)
-            _PARQUET_DATASET_CACHES[file_io] = cache
+            cache = _FileFormatDatasetCache(max_size)
+            _FILE_FORMAT_DATASET_CACHES[file_io] = cache
         else:
-            cache.ensure_capacity(max_entries)
+            cache.ensure_capacity(max_size)
         return cache
 
 
-def _reset_parquet_dataset_cache():
-    global _PARQUET_DATASET_CACHES
-    _ensure_parquet_dataset_cache_process()
-    with _PARQUET_DATASET_CACHE_LOCK:
-        _PARQUET_DATASET_CACHES = weakref.WeakKeyDictionary()
+def _reset_file_format_dataset_cache():
+    global _FILE_FORMAT_DATASET_CACHES
+    _ensure_file_format_dataset_cache_process()
+    with _FILE_FORMAT_DATASET_CACHE_LOCK:
+        _FILE_FORMAT_DATASET_CACHES = weakref.WeakKeyDictionary()
 
 
-def _parquet_dataset(file_io: FileIO, file_path: str, cache_enabled: bool,
-                     cache_max_entries: int):
+def _estimate_file_format_dataset_size(dataset, file_format: str) -> Optional[int]:
+    try:
+        if file_format == 'parquet':
+            footer_size = 0
+            for fragment in dataset.get_fragments():
+                metadata = fragment.metadata
+                if metadata is not None:
+                    footer_size += int(metadata.serialized_size)
+            if footer_size > 0:
+                return footer_size
+        return int(dataset.schema.serialize().size)
+    except Exception:
+        return None
+
+
+def _file_format_dataset(file_io: FileIO, file_format: str, file_path: str,
+                         cache_max_size: int):
     file_path_for_pyarrow = file_io.to_filesystem_path(file_path)
     filesystem = file_io.filesystem
 
     def load():
         return ds.dataset(
-            file_path_for_pyarrow, format='parquet', filesystem=filesystem)
+            file_path_for_pyarrow, format=file_format, filesystem=filesystem)
 
-    if not cache_enabled or cache_max_entries <= 0:
+    if cache_max_size <= 0:
         return load()
 
-    key = (_FilesystemIdentity(filesystem), file_path_for_pyarrow)
-    return _parquet_dataset_cache(
-        file_io, cache_max_entries).get_or_load(key, load)
+    key = (_FilesystemIdentity(filesystem), file_format, file_path_for_pyarrow)
+    return _file_format_dataset_cache(
+        file_io, cache_max_size).get_or_load(
+            key, load,
+            lambda dataset: _estimate_file_format_dataset_size(
+                dataset, file_format))
 
 
 class FormatPyArrowReader(RecordBatchReader):
@@ -167,22 +193,12 @@ def __init__(self, file_io: FileIO, file_format: str, file_path: str,
                  push_down_predicate: Any, batch_size: int = 1024,
                  options: CoreOptions = None,
                  nested_name_paths: Optional[List[List[str]]] = None):
-        if file_format == 'parquet':
-            cache_enabled = (
-                options.parquet_metadata_cache_enabled()
-                if options is not None else False
-            )
-            cache_max_entries = (
-                options.parquet_metadata_cache_max_entries()
-                if options is not None else 256
-            )
-            self.dataset = _parquet_dataset(
-                file_io, file_path, cache_enabled, cache_max_entries)
-        else:
-            file_path_for_pyarrow = file_io.to_filesystem_path(file_path)
-            self.dataset = ds.dataset(
-                file_path_for_pyarrow, format=file_format,
-                filesystem=file_io.filesystem)
+        cache_max_size = (
+            options.file_format_metadata_cache_max_size().get_bytes()
+            if options is not None else 10 * 1024 * 1024
+        )
+        self.dataset = _file_format_dataset(
+            file_io, file_format, file_path, cache_max_size)
         self._file_format = file_format
         self.read_fields = read_fields
         self._read_field_names = [f.name for f in read_fields]
diff --git a/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py b/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
index 44eef290a88d..1a90d81cf29b 100644
--- a/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
+++ b/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
@@ -36,6 +36,9 @@
 from pypaimon.schema.data_types import AtomicType, DataField
 
 
+DEFAULT_CACHE_SIZE = 10 * 1024 * 1024
+
+
 class _CountingInputFile:
     def __init__(self, wrapped, file_system):
         self._wrapped = wrapped
@@ -75,9 +78,9 @@ def reset_counts(self):
         self.reads = []
 
 
-class ParquetMetadataCacheTest(unittest.TestCase):
+class FileFormatMetadataCacheTest(unittest.TestCase):
     def setUp(self):
-        reader_module._reset_parquet_dataset_cache()
+        reader_module._reset_file_format_dataset_cache()
         self.temp_dir = tempfile.TemporaryDirectory()
         self.file_io = LocalFileIO(self.temp_dir.name, Options({}))
         self.paths = []
@@ -91,14 +94,13 @@ def setUp(self):
             self.paths.append(path)
 
     def tearDown(self):
-        reader_module._reset_parquet_dataset_cache()
+        reader_module._reset_file_format_dataset_cache()
         self.temp_dir.cleanup()
 
     @staticmethod
-    def _options(enabled, max_entries=256):
+    def _options(max_size="10 mb"):
         return CoreOptions(Options({
-            "parquet.metadata-cache-enabled": enabled,
-            "parquet.metadata-cache-max-entries": max_entries,
+            "file-format.metadata-cache.max-size": max_size,
         }))
 
     def _read(self, path, options, file_io=None):
@@ -120,11 +122,20 @@ def _read(self, path, options, file_io=None):
         finally:
             reader.close()
 
-    def test_disabled_by_default(self):
+    def test_enabled_by_default(self):
         options = CoreOptions(Options({}))
-        self.assertFalse(options.parquet_metadata_cache_enabled())
-        self.assertEqual(256, options.parquet_metadata_cache_max_entries())
+        self.assertEqual(
+            DEFAULT_CACHE_SIZE,
+            options.file_format_metadata_cache_max_size().get_bytes())
+
+        original = reader_module.ds.dataset
+        with patch.object(reader_module.ds, "dataset", wraps=original) as dataset:
+            self._read(self.paths[0], options)
+            self._read(self.paths[0], options)
+        self.assertEqual(1, dataset.call_count)
 
+    def test_zero_size_disables_cache(self):
+        options = self._options("0 b")
         original = reader_module.ds.dataset
         with patch.object(reader_module.ds, "dataset", wraps=original) as dataset:
             self._read(self.paths[0], options)
@@ -132,7 +143,7 @@ def test_disabled_by_default(self):
         self.assertEqual(2, dataset.call_count)
 
     def test_reuses_dataset(self):
-        options = self._options(True)
+        options = self._options()
         original = reader_module.ds.dataset
         with patch.object(reader_module.ds, "dataset", wraps=original) as dataset:
             first = self._read(self.paths[0], options)
@@ -157,58 +168,126 @@ def test_repeated_scan_skips_footer_io(self):
         file_io = LocalFileIO(self.temp_dir.name, Options({}))
         file_io.filesystem = pafs.PyFileSystem(pafs.FSSpecHandler(counting))
 
-        uncached = self._read(path, self._options(False), file_io)
+        uncached = self._read(path, self._options("0 b"), file_io)
         uncached_opens = counting.opens
         uncached_reads = len(counting.reads)
 
         counting.reset_counts()
-        reader_module._reset_parquet_dataset_cache()
-        self._read(path, self._options(True), file_io)
+        reader_module._reset_file_format_dataset_cache()
+        self._read(path, self._options(), file_io)
         counting.reset_counts()
-        cached = self._read(path, self._options(True), file_io)
+        cached = self._read(path, self._options(), file_io)
 
         self.assertEqual(uncached, cached)
         self.assertLess(counting.opens, uncached_opens)
         self.assertLess(len(counting.reads), uncached_reads)
 
-    def test_evicts_least_recently_used_entry(self):
-        options = self._options(True, max_entries=2)
-        original = reader_module.ds.dataset
-        with patch.object(reader_module.ds, "dataset", wraps=original) as dataset:
-            for path in self.paths:
-                self._read(path, options)
-            self._read(self.paths[0], options)
-        self.assertEqual(4, dataset.call_count)
+    def test_evicts_least_recently_used_entry_by_estimated_size(self):
+        cache = reader_module._FileFormatDatasetCache(10)
+        first_key = (None, "parquet", "first")
+        second_key = (None, "parquet", "second")
+        third_key = (None, "parquet", "third")
 
-    def test_cache_capacity_does_not_shrink(self):
-        original = reader_module.ds.dataset
-        with patch.object(reader_module.ds, "dataset", wraps=original) as dataset:
-            for path in self.paths:
-                self._read(path, self._options(True, max_entries=3))
-            self._read(self.paths[2], self._options(True, max_entries=1))
-            self._read(self.paths[0], self._options(True, max_entries=1))
-        self.assertEqual(3, dataset.call_count)
+        cache.get_or_load(first_key, lambda: "first", lambda _: 4)
+        cache.get_or_load(second_key, lambda: "second", lambda _: 4)
+        cache.get_or_load(first_key, lambda: "unused", lambda _: 4)
+        cache.get_or_load(third_key, lambda: "third", lambda _: 4)
+
+        self.assertEqual([first_key, third_key], list(cache._entries.keys()))
+        self.assertEqual(8, cache.estimated_size)
+
+    def test_does_not_retain_entry_larger_than_size_limit(self):
+        cache = reader_module._FileFormatDatasetCache(5)
+        loads = []
+        key = (None, "parquet", "large")
+
+        def load():
+            loads.append(True)
+            return "large"
+
+        self.assertEqual(
+            "large", cache.get_or_load(key, load, lambda _: 6))
         self.assertEqual(
-            3, reader_module._parquet_dataset_cache(self.file_io, 1).max_entries)
+            "large", cache.get_or_load(key, load, lambda _: 6))
+        self.assertEqual(2, len(loads))
+        self.assertEqual(0, len(cache._entries))
+        self.assertEqual(0, cache.estimated_size)
+
+    def test_does_not_retain_entry_without_size_estimate(self):
+        cache = reader_module._FileFormatDatasetCache(10)
+        key = (None, "unknown", "data")
+        loads = []
+
+        def load():
+            loads.append(True)
+            return "unknown"
+
+        self.assertEqual(
+            "unknown", cache.get_or_load(key, load, lambda _: None))
+        self.assertEqual(
+            "unknown", cache.get_or_load(key, load, lambda _: None))
+        self.assertEqual(2, len(loads))
+        self.assertEqual(0, len(cache._entries))
+
+    def test_estimates_serialized_parquet_footer_size(self):
+        dataset = reader_module.ds.dataset(self.paths[0], format="parquet")
+        expected = sum(
+            fragment.metadata.serialized_size
+            for fragment in dataset.get_fragments()
+        )
+        self.assertGreater(expected, 0)
+        self.assertEqual(
+            expected,
+            reader_module._estimate_file_format_dataset_size(
+                dataset, "parquet"))
+
+    def test_cache_capacity_does_not_shrink(self):
+        cache = reader_module._file_format_dataset_cache(
+            self.file_io, 3 * 1024 * 1024)
+        same_cache = reader_module._file_format_dataset_cache(
+            self.file_io, 1024 * 1024)
+
+        self.assertIs(cache, same_cache)
+        self.assertEqual(3 * 1024 * 1024, cache.max_size)
 
     def test_cache_capacity_is_isolated_by_file_io(self):
         other_file_io = LocalFileIO(self.temp_dir.name, Options({}))
 
-        first_cache = reader_module._parquet_dataset_cache(self.file_io, 3)
-        second_cache = reader_module._parquet_dataset_cache(other_file_io, 1)
+        first_cache = reader_module._file_format_dataset_cache(
+            self.file_io, 3 * 1024 * 1024)
+        second_cache = reader_module._file_format_dataset_cache(
+            other_file_io, 1024 * 1024)
 
         self.assertIsNot(first_cache, second_cache)
-        self.assertEqual(3, first_cache.max_entries)
-        self.assertEqual(1, second_cache.max_entries)
+        self.assertEqual(3 * 1024 * 1024, first_cache.max_size)
+        self.assertEqual(1024 * 1024, second_cache.max_size)
 
     def test_does_not_share_across_filesystems(self):
         other_file_io = LocalFileIO(self.temp_dir.name, Options({}))
         original = reader_module.ds.dataset
         with patch.object(reader_module.ds, "dataset", wraps=original) as dataset:
-            reader_module._parquet_dataset(
-                self.file_io, self.paths[0], True, 256)
-            reader_module._parquet_dataset(
-                other_file_io, self.paths[0], True, 256)
+            reader_module._file_format_dataset(
+                self.file_io, "parquet", self.paths[0], DEFAULT_CACHE_SIZE)
+            reader_module._file_format_dataset(
+                other_file_io, "parquet", self.paths[0], DEFAULT_CACHE_SIZE)
+        self.assertEqual(2, dataset.call_count)
+
+    def test_does_not_share_across_file_formats(self):
+        parquet_dataset = object()
+        orc_dataset = object()
+        with patch.object(
+                reader_module.ds, "dataset",
+                side_effect=[parquet_dataset, orc_dataset]) as dataset:
+            with patch.object(
+                    reader_module, "_estimate_file_format_dataset_size",
+                    return_value=1):
+                first = reader_module._file_format_dataset(
+                    self.file_io, "parquet", self.paths[0], DEFAULT_CACHE_SIZE)
+                second = reader_module._file_format_dataset(
+                    self.file_io, "orc", self.paths[0], DEFAULT_CACHE_SIZE)
+
+        self.assertIs(parquet_dataset, first)
+        self.assertIs(orc_dataset, second)
         self.assertEqual(2, dataset.call_count)
 
     def test_cache_key_retains_filesystem_wrapper(self):
@@ -218,8 +297,9 @@ def test_cache_key_retains_filesystem_wrapper(self):
         file_io = LocalFileIO(self.temp_dir.name, Options({}))
         file_io.filesystem = filesystem
 
-        reader_module._parquet_dataset(
-            file_io, os.path.basename(self.paths[0]), True, 256)
+        reader_module._file_format_dataset(
+            file_io, "parquet", os.path.basename(self.paths[0]),
+            DEFAULT_CACHE_SIZE)
         file_io.filesystem = root
         del filesystem
         gc.collect()
@@ -244,11 +324,13 @@ def test_filesystem_hash_collision_does_not_share_dataset(self):
             with patch.object(
                     reader_module._FilesystemIdentity, "__hash__", return_value=1):
                 file_io.filesystem = first_filesystem
-                first = reader_module._parquet_dataset(
-                    file_io, file_name, True, 256).to_table()
+                first = reader_module._file_format_dataset(
+                    file_io, "parquet", file_name,
+                    DEFAULT_CACHE_SIZE).to_table()
                 file_io.filesystem = second_filesystem
-                second = reader_module._parquet_dataset(
-                    file_io, file_name, True, 256).to_table()
+                second = reader_module._file_format_dataset(
+                    file_io, "parquet", file_name,
+                    DEFAULT_CACHE_SIZE).to_table()
 
             self.assertEqual([1], first.column("value").to_pylist())
             self.assertEqual([2], second.column("value").to_pylist())
@@ -257,9 +339,11 @@ def test_filesystem_hash_collision_does_not_share_dataset(self):
             second_dir.cleanup()
 
     def test_resets_after_process_change(self):
-        parent_cache = reader_module._parquet_dataset_cache(self.file_io, 256)
+        parent_cache = reader_module._file_format_dataset_cache(
+            self.file_io, DEFAULT_CACHE_SIZE)
         with patch.object(reader_module.os, "getpid", return_value=os.getpid() + 1):
-            child_cache = reader_module._parquet_dataset_cache(self.file_io, 256)
+            child_cache = reader_module._file_format_dataset_cache(
+                self.file_io, DEFAULT_CACHE_SIZE)
         self.assertIsNot(parent_cache, child_cache)
 
     def test_coalesces_concurrent_loads(self):
@@ -273,7 +357,7 @@ def delayed_dataset(*args, **kwargs):
                 reader_module.ds, "dataset", side_effect=delayed_dataset) as dataset:
             with ThreadPoolExecutor(max_workers=8) as executor:
                 results = list(executor.map(
-                    lambda _: self._read(self.paths[0], self._options(True)),
+                    lambda _: self._read(self.paths[0], self._options()),
                     range(8),
                 ))
 

From 1a8cad72ca3aa64e84ef6500e539bffe07fdbe92 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Wed, 22 Jul 2026 16:52:31 +0800
Subject: [PATCH 7/9] [python] Clear metadata cache when disabled

---
 .../pypaimon/read/reader/format_pyarrow_reader.py    |  7 +++++++
 .../pypaimon/tests/parquet_metadata_cache_test.py    | 12 +++++++-----
 2 files changed, 14 insertions(+), 5 deletions(-)

diff --git a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
index a57e7f3ba488..7857d7a2fb73 100644
--- a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
@@ -143,6 +143,12 @@ def _reset_file_format_dataset_cache():
         _FILE_FORMAT_DATASET_CACHES = weakref.WeakKeyDictionary()
 
 
+def _remove_file_format_dataset_cache(file_io: FileIO):
+    _ensure_file_format_dataset_cache_process()
+    with _FILE_FORMAT_DATASET_CACHE_LOCK:
+        _FILE_FORMAT_DATASET_CACHES.pop(file_io, None)
+
+
 def _estimate_file_format_dataset_size(dataset, file_format: str) -> Optional[int]:
     try:
         if file_format == 'parquet':
@@ -168,6 +174,7 @@ def load():
             file_path_for_pyarrow, format=file_format, filesystem=filesystem)
 
     if cache_max_size <= 0:
+        _remove_file_format_dataset_cache(file_io)
         return load()
 
     key = (_FilesystemIdentity(filesystem), file_format, file_path_for_pyarrow)
diff --git a/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py b/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
index 1a90d81cf29b..3bc879a9f07e 100644
--- a/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
+++ b/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
@@ -134,13 +134,15 @@ def test_enabled_by_default(self):
             self._read(self.paths[0], options)
         self.assertEqual(1, dataset.call_count)
 
-    def test_zero_size_disables_cache(self):
-        options = self._options("0 b")
+    def test_zero_size_disables_and_clears_cache(self):
+        enabled = self._options()
+        disabled = self._options("0 b")
         original = reader_module.ds.dataset
         with patch.object(reader_module.ds, "dataset", wraps=original) as dataset:
-            self._read(self.paths[0], options)
-            self._read(self.paths[0], options)
-        self.assertEqual(2, dataset.call_count)
+            self._read(self.paths[0], enabled)
+            self._read(self.paths[0], disabled)
+            self._read(self.paths[0], enabled)
+        self.assertEqual(3, dataset.call_count)
 
     def test_reuses_dataset(self):
         options = self._options()

From 67a2100ad7cd00d94e3c0f56fdeb11f241c2c108 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Wed, 22 Jul 2026 20:08:53 +0800
Subject: [PATCH 8/9] [python] Use process-wide metadata cache

---
 docs/docs/pypaimon/pytorch.md                 | 12 ++---
 .../pypaimon/common/options/core_options.py   | 12 ++---
 .../read/reader/format_pyarrow_reader.py      | 53 ++++++++++---------
 .../tests/parquet_metadata_cache_test.py      | 48 ++++++++++-------
 4 files changed, 69 insertions(+), 56 deletions(-)

diff --git a/docs/docs/pypaimon/pytorch.md b/docs/docs/pypaimon/pytorch.md
index ef14a7f86bf4..008fced72fb0 100644
--- a/docs/docs/pypaimon/pytorch.md
+++ b/docs/docs/pypaimon/pytorch.md
@@ -66,16 +66,16 @@ size limit with:
 
 ```python
 table = table.copy({
-    "file-format.metadata-cache.max-size": "10 mb",
+    "file-format.metadata-cache.max-size": "50 mb",
 })
 read_builder = table.new_read_builder()
 ```
 
-The default limit is 10 MB; set it to `0 b` to disable the cache. The cache is
-local to each process and `FileIO`, and benefits workers reused with
-`DataLoader(..., persistent_workers=True)`. The limit estimates retained
-metadata size, so actual memory usage may be higher. The cache assumes immutable
-Paimon data files.
+The default limit is 50 MB; set it to `0 b` to bypass the cache. The cache is
+local to each process and benefits workers reused with
+`DataLoader(..., persistent_workers=True)`. The limit estimates retained metadata
+size, so actual memory usage may be higher. The cache assumes immutable Paimon
+data files.
 
 ## Shuffle
 
diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py
index 67e3ee26fd97..7bfe2342312b 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -267,12 +267,12 @@ class CoreOptions:
     FILE_FORMAT_METADATA_CACHE_MAX_SIZE: ConfigOption[MemorySize] = (
         ConfigOptions.key("file-format.metadata-cache.max-size")
         .memory_type()
-        .default_value(MemorySize.of_mebi_bytes(10))
+        .default_value(MemorySize.of_mebi_bytes(50))
         .with_description(
-            "Maximum estimated size of reusable PyArrow Dataset metadata cached per "
-            "FileIO in the current process. Actual retained memory may be higher. "
-            "The cache assumes immutable Paimon data files. Set to 0 to disable it. "
-            "When readers share a FileIO, the largest requested value is used."
+            "Maximum estimated size of reusable PyArrow Dataset metadata cached in the "
+            "current process. Actual retained memory may be higher. The cache assumes "
+            "immutable Paimon data files. Set to 0 to bypass it. When readers request "
+            "different positive values, the largest value is used."
         )
     )
 
@@ -1038,7 +1038,7 @@ def file_format(self, default=None):
     def file_format_metadata_cache_max_size(self) -> MemorySize:
         return self.options.get(
             CoreOptions.FILE_FORMAT_METADATA_CACHE_MAX_SIZE,
-            MemorySize.of_mebi_bytes(10),
+            MemorySize.of_mebi_bytes(50),
         )
 
     def file_compression(self, default=None):
diff --git a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
index 7857d7a2fb73..388d194c17de 100644
--- a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
@@ -17,7 +17,6 @@
 
 import os
 import threading
-import weakref
 from collections import OrderedDict
 from concurrent.futures import Future
 from typing import Any, Callable, Dict, List, Optional, Tuple
@@ -106,47 +105,53 @@ def ensure_capacity(self, max_size: int):
         with self._lock:
             self.max_size = max(self.max_size, max_size)
 
+    def remove(self, key: Tuple[Any, str, str]):
+        with self._lock:
+            entry = self._entries.pop(key, None)
+            if entry is not None:
+                self.estimated_size -= entry[1]
+
 
-_FILE_FORMAT_DATASET_CACHES = weakref.WeakKeyDictionary()
+_FILE_FORMAT_DATASET_CACHE = None
 _FILE_FORMAT_DATASET_CACHE_LOCK = threading.Lock()
 _FILE_FORMAT_DATASET_CACHE_PID = os.getpid()
 
 
 def _ensure_file_format_dataset_cache_process():
-    global _FILE_FORMAT_DATASET_CACHES
+    global _FILE_FORMAT_DATASET_CACHE
     global _FILE_FORMAT_DATASET_CACHE_LOCK
     global _FILE_FORMAT_DATASET_CACHE_PID
     current_pid = os.getpid()
     if current_pid != _FILE_FORMAT_DATASET_CACHE_PID:
-        _FILE_FORMAT_DATASET_CACHES = weakref.WeakKeyDictionary()
+        _FILE_FORMAT_DATASET_CACHE = None
         _FILE_FORMAT_DATASET_CACHE_LOCK = threading.Lock()
         _FILE_FORMAT_DATASET_CACHE_PID = current_pid
 
 
-def _file_format_dataset_cache(file_io: FileIO,
-                               max_size: int) -> _FileFormatDatasetCache:
+def _file_format_dataset_cache(max_size: int) -> _FileFormatDatasetCache:
+    global _FILE_FORMAT_DATASET_CACHE
     _ensure_file_format_dataset_cache_process()
     with _FILE_FORMAT_DATASET_CACHE_LOCK:
-        cache = _FILE_FORMAT_DATASET_CACHES.get(file_io)
-        if cache is None:
-            cache = _FileFormatDatasetCache(max_size)
-            _FILE_FORMAT_DATASET_CACHES[file_io] = cache
+        if _FILE_FORMAT_DATASET_CACHE is None:
+            _FILE_FORMAT_DATASET_CACHE = _FileFormatDatasetCache(max_size)
         else:
-            cache.ensure_capacity(max_size)
-        return cache
+            _FILE_FORMAT_DATASET_CACHE.ensure_capacity(max_size)
+        return _FILE_FORMAT_DATASET_CACHE
 
 
 def _reset_file_format_dataset_cache():
-    global _FILE_FORMAT_DATASET_CACHES
+    global _FILE_FORMAT_DATASET_CACHE
     _ensure_file_format_dataset_cache_process()
     with _FILE_FORMAT_DATASET_CACHE_LOCK:
-        _FILE_FORMAT_DATASET_CACHES = weakref.WeakKeyDictionary()
+        _FILE_FORMAT_DATASET_CACHE = None
 
 
-def _remove_file_format_dataset_cache(file_io: FileIO):
+def _remove_file_format_dataset_cache_entry(key: Tuple[Any, str, str]):
     _ensure_file_format_dataset_cache_process()
     with _FILE_FORMAT_DATASET_CACHE_LOCK:
-        _FILE_FORMAT_DATASET_CACHES.pop(file_io, None)
+        cache = _FILE_FORMAT_DATASET_CACHE
+    if cache is not None:
+        cache.remove(key)
 
 
 def _estimate_file_format_dataset_size(dataset, file_format: str) -> Optional[int]:
@@ -173,16 +178,16 @@ def load():
         return ds.dataset(
             file_path_for_pyarrow, format=file_format, filesystem=filesystem)
 
+    key = (_FilesystemIdentity(filesystem), file_format, file_path_for_pyarrow)
     if cache_max_size <= 0:
-        _remove_file_format_dataset_cache(file_io)
+        _remove_file_format_dataset_cache_entry(key)
         return load()
 
-    key = (_FilesystemIdentity(filesystem), file_format, file_path_for_pyarrow)
-    return _file_format_dataset_cache(
-        file_io, cache_max_size).get_or_load(
-            key, load,
-            lambda dataset: _estimate_file_format_dataset_size(
-                dataset, file_format))
+    return _file_format_dataset_cache(cache_max_size).get_or_load(
+        key,
+        load,
+        lambda dataset: _estimate_file_format_dataset_size(
+            dataset, file_format))
 
 
 class FormatPyArrowReader(RecordBatchReader):
@@ -202,7 +207,7 @@ def __init__(self, file_io: FileIO, file_format: str, file_path: str,
                  nested_name_paths: Optional[List[List[str]]] = None):
         cache_max_size = (
             options.file_format_metadata_cache_max_size().get_bytes()
-            if options is not None else 10 * 1024 * 1024
+            if options is not None else 50 * 1024 * 1024
         )
         self.dataset = _file_format_dataset(
             file_io, file_format, file_path, cache_max_size)
diff --git a/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py b/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
index 3bc879a9f07e..29a36343264d 100644
--- a/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
+++ b/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
@@ -36,7 +36,7 @@
 from pypaimon.schema.data_types import AtomicType, DataField
 
 
-DEFAULT_CACHE_SIZE = 10 * 1024 * 1024
+DEFAULT_CACHE_SIZE = 50 * 1024 * 1024
 
 
 class _CountingInputFile:
@@ -98,7 +98,7 @@ def tearDown(self):
         self.temp_dir.cleanup()
 
     @staticmethod
-    def _options(max_size="10 mb"):
+    def _options(max_size="50 mb"):
         return CoreOptions(Options({
             "file-format.metadata-cache.max-size": max_size,
         }))
@@ -134,7 +134,7 @@ def test_enabled_by_default(self):
             self._read(self.paths[0], options)
         self.assertEqual(1, dataset.call_count)
 
-    def test_zero_size_disables_and_clears_cache(self):
+    def test_zero_size_bypasses_and_removes_entry(self):
         enabled = self._options()
         disabled = self._options("0 b")
         original = reader_module.ds.dataset
@@ -144,6 +144,17 @@ def test_zero_size_disables_and_clears_cache(self):
             self._read(self.paths[0], enabled)
         self.assertEqual(3, dataset.call_count)
 
+    def test_zero_size_keeps_other_entries(self):
+        enabled = self._options()
+        disabled = self._options("0 b")
+        original = reader_module.ds.dataset
+        with patch.object(reader_module.ds, "dataset", wraps=original) as dataset:
+            self._read(self.paths[0], enabled)
+            self._read(self.paths[1], enabled)
+            self._read(self.paths[0], disabled)
+            self._read(self.paths[1], enabled)
+        self.assertEqual(3, dataset.call_count)
+
     def test_reuses_dataset(self):
         options = self._options()
         original = reader_module.ds.dataset
@@ -243,26 +254,25 @@ def test_estimates_serialized_parquet_footer_size(self):
             reader_module._estimate_file_format_dataset_size(
                 dataset, "parquet"))
 
-    def test_cache_capacity_does_not_shrink(self):
-        cache = reader_module._file_format_dataset_cache(
-            self.file_io, 3 * 1024 * 1024)
-        same_cache = reader_module._file_format_dataset_cache(
-            self.file_io, 1024 * 1024)
+    def test_process_cache_uses_largest_requested_capacity(self):
+        cache = reader_module._file_format_dataset_cache(3 * 1024 * 1024)
+        same_cache = reader_module._file_format_dataset_cache(1024 * 1024)
 
         self.assertIs(cache, same_cache)
         self.assertEqual(3 * 1024 * 1024, cache.max_size)
 
-    def test_cache_capacity_is_isolated_by_file_io(self):
+    def test_shares_cache_across_file_io_with_same_filesystem(self):
         other_file_io = LocalFileIO(self.temp_dir.name, Options({}))
+        other_file_io.filesystem = self.file_io.filesystem
 
-        first_cache = reader_module._file_format_dataset_cache(
-            self.file_io, 3 * 1024 * 1024)
-        second_cache = reader_module._file_format_dataset_cache(
-            other_file_io, 1024 * 1024)
+        original = reader_module.ds.dataset
+        with patch.object(reader_module.ds, "dataset", wraps=original) as dataset:
+            reader_module._file_format_dataset(
+                self.file_io, "parquet", self.paths[0], DEFAULT_CACHE_SIZE)
+            reader_module._file_format_dataset(
+                other_file_io, "parquet", self.paths[0], DEFAULT_CACHE_SIZE)
 
-        self.assertIsNot(first_cache, second_cache)
-        self.assertEqual(3 * 1024 * 1024, first_cache.max_size)
-        self.assertEqual(1024 * 1024, second_cache.max_size)
+        self.assertEqual(1, dataset.call_count)
 
     def test_does_not_share_across_filesystems(self):
         other_file_io = LocalFileIO(self.temp_dir.name, Options({}))
@@ -341,11 +351,9 @@ def test_filesystem_hash_collision_does_not_share_dataset(self):
             second_dir.cleanup()
 
     def test_resets_after_process_change(self):
-        parent_cache = reader_module._file_format_dataset_cache(
-            self.file_io, DEFAULT_CACHE_SIZE)
+        parent_cache = reader_module._file_format_dataset_cache(DEFAULT_CACHE_SIZE)
         with patch.object(reader_module.os, "getpid", return_value=os.getpid() + 1):
-            child_cache = reader_module._file_format_dataset_cache(
-                self.file_io, DEFAULT_CACHE_SIZE)
+            child_cache = reader_module._file_format_dataset_cache(DEFAULT_CACHE_SIZE)
         self.assertIsNot(parent_cache, child_cache)
 
     def test_coalesces_concurrent_loads(self):

From b1beeb9c5a00f6c813ac510334db7060d1bb5242 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Wed, 22 Jul 2026 20:18:08 +0800
Subject: [PATCH 9/9] [python] Keep in-flight loads until completion

---
 .../read/reader/format_pyarrow_reader.py      |  5 +-
 .../tests/parquet_metadata_cache_test.py      | 51 ++++++++++++++++++-
 2 files changed, 53 insertions(+), 3 deletions(-)

diff --git a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
index 388d194c17de..ae6e202abbd5 100644
--- a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
@@ -83,9 +83,9 @@ def get_or_load(self, key: Tuple[Any, str, str], loader: Callable[[], Any],
             dataset = loader()
             estimated_size = size_estimator(dataset)
         except BaseException as exception:
+            future.set_exception(exception)
             with self._lock:
                 self._loads.pop(key, None)
-            future.set_exception(exception)
             raise
 
         with self._lock:
@@ -97,8 +97,9 @@ def get_or_load(self, key: Tuple[Any, str, str], loader: Callable[[], Any],
                 while self.estimated_size > self.max_size:
                     _, (_, evicted_size) = self._entries.popitem(last=False)
                     self.estimated_size -= evicted_size
-            self._loads.pop(key, None)
         future.set_result(dataset)
+        with self._lock:
+            self._loads.pop(key, None)
         return dataset
 
     def ensure_capacity(self, max_size: int):
diff --git a/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py b/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
index 29a36343264d..77749d26a9ff 100644
--- a/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
+++ b/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
@@ -17,10 +17,11 @@
 import gc
 import os
 import tempfile
+import threading
 import time
 import unittest
 import weakref
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import Future, ThreadPoolExecutor
 from unittest.mock import patch
 
 import pyarrow as pa
@@ -242,6 +243,54 @@ def load():
         self.assertEqual(2, len(loads))
         self.assertEqual(0, len(cache._entries))
 
+    def test_coalesces_load_while_uncached_result_completes(self):
+        cache = reader_module._FileFormatDatasetCache(10)
+        key = (None, "unknown", "data")
+        setting_result = threading.Event()
+        waiter_started = threading.Event()
+        release_result = threading.Event()
+        loads = []
+        future_count = []
+
+        def new_future():
+            future = Future()
+            future_count.append(future)
+            if len(future_count) == 1:
+                original_set_result = future.set_result
+                original_result = future.result
+
+                def delayed_set_result(result):
+                    setting_result.set()
+                    release_result.wait()
+                    original_set_result(result)
+
+                def observed_result(*args, **kwargs):
+                    waiter_started.set()
+                    return original_result(*args, **kwargs)
+
+                future.set_result = delayed_set_result
+                future.result = observed_result
+            return future
+
+        def load():
+            loads.append(True)
+            return "unknown"
+
+        with patch.object(reader_module, "Future", side_effect=new_future):
+            with ThreadPoolExecutor(max_workers=2) as executor:
+                first = executor.submit(
+                    cache.get_or_load, key, load, lambda _: None)
+                self.assertTrue(setting_result.wait(1))
+                second = executor.submit(
+                    cache.get_or_load, key, load, lambda _: None)
+                waited = waiter_started.wait(1)
+                release_result.set()
+
+                self.assertTrue(waited)
+                self.assertEqual("unknown", first.result())
+                self.assertEqual("unknown", second.result())
+        self.assertEqual(1, len(loads))
+
     def test_estimates_serialized_parquet_footer_size(self):
         dataset = reader_module.ds.dataset(self.paths[0], format="parquet")
         expected = sum(