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 933421aaf1 [python] Resolve BlobView blobs with upstream FileIO (#8488)
933421aaf1 is described below
commit 933421aaf127890315c0d79a050c588dec154076
Author: umi <[email protected]>
AuthorDate: Tue Jul 7 15:02:14 2026 +0800
[python] Resolve BlobView blobs with upstream FileIO (#8488)
---
.../read/reader/blob_descriptor_convert_reader.py | 77 +++++++++++----
paimon-python/pypaimon/tests/blob_table_test.py | 108 +++++++++++++++++++++
paimon-python/pypaimon/utils/blob_view_lookup.py | 32 +++++-
3 files changed, 195 insertions(+), 22 deletions(-)
diff --git
a/paimon-python/pypaimon/read/reader/blob_descriptor_convert_reader.py
b/paimon-python/pypaimon/read/reader/blob_descriptor_convert_reader.py
index 92b30ebbc4..f6c5f73cfe 100644
--- a/paimon-python/pypaimon/read/reader/blob_descriptor_convert_reader.py
+++ b/paimon-python/pypaimon/read/reader/blob_descriptor_convert_reader.py
@@ -33,10 +33,12 @@ class BlobInlineConvertReader(RecordBatchReader):
prescan reader (only projecting view columns) to collect
BlobViewStructs, bulk-preload their descriptors, then read
full data from the main reader and replace view field values
- with the corresponding BlobDescriptor serialized bytes.
- Stage 2 (BlobData resolution): Controlled by blob-as-descriptor option.
- If false, resolve all BlobDescriptor bytes (from both descriptor
- fields and view fields) into real blob data bytes.
+ with descriptor bytes or real blob data according to the
+ blob-as-descriptor option.
+ Stage 2 (BlobDescriptor resolution): Controlled by blob-as-descriptor
option.
+ If false, resolve BlobDescriptor bytes from descriptor fields
+ into real blob data bytes. BlobView fields are already resolved
+ in Stage 1 with the upstream table FileIO.
If true, return as-is.
"""
@@ -72,8 +74,6 @@ class BlobInlineConvertReader(RecordBatchReader):
def read_arrow_batch(self) -> Optional[RecordBatch]:
# Align with Java: only enter blob view resolution when catalog_loader
is available
# If catalog_loader is None, skip both Stage 1 (view resolution) and
Stage 2 (descriptor resolution)
- # This matches Java's behavior in DataEvolutionTableRead.createReader
where blob view reader
- # is only created when catalogContext != null
if self._view_fields and not self._prescan_done:
self._prescan_view_structs()
@@ -81,10 +81,11 @@ class BlobInlineConvertReader(RecordBatchReader):
if batch is None:
return None
# Resolve view fields using the preloaded lookup
+ view_file_ios = {}
if self._view_fields and self._blob_view_lookup is not None:
- batch = self._resolve_view_fields(batch, self._blob_view_lookup)
+ batch, view_file_ios = self._resolve_view_fields(batch,
self._blob_view_lookup)
# Resolve BlobDescriptor -> real bytes (if blob-as-descriptor=false)
- return self._resolve_descriptor_fields(batch)
+ return self._resolve_descriptor_fields(batch, view_file_ios)
# ------------------------------------------------------------------
# Stage 1: BlobView prescan (lightweight, only reads view columns)
@@ -128,29 +129,30 @@ class BlobInlineConvertReader(RecordBatchReader):
self._prescan_done = True
def _resolve_view_fields(self, batch, blob_view_lookup):
- """Replace BlobViewStruct bytes in view fields with the corresponding
- BlobDescriptor serialized bytes."""
+ """Replace BlobViewStruct bytes in view fields with descriptor
bytes."""
+ view_file_ios = {}
for field_name in self._view_fields:
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()]
converted_values = []
+ field_file_ios = []
for value in values:
- if value is None:
- converted_values.append(None)
- continue
- if not isinstance(value, bytes):
- converted_values.append(value)
- continue
- if not BlobViewStruct.is_blob_view_struct(value):
+ if value is None or not (
+ isinstance(value, bytes) and
BlobViewStruct.is_blob_view_struct(value)):
converted_values.append(value)
+ field_file_ios.append(None)
continue
+
view_struct = BlobViewStruct.deserialize(value)
if blob_view_lookup.resolve_to_null(view_struct):
converted_values.append(None)
+ field_file_ios.append(None)
else:
descriptor =
blob_view_lookup.resolve_descriptor(view_struct)
converted_values.append(descriptor.serialize())
+ file_io = blob_view_lookup.resolve_file_io(view_struct)
+ field_file_ios.append(file_io)
column_idx = batch.schema.names.index(field_name)
batch = batch.set_column(
@@ -158,18 +160,18 @@ class BlobInlineConvertReader(RecordBatchReader):
pyarrow.field(field_name, pyarrow.large_binary(),
nullable=True),
pyarrow.array(converted_values, type=pyarrow.large_binary()),
)
- return batch
+ view_file_ios[field_name] = field_file_ios
+ return batch, view_file_ios
# ------------------------------------------------------------------
# Stage 2: BlobData resolution (unified exit)
# ------------------------------------------------------------------
- def _resolve_descriptor_fields(self, batch):
+ def _resolve_descriptor_fields(self, batch, view_file_ios=None):
if self._blob_as_descriptor:
return batch
- all_inline_blob_fields = self._descriptor_fields | self._view_fields
- for field_name in all_inline_blob_fields:
+ for field_name in self._descriptor_fields:
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()]
@@ -187,6 +189,39 @@ class BlobInlineConvertReader(RecordBatchReader):
pyarrow.field(field_name, pyarrow.large_binary(),
nullable=True),
pyarrow.array(converted_values, type=pyarrow.large_binary()),
)
+
+ view_file_ios = view_file_ios or {}
+ for field_name in self._view_fields:
+ field_file_ios = view_file_ios.get(field_name)
+ if field_name not in batch.schema.names or field_file_ios is None:
+ continue
+ values = [self._normalize_blob_to_bytes(v) for v in
batch.column(field_name).to_pylist()]
+ blobs_by_file_io = {}
+ converted_values = []
+
+ for idx, value in enumerate(values):
+ file_io = field_file_ios[idx] or self._table.file_io
+ blob = Blob.from_bytes(value, file_io)
+ if self._blob_parallelism > 1:
+ converted_values.append(None)
+ if blob is not None:
+ blobs_by_file_io.setdefault(file_io, []).append((idx,
blob))
+ else:
+ converted_values.append(blob.to_data() if blob else None)
+
+ for file_io, indexed_blobs in blobs_by_file_io.items():
+ blobs = [item[1] for item in indexed_blobs]
+ results = file_io.read_blobs_concurrent(blobs,
self._blob_parallelism)
+ for (idx, _), data in zip(indexed_blobs, results):
+ converted_values[idx] = data
+
+ column_idx = batch.schema.names.index(field_name)
+ batch = batch.set_column(
+ column_idx,
+ pyarrow.field(field_name, pyarrow.large_binary(),
nullable=True),
+ pyarrow.array(converted_values, type=pyarrow.large_binary()),
+ )
+
return batch
# ------------------------------------------------------------------
diff --git a/paimon-python/pypaimon/tests/blob_table_test.py
b/paimon-python/pypaimon/tests/blob_table_test.py
index e45b4c77bf..57228c7cbf 100755
--- a/paimon-python/pypaimon/tests/blob_table_test.py
+++ b/paimon-python/pypaimon/tests/blob_table_test.py
@@ -2143,6 +2143,114 @@ class DedicatedFormatWriterTest(unittest.TestCase):
"Expected BlobDescriptor bytes when blob-as-descriptor=true"
)
+ def test_blob_view_default_read_uses_upstream_file_io(self):
+ from pypaimon import Schema
+ from pypaimon.table.row.blob import BlobViewStruct
+
+ class GuardedFileIO:
+ def __init__(self, wrapped, forbidden_uris):
+ self._wrapped = wrapped
+ self._forbidden_uris = forbidden_uris
+ self.forbidden_reads = []
+
+ def new_input_stream(self, path):
+ path = str(path)
+ if path in self._forbidden_uris:
+ self.forbidden_reads.append(path)
+ raise AssertionError(
+ "Downstream file_io must not read upstream blob
{}.".format(path)
+ )
+ return self._wrapped.new_input_stream(path)
+
+ def __getattr__(self, name):
+ return getattr(self._wrapped, name)
+
+ source_schema = pa.schema([
+ ('id', pa.int32()),
+ ('picture', pa.large_binary()),
+ ])
+ source = Schema.from_pyarrow_schema(
+ source_schema,
+ options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ }
+ )
+ self.catalog.create_table('test_db.blob_view_guard_source', source,
False)
+ source_table = self.catalog.get_table('test_db.blob_view_guard_source')
+ payloads = [b'guard-source-0', b'guard-source-1']
+
+ write_builder = source_table.new_batch_write_builder()
+ writer = write_builder.new_write()
+ writer.write_arrow(pa.Table.from_pydict({
+ 'id': [1, 2],
+ 'picture': payloads,
+ }, schema=source_schema))
+ source_commit_messages = writer.prepare_commit()
+ write_builder.new_commit().commit(source_commit_messages)
+ writer.close()
+
+ source_blob_paths = {
+ str(f.file_path)
+ for msg in source_commit_messages
+ for f in msg.new_files
+ if f.file_name.endswith('.blob')
+ }
+ self.assertGreater(len(source_blob_paths), 0)
+
+ picture_field_id = next(
+ field.id for field in source_table.table_schema.fields if
field.name == 'picture'
+ )
+ view_values = [
+ BlobViewStruct('test_db.blob_view_guard_source', picture_field_id,
0).serialize(),
+ BlobViewStruct('test_db.blob_view_guard_source', picture_field_id,
1).serialize(),
+ ]
+
+ target_schema = pa.schema([
+ ('id', pa.int32()),
+ ('picture', pa.large_binary()),
+ ])
+ target = Schema.from_pyarrow_schema(
+ target_schema,
+ options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'blob-view-field': 'picture',
+ }
+ )
+ self.catalog.create_table('test_db.blob_view_guard_target', target,
False)
+ target_table = self.catalog.get_table('test_db.blob_view_guard_target')
+
+ target_write_builder = target_table.new_batch_write_builder()
+ target_writer = target_write_builder.new_write()
+ target_writer.write_arrow(pa.Table.from_pydict({
+ 'id': [10, 11],
+ 'picture': view_values,
+ }, schema=target_schema))
+ target_commit_messages = target_writer.prepare_commit()
+ target_write_builder.new_commit().commit(target_commit_messages)
+ target_writer.close()
+
+ all_target_files = [f for msg in target_commit_messages for f in
msg.new_files]
+ self.assertFalse(
+ any(f.file_name.endswith('.blob') for f in all_target_files),
+ "Blob view fields should be stored inline without writing new blob
files",
+ )
+
+ original_file_io = target_table.file_io
+ guarded_file_io = GuardedFileIO(original_file_io, source_blob_paths)
+ target_table.file_io = guarded_file_io
+ try:
+ read_builder = target_table.new_read_builder()
+ result = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits()
+ ).sort_by('id')
+ finally:
+ target_table.file_io = original_file_io
+
+ self.assertEqual(result.column('picture').to_pylist(), payloads)
+ self.assertEqual(guarded_file_io.forbidden_reads, [])
+
def test_blob_view_resolve_disabled_preserves_references(self):
from pypaimon import Schema
from pypaimon.common.options.core_options import CoreOptions
diff --git a/paimon-python/pypaimon/utils/blob_view_lookup.py
b/paimon-python/pypaimon/utils/blob_view_lookup.py
index 1acff6bcc8..2e6fd1e8be 100644
--- a/paimon-python/pypaimon/utils/blob_view_lookup.py
+++ b/paimon-python/pypaimon/utils/blob_view_lookup.py
@@ -19,8 +19,9 @@ from concurrent.futures import ThreadPoolExecutor,
as_completed
from typing import Dict, List, Tuple, Set
from pypaimon.common.identifier import Identifier
+from pypaimon.common.uri_reader import FileUriReader, UriReader
from pypaimon.common.options.core_options import CoreOptions
-from pypaimon.table.row.blob import BlobDescriptor, BlobViewStruct
+from pypaimon.table.row.blob import Blob, BlobDescriptor, BlobViewStruct
from pypaimon.table.special_fields import SpecialFields
from pypaimon.utils.range import Range
@@ -58,6 +59,7 @@ class BlobViewLookup:
def __init__(self, table):
self._table = table
self._descriptor_cache: Dict[BlobViewStruct, BlobDescriptor] = {}
+ self._uri_reader_cache: Dict[str, UriReader] = {}
self._null_value_cache: Set[BlobViewStruct] = set()
def preload(self, view_structs: List[BlobViewStruct]):
@@ -112,6 +114,31 @@ class BlobViewLookup:
)
return descriptor
+ def resolve_blob(self, view_struct: BlobViewStruct) -> Blob:
+ descriptor = self.resolve_descriptor(view_struct)
+ uri_reader = self.resolve_uri_reader(view_struct)
+ return Blob.from_descriptor(uri_reader, descriptor)
+
+ def resolve_file_io(self, view_struct: BlobViewStruct):
+ uri_reader = self.resolve_uri_reader(view_struct)
+ if not isinstance(uri_reader, FileUriReader):
+ raise ValueError(
+ "Cannot resolve BlobViewStruct {} with parallel blob reads
because "
+ "upstream table {} does not use a file-backed
UriReader.".format(
+ view_struct, view_struct.identifier.get_full_name())
+ )
+ return uri_reader._file_io
+
+ def resolve_uri_reader(self, view_struct: BlobViewStruct) -> UriReader:
+ table_key = view_struct.identifier.get_full_name()
+ uri_reader = self._uri_reader_cache.get(table_key)
+ if uri_reader is None:
+ raise ValueError(
+ "Cannot resolve BlobViewStruct {} because upstream table {} "
+ "was not loaded.".format(view_struct, table_key)
+ )
+ return uri_reader
+
def resolve_to_null(self, view_struct: BlobViewStruct) -> bool:
if view_struct in self._null_value_cache:
return True
@@ -135,6 +162,9 @@ class BlobViewLookup:
def _create_table_read_plan(self, table_refs: TableReferences) ->
TableReadPlan:
upstream_table = self._load_table(table_refs.identifier)
+ self._uri_reader_cache[table_refs.identifier.get_full_name()] = (
+ UriReader.from_file(upstream_table.file_io)
+ )
fields: List = []
for field_id in table_refs.references_by_field:
