From 252cb6e9eb47c3483ec7b8c472a019ddb7422d6d Mon Sep 17 00:00:00 2001
From: "wenchao.wu" <wenchao.wu@shopee.com>
Date: Mon, 31 Aug 2026 18:44:00 +0800
Subject: [PATCH] [python] Parse BlobDescriptor v1/v2 bytes without
 misclassifying inline payload.

Align serialize() with Java (always CURRENT_VERSION + magic) and add
explicit parse APIs for known descriptor bytes. Writer validation uses
exact wire length so exact v1 input still lands; from_bytes() stays
v2-magic-only so inline blob payload is not treated as a v1 descriptor.
---
 .../pypaimon/common/options/core_options.py   |   8 +
 paimon-python/pypaimon/table/row/blob.py      | 152 ++++++++++--
 .../pypaimon/tests/blob_table_test.py         |  46 ++++
 paimon-python/pypaimon/tests/blob_test.py     | 233 +++++++++++++++++-
 .../write/writer/dedicated_format_writer.py   |  19 +-
 5 files changed, 431 insertions(+), 27 deletions(-)

diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py
index 1ae2f3187d9a..6e92ceb26007 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/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)
 
diff --git a/paimon-python/pypaimon/table/row/blob.py b/paimon-python/pypaimon/table/row/blob.py
index 4988f61c248b..fb700721de3e 100644
--- a/paimon-python/pypaimon/table/row/blob.py
+++ b/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
diff --git a/paimon-python/pypaimon/tests/blob_table_test.py b/paimon-python/pypaimon/tests/blob_table_test.py
index c08289b4efdb..1ab02b56ea1f 100755
--- a/paimon-python/pypaimon/tests/blob_table_test.py
+++ b/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):
 
diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py
index 544e00cdee8f..2f3d8abd35e0 100644
--- a/paimon-python/pypaimon/tests/blob_test.py
+++ b/paimon-python/pypaimon/tests/blob_test.py
@@ -41,7 +41,16 @@
 from pypaimon.read.reader.concat_batch_reader import BlobFallbackBatchReader
 from pypaimon.read.reader.format_blob_reader import BlobRecordIterator, FormatBlobReader
 from pypaimon.schema.data_types import ArrayType, AtomicType, DataField, MapType
-from pypaimon.table.row.blob import Blob, BlobData, BlobRef, BlobDescriptor, BlobViewStruct, BlobView
+from pypaimon.table.row.blob import (
+    Blob,
+    BlobData,
+    BlobRef,
+    BlobDescriptor,
+    BlobDescriptorSerde,
+    BlobViewStruct,
+    BlobView,
+    VideoFrameDescriptor,
+)
 from pypaimon.table.row.generic_row import GenericRowDeserializer, GenericRowSerializer, GenericRow
 from pypaimon.table.row.row_kind import RowKind
 from pypaimon.utils.range import Range
@@ -1282,6 +1291,14 @@ def test_blob_descriptor_deserialization_invalid_data(self):
             BlobDescriptor.deserialize(incomplete_data)
         self.assertIn("URI length exceeds data size", str(context.exception))
 
+        # Java reads uri length as a signed int and rejects negatives.
+        negative_uri = bytearray(valid_descriptor.serialize())
+        struct.pack_into('<i', negative_uri, 1 + 8, -1)
+        with self.assertRaises(ValueError) as context:
+            BlobDescriptor.deserialize(bytes(negative_uri))
+        self.assertIn("negative URI length", str(context.exception))
+        self.assertIsNone(BlobDescriptor.parse_if_serialized(bytes(negative_uri)))
+
     def test_blob_descriptor_equality_and_hashing(self):
         """Test BlobDescriptor equality and hashing."""
         # Create identical descriptors
@@ -1411,6 +1428,16 @@ def test_blob_descriptor_detection(self):
         )
         random_bytes = b"not-a-descriptor"
         fake_v1_prefix = b"\x01not-a-descriptor"
+        # v1-shaped inline payload: passes len/version/uri-length checks but fails
+        # exact total-length match, so it must not be parsed as a descriptor.
+        v1_shaped_inline = (
+            bytes([1])
+            + struct.pack('<I', 5)
+            + b"hello"
+            + struct.pack('<q', 0)
+            + struct.pack('<q', 5)
+            + b"\xff"
+        )
         v2_magic_only = bytes([2]) + struct.pack('<Q', BlobDescriptor.MAGIC)
 
         self.assertTrue(BlobDescriptor.is_blob_descriptor(descriptor_v2.serialize()))
@@ -1421,6 +1448,210 @@ def test_blob_descriptor_detection(self):
         self.assertFalse(BlobDescriptor.is_blob_descriptor(fake_v1_prefix))
         self.assertFalse(BlobDescriptor.is_blob_descriptor(b"tiny"))
 
+        self.assertIsNotNone(BlobDescriptor.parse_if_serialized(descriptor_v1_bytes))
+        self.assertIsNotNone(BlobDescriptor.parse_if_serialized(descriptor_v2.serialize()))
+        self.assertIsNone(BlobDescriptor.parse_if_serialized(random_bytes))
+        self.assertIsNone(BlobDescriptor.parse_if_serialized(fake_v1_prefix))
+        self.assertIsNone(BlobDescriptor.parse_if_serialized(v1_shaped_inline))
+        self.assertIsNone(BlobDescriptor.parse_if_serialized(b"tiny"))
+
+        video = VideoFrameDescriptor("file:///v.mp4", 0, 10, 2)
+        video_bytes = video.serialize()
+        self.assertEqual(video_bytes, BlobDescriptor.deserialize(video_bytes).serialize())
+        self.assertEqual(video, BlobDescriptor.parse_if_serialized(video_bytes))
+        self.assertEqual(video, BlobDescriptorSerde.parse_if_serialized(video_bytes))
+        self.assertIsNone(BlobDescriptor.parse_if_serialized(video_bytes + b"x"))
+
+    def test_from_descriptor_bytes_rejects_non_descriptor_bytes(self):
+        with self.assertRaises(ValueError) as ctx:
+            Blob.from_descriptor_bytes(b"hello blob", file_io=LocalFileIO())
+        self.assertIn("Expected BlobDescriptor bytes", str(ctx.exception))
+
+    def test_from_descriptor_bytes_accepts_v1_descriptor_with_trailing_bytes(self):
+        uri = b"file:///tmp/blob.bin"
+        serialized_v1 = (
+            bytes([1])
+            + struct.pack('<I', len(uri))
+            + uri
+            + struct.pack('<q', 0)
+            + struct.pack('<q', 10)
+        )
+        padded = serialized_v1 + b"extra"
+        blob = Blob.from_descriptor_bytes(padded, file_io=LocalFileIO())
+        self.assertIsInstance(blob, BlobRef)
+        self.assertEqual(blob.to_descriptor().uri, "file:///tmp/blob.bin")
+
+    def test_from_descriptor_bytes_accepts_v2_descriptor_with_trailing_bytes(self):
+        descriptor = BlobDescriptor("file:///tmp/blob.bin", 0, 10)
+        padded = descriptor.serialize() + b"extra"
+        blob = Blob.from_descriptor_bytes(padded, file_io=LocalFileIO())
+        self.assertIsInstance(blob, BlobRef)
+        self.assertEqual(blob.to_descriptor().uri, descriptor.uri)
+
+    def test_from_bytes_allow_blob_data_false_rejects_raw_bytes(self):
+        with self.assertRaises(ValueError) as ctx:
+            Blob.from_bytes(b"hello blob", allow_blob_data=False)
+        self.assertIn("Expected BlobDescriptor bytes", str(ctx.exception))
+        self.assertIn("allow_blob_data=False", str(ctx.exception))
+
+    def test_from_bytes_allow_blob_data_false_requires_file_io_for_valid_descriptor(self):
+        descriptor = BlobDescriptor("file:///tmp/blob.bin", 0, 10)
+        with self.assertRaises(ValueError) as ctx:
+            Blob.from_bytes(descriptor.serialize(), allow_blob_data=False)
+        self.assertIn("file_io is required", str(ctx.exception))
+        self.assertNotIn("allow_blob_data=False", str(ctx.exception))
+
+    def test_from_bytes_allow_blob_data_false_accepts_v1_descriptor(self):
+        data = b"actual blob content"
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            blob_path = os.path.join(tmp_dir, "blob.bin")
+            with open(blob_path, 'wb') as f:
+                f.write(data)
+            uri = blob_path.encode('utf-8')
+            serialized_v1 = (
+                bytes([1])
+                + struct.pack('<I', len(uri))
+                + uri
+                + struct.pack('<q', 0)
+                + struct.pack('<q', len(data))
+            )
+            file_io = FileIO.get(f"file://{tmp_dir}", {})
+            blob = Blob.from_bytes(serialized_v1, file_io=file_io, allow_blob_data=False)
+            self.assertIsInstance(blob, BlobRef)
+            self.assertEqual(blob.to_data(), data)
+
+    def test_from_descriptor_bytes_with_v1_descriptor_bytes(self):
+        data = b"actual blob content"
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            blob_path = os.path.join(tmp_dir, "blob.bin")
+            with open(blob_path, 'wb') as f:
+                f.write(data)
+            uri = blob_path.encode('utf-8')
+            serialized_v1 = (
+                bytes([1])
+                + struct.pack('<I', len(uri))
+                + uri
+                + struct.pack('<q', 0)
+                + struct.pack('<q', len(data))
+            )
+            file_io = FileIO.get(f"file://{tmp_dir}", {})
+            blob = Blob.from_descriptor_bytes(serialized_v1, file_io)
+            self.assertIsInstance(blob, BlobRef)
+            self.assertEqual(blob.to_data(), data)
+
+    def test_from_bytes_does_not_misparse_v1_shaped_inline_payload(self):
+        # Exact-length inline bytes that match the v1 descriptor wire layout must
+        # stay as BlobData when parsed via the heuristic from_bytes entry point.
+        v1_shaped_inline = (
+            bytes([1])
+            + struct.pack('<I', 5)
+            + b"hello"
+            + struct.pack('<q', 0)
+            + struct.pack('<q', 5)
+        )
+        blob = Blob.from_bytes(v1_shaped_inline)
+        self.assertIsInstance(blob, BlobData)
+        self.assertEqual(blob.to_data(), v1_shaped_inline)
+
+    def test_blob_descriptor_serialize_uses_current_version(self):
+        uri = b"file:///tmp/blob.bin"
+        serialized_v1 = (
+            bytes([1])
+            + struct.pack('<I', len(uri))
+            + uri
+            + struct.pack('<q', 0)
+            + struct.pack('<q', 10)
+        )
+        descriptor = BlobDescriptor.deserialize(serialized_v1)
+        self.assertEqual(descriptor.version, 1)
+
+        serialized = descriptor.serialize()
+        self.assertEqual(serialized[0], BlobDescriptor.CURRENT_VERSION)
+        self.assertTrue(BlobDescriptor.is_blob_descriptor(serialized))
+        self.assertEqual(
+            BlobDescriptor.deserialize(serialized),
+            BlobDescriptor("file:///tmp/blob.bin", 0, 10),
+        )
+
+    def test_blob_descriptor_fields_ignores_legacy_stored_key(self):
+        from pypaimon.common.options.core_options import CoreOptions
+
+        # Python master ignored this key and wrote dedicated .blob files.
+        # A global fallback would break rolling upgrades.
+        legacy_only = CoreOptions(
+            Options({"blob.stored-descriptor-fields": "legacy_col"}))
+        self.assertEqual(set(), legacy_only.blob_descriptor_fields())
+
+        canonical_wins = CoreOptions(Options({
+            "blob-descriptor-field": "canon",
+            "blob.stored-descriptor-fields": "legacy_col",
+        }))
+        self.assertEqual({"canon"}, canonical_wins.blob_descriptor_fields())
+
+        blank_canonical = CoreOptions(Options({
+            "blob-descriptor-field": "",
+            "blob.stored-descriptor-fields": "legacy_col",
+        }))
+        self.assertEqual(set(), blank_canonical.blob_descriptor_fields())
+
+    def test_dedicated_writer_accepts_exact_v1_descriptor_bytes(self):
+        from pypaimon.write.writer.dedicated_format_writer import (
+            DedicatedFormatWriter)
+
+        uri = b"file:///tmp/blob.bin"
+        v1 = (
+            bytes([1])
+            + struct.pack('<I', len(uri))
+            + uri
+            + struct.pack('<q', 0)
+            + struct.pack('<q', 10)
+        )
+        writer = object.__new__(DedicatedFormatWriter)
+        writer.blob_inline_fields = {'payload'}
+        writer.blob_descriptor_fields = {'payload'}
+        writer.blob_view_fields = set()
+        v2 = BlobDescriptor("file:///tmp/blob.bin", 0, 10).serialize()
+        writer._validate_inline_stored_fields_input(
+            pa.RecordBatch.from_arrays(
+                [pa.array([v2], type=pa.large_binary())], names=['payload']))
+        writer._validate_inline_stored_fields_input(
+            pa.RecordBatch.from_arrays(
+                [pa.array([v1], type=pa.large_binary())], names=['payload']))
+
+        padded = pa.RecordBatch.from_arrays(
+            [pa.array([v1 + b"x"], type=pa.large_binary())], names=['payload'])
+        with self.assertRaisesRegex(ValueError, "trailing bytes"):
+            writer._validate_inline_stored_fields_input(padded)
+
+        v0 = bytes([0]) + v1[1:]
+        with self.assertRaisesRegex(ValueError, r"in \[1, 2\], but found 0"):
+            writer._validate_inline_stored_fields_input(
+                pa.RecordBatch.from_arrays(
+                    [pa.array([v0], type=pa.large_binary())], names=['payload']))
+
+        v3 = bytes([3]) + v2[1:]
+        with self.assertRaisesRegex(ValueError, r"in \[1, 2\], but found 3"):
+            writer._validate_inline_stored_fields_input(
+                pa.RecordBatch.from_arrays(
+                    [pa.array([v3], type=pa.large_binary())], names=['payload']))
+
+        # write_row() keeps raw bytes from _normal_row_value(); the same
+        # validator must accept exact v1 and reject padding.
+        self.assertEqual(writer._normal_row_value("payload", v1), v1)
+        writer._validate_inline_stored_fields_input(
+            pa.RecordBatch.from_arrays(
+                [pa.array([v1], type=pa.large_binary())], names=["payload"]))
+
+        video_bytes = VideoFrameDescriptor("file:///v.mp4", 0, 10, 2).serialize()
+        writer._validate_inline_stored_fields_input(
+            pa.RecordBatch.from_arrays(
+                [pa.array([video_bytes], type=pa.large_binary())], names=["payload"]))
+        with self.assertRaisesRegex(ValueError, "serialized"):
+            writer._validate_inline_stored_fields_input(
+                pa.RecordBatch.from_arrays(
+                    [pa.array([video_bytes + b"x"], type=pa.large_binary())],
+                    names=["payload"]))
+
 
 class BlobEndToEndTest(unittest.TestCase):
     """End-to-end tests for blob functionality with schema definition, file writing, and reading."""
diff --git a/paimon-python/pypaimon/write/writer/dedicated_format_writer.py b/paimon-python/pypaimon/write/writer/dedicated_format_writer.py
index 8b2097741f21..9c80ec281aee 100644
--- a/paimon-python/pypaimon/write/writer/dedicated_format_writer.py
+++ b/paimon-python/pypaimon/write/writer/dedicated_format_writer.py
@@ -296,6 +296,7 @@ def write_row(self, row):
                     self.table.table_schema.fields,
                     self.normal_column_names,
                 ).to_batches()[0]
+                self._validate_inline_stored_fields_input(normal_data)
                 processed_normal = self._process_normal_data(normal_data)
                 if processed_normal is not None:
                     self._normal_buffer.append(processed_normal)
@@ -436,16 +437,26 @@ def _validate_inline_stored_fields_input(self, data: pa.RecordBatch):
                         "blob-descriptor-field requires blob field value to be a serialized "
                         "BlobDescriptor."
                     )
+                descriptor_bytes = bytes(value)
+                if descriptor_bytes:
+                    version = descriptor_bytes[0]
+                    if version < 1 or version > BlobDescriptor.CURRENT_VERSION:
+                        raise ValueError(
+                            f"blob-descriptor-field requires BlobDescriptor version "
+                            f"in [1, {BlobDescriptor.CURRENT_VERSION}], but found "
+                            f"{version}."
+                        )
                 try:
-                    descriptor_bytes = bytes(value)
-                    descriptor = BlobDescriptor.deserialize(descriptor_bytes)
-                    if descriptor.serialize() != descriptor_bytes:
-                        raise ValueError("Descriptor payload contains trailing bytes.")
+                    BlobDescriptor.deserialize(descriptor_bytes)
                 except Exception as e:
                     raise ValueError(
                         "blob-descriptor-field requires blob field value to be a serialized "
                         "BlobDescriptor."
                     ) from e
+                # serialize() always emits CURRENT_VERSION, so a round-trip
+                # would reject exact v1 bytes. Check exact wire length instead.
+                if BlobDescriptor.parse_if_serialized(descriptor_bytes) is None:
+                    raise ValueError("Descriptor payload contains trailing bytes.")
 
         for field_name in self.blob_view_fields:
             if field_name not in data.schema.names: