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 16767ea96e [python] Validate upstream table row-tracking for BlobView
(#10647)
16767ea96e is described below
commit 16767ea96e0c714ab34e47bbac702305478b7a72
Author: zhuanshenbsj1 <[email protected]>
AuthorDate: Tue Oct 20 14:50:35 2026 +0800
[python] Validate upstream table row-tracking for BlobView (#10647)
---
paimon-python/pypaimon/table/row/blob.py | 85 +++++++++++++++++++++---
paimon-python/pypaimon/tests/blob_table_test.py | 79 ++++++++++++++++++++++
paimon-python/pypaimon/tests/blob_test.py | 61 +++++++++++++++++
paimon-python/pypaimon/utils/blob_view_lookup.py | 5 ++
4 files changed, 220 insertions(+), 10 deletions(-)
diff --git a/paimon-python/pypaimon/table/row/blob.py
b/paimon-python/pypaimon/table/row/blob.py
index 769fdfb717..b7360a9da4 100644
--- a/paimon-python/pypaimon/table/row/blob.py
+++ b/paimon-python/pypaimon/table/row/blob.py
@@ -427,12 +427,51 @@ class BlobViewStruct:
return upstream
raise TypeError("Expected Table, TableSchema, RowType, or sequence of
DataField.")
+ @staticmethod
+ def _extract_options(upstream):
+ if hasattr(upstream, 'options') and upstream.options is not None:
+ return upstream.options() if callable(upstream.options) else
upstream.options
+ if hasattr(upstream, 'table_schema') and getattr(
+ upstream.table_schema, 'options', None) is not None:
+ return upstream.table_schema.options
+ return None
+
+ @classmethod
+ def validate_upstream_row_tracking(
+ cls,
+ identifier: Union[Identifier, str],
+ upstream_options,
+ ) -> None:
+ from pypaimon.common.options.core_options import CoreOptions
+ from pypaimon.common.options.options import Options
+
+ identifier = cls._coerce_identifier(identifier)
+ if isinstance(upstream_options, CoreOptions):
+ core_options = upstream_options
+ elif isinstance(upstream_options, Options):
+ core_options = CoreOptions(upstream_options)
+ elif isinstance(upstream_options, dict):
+ core_options = CoreOptions.from_dict(upstream_options)
+ elif upstream_options is None:
+ core_options = CoreOptions.from_dict({})
+ else:
+ raise TypeError(
+ "Expected upstream_options to be dict, Options, or
CoreOptions."
+ )
+ if not core_options.row_tracking_enabled():
+ raise ValueError(
+ f"Upstream table {identifier.get_full_name()} "
+ f"referenced by blob view must enable "
+ f"'{CoreOptions.ROW_TRACKING_ENABLED.key()}'."
+ )
+
@classmethod
def resolve_blob_field_id(
cls,
identifier: Union[Identifier, str],
upstream,
field_name: str,
+ upstream_options=None,
) -> int:
from pypaimon.schema.data_types import is_blob_type
@@ -447,12 +486,18 @@ class BlobViewStruct:
break
if matched_field is None:
raise ValueError(
- f"Cannot find blob field {field_name} in upstream table
{identifier.get_full_name()}."
+ f"Cannot find blob field {field_name} in upstream table "
+ f"{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."
+ f"Field {field_name} in upstream table "
+ f"{identifier.get_full_name()} is not a BLOB field."
)
+ if upstream_options is None:
+ upstream_options = cls._extract_options(upstream)
+ if upstream_options is not None:
+ cls.validate_upstream_row_tracking(identifier, upstream_options)
return matched_field.id
@classmethod
@@ -461,18 +506,37 @@ class BlobViewStruct:
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."
+ "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)
+ upstream_options = cls._extract_options(upstream_table)
+ if upstream_options is None:
+ upstream_options = {}
+ field_id = cls.resolve_blob_field_id(
+ identifier, upstream_table, field_name,
+ upstream_options=upstream_options
+ )
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)
+ field_id = cls.resolve_blob_field_id(
+ identifier, upstream, field_name)
+ return cls(identifier, field_id, row_id)
+ if len(args) == 5:
+ identifier, upstream, upstream_options, field_name, row_id = args
+ if upstream_options is None:
+ upstream_options = {}
+ field_id = cls.resolve_blob_field_id(
+ identifier, upstream, field_name,
+ upstream_options=upstream_options
+ )
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)."
+ "BlobViewStruct.of expects (upstream_table, field_name, row_id), "
+ "(identifier, upstream_schema, field_name, row_id), or "
+ "(identifier, upstream_schema, upstream_options, field_name, "
+ "row_id)."
)
@property
@@ -749,12 +813,13 @@ class Blob(ABC):
return BlobView(BlobViewStruct(target, field, row_id))
if isinstance(field, str):
return BlobView(BlobViewStruct.of(target, field, row_id))
- if len(args) == 4:
+ if len(args) in (4, 5):
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)."
+ "(upstream_table, field_name, row_id), "
+ "(identifier, upstream_schema, field_name, row_id), or "
+ "(identifier, upstream_schema, upstream_options, field_name,
row_id)."
)
@staticmethod
diff --git a/paimon-python/pypaimon/tests/blob_table_test.py
b/paimon-python/pypaimon/tests/blob_table_test.py
index 458792633f..efb636839f 100755
--- a/paimon-python/pypaimon/tests/blob_table_test.py
+++ b/paimon-python/pypaimon/tests/blob_table_test.py
@@ -3066,6 +3066,85 @@ class DedicatedFormatWriterTest(unittest.TestCase):
writer.write_arrow(bad_data)
self.assertIn("blob-view-field", str(context.exception))
+ def test_blob_view_rejects_non_row_tracking_upstream_table(self):
+ from pypaimon import Schema
+ from pypaimon.common.identifier import Identifier
+ from pypaimon.common.json_util import JSON
+ from pypaimon.schema.table_schema import TableSchema
+ from pypaimon.table.row.blob import Blob, BlobViewStruct
+
+ # 1. Persist schema-0 directly to simulate a primary-key BLOB table
created by Java/Spark
+ # in the shared catalog (pypaimon's create_table currently disallows
primary_keys + BLOB).
+ source_schema = pa.schema([
+ ('id', pa.int32()),
+ ('picture', pa.large_binary()),
+ ])
+ source_identifier =
Identifier.from_string('test_db.blob_view_non_rt_source')
+ source_table_path = self.catalog.get_table_path(source_identifier)
+ non_rt_table_schema = TableSchema.from_schema(
+ 0,
+ Schema.from_pyarrow_schema(
+ source_schema,
+ primary_keys=['id'],
+ options={'bucket': '1'},
+ ),
+ )
+ self.catalog.file_io.try_to_write_atomic(
+ f"{source_table_path}/schema/schema-0",
+ JSON.to_json(non_rt_table_schema, indent=2),
+ )
+ source_table =
self.catalog.get_table('test_db.blob_view_non_rt_source')
+
+ # 2. Verify write/construction-time validation rejects the upstream
table immediately.
+ expected_msg = (
+ r"Upstream table test_db\.blob_view_non_rt_source referenced by
blob view "
+ r"must enable 'row-tracking\.enabled'\."
+ )
+ with self.assertRaisesRegex(ValueError, expected_msg):
+ BlobViewStruct.of(source_table, 'picture', 0)
+
+ with self.assertRaisesRegex(ValueError, expected_msg):
+ Blob.from_view(source_table, 'picture', 0)
+
+ # 3. Simulate a caller bypassing BlobViewStruct.of by serializing a
raw BlobViewStruct
+ # with integer field_id directly, and verify BlobViewLookup rejects it
on read.
+ 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_non_rt_target', target,
False)
+ target_table =
self.catalog.get_table('test_db.blob_view_non_rt_target')
+
+ picture_field_id = next(
+ field.id for field in source_table.table_schema.fields if
field.name == 'picture'
+ )
+ raw_view_bytes = BlobViewStruct(
+ 'test_db.blob_view_non_rt_source', picture_field_id, 0
+ ).serialize()
+
+ 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],
+ 'picture': [raw_view_bytes],
+ }, schema=target_schema))
+ target_commit_messages = target_writer.prepare_commit()
+ target_write_builder.new_commit().commit(target_commit_messages)
+ target_writer.close()
+
+ with self.assertRaisesRegex(ValueError, expected_msg):
+ target_table.new_read_builder().new_read().to_arrow(
+ target_table.new_read_builder().new_scan().plan().splits()
+ )
+
def test_blob_inline_fields_reject_overlap_and_unknown_fields(self):
from pypaimon import Schema
diff --git a/paimon-python/pypaimon/tests/blob_test.py
b/paimon-python/pypaimon/tests/blob_test.py
index 3f4f2ebd44..77e0142262 100644
--- a/paimon-python/pypaimon/tests/blob_test.py
+++ b/paimon-python/pypaimon/tests/blob_test.py
@@ -393,6 +393,7 @@ class BlobTest(unittest.TestCase):
fake_table = SimpleNamespace(
identifier=Identifier.from_string("test_db.source_table"),
fields=fields,
+ options={"row-tracking.enabled": "true"},
)
expected = BlobViewStruct("test_db.source_table", 7, 42)
@@ -429,6 +430,66 @@ class BlobTest(unittest.TestCase):
):
BlobViewStruct.of(fake_table, "id", 42)
+ def test_validate_upstream_row_tracking(self):
+ from types import SimpleNamespace
+ from pypaimon.common.identifier import Identifier
+ from pypaimon.schema.data_types import AtomicType, DataField, RowType
+
+ identifier = Identifier.from_string("default.source")
+ fields = [
+ DataField(0, "id", AtomicType("INT")),
+ DataField(7, "picture", AtomicType("BLOB")),
+ ]
+ row_type = RowType(True, fields)
+ row_tracking_options = {"row-tracking.enabled": "true"}
+
+ view_struct = BlobViewStruct.of(
+ identifier, row_type, row_tracking_options, "picture", 5
+ )
+ self.assertEqual(view_struct.field_id, 7)
+ self.assertEqual(
+ Blob.from_view(identifier, row_type, row_tracking_options,
"picture", 5),
+ Blob.from_view(view_struct),
+ )
+
+ with self.assertRaisesRegex(
+ ValueError,
+ r"Upstream table default\.source referenced by blob view must
enable 'row-tracking\.enabled'\.",
+ ):
+ BlobViewStruct.of(identifier, row_type, {}, "picture", 5)
+
+ non_row_tracking_table = SimpleNamespace(
+ identifier=identifier,
+ fields=fields,
+ options={},
+ )
+ with self.assertRaisesRegex(
+ ValueError,
+ r"Upstream table default\.source referenced by blob view must
enable 'row-tracking\.enabled'\.",
+ ):
+ BlobViewStruct.of(non_row_tracking_table, "picture", 5)
+
+ callable_options_table = SimpleNamespace(
+ identifier=identifier,
+ fields=fields,
+ options=lambda: {},
+ )
+ with self.assertRaisesRegex(
+ ValueError,
+ r"Upstream table default\.source referenced by blob view must
enable 'row-tracking\.enabled'\.",
+ ):
+ BlobViewStruct.of(callable_options_table, "picture", 5)
+
+ schema_only_table = SimpleNamespace(
+ identifier=identifier,
+ table_schema=SimpleNamespace(fields=fields, options={}),
+ )
+ with self.assertRaisesRegex(
+ ValueError,
+ r"Upstream table default\.source referenced by blob view must
enable 'row-tracking\.enabled'\.",
+ ):
+ BlobViewStruct.of(schema_only_table, "picture", 5)
+
def test_blob_fallback_batch_reader_respects_batch_size(self):
created_readers = []
diff --git a/paimon-python/pypaimon/utils/blob_view_lookup.py
b/paimon-python/pypaimon/utils/blob_view_lookup.py
index e24eec5a67..89a7d5b6a9 100644
--- a/paimon-python/pypaimon/utils/blob_view_lookup.py
+++ b/paimon-python/pypaimon/utils/blob_view_lookup.py
@@ -186,6 +186,11 @@ class BlobViewLookup:
for field_id in table_refs.references_by_field:
fields.append(self._field_by_id(upstream_table, field_id))
+ BlobViewStruct.validate_upstream_row_tracking(
+ table_refs.identifier,
+ BlobViewStruct._extract_options(upstream_table),
+ )
+
read_fields = SpecialFields.row_type_with_row_id(fields)
return TableReadPlan(
table_refs.identifier, upstream_table, read_fields,
