[python] BLOB descriptor parsing and read-path compatibility.

paimon-python/pypaimon/common/options/core_options.py

@@ -409,6 +409,15 @@ class CoreOptions:
         )
     )
 
+    BLOB_STORED_DESCRIPTOR_FIELDS: ConfigOption[str] = (
+        ConfigOptions.key("blob.stored-descriptor-fields")
+        .string_type()
+        .no_default_value()
+        .with_description(
+            "Legacy Java option name for blob-descriptor-field."
+        )
+    )
+
     BLOB_VIEW_FIELD: ConfigOption[str] = (
         ConfigOptions.key("blob-view-field")
         .string_type()
@@ -1213,7 +1222,14 @@ def variant_shredding_schema(self) -> Optional[str]:
         return val
 
     def blob_descriptor_fields(self, default=None):
-        value = self.options.get(CoreOptions.BLOB_DESCRIPTOR_FIELD, default)
+        value = self.options.get(CoreOptions.BLOB_DESCRIPTOR_FIELD, None)
+        if value is None:
+            legacy = self.options.data.get(
+                CoreOptions.BLOB_STORED_DESCRIPTOR_FIELDS.key())
+            if legacy is not None:
+                value = legacy
+            else:
+                value = default
         return CoreOptions._parse_field_set(value)
 
     def blob_view_fields(self, default=None):

paimon-python/pypaimon/common/uri_reader.py

@@ -109,8 +109,13 @@ 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()
+        self._owned_file_ios = []
+        self._closing = False
+        self._readers = self._new_reader_cache()
+
+    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 +153,52 @@ 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()

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

@@ -473,6 +473,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

@@ -100,6 +100,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/filesystem/resolving_file_io.py

@@ -37,6 +37,7 @@ def __init__(self, catalog_options: Options):
         opts_map.pop(CatalogOptions.RESOLVING_FILE_IO_ENABLED.key(), None)
         self._options = Options(opts_map)
         self._fileio_cache: Dict[tuple, FileIO] = {}
+        self._uri_reader_factory = None
 
     def _cache_key(self, path: str) -> tuple:
         uri = urlparse(path)
@@ -137,7 +138,18 @@ def write_row(self, path: str, data, fields=None, zstd_level: int = 1, **kwargs)
         return self._get_fileio(path).write_row(path, data, fields,
                                                 zstd_level, **kwargs)
 
+    @property
+    def uri_reader_factory(self):
+        if self._uri_reader_factory is None:
+            from pypaimon.common.uri_reader import UriReaderFactory
+            self._uri_reader_factory = UriReaderFactory(self._options)
+        return self._uri_reader_factory
+
     def close(self):
+        factory = self._uri_reader_factory
+        self._uri_reader_factory = None
+        if factory is not None:
+            factory.close()
         for fileio in self._fileio_cache.values():
             fileio.close()
         self._fileio_cache.clear()

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]:
@@ -95,20 +98,35 @@ 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)
+        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 +136,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 +149,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()

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

@@ -67,12 +67,15 @@ def __init__(self, inner: RecordBatchReader, table,
         self._view_fields = CoreOptions.blob_view_fields(table.options) if resolve_enabled else set()
         self._descriptor_fields = CoreOptions.blob_descriptor_fields(table.options)
         self._blob_as_descriptor = CoreOptions.blob_as_descriptor(table.options)
+        if not self._blob_as_descriptor:
+            # Stage 2 materializes descriptor/view fields to payload bytes.
+            # Row-level descriptor routing must not re-parse that content.
+            self.descriptor_field_indices = set()
         self._prescan_done = False
         self._blob_view_lookup = None
 
     def read_arrow_batch(self) -> Optional[RecordBatch]:
-        # Align with Java: only enter blob view resolution when catalog_loader is available
-        # If catalog_loader is None, skip both Stage 1 (view resolution) and Stage 2 (descriptor resolution)
+        # Align with Java: only enter blob view resolution when catalog_loader is available.
         if self._view_fields and not self._prescan_done:
             self._prescan_view_structs()
 
@@ -174,7 +177,10 @@ def _resolve_descriptor_fields(self, batch, view_file_ios=None):
             if field_name not in batch.schema.names:
                 continue
             values = [self._normalize_blob_to_bytes(v) for v in batch.column(field_name).to_pylist()]
-            blobs = [Blob.from_bytes(v, self._table.file_io) for v in values]
+            blobs = [
+                self._descriptor_field_to_blob(value, self._table.file_io)
+                for value in values
+            ]
 
             if self._blob_parallelism > 1:
                 converted_values = self._table.file_io.read_blobs_concurrent(
@@ -200,7 +206,7 @@ def _resolve_descriptor_fields(self, batch, view_file_ios=None):
 
             for idx, value in enumerate(values):
                 file_io = field_file_ios[idx] or self._table.file_io
-                blob = Blob.from_bytes(value, file_io)
+                blob = self._descriptor_field_to_blob(value, file_io)
                 if self._blob_parallelism > 1:
                     converted_values.append(None)
                     if blob is not None:
@@ -239,5 +245,15 @@ def _normalize_blob_to_bytes(value):
             value = bytes(value)
         return value
 
+    @staticmethod
+    def _descriptor_field_to_blob(value, file_io):
+        if value is None:
+            return None
+        return Blob.from_descriptor_bytes(
+            value,
+            file_io=file_io,
+            uri_reader_factory=getattr(file_io, 'uri_reader_factory', None),
+        )
+
     def close(self):
         self._inner.close()

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

@@ -0,0 +1,65 @@
+# 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.
+
+"""Helpers for eager blob-view/descriptor inline conversion on read."""
+
+from typing import List
+
+from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.read.reader.iface.record_reader import RecordReader
+from pypaimon.schema.data_types import DataField, PyarrowFieldParser
+
+
+def needs_blob_inline_convert(table) -> bool:
+    view_fields = CoreOptions.blob_view_fields(table.options)
+    descriptor_fields = CoreOptions.blob_descriptor_fields(table.options)
+    if descriptor_fields:
+        # Materialize when blob-as-descriptor=false; otherwise still wrap so
+        # merge to_iterator()+get_blob() receives descriptor field metadata.
+        return True
+    if not view_fields:
+        return False
+    if CoreOptions.blob_as_descriptor(table.options):
+        return True
+    return CoreOptions.blob_view_resolve_enabled(table.options)
+
+
+def wrap_record_reader_with_blob_inline_convert(
+        reader: RecordReader,
+        split_read,
+        read_fields: List[DataField],
+) -> RecordReader:
+    from pypaimon.read.reader.auth_masking_reader import (
+        BatchToRecordReaderAdapter, RecordReaderToBatchAdapter)
+    from pypaimon.read.reader.blob_descriptor_convert_reader import BlobInlineConvertReader
+    from pypaimon.read.reader.field_indices import (
+        blob_field_indices, descriptor_field_indices_for_table, vector_field_indices)
+
+    schema = PyarrowFieldParser.from_paimon_schema(read_fields)
+    batch_reader = RecordReaderToBatchAdapter(reader, schema)
+    batch_reader.file_io = split_read.table.file_io
+    batch_reader.blob_field_indices = blob_field_indices(read_fields)
+    batch_reader.descriptor_field_indices = descriptor_field_indices_for_table(
+        split_read.table, read_fields)
+    batch_reader.vector_field_indices = vector_field_indices(read_fields)
+    batch_reader = BlobInlineConvertReader(
+        batch_reader,
+        split_read.table,
+        prescan_reader_factory=lambda names: split_read._create_blob_view_prescan_reader(names),
+        blob_parallelism=split_read._blob_parallelism,
+    )
+    return BatchToRecordReaderAdapter(batch_reader)

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

@@ -53,11 +53,15 @@ def __init__(
 class ConcatBatchReader(RecordBatchReader):
 
     def __init__(self, reader_suppliers: List[Callable], file_io=None,
-                 blob_field_indices=None, vector_field_indices=None):
+                 blob_field_indices=None, vector_field_indices=None,
+                 descriptor_field_indices=None,
+                 blob_view_lookup=None):
         self.queue: collections.deque[Callable] = collections.deque(reader_suppliers)
         self.current_reader: Optional[RecordBatchReader] = None
         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 read_arrow_batch(self) -> Optional[RecordBatch]:

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

@@ -52,6 +52,9 @@ def read_arrow_batch(self) -> Optional[RecordBatch]:
             if column_index < 0:
                 continue
             values = batch.column(column_index).to_pylist()
+            # Dedicated .blob files live on the table filesystem. Resolve
+            # through this FileIO so REST tokens and test wrappers apply;
+            # UriReaderFactory would rebuild an unscoped FileIO.
             blobs = [Blob.from_bytes(value, self._file_io) for value in values]
             if self._blob_parallelism > 1:
                 payloads = self._file_io.read_blobs_concurrent(