From 7fcf861787f916c06b11ad85f6235f6b51f0ba49 Mon Sep 17 00:00:00 2001 From: "wenchao.wu" Date: Thu, 30 Jul 2026 15:24:58 +0800 Subject: [PATCH] [core] Support BLOB descriptors in partial-update tables. --- docs/docs/multimodal-table/blob.mdx | 6 +- docs/docs/primary-key-table/blob-storage.md | 36 +++- .../paimon/schema/SchemaValidation.java | 66 ++++++-- .../PartialUpdateMergeFunctionTest.java | 59 +++++++ .../paimon/schema/SchemaValidationTest.java | 62 ++++++- .../paimon/flink/PartialUpdateITCase.java | 154 ++++++++++++++++++ 6 files changed, 363 insertions(+), 20 deletions(-) diff --git a/docs/docs/multimodal-table/blob.mdx b/docs/docs/multimodal-table/blob.mdx index 7d3f14d8c388..ce7fa9488a77 100644 --- a/docs/docs/multimodal-table/blob.mdx +++ b/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`, and `MAP` 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`. diff --git a/docs/docs/primary-key-table/blob-storage.md b/docs/docs/primary-key-table/blob-storage.md index 51d7fc27562d..2468ee8f4012 100644 --- a/docs/docs/primary-key-table/blob-storage.md +++ b/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` diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java index cdd2555c361c..c28a673a14ee 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java @@ -220,6 +220,8 @@ public static void validateTableSchema(TableSchema schema, Set dynamicOp Set blobDescriptorFields = validateBlobDescriptorFields(tableRowType, options); Set blobViewFields = validateBlobViewFields(tableRowType, options, blobDescriptorFields); + validatePrimaryKeyBlobKeyConfiguration(schema, options); + validatePartialUpdateBlobConfiguration(schema, options); validatePrimaryKeyBlobConfiguration(schema, options); Set 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 managedBlobFields = + fieldNamesInBlobFile(new RowType(schema.fields()), options.blobInlineField()); + if (managedBlobFields.isEmpty()) { + return; + } + List 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 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) { diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunctionTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunctionTest.java index 050f4b855b39..9de9d2295731 100644 --- a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunctionTest.java +++ b/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 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 function, RowKind rowKind, Integer... f new KeyValue().replace(GenericRow.of(1), sequence++, rowKind, GenericRow.of(f))); } + private void addBlobRow( + MergeFunction 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 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 function, Integer... f) { assertThat(function.getResult().value()).isEqualTo(GenericRow.of(f)); } diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java index 36a085c6a819..3489741ed4cd 100644 --- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java +++ b/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 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 fields = + Arrays.asList( + new DataField(0, "id", DataTypes.INT()), + new DataField(1, "view", DataTypes.BLOB())); + Map 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 fields = + Arrays.asList( + new DataField(0, "id", DataTypes.INT()), + new DataField(1, "payload", DataTypes.BLOB()), + new DataField(2, "view", DataTypes.BLOB())); + Map 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 fields = @@ -465,13 +504,34 @@ sequenceOptions, singletonList("id"), emptyList()))) Map 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 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 changelogOptions = new HashMap<>(options); changelogOptions.put(CoreOptions.CHANGELOG_PRODUCER.key(), "input"); diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java index c1fcc2ca0914..a691af65d272 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java @@ -18,11 +18,16 @@ package org.apache.paimon.flink; +import org.apache.paimon.Snapshot; +import org.apache.paimon.data.BlobDescriptor; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.utils.BlockingIterator; import org.apache.paimon.utils.CommonTestUtils; import org.apache.flink.configuration.RestartStrategyOptions; import org.apache.flink.table.api.config.ExecutionConfigOptions; +import org.apache.flink.table.api.config.TableConfigOptions; import org.apache.flink.table.planner.factories.TestValuesTableFactory; import org.apache.flink.types.Row; import org.apache.flink.types.RowKind; @@ -32,6 +37,7 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; +import java.io.OutputStream; import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; @@ -772,4 +778,152 @@ public void testSequenceGroupWithDefaultAgg() { assertThat(sql("SELECT * FROM seq_default_agg")).containsExactly(Row.of(0, 2, 3)); } + + @Test + public void testBlobDescriptorPartialUpdate() throws Exception { + byte[] first = "video-v1".getBytes(); + byte[] second = "video-v2".getBytes(); + String firstUri = writeExternalBlob("pu_blob_v1", first); + String secondUri = writeExternalBlob("pu_blob_v2", second); + + sql( + "CREATE TABLE pu_blob (" + + " id INT PRIMARY KEY NOT ENFORCED," + + " name STRING," + + " payload BYTES" + + ") WITH (" + + " 'merge-engine'='partial-update'," + + " 'blob-descriptor-field'='payload'," + + " 'bucket'='1'," + + " 'write-only'='true'," + + " 'num-sorted-run.compaction-trigger'='100'" + + ")"); + + sql("INSERT INTO pu_blob VALUES (1, 'a', sys.path_to_descriptor('" + firstUri + "'))"); + sql("INSERT INTO pu_blob VALUES (1, 'b', CAST(NULL AS BYTES))"); + assertThat(sql("SELECT id, name, payload FROM pu_blob")) + .containsExactly(Row.of(1, "b", first)); + + sql( + "INSERT INTO pu_blob VALUES (1, CAST(NULL AS STRING), sys.path_to_descriptor('" + + secondUri + + "'))"); + assertThat(sql("SELECT id, name, payload FROM pu_blob")) + .containsExactly(Row.of(1, "b", second)); + + assertThat((long) sql("SELECT COUNT(*) FROM `pu_blob$files`").get(0).getField(0)) + .isGreaterThan(1); + tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true); + sql("CALL sys.compact(`table` => 'default.pu_blob', compact_strategy => 'full')"); + Snapshot snapshot = findLatestSnapshot("pu_blob"); + assertThat(snapshot).isNotNull(); + assertThat(snapshot.commitKind()).isEqualTo(Snapshot.CommitKind.COMPACT); + assertThat(sql("SELECT COUNT(*) FROM `pu_blob$files`")).containsExactly(Row.of(1L)); + assertThat(sql("SELECT id, name, payload FROM pu_blob")) + .containsExactly(Row.of(1, "b", second)); + + sql("ALTER TABLE pu_blob SET ('blob-as-descriptor'='true')"); + byte[] descriptorBytes = (byte[]) sql("SELECT payload FROM pu_blob").get(0).getField(0); + BlobDescriptor descriptor = BlobDescriptor.deserialize(descriptorBytes); + assertThat(descriptor.uri()).isEqualTo(secondUri); + assertThat(descriptor.offset()).isEqualTo(0); + // path_to_descriptor may encode whole-file length as -1. + assertThat(descriptor.length()).isIn(-1L, (long) second.length); + } + + @Test + public void testBlobDescriptorPartialUpdateSequenceGroup() throws Exception { + byte[] first = "seq-v1".getBytes(); + byte[] second = "seq-v2".getBytes(); + String firstUri = writeExternalBlob("pu_blob_seq_v1", first); + String secondUri = writeExternalBlob("pu_blob_seq_v2", second); + + sql( + "CREATE TABLE pu_blob_seq (" + + " id INT PRIMARY KEY NOT ENFORCED," + + " name STRING," + + " payload BYTES," + + " ts INT" + + ") WITH (" + + " 'merge-engine'='partial-update'," + + " 'blob-descriptor-field'='payload'," + + " 'fields.ts.sequence-group'='name,payload'," + + " 'bucket'='1'," + + " 'write-only'='true'," + + " 'num-sorted-run.compaction-trigger'='100'" + + ")"); + + sql( + "INSERT INTO pu_blob_seq VALUES (1, 'a', sys.path_to_descriptor('" + + firstUri + + "'), 1)"); + // null sequence should not overwrite name/payload + sql( + "INSERT INTO pu_blob_seq VALUES (1, 'b', sys.path_to_descriptor('" + + secondUri + + "'), CAST(NULL AS INT))"); + assertThat(sql("SELECT id, name, payload, ts FROM pu_blob_seq")) + .containsExactly(Row.of(1, "a", first, 1)); + + sql( + "INSERT INTO pu_blob_seq VALUES (1, 'c', sys.path_to_descriptor('" + + secondUri + + "'), 2)"); + assertThat(sql("SELECT id, name, payload, ts FROM pu_blob_seq")) + .containsExactly(Row.of(1, "c", second, 2)); + + // equal sequence overwrites the entire group, including with null + sql("INSERT INTO pu_blob_seq VALUES " + "(1, 'd', CAST(NULL AS BYTES), 2)"); + assertThat(sql("SELECT id, name, payload, ts FROM pu_blob_seq")) + .containsExactly(Row.of(1, "d", null, 2)); + + // older sequence must not overwrite + sql( + "INSERT INTO pu_blob_seq VALUES (1, 'e', sys.path_to_descriptor('" + + secondUri + + "'), 1)"); + assertThat(sql("SELECT id, name, payload, ts FROM pu_blob_seq")) + .containsExactly(Row.of(1, "d", null, 2)); + + assertThat((long) sql("SELECT COUNT(*) FROM `pu_blob_seq$files`").get(0).getField(0)) + .isGreaterThan(1); + tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true); + sql("CALL sys.compact(`table` => 'default.pu_blob_seq', compact_strategy => 'full')"); + Snapshot snapshot = findLatestSnapshot("pu_blob_seq"); + assertThat(snapshot).isNotNull(); + assertThat(snapshot.commitKind()).isEqualTo(Snapshot.CommitKind.COMPACT); + assertThat(sql("SELECT COUNT(*) FROM `pu_blob_seq$files`")).containsExactly(Row.of(1L)); + assertThat(sql("SELECT id, name, payload, ts FROM pu_blob_seq")) + .containsExactly(Row.of(1, "d", null, 2)); + + sql("ALTER TABLE pu_blob_seq SET ('blob-as-descriptor'='true')"); + assertThat(sql("SELECT payload FROM pu_blob_seq")).containsExactly(Row.of((Object) null)); + } + + @Test + public void testPartialUpdateRejectsBlobViewField() { + assertThatThrownBy( + () -> + sql( + "CREATE TABLE pu_blob_view (" + + " id INT PRIMARY KEY NOT ENFORCED," + + " payload BYTES" + + ") WITH (" + + " 'merge-engine'='partial-update'," + + " 'blob-view-field'='payload'" + + ")")) + .hasRootCauseMessage( + "Partial-update does not support blob-view-field. " + + "Only scalar blob-descriptor-field is supported."); + } + + private String writeExternalBlob(String name, byte[] data) throws Exception { + FileIO fileIO = new LocalFileIO(); + String uri = "file://" + path + "/" + name; + try (OutputStream outputStream = + fileIO.newOutputStream(new org.apache.paimon.fs.Path(uri), true)) { + outputStream.write(data); + } + return uri; + } }