This is an automated email from the ASF dual-hosted git repository. discivigour pushed a commit to branch feat/blobPara in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 295e97466b4adea498604196082a0f7880f3a093 Author: umi <[email protected]> AuthorDate: Tue Jul 14 17:22:06 2026 +0800 feat(python): parallelize fallback blob reads --- .../pypaimon/read/reader/concat_batch_reader.py | 35 ++++++- paimon-python/pypaimon/read/split_read.py | 7 +- paimon-python/pypaimon/tests/blob_test.py | 107 ++++++++++++++++++++- 3 files changed, 144 insertions(+), 5 deletions(-) diff --git a/paimon-python/pypaimon/read/reader/concat_batch_reader.py b/paimon-python/pypaimon/read/reader/concat_batch_reader.py index f17092279c..9b70287c2b 100644 --- a/paimon-python/pypaimon/read/reader/concat_batch_reader.py +++ b/paimon-python/pypaimon/read/reader/concat_batch_reader.py @@ -259,7 +259,8 @@ class BlobFallbackBatchReader(RecordBatchReader): def __init__(self, file_reader_suppliers: List[Tuple[DataFileMeta, Callable]], field_name: str, output_type, row_ranges: Optional[List[Range]] = None, - blob_as_descriptor: bool = False, deletion_vector=None, batch_size: int = 1024): + blob_as_descriptor: bool = False, deletion_vector=None, batch_size: int = 1024, + blob_parallelism: int = 1): self._file_reader_suppliers = file_reader_suppliers self._field_name = field_name self._output_type = output_type @@ -272,6 +273,8 @@ class BlobFallbackBatchReader(RecordBatchReader): self._deletion_vector_range, self._deletion_vector = deletion_vector self._returned = False self._batch_size = max(1, batch_size) + self._blob_parallelism = max(1, blob_parallelism) + self._file_io = None self._target_ranges = self._compute_target_ranges() self._target_range_index = 0 self._next_row_id = ( @@ -290,6 +293,9 @@ class BlobFallbackBatchReader(RecordBatchReader): if not batch_row_ids: return None + resolve_blobs_concurrently = ( + self._blob_parallelism > 1 and not self._blob_as_descriptor + ) groups: Dict[int, Dict[int, Tuple[object, bool]]] = {} batch_first = batch_row_ids[0] @@ -301,6 +307,8 @@ class BlobFallbackBatchReader(RecordBatchReader): if not blob_values: continue group = groups.setdefault(state.file.max_sequence_number, {}) + if resolve_blobs_concurrently and self._file_io is None: + self._file_io = state.reader._file_io for row_id, blob in blob_values.items(): if row_id in group: raise ValueError( @@ -313,6 +321,11 @@ class BlobFallbackBatchReader(RecordBatchReader): else: if self._blob_as_descriptor: group[row_id] = (blob.to_descriptor().serialize(), False) + elif resolve_blobs_concurrently: + # Keep values lazy until fallback selects the newest + # non-placeholder version for each row. Otherwise older + # overridden BLOB versions would be read unnecessarily. + group[row_id] = (blob, False) else: group[row_id] = (blob.to_data(), False) @@ -334,11 +347,31 @@ class BlobFallbackBatchReader(RecordBatchReader): if not found: raise ValueError("All blob files at the same row id store a placeholder.") + if resolve_blobs_concurrently: + result = self._resolve_selected_blobs(result) + return pa.RecordBatch.from_arrays( [pa.array(result, type=self._output_type)], names=[self._field_name], ) + def _resolve_selected_blobs(self, values: List[object]) -> List[object]: + """Materialize selected BLOBs concurrently with the shared FileIO.""" + resolved = list(values) + indexed_blobs = [ + (index, value) + for index, value in enumerate(values) + if isinstance(value, Blob) + ] + if not indexed_blobs: + return resolved + + bodies = self._file_io.read_blobs_concurrent( + [blob for _, blob in indexed_blobs], self._blob_parallelism) + for (index, _), body in zip(indexed_blobs, bodies): + resolved[index] = body + return resolved + def _compute_target_ranges(self) -> List[Range]: ranges = Range.sort_and_merge_overlap([ file.row_id_range() diff --git a/paimon-python/pypaimon/read/split_read.py b/paimon-python/pypaimon/read/split_read.py index 0b01ae4507..ed690a78da 100644 --- a/paimon-python/pypaimon/read/split_read.py +++ b/paimon-python/pypaimon/read/split_read.py @@ -284,7 +284,7 @@ class SplitRead(ABC): raise NotImplementedError( "Nested-field projection is not supported on BLOB files") blob_as_descriptor = CoreOptions.blob_as_descriptor(self.table.options) - blob_parallelism = getattr(self, '_blob_parallelism', 1) + blob_parallelism = self._blob_parallelism format_reader = FormatBlobReader(self.table.file_io, file_path, read_file_fields, self.read_fields, read_arrow_predicate, blob_as_descriptor, batch_size=batch_size, @@ -1005,7 +1005,7 @@ class DataEvolutionSplitRead(SplitRead): self.table.options)) or (not CoreOptions.blob_as_descriptor(self.table.options) and CoreOptions.blob_descriptor_fields(self.table.options))): - blob_parallelism = getattr(self, '_blob_parallelism', 1) + blob_parallelism = self._blob_parallelism reader = BlobInlineConvertReader( reader, self.table, prescan_reader_factory=lambda names: self._create_prescan_reader(names), @@ -1284,6 +1284,7 @@ class DataEvolutionSplitRead(SplitRead): CoreOptions.blob_as_descriptor(self.table.options), deletion_vector=deletion_vector, batch_size=batch_size, + blob_parallelism=self._blob_parallelism, ) else: # Create concatenated reader for multiple files @@ -1328,7 +1329,7 @@ class DataEvolutionSplitRead(SplitRead): return None file_path = file.external_path if file.external_path else file.file_path - blob_parallelism = getattr(self, '_blob_parallelism', 1) + blob_parallelism = self._blob_parallelism return FormatBlobReader( self.table.file_io, file_path, diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index cd79d5d432..90225366c0 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -251,6 +251,12 @@ class BlobTest(unittest.TestCase): def test_blob_fallback_batch_reader_respects_batch_size(self): created_readers = [] + class DescriptorBlobFallbackBatchReader(BlobFallbackBatchReader): + def _resolve_selected_blobs(self, values): + raise AssertionError( + "Descriptor reads should not materialize BLOB data." + ) + class FakeBlobReader: def __init__(self): self._file_io = None @@ -284,12 +290,13 @@ class BlobTest(unittest.TestCase): first_row_id=10, file_path="fake.blob", ) - reader = BlobFallbackBatchReader( + reader = DescriptorBlobFallbackBatchReader( [(data_file, supplier)], "picture", pa.large_binary(), blob_as_descriptor=True, batch_size=2, + blob_parallelism=4, ) first = reader.read_arrow_batch() @@ -399,6 +406,86 @@ class BlobTest(unittest.TestCase): self.assertTrue(created_by_file["old.blob"][0].closed) self.assertTrue(created_by_file["new.blob"][0].closed) + def test_blob_fallback_batch_reader_materializes_selected_values_in_parallel(self): + class RecordingFileIO: + def __init__(self): + self.calls = [] + + def read_blobs_concurrent(self, blobs, parallelism): + descriptors = [blob.to_descriptor() for blob in blobs] + self.calls.append((descriptors, parallelism)) + return [ + "{}:{}".format(descriptor.uri, descriptor.offset).encode() + for descriptor in descriptors + ] + + class FakeBlobReader: + def __init__(self, file_io, file_path, blob_lengths, blob_offsets): + self._file_io = file_io + self.file_path = file_path + self.blob_lengths = blob_lengths + self.blob_offsets = blob_offsets + self._input_stream = None + + def close(self): + pass + + def data_file(name, max_sequence_number): + return DataFileMeta( + file_name=name, + file_size=0, + row_count=3, + min_key=None, + max_key=None, + key_stats=None, + value_stats=None, + min_sequence_number=max_sequence_number, + max_sequence_number=max_sequence_number, + schema_id=0, + level=0, + extra_files=[], + first_row_id=0, + file_path=name, + ) + + file_io = RecordingFileIO() + old_file = data_file("old.blob", 1) + new_file = data_file("new.blob", 2) + reader = BlobFallbackBatchReader( + [ + ( + old_file, + lambda: FakeBlobReader( + file_io, "old.blob", [20, 20, 20], [0, 100, 200] + ), + ), + ( + new_file, + lambda: FakeBlobReader( + file_io, "new.blob", [-2, 20, -2], [-1, 1000, -1] + ), + ), + ], + "picture", + pa.large_binary(), + batch_size=3, + blob_parallelism=4, + ) + + batch = reader.read_arrow_batch() + + self.assertEqual( + [b"old.blob:4", b"new.blob:1004", b"old.blob:204"], + batch.column("picture").to_pylist(), + ) + self.assertEqual(1, len(file_io.calls)) + descriptors, parallelism = file_io.calls[0] + self.assertEqual(4, parallelism) + self.assertEqual( + [("old.blob", 4), ("new.blob", 1004), ("old.blob", 204)], + [(descriptor.uri, descriptor.offset) for descriptor in descriptors], + ) + def test_blob_data_interface_compliance(self): """Test that BlobData properly implements Blob interface.""" test_data = b"interface test data" @@ -1757,6 +1844,24 @@ class BlobParallelismTest(unittest.TestCase): for i in range(20): self.assertEqual(got[i], self.payloads[i]) + def test_blob_parallelism_with_multiple_blob_files(self): + t = self.catalog.get_table('default.bp_test') + extra_payloads = [os.urandom(512) for _ in range(20)] + w = t.new_batch_write_builder().new_write() + w.write_arrow(pa.Table.from_pydict( + {'id': list(range(20, 40)), 'img': extra_payloads}, + schema=pa.schema([('id', pa.int32()), ('img', pa.large_binary())]))) + t.new_batch_write_builder().new_commit().commit(w.prepare_commit()) + w.close() + + rb = t.new_read_builder().with_projection(['id', 'img']) + splits = rb.new_scan().plan().splits() + serial = rb.new_read().to_arrow(splits, blob_parallelism=1) + parallel = rb.new_read().to_arrow(splits, blob_parallelism=4) + + self.assertEqual(40, serial.num_rows) + self.assertEqual(serial.to_pydict(), parallel.to_pydict()) + class CapBlobParallelismTest(unittest.TestCase): """Peak blob threads on the parallel path (workers * blob_parallelism)
