[python] Support creating primary-key BLOB tables and writing inline BLOB fields on PK tables

Author: zhuanshenbsj1

paimon-python/pypaimon/schema/schema_manager.py

@@ -437,35 +437,125 @@ def _validate_blob_fields(
             )
         )
 
+    if video_fields and primary_keys:
+        raise ValueError(
+            f"'{CoreOptions.VIDEO_FRAME_FIELD.key()}' only supports append-only tables."
+        )
+
     if blob_field_names:
-        required_options = {
-            CoreOptions.ROW_TRACKING_ENABLED.key(): 'true',
-            CoreOptions.DATA_EVOLUTION_ENABLED.key(): 'true'
-        }
+        if blob_field_names.intersection(partition_keys):
+            raise ValueError(
+                "The BLOB, ARRAY<BLOB> or MAP<X, BLOB> type column can not be "
+                "part of partition keys."
+            )
 
-        missing_options = []
-        for key, expected_value in required_options.items():
-            if key not in options or options[key] != expected_value:
-                missing_options.append(f"{key}='{expected_value}'")
+        if not primary_keys:
+            required_options = {
+                CoreOptions.ROW_TRACKING_ENABLED.key(): 'true',
+                CoreOptions.DATA_EVOLUTION_ENABLED.key(): 'true'
+            }
+
+            missing_options = []
+            for key, expected_value in required_options.items():
+                if key not in options or options[key] != expected_value:
+                    missing_options.append(f"{key}='{expected_value}'")
+
+            if missing_options:
+                raise ValueError(
+                    f"Schema contains BLOB, ARRAY<BLOB> or MAP<X, BLOB> type but is "
+                    f"missing required options: "
+                    f"{', '.join(missing_options)}. "
+                    f"Please add these options to the schema."
+                )
+        else:
+            _validate_primary_key_blob_fields(
+                fields,
+                options,
+                core_options,
+                primary_keys,
+                partition_keys,
+                blob_field_names,
+                all_inline_fields,
+            )
+
+
+def _validate_primary_key_blob_fields(
+    fields: List[DataField],
+    options: dict,
+    core_options: CoreOptions,
+    primary_keys: List[str],
+    partition_keys: List[str],
+    blob_field_names: set,
+    all_inline_fields: set,
+):
+    """Mirror Java SchemaValidation#validatePrimaryKeyBlobKeyConfiguration and
+    validatePrimaryKeyBlobConfiguration."""
+    from pypaimon.common.options.core_options import ChangelogProducer, MergeEngine
+
+    managed_blob_fields = [
+        field.name
+        for field in fields
+        if field.name in blob_field_names and field.name not in all_inline_fields
+    ]
+    configured_bucket_key = options.get(CoreOptions.BUCKET_KEY.key())
+    if configured_bucket_key and configured_bucket_key.strip():
+        bucket_keys = [
+            key.strip() for key in configured_bucket_key.split(',') if key.strip()
+        ]
+    else:
+        bucket_keys = [pk for pk in primary_keys if pk not in partition_keys]
 
-        if missing_options:
+    if managed_blob_fields:
+        pk_blob_fields = [f for f in managed_blob_fields if f in primary_keys]
+        if pk_blob_fields:
             raise ValueError(
-                f"Schema contains BLOB, ARRAY<BLOB> or MAP<X, BLOB> type but is "
-                f"missing required options: "
-                f"{', '.join(missing_options)}. "
-                f"Please add these options to the schema."
+                f"Managed BLOB fields cannot be primary keys: {pk_blob_fields}."
             )
-
-        if primary_keys:
+        bucket_blob_fields = [f for f in managed_blob_fields if f in bucket_keys]
+        if bucket_blob_fields:
             raise ValueError(
-                "BLOB, ARRAY<BLOB> or MAP<X, BLOB> type is not supported with primary key."
+                f"Managed BLOB fields cannot be bucket keys: {bucket_blob_fields}."
+            )
+        sequence_fields = core_options.sequence_field()
+        sequence_blob_fields = [f for f in managed_blob_fields if f in sequence_fields]
+        if sequence_blob_fields:
+            raise ValueError(
+                f"Managed BLOB fields cannot be sequence fields: {sequence_blob_fields}."
             )
 
-        if blob_field_names.intersection(partition_keys):
+        if core_options.merge_engine() not in (
+            MergeEngine.DEDUPLICATE,
+            MergeEngine.PARTIAL_UPDATE,
+            MergeEngine.FIRST_ROW,
+        ):
             raise ValueError(
-                "The BLOB, ARRAY<BLOB> or MAP<X, BLOB> type column can not be "
-                "part of partition keys."
+                "Primary-key managed BLOB tables only support the deduplicate, "
+                "partial-update or first-row merge engine."
+            )
+        if core_options.changelog_producer() != ChangelogProducer.NONE:
+            raise ValueError(
+                "Primary-key managed BLOB tables only support changelog-producer 'none'."
             )
+        if core_options.data_file_external_paths() is not None:
+            raise ValueError(
+                f"Primary-key managed BLOB tables do not support "
+                f"'{CoreOptions.DATA_FILE_EXTERNAL_PATHS.key()}'."
+            )
+        if str(options.get('pk-clustering-override', 'false')).lower() == 'true':
+            raise ValueError(
+                "Primary-key managed BLOB tables do not support 'pk-clustering-override'."
+            )
+
+    if blob_field_names.intersection(primary_keys):
+        raise ValueError(
+            "The BLOB, ARRAY<BLOB> or MAP<X, BLOB> type column can not be "
+            "part of primary keys."
+        )
+    if blob_field_names.intersection(bucket_keys):
+        raise ValueError(
+            "The BLOB, ARRAY<BLOB> or MAP<X, BLOB> type column can not be "
+            "part of bucket keys."
+        )
 
 
 def _validate_options(options: dict):

paimon-python/pypaimon/tests/blob_table_test.py

@@ -5811,6 +5811,219 @@ def test_blob_view_as_descriptor_projection_with_predicate_extra_field(self):
         )
         self.assertFalse(BlobViewStruct.is_blob_view_struct(picture_bytes))
 
+    def test_primary_key_blob_descriptor_field_write_and_upsert(self):
+        from pypaimon import Schema
+        from pypaimon.table.row.blob import Blob, BlobDescriptor
+        from pypaimon.table.row.generic_row import GenericRow
+
+        blob_file = os.path.join(self.temp_dir, 'pk_descriptor_payloads.bin')
+        with open(blob_file, 'wb') as f:
+            f.write(b'payload-v1payload-v2payload-row-3')
+
+        desc_v1 = BlobDescriptor(blob_file, 0, 10)
+        desc_v2 = BlobDescriptor(blob_file, 10, 10)
+        desc_row3 = BlobDescriptor(blob_file, 20, 13)
+
+        pa_schema = pa.schema([
+            pa.field('id', pa.int32(), nullable=False),
+            pa.field('name', pa.string()),
+            pa.field('picture', pa.large_binary()),
+        ])
+        schema = Schema.from_pyarrow_schema(
+            pa_schema,
+            primary_keys=['id'],
+            options={
+                'bucket': '1',
+                'blob-descriptor-field': 'picture',
+            },
+        )
+        self.catalog.create_table('test_db.pk_blob_descriptor_table', schema, False)
+        table = self.catalog.get_table('test_db.pk_blob_descriptor_table')
+
+        # 1. Invalid descriptor payload is rejected by the PK writer
+        write_builder = table.new_batch_write_builder()
+        bad_writer = write_builder.new_write()
+        with self.assertRaisesRegex(
+            ValueError,
+            "blob-descriptor-field requires blob field value to be a serialized BlobDescriptor.",
+        ):
+            bad_writer.write_arrow(pa.Table.from_pydict({
+                'id': [1],
+                'name': ['bad'],
+                'picture': [b'not-a-descriptor'],
+            }, schema=pa_schema))
+        bad_writer.close()
+
+        # 2. Initial commit via write_arrow + write_row
+        writer1 = write_builder.new_write()
+        writer1.write_arrow(pa.Table.from_pydict({
+            'id': [1, 2],
+            'name': ['alice-v1', 'bob'],
+            'picture': [desc_v1.serialize(), None],
+        }, schema=pa_schema))
+        writer1.write_row(GenericRow(
+            [
+                3,
+                'charlie',
+                Blob.from_descriptor_bytes(
+                    desc_row3.serialize(), file_io=table.file_io
+                ),
+            ],
+            table.fields,
+        ))
+        write_builder.new_commit().commit(writer1.prepare_commit())
+        writer1.close()
+
+        # 3. Second commit upserting id=1 with desc_v2
+        writer2 = write_builder.new_write()
+        writer2.write_arrow(pa.Table.from_pydict({
+            'id': [1],
+            'name': ['alice-v2'],
+            'picture': [desc_v2.serialize()],
+        }, schema=pa_schema))
+        write_builder.new_commit().commit(writer2.prepare_commit())
+        writer2.close()
+
+        read_builder = table.new_read_builder()
+        result = read_builder.new_read().to_arrow(
+            read_builder.new_scan().plan().splits()
+        ).sort_by('id')
+        self.assertEqual(result.column('id').to_pylist(), [1, 2, 3])
+        self.assertEqual(
+            result.column('name').to_pylist(),
+            ['alice-v2', 'bob', 'charlie'],
+        )
+        self.assertEqual(
+            result.column('picture').to_pylist(),
+            [b'payload-v2', None, b'payload-row-3'],
+        )
+
+    def test_primary_key_blob_view_field_write_and_upsert(self):
+        from pypaimon import Schema
+        from pypaimon.table.row.blob import Blob, BlobViewStruct
+        from pypaimon.table.row.generic_row import GenericRow
+
+        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.pk_blob_view_source', source, False)
+        source_table = self.catalog.get_table('test_db.pk_blob_view_source')
+
+        src_wb = source_table.new_batch_write_builder()
+        src_writer = src_wb.new_write()
+        src_writer.write_arrow(pa.Table.from_pydict({
+            'id': [10, 20, 30],
+            'picture': [b'upstream-0', b'upstream-1', b'upstream-2'],
+        }, schema=source_schema))
+        src_wb.new_commit().commit(src_writer.prepare_commit())
+        src_writer.close()
+
+        target_schema = pa.schema([
+            pa.field('id', pa.int32(), nullable=False),
+            pa.field('label', pa.string()),
+            pa.field('picture', pa.large_binary()),
+        ])
+        target = Schema.from_pyarrow_schema(
+            target_schema,
+            primary_keys=['id'],
+            options={
+                'bucket': '1',
+                'blob-view-field': 'picture',
+            },
+        )
+        self.catalog.create_table('test_db.pk_blob_view_target', target, False)
+        target_table = self.catalog.get_table('test_db.pk_blob_view_target')
+
+        view_0 = BlobViewStruct.of('test_db.pk_blob_view_source', source_table, 'picture', 0)
+        view_1 = BlobViewStruct.of('test_db.pk_blob_view_source', source_table, 'picture', 1)
+        view_2 = BlobViewStruct.of('test_db.pk_blob_view_source', source_table, 'picture', 2)
+
+        # 1. Invalid BlobViewStruct bytes are rejected by the PK writer
+        target_wb = target_table.new_batch_write_builder()
+        bad_writer = target_wb.new_write()
+        with self.assertRaisesRegex(
+            ValueError,
+            "blob-view-field requires blob field value to be a serialized BlobViewStruct.",
+        ):
+            bad_writer.write_arrow(pa.Table.from_pydict({
+                'id': [1],
+                'label': ['bad'],
+                'picture': [b'invalid-view-bytes'],
+            }, schema=target_schema))
+        bad_writer.close()
+
+        # 2. First commit via write_arrow + write_row
+        writer1 = target_wb.new_write()
+        writer1.write_arrow(pa.Table.from_pydict({
+            'id': [1, 2],
+            'label': ['row1-v1', 'row2'],
+            'picture': [view_0.serialize(), None],
+        }, schema=target_schema))
+        writer1.write_row(GenericRow(
+            [3, 'row3', Blob.from_view(view_2)],
+            target_table.fields,
+        ))
+        target_wb.new_commit().commit(writer1.prepare_commit())
+        writer1.close()
+
+        # 3. Second commit upserting id=1 to point to upstream row 1
+        writer2 = target_wb.new_write()
+        writer2.write_arrow(pa.Table.from_pydict({
+            'id': [1],
+            'label': ['row1-v2'],
+            'picture': [view_1.serialize()],
+        }, schema=target_schema))
+        target_wb.new_commit().commit(writer2.prepare_commit())
+        writer2.close()
+
+        read_builder = target_table.new_read_builder()
+        result = read_builder.new_read().to_arrow(
+            read_builder.new_scan().plan().splits()
+        ).sort_by('id')
+        self.assertEqual(result.column('id').to_pylist(), [1, 2, 3])
+        self.assertEqual(
+            result.column('label').to_pylist(),
+            ['row1-v2', 'row2', 'row3'],
+        )
+        self.assertEqual(
+            result.column('picture').to_pylist(),
+            [b'upstream-1', None, b'upstream-2'],
+        )
+
+    def test_primary_key_managed_blob_requires_native_writer(self):
+        from pypaimon import Schema
+
+        pa_schema = pa.schema([
+            pa.field('id', pa.int32(), nullable=False),
+            pa.field('picture', pa.large_binary()),
+        ])
+        schema = Schema.from_pyarrow_schema(
+            pa_schema,
+            primary_keys=['id'],
+            options={'bucket': '1'},
+        )
+        self.catalog.create_table('test_db.pk_managed_blob_table', schema, False)
+        table = self.catalog.get_table('test_db.pk_managed_blob_table')
+
+        writer = table.new_batch_write_builder().new_write()
+        with self.assertRaisesRegex(
+            NotImplementedError,
+            "Primary-key Blob writes require the native writer",
+        ):
+            writer.write_arrow(pa.Table.from_pydict({
+                'id': [1],
+                'picture': [b'raw-blob-bytes'],
+            }, schema=pa_schema))
+        writer.close()
+
 
 class GetBlobTest(unittest.TestCase):