[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/read/reader/auth_masking_reader.py

@@ -41,6 +41,7 @@ def __init__(self, inner, schema: pa.Schema, chunk_size: int = 65536, include_ro
         self._pending_iterator = None
         self._include_row_kind = include_row_kind
         self.blob_field_indices = getattr(inner, 'blob_field_indices', None)
+        self.descriptor_field_indices = getattr(inner, 'descriptor_field_indices', None)
         self.vector_field_indices = getattr(inner, 'vector_field_indices', None)
 
     def read_arrow_batch(self) -> Optional[pa.RecordBatch]:

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

@@ -174,7 +174,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 +203,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 +242,11 @@ 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)
+
     def close(self):
         self._inner.close()

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

@@ -53,11 +53,13 @@ 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):
         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.vector_field_indices = vector_field_indices
 
     def read_arrow_batch(self) -> Optional[RecordBatch]:

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

@@ -29,6 +29,23 @@ def blob_field_indices(fields: List[DataField]) -> Set[int]:
     }
 
 
+def descriptor_field_indices(
+        fields: List[DataField], descriptor_field_names: Iterable[str]) -> Set[int]:
+    names = set(descriptor_field_names)
+    if not names:
+        return set()
+    return {i for i, f in enumerate(fields) if f.name in names}
+
+
+def descriptor_field_indices_for_table(table, fields: List[DataField]) -> Set[int]:
+    from pypaimon.common.options.core_options import CoreOptions
+
+    if not CoreOptions.blob_as_descriptor(table.options):
+        return set()
+    return descriptor_field_indices(
+        fields, CoreOptions.blob_descriptor_fields(table.options))
+
+
 def vector_field_indices(fields: List[DataField]) -> Set[int]:
     return {i for i, f in enumerate(fields) if isinstance(f.type, VectorType)}
 

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

@@ -98,6 +98,7 @@ def _filter_batch_by_row(self, batch: pa.RecordBatch) -> Optional[pa.RecordBatch
             self.file_io,
             self.blob_field_indices,
             self.vector_field_indices,
+            self.descriptor_field_indices,
         )
         selected = []
         pos = 0

paimon-python/pypaimon/read/reader/iface/record_batch_reader.py

@@ -36,11 +36,14 @@ class RecordBatchReader(RecordReader):
 
     file_io = None
     blob_field_indices = None
+    descriptor_field_indices = None
     vector_field_indices = None
 
     def _adopt_metadata(self, reader: "RecordBatchReader") -> None:
         self.file_io = reader.file_io
         self.blob_field_indices = reader.blob_field_indices
+        self.descriptor_field_indices = getattr(
+            reader, 'descriptor_field_indices', None)
         self.vector_field_indices = reader.vector_field_indices
 
     @abstractmethod
@@ -73,7 +76,8 @@ def read_batch(self) -> Optional[RecordIterator[InternalRow]]:
             return None
         return InternalRowWrapperIterator(
             self._iter_df_rows(df), df.width, self.file_io,
-            self.blob_field_indices, self.vector_field_indices)
+            self.blob_field_indices, self.vector_field_indices,
+            self.descriptor_field_indices)
 
     @staticmethod
     def _iter_df_rows(df) -> Iterator[tuple]:
@@ -87,12 +91,14 @@ def _iter_df_rows(df) -> Iterator[tuple]:
 class InternalRowWrapperIterator(RecordIterator[InternalRow]):
     def __init__(self, iterator: Iterator[tuple], width: int,
                  file_io=None, blob_field_indices=None,
-                 vector_field_indices=None):
+                 vector_field_indices=None,
+                 descriptor_field_indices=None):
         self._iterator = iterator
         self._reused_row = OffsetRow(None, 0, width,
                                      file_io=file_io,
                                      blob_field_indices=blob_field_indices,
-                                     vector_field_indices=vector_field_indices)
+                                     vector_field_indices=vector_field_indices,
+                                     descriptor_field_indices=descriptor_field_indices)
 
     def next(self) -> Optional[InternalRow]:
         row_tuple = next(self._iterator, None)

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

@@ -21,7 +21,8 @@
 import pyarrow.compute as pc
 from pyarrow import RecordBatch
 
-from pypaimon.read.reader.field_indices import blob_field_indices, vector_field_indices
+from pypaimon.read.reader.field_indices import (
+    blob_field_indices, descriptor_field_indices, vector_field_indices)
 from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
 from pypaimon.schema.data_types import DataField, PyarrowFieldParser
 
@@ -37,7 +38,8 @@ class NestedLeafBatchReader(RecordBatchReader):
     """
 
     def __init__(self, inner: RecordBatchReader, name_paths: List[List[str]],
-                 output_fields: List[DataField]):
+                 output_fields: List[DataField],
+                 descriptor_field_names=None):
         if len(name_paths) != len(output_fields):
             raise ValueError(
                 "name_paths length {} does not match output_fields length {}".format(
@@ -47,6 +49,8 @@ def __init__(self, inner: RecordBatchReader, name_paths: List[List[str]],
         self._schema = PyarrowFieldParser.from_paimon_schema(output_fields)
         self.file_io = inner.file_io
         self.blob_field_indices = blob_field_indices(output_fields)
+        self.descriptor_field_indices = descriptor_field_indices(
+            output_fields, descriptor_field_names or ())
         self.vector_field_indices = vector_field_indices(output_fields)
 
     def read_arrow_batch(self) -> Optional[RecordBatch]:

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

@@ -45,6 +45,7 @@ def __init__(
         file_io=None,
         blob_field_indices=None,
         vector_field_indices=None,
+        descriptor_field_indices=None,
     ):
         if not name_paths:
             raise ValueError("name_paths must be non-empty")
@@ -65,6 +66,8 @@ def __init__(
         self._file_io = file_io
         self._blob_field_indices = project_top_level_field_indices(
             blob_field_indices, self._specs)
+        self._descriptor_field_indices = project_top_level_field_indices(
+            descriptor_field_indices, self._specs)
         self._vector_field_indices = project_top_level_field_indices(
             vector_field_indices, self._specs)
 
@@ -74,7 +77,8 @@ def read_batch(self) -> Optional[RecordIterator[InternalRow]]:
             return None
         return _OuterProjectionIterator(
             inner_batch, self._specs, self._flat_arity, self._file_io,
-            self._blob_field_indices, self._vector_field_indices)
+            self._blob_field_indices, self._vector_field_indices,
+            self._descriptor_field_indices)
 
     def close(self) -> None:
         self._inner.close()
@@ -91,14 +95,16 @@ def __init__(
         file_io=None,
         blob_field_indices=None,
         vector_field_indices=None,
+        descriptor_field_indices=None,
     ):
         self._inner = inner
         self._specs = specs
         self._flat_arity = flat_arity
         self._reused_row = OffsetRow(None, 0, flat_arity,
                                      file_io=file_io,
                                      blob_field_indices=blob_field_indices,
-                                     vector_field_indices=vector_field_indices)
+                                     vector_field_indices=vector_field_indices,
+                                     descriptor_field_indices=descriptor_field_indices)
 
     def next(self) -> Optional[InternalRow]:
         inner_row = self._inner.next()