[python] Route descriptor-backed BLOB reads through the table FileIO.

paimon-python/pypaimon/catalog/rest/rest_token_file_io.py

@@ -299,4 +299,7 @@ def valid_token(self):
         return self.token
 
     def close(self):
-        pass
+        factory = self._uri_reader_factory_cache
+        self._uri_reader_factory_cache = None
+        if factory is not None:
+            factory.close()

paimon-python/pypaimon/common/file_io.py

@@ -418,26 +418,51 @@ def _run_lane(lane):
     def read_blobs_concurrent(self, blobs, parallelism):
         """Read a list of Blobs concurrently, coalescing same-file ranged reads.
 
-        ``BlobRef`` values expose a file range and are coalesced; in-memory
-        ``BlobData`` values are returned directly.
+        Exact ``BlobRef`` values (not subclasses) with a file-backed UriReader
+        are coalesced through that FileIO so table-scoped credentials are
+        preserved. Subclasses may override ``new_input_stream()`` and must not
+        be bypassed. Other readers (for example HTTP) read through the Blob.
         """
-        from pypaimon.table.row.blob import BlobRef
+        from concurrent.futures import ThreadPoolExecutor
+
+        from pypaimon.common.uri_reader import FileUriReader
+        from pypaimon.table.row.blob import BlobData, BlobRef
+
         results: List[Optional[bytes]] = [None] * len(blobs)
-        ranges: List[Optional[tuple]] = [None] * len(blobs)
-        inmem = []
-        for i, b in enumerate(blobs):
-            if b is None:
+        file_groups = {}
+        other_blobs = []
+        for index, blob in enumerate(blobs):
+            if blob is None:
                 continue
-            if isinstance(b, BlobRef):
-                d = b.to_descriptor()
-                ranges[i] = (d.uri, d.offset, d.length)
+            if isinstance(blob, BlobData):
+                results[index] = blob.to_data()
+            elif type(blob) is BlobRef and isinstance(
+                    blob.uri_reader, FileUriReader):
+                descriptor = blob.to_descriptor()
+                source_file_io = blob.uri_reader.file_io
+                group = file_groups.setdefault(
+                    id(source_file_io), (source_file_io, []))[1]
+                group.append((index, (
+                    descriptor.uri, descriptor.offset, descriptor.length)))
             else:
-                inmem.append((i, b))
-        for i, v in enumerate(self.read_ranges_coalesced(ranges, parallelism)):
-            if v is not None:
-                results[i] = v
-        for idx, b in inmem:
-            results[idx] = b.to_data()
+                other_blobs.append((index, blob))
+
+        for source_file_io, indexed_ranges in file_groups.values():
+            ranges = [value for _, value in indexed_ranges]
+            values = source_file_io.read_ranges_coalesced(ranges, parallelism)
+            for (index, _), value in zip(indexed_ranges, values):
+                results[index] = value
+
+        if other_blobs:
+            workers = max(1, min(parallelism, len(other_blobs)))
+
+            def _read_blob(indexed_blob):
+                return indexed_blob[1].to_data()
+
+            with ThreadPoolExecutor(workers) as pool:
+                values = pool.map(_read_blob, other_blobs)
+                for (index, _), value in zip(other_blobs, values):
+                    results[index] = value
         return results
 
     def read_file_utf8(self, path: str) -> str:

paimon-python/pypaimon/common/uri_reader.py

@@ -16,6 +16,7 @@
 # under the License.
 
 import io
+import weakref
 from abc import ABC, abstractmethod
 from typing import Any, Optional, Union
 from urllib.parse import urlparse, ParseResult
@@ -57,6 +58,10 @@ class FileUriReader(UriReader):
     def __init__(self, file_io: Any):
         self._file_io = file_io
 
+    @property
+    def file_io(self) -> Any:
+        return self._file_io
+
     def new_input_stream(self, uri: str):
         try:
             return self._file_io.new_input_stream(uri)
@@ -109,8 +114,33 @@ class UriReaderFactory:
 
     def __init__(self, catalog_options: Union[Options, dict]) -> None:
         self.catalog_options = catalog_options if isinstance(catalog_options, Options) else Options(catalog_options)
-        self._readers = LRUCache(CatalogOptions.BLOB_FILE_IO_DEFAULT_CACHE_SIZE)
         self._readers_lock = rwlock.RWLockFair()
+        # FileIOs created by this factory. Do not close them on LRU eviction:
+        # live BlobRefs may still hold the corresponding UriReader.
+        self._owned_file_ios = []
+        self._closing = False
+        self._readers = self._new_reader_cache()
+
+    _FROM_FILE_IO_FACTORIES = weakref.WeakKeyDictionary()
+
+    @staticmethod
+    def from_file_io(file_io: Any) -> 'UriReaderFactory':
+        """Reuse a token-aware FileIO for non-HTTP URIs (Java fromFileIO)."""
+        try:
+            cached = UriReaderFactory._FROM_FILE_IO_FACTORIES.get(file_io)
+        except TypeError:
+            return _ProvidedFileIOUriReaderFactory(file_io)
+        if cached is not None:
+            return cached
+        factory = _ProvidedFileIOUriReaderFactory(file_io)
+        try:
+            UriReaderFactory._FROM_FILE_IO_FACTORIES[file_io] = factory
+        except TypeError:
+            pass
+        return factory
+
+    def _new_reader_cache(self) -> LRUCache:
+        return LRUCache(CatalogOptions.BLOB_FILE_IO_DEFAULT_CACHE_SIZE)
 
     def create(self, input_uri: str) -> UriReader:
         try:
@@ -148,21 +178,106 @@ def _new_reader(self, key: UriKey, parsed_uri: ParseResult) -> UriReader:
             from pypaimon.common.file_io import FileIO
             uri_string = parsed_uri.geturl()
             file_io = FileIO.get(uri_string, self.catalog_options)
+            self._owned_file_ios.append(file_io)
             return UriReader.from_file(file_io)
         except Exception as e:
             raise RuntimeError(f"Failed to create reader for URI {parsed_uri.geturl()}") from e
 
     def clear_cache(self) -> None:
-        self._readers.clear()
+        if self._closing:
+            return
+        self._closing = True
+        wlock = self._readers_lock.gen_wlock()
+        wlock.acquire()
+        try:
+            file_ios = list(self._owned_file_ios)
+            self._owned_file_ios = []
+            self._readers = self._new_reader_cache()
+        finally:
+            wlock.release()
+        first_error = None
+        try:
+            for file_io in file_ios:
+                try:
+                    file_io.close()
+                except Exception as error:
+                    if first_error is None:
+                        first_error = error
+        finally:
+            self._closing = False
+        if first_error is not None:
+            raise first_error
+
+    def close(self) -> None:
+        self.clear_cache()
 
     def get_cache_size(self) -> int:
         return len(self._readers)
 
     def __getstate__(self):
         state = self.__dict__.copy()
         del state['_readers_lock']
+        del state['_readers']
+        del state['_owned_file_ios']
         return state
 
     def __setstate__(self, state):
         self.__dict__.update(state)
         self._readers_lock = rwlock.RWLockFair()
+        self._owned_file_ios = []
+        self._closing = False
+        self._readers = self._new_reader_cache()
+
+
+class _ProvidedFileIOUriReaderFactory(UriReaderFactory):
+    """Resolves HTTP(S) via HttpUriReader and every other URI through file_io."""
+
+    def __init__(self, file_io: Any) -> None:
+        super().__init__({})
+        self._bind_provided_file_io(file_io)
+
+    def _bind_provided_file_io(self, file_io: Any) -> None:
+        try:
+            self._provided_file_io = weakref.ref(file_io)
+        except TypeError:
+            # Not weakref-able, and therefore also not a WeakKeyDictionary
+            # key — from_file_io does not cache these objects.
+            self._provided_file_io = lambda: file_io
+
+    def _resolved_file_io(self):
+        file_io = self._provided_file_io()
+        if file_io is None:
+            raise RuntimeError(
+                "FileIO used by UriReaderFactory.from_file_io was garbage collected")
+        return file_io
+
+    def __getstate__(self):
+        state = super().__getstate__()
+        # weakref.ref (and the TypeError fallback lambda) cannot be pickled.
+        # Resolve to a strong FileIO for the wire; __setstate__ re-wraps.
+        state['_provided_file_io'] = self._resolved_file_io()
+        return state
+
+    def __setstate__(self, state):
+        file_io = state.pop('_provided_file_io')
+        super().__setstate__(state)
+        self._bind_provided_file_io(file_io)
+
+    def create(self, input_uri: str) -> UriReader:
+        try:
+            parsed_uri = urlparse(input_uri)
+        except Exception as e:
+            raise ValueError("Invalid URI: %s" % input_uri) from e
+        scheme = (parsed_uri.scheme or '').lower()
+        if scheme in ('http', 'https'):
+            return super().create(input_uri)
+        # Do not LRU-cache FileUriReader: it holds FileIO strongly and would
+        # pin the WeakKeyDictionary key. Every non-HTTP URI already wraps the
+        # same provided FileIO, so the cache buys nothing here.
+        return UriReader.from_file(self._resolved_file_io())
+
+    def _new_reader(self, key: UriKey, parsed_uri: ParseResult) -> UriReader:
+        scheme = (key.scheme or '').lower()
+        if scheme in ('http', 'https'):
+            return UriReader.from_http()
+        return UriReader.from_file(self._resolved_file_io())

paimon-python/pypaimon/filesystem/hdfs_native_file_io.py

@@ -696,4 +696,7 @@ def write_vortex(self, path: str, data: pyarrow.Table, **kwargs):
             raise RuntimeError(f"Failed to write Vortex file {path}: {e}") from e
 
     def close(self):
+        uri_reader_factory = getattr(self, 'uri_reader_factory', None)
+        if uri_reader_factory is not None:
+            uri_reader_factory.close()
         self._client = None

paimon-python/pypaimon/filesystem/local_file_io.py

@@ -490,6 +490,11 @@ def write_blob(self, path: str, data: pyarrow.Table, **kwargs):
             self.delete_quietly(path)
             raise RuntimeError(f"Failed to write blob file {path}: {e}") from e
 
+    def close(self):
+        uri_reader_factory = getattr(self, 'uri_reader_factory', None)
+        if uri_reader_factory is not None:
+            uri_reader_factory.close()
+
 
 class FuseLocalFileIO(LocalFileIO):
     """LocalFileIO that translates remote OSS paths to FUSE-mounted local paths.

paimon-python/pypaimon/filesystem/pyarrow_file_io.py

@@ -104,6 +104,11 @@ def __setstate__(self, state):
         self.__dict__.update(state)
         self._legacy_bucket_lock = threading.Lock()
 
+    def close(self):
+        uri_reader_factory = getattr(self, 'uri_reader_factory', None)
+        if uri_reader_factory is not None:
+            uri_reader_factory.close()
+
     @staticmethod
     def parse_location(location: str):
         uri = urlparse(location)

paimon-python/pypaimon/read/reader/auth_masking_reader.py

@@ -40,7 +40,10 @@ def __init__(self, inner, schema: pa.Schema, chunk_size: int = 65536, include_ro
         self._exhausted = False
         self._pending_iterator = None
         self._include_row_kind = include_row_kind
+        self.file_io = getattr(inner, 'file_io', None)
         self.blob_field_indices = getattr(inner, 'blob_field_indices', None)
+        self.descriptor_field_indices = getattr(inner, 'descriptor_field_indices', None)
+        self.blob_view_lookup = getattr(inner, 'blob_view_lookup', None)
         self.vector_field_indices = getattr(inner, 'vector_field_indices', None)
 
     def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
@@ -66,6 +69,7 @@ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
                 self._exhausted = True
                 break
             self._pending_iterator = row_iterator
+            self._refresh_blob_view_lookup(self._inner)
 
         if not row_tuples:
             return None
@@ -95,20 +99,36 @@ class BatchToRecordReaderAdapter(RecordReader):
 
     def __init__(self, inner: RecordBatchReader):
         self._inner = inner
+        self.file_io = getattr(inner, 'file_io', None)
+        self.blob_field_indices = getattr(inner, 'blob_field_indices', None)
+        self.descriptor_field_indices = getattr(inner, 'descriptor_field_indices', None)
+        self.blob_view_lookup = getattr(inner, 'blob_view_lookup', None)
+        self.vector_field_indices = getattr(inner, 'vector_field_indices', None)
 
     def read_batch(self):
         batch = self._inner.read_arrow_batch()
         if batch is None:
             return None
-        return _ArrowBatchIterator(batch)
+        self._refresh_blob_view_lookup(self._inner)
+        return _ArrowBatchIterator(
+            batch,
+            file_io=self.file_io,
+            blob_field_indices=self.blob_field_indices,
+            descriptor_field_indices=self.descriptor_field_indices,
+            blob_view_lookup=self.blob_view_lookup,
+            vector_field_indices=self.vector_field_indices,
+        )
 
     def close(self):
         self._inner.close()
 
 
 class _ArrowBatchIterator(RecordIterator):
 
-    def __init__(self, batch: pa.RecordBatch):
+    def __init__(self, batch: pa.RecordBatch,
+                 file_io=None, blob_field_indices=None,
+                 descriptor_field_indices=None, blob_view_lookup=None,
+                 vector_field_indices=None):
         self._batch = batch
         self._idx = 0
         self._has_rk = "_row_kind" in batch.schema.names
@@ -118,6 +138,11 @@ def __init__(self, batch: pa.RecordBatch):
         else:
             self._rk_idx = -1
             self._data_cols = list(range(batch.num_columns))
+        self._file_io = file_io
+        self._blob_field_indices = blob_field_indices
+        self._descriptor_field_indices = descriptor_field_indices
+        self._blob_view_lookup = blob_view_lookup
+        self._vector_field_indices = vector_field_indices
 
     def next(self):
         if self._idx >= self._batch.num_rows:
@@ -126,7 +151,13 @@ def next(self):
             self._batch.column(j)[self._idx].as_py()
             for j in self._data_cols
         )
-        row = OffsetRow(row_tuple, 0, len(self._data_cols))
+        row = OffsetRow(
+            row_tuple, 0, len(self._data_cols),
+            file_io=self._file_io,
+            blob_field_indices=self._blob_field_indices,
+            descriptor_field_indices=self._descriptor_field_indices,
+            blob_view_lookup=self._blob_view_lookup,
+            vector_field_indices=self._vector_field_indices)
         if self._has_rk:
             from pypaimon.table.row.row_kind import RowKind
             kind_str = self._batch.column(self._rk_idx)[self._idx].as_py()
@@ -146,6 +177,7 @@ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
         batch = self._inner.read_arrow_batch()
         if batch is None:
             return None
+        self._refresh_blob_view_lookup(self._inner)
         mask = self._filter_fn(batch)
         return batch.filter(mask)
 
@@ -184,6 +216,7 @@ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
         batch = self._inner.read_arrow_batch()
         if batch is None:
             return None
+        self._refresh_blob_view_lookup(self._inner)
         original_batch = batch
         masked_columns = {}
         for col_name, transform in self._parsed_rules.items():
@@ -223,6 +256,7 @@ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
         batch = self._inner.read_arrow_batch()
         if batch is None:
             return None
+        self._refresh_blob_view_lookup(self._inner)
         columns = self._columns
         if "_row_kind" in batch.schema.names and "_row_kind" not in columns:
             columns = ["_row_kind"] + list(columns)