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 c75ebb8fe1 [python] Keep BlobViewStruct bytes when blob-as-descriptor
is true. (#10057)
c75ebb8fe1 is described below
commit c75ebb8fe189a9910a029a927d28a98e0880d6f4
Author: Wenchao Wu <[email protected]>
AuthorDate: Tue Sep 22 10:45:01 2026 +0800
[python] Keep BlobViewStruct bytes when blob-as-descriptor is true. (#10057)
---
.../read/reader/blob_descriptor_convert_reader.py | 17 +-
paimon-python/pypaimon/read/table_read.py | 53 +++-
paimon-python/pypaimon/tests/blob_table_test.py | 230 +++++++++++++++++
paimon-python/pypaimon/tests/blob_test.py | 285 +++++++++++++++++++++
paimon-python/pypaimon/utils/blob_view_lookup.py | 17 ++
5 files changed, 595 insertions(+), 7 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 f7bc46612f..1d671f470a 100644
--- a/paimon-python/pypaimon/read/reader/blob_descriptor_convert_reader.py
+++ b/paimon-python/pypaimon/read/reader/blob_descriptor_convert_reader.py
@@ -31,9 +31,12 @@ class BlobInlineConvertReader(RecordBatchReader):
Processing is split into two clear stages:
Stage 1 (BlobView resolution): If view fields exist, use a lightweight
prescan reader (only projecting view columns) to collect
- BlobViewStructs and bulk-preload their descriptors, then replace
- view field values with descriptor bytes so Stage 2 can
- materialize payloads with the originating table FileIO.
+ BlobViewStructs and bulk-preload their descriptors. When
+ blob-as-descriptor=false, replace view field values with
+ descriptor bytes so Stage 2 can materialize payloads with the
+ originating table FileIO. When blob-as-descriptor=true, leave
+ BlobViewStruct bytes in place so get_blob() keeps that
+ association; Arrow output serializes descriptors later.
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
@@ -81,9 +84,13 @@ class BlobInlineConvertReader(RecordBatchReader):
batch = self._inner.read_arrow_batch()
if batch is None:
return None
- # Resolve view fields using the preloaded lookup.
+ # Resolve view fields using the preloaded lookup. When
+ # blob-as-descriptor=true, leave BlobViewStruct bytes in place so
+ # get_blob() keeps the originating table's FileIO. Arrow paths
+ # serialize descriptors later.
view_blobs = {}
- if self._view_fields and self._blob_view_lookup is not None:
+ if (self._view_fields and self._blob_view_lookup is not None
+ and not self._blob_as_descriptor):
batch, view_blobs = self._resolve_view_fields(batch,
self._blob_view_lookup)
# Resolve BlobDescriptor -> real bytes (if blob-as-descriptor=false)
return self._resolve_descriptor_fields(batch, view_blobs)
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index 1b3a4585d3..fb2926a467 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -168,6 +168,16 @@ class TableRead:
self._predicate_extra_fields =
self._predicate_fields_outside_read_type()
self._scan_read_type = self.read_type + self._predicate_extra_fields
self._output_column_names = [f.name for f in self.read_type]
+ # Output positions in ``read_type``, not ``_scan_read_type``. Predicate
+ # extras are appended for the inner scan and projected away before
+ # ``_serialize_blob_view_row_tuple``; prepending them would make these
+ # indices rewrite the wrong columns.
+ blob_view_fields = self.table.options.blob_view_fields()
+ self._blob_view_output_indices = tuple(
+ i for i, field in enumerate(self.read_type)
+ if field.name in blob_view_fields
+ )
+ self._blob_as_descriptor =
bool(self.table.options.blob_as_descriptor())
self._deferred_blob_fields = (
deferred_blob_field_names(
self.table,
@@ -813,6 +823,8 @@ class TableRead:
continue
if remaining is not None and batch.num_rows >
remaining:
batch = batch.slice(0, remaining)
+ batch = self._serialize_blob_views_as_descriptors(
+ batch, reader)
batch = self._project_batch_to_output(batch)
if self.include_row_kind:
if "_row_kind" not in batch.schema.names:
@@ -833,7 +845,8 @@ class TableRead:
for row in iter(row_iterator.next, None):
if not isinstance(row, OffsetRow):
raise TypeError(f"Expected OffsetRow, but got
{type(row).__name__}")
- row_tuple_chunk.append(row.row_tuple[row.offset:
row.offset + row.arity])
+ row_tuple_chunk.append(
+ self._serialize_blob_view_row_tuple(row,
reader))
if self.include_row_kind:
row_kind_chunk.append(row.get_row_kind().to_string())
@@ -1045,6 +1058,8 @@ class TableRead:
break
if allowed < batch.num_rows:
batch = batch.slice(0, allowed)
+ batch = self._serialize_blob_views_as_descriptors(
+ batch, reader)
batch = self._project_batch_to_output(batch)
if self.include_row_kind:
if "_row_kind" not in batch.schema.names:
@@ -1068,7 +1083,7 @@ class TableRead:
stop = True
break
row_tuple_chunk.append(
- row.row_tuple[row.offset: row.offset + row.arity])
+ self._serialize_blob_view_row_tuple(row, reader))
if self.include_row_kind:
row_kind_chunk.append(row.get_row_kind().to_string())
@@ -1530,6 +1545,40 @@ class TableRead:
limit=effective_limit,
)
+ def _serialize_blob_views_as_descriptors(self, batch: pyarrow.RecordBatch,
reader) -> pyarrow.RecordBatch:
+ if not self._blob_as_descriptor or not self._blob_view_output_indices:
+ return batch
+ lookup = getattr(reader, 'blob_view_lookup', None)
+ if lookup is None:
+ return batch
+ for index in self._blob_view_output_indices:
+ field_name = self.read_type[index].name
+ if field_name not in batch.schema.names:
+ continue
+ converted = [
+ lookup.serialize_view_field_value(value)
+ for value in batch.column(field_name).to_pylist()
+ ]
+ 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, type=pyarrow.large_binary()),
+ )
+ return batch
+
+ def _serialize_blob_view_row_tuple(self, row: OffsetRow, reader) -> tuple:
+ values = row.row_tuple[row.offset: row.offset + row.arity]
+ if not self._blob_as_descriptor or not self._blob_view_output_indices:
+ return values
+ lookup = getattr(reader, 'blob_view_lookup', None)
+ if lookup is None:
+ return values
+ converted = list(values)
+ for index in self._blob_view_output_indices:
+ converted[index] =
lookup.serialize_view_field_value(converted[index])
+ return tuple(converted)
+
def _project_batch_to_output(self, batch: pyarrow.RecordBatch) ->
pyarrow.RecordBatch:
if not self._needs_output_projection():
return batch
diff --git a/paimon-python/pypaimon/tests/blob_table_test.py
b/paimon-python/pypaimon/tests/blob_table_test.py
index 9adb80e1ac..6b4133242c 100755
--- a/paimon-python/pypaimon/tests/blob_table_test.py
+++ b/paimon-python/pypaimon/tests/blob_table_test.py
@@ -5593,6 +5593,236 @@ class DedicatedFormatWriterTest(unittest.TestCase):
self.assertEqual(
result.column('picture').to_pylist(), [b'raw-payload-9'])
+ def test_blob_view_as_descriptor_get_blob_uses_upstream_file_io(self):
+ """blob-as-descriptor=true still must read source .blob with source
FileIO.
+
+ Stage 1 retains each BlobViewStruct so get_blob().to_data() can select
+ the source table token instead of falling back to the target table
token.
+ """
+ from pypaimon import Schema
+ from pypaimon.common.options.core_options import CoreOptions
+ 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_desc_guard_source', source, False)
+ source_table = self.catalog.get_table(
+ 'test_db.blob_view_desc_guard_source')
+ payloads = [b'desc-guard-source-0', b'desc-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_desc_guard_source', picture_field_id, 0
+ ).serialize(),
+ BlobViewStruct(
+ 'test_db.blob_view_desc_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_desc_guard_target', target, False)
+ target_table = self.catalog.get_table(
+ 'test_db.blob_view_desc_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_write_builder.new_commit().commit(
+ target_writer.prepare_commit())
+ target_writer.close()
+
+ descriptor_table = target_table.copy({
+ CoreOptions.BLOB_AS_DESCRIPTOR.key(): 'true'
+ })
+ original_file_io = descriptor_table.file_io
+ guarded_file_io = GuardedFileIO(original_file_io, source_blob_paths)
+ descriptor_table.file_io = guarded_file_io
+ try:
+ read_builder = descriptor_table.new_read_builder()
+ data = []
+ for row in read_builder.new_read().to_iterator(
+ read_builder.new_scan().plan().splits()):
+ data.append(row.get_blob(1).to_data())
+ finally:
+ descriptor_table.file_io = original_file_io
+
+ self.assertEqual(sorted(data), sorted(payloads))
+ self.assertEqual(guarded_file_io.forbidden_reads, [])
+
+ def
test_blob_view_as_descriptor_projection_with_predicate_extra_field(self):
+ """Arrow serialize uses read_type positions, not the widened scan type.
+
+ Projection puts picture first and drops grp; the filter still needs
+ grp internally. _blob_view_output_indices must be (0,) — picture in
+ the output schema — so a later change that indexes _scan_read_type
+ or the table field order cannot silently rewrite the wrong column.
+ """
+ from pypaimon import Schema
+ from pypaimon.common.options.core_options import CoreOptions
+ from pypaimon.table.row.blob import BlobDescriptor, BlobViewStruct
+
+ 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_proj_pred_source', source, False)
+ source_table = self.catalog.get_table(
+ 'test_db.blob_view_proj_pred_source')
+ payloads = [b'proj-pred-source-1', b'proj-pred-source-2']
+
+ 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))
+ write_builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+
+ 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_proj_pred_source', picture_field_id, 0
+ ).serialize(),
+ BlobViewStruct(
+ 'test_db.blob_view_proj_pred_source', picture_field_id, 1
+ ).serialize(),
+ ]
+
+ target_schema = pa.schema([
+ ('id', pa.int32()),
+ ('grp', 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_proj_pred_target', target, False)
+ target_table = self.catalog.get_table(
+ 'test_db.blob_view_proj_pred_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],
+ 'grp': [1, 2],
+ 'picture': view_values,
+ }, schema=target_schema))
+ target_write_builder.new_commit().commit(
+ target_writer.prepare_commit())
+ target_writer.close()
+
+ descriptor_table = target_table.copy({
+ CoreOptions.BLOB_AS_DESCRIPTOR.key(): 'true'
+ })
+ predicate =
descriptor_table.new_read_builder().new_predicate_builder().equal(
+ 'grp', 2)
+ read_builder = descriptor_table.new_read_builder().with_projection(
+ ['picture', 'id']).with_filter(predicate)
+ table_read = read_builder.new_read()
+
+ self.assertEqual(
+ [field.name for field in table_read.read_type],
+ ['picture', 'id'])
+ self.assertEqual(
+ [field.name for field in table_read._scan_read_type],
+ ['picture', 'id', 'grp'])
+ self.assertEqual(table_read._blob_view_output_indices, (0,))
+ self.assertTrue(table_read._blob_as_descriptor)
+
+ result = table_read.to_arrow(read_builder.new_scan().plan().splits())
+ self.assertEqual(list(result.column_names), ['picture', 'id'])
+ self.assertEqual(result.num_rows, 1)
+ self.assertEqual(result.column('id').to_pylist(), [11])
+ picture_bytes = result.column('picture').to_pylist()[0]
+ self.assertTrue(
+ BlobDescriptor.is_blob_descriptor(picture_bytes),
+ "Projected view column must be descriptor bytes, not
BlobViewStruct",
+ )
+ self.assertFalse(BlobViewStruct.is_blob_view_struct(picture_bytes))
+
class GetBlobTest(unittest.TestCase):
diff --git a/paimon-python/pypaimon/tests/blob_test.py
b/paimon-python/pypaimon/tests/blob_test.py
index 736c7ddd97..dee9d21488 100644
--- a/paimon-python/pypaimon/tests/blob_test.py
+++ b/paimon-python/pypaimon/tests/blob_test.py
@@ -2867,6 +2867,291 @@ class BlobTest(unittest.TestCase):
return TokenFileIO()
+ def test_offset_row_get_blob_view_keeps_per_table_uri_reader(self):
+ from pypaimon.common.identifier import Identifier
+ from pypaimon.common.uri_reader import FileUriReader, UriReaderFactory
+ from pypaimon.table.row.offset_row import OffsetRow
+ from pypaimon.utils.blob_view_lookup import BlobViewLookup
+
+ shared_uri = "s3://shared/blob"
+ data_a = b"AAAA"
+ data_b = b"BBBB"
+ descriptor = BlobDescriptor(shared_uri, 0, len(data_a))
+ view_a = BlobViewStruct(Identifier.from_string("db.src_a"), 1, 0)
+ view_b = BlobViewStruct(Identifier.from_string("db.src_b"), 1, 0)
+
+ class TokenFileIO:
+ def __init__(self, payload):
+ self._payload = payload
+ self.opened = []
+
+ def new_input_stream(self, path):
+ self.opened.append(path)
+ return io.BytesIO(self._payload)
+
+ file_io_a = TokenFileIO(data_a)
+ file_io_b = TokenFileIO(data_b)
+ lookup = BlobViewLookup(object())
+ lookup._uri_reader_cache["db.src_a"] = FileUriReader(file_io_a)
+ lookup._uri_reader_cache["db.src_b"] = FileUriReader(file_io_b)
+ lookup._uri_reader_factory_cache["db.src_a"] = (
+ UriReaderFactory.from_file_io(file_io_a))
+ lookup._uri_reader_factory_cache["db.src_b"] = (
+ UriReaderFactory.from_file_io(file_io_b))
+ lookup._store_chunk_results(
+ {view_a: descriptor, view_b: descriptor}, set())
+
+ target_file_io = TokenFileIO(b"target-must-not-be-used")
+ row_a = OffsetRow(
+ (view_a.serialize(),), 0, 1,
+ file_io=target_file_io,
+ blob_field_indices=[0],
+ descriptor_field_indices=[0],
+ blob_view_lookup=lookup,
+ )
+ row_b = OffsetRow(
+ (view_b.serialize(),), 0, 1,
+ file_io=target_file_io,
+ blob_field_indices=[0],
+ descriptor_field_indices=[0],
+ blob_view_lookup=lookup,
+ )
+ self.assertEqual(row_a.get_blob(0).to_data(), data_a)
+ self.assertEqual(row_b.get_blob(0).to_data(), data_b)
+ self.assertEqual(file_io_a.opened, [shared_uri])
+ self.assertEqual(file_io_b.opened, [shared_uri])
+ self.assertEqual(target_file_io.opened, [])
+
+ def test_table_read_serializes_only_configured_blob_view_fields(self):
+ from types import SimpleNamespace
+
+ from pypaimon.common.identifier import Identifier
+ from pypaimon.common.options.core_options import CoreOptions
+ from pypaimon.read.table_read import TableRead
+ from pypaimon.schema.data_types import AtomicType, DataField
+ from pypaimon.table.row.offset_row import OffsetRow
+ from pypaimon.utils.blob_view_lookup import BlobViewLookup
+
+ view_struct = BlobViewStruct(Identifier.from_string("db.source"), 1, 0)
+ view_bytes = view_struct.serialize()
+ descriptor = BlobDescriptor("s3://source/blob", 0, 4)
+ lookup = BlobViewLookup(object())
+ lookup._store_chunk_results({view_struct: descriptor}, set())
+
+ table_read = TableRead.__new__(TableRead)
+ table_read.table = SimpleNamespace(options=CoreOptions(Options({
+ "blob-as-descriptor": "true",
+ "blob-view-field": "picture",
+ })))
+ table_read.read_type = [
+ DataField(0, "id", AtomicType("INT")),
+ DataField(1, "picture", AtomicType("BYTES")),
+ ]
+ table_read._blob_as_descriptor = True
+ table_read._blob_view_output_indices = (1,)
+ reader = SimpleNamespace(blob_view_lookup=lookup)
+ row = OffsetRow((view_bytes, view_bytes), 0, 2)
+
+ converted = table_read._serialize_blob_view_row_tuple(row, reader)
+
+ self.assertEqual(converted[0], view_bytes)
+ self.assertEqual(converted[1], descriptor.serialize())
+
+ table_read._blob_view_output_indices = ()
+ skipped = table_read._serialize_blob_view_row_tuple(
+ row, SimpleNamespace())
+ self.assertEqual(skipped, (view_bytes, view_bytes))
+
+ table_read._blob_view_output_indices = (1,)
+ table_read._blob_as_descriptor = False
+ not_descriptor = table_read._serialize_blob_view_row_tuple(row, reader)
+ self.assertEqual(not_descriptor, (view_bytes, view_bytes))
+
+ def test_offset_row_get_blob_v1_resolved_blob_view_field(self):
+ from pypaimon.common.options import Options
+ from pypaimon.common.options.core_options import CoreOptions
+ from pypaimon.read.reader.field_indices import
descriptor_field_indices_for_table
+ from pypaimon.schema.data_types import AtomicType, DataField
+ from pypaimon.table.row.offset_row import OffsetRow
+
+ data = b"resolved blob-view payload"
+ with tempfile.TemporaryDirectory() as tmp_dir:
+ blob_path = os.path.join(tmp_dir, "blob.bin")
+ with open(blob_path, 'wb') as f:
+ f.write(data)
+ uri = blob_path.encode('utf-8')
+ serialized_v1 = (
+ bytes([1])
+ + struct.pack('<I', len(uri))
+ + uri
+ + struct.pack('<q', 0)
+ + struct.pack('<q', len(data))
+ )
+
+ class _Table:
+ options = CoreOptions(Options({
+ "blob-as-descriptor": "true",
+ "blob-view-field": "picture",
+ }))
+
+ fields = [DataField(0, "picture", AtomicType("BYTES"))]
+ descriptor_indices = descriptor_field_indices_for_table(_Table(),
fields)
+ file_io = FileIO.get(f"file://{tmp_dir}", {})
+ row = OffsetRow(
+ (serialized_v1,), 0, 1, file_io=file_io,
+ blob_field_indices=[0],
+ descriptor_field_indices=descriptor_indices)
+ blob = row.get_blob(0)
+ self.assertIsInstance(blob, BlobRef)
+ self.assertEqual(blob.to_data(), data)
+
+ def test_offset_row_get_blob_resolves_null_blob_view(self):
+ from unittest.mock import MagicMock
+
+ from pypaimon.table.row.offset_row import OffsetRow
+ from pypaimon.table.row.blob import BlobViewStruct
+ from pypaimon.common.identifier import Identifier
+
+ view_struct = BlobViewStruct(Identifier.from_string("db.source"), 1,
42)
+ lookup = MagicMock()
+ lookup.resolve_to_null.return_value = True
+ row = OffsetRow(
+ (view_struct.serialize(),), 0, 1,
+ blob_field_indices=[0],
+ blob_view_lookup=lookup,
+ )
+ self.assertIsNone(row.get_blob(0))
+ lookup.resolve_to_null.assert_called_once()
+
+ def test_offset_row_get_blob_view_struct_without_view_field_indices(self):
+ from unittest.mock import MagicMock
+
+ from pypaimon.common.options import Options
+ from pypaimon.common.options.core_options import CoreOptions
+ from pypaimon.read.reader.field_indices import
descriptor_field_indices_for_table
+ from pypaimon.schema.data_types import AtomicType, DataField
+ from pypaimon.common.uri_reader import FileUriReader
+ from pypaimon.table.row.blob import BlobRef, BlobViewStruct
+ from pypaimon.table.row.offset_row import OffsetRow
+ from pypaimon.common.identifier import Identifier
+
+ data = b"resolved via view struct"
+ with tempfile.TemporaryDirectory() as tmp_dir:
+ blob_path = os.path.join(tmp_dir, "blob.bin")
+ with open(blob_path, 'wb') as f:
+ f.write(data)
+ descriptor = BlobDescriptor(blob_path, 0, len(data))
+
+ class _Table:
+ options = CoreOptions(Options({
+ "blob-as-descriptor": "true",
+ "blob-view-field": "picture",
+ }))
+
+ fields = [DataField(0, "picture", AtomicType("BYTES"))]
+ descriptor_indices = descriptor_field_indices_for_table(_Table(),
fields)
+ view_struct = BlobViewStruct(Identifier.from_string("db.source"),
1, 42)
+ lookup = MagicMock()
+ lookup.resolve_to_null.return_value = False
+ lookup.resolve_blob.return_value = BlobRef(
+ FileUriReader(FileIO.get(f"file://{tmp_dir}", {})), descriptor)
+
+ row = OffsetRow(
+ (view_struct.serialize(),), 0, 1,
+ blob_field_indices=[0],
+ descriptor_field_indices=descriptor_indices,
+ blob_view_lookup=lookup,
+ )
+ blob = row.get_blob(0)
+ self.assertIsInstance(blob, BlobRef)
+ self.assertEqual(blob.to_data(), data)
+ lookup.resolve_blob.assert_called_once_with(view_struct)
+
+ def
test_to_iterator_adapters_refresh_blob_view_lookup_after_first_read(self):
+ """Merge to_iterator rebuilds OffsetRow after wrap; lookup is filled on
+ the first convert read, so adapters must copy it then, not at wrap
time.
+ """
+ from unittest.mock import MagicMock
+
+ from pypaimon.common.identifier import Identifier
+ from pypaimon.common.uri_reader import FileUriReader
+ from pypaimon.read.reader.auth_masking_reader import (
+ BatchToRecordReaderAdapter, RecordReaderToBatchAdapter)
+ from pypaimon.read.reader.iface.record_iterator import RecordIterator
+ from pypaimon.read.reader.iface.record_reader import RecordReader
+ from pypaimon.read.reader.limited_record_reader import
LimitedRecordReader
+ from pypaimon.schema.data_types import AtomicType, DataField,
PyarrowFieldParser
+ from pypaimon.table.row.offset_row import OffsetRow
+
+ data = b"upstream blob via refreshed lookup"
+ with tempfile.TemporaryDirectory() as tmp_dir:
+ blob_path = os.path.join(tmp_dir, "upstream.bin")
+ with open(blob_path, 'wb') as f:
+ f.write(data)
+ view_struct = BlobViewStruct(Identifier.from_string("db.src"), 1,
0)
+ source_file_io = FileIO.get(f"file://{tmp_dir}", {})
+ target_file_io = MagicMock()
+ target_file_io.new_input_stream.side_effect = AssertionError(
+ "target FileIO must not read upstream blob")
+ lookup = MagicMock()
+ lookup.resolve_to_null.return_value = False
+ lookup.resolve_blob.return_value = BlobRef(
+ FileUriReader(source_file_io),
+ BlobDescriptor(blob_path, 0, len(data)),
+ )
+
+ class _OnceIterator(RecordIterator):
+ def __init__(self, row):
+ self._row = row
+ self._done = False
+
+ def next(self):
+ if self._done:
+ return None
+ self._done = True
+ return self._row
+
+ class _ConvertLikeReader(RecordReader):
+ def __init__(self, row):
+ self._row = row
+ self._done = False
+ self.blob_view_lookup = None
+ self.file_io = target_file_io
+ self.blob_field_indices = {0}
+ self.descriptor_field_indices = {0}
+
+ def read_batch(self):
+ if self._done:
+ return None
+ self._done = True
+ self.blob_view_lookup = lookup
+ return _OnceIterator(self._row)
+
+ def close(self):
+ pass
+
+ fields = [DataField(0, "picture", AtomicType("BLOB"))]
+ schema = PyarrowFieldParser.from_paimon_schema(fields)
+ inner = _ConvertLikeReader(OffsetRow(
+ (view_struct.serialize(),), 0, 1,
+ file_io=target_file_io,
+ blob_field_indices=[0],
+ descriptor_field_indices=[0],
+ ))
+ limited = LimitedRecordReader(inner, 10)
+ self.assertIsNone(limited.blob_view_lookup)
+ batch_reader = RecordReaderToBatchAdapter(limited, schema)
+ self.assertIsNone(batch_reader.blob_view_lookup)
+ wrapped = BatchToRecordReaderAdapter(batch_reader)
+ self.assertIsNone(wrapped.blob_view_lookup)
+ row = wrapped.read_batch().next()
+ blob = row.get_blob(0)
+ self.assertIsInstance(blob, BlobRef)
+ self.assertEqual(blob.to_data(), data)
+ lookup.resolve_blob.assert_called_once_with(view_struct)
+ target_file_io.new_input_stream.assert_not_called()
+ wrapped.close()
+
class BlobEndToEndTest(unittest.TestCase):
"""End-to-end tests for blob functionality with schema definition, file
writing, and reading."""
diff --git a/paimon-python/pypaimon/utils/blob_view_lookup.py
b/paimon-python/pypaimon/utils/blob_view_lookup.py
index 37d6df5c7e..e24eec5a67 100644
--- a/paimon-python/pypaimon/utils/blob_view_lookup.py
+++ b/paimon-python/pypaimon/utils/blob_view_lookup.py
@@ -132,6 +132,23 @@ class BlobViewLookup:
)
return uri_reader
+ def serialize_view_field_value(self, value):
+ """Replace BlobViewStruct bytes with descriptor bytes for Arrow
output."""
+ if value is None:
+ return None
+ if hasattr(value, 'as_py'):
+ value = value.as_py()
+ if isinstance(value, str):
+ value = value.encode('utf-8')
+ if isinstance(value, bytearray):
+ value = bytes(value)
+ if not (isinstance(value, bytes) and
BlobViewStruct.is_blob_view_struct(value)):
+ return value
+ view_struct = BlobViewStruct.deserialize(value)
+ if self.resolve_to_null(view_struct):
+ return None
+ return self.resolve_descriptor(view_struct).serialize()
+
def _store_chunk_results(self, descriptors, null_values):
self._descriptor_cache.update(descriptors)
self._null_value_cache.update(null_values)
