This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new acf19b491e [python] Replace blob index entry limit with a memory
budget (#10151)
acf19b491e is described below
commit acf19b491e88bcdbf7a0c6a5cf711f74eaa01920
Author: XiaoHongbo <[email protected]>
AuthorDate: Fri Oct 2 18:27:54 2026 +0800
[python] Replace blob index entry limit with a memory budget (#10151)
---
docs/docs/pypaimon/blob.md | 6 +
paimon-python/pypaimon/catalog/catalog_context.py | 39 ++++++-
.../pypaimon/catalog/catalog_environment.py | 10 +-
paimon-python/pypaimon/catalog/catalog_factory.py | 2 +-
.../pypaimon/catalog/filesystem_catalog.py | 9 +-
.../pypaimon/catalog/filesystem_catalog_loader.py | 2 +-
.../pypaimon/catalog/rest/rest_catalog.py | 3 +-
paimon-python/pypaimon/common/blob_index_cache.py | 93 +++++++++++++++
paimon-python/pypaimon/common/options/config.py | 7 ++
.../pypaimon/read/reader/format_blob_reader.py | 16 ++-
paimon-python/pypaimon/read/split_read.py | 4 +-
paimon-python/pypaimon/tests/blob_table_test.py | 29 +++--
paimon-python/pypaimon/tests/blob_test.py | 129 ++++++++++++++++++++-
.../pypaimon/tests/catalog_environment_test.py | 35 ++++++
14 files changed, 345 insertions(+), 39 deletions(-)
diff --git a/docs/docs/pypaimon/blob.md b/docs/docs/pypaimon/blob.md
index 56126ef5b2..d09b3c252e 100644
--- a/docs/docs/pypaimon/blob.md
+++ b/docs/docs/pypaimon/blob.md
@@ -176,6 +176,12 @@ The factory auto-dispatches based on the bytes content
(`BLOBDESC`,
`VIDEOFRM`, or blob-view magic header). This mirrors Java's
`Blob.fromBytes(...)`.
+## Python BLOB index cache
+
+Set `cache.blob-index.max-size` in Catalog options to limit parsed indexes per
+Catalog context (default: `64 mb`; `0 b` disables). The cache is process-local
+and only used by the Python BLOB reader.
+
## See Also
- [Blob Storage](../multimodal-table/blob) — concept, storage modes,
diff --git a/paimon-python/pypaimon/catalog/catalog_context.py
b/paimon-python/pypaimon/catalog/catalog_context.py
index ea9138a7a2..2facc7aa2f 100644
--- a/paimon-python/pypaimon/catalog/catalog_context.py
+++ b/paimon-python/pypaimon/catalog/catalog_context.py
@@ -15,20 +15,53 @@
# specific language governing permissions and limitations
# under the License.
+from pypaimon.common.blob_index_cache import BlobIndexCache
from pypaimon.common.options import Options
+from pypaimon.common.options.config import CatalogOptions
class CatalogContext:
- def __init__(self, options: Options, hadoop_conf, prefer_loader,
fallback_io_loader):
+ def __init__(self, options: Options, hadoop_conf, prefer_loader,
+ fallback_io_loader, blob_index_cache=None):
+ if isinstance(options, dict):
+ options = Options(options)
self.options = options
+ self.blob_index_cache = blob_index_cache
+ if self.blob_index_cache is None:
+ self.blob_index_cache = BlobIndexCache(
+ options.get(CatalogOptions.CACHE_BLOB_INDEX_MAX_SIZE))
self.hadoop_conf = hadoop_conf
self.prefer_io_loader = prefer_loader
self.fallback_io_loader = fallback_io_loader
@staticmethod
- def create(options: Options, hadoop_conf, prefer_loader,
fallback_io_loader):
- return CatalogContext(options, hadoop_conf, prefer_loader,
fallback_io_loader)
+ def create(options: Options, hadoop_conf, prefer_loader,
fallback_io_loader,
+ blob_index_cache=None):
+ return CatalogContext(
+ options,
+ hadoop_conf,
+ prefer_loader,
+ fallback_io_loader,
+ blob_index_cache,
+ )
@staticmethod
def create_from_options(options: Options):
return CatalogContext(options, None, None, None)
+
+ def with_options(self, options: Options):
+ if isinstance(options, dict):
+ options = Options(options)
+ max_size =
options.get(CatalogOptions.CACHE_BLOB_INDEX_MAX_SIZE).get_bytes()
+ cache = (
+ self.blob_index_cache
+ if self.blob_index_cache.max_size_bytes == max_size
+ else None
+ )
+ return CatalogContext(
+ options,
+ self.hadoop_conf,
+ self.prefer_io_loader,
+ self.fallback_io_loader,
+ cache,
+ )
diff --git a/paimon-python/pypaimon/catalog/catalog_environment.py
b/paimon-python/pypaimon/catalog/catalog_environment.py
index badb147340..0f15594cad 100644
--- a/paimon-python/pypaimon/catalog/catalog_environment.py
+++ b/paimon-python/pypaimon/catalog/catalog_environment.py
@@ -101,6 +101,9 @@ class CatalogEnvironment:
context = getattr(self.catalog_loader, "context", None)
return context() if callable(context) else None
+ def blob_index_cache(self):
+ return getattr(self.catalog_context(), "blob_index_cache", None)
+
def dependency_read_context(self) -> Optional[CatalogContext]:
context = self.catalog_context()
if self.identifier is None or context is None:
@@ -123,12 +126,7 @@ class CatalogEnvironment:
dependency_options.to_map()[self._READ_VIA_OPTION] =
RESTUtil.encode_string(
JSON.to_json(self.identifier, separators=(",", ":"))
)
- return CatalogContext.create(
- dependency_options,
- context.hadoop_conf,
- context.prefer_io_loader,
- context.fallback_io_loader,
- )
+ return context.with_options(dependency_options)
def copy(self, identifier: Identifier) -> 'CatalogEnvironment':
"""
diff --git a/paimon-python/pypaimon/catalog/catalog_factory.py
b/paimon-python/pypaimon/catalog/catalog_factory.py
index b28de94e70..0712e1e032 100644
--- a/paimon-python/pypaimon/catalog/catalog_factory.py
+++ b/paimon-python/pypaimon/catalog/catalog_factory.py
@@ -57,7 +57,7 @@ class CatalogFactory:
"Unknown catalog identifier: {}. Available types: {}".format(
identifier, list(CatalogFactory.CATALOG_REGISTRY.keys())))
if identifier == "filesystem":
- return catalog_class(context.options)
+ return catalog_class(context.options, context)
if identifier == "rest":
return catalog_class(context, config_required=config_required)
return catalog_class(context)
diff --git a/paimon-python/pypaimon/catalog/filesystem_catalog.py
b/paimon-python/pypaimon/catalog/filesystem_catalog.py
index ccd5b83296..52c42f163d 100644
--- a/paimon-python/pypaimon/catalog/filesystem_catalog.py
+++ b/paimon-python/pypaimon/catalog/filesystem_catalog.py
@@ -50,12 +50,17 @@ from pypaimon.table.table import Table
class FileSystemCatalog(Catalog):
- def __init__(self, catalog_options: Options):
+ def __init__(self, catalog_options: Options,
+ catalog_context: Optional[CatalogContext] = None):
if not catalog_options.contains(CatalogOptions.WAREHOUSE):
raise ValueError(f"Paimon '{CatalogOptions.WAREHOUSE.key()}' path
must be set")
self.warehouse = catalog_options.get(CatalogOptions.WAREHOUSE)
self.catalog_options = catalog_options
- self.catalog_context =
CatalogContext.create_from_options(catalog_options)
+ self.catalog_context = (
+ catalog_context
+ if catalog_context is not None
+ else CatalogContext.create_from_options(catalog_options)
+ )
self._cache_manager =
CachingFileIO.create_cache_manager(self.catalog_options)
self.file_io = CachingFileIO.wrap_with_caching_if_needed(
FileIO.get(self.warehouse, self.catalog_options),
self.catalog_options,
diff --git a/paimon-python/pypaimon/catalog/filesystem_catalog_loader.py
b/paimon-python/pypaimon/catalog/filesystem_catalog_loader.py
index f7a29d0970..46b840d16e 100644
--- a/paimon-python/pypaimon/catalog/filesystem_catalog_loader.py
+++ b/paimon-python/pypaimon/catalog/filesystem_catalog_loader.py
@@ -64,4 +64,4 @@ class FileSystemCatalogLoader(CatalogLoader):
Returns:
A new FileSystemCatalog instance
"""
- return FileSystemCatalog(self._context.options)
+ return FileSystemCatalog(self._context.options, self._context)
diff --git a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
index 1b7f58e4f6..6f6b72d043 100644
--- a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
+++ b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
@@ -71,8 +71,7 @@ class RESTCatalog(Catalog):
def __init__(self, context: CatalogContext, config_required:
Optional[bool] = True):
self.warehouse = context.options.get(CatalogOptions.WAREHOUSE)
self.rest_api = RESTApi(context.options, config_required)
- self.context = CatalogContext.create(self.rest_api.options,
context.hadoop_conf,
- context.prefer_io_loader,
context.fallback_io_loader)
+ self.context = context.with_options(self.rest_api.options)
self.data_token_enabled =
self.rest_api.options.get(CatalogOptions.DATA_TOKEN_ENABLED)
self._cache_manager =
CachingFileIO.create_cache_manager(self.context.options)
diff --git a/paimon-python/pypaimon/common/blob_index_cache.py
b/paimon-python/pypaimon/common/blob_index_cache.py
new file mode 100644
index 0000000000..611ed14312
--- /dev/null
+++ b/paimon-python/pypaimon/common/blob_index_cache.py
@@ -0,0 +1,93 @@
+# 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 sys
+from collections import namedtuple
+from threading import Lock
+
+from cachetools import LRUCache
+
+from pypaimon.common.memory_size import MemorySize
+
+
+_CachedBlobIndex = namedtuple(
+ "_CachedBlobIndex", ["blob_lengths", "blob_offsets", "size_bytes"]
+)
+_CACHED_ENTRY_OVERHEAD = sys.getsizeof(_CachedBlobIndex((), (), 0))
+_ESTIMATED_INT_SIZE = sys.getsizeof(1 << 60)
+# Allow for cachetools' two dict slots, OrderedDict node, and allocator slack.
+_LRU_ENTRY_OVERHEAD = 256
+
+
+class BlobIndexCache:
+ """Catalog-owned parsed indexes bounded by estimated retained bytes."""
+
+ def __init__(self, max_size):
+ if not isinstance(max_size, MemorySize):
+ raise ValueError("cache.blob-index.max-size must be a memory size")
+ self.max_size_bytes = max_size.get_bytes()
+ self._cache = LRUCache(
+ maxsize=self.max_size_bytes,
+ getsizeof=lambda entry: entry.size_bytes,
+ )
+ self._lock = Lock()
+
+ def get(self, file_path):
+ if self.max_size_bytes == 0:
+ return None
+ with self._lock:
+ entry = self._cache.get(file_path)
+ if entry is None:
+ return None
+ return entry.blob_lengths, entry.blob_offsets
+
+ def put(self, file_path, blob_lengths, blob_offsets):
+ if self.max_size_bytes == 0:
+ return
+ size_bytes = (
+ sys.getsizeof(file_path)
+ + sys.getsizeof(blob_lengths)
+ + sys.getsizeof(blob_offsets)
+ + _CACHED_ENTRY_OVERHEAD
+ + _LRU_ENTRY_OVERHEAD
+ + (len(blob_lengths) + len(blob_offsets)) * _ESTIMATED_INT_SIZE
+ )
+ if size_bytes > self.max_size_bytes:
+ return
+ entry = _CachedBlobIndex(blob_lengths, blob_offsets, size_bytes)
+ with self._lock:
+ self._cache[file_path] = entry
+
+ def __len__(self):
+ with self._lock:
+ return len(self._cache)
+
+ def __contains__(self, file_path):
+ with self._lock:
+ return file_path in self._cache
+
+ def clear(self):
+ with self._lock:
+ self._cache.clear()
+
+ @property
+ def size_bytes(self):
+ with self._lock:
+ return self._cache.currsize
+
+ def __reduce__(self):
+ return type(self), (MemorySize.of_bytes(self.max_size_bytes),)
diff --git a/paimon-python/pypaimon/common/options/config.py
b/paimon-python/pypaimon/common/options/config.py
index d6d627bd3e..e670309579 100644
--- a/paimon-python/pypaimon/common/options/config.py
+++ b/paimon-python/pypaimon/common/options/config.py
@@ -87,6 +87,13 @@ class PVFSOptions:
class CatalogOptions:
+ CACHE_BLOB_INDEX_MAX_SIZE = (
+ ConfigOptions.key("cache.blob-index.max-size")
+ .memory_type()
+ .default_value(MemorySize.of_mebi_bytes(64))
+ .with_description("Maximum estimated Python BLOB index cache size per
catalog.")
+ )
+
URI =
ConfigOptions.key("uri").string_type().no_default_value().with_description("Catalog
URI")
METASTORE =
ConfigOptions.key("metastore").string_type().default_value("filesystem").with_description(
"Metastore type")
diff --git a/paimon-python/pypaimon/read/reader/format_blob_reader.py
b/paimon-python/pypaimon/read/reader/format_blob_reader.py
index e39e08b7b2..c959228cff 100644
--- a/paimon-python/pypaimon/read/reader/format_blob_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_blob_reader.py
@@ -16,17 +16,17 @@
# under the License.
import struct
-from threading import Lock
from typing import List, Optional, Any, Iterator, BinaryIO
import pyarrow as pa
import pyarrow.dataset as ds
-from cachetools import LRUCache
from pyarrow import RecordBatch
+from pypaimon.common.blob_index_cache import BlobIndexCache
from pypaimon.common.delta_varint_compressor import DeltaVarintCompressor
from pypaimon.common.file_io import FileIO
from pypaimon.common.map_blob_key_serializer import
create_map_blob_key_serializer
+from pypaimon.common.memory_size import MemorySize
from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
from pypaimon.read.reader.video_format_reader import (
VideoFileMeta,
@@ -45,8 +45,7 @@ from pypaimon.table.row.generic_row import GenericRow
from pypaimon.table.row.row_kind import RowKind
-_BLOB_INDEX_CACHE = LRUCache(maxsize=16)
-_BLOB_INDEX_CACHE_LOCK = Lock()
+_BLOB_INDEX_CACHE = BlobIndexCache(MemorySize.of_mebi_bytes(64))
def _decode_blob_index(index_bytes):
@@ -70,7 +69,8 @@ class FormatBlobReader(RecordBatchReader):
def __init__(self, file_io: FileIO, file_path: str, read_fields: List[str],
full_fields: List[DataField], push_down_predicate: Any,
blob_as_descriptor: bool,
batch_size: int = 1024, row_indices: Optional[Any] = None,
- blob_parallelism: int = 1, file_size: Optional[int] = None):
+ blob_parallelism: int = 1, file_size: Optional[int] = None,
index_cache=None):
+ self._index_cache = _BLOB_INDEX_CACHE if index_cache is None else
index_cache
self._file_io = file_io
self._file_path = file_path
self._push_down_predicate = push_down_predicate
@@ -374,8 +374,7 @@ class FormatBlobReader(RecordBatchReader):
)
return
- with _BLOB_INDEX_CACHE_LOCK:
- cached_index = _BLOB_INDEX_CACHE.get(self.file_path)
+ cached_index = self._index_cache.get(self.file_path)
if cached_index is not None:
blob_lengths, blob_offsets = cached_index
self.blob_lengths = list(blob_lengths)
@@ -407,8 +406,7 @@ class FormatBlobReader(RecordBatchReader):
raise IOError("Invalid blob file: cannot read index")
blob_lengths, blob_offsets = _decode_blob_index(index_bytes)
- with _BLOB_INDEX_CACHE_LOCK:
- _BLOB_INDEX_CACHE[self.file_path] = blob_lengths, blob_offsets
+ self._index_cache.put(self.file_path, blob_lengths, blob_offsets)
self.blob_lengths = list(blob_lengths)
self.blob_offsets = list(blob_offsets)
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index 50a2c0ea7a..88883f61a2 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -405,7 +405,8 @@ class SplitRead(ABC):
batch_size=batch_size,
row_indices=row_indices,
blob_parallelism=blob_parallelism,
- file_size=file.file_size)
+ file_size=file.file_size,
+
index_cache=self.table.catalog_environment.blob_index_cache())
elif file_format == CoreOptions.FILE_FORMAT_LANCE:
if has_nested:
raise NotImplementedError(
@@ -1761,6 +1762,7 @@ class DataEvolutionSplitRead(SplitRead):
row_indices=row_indices,
blob_parallelism=blob_parallelism,
file_size=file.file_size,
+ index_cache=self.table.catalog_environment.blob_index_cache(),
)
def _split_field_bunches(self, need_merge_files: List[DataFileMeta]) ->
List[FieldBunch]:
diff --git a/paimon-python/pypaimon/tests/blob_table_test.py
b/paimon-python/pypaimon/tests/blob_table_test.py
index ae4c5aa032..75b6aa405b 100755
--- a/paimon-python/pypaimon/tests/blob_table_test.py
+++ b/paimon-python/pypaimon/tests/blob_table_test.py
@@ -2692,9 +2692,13 @@ class DedicatedFormatWriterTest(unittest.TestCase):
self.assertEqual(result.column('pic1').to_pylist()[0], pic1_data)
self.assertEqual(result.column('pic2').to_pylist()[0], pic2_data)
+ @pytest.mark.python_read
def test_blob_view_fields_resolve_upstream_blob(self):
+ from unittest import mock
+
from pypaimon import Schema
from pypaimon.common.options.core_options import CoreOptions
+ from pypaimon.read.reader import format_blob_reader
from pypaimon.table.row.blob import BlobViewStruct
source_schema = pa.schema([
@@ -2761,15 +2765,24 @@ class DedicatedFormatWriterTest(unittest.TestCase):
"Blob view fields should be stored inline without writing new blob
files",
)
- result = target_table.new_read_builder().new_read().to_arrow(
- target_table.new_read_builder().new_scan().plan().splits()
- ).sort_by('id')
- self.assertEqual(result.column('picture').to_pylist(), payloads)
+ with mock.patch.object(
+ format_blob_reader,
+ '_decode_blob_index',
+ wraps=format_blob_reader._decode_blob_index,
+ ) as decode_index:
+ result = target_table.new_read_builder().new_read().to_arrow(
+ target_table.new_read_builder().new_scan().plan().splits()
+ ).sort_by('id')
+ self.assertEqual(result.column('picture').to_pylist(), payloads)
- descriptor_table =
target_table.copy({CoreOptions.BLOB_AS_DESCRIPTOR.key(): 'true'})
- descriptor_result =
descriptor_table.new_read_builder().new_read().to_arrow(
- descriptor_table.new_read_builder().new_scan().plan().splits()
- ).sort_by('id')
+ descriptor_table = target_table.copy({
+ CoreOptions.BLOB_AS_DESCRIPTOR.key(): 'true'
+ })
+ descriptor_result =
descriptor_table.new_read_builder().new_read().to_arrow(
+ descriptor_table.new_read_builder().new_scan().plan().splits()
+ ).sort_by('id')
+
+ self.assertEqual(1, decode_index.call_count)
# With blob-as-descriptor=true, view fields return BlobDescriptor bytes
from pypaimon.table.row.blob import BlobDescriptor
for value in descriptor_result.column('picture').to_pylist():
diff --git a/paimon-python/pypaimon/tests/blob_test.py
b/paimon-python/pypaimon/tests/blob_test.py
index 9beb2829cb..f08eab83df 100644
--- a/paimon-python/pypaimon/tests/blob_test.py
+++ b/paimon-python/pypaimon/tests/blob_test.py
@@ -3181,6 +3181,120 @@ class BlobEndToEndTest(unittest.TestCase):
[BlobData(value)], [field], RowKind.INSERT))
writer.close()
+ def test_blob_index_cache_size_controls_repeated_file_reads(self):
+ from pypaimon.catalog.catalog_context import CatalogContext
+
+ field = DataField(0, "blob_field", AtomicType("BLOB"))
+ file_io = LocalFileIO(self.temp_dir, Options({}))
+ paths = [os.path.join(self.temp_dir, "cache-{}.blob".format(i)) for i
in range(17)]
+ for i, path in enumerate(paths):
+ self._write_single_blob(path, field, bytes([i]))
+ open_stream = file_io.new_input_stream
+ for max_size, expected_reads, expected_entries in [
+ ("1 kb", 34, None),
+ ("1 mb", 0, 17),
+ ("0 b", 34, 0),
+ ]:
+ with self.subTest(max_size=max_size):
+ cache = CatalogContext.create_from_options(
+ Options({"cache.blob-index.max-size":
max_size})).blob_index_cache
+ streams = []
+
+ def counted_open(path):
+ stream = MagicMock(wraps=open_stream(path))
+ streams.append(stream)
+ return stream
+
+ with patch.object(file_io, "new_input_stream",
side_effect=counted_open):
+ results = []
+ reads = []
+ for _ in range(2):
+ streams.clear()
+ descriptors = []
+ for path in paths:
+ reader = FormatBlobReader(
+ file_io, path, [field.name], [field], None,
True,
+ index_cache=cache)
+ try:
+
descriptors.extend(reader.read_arrow_batch().column(0).to_pylist())
+ finally:
+ reader.close()
+ results.append(descriptors)
+ reads.append(sum(stream.read.call_count for stream in
streams))
+ self.assertEqual(reads, [34, expected_reads])
+ self.assertEqual(results[0], results[1])
+ self.assertLessEqual(cache.size_bytes,
cache.max_size_bytes)
+ if expected_entries is not None:
+ self.assertEqual(len(cache), expected_entries)
+
+ def test_blob_index_cache_skips_oversized_entry(self):
+ from pypaimon.common.blob_index_cache import BlobIndexCache
+ from pypaimon.common.memory_size import MemorySize
+
+ cache = BlobIndexCache(MemorySize.of_kibi_bytes(1))
+ cache.put("small.blob", (1,), (0,))
+ self.assertEqual(((1,), (0,)), cache.get("small.blob"))
+
+ values = tuple(range(100))
+ cache.put("oversized.blob", values, values)
+ self.assertIsNone(cache.get("oversized.blob"))
+ self.assertEqual(((1,), (0,)), cache.get("small.blob"))
+
+ def test_blob_index_cache_accounts_for_many_small_entries(self):
+ from pypaimon.common.blob_index_cache import BlobIndexCache
+ from pypaimon.common.memory_size import MemorySize
+
+ cache = BlobIndexCache(MemorySize.of_mebi_bytes(1))
+ for i in range(10000):
+ cache.put("small-{:05d}.blob".format(i), (1,), (0,))
+
+ self.assertLess(len(cache), 2048)
+ self.assertLessEqual(cache.size_bytes, cache.max_size_bytes)
+ self.assertNotIn("small-00000.blob", cache)
+ self.assertIn("small-09999.blob", cache)
+
+ def test_blob_index_cache_catalog_scope_and_serialization(self):
+ import pickle
+ from pypaimon import CatalogFactory, Schema
+ from pypaimon.catalog.catalog_context import CatalogContext
+
+ self.assertEqual(
+ 64 * 1024 * 1024,
+ CatalogContext.create_from_options(
+ Options({})).blob_index_cache.max_size_bytes,
+ )
+ for value in ("-1 mb", "invalid"):
+ with self.subTest(value=value), self.assertRaises(ValueError):
+ CatalogContext.create_from_options(
+ Options({"cache.blob-index.max-size": value}))
+
+ options = {"warehouse": self.temp_dir, "cache.blob-index.max-size":
"64 kb"}
+ catalog = CatalogFactory.create(options)
+ catalog.create_database("db", True)
+ for name in ("a", "b"):
+ catalog.create_table("db." + name, Schema.from_pyarrow_schema(
+ pa.schema([("value", pa.binary())])), False)
+ first = catalog.get_table("db.a")
+ second = catalog.get_table("db.b")
+ cache = first.catalog_environment.blob_index_cache()
+ self.assertIs(cache, second.catalog_environment.blob_index_cache())
+ self.assertIs(cache, first.copy({"blob-as-descriptor": "true"})
+ .catalog_environment.blob_index_cache())
+ cache.put("file.blob", (1,), (0,))
+
+ other = CatalogFactory.create(dict(options,
**{"cache.blob-index.max-size": "32 kb"}))
+ other_cache =
other.get_table("db.a").catalog_environment.blob_index_cache()
+ self.assertIsNot(cache, other_cache)
+ self.assertEqual(other_cache.max_size_bytes, 32 * 1024)
+ self.assertEqual(len(other_cache), 0)
+
+ restored = pickle.loads(pickle.dumps(first.catalog_environment))
+ restored_cache = restored.blob_index_cache()
+ self.assertEqual(restored_cache.max_size_bytes, 64 * 1024)
+ self.assertEqual(len(restored_cache), 0)
+ restored_cache.put("worker.blob", (1,), (0,))
+ self.assertIsNone(cache.get("worker.blob"))
+
def test_blob_end_to_end(self):
# Set up file I/O
file_io = LocalFileIO(self.temp_dir, Options({}))
@@ -3292,14 +3406,13 @@ class BlobEndToEndTest(unittest.TestCase):
reader.close()
def test_blob_readers_reuse_index_by_path(self):
- from pypaimon.read.reader.format_blob_reader import _BLOB_INDEX_CACHE,
_BLOB_INDEX_CACHE_LOCK
+ from pypaimon.read.reader.format_blob_reader import _BLOB_INDEX_CACHE
field = DataField(0, "blob_field", AtomicType("BLOB"))
path = os.path.join(self.temp_dir, "cached-index.blob")
file_io = CountingBlobFileIO(self.temp_dir)
self._write_single_blob(path, field, b"cached-value")
- with _BLOB_INDEX_CACHE_LOCK:
- _BLOB_INDEX_CACHE.clear()
+ _BLOB_INDEX_CACHE.clear()
results = []
try:
@@ -3331,8 +3444,7 @@ class BlobEndToEndTest(unittest.TestCase):
self.assertEqual((path, 4, len(b"cached-value")),
(descriptor.uri, descriptor.offset,
descriptor.length))
finally:
- with _BLOB_INDEX_CACHE_LOCK:
- _BLOB_INDEX_CACHE.clear()
+ _BLOB_INDEX_CACHE.clear()
def test_blob_cached_index_keeps_payload_reads(self):
field = DataField(0, "blob_field", AtomicType("BLOB"))
@@ -3556,13 +3668,18 @@ class BlobEndToEndTest(unittest.TestCase):
evolution_read = DataEvolutionSplitRead(
table, None, [field], split, False)
- with patch("pypaimon.read.split_read.FormatBlobReader") as reader_cls:
+ from pypaimon.common.blob_index_cache import BlobIndexCache
+ from pypaimon.common.memory_size import MemorySize
+ cache = BlobIndexCache(MemorySize.of_kibi_bytes(32))
+ with patch.object(table.catalog_environment, "blob_index_cache",
return_value=cache), \
+ patch("pypaimon.read.split_read.FormatBlobReader") as
reader_cls:
def assert_file_size():
args, kwargs = reader_cls.call_args
arguments = inspect.signature(FormatBlobReader).bind_partial(
*args, **kwargs
).arguments
self.assertEqual(123, arguments.get("file_size"))
+ self.assertIs(cache, arguments.get("index_cache"))
raw_read.file_reader_supplier(file, False, [field.name], False)
assert_file_size()
diff --git a/paimon-python/pypaimon/tests/catalog_environment_test.py
b/paimon-python/pypaimon/tests/catalog_environment_test.py
index 3908b6cb13..9ac2e56bf9 100644
--- a/paimon-python/pypaimon/tests/catalog_environment_test.py
+++ b/paimon-python/pypaimon/tests/catalog_environment_test.py
@@ -48,6 +48,10 @@ class CatalogEnvironmentTest(unittest.TestCase):
dependency_context = environment.dependency_read_context()
self.assertIsNot(dependency_context, context)
+ self.assertIs(
+ dependency_context.blob_index_cache,
+ context.blob_index_cache,
+ )
self.assertFalse(context.options.contains_key(self._READ_VIA_OPTION))
self.assertEqual(
dependency_context.options.get(CatalogOptions.METASTORE), "rest")
@@ -98,6 +102,37 @@ class CatalogEnvironmentTest(unittest.TestCase):
self.assertIsNot(environment.dependency_read_context(), context)
+ def test_filesystem_catalog_loader_preserves_context_cache(self):
+ context = CatalogContext.create_from_options(Options({
+ CatalogOptions.WAREHOUSE.key(): "/tmp/warehouse",
+ }))
+
+ catalog = FileSystemCatalogLoader(context).load()
+
+ self.assertIs(catalog.catalog_context, context)
+ self.assertIs(
+ catalog.catalog_context.blob_index_cache,
+ context.blob_index_cache,
+ )
+
+ def test_rest_catalog_preserves_context_cache(self):
+ from pypaimon.catalog.rest.rest_catalog import RESTCatalog
+
+ context = CatalogContext.create_from_options(Options({
+ CatalogOptions.WAREHOUSE.key(): "/tmp/warehouse",
+ CatalogOptions.METASTORE.key(): "rest",
+ }))
+ with mock.patch(
+ "pypaimon.catalog.rest.rest_catalog.RESTApi") as rest_api:
+ rest_api.return_value.options = context.options
+
+ catalog = RESTCatalog(context)
+
+ self.assertIs(
+ catalog.context.blob_index_cache,
+ context.blob_index_cache,
+ )
+
def test_dependency_read_context_preserves_custom_rest_metastore(self):
context = CatalogContext.create_from_options(Options({
CatalogOptions.METASTORE.key(): "custom-rest",
