[python] Parse BlobDescriptor v1/v2 bytes without misclassifying inline payload

paimon-python/pypaimon/common/options/core_options.py

@@ -1238,6 +1238,14 @@ def variant_shredding_schema(self) -> Optional[str]:
         return val
 
     def blob_descriptor_fields(self, default=None):
+        # Do not treat blob.stored-descriptor-fields as a layout switch.
+        # Python master ignored that key and wrote dedicated .blob payloads;
+        # a global fallback would mis-parse those files during a rolling
+        # upgrade. The cost is that Java tables which only set the fallback
+        # key store inline descriptors, and Python returns those bytes
+        # instead of fetching payload. Migrate explicitly to
+        # blob-descriptor-field (column directives already copy the legacy
+        # key onto the canonical option).
         value = self.options.get(CoreOptions.BLOB_DESCRIPTOR_FIELD, default)
         return CoreOptions._parse_field_set(value)
 

paimon-python/pypaimon/table/row/blob.py

@@ -28,6 +28,8 @@
 class BlobDescriptor:
     CURRENT_VERSION = 2
     MAGIC = 0x424C4F4244455343  # "BLOBDESC"
+    # v1 wire: version (1) + uri_length (4) + offset (8) + length (8)
+    _V1_MIN_WIRE_SIZE = 1 + 4 + 16
 
     def __init__(self, uri: str, offset: int, length: int):
         self._version = self.CURRENT_VERSION
@@ -54,13 +56,13 @@ def version(self) -> int:
     def serialize(self) -> bytes:
         uri_bytes = self._uri.encode('utf-8')
         uri_length = len(uri_bytes)
-        data = struct.pack('<B', self._version)  # version (1 byte)
-        if self._version > 1:
-            data += struct.pack('<Q', self.MAGIC)  # magic (8 bytes, unsigned)
-        data += struct.pack('<I', uri_length)  # uri length (4 bytes)
-        data += uri_bytes  # uri bytes
-        data += struct.pack('<q', self._offset)  # offset (8 bytes, signed)
-        data += struct.pack('<q', self._length)  # length (8 bytes, signed)
+        # Always write CURRENT_VERSION with magic, matching Java BlobDescriptor.serialize().
+        data = struct.pack('<B', self.CURRENT_VERSION)
+        data += struct.pack('<Q', self.MAGIC)
+        data += struct.pack('<i', uri_length)
+        data += uri_bytes
+        data += struct.pack('<q', self._offset)
+        data += struct.pack('<q', self._length)
         return data
 
     @classmethod
@@ -100,8 +102,12 @@ def _deserialize(cls, data: bytes) -> 'BlobDescriptor':
         # Read URI length
         if offset + 4 > len(data):
             raise ValueError("Invalid BlobDescriptor data: too short")
-        uri_length = struct.unpack('<I', data[offset:offset + 4])[0]
+        uri_length = struct.unpack('<i', data[offset:offset + 4])[0]
         offset += 4
+        if uri_length < 0:
+            raise ValueError(
+                f"Invalid BlobDescriptor data: negative URI length: {uri_length}"
+            )
 
         # Read URI bytes
         if offset + uri_length > len(data):
@@ -124,6 +130,51 @@ def _deserialize(cls, data: bytes) -> 'BlobDescriptor':
         descriptor._version = version
         return descriptor
 
+    @classmethod
+    def parse_if_serialized(cls, data: bytes) -> Optional['BlobDescriptor']:
+        """Parse when data is exactly a serialized descriptor (no trailing bytes).
+
+        Dispatches through :class:`BlobDescriptorSerde` so an exact
+        :class:`VideoFrameDescriptor` is accepted before the ordinary v1/v2
+        BlobDescriptor length check. Unlike :meth:`is_blob_descriptor` (v2
+        magic header only), this accepts v1 descriptors without a magic
+        prefix. Unlike ordinary :meth:`deserialize`, the encoded length must
+        match the buffer exactly. Still heuristic: arbitrary inline blob
+        bytes could theoretically match.
+        """
+        if not isinstance(data, (bytes, bytearray)):
+            return None
+        return BlobDescriptorSerde.parse_if_serialized(bytes(data))
+
+    @classmethod
+    def _parse_ordinary_if_serialized(cls, raw: bytes) -> Optional['BlobDescriptor']:
+        if len(raw) < cls._V1_MIN_WIRE_SIZE:
+            return None
+        try:
+            offset = 0
+            version = raw[offset]
+            offset += 1
+            if version < 1 or version > cls.CURRENT_VERSION:
+                return None
+            if version > 1:
+                if offset + 8 > len(raw):
+                    return None
+                magic = struct.unpack('<Q', raw[offset:offset + 8])[0]
+                if magic != cls.MAGIC:
+                    return None
+                offset += 8
+            if offset + 4 > len(raw):
+                return None
+            uri_length = struct.unpack('<i', raw[offset:offset + 4])[0]
+            if uri_length < 0:
+                return None
+            total = offset + 4 + uri_length + 16
+            if total != len(raw):
+                return None
+            return cls._deserialize(raw)
+        except (ValueError, struct.error, UnicodeDecodeError):
+            return None
+
     @classmethod
     def is_blob_descriptor(cls, data: bytes) -> bool:
         if not isinstance(data, (bytes, bytearray)):
@@ -296,6 +347,19 @@ def deserialize(data: bytes) -> BlobDescriptor:
             return VideoFrameDescriptor.deserialize(data)
         return BlobDescriptor._deserialize(data)
 
+    @staticmethod
+    def parse_if_serialized(data: bytes) -> Optional[BlobDescriptor]:
+        """Exact-length parse for any persisted BlobDescriptor wire type."""
+        if not isinstance(data, (bytes, bytearray)):
+            return None
+        raw = bytes(data)
+        if VideoFrameDescriptor.is_video_frame_descriptor(raw):
+            try:
+                return VideoFrameDescriptor.deserialize(raw)
+            except (ValueError, struct.error, UnicodeDecodeError):
+                return None
+        return BlobDescriptor._parse_ordinary_if_serialized(raw)
+
 
 class BlobViewStruct:
     CURRENT_VERSION = 1
@@ -520,6 +584,49 @@ def from_file(file_io, file_path: str, offset: int, length: int) -> 'Blob':
     def from_descriptor(uri_reader: UriReader, descriptor: BlobDescriptor) -> 'Blob':
         return BlobRef(uri_reader, descriptor)
 
+    @staticmethod
+    def _blob_ref_from_descriptor(
+            descriptor: 'BlobDescriptor', file_io=None, uri_reader_factory=None,
+    ) -> 'BlobRef':
+        if uri_reader_factory is None:
+            if file_io is None:
+                raise ValueError("file_io is required to resolve BlobDescriptor bytes")
+            uri_reader = UriReader.from_file(file_io)
+        else:
+            uri_reader = uri_reader_factory.create(descriptor.uri)
+        return BlobRef(uri_reader, descriptor)
+
+    @staticmethod
+    def from_descriptor_bytes(
+            data: Optional[bytes], file_io=None, uri_reader_factory=None,
+    ) -> Optional['Blob']:
+        """Build a Blob from bytes known to contain a descriptor.
+
+        Version 1 descriptors have no magic header, so they cannot be
+        distinguished safely from arbitrary payload bytes. Callers which know
+        from schema or storage context that a value is a descriptor must use
+        this method instead of the heuristic :meth:`from_bytes` entry point.
+
+        Parsing uses :meth:`BlobDescriptor.deserialize`, matching Java: a
+        valid v1/v2 prefix is accepted and trailing bytes after that prefix
+        are ignored. This is not a detector; garbage that happens to look
+        like a v1 prefix can produce a BlobRef with a nonsense URI.
+        Bytes that are not a parseable prefix raise :class:`ValueError`.
+        """
+        if data is None:
+            return None
+        if not isinstance(data, (bytes, bytearray)):
+            raise TypeError(
+                f"Blob.from_descriptor_bytes expects bytes, got {type(data)}")
+
+        try:
+            descriptor = BlobDescriptor.deserialize(bytes(data))
+        except (ValueError, struct.error, UnicodeDecodeError) as exc:
+            raise ValueError(
+                "Expected BlobDescriptor bytes, got raw bytes") from exc
+        return Blob._blob_ref_from_descriptor(
+            descriptor, file_io=file_io, uri_reader_factory=uri_reader_factory)
+
     @staticmethod
     def from_view(view_struct: BlobViewStruct) -> 'BlobView':
         return BlobView(view_struct)
@@ -535,20 +642,16 @@ def from_bytes(
         data = bytes(data)
         if BlobViewStruct.is_blob_view_struct(data):
             return Blob.from_view(BlobViewStruct.deserialize(data))
-        is_descriptor = BlobDescriptorSerde.is_descriptor(data)
-        if not allow_blob_data and not is_descriptor:
-            raise ValueError(
-                "Expected BlobDescriptor bytes, got raw bytes (allow_blob_data=False)"
-            )
-        if is_descriptor:
-            descriptor = BlobDescriptorSerde.deserialize(data)
-            if uri_reader_factory is None:
-                if file_io is None:
-                    raise ValueError("file_io is required to resolve BlobDescriptor bytes")
-                uri_reader = UriReader.from_file(file_io)
-            else:
-                uri_reader = uri_reader_factory.create(descriptor.uri)
-            return BlobRef(uri_reader, descriptor)
+        if BlobDescriptorSerde.is_descriptor(data) or not allow_blob_data:
+            try:
+                descriptor = BlobDescriptor.deserialize(data)
+            except (ValueError, struct.error, UnicodeDecodeError) as exc:
+                raise ValueError(
+                    "Expected BlobDescriptor bytes, got raw bytes"
+                    + ("" if allow_blob_data else " (allow_blob_data=False)")
+                ) from exc
+            return Blob._blob_ref_from_descriptor(
+                descriptor, file_io=file_io, uri_reader_factory=uri_reader_factory)
         return BlobData(data)
 
 
@@ -637,6 +740,11 @@ def to_data(self) -> bytes:
     def to_descriptor(self) -> BlobDescriptor:
         return self._descriptor
 
+    @property
+    def uri_reader(self) -> UriReader:
+        """UriReader used to fetch this blob's payload."""
+        return self._uri_reader
+
     def new_input_stream(self) -> BinaryIO:
         uri = self._descriptor.uri
         offset = self._descriptor.offset

paimon-python/pypaimon/tests/blob_table_test.py

@@ -5247,6 +5247,52 @@ def test_blob_table_partial_update_non_blob_column_with_rolling_files(self):
         self.assertEqual(result['id'], list(range(2000)))
         self.assertEqual(result['name'], ['updated'] * 2000)
 
+    def test_legacy_stored_descriptor_fields_keeps_dedicated_blob_layout(self):
+        """blob.stored-descriptor-fields must not switch Python to inline descriptors.
+
+        Master ignored that key and wrote dedicated .blob payloads. Head write
+        with the same option must keep that layout so old readers still see
+        payloads, and head read must not fail-fast on those bytes.
+        """
+        from pypaimon import Schema
+
+        pa_schema = pa.schema([
+            ('id', pa.int32()),
+            ('picture', pa.large_binary()),
+        ])
+        schema = Schema.from_pyarrow_schema(
+            pa_schema,
+            options={
+                'row-tracking.enabled': 'true',
+                'data-evolution.enabled': 'true',
+                'blob.stored-descriptor-fields': 'picture',
+            }
+        )
+        self.catalog.create_table(
+            'test_db.legacy_stored_descriptor_fields', schema, False)
+        table = self.catalog.get_table('test_db.legacy_stored_descriptor_fields')
+
+        payload = b'legacy-dedicated-blob-payload'
+        write_builder = table.new_batch_write_builder()
+        writer = write_builder.new_write()
+        writer.write_arrow(pa.Table.from_pydict({
+            'id': [1],
+            'picture': [payload],
+        }, schema=pa_schema))
+        commit_messages = writer.prepare_commit()
+        write_builder.new_commit().commit(commit_messages)
+        writer.close()
+
+        all_files = [f for msg in commit_messages for f in msg.new_files]
+        blob_files = [f for f in all_files if f.file_name.endswith('.blob')]
+        self.assertGreaterEqual(len(blob_files), 1)
+        self.assertTrue(all(f.write_cols == ['picture'] for f in blob_files))
+
+        result = table.new_read_builder().new_read().to_arrow(
+            table.new_read_builder().new_scan().plan().splits())
+        self.assertEqual(result.num_rows, 1)
+        self.assertEqual(result.column('picture').to_pylist()[0], payload)
+
 
 class GetBlobTest(unittest.TestCase):