[python] Support creating primary-key BLOB tables and writing inline …

paimon-python/pypaimon/execution/python/write/file_store_write.py

@@ -22,9 +22,13 @@
 import pyarrow as pa
 
 from pypaimon.common.options.core_options import CoreOptions
-from pypaimon.schema.data_types import is_blob_file_field
+from pypaimon.schema.data_types import is_blob_file_field, is_blob_type
 from pypaimon.write.commit_message import CommitMessage
-from pypaimon.write.row_utils import inline_blob_value, require_columns, row_values_to_arrow_table
+from pypaimon.write.row_utils import (
+    normalize_inline_blob_row,
+    require_columns,
+    row_values_to_arrow_table,
+)
 from pypaimon.schema.arrow_schema import project_arrow_batch
 from pypaimon.execution.python.write.writer.append_only_data_writer import AppendOnlyDataWriter
 from pypaimon.execution.python.write.writer.dedicated_format_writer import DedicatedFormatWriter
@@ -127,27 +131,32 @@ def write_row(
         require_columns(values_by_name, column_names, 'write_row')
         writer = self.data_writers.get(key)
         data = None
-        if writer is None and self._has_blob_columns():
-            # Dedicated writers consume live payloads themselves. Inline fields
-            # instead encode to Arrow before the writer captures its layout.
-            descriptor_fields = self.options.blob_descriptor_fields()
-            view_fields = self.options.blob_view_fields()
+        if self._has_blob_columns():
+            descriptor_fields = CoreOptions.blob_descriptor_fields(self.options)
+            view_fields = CoreOptions.blob_view_fields(self.options)
             inline_columns = [name for name in column_names if name in descriptor_fields or name in view_fields]
-            if inline_columns:
-                values_by_name = dict(values_by_name)
-                for name in inline_columns:
-                    values_by_name[name] = inline_blob_value(
-                        values_by_name[name], name in descriptor_fields, name in view_fields)
-                # Reuse the encoded bytes in the dedicated row path. Keep the
-                # original event kind and leave unselected Blob objects alone.
-                from pypaimon.table.row.generic_row import GenericRow
-                row = GenericRow([values_by_name[name] for name in column_names],
-                                 [self.table.field_dict[name] for name in column_names], row.get_row_kind())
-            ordinary_columns = [name for name in column_names
-                                if name in inline_columns or not is_blob_file_field(self.table.field_dict[name])]
-            if ordinary_columns:
-                row_values_to_arrow_table(
-                    values_by_name, self.table.table_schema.fields, ordinary_columns)
+            row, values_by_name = normalize_inline_blob_row(
+                row,
+                values_by_name,
+                self.table.field_dict,
+                column_names,
+                descriptor_fields,
+                view_fields,
+            )
+            if not self.table.is_primary_key_table:
+                if writer is None:
+                    ordinary_columns = [
+                        name for name in column_names
+                        if name in inline_columns or not is_blob_file_field(self.table.field_dict[name])
+                    ]
+                    if ordinary_columns:
+                        row_values_to_arrow_table(
+                            values_by_name, self.table.table_schema.fields, ordinary_columns)
+            else:
+                data = row_values_to_arrow_table(
+                    values_by_name, self.table.table_schema.fields, column_names)
+                from pypaimon.write.row_kind import with_row_kind
+                data = with_row_kind(self.table, data, row)
         elif writer is None or not hasattr(writer, 'write_row'):
             data = row_values_to_arrow_table(
                 values_by_name, self.table.table_schema.fields, column_names)
@@ -220,13 +229,12 @@ def max_seq_number():
             default = -1 if self.table.is_primary_key_table else 1
             return self._seq_number_stats(partition).get(bucket, default)
 
-        # Dedicated Blob files are an append-table layout. PK tables require
-        # managed packs and references attached to their key-value Parquet files.
-        if self._has_blob_columns():
-            if self.table.is_primary_key_table:
-                raise NotImplementedError(
-                    'Primary-key Blob writes require the native writer; '
-                    'enable write.native.enabled and write Arrow batches')
+        # Dedicated Blob files are an append-table layout. PK tables with
+        # managed BLOB columns require packs and references attached to their
+        # key-value Parquet files, whereas inline BLOB fields
+        # (blob-descriptor-field and blob-view-field) are stored directly in
+        # the normal key-value data files.
+        if self._has_blob_columns() and not self.table.is_primary_key_table:
             return DedicatedFormatWriter(
                 table=self.table,
                 partition=partition,
@@ -238,6 +246,10 @@ def max_seq_number():
                 changelog_producer=self.changelog_producer,
                 blob_uri_reader_factory=self.blob_uri_reader_factory,
             )
+        elif self._has_managed_blob_columns() and self.table.is_primary_key_table:
+            raise NotImplementedError(
+                'Primary-key Blob writes require the native writer; '
+                'enable write.native.enabled and write Arrow batches')
         elif self._has_vector_columns() and options.with_vector_format():
             return DataVectorWriter(
                 table=self.table,
@@ -364,6 +376,18 @@ def _has_blob_columns(self) -> bool:
         """Check if the table schema contains blob columns."""
         return any(is_blob_file_field(field) for field in self.table.table_schema.fields)
 
+    def _has_managed_blob_columns(self) -> bool:
+        """Check if the table schema contains non-inline (managed) blob columns."""
+        inline_fields = (
+            CoreOptions.blob_descriptor_fields(self.options)
+            | CoreOptions.blob_view_fields(self.options)
+        )
+        return any(
+            is_blob_file_field(field)
+            and not (is_blob_type(field.type) and field.name in inline_fields)
+            for field in self.table.table_schema.fields
+        )
+
     def _has_vector_columns(self) -> bool:
         from pypaimon.schema.data_types import VectorType
         return any(isinstance(f.type, VectorType) for f in self.table.table_schema.fields)

paimon-python/pypaimon/execution/python/write/postpone_batch_table_write.py

@@ -18,12 +18,14 @@
 
 import pyarrow as pa
 
+from pypaimon.common.options.core_options import CoreOptions
 from pypaimon.write.commit_message import CommitMessage
 from pypaimon.execution.python.write.file_store_write import PostponeFixedBucketFileStoreWrite
 from pypaimon.execution.python.write.postpone_bucket import PythonPostponeBucketPlanner as PostponeBucketPlanner
 from pypaimon.execution.python.write.table_write import PythonTableWrite
 from pypaimon.execution.python.write.row_key_extractor import PostponeFixedBucketRowKeyExtractor
 from pypaimon.write.row_utils import (
+    normalize_inline_blob_row,
     require_columns,
     row_to_named_values,
     row_values_to_arrow_table,
@@ -129,6 +131,14 @@ def write_row(self, row):
         if self._bucket_plan.contains(partition):
             return super().write_row(row)
 
+        row, values_by_name = normalize_inline_blob_row(
+            row,
+            values_by_name,
+            self.table.field_dict,
+            column_names,
+            CoreOptions.blob_descriptor_fields(self.table.options),
+            CoreOptions.blob_view_fields(self.table.options),
+        )
         arrow_row = row_values_to_arrow_table(
             values_by_name, self.table.table_schema.fields, column_names)
         size = self._planner.input_partition_stats(arrow_row)[partition][1]

paimon-python/pypaimon/execution/python/write/writer/data_writer.py

@@ -98,6 +98,9 @@ def __init__(self, table, partition: Tuple, bucket: int, max_seq_number: int, op
         )
         self.write_cols = write_cols
         self.blob_as_descriptor = self.options.blob_as_descriptor()
+        self.blob_descriptor_fields = CoreOptions.blob_descriptor_fields(self.options)
+        self.blob_view_fields = CoreOptions.blob_view_fields(self.options)
+        self.blob_inline_fields = self.blob_descriptor_fields.union(self.blob_view_fields)
 
         self.path_factory = self.table.path_factory()
         self.external_path_provider: Optional[ExternalPathProvider] = self.path_factory.create_external_path_provider(
@@ -138,6 +141,7 @@ def pending_row_count(self) -> int:
 
     def write(self, data: pa.RecordBatch):
         try:
+            self._validate_inline_stored_fields_input(data)
             processed_data = self._process_data(data)
             self._buffer.append(processed_data)
             self._check_and_roll_if_needed()
@@ -148,6 +152,67 @@ def write(self, data: pa.RecordBatch):
             self.abort()
             raise e
 
+    def _validate_inline_stored_fields_input(self, data: pa.RecordBatch):
+        if not getattr(self, 'blob_inline_fields', None):
+            return
+
+        from pypaimon.table.row.blob import BlobDescriptor, BlobViewStruct
+
+        for field_name in getattr(self, 'blob_descriptor_fields', ()):
+            if field_name not in data.schema.names:
+                continue
+            values = data.column(data.schema.get_field_index(field_name)).to_pylist()
+            for value in values:
+                if value is None:
+                    continue
+                if hasattr(value, 'as_py'):
+                    value = value.as_py()
+                if isinstance(value, str):
+                    value = value.encode('utf-8')
+                if not isinstance(value, (bytes, bytearray)):
+                    raise ValueError(
+                        "blob-descriptor-field requires blob field value to be a serialized "
+                        "BlobDescriptor."
+                    )
+                descriptor_bytes = bytes(value)
+                # Like Java, schema-declared descriptors use the prefix parser.
+                # Exact wire length is only needed when detecting descriptors
+                # among arbitrary payload bytes.
+                try:
+                    BlobDescriptor.deserialize(descriptor_bytes)
+                except Exception as e:
+                    raise ValueError(
+                        "blob-descriptor-field requires blob field value to be a serialized "
+                        "BlobDescriptor."
+                    ) from e
+
+        for field_name in getattr(self, 'blob_view_fields', ()):
+            if field_name not in data.schema.names:
+                continue
+            values = data.column(data.schema.get_field_index(field_name)).to_pylist()
+            for value in values:
+                if value is None:
+                    continue
+                if hasattr(value, 'as_py'):
+                    value = value.as_py()
+                if isinstance(value, str):
+                    value = value.encode('utf-8')
+                if not isinstance(value, (bytes, bytearray)):
+                    raise ValueError(
+                        "blob-view-field requires blob field value to be a serialized "
+                        "BlobViewStruct."
+                    )
+                try:
+                    view_bytes = bytes(value)
+                    view_struct = BlobViewStruct.deserialize(view_bytes)
+                    if view_struct.serialize() != view_bytes:
+                        raise ValueError("BlobViewStruct payload contains trailing bytes.")
+                except Exception as e:
+                    raise ValueError(
+                        "blob-view-field requires blob field value to be a serialized "
+                        "BlobViewStruct."
+                    ) from e
+
     def prepare_commit(self) -> List[DataFileMeta]:
         if self._buffer.num_rows > 0:
             # Clear only once the write lands: a caller that retries

paimon-python/pypaimon/execution/python/write/writer/dedicated_format_writer.py

@@ -350,67 +350,6 @@ def _split_data(self, data: pa.RecordBatch) -> Tuple[
     def _project_columns(data: pa.RecordBatch, column_names: List[str]) -> pa.RecordBatch:
         return project_arrow_batch(data, column_names)
 
-    def _validate_inline_stored_fields_input(self, data: pa.RecordBatch):
-        if not self.blob_inline_fields:
-            return
-
-        from pypaimon.table.row.blob import BlobDescriptor, BlobViewStruct
-
-        for field_name in self.blob_descriptor_fields:
-            if field_name not in data.schema.names:
-                continue
-            values = data.column(data.schema.get_field_index(field_name)).to_pylist()
-            for value in values:
-                if value is None:
-                    continue
-                if hasattr(value, 'as_py'):
-                    value = value.as_py()
-                if isinstance(value, str):
-                    value = value.encode('utf-8')
-                if not isinstance(value, (bytes, bytearray)):
-                    raise ValueError(
-                        "blob-descriptor-field requires blob field value to be a serialized "
-                        "BlobDescriptor."
-                    )
-                descriptor_bytes = bytes(value)
-                # Like Java, schema-declared descriptors use the prefix parser.
-                # Exact wire length is only needed when detecting descriptors
-                # among arbitrary payload bytes.
-                try:
-                    BlobDescriptor.deserialize(descriptor_bytes)
-                except Exception as e:
-                    raise ValueError(
-                        "blob-descriptor-field requires blob field value to be a serialized "
-                        "BlobDescriptor."
-                    ) from e
-
-        for field_name in self.blob_view_fields:
-            if field_name not in data.schema.names:
-                continue
-            values = data.column(data.schema.get_field_index(field_name)).to_pylist()
-            for value in values:
-                if value is None:
-                    continue
-                if hasattr(value, 'as_py'):
-                    value = value.as_py()
-                if isinstance(value, str):
-                    value = value.encode('utf-8')
-                if not isinstance(value, (bytes, bytearray)):
-                    raise ValueError(
-                        "blob-view-field requires blob field value to be a serialized "
-                        "BlobViewStruct."
-                    )
-                try:
-                    view_bytes = bytes(value)
-                    view_struct = BlobViewStruct.deserialize(view_bytes)
-                    if view_struct.serialize() != view_bytes:
-                        raise ValueError("BlobViewStruct payload contains trailing bytes.")
-                except Exception as e:
-                    raise ValueError(
-                        "blob-view-field requires blob field value to be a serialized "
-                        "BlobViewStruct."
-                    ) from e
-
     @staticmethod
     def _process_normal_data(data: pa.RecordBatch) -> Optional[pa.Table]:
         """Process normal data (similar to base DataWriter)."""