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 39a7c4cf0a [python] Deduplicate concurrent BLOB index loads (#10475)
39a7c4cf0a is described below
commit 39a7c4cf0a81cafb9794e59efba1b6f977691d4f
Author: QuakeWang <[email protected]>
AuthorDate: Tue Oct 13 13:31:11 2026 +0800
[python] Deduplicate concurrent BLOB index loads (#10475)
---
docs/docs/pypaimon/blob.md | 3 +
.../pypaimon/benchmark/blob_index_cold_load.py | 273 ++++++++++++++++
paimon-python/pypaimon/common/blob_index_cache.py | 83 ++++-
.../pypaimon/read/reader/format_blob_reader.py | 16 +-
.../pypaimon/tests/blob_index_cache_test.py | 353 +++++++++++++++++++++
5 files changed, 701 insertions(+), 27 deletions(-)
diff --git a/docs/docs/pypaimon/blob.md b/docs/docs/pypaimon/blob.md
index d09b3c252e..085dc2a0dd 100644
--- a/docs/docs/pypaimon/blob.md
+++ b/docs/docs/pypaimon/blob.md
@@ -181,6 +181,9 @@ The factory auto-dispatches based on the bytes content
(`BLOBDESC`,
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.
+With caching enabled, concurrent readers of the same file share one in-flight
+index load. Failed loads can be retried; indexes exceeding the budget are
shared
+only with current waiters and are not retained.
## See Also
diff --git a/paimon-python/pypaimon/benchmark/blob_index_cold_load.py
b/paimon-python/pypaimon/benchmark/blob_index_cold_load.py
new file mode 100644
index 0000000000..be06d78028
--- /dev/null
+++ b/paimon-python/pypaimon/benchmark/blob_index_cold_load.py
@@ -0,0 +1,273 @@
+# 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.
+
+"""Observe BLOB index cold loads through public, disjoint shard reads.
+
+Requires Python 3.7+ for thread CPU timing. Cold means an empty Catalog
+index cache, not an empty OS page cache. No delay or barrier is injected
+inside a reader. Timing-only runs do not patch any production method.
+"""
+
+import argparse
+import json
+import platform
+import random
+import sys
+import tempfile
+import threading
+import time
+from collections import Counter
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import ExitStack
+from pathlib import Path
+from unittest.mock import patch
+
+import pyarrow as pa
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.read.reader import format_blob_reader
+
+
+class _IndexProbe:
+ def __init__(self, file_io, cache):
+ self.file_io = file_io
+ self.cache = cache
+ self.local = threading.local()
+ self.lock = threading.Lock()
+ self.files = {}
+
+ def add(self, path, **values):
+ with self.lock:
+ counters = self.files.setdefault(path, Counter())
+ counters.update(values)
+
+ def install(self, stack):
+ original_index = format_blob_reader.FormatBlobReader._read_index
+ original_decode = format_blob_reader._decode_blob_index
+ io_type = type(self.file_io)
+ original_open = io_type.new_input_stream
+ probe = self
+
+ def read_index(reader):
+ assert reader._index_cache is probe.cache
+ probe.local.path = reader.file_path
+ probe.add(reader.file_path, reader_entries=1)
+ try:
+ return original_index(reader)
+ finally:
+ probe.local.path = None
+
+ def decode(data):
+ path = probe.local.path
+ started = time.thread_time()
+ result = original_decode(data)
+ probe.add(path, decodes=1, decoded_bytes=len(data),
+ decode_cpu_ms=(time.thread_time() - started) * 1000)
+ return result
+
+ class Stream:
+ def __init__(self, wrapped, path):
+ self.wrapped = wrapped
+ self.path = path
+
+ def __getattr__(self, name):
+ return getattr(self.wrapped, name)
+
+ def read(self, size=-1):
+ data = self.wrapped.read(size)
+ if getattr(probe.local, 'path', None) == self.path:
+ probe.add(self.path, index_read_calls=1,
+ index_read_bytes=len(data))
+ return data
+
+ def __enter__(self):
+ self.wrapped.__enter__()
+ return self
+
+ def __exit__(self, *args):
+ return self.wrapped.__exit__(*args)
+
+ def open_stream(file_io, path):
+ stream = original_open(file_io, path)
+ if str(path).endswith('.blob'):
+ return Stream(stream, str(path))
+ return stream
+
+ stack.enter_context(patch.object(
+ format_blob_reader.FormatBlobReader, '_read_index', read_index))
+ stack.enter_context(patch.object(
+ format_blob_reader, '_decode_blob_index', decode))
+ stack.enter_context(patch.object(io_type, 'new_input_stream',
open_stream))
+
+
+def _create_table(catalog, rows):
+ schema = pa.schema([('id', pa.int64()), ('payload', pa.large_binary())])
+ catalog.create_database('default', False)
+ catalog.create_table('default.blobs', Schema.from_pyarrow_schema(
+ schema, options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ }), False)
+ table = catalog.get_table('default.blobs')
+ expected = pa.table({
+ 'id': list(range(rows)),
+ 'payload': [('payload-%08d' % i).encode('ascii') for i in range(rows)],
+ }, schema=schema)
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(expected)
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ return table, expected
+
+
+def _plans(catalog, workers, cache):
+ jobs, details = [], []
+ for shard in range(workers):
+ table = catalog.get_table('default.blobs')
+ assert table.catalog_environment.blob_index_cache() is cache
+ assert not table.options.native_read_enabled()
+ builder = table.new_read_builder()
+ splits = builder.new_scan().with_shard(shard, workers).plan().splits()
+ assert len(splits) == 1
+ files = [f for f in splits[0].files if f.file_name.endswith('.blob')]
+ assert len(files) == 1
+ ranges = [(r.from_, r.to) for r in splits[0].row_ranges()]
+ details.append({'shard': shard, 'blob_file': files[0].file_name,
+ 'row_ranges': ranges})
+ jobs.append((builder, splits))
+ assert len({detail['blob_file'] for detail in details}) == 1
+ return jobs, details
+
+
+def _read_jobs(jobs, concurrent, parallelism):
+ barrier = threading.Barrier(len(jobs), timeout=60) if concurrent else None
+
+ def read(job):
+ builder, splits = job
+ reader = builder.new_read()
+ if barrier is not None:
+ barrier.wait()
+ return reader.to_arrow(splits, parallelism=parallelism)
+
+ if concurrent:
+ with ThreadPoolExecutor(max_workers=len(jobs)) as executor:
+ return list(executor.map(read, jobs))
+ return [read(job) for job in jobs]
+
+
+def _measure(name, jobs, concurrent, parallelism, cache, file_io,
+ expected, warm, instrumented):
+ cache.clear()
+ if warm:
+ _read_jobs(jobs, False, parallelism)
+ assert len(cache) == 1
+ probe = _IndexProbe(file_io, cache)
+ with ExitStack() as stack:
+ if instrumented:
+ probe.install(stack)
+ cpu_start, wall_start = time.process_time(), time.perf_counter()
+ results = _read_jobs(jobs, concurrent, parallelism)
+ wall_ms = (time.perf_counter() - wall_start) * 1000
+ cpu_ms = (time.process_time() - cpu_start) * 1000
+ # Validate every ID and payload, including shard order, outside the timer.
+ assert pa.concat_tables(results).equals(expected)
+ assert len(cache) == 1
+ totals = Counter()
+ for counters in probe.files.values():
+ totals.update(counters)
+ if instrumented:
+ assert len(probe.files) == 1
+ assert totals['reader_entries'] == len(jobs)
+ assert totals['index_read_calls'] == 2 * totals['decodes']
+ assert totals['index_read_bytes'] == (
+ totals['decoded_bytes'] + 5 * totals['decodes'])
+ if warm:
+ assert totals['decodes'] == 0
+ elif not concurrent:
+ assert totals['decodes'] == 1
+ else:
+ assert 1 <= totals['decodes'] <= len(jobs)
+ return {
+ 'scenario': name, 'warm': warm, 'instrumented': instrumented,
+ 'wall_ms': wall_ms, 'process_cpu_ms': cpu_ms,
+ 'cache_bytes_after': cache.size_bytes,
+ 'files': {Path(path).name: dict(counts)
+ for path, counts in probe.files.items()},
+ }
+
+
+def run(rows, workers, repeats):
+ with tempfile.TemporaryDirectory(prefix='blob-cold-load-') as root:
+ catalog = CatalogFactory.create({'warehouse': root})
+ table, expected = _create_table(catalog, rows)
+ cache = table.catalog_environment.blob_index_cache()
+ shard_jobs, details = _plans(catalog, workers, cache)
+ builder = table.new_read_builder()
+ full_splits = builder.new_scan().plan().splits()
+ assert len(full_splits) == 1
+ full_jobs = [(builder, full_splits)]
+ assert _read_jobs(full_jobs, False, workers)[0].equals(expected)
+ scenarios = [
+ ('single_query', full_jobs, False, workers),
+ ('serial_shards', shard_jobs, False, 1),
+ ('concurrent_shards', shard_jobs, True, 1),
+ ]
+ runs = []
+ rng = random.Random(10151)
+ for instrumented in (True, False):
+ for repeat in range(repeats):
+ cases = [(scenario, warm) for scenario in scenarios
+ for warm in (False, True)]
+ rng.shuffle(cases)
+ for (name, jobs, concurrent, parallelism), warm in cases:
+ result = _measure(
+ name, jobs, concurrent, parallelism, cache,
+ table.file_io, expected, warm, instrumented)
+ result['repeat'] = repeat
+ runs.append(result)
+ print('instrumented=%s repeat=%d/%d passed' % (
+ instrumented, repeat + 1, repeats), file=sys.stderr)
+ return {
+ 'python': sys.version, 'pyarrow': pa.__version__,
+ 'platform': platform.platform(), 'rows': rows,
+ 'workers': workers, 'repeats': repeats,
+ 'native_read_enabled': table.options.native_read_enabled(),
+ 'cache_budget_bytes': cache.max_size_bytes,
+ 'shards': details, 'runs': runs,
+ }
+
+
+def main():
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument('--rows', type=int, default=200000)
+ parser.add_argument('--workers', type=int, default=8)
+ parser.add_argument('--repeats', type=int, default=5)
+ parser.add_argument('--output', type=Path, required=True)
+ args = parser.parse_args()
+ if args.workers < 2 or args.rows < args.workers or args.repeats < 1:
+ parser.error('Require workers >= 2, rows >= workers, repeats >= 1')
+ result = run(args.rows, args.workers, args.repeats)
+ args.output.write_text(json.dumps(result, indent=2) + '\n',
encoding='utf-8')
+ print(str(args.output))
+
+
+if __name__ == '__main__':
+ main()
diff --git a/paimon-python/pypaimon/common/blob_index_cache.py
b/paimon-python/pypaimon/common/blob_index_cache.py
index 611ed14312..acf3c47b9e 100644
--- a/paimon-python/pypaimon/common/blob_index_cache.py
+++ b/paimon-python/pypaimon/common/blob_index_cache.py
@@ -15,9 +15,12 @@
# specific language governing permissions and limitations
# under the License.
+import os
import sys
from collections import namedtuple
+from concurrent.futures import Future
from threading import Lock
+from types import SimpleNamespace
from cachetools import LRUCache
@@ -40,17 +43,27 @@ class BlobIndexCache:
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()
+ self._states = {}
+
+ def _state(self):
+ pid = os.getpid()
+ states = self._states
+ state = states.get(pid)
+ if state is None:
+ # Publish a complete state atomically; never touch inherited locks
or Futures.
+ state = states.setdefault(pid, SimpleNamespace(
+ lock=Lock(), loading={}, cache=LRUCache(
+ maxsize=self.max_size_bytes,
+ getsizeof=lambda entry: entry.size_bytes)))
+ self._states = {pid: state}
+ return state
def get(self, file_path):
if self.max_size_bytes == 0:
return None
- with self._lock:
- entry = self._cache.get(file_path)
+ state = self._state()
+ with state.lock:
+ entry = state.cache.get(file_path)
if entry is None:
return None
return entry.blob_lengths, entry.blob_offsets
@@ -69,25 +82,61 @@ class BlobIndexCache:
if size_bytes > self.max_size_bytes:
return
entry = _CachedBlobIndex(blob_lengths, blob_offsets, size_bytes)
- with self._lock:
- self._cache[file_path] = entry
+ state = self._state()
+ with state.lock:
+ state.cache[file_path] = entry
+
+ def get_or_load(self, file_path, loader):
+ """Share an in-flight load without holding the cache lock during
I/O."""
+ if self.max_size_bytes == 0:
+ return loader()
+ state = self._state()
+ with state.lock:
+ entry = state.cache.get(file_path)
+ if entry is not None:
+ return entry.blob_lengths, entry.blob_offsets
+ pending = state.loading.get(file_path)
+ owner = pending is None
+ if owner:
+ pending = Future()
+ state.loading[file_path] = pending
+
+ if not owner:
+ return pending.result()
+ try:
+ index = loader()
+ self.put(file_path, *index)
+ except BaseException as error:
+ # Unblock waiters even if the loading thread is interrupted.
+ pending.set_exception(error)
+ raise
+ else:
+ pending.set_result(index)
+ return index
+ finally:
+ with state.lock:
+ del state.loading[file_path]
def __len__(self):
- with self._lock:
- return len(self._cache)
+ state = self._state()
+ with state.lock:
+ return len(state.cache)
def __contains__(self, file_path):
- with self._lock:
- return file_path in self._cache
+ state = self._state()
+ with state.lock:
+ return file_path in state.cache
def clear(self):
- with self._lock:
- self._cache.clear()
+ state = self._state()
+ with state.lock:
+ state.cache.clear()
@property
def size_bytes(self):
- with self._lock:
- return self._cache.currsize
+ state = self._state()
+ with state.lock:
+ return state.cache.currsize
def __reduce__(self):
return type(self), (MemorySize.of_bytes(self.max_size_bytes),)
diff --git a/paimon-python/pypaimon/read/reader/format_blob_reader.py
b/paimon-python/pypaimon/read/reader/format_blob_reader.py
index c959228cff..c95626d7fe 100644
--- a/paimon-python/pypaimon/read/reader/format_blob_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_blob_reader.py
@@ -374,13 +374,12 @@ class FormatBlobReader(RecordBatchReader):
)
return
- 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)
- self.blob_offsets = list(blob_offsets)
- return
+ blob_lengths, blob_offsets = self._index_cache.get_or_load(
+ self.file_path, self._load_index)
+ self.blob_lengths = list(blob_lengths)
+ self.blob_offsets = list(blob_offsets)
+ def _load_index(self):
self._input_stream = self._file_io.new_input_stream(self.file_path)
f = self._input_stream
@@ -405,10 +404,7 @@ class FormatBlobReader(RecordBatchReader):
if len(index_bytes) != index_length:
raise IOError("Invalid blob file: cannot read index")
- blob_lengths, blob_offsets = _decode_blob_index(index_bytes)
- self._index_cache.put(self.file_path, blob_lengths, blob_offsets)
- self.blob_lengths = list(blob_lengths)
- self.blob_offsets = list(blob_offsets)
+ return _decode_blob_index(index_bytes)
def _apply_row_indices(self, row_indices: Optional[Any]) -> None:
if row_indices is None:
diff --git a/paimon-python/pypaimon/tests/blob_index_cache_test.py
b/paimon-python/pypaimon/tests/blob_index_cache_test.py
new file mode 100644
index 0000000000..bb6f07b90b
--- /dev/null
+++ b/paimon-python/pypaimon/tests/blob_index_cache_test.py
@@ -0,0 +1,353 @@
+# 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 multiprocessing
+import os
+import pickle
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from unittest.mock import Mock, patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.common.blob_index_cache import BlobIndexCache
+from pypaimon.common.memory_size import MemorySize
+from pypaimon.common.options import Options
+from pypaimon.filesystem.local_file_io import LocalFileIO
+from pypaimon.read.reader import format_blob_reader
+from pypaimon.schema.data_types import AtomicType, DataField
+
+
[email protected]('budget', [1, 1024 * 1024])
[email protected]('error_type', [None, IOError, KeyboardInterrupt])
+def test_concurrent_load_result_and_failure_are_shared(budget, error_type):
+ cache = BlobIndexCache(MemorySize.of_bytes(budget))
+ entered, release, waiting = (threading.Event() for _ in range(3))
+ index = ((3,), (0,))
+
+ def load():
+ entered.set()
+ assert release.wait(10)
+ if error_type is not None:
+ raise error_type('index load failed')
+ return index
+
+ loader = Mock(side_effect=load)
+ with ThreadPoolExecutor(max_workers=2) as executor:
+ owner = executor.submit(cache.get_or_load, 'file.blob', loader)
+ try:
+ assert entered.wait(10)
+ pending = cache._state().loading['file.blob']
+ original_result = pending.result
+
+ def wait_for_result():
+ waiting.set()
+ return original_result()
+
+ with patch.object(pending, 'result', wait_for_result):
+ waiter = executor.submit(cache.get_or_load, 'file.blob',
loader)
+ assert waiting.wait(10)
+ release.set()
+ for future in (owner, waiter):
+ if error_type is None:
+ assert future.result(timeout=10) is index
+ else:
+ with pytest.raises(error_type, match='index load
failed'):
+ future.result(timeout=10)
+ finally:
+ release.set()
+ assert loader.call_count == 1
+ assert not cache._state().loading
+ retained = error_type is None and budget > 1
+ assert len(cache) == int(retained)
+ retry = Mock(return_value=index)
+ assert cache.get_or_load('file.blob', retry) == index
+ assert retry.call_count == int(not retained)
+
+
[email protected]('separate_caches', [False, True])
+def test_other_files_and_catalogs_do_not_wait(separate_caches):
+ first = BlobIndexCache(MemorySize.of_mebi_bytes(1))
+ second = BlobIndexCache(MemorySize.of_mebi_bytes(1)) if separate_caches
else first
+ second_path = 'a.blob' if separate_caches else 'b.blob'
+ both_loading = threading.Barrier(2, timeout=10)
+
+ def load():
+ both_loading.wait()
+ return ((1,), (0,))
+
+ with ThreadPoolExecutor(max_workers=2) as executor:
+ futures = [executor.submit(first.get_or_load, 'a.blob', load),
+ executor.submit(second.get_or_load, second_path, load)]
+ for future in futures:
+ assert future.result(timeout=10) == ((1,), (0,))
+
+
+def test_disabled_cache_does_not_share_or_retain_loads():
+ cache = BlobIndexCache(MemorySize.of_bytes(0))
+ both_loading = threading.Barrier(2, timeout=10)
+
+ def load():
+ both_loading.wait()
+ return ((1,), (0,))
+
+ loader = Mock(side_effect=load)
+ with ThreadPoolExecutor(max_workers=2) as executor:
+ futures = [executor.submit(cache.get_or_load, 'a.blob', loader) for _
in range(2)]
+ for future in futures:
+ assert future.result(timeout=10) == ((1,), (0,))
+ assert loader.call_count == 2
+ assert len(cache) == 0
+ assert not cache._state().loading
+
+
+def test_hot_cache_does_not_create_a_future():
+ cache = BlobIndexCache(MemorySize.of_mebi_bytes(1))
+ cache.put('a.blob', (1,), (0,))
+ loader = Mock(side_effect=AssertionError('hot cache invoked loader'))
+ with patch('pypaimon.common.blob_index_cache.Future',
+ side_effect=AssertionError('hot cache created future')):
+ assert cache.get_or_load('a.blob', loader) == ((1,), (0,))
+
+
+def test_serialization_does_not_copy_inflight_loads():
+ cache = BlobIndexCache(MemorySize.of_mebi_bytes(1))
+ entered, release = threading.Event(), threading.Event()
+
+ def load():
+ entered.set()
+ assert release.wait(10)
+ return ((1,), (0,))
+
+ with ThreadPoolExecutor(max_workers=1) as executor:
+ owner = executor.submit(cache.get_or_load, 'a.blob', load)
+ try:
+ assert entered.wait(10)
+ restored = pickle.loads(pickle.dumps(cache))
+ assert restored.max_size_bytes == cache.max_size_bytes
+ assert len(restored) == 0
+ assert not restored._state().loading
+ assert restored.get_or_load('a.blob', lambda: ((2,), (0,))) ==
((2,), (0,))
+ finally:
+ release.set()
+ assert owner.result(timeout=10) == ((1,), (0,))
+ assert cache.get('a.blob') == ((1,), (0,))
+
+
[email protected]
+def fork_context():
+ if 'fork' not in multiprocessing.get_all_start_methods():
+ pytest.skip('fork required')
+ return multiprocessing.get_context('fork')
+
+
+def test_forked_blob_reader_does_not_wait_for_parent_load(tmp_path,
fork_context):
+ cache = BlobIndexCache(MemorySize.of_mebi_bytes(1))
+ file_io = LocalFileIO(str(tmp_path), Options({}))
+ path = str(tmp_path / 'payload.blob')
+ expected = [b'first', None, b'last']
+ file_io.write_blob(path, pa.table({'payload': expected}, schema=pa.schema([
+ ('payload', pa.large_binary())])))
+ field = DataField(0, 'payload', AtomicType('BLOB'))
+ entered, release = threading.Event(), threading.Event()
+ parent_pid = os.getpid()
+ original_decode = format_blob_reader._decode_blob_index
+
+ def decode(data):
+ if os.getpid() == parent_pid:
+ entered.set()
+ assert release.wait(10)
+ return original_decode(data)
+
+ def read():
+ reader = format_blob_reader.FormatBlobReader(
+ file_io, path, [field.name], [field], None, False,
index_cache=cache)
+ try:
+ return reader.read_arrow_batch().column(0).to_pylist()
+ finally:
+ reader.close()
+
+ def child(connection):
+ try:
+ # The parent finishes first, but its Future remains incomplete
here.
+ connection.recv()
+ with patch.object(format_blob_reader, '_decode_blob_index',
+ wraps=original_decode) as child_decode:
+ values = [read(), read()]
+ connection.send((values, child_decode.call_count, len(cache)))
+ except BaseException as error:
+ connection.send(('error', repr(error)))
+ finally:
+ connection.close()
+
+ receiving, sending = fork_context.Pipe()
+ process = fork_context.Process(target=child, args=(sending,))
+ with patch.object(format_blob_reader, '_decode_blob_index',
side_effect=decode), \
+ ThreadPoolExecutor(max_workers=1) as executor:
+ owner = executor.submit(read)
+ try:
+ assert entered.wait(5)
+ assert cache._state().lock.acquire(blocking=False)
+ cache._state().lock.release()
+ process.start()
+ sending.close()
+ release.set()
+ assert owner.result(timeout=5) == expected
+ assert not cache._state().loading
+ receiving.send('parent finished')
+ assert receiving.poll(5), 'child waited for an inherited Future'
+ assert receiving.recv() == ([expected, expected], 1, 1)
+ process.join(5)
+ assert process.exitcode == 0
+ assert len(cache) == 1
+ finally:
+ release.set()
+ if process.is_alive():
+ process.terminate()
+ process.join(5)
+ receiving.close()
+ sending.close()
+
+
+def test_forked_cache_concurrent_first_access_avoids_parent_lock(fork_context):
+ cache = BlobIndexCache(MemorySize.of_mebi_bytes(1))
+ cache.put('parent.blob', (1,), (0,))
+
+ def child(connection):
+ creating, resume = threading.Event(), threading.Event()
+ results = []
+ loader = Mock(return_value=((2,), (0,)))
+ original_lock = threading.Lock
+
+ def new_lock():
+ if threading.current_thread().name == 'first':
+ creating.set()
+ assert resume.wait(5)
+ return original_lock()
+
+ def access():
+ results.append(cache.get_or_load('child.blob', loader))
+
+ try:
+ with patch('pypaimon.common.blob_index_cache.Lock',
side_effect=new_lock):
+ first = threading.Thread(target=access, name='first',
daemon=True)
+ second = threading.Thread(target=access, name='second',
daemon=True)
+ first.start()
+ assert creating.wait(3), 'child reused the parent lock'
+ second.start()
+ second.join(3)
+ assert not second.is_alive()
+ resume.set()
+ first.join(3)
+ assert not first.is_alive()
+ connection.send((results, loader.call_count, len(cache),
+ cache.get('parent.blob'), cache.size_bytes > 0))
+ cache.clear()
+ assert len(cache) == 0 and 'child.blob' not in cache
+ except BaseException as error:
+ connection.send(('error', repr(error)))
+ finally:
+ resume.set()
+ connection.close()
+
+ receiving, sending = fork_context.Pipe(duplex=False)
+ process = fork_context.Process(target=child, args=(sending,))
+ with cache._state().lock:
+ try:
+ process.start()
+ sending.close()
+ assert receiving.poll(10), 'child deadlocked on the inherited lock'
+ assert receiving.recv() == ([((2,), (0,))] * 2, 1, 1, None, True)
+ process.join(5)
+ assert process.exitcode == 0
+ finally:
+ if process.is_alive():
+ process.terminate()
+ process.join(5)
+ receiving.close()
+ sending.close()
+ assert cache.get('parent.blob') == ((1,), (0,))
+ assert cache.get('child.blob') is None
+
+
[email protected]_plan
[email protected]_read
[email protected]('blob_parallelism', [1, 4])
+def test_public_shard_reads_share_index_and_keep_payloads(tmp_path,
blob_parallelism):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', False)
+ schema = pa.schema([('id', pa.int64()), ('payload', pa.large_binary())])
+ catalog.create_table('default.blobs', Schema.from_pyarrow_schema(schema,
options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ }), False)
+ table = catalog.get_table('default.blobs')
+ expected = pa.table({'id': list(range(1000)),
+ 'payload': [('value-%d' % i).encode() for i in
range(1000)]},
+ schema=schema)
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(expected)
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+
+ cache = table.catalog_environment.blob_index_cache()
+ jobs = []
+ for shard in range(4):
+ shard_table = catalog.get_table('default.blobs')
+ assert shard_table.catalog_environment.blob_index_cache() is cache
+ builder = shard_table.new_read_builder()
+ splits = builder.new_scan().with_shard(shard, 4).plan().splits()
+ assert len(splits) == 1
+ jobs.append((builder, splits))
+ paths = [file.physical_path() for _, splits in jobs
+ for file in splits[0].files if file.file_name.endswith('.blob')]
+ assert len(paths) == 4 and len(set(paths)) == 1
+
+ start = threading.Barrier(4, timeout=10)
+ streams = []
+ stream_lock = threading.Lock()
+ original_open = LocalFileIO.new_input_stream
+
+ def open_stream(file_io, path):
+ stream = original_open(file_io, path)
+ if str(path).endswith('.blob'):
+ with stream_lock:
+ streams.append(stream)
+ return stream
+
+ def read(job):
+ builder, splits = job
+ start.wait()
+ return builder.new_read().to_arrow(
+ splits, parallelism=1, blob_parallelism=blob_parallelism)
+
+ with patch.object(LocalFileIO, 'new_input_stream', open_stream),
patch.object(
+ format_blob_reader, '_decode_blob_index',
+ wraps=format_blob_reader._decode_blob_index) as decode:
+ for _ in range(2):
+ with ThreadPoolExecutor(max_workers=4) as executor:
+ results = list(executor.map(read, jobs))
+ assert pa.concat_tables(results).equals(expected)
+ assert decode.call_count == 1
+ assert all(stream.closed for stream in streams)
+ assert len(cache) == 1
+ assert not cache._state().loading
