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 4f289264bf [python] Support constructing BlobView and BlobViewStruct
by field name (#10608)
4f289264bf is described below
commit 4f289264bf32393d0b09c5b54e7cc9a059dc5ff6
Author: zhuanshenbsj1 <[email protected]>
AuthorDate: Sun Oct 18 16:51:33 2026 +0800
[python] Support constructing BlobView and BlobViewStruct by field name
(#10608)
---
paimon-python/pypaimon/table/row/blob.py | 94 +++++++++++++++++++++++--
paimon-python/pypaimon/tests/blob_table_test.py | 38 +++-------
paimon-python/pypaimon/tests/blob_test.py | 87 +++++++++++++++++++++++
3 files changed, 185 insertions(+), 34 deletions(-)
diff --git a/paimon-python/pypaimon/table/row/blob.py
b/paimon-python/pypaimon/table/row/blob.py
index 317763ab12..769fdfb717 100644
--- a/paimon-python/pypaimon/table/row/blob.py
+++ b/paimon-python/pypaimon/table/row/blob.py
@@ -405,13 +405,75 @@ class BlobViewStruct:
MAGIC = 0x424C4F4256494557 # "BLOBVIEW"
def __init__(self, identifier: Union[Identifier, str], field_id: int,
row_id: int):
+ self._identifier = self._coerce_identifier(identifier)
+ self._field_id = field_id
+ self._row_id = row_id
+
+ @staticmethod
+ def _coerce_identifier(identifier: Union[Identifier, str]) -> Identifier:
if isinstance(identifier, str):
- identifier = Identifier.from_string(identifier)
+ return Identifier.from_string(identifier)
if not isinstance(identifier, Identifier):
raise TypeError("BlobViewStruct identifier must be Identifier or
str.")
- self._identifier = identifier
- self._field_id = field_id
- self._row_id = row_id
+ return identifier
+
+ @staticmethod
+ def _extract_fields(upstream):
+ if hasattr(upstream, 'fields') and upstream.fields is not None:
+ return upstream.fields
+ if hasattr(upstream, 'table_schema') and
getattr(upstream.table_schema, 'fields', None) is not None:
+ return upstream.table_schema.fields
+ if isinstance(upstream, (list, tuple)):
+ return upstream
+ raise TypeError("Expected Table, TableSchema, RowType, or sequence of
DataField.")
+
+ @classmethod
+ def resolve_blob_field_id(
+ cls,
+ identifier: Union[Identifier, str],
+ upstream,
+ field_name: str,
+ ) -> int:
+ from pypaimon.schema.data_types import is_blob_type
+
+ identifier = cls._coerce_identifier(identifier)
+ if not isinstance(field_name, str):
+ raise TypeError("BlobViewStruct field_name must be a str.")
+ fields = cls._extract_fields(upstream)
+ matched_field = None
+ for field in fields:
+ if field.name == field_name:
+ matched_field = field
+ break
+ if matched_field is None:
+ raise ValueError(
+ f"Cannot find blob field {field_name} in upstream table
{identifier.get_full_name()}."
+ )
+ if not is_blob_type(matched_field.type):
+ raise ValueError(
+ f"Field {field_name} in upstream table
{identifier.get_full_name()} is not a BLOB field."
+ )
+ return matched_field.id
+
+ @classmethod
+ def of(cls, *args) -> 'BlobViewStruct':
+ if len(args) == 3:
+ upstream_table, field_name, row_id = args
+ if not hasattr(upstream_table, 'identifier'):
+ raise TypeError(
+ "3-argument BlobViewStruct.of expects an upstream Table
with an identifier."
+ )
+ identifier = upstream_table.identifier
+ field_id = cls.resolve_blob_field_id(identifier, upstream_table,
field_name)
+ return cls(identifier, field_id, row_id)
+ if len(args) == 4:
+ identifier, upstream, field_name, row_id = args
+ field_id = cls.resolve_blob_field_id(identifier, upstream,
field_name)
+ return cls(identifier, field_id, row_id)
+ raise TypeError(
+ "BlobViewStruct.of expects (upstream_table, field_name, row_id) or
"
+ "(identifier, upstream_schema, field_name, row_id)."
+ )
@property
def identifier(self) -> Identifier:
@@ -672,8 +734,28 @@ class Blob(ABC):
descriptor, file_io=file_io, uri_reader_factory=uri_reader_factory)
@staticmethod
- def from_view(view_struct: BlobViewStruct) -> 'BlobView':
- return BlobView(view_struct)
+ def from_view(*args, view_struct: Optional[BlobViewStruct] = None) ->
'BlobView':
+ # Case 1: Keyword argument form (backward compatible)
+ if view_struct is not None:
+ if len(args) > 0:
+ raise ValueError("Cannot provide both view_struct keyword
argument and positional arguments")
+ return BlobView(view_struct)
+ # Case 2: Positional argument forms
+ if len(args) == 1 and isinstance(args[0], BlobViewStruct):
+ return BlobView(args[0])
+ if len(args) == 3:
+ target, field, row_id = args
+ if isinstance(field, int) and not isinstance(field, bool):
+ return BlobView(BlobViewStruct(target, field, row_id))
+ if isinstance(field, str):
+ return BlobView(BlobViewStruct.of(target, field, row_id))
+ if len(args) == 4:
+ return BlobView(BlobViewStruct.of(*args))
+ raise TypeError(
+ "Blob.from_view expects (BlobViewStruct), (identifier, field_id,
row_id), "
+ "(upstream_table, field_name, row_id), or "
+ "(identifier, upstream_schema, field_name, row_id)."
+ )
@staticmethod
def from_bytes(
diff --git a/paimon-python/pypaimon/tests/blob_table_test.py
b/paimon-python/pypaimon/tests/blob_table_test.py
index bf6612bf15..458792633f 100755
--- a/paimon-python/pypaimon/tests/blob_table_test.py
+++ b/paimon-python/pypaimon/tests/blob_table_test.py
@@ -2722,12 +2722,9 @@ class DedicatedFormatWriterTest(unittest.TestCase):
write_builder.new_commit().commit(commit_messages)
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_source', picture_field_id,
0).serialize(),
- BlobViewStruct('test_db.blob_view_source', picture_field_id,
1).serialize(),
+ BlobViewStruct.of(source_table, 'picture', 0).serialize(),
+ BlobViewStruct.of(source_table, 'picture', 1).serialize(),
]
target_schema = pa.schema([
@@ -2842,12 +2839,9 @@ class DedicatedFormatWriterTest(unittest.TestCase):
}
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(),
+ BlobViewStruct.of(source_table, 'picture', 0).serialize(),
+ BlobViewStruct.of(source_table, 'picture', 1).serialize(),
]
target_schema = pa.schema([
@@ -2925,12 +2919,9 @@ class DedicatedFormatWriterTest(unittest.TestCase):
write_builder.new_commit().commit(commit_messages)
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_resolve_source',
picture_field_id, 0).serialize(),
- BlobViewStruct('test_db.blob_view_resolve_source',
picture_field_id, 1).serialize(),
+ BlobViewStruct.of(source_table, 'picture', 0).serialize(),
+ BlobViewStruct.of(source_table, 'picture', 1).serialize(),
]
target_schema = pa.schema([
@@ -3009,12 +3000,9 @@ class DedicatedFormatWriterTest(unittest.TestCase):
write_builder.new_commit().commit(commit_messages)
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_null_source', picture_field_id,
0).serialize(),
- BlobViewStruct('test_db.blob_view_null_source', picture_field_id,
1).serialize(),
+ BlobViewStruct.of(source_table, 'picture', 0).serialize(),
+ BlobViewStruct.of(source_table, 'picture', 1).serialize(),
]
target_schema = pa.schema([
@@ -3146,11 +3134,8 @@ class DedicatedFormatWriterTest(unittest.TestCase):
write_builder.new_commit().commit(commit_messages)
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_limit_source', picture_field_id,
i).serialize()
+ BlobViewStruct.of(source_table, 'picture', i).serialize()
for i in range(num_rows)
]
@@ -3230,11 +3215,8 @@ class DedicatedFormatWriterTest(unittest.TestCase):
write_builder.new_commit().commit(commit_messages)
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_prescan_count_source',
picture_field_id, i).serialize()
+ BlobViewStruct.of(source_table, 'picture', i).serialize()
for i in range(num_rows)
]
diff --git a/paimon-python/pypaimon/tests/blob_test.py
b/paimon-python/pypaimon/tests/blob_test.py
index 0043fd9833..3f4f2ebd44 100644
--- a/paimon-python/pypaimon/tests/blob_test.py
+++ b/paimon-python/pypaimon/tests/blob_test.py
@@ -380,6 +380,55 @@ class BlobTest(unittest.TestCase):
self.assertFalse(blob.is_resolved())
self.assertEqual(blob.view_struct, view_struct)
+ def test_blob_view_struct_construct_by_field_name(self):
+ from types import SimpleNamespace
+ from pypaimon.common.identifier import Identifier
+ from pypaimon.schema.data_types import AtomicType, DataField, RowType
+
+ fields = [
+ DataField(0, "id", AtomicType("INT")),
+ DataField(7, "picture", AtomicType("BLOB")),
+ ]
+ row_type = RowType(True, fields)
+ fake_table = SimpleNamespace(
+ identifier=Identifier.from_string("test_db.source_table"),
+ fields=fields,
+ )
+
+ expected = BlobViewStruct("test_db.source_table", 7, 42)
+ self.assertEqual(
+ BlobViewStruct.of("test_db.source_table", row_type, "picture", 42),
+ expected,
+ )
+ self.assertEqual(
+ BlobViewStruct.of(fake_table, "picture", 42),
+ expected,
+ )
+ self.assertEqual(
+ Blob.from_view("test_db.source_table", 7, 42),
+ Blob.from_view(expected),
+ )
+ self.assertEqual(
+ Blob.from_view("test_db.source_table", row_type, "picture", 42),
+ Blob.from_view(expected),
+ )
+ self.assertEqual(
+ Blob.from_view(fake_table, "picture", 42),
+ Blob.from_view(expected),
+ )
+
+ with self.assertRaisesRegex(
+ ValueError,
+ r"Cannot find blob field missing in upstream table
test_db\.source_table\.",
+ ):
+ BlobViewStruct.of(fake_table, "missing", 42)
+
+ with self.assertRaisesRegex(
+ ValueError,
+ r"Field id in upstream table test_db\.source_table is not a BLOB
field\.",
+ ):
+ BlobViewStruct.of(fake_table, "id", 42)
+
def test_blob_fallback_batch_reader_respects_batch_size(self):
created_readers = []
@@ -6757,5 +6806,43 @@ class OffsetInputStreamTest(unittest.TestCase):
os.remove(tmp_path)
+class BlobFromViewBackwardCompatibilityTest(unittest.TestCase):
+ """Test backward compatibility of Blob.from_view keyword argument."""
+
+ def test_from_view_keyword_argument_backward_compatible(self):
+ """Test that existing keyword-argument calls still work."""
+ from pypaimon.common.identifier import Identifier
+ from pypaimon.table.row.blob import Blob, BlobViewStruct
+
+ view_struct =
BlobViewStruct(Identifier.from_string("default.test_compat"), 1, 42)
+ blob_view = Blob.from_view(view_struct=view_struct)
+
+ self.assertEqual(blob_view.view_struct.identifier.get_full_name(),
"default.test_compat")
+ self.assertEqual(blob_view.view_struct.field_id, 1)
+ self.assertEqual(blob_view.view_struct.row_id, 42)
+
+ def test_from_view_positional_argument_still_works(self):
+ """Test that positional argument form still works."""
+ from pypaimon.common.identifier import Identifier
+ from pypaimon.table.row.blob import Blob, BlobViewStruct
+
+ view_struct =
BlobViewStruct(Identifier.from_string("default.test_pos"), 2, 100)
+ blob_view = Blob.from_view(view_struct)
+
+ self.assertEqual(blob_view.view_struct.identifier.get_full_name(),
"default.test_pos")
+ self.assertEqual(blob_view.view_struct.field_id, 2)
+ self.assertEqual(blob_view.view_struct.row_id, 100)
+
+ def test_from_view_cannot_mix_keyword_and_positional(self):
+ """Test that mixing keyword and positional is rejected."""
+ from pypaimon.common.identifier import Identifier
+ from pypaimon.table.row.blob import Blob, BlobViewStruct
+
+ view_struct =
BlobViewStruct(Identifier.from_string("default.test_mix"), 3, 200)
+ with self.assertRaises(ValueError) as cm:
+ Blob.from_view(view_struct, view_struct=view_struct)
+ self.assertIn("Cannot provide both", str(cm.exception))
+
+
if __name__ == '__main__':
unittest.main()
