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
commit 82303b78ca1c87b32807de69ec125706a097ee7d Author: JingsongLi <[email protected]> AuthorDate: Tue Oct 20 15:04:03 2026 +0800 Revert "[python] Validate upstream table row-tracking for BlobView (#10647)" This reverts commit 16767ea96e0c714ab34e47bbac702305478b7a72. --- 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, 10 insertions(+), 220 deletions(-) diff --git a/paimon-python/pypaimon/table/row/blob.py b/paimon-python/pypaimon/table/row/blob.py index b7360a9da4..769fdfb717 100644 --- a/paimon-python/pypaimon/table/row/blob.py +++ b/paimon-python/pypaimon/table/row/blob.py @@ -427,51 +427,12 @@ 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 @@ -486,18 +447,12 @@ class BlobViewStruct: break if matched_field is None: raise ValueError( - f"Cannot find blob field {field_name} in upstream table " - f"{identifier.get_full_name()}." + 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 " - f"{identifier.get_full_name()} is not a BLOB field." + f"Field {field_name} in upstream table {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 @@ -506,37 +461,18 @@ 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 - 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 - ) + 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) - 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 - ) + 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), " - "(identifier, upstream_schema, field_name, row_id), or " - "(identifier, upstream_schema, upstream_options, field_name, " - "row_id)." + "BlobViewStruct.of expects (upstream_table, field_name, row_id) or " + "(identifier, upstream_schema, field_name, row_id)." ) @property @@ -813,13 +749,12 @@ 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) in (4, 5): + 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), " - "(identifier, upstream_schema, field_name, row_id), or " - "(identifier, upstream_schema, upstream_options, field_name, row_id)." + "(upstream_table, field_name, row_id), or " + "(identifier, upstream_schema, 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 efb636839f..458792633f 100755 --- a/paimon-python/pypaimon/tests/blob_table_test.py +++ b/paimon-python/pypaimon/tests/blob_table_test.py @@ -3066,85 +3066,6 @@ 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 77e0142262..3f4f2ebd44 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -393,7 +393,6 @@ 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) @@ -430,66 +429,6 @@ 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 89a7d5b6a9..e24eec5a67 100644 --- a/paimon-python/pypaimon/utils/blob_view_lookup.py +++ b/paimon-python/pypaimon/utils/blob_view_lookup.py @@ -186,11 +186,6 @@ 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,
