[core] Support BLOB descriptors in partial-update tables.

docs/docs/multimodal-table/blob.mdx

@@ -79,7 +79,7 @@ Paimon supports three storage modes for BLOB fields, selected via **comment dire
    This mode supports `BLOB`, `ARRAY<BLOB>`, and `MAP<K, BLOB>` fields.
 
 2. **Descriptor-only storage** (`__BLOB_DESCRIPTOR_FIELD`)
-   Only serialized `BlobDescriptor` bytes are stored inline in data files. Paimon does not write `.blob` files for these fields, and writes must provide descriptor-based input.
+   Only serialized `BlobDescriptor` bytes are stored inline in data files. Paimon does not write `.blob` files for these fields, and non-null values must provide descriptor-based input.
 
 3. **Blob view storage** (`__BLOB_VIEW_FIELD`)
    Serialized `BlobViewStruct` bytes are stored inline. The struct points to a BLOB value in an upstream table by table identifier, BLOB field, and row id. The actual blob bytes are resolved from the upstream table at read time.
@@ -500,7 +500,7 @@ CREATE TABLE descriptor_table (
 );
 ```
 
-Paimon stores only serialized `BlobDescriptor` bytes in normal data files. Reading the blob follows the descriptor URI to access bytes, and writing requires descriptor input for those fields.
+Paimon stores only serialized `BlobDescriptor` bytes in normal data files. Reading the blob follows the descriptor URI to access bytes, and non-null values require descriptor input for those fields.
 
 ## Presigned URLs for OSS Blobs
 
@@ -674,7 +674,7 @@ For the Python equivalent, see [Blob Storage in pypaimon](../pypaimon/blob).
 
 ## Limitations
 
-1. **Primary Key Tables**: Managed BLOB storage in primary-key tables has additional requirements; see [Primary-Key BLOB Storage](../primary-key-table/blob-storage).
+1. **Primary Key Tables**: Managed BLOB storage in primary-key tables has additional requirements; see [Primary-Key BLOB Storage](../primary-key-table/blob-storage). For `merge-engine=partial-update`, only scalar `__BLOB_DESCRIPTOR_FIELD` / `blob-descriptor-field` is supported; managed `blob-field` and `blob-view-field` are rejected.
 2. **No Predicate Pushdown**: Blob columns cannot be used in filter predicates.
 3. **No Statistics**: Statistics collection is not supported for blob columns.
 4. **Append Table Options**: For append tables, `row-tracking.enabled` and `data-evolution.enabled` must be set to `true`.

docs/docs/primary-key-table/blob-storage.md

@@ -101,9 +101,9 @@ multiple rows, and a row descriptor records its URI, offset, and length.
 
 :::note
 
-On append tables, `blob-descriptor-field` is descriptor-only storage and writes must provide a descriptor. Append-table
-`blob-field` storage still requires row tracking and data evolution. The managed raw-byte externalization described here
-is specific to supported primary-key tables.
+On append tables, `blob-descriptor-field` is descriptor-only storage and non-null values must provide a descriptor.
+Append-table `blob-field` storage still requires row tracking and data evolution. The managed raw-byte externalization
+described here is specific to supported primary-key tables.
 
 :::
 
@@ -122,7 +122,35 @@ Primary-key managed BLOB storage has the following requirements:
 
 `row-tracking.enabled` and `data-evolution.enabled` are not required for this primary-key mode.
 
-## Update, Delete, and Compaction
+### Partial-update with descriptor fields
+
+Primary-key tables may use `merge-engine=partial-update` with **scalar** `blob-descriptor-field` only:
+
+| Allowed | Not allowed |
+|---------|-------------|
+| Scalar `blob-descriptor-field` | Managed `blob-field` |
+| Sequence groups over normal columns + descriptor columns | `blob-view-field` |
+| | `ARRAY` / `MAP` descriptor or view fields |
+
+```sql
+CREATE TABLE media_meta (
+    id BIGINT,
+    name STRING,
+    content BYTES COMMENT '__BLOB_DESCRIPTOR_FIELD; external video',
+    ts INT,
+    PRIMARY KEY (id) NOT ENFORCED
+) WITH (
+    'merge-engine' = 'partial-update',
+    'fields.ts.sequence-group' = 'name,content'
+);
+```
+
+Non-null values must be descriptors (for example Flink `sys.path_to_descriptor`). Outside a sequence group, a null
+value does not update the field. Within a sequence group, a null sequence value skips the entire group. When the
+incoming sequence is newer or equal, every field in the group is replaced, including replacing a descriptor with
+null. When the sequence is older, no field in the group is updated.
+
+## Managed BLOB Update, Delete, and Compaction
 
 An update writes a new descriptor and managed payload when a scalar value, array element, or map value changes. A delete
 record does not write a new payload. Deduplication determines the final rows of each data file, and its `.blobref`

paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java

@@ -220,6 +220,8 @@ public static void validateTableSchema(TableSchema schema, Set<String> dynamicOp
         Set<String> blobDescriptorFields = validateBlobDescriptorFields(tableRowType, options);
         Set<String> blobViewFields =
                 validateBlobViewFields(tableRowType, options, blobDescriptorFields);
+        validatePrimaryKeyBlobKeyConfiguration(schema, options);
+        validatePartialUpdateBlobConfiguration(schema, options);
         validatePrimaryKeyBlobConfiguration(schema, options);
         Set<String> blobInlineFields = new HashSet<>(blobDescriptorFields);
         blobInlineFields.addAll(blobViewFields);
@@ -1427,6 +1429,34 @@ private static void validatePrimaryKeyBlobConfiguration(
             return;
         }
 
+        checkArgument(
+                options.mergeEngine() == MergeEngine.DEDUPLICATE,
+                "Primary-key managed BLOB tables only support the deduplicate merge engine.");
+        checkArgument(
+                options.changelogProducer() == ChangelogProducer.NONE,
+                "Primary-key managed BLOB tables only support changelog-producer 'none'.");
+        checkArgument(
+                options.dataFileExternalPaths() == null,
+                "Primary-key managed BLOB tables do not support '%s'.",
+                CoreOptions.DATA_FILE_EXTERNAL_PATHS.key());
+        checkArgument(
+                !options.pkClusteringOverride(),
+                "Primary-key managed BLOB tables do not support '%s'.",
+                CoreOptions.PK_CLUSTERING_OVERRIDE.key());
+    }
+
+    private static void validatePrimaryKeyBlobKeyConfiguration(
+            TableSchema schema, CoreOptions options) {
+        if (schema.primaryKeys().isEmpty()) {
+            return;
+        }
+
+        Set<String> managedBlobFields =
+                fieldNamesInBlobFile(new RowType(schema.fields()), options.blobInlineField());
+        if (managedBlobFields.isEmpty()) {
+            return;
+        }
+
         List<String> primaryKeyBlobFields =
                 managedBlobFields.stream()
                         .filter(schema.primaryKeys()::contains)
@@ -1453,21 +1483,33 @@ private static void validatePrimaryKeyBlobConfiguration(
                 sequenceBlobFields.isEmpty(),
                 "Managed BLOB fields cannot be sequence fields: %s.",
                 sequenceBlobFields);
+    }
 
+    /**
+     * Partial-update only supports scalar {@link CoreOptions#BLOB_DESCRIPTOR_FIELD}. Managed {@link
+     * CoreOptions#BLOB_FIELD} and {@link CoreOptions#BLOB_VIEW_FIELD} are rejected.
+     */
+    private static void validatePartialUpdateBlobConfiguration(
+            TableSchema schema, CoreOptions options) {
+        if (options.mergeEngine() != MergeEngine.PARTIAL_UPDATE) {
+            return;
+        }
+        if (schema.primaryKeys().isEmpty()) {
+            return;
+        }
+
+        Set<String> managedBlobFields =
+                fieldNamesInBlobFile(new RowType(schema.fields()), options.blobInlineField());
         checkArgument(
-                options.mergeEngine() == MergeEngine.DEDUPLICATE,
-                "Primary-key managed BLOB tables only support the deduplicate merge engine.");
-        checkArgument(
-                options.changelogProducer() == ChangelogProducer.NONE,
-                "Primary-key managed BLOB tables only support changelog-producer 'none'.");
-        checkArgument(
-                options.dataFileExternalPaths() == null,
-                "Primary-key managed BLOB tables do not support '%s'.",
-                CoreOptions.DATA_FILE_EXTERNAL_PATHS.key());
+                managedBlobFields.isEmpty(),
+                "Partial-update does not support managed BLOB fields (blob-field): %s. "
+                        + "Only scalar blob-descriptor-field is supported.",
+                managedBlobFields);
+
         checkArgument(
-                !options.pkClusteringOverride(),
-                "Primary-key managed BLOB tables do not support '%s'.",
-                CoreOptions.PK_CLUSTERING_OVERRIDE.key());
+                options.blobViewField().isEmpty(),
+                "Partial-update does not support blob-view-field. "
+                        + "Only scalar blob-descriptor-field is supported.");
     }
 
     private static void validateIncrementalClustering(TableSchema schema, CoreOptions options) {

paimon-core/src/test/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunctionTest.java

@@ -19,6 +19,7 @@
 package org.apache.paimon.mergetree.compact;
 
 import org.apache.paimon.KeyValue;
+import org.apache.paimon.data.Blob;
 import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.types.DataType;
@@ -96,6 +97,42 @@ public void testSequenceGroup() {
         validate(func, 1, null, null, 6, null, null, 6);
     }
 
+    @Test
+    public void testSequenceGroupWithBlobField() {
+        Options options = new Options();
+        options.set("fields.f3.sequence-group", "f1,f2");
+        RowType rowType =
+                RowType.of(DataTypes.INT(), DataTypes.INT(), DataTypes.BLOB(), DataTypes.INT());
+        MergeFunction<KeyValue> func =
+                PartialUpdateMergeFunction.factory(options, rowType, ImmutableList.of("f0"))
+                        .create();
+        func.reset();
+
+        Blob first = Blob.fromData(new byte[] {1, 2, 3});
+        Blob second = Blob.fromData(new byte[] {4, 5, 6});
+        Blob third = Blob.fromData(new byte[] {7, 8, 9});
+
+        addBlobRow(func, RowKind.INSERT, 1, 10, first, 1);
+        addBlobRow(func, RowKind.INSERT, 1, 20, second, null);
+        // null sequence group should not overwrite f1/f2
+        assertBlobRow(func, 1, 10, first, 1);
+
+        addBlobRow(func, RowKind.INSERT, 1, 30, third, 2);
+        assertBlobRow(func, 1, 30, third, 2);
+
+        // equal sequence should overwrite the entire group
+        addBlobRow(func, RowKind.INSERT, 1, 40, second, 2);
+        assertBlobRow(func, 1, 40, second, 2);
+
+        // valid sequence should overwrite the entire group, including with null
+        addBlobRow(func, RowKind.INSERT, 1, 50, null, 3);
+        assertBlobRow(func, 1, 50, null, 3);
+
+        // older sequence should not overwrite
+        addBlobRow(func, RowKind.INSERT, 1, 60, first, 2);
+        assertBlobRow(func, 1, 50, null, 3);
+    }
+
     @Test
     public void testSequenceGroupPartialDelete() {
         Options options = new Options();
@@ -1007,6 +1044,28 @@ private void add(MergeFunction<KeyValue> function, RowKind rowKind, Integer... f
                 new KeyValue().replace(GenericRow.of(1), sequence++, rowKind, GenericRow.of(f)));
     }
 
+    private void addBlobRow(
+            MergeFunction<KeyValue> function,
+            RowKind rowKind,
+            Integer pk,
+            Integer name,
+            Blob payload,
+            Integer ts) {
+        function.add(
+                new KeyValue()
+                        .replace(
+                                GenericRow.of(pk),
+                                sequence++,
+                                rowKind,
+                                GenericRow.of(pk, name, payload, ts)));
+    }
+
+    private void assertBlobRow(
+            MergeFunction<KeyValue> function, Integer pk, Integer name, Blob payload, Integer ts) {
+        GenericRow expected = GenericRow.of(pk, name, payload, ts);
+        assertThat(function.getResult().value()).isEqualTo(expected);
+    }
+
     private void validate(MergeFunction<KeyValue> function, Integer... f) {
         assertThat(function.getResult().value()).isEqualTo(GenericRow.of(f));
     }

paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java

@@ -408,6 +408,7 @@ public void testPrimaryKeyBlobFileField() {
 
     @Test
     public void testPrimaryKeyInlineBlobDoesNotTriggerManagedRestrictions() {
+        // Scheme B: partial-update allows scalar blob-descriptor-field only.
         List<DataField> fields =
                 Arrays.asList(
                         new DataField(0, "id", DataTypes.INT()),
@@ -423,6 +424,44 @@ public void testPrimaryKeyInlineBlobDoesNotTriggerManagedRestrictions() {
         assertThatCode(() -> validateTableSchema(schema)).doesNotThrowAnyException();
     }
 
+    @Test
+    public void testPartialUpdateRejectsBlobViewField() {
+        List<DataField> fields =
+                Arrays.asList(
+                        new DataField(0, "id", DataTypes.INT()),
+                        new DataField(1, "view", DataTypes.BLOB()));
+        Map<String, String> options = new HashMap<>();
+        options.put(BUCKET.key(), "1");
+        options.put(CoreOptions.BLOB_VIEW_FIELD.key(), "view");
+        options.put(CoreOptions.MERGE_ENGINE.key(), "partial-update");
+
+        TableSchema schema =
+                new TableSchema(1, fields, 10, emptyList(), singletonList("id"), options, "");
+
+        assertThatThrownBy(() -> validateTableSchema(schema))
+                .hasMessageContaining("Partial-update does not support blob-view-field");
+    }
+
+    @Test
+    public void testPartialUpdateRejectsDescriptorWithBlobViewField() {
+        List<DataField> fields =
+                Arrays.asList(
+                        new DataField(0, "id", DataTypes.INT()),
+                        new DataField(1, "payload", DataTypes.BLOB()),
+                        new DataField(2, "view", DataTypes.BLOB()));
+        Map<String, String> options = new HashMap<>();
+        options.put(BUCKET.key(), "1");
+        options.put(CoreOptions.BLOB_DESCRIPTOR_FIELD.key(), "payload");
+        options.put(CoreOptions.BLOB_VIEW_FIELD.key(), "view");
+        options.put(CoreOptions.MERGE_ENGINE.key(), "partial-update");
+
+        TableSchema schema =
+                new TableSchema(1, fields, 10, emptyList(), singletonList("id"), options, "");
+
+        assertThatThrownBy(() -> validateTableSchema(schema))
+                .hasMessageContaining("Partial-update does not support blob-view-field");
+    }
+
     @Test
     public void testPrimaryKeyBlobViewCoexistsWithManagedBlob() {
         List<DataField> fields =
@@ -465,13 +504,34 @@ sequenceOptions, singletonList("id"), emptyList())))
 
         Map<String, String> mergeOptions = new HashMap<>(options);
         mergeOptions.put(CoreOptions.MERGE_ENGINE.key(), "partial-update");
+        assertThatThrownBy(
+                        () ->
+                                validateTableSchema(
+                                        primaryKeyBlobSchema(
+                                                mergeOptions,
+                                                singletonList("payload"),
+                                                emptyList())))
+                .hasMessage("Managed BLOB fields cannot be primary keys: [payload].");
+
+        Map<String, String> mergeSequenceOptions = new HashMap<>(mergeOptions);
+        mergeSequenceOptions.put(CoreOptions.SEQUENCE_FIELD.key(), "payload");
+        assertThatThrownBy(
+                        () ->
+                                validateTableSchema(
+                                        primaryKeyBlobSchema(
+                                                mergeSequenceOptions,
+                                                singletonList("id"),
+                                                emptyList())))
+                .hasMessage("Managed BLOB fields cannot be sequence fields: [payload].");
+
         assertThatThrownBy(
                         () ->
                                 validateTableSchema(
                                         primaryKeyBlobSchema(
                                                 mergeOptions, singletonList("id"), emptyList())))
                 .hasMessage(
-                        "Primary-key managed BLOB tables only support the deduplicate merge engine.");
+                        "Partial-update does not support managed BLOB fields (blob-field): "
+                                + "[payload]. Only scalar blob-descriptor-field is supported.");
 
         Map<String, String> changelogOptions = new HashMap<>(options);
         changelogOptions.put(CoreOptions.CHANGELOG_PRODUCER.key(), "input");