[python] Support structured-data local cache selection (#10225)

Author: XiaoHongbo-Hope

docs/docs/program-api/file-cache.mdx

@@ -50,6 +50,7 @@ The cache classifies files by path. Set `local-cache.whitelist` to select the fi
 | GLOBAL_INDEX | global-index | BTree and full-text index files | Yes |
 | BUCKET_INDEX | bucket-index | Hash, deletion vector index files | No |
 | DATA | data | Data files (ORC, Parquet, etc.) | No |
+| STRUCTURED_DATA | structured-data | Parquet, ORC, Avro, ROW, Mosaic (PyPaimon/Rust) | No |
 | FILE_INDEX | file-index | Data-file level bloom filter, bitmap | No |
 
 The default whitelist is `meta,global-index`, and `*` selects every file type. Files rewritten in
@@ -128,12 +129,15 @@ The Java snippet is a method-body fragment; declare or handle the catalog except
 | `local-cache.dir` | String | (none) | Directory for storing cached blocks on disk. If not configured, memory cache is used. |
 | `local-cache.max-size` | MemorySize | See below | Maximum cached block bytes per cache manager. Least recently used blocks are evicted when the limit is exceeded. |
 | `local-cache.block-size` | MemorySize | 1 mb | Block size for caching. Files are logically divided into fixed-size blocks and cached independently. |
-| `local-cache.whitelist` | String | meta,global-index | Comma-separated list of file types to cache. Supported values: `meta`, `global-index`, `bucket-index`, `data`, `file-index`, or `*` for all of them. |
+| `local-cache.whitelist` | String | meta,global-index | File types to cache (see [Cached File Types](#cached-file-types)); `*` selects all. |
 
 When `local-cache.max-size` is omitted, Java has no configured size limit. PyPaimon uses a 256 MiB
 memory limit or a 10 GiB disk limit. Set the option explicitly when you want the same limit across
 clients. The limit accounts for cached block bytes, not all reader buffers or process memory.
 
+Use `local-cache.whitelist=meta,global-index,structured-data` to cache structured files without
+BLOB or video bodies. PyPaimon's Vortex and Lance readers bypass this cache.
+
 ## How It Works
 
 1. The reader requests bytes from an eligible file. The cache maps the byte range to fixed-size

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

@@ -1150,8 +1150,8 @@ class CoreOptions:
         .default_value("meta,global-index")
         .with_description(
             "Comma-separated list of file types to cache. "
-            "Supported values: meta, global-index, bucket-index, data, file-index, "
-            "or * for all of them."
+            "Supported values: meta, global-index, bucket-index, data, "
+            "structured-data, file-index, or * for all of them."
         )
     )
 

paimon-python/pypaimon/execution/python/read/reader/format_avro_reader.py

@@ -36,8 +36,12 @@ class FormatAvroReader(RecordBatchReader):
     def __init__(self, file_io: FileIO, file_path: str, read_fields: List[str], full_fields: List[DataField],
                  push_down_predicate: Any, batch_size: int = 1024,
                  nested_name_paths: Optional[List[List[str]]] = None):
-        file_path_for_io = file_io.to_filesystem_path(file_path)
-        self._file = file_io.filesystem.open_input_file(file_path_for_io)
+        from pypaimon.filesystem.caching_file_io import CachingFileIO
+        if isinstance(file_io, CachingFileIO) and file_io._is_cacheable(file_path):
+            self._file = file_io.new_input_stream(file_path)
+        else:
+            file_path_for_io = file_io.to_filesystem_path(file_path)
+            self._file = file_io.filesystem.open_input_file(file_path_for_io)
         self._avro_reader = fastavro.reader(self._file)
         self._batch_size = batch_size
         self._push_down_predicate = push_down_predicate

paimon-python/pypaimon/execution/python/read/reader/format_pyarrow_reader.py

@@ -275,6 +275,18 @@ def _file_format_metadata_cache_max_size(file_io: FileIO) -> int:
         CatalogOptions.FILE_FORMAT_METADATA_CACHE_MAX_SIZE).get_bytes()
 
 
+def _uses_block_cache(file_io: FileIO, file_path: str) -> bool:
+    from pypaimon.filesystem.caching_file_io import CachingFileIO
+    return isinstance(file_io, CachingFileIO) and file_io._is_cacheable(file_path)
+
+
+def _open_pyarrow_input_file(file_io: FileIO, file_path: str, file_size=None):
+    if _uses_block_cache(file_io, file_path):
+        return pa.PythonFile(
+            file_io._new_input_stream_with_known_size(file_path, file_size), mode='r')
+    return file_io.filesystem.open_input_file(file_io.to_filesystem_path(file_path))
+
+
 def _file_format_dataset(file_io: FileIO, file_format: str, file_path: str,
                          cache_max_size: int,
                          file_size: Optional[int] = None):
@@ -291,6 +303,11 @@ def _file_format_dataset(file_io: FileIO, file_format: str, file_path: str,
     if known_size is not None and register_file_size is not None:
         register_file_size(file_path_for_pyarrow, known_size)
 
+    if _uses_block_cache(file_io, file_path):
+        from pypaimon.filesystem.caching_file_system import CachedFileSystemHandler
+        import pyarrow.fs as pafs
+        filesystem = pafs.PyFileSystem(CachedFileSystemHandler(file_io, file_path, known_size))
+
     def load():
         if file_format == 'parquet':
             parquet_format = ds.ParquetFileFormat()
@@ -315,7 +332,8 @@ def load():
             file_path_for_pyarrow, format=file_format, filesystem=filesystem)
 
     key = (
-        _FilesystemIdentity(filesystem), file_format, file_path_for_pyarrow,
+        _FilesystemIdentity(file_io if _uses_block_cache(file_io, file_path) else filesystem),
+        file_format, file_path_for_pyarrow,
         known_size)
     if cache_max_size <= 0:
         _reset_file_format_dataset_cache()
@@ -335,8 +353,7 @@ def _orc_schema_with_field_metadata(file_io: FileIO, file_path: str,
         return fallback
 
     import pyarrow.orc as orc
-    source = file_io.filesystem.open_input_file(
-        file_io.to_filesystem_path(file_path))
+    source = _open_pyarrow_input_file(file_io, file_path)
     try:
         metadata = orc.ORCFile(source).metadata
         arrow_schema = metadata.get(b"ARROW:schema")
@@ -383,7 +400,7 @@ def __init__(self, file_io: FileIO, file_format: str, file_path: str,
         file_path_for_pyarrow = file_io.to_filesystem_path(file_path)
         self._row_group_cache = row_group_cache
         self._row_group_cache_filesystem = _FilesystemIdentity(
-            file_io.filesystem)
+            file_io if _uses_block_cache(file_io, file_path) else file_io.filesystem)
         self._row_group_cache_path = file_path_for_pyarrow
         cache_max_size = _file_format_metadata_cache_max_size(file_io)
         self.dataset = _file_format_dataset(
@@ -561,8 +578,7 @@ def __init__(self, file_io: FileIO, file_format: str, file_path: str,
                     and self._selected_shared_map_paths)):
             import pyarrow.parquet as pq
             # ParquetFile(filesystem=...) is unavailable in PyArrow 6.
-            self._parquet_source = file_io.filesystem.open_input_file(
-                file_path_for_pyarrow)
+            self._parquet_source = _open_pyarrow_input_file(file_io, file_path, file_size)
             try:
                 self._parquet_file = pq.ParquetFile(self._parquet_source)
                 if (self._selected_parquet_row_groups is not None
@@ -581,8 +597,7 @@ def __init__(self, file_io: FileIO, file_format: str, file_path: str,
                 raise
         if file_format == 'orc' and self._selected_shared_map_paths:
             import pyarrow.orc as orc
-            self._orc_source = file_io.filesystem.open_input_file(
-                file_path_for_pyarrow)
+            self._orc_source = _open_pyarrow_input_file(file_io, file_path)
             self._orc_file = orc.ORCFile(self._orc_source)
         if self._exhausted:
             self._raw_batches = iter(())

paimon-python/pypaimon/filesystem/caching_file_io.py

@@ -26,6 +26,7 @@
 """
 
 import hashlib
+import io
 import os
 import threading
 import uuid
@@ -262,7 +263,7 @@ def put_file_size(self, file_path: str, size: int) -> None:
         self._file_size_cache[file_path] = size
 
 
-class CachingInputStream:
+class CachingInputStream(io.RawIOBase):
     """Wraps a remote stream with block-level caching."""
 
     def __init__(self, file_io, file_path: str, cache, file_size=None):
@@ -275,6 +276,18 @@ def __init__(self, file_io, file_path: str, cache, file_size=None):
         self._io_lock = threading.Lock()
         self._remote_supports_pread = None
 
+    def readable(self):
+        return True
+
+    def seekable(self):
+        return True
+
+    def readinto(self, buffer):
+        view = memoryview(buffer).cast('B')
+        data = self.read(len(view))
+        view[:len(data)] = data
+        return len(data)
+
     def _get_file_size(self) -> int:
         if self._file_size == -1:
             cached = self._cache.get_file_size(self._file_path)
@@ -320,14 +333,14 @@ def read_at(self, nbytes: int, offset: int) -> bytes:
         first_block = offset // block_size
         last_block = (end - 1) // block_size
 
-        result = bytearray()
+        parts = []
         for bi, block_data in self._read_blocks(first_block, last_block):
             block_start = bi * block_size
             start_in_block = max(offset - block_start, 0)
             end_in_block = min(end - block_start, len(block_data))
-            result.extend(block_data[start_in_block:end_in_block])
+            parts.append(block_data[start_in_block:end_in_block])
 
-        return bytes(result)
+        return b''.join(parts)
 
     def _read_blocks(self, first: int, last: int):
         block_size = self._cache.block_size
@@ -393,6 +406,7 @@ def close(self):
         if self._stream is not None:
             self._stream.close()
             self._stream = None
+        super().close()
 
     def __enter__(self):
         return self
@@ -471,12 +485,17 @@ def wrap_with_caching_if_needed(file_io, options, cache=None):
     def properties(self):
         return self._delegate.properties
 
+    def _is_cacheable(self, path: str):
+        file_type = FileType.classify(path)
+        return (self._cache is not None
+                and file_type in self._whitelist
+                and not FileType.is_mutable(path))
+
     def new_input_stream(self, path: str):
         return self._new_input_stream_with_known_size(path, None)
 
     def _new_input_stream_with_known_size(self, path: str, file_size):
-        file_type = FileType.classify(path)
-        if self._cache is None or file_type not in self._whitelist or FileType.is_mutable(path):
+        if not self._is_cacheable(path):
             return self._delegate.new_input_stream(path)
         return CachingInputStream(self._delegate, path, self._cache, file_size)
 

paimon-python/pypaimon/filesystem/caching_file_system.py

@@ -0,0 +1,78 @@
+# 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 pyarrow as pa
+import pyarrow.fs as pafs
+
+
+class CachedFileSystemHandler(pafs.FileSystemHandler):
+    """Route one fragment's input through FileIO while preserving filesystem operations."""
+
+    def __init__(self, file_io, path, file_size=None):
+        self._file_io = file_io
+        self._path = path
+        self._delegate = file_io.filesystem
+        self._filesystem_path = self._delegate.normalize_path(file_io.to_filesystem_path(path))
+        self._file_size = file_size
+
+    def get_type_name(self):
+        return 'paimon-cached-file'
+
+    def open_input_file(self, path):
+        if self._delegate.normalize_path(path) != self._filesystem_path:
+            return self._delegate.open_input_file(path)
+        return pa.PythonFile(
+            self._file_io._new_input_stream_with_known_size(self._path, self._file_size), mode='r')
+
+    def open_input_stream(self, path):
+        return self.open_input_file(path)
+
+    def normalize_path(self, path):
+        return self._delegate.normalize_path(path)
+
+    def get_file_info(self, paths):
+        return self._delegate.get_file_info(paths)
+
+    def get_file_info_selector(self, selector):
+        return self._delegate.get_file_info(selector)
+
+    def create_dir(self, path, recursive):
+        return self._delegate.create_dir(path, recursive=recursive)
+
+    def delete_dir(self, path):
+        return self._delegate.delete_dir(path)
+
+    def delete_dir_contents(self, path, missing_dir_ok):
+        return self._delegate.delete_dir_contents(path, missing_dir_ok=missing_dir_ok)
+
+    def delete_root_dir_contents(self):
+        return self._delegate.delete_dir_contents("", accept_root_dir=True)
+
+    def delete_file(self, path):
+        return self._delegate.delete_file(path)
+
+    def move(self, src, dest):
+        return self._delegate.move(src, dest)
+
+    def copy_file(self, src, dest):
+        return self._delegate.copy_file(src, dest)
+
+    def open_output_stream(self, path, metadata):
+        return self._delegate.open_output_stream(path, metadata=metadata)
+
+    def open_append_stream(self, path, metadata):
+        return self._delegate.open_append_stream(path, metadata=metadata)

paimon-python/pypaimon/tests/caching_file_io_test.py

@@ -26,6 +26,7 @@
 import shutil
 import tempfile
 import threading
+import tracemalloc
 import unittest
 from concurrent.futures import ThreadPoolExecutor
 from unittest.mock import MagicMock, patch
@@ -242,6 +243,25 @@ def test_read_spanning_multiple_blocks(self):
         result = stream.read(30)  # 90..120, spans blocks 0 and 1
         self.assertEqual(data[90:120], result)
 
+    def test_large_cached_read_uses_one_result_buffer(self):
+        block_size = 256 * 1024
+        data = b'x' * (32 * block_size)
+        cache = LocalMemoryCacheManager(len(data), block_size=block_size)
+        for i in range(32):
+            cache.put_block('test', i, data[i * block_size:(i + 1) * block_size])
+        for positioned in (False, True):
+            with self.subTest(positioned=positioned):
+                stream = CachingInputStream(self._make_file_io(data), 'test', cache)
+                tracemalloc.start()
+                try:
+                    result = (stream.read_at(len(data), 0) if positioned else stream.read())
+                    _, peak = tracemalloc.get_traced_memory()
+                finally:
+                    tracemalloc.stop()
+                self.assertEqual(data, result)
+                self.assertEqual(0 if positioned else len(data), stream.tell())
+                self.assertLess(peak, len(data) * 3 // 2)
+
     def test_cache_hit_avoids_remote_read(self):
         data = b"0123456789abcdef"
         cache = LocalDiskCacheManager(self.cache_dir, 2 ** 63 - 1, block_size=8)
@@ -472,6 +492,57 @@ def test_custom_whitelist_meta_only(self):
         result = caching_io.new_input_stream("global-index-uuid.index")
         self.assertNotIsInstance(result, CachingInputStream)
 
+    def test_structured_data_cache_preserves_existing_selectors(self):
+        from pypaimon.filesystem.caching_file_io import LocalMemoryCacheManager
+        from pypaimon.utils.file_type import FileType
+
+        for disk in (False, True):
+            for whitelist in ('meta,global-index', 'data', 'structured-data',
+                              'meta,global-index,structured-data', '*'):
+                for name in ('data.parquet', 'data.blob', 'data.orc', 'data.avro',
+                             'data.row', 'data.mosaic', 'data.vortex', 'data.lance',
+                             'data.video', 'data.unknown', 'data.parquet.index',
+                             'snapshot-1', 'snapshot-1.blob'):
+                    with self.subTest(disk=disk, whitelist=whitelist, name=name):
+                        with tempfile.TemporaryDirectory() as directory:
+                            cache = (LocalDiskCacheManager(directory, 1024, block_size=4) if disk
+                                     else LocalMemoryCacheManager(1024, block_size=4))
+                            delegate = self._make_delegate({name: b'abcdefgh'})
+                            caching_io = CachingFileIO(
+                                delegate, cache, FileType.parse_whitelist(whitelist))
+                            if whitelist == '*':
+                                cached = True
+                            elif whitelist == 'data':
+                                cached = name.startswith('data.') and not name.endswith('.index')
+                            elif whitelist == 'structured-data':
+                                cached = name in (
+                                    'data.parquet', 'data.orc', 'data.avro',
+                                    'data.row', 'data.mosaic')
+                            elif whitelist == 'meta,global-index,structured-data':
+                                cached = name in (
+                                    'data.parquet', 'data.orc', 'data.avro', 'data.row',
+                                    'data.mosaic', 'snapshot-1', 'snapshot-1.blob')
+                            else:
+                                cached = name.startswith('snapshot-')
+                            for _ in range(2):
+                                with caching_io.new_input_stream(name) as stream:
+                                    stream.seek(1)
+                                    self.assertEqual(b'bcdefg', stream.read(6))
+                            self.assertEqual(1 if cached else 2,
+                                             delegate.new_input_stream.call_count)
+                            if not cached:
+                                delegate.get_file_size.assert_not_called()
+
+    def test_data_whitelist_still_caches_blob(self):
+        from pypaimon.utils.file_type import FileType
+        delegate = self._make_delegate({"data.blob": b"abcdefgh"})
+        cache = LocalDiskCacheManager(self.cache_dir, 1024, block_size=4)
+        caching_io = CachingFileIO(delegate, cache, {FileType.DATA})
+        for _ in range(2):
+            with caching_io.new_input_stream("data.blob") as stream:
+                self.assertEqual(b"abcdefgh", stream.read())
+        delegate.new_input_stream.assert_called_once()
+
     def test_data_file_not_cached(self):
         data = b"data content"
         delegate = self._make_delegate({"data-abc.orc": data})