diff --git a/docs/docs/multimodal-table/blob.mdx b/docs/docs/multimodal-table/blob.mdx index 7d3f14d8c388..44d9edd9b393 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). Primary-key tables support `merge-engine=partial-update` with scalar `blob-descriptor-field`, managed `blob-field` (`changelog-producer=none`), and scalar `blob-view-field`; `ARRAY` / `MAP` descriptor or view fields are not supported. 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..544b8cc3e2da 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. ::: @@ -114,7 +114,7 @@ Primary-key managed BLOB storage has the following requirements: | Item | Requirement | |------|-------------| | Managed BLOB declaration | `BLOB`, `ARRAY`, and `MAP` use `blob-field`; `blob-descriptor-field` remains inline | -| Merge engine | `deduplicate` only | +| Merge engine | `deduplicate` or `partial-update` | | Changelog producer | `none` only | | Key usage | A managed BLOB column cannot be a primary, partition, bucket, or sequence key | | External data paths | `data-file.external-paths` is not supported | @@ -122,11 +122,68 @@ 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 BLOB fields -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` -sidecar contains only the managed packs referenced by those rows. +Primary-key tables may use `merge-engine=partial-update` with scalar `blob-descriptor-field`, +managed `blob-field` (`BLOB`, `ARRAY`, or `MAP`), and scalar +`blob-view-field`: + +| Mode | Partial-update | Changelog | Notes | +|------|----------------|-----------|-------| +| Scalar `blob-descriptor-field` | Supported | Same as other PU tables | Inline descriptor bytes | +| `blob-field` (managed scalar, array, or map) | Supported | `changelog-producer=none` only | Payload in `.managed.blob` | +| Scalar `blob-view-field` | Supported | Same as other PU tables | Resolve on read via catalog | +| `ARRAY` / `MAP` descriptor or view | Not supported | — | Scalar only | + +```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' +); +``` + +For managed BLOB columns, set `changelog-producer=none` (same as deduplicate managed BLOB tables). + +```sql +CREATE TABLE training_chunks ( + id BIGINT, + name STRING, + chunk BYTES COMMENT '__BLOB_FIELD; raw training block', + PRIMARY KEY (id) NOT ENFORCED +) WITH ( + 'merge-engine' = 'partial-update', + 'changelog-producer' = 'none', + 'blob-field' = 'chunk' +); +``` + +Non-null values update the corresponding column; null values do not update the field (standard partial-update +semantics). 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 BLOB value with null. When the sequence +is older, no field in the group is updated. + +Managed BLOB partial updates externalize each non-null scalar BLOB, array element, or map value into a +`.managed.blob` pack. Empty collections and collections containing only null values write no payload. BLOB garbage +collection for orphaned packs is not implemented yet; repeated updates can leave unreachable storage until a future +collector is available. + +`blob-view-field` columns store serialized view structs inline. Reads resolve upstream blob bytes through the catalog +when `blob-view.resolve.enabled` is true (default). Append upstream tables used by `sys.blob_view(...)` must enable +`row-tracking.enabled` and `data-evolution.enabled`. + +## Managed BLOB Update, Delete, and Compaction + +Each incoming non-null scalar BLOB, array element, or map value is written as a new descriptor and payload before merge +rules are applied. An update later discarded by partial-update or sequence rules can therefore leave an orphaned +payload pack. A delete record does not write a new payload. The merge engine determines the logical final row during +reads and compaction. Each data file's `.blobref` sidecar records managed packs referenced by non-retract key-values in +that file, which may include intermediate partial-update payloads before compaction. Compaction preserves descriptors for surviving values and creates new `.blobref` sidecars from the compacted output. It does not copy the referenced payload bytes into new `.managed.blob` packs. This keeps ordinary compaction cost diff --git a/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceCollector.java b/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceCollector.java index 529b077888e7..841d3d1eedc6 100644 --- a/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceCollector.java +++ b/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceCollector.java @@ -37,7 +37,7 @@ import static org.apache.paimon.types.BlobType.isBlobFileField; import static org.apache.paimon.utils.Preconditions.checkState; -/** Collects exact managed BLOB dependencies from the final rows of one data file. */ +/** Collects managed BLOB dependencies from non-retract key-values written to one data file. */ public class ManagedBlobReferenceCollector { private final FileIO fileIO; 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..6e7b4ff7e6b2 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,7 @@ public static void validateTableSchema(TableSchema schema, Set dynamicOp Set blobDescriptorFields = validateBlobDescriptorFields(tableRowType, options); Set blobViewFields = validateBlobViewFields(tableRowType, options, blobDescriptorFields); + validatePrimaryKeyBlobKeyConfiguration(schema, options); validatePrimaryKeyBlobConfiguration(schema, options); Set blobInlineFields = new HashSet<>(blobDescriptorFields); blobInlineFields.addAll(blobViewFields); @@ -1427,6 +1428,36 @@ private static void validatePrimaryKeyBlobConfiguration( return; } + checkArgument( + options.mergeEngine() == MergeEngine.DEDUPLICATE + || options.mergeEngine() == MergeEngine.PARTIAL_UPDATE, + "Primary-key managed BLOB tables only support the deduplicate or " + + "partial-update 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 +1484,6 @@ private static void validatePrimaryKeyBlobConfiguration( sequenceBlobFields.isEmpty(), "Managed BLOB fields cannot be sequence fields: %s.", sequenceBlobFields); - - 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 validateIncrementalClustering(TableSchema schema, CoreOptions options) { diff --git a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java index 3030b3504e1d..f73b317c439d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java @@ -149,7 +149,10 @@ protected BiConsumer nonPartitionFilterConsumer() { @Override public InnerTableRead newRead() { return new KeyValueTableRead( - () -> store().newRead(), () -> store().newBatchRawFileRead(), schema()); + () -> store().newRead(), + () -> store().newBatchRawFileRead(), + schema(), + catalogEnvironment.dependencyReadContext()); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableRead.java index f349047fe8ea..c46e1f299f64 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableRead.java @@ -103,6 +103,10 @@ protected Predicate predicate() { return predicate; } + protected boolean shouldExecuteFilter() { + return executeFilter; + } + @Override public RecordReader createReader(Split split) throws IOException { QueryAuthContext queryAuthContext = unwrapQueryAuthSplit(split); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/BlobViewTableReadSupport.java b/paimon-core/src/main/java/org/apache/paimon/table/source/BlobViewTableReadSupport.java new file mode 100644 index 000000000000..6602d40a6ebf --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/BlobViewTableReadSupport.java @@ -0,0 +1,142 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table.source; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.catalog.CatalogContext; +import org.apache.paimon.catalog.TableQueryAuthResult; +import org.apache.paimon.data.Blob; +import org.apache.paimon.data.BlobView; +import org.apache.paimon.data.BlobViewResolver; +import org.apache.paimon.data.BlobViewStruct; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.TopN; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.types.DataTypeRoot; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.BlobViewLookup; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.LinkedHashSet; +import java.util.Set; +import java.util.function.Supplier; + +/** Shared helpers for resolving {@link org.apache.paimon.CoreOptions#BLOB_VIEW_FIELD} on read. */ +final class BlobViewTableReadSupport { + + private BlobViewTableReadSupport() {} + + static int[] blobViewFieldIndexes(RowType rowType, CoreOptions options) { + if (!options.blobViewResolveEnabled()) { + return new int[0]; + } + + Set blobViewFieldNames = options.blobViewField(); + if (blobViewFieldNames.isEmpty()) { + return new int[0]; + } + + return rowType.getFields().stream() + .filter( + field -> + field.type().is(DataTypeRoot.BLOB) + && blobViewFieldNames.contains(field.name())) + .mapToInt(field -> rowType.getFieldIndex(field.name())) + .toArray(); + } + + static RecordReader createBlobViewReader( + CatalogContext catalogContext, + Split split, + @Nullable TableQueryAuthResult authResult, + int[] blobViewFields, + RowType readType, + @Nullable Predicate predicate, + @Nullable TopN topN, + @Nullable Integer limit, + boolean executeFilter, + RecordReaderSupplier dataReaderSupplier, + Supplier prescanReadSupplier) + throws IOException { + RowType prescanType = executeFilter ? readType : readType.project(blobViewFields); + int[] prescanBlobViewFields = new int[blobViewFields.length]; + if (executeFilter) { + System.arraycopy(blobViewFields, 0, prescanBlobViewFields, 0, blobViewFields.length); + } else { + for (int i = 0; i < prescanBlobViewFields.length; i++) { + prescanBlobViewFields[i] = i; + } + } + InnerTableRead prescanRead = prescanReadSupplier.get(); + prescanRead.withReadType(prescanType); + if (predicate != null) { + prescanRead.withFilter(predicate); + } + if (topN != null) { + prescanRead.withTopN(topN); + } + if (limit != null) { + prescanRead.withLimit(limit); + } + + Split prescanSplit = authResult != null ? new QueryAuthSplit(split, authResult) : split; + LinkedHashSet viewStructs = new LinkedHashSet<>(); + RecordReader prescanReader = prescanRead.createReader(prescanSplit); + try { + prescanReader.forEachRemaining( + row -> { + for (int field : prescanBlobViewFields) { + if (row.isNullAt(field)) { + continue; + } + Blob blob = row.getBlob(field); + if (!(blob instanceof BlobView)) { + throw new IllegalArgumentException( + "blob-view-field requires blob field value to be a " + + "serialized BlobViewStruct."); + } + viewStructs.add(((BlobView) blob).viewStruct()); + } + }); + } finally { + prescanReader.close(); + } + + BlobViewResolver resolver = + BlobViewLookup.createResolver(catalogContext, new ArrayList<>(viewStructs)); + + RecordReader reader = dataReaderSupplier.get(); + Set blobViewFieldSet = new HashSet<>(); + for (int field : blobViewFields) { + blobViewFieldSet.add(field); + } + return reader.transform(row -> new BlobViewResolvingRow(row, blobViewFieldSet, resolver)); + } + + @FunctionalInterface + interface RecordReaderSupplier { + + RecordReader get() throws IOException; + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionTableRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionTableRead.java index bf41ff70b62f..60bd2b99fe83 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionTableRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionTableRead.java @@ -20,29 +20,16 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.catalog.CatalogContext; -import org.apache.paimon.catalog.TableQueryAuthResult; -import org.apache.paimon.data.Blob; -import org.apache.paimon.data.BlobView; -import org.apache.paimon.data.BlobViewResolver; -import org.apache.paimon.data.BlobViewStruct; import org.apache.paimon.data.InternalRow; -import org.apache.paimon.predicate.Predicate; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.source.splitread.SplitReadConfig; import org.apache.paimon.table.source.splitread.SplitReadProvider; -import org.apache.paimon.types.DataTypeRoot; -import org.apache.paimon.types.RowType; -import org.apache.paimon.utils.BlobViewLookup; import javax.annotation.Nullable; import java.io.IOException; -import java.util.ArrayList; -import java.util.HashSet; -import java.util.LinkedHashSet; import java.util.List; -import java.util.Set; import java.util.function.Function; import java.util.function.Supplier; @@ -65,91 +52,33 @@ public DataEvolutionTableRead( @Override public RecordReader createReader(Split split) throws IOException { QueryAuthContext queryAuthContext = unwrapQueryAuthSplit(split); - if (catalogContext != null) { - int[] blobViewFields = blobViewFields(currentReadType()); - if (blobViewFields.length > 0) { - if (readFactory == null) { - throw new IllegalStateException( - "Cannot read blob-view-field fields without a readFactory."); - } - return createBlobViewReader( - queryAuthContext.split(), queryAuthContext.authResult(), blobViewFields); - } - } - return createDataReader(queryAuthContext.split(), queryAuthContext.authResult()); - } - - private int[] blobViewFields(RowType rowType) { CoreOptions options = CoreOptions.fromMap(schema().options()); - if (!options.blobViewResolveEnabled()) { - return new int[0]; - } - - Set blobViewFieldNames = options.blobViewField(); - if (blobViewFieldNames.isEmpty()) { - return new int[0]; - } - - return rowType.getFields().stream() - .filter( - field -> - field.type().is(DataTypeRoot.BLOB) - && blobViewFieldNames.contains(field.name())) - .mapToInt(field -> rowType.getFieldIndex(field.name())) - .toArray(); - } - - private RecordReader createBlobViewReader( - Split split, @Nullable TableQueryAuthResult authResult, int[] blobViewFields) - throws IOException { - RowType blobViewOnlyType = currentReadType().project(blobViewFields); - InnerTableRead prescanRead = readFactory.get(); - prescanRead.withReadType(blobViewOnlyType); - Predicate predicate = predicate(); - if (predicate != null) { - prescanRead.withFilter(predicate); - } - configureBlobViewPrescanRead(prescanRead); - Split prescanSplit = authResult != null ? new QueryAuthSplit(split, authResult) : split; - LinkedHashSet viewStructs = new LinkedHashSet<>(); - RecordReader prescanReader = prescanRead.createReader(prescanSplit); - try { - prescanReader.forEachRemaining( - row -> { - for (int i = 0; i < blobViewFields.length; i++) { - if (row.isNullAt(i)) { - continue; - } - Blob blob = row.getBlob(i); - if (!(blob instanceof BlobView)) { - throw new IllegalArgumentException( - "blob-view-field requires blob field value to be a " - + "serialized BlobViewStruct."); - } - viewStructs.add(((BlobView) blob).viewStruct()); + int[] blobViewFields = + BlobViewTableReadSupport.blobViewFieldIndexes(currentReadType(), options); + if (catalogContext != null && blobViewFields.length > 0) { + if (readFactory == null) { + throw new IllegalStateException( + "Cannot read blob-view-field fields without a readFactory."); + } + return BlobViewTableReadSupport.createBlobViewReader( + catalogContext, + queryAuthContext.split(), + queryAuthContext.authResult(), + blobViewFields, + currentReadType(), + predicate(), + topN, + limit, + shouldExecuteFilter(), + () -> createDataReader(queryAuthContext.split(), queryAuthContext.authResult()), + () -> { + InnerTableRead prescanRead = readFactory.get(); + if (shouldExecuteFilter()) { + prescanRead.executeFilter(); } + return prescanRead; }); - } finally { - prescanReader.close(); - } - - BlobViewResolver resolver = - BlobViewLookup.createResolver(catalogContext, new ArrayList<>(viewStructs)); - - RecordReader reader = createDataReader(split, authResult); - Set blobViewFieldSet = new HashSet<>(); - for (int field : blobViewFields) { - blobViewFieldSet.add(field); - } - return reader.transform(row -> new BlobViewResolvingRow(row, blobViewFieldSet, resolver)); - } - - private void configureBlobViewPrescanRead(InnerTableRead prescanRead) { - if (topN != null) { - prescanRead.withTopN(topN); - } - if (limit != null) { - prescanRead.withLimit(limit); } + return createDataReader(queryAuthContext.split(), queryAuthContext.authResult()); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java index 0b19df5c0c7b..9c02bb470f66 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java @@ -21,6 +21,7 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.KeyValue; import org.apache.paimon.annotation.VisibleForTesting; +import org.apache.paimon.catalog.CatalogContext; import org.apache.paimon.data.InternalRow; import org.apache.paimon.disk.IOManager; import org.apache.paimon.operation.MergeFileSplitRead; @@ -53,7 +54,10 @@ */ public final class KeyValueTableRead extends AbstractDataTableRead { + private final Supplier mergeReadSupplier; + private final Supplier batchRawReadSupplier; private final List readProviders; + @Nullable private final CatalogContext catalogContext; @Nullable private RowType readType = null; private boolean forceKeepDelete = false; @@ -66,7 +70,18 @@ public KeyValueTableRead( Supplier mergeReadSupplier, Supplier batchRawReadSupplier, TableSchema schema) { + this(mergeReadSupplier, batchRawReadSupplier, schema, null); + } + + public KeyValueTableRead( + Supplier mergeReadSupplier, + Supplier batchRawReadSupplier, + TableSchema schema, + @Nullable CatalogContext catalogContext) { super(schema); + this.mergeReadSupplier = mergeReadSupplier; + this.batchRawReadSupplier = batchRawReadSupplier; + this.catalogContext = catalogContext; this.readProviders = Arrays.asList( new PrimaryKeyIndexedSplitReadProvider(batchRawReadSupplier, this::config), @@ -140,7 +155,51 @@ public RecordReader createReader(List splits) throws IOExcep @Override public RecordReader createReader(Split split) throws IOException { - return LimitRecordReader.limit(super.createReader(split), limit); + QueryAuthContext queryAuthContext = unwrapQueryAuthSplit(split); + RecordReader reader; + if (catalogContext != null) { + CoreOptions options = CoreOptions.fromMap(schema().options()); + int[] blobViewFields = + BlobViewTableReadSupport.blobViewFieldIndexes(currentReadType(), options); + if (blobViewFields.length > 0) { + reader = + BlobViewTableReadSupport.createBlobViewReader( + catalogContext, + queryAuthContext.split(), + queryAuthContext.authResult(), + blobViewFields, + currentReadType(), + predicate(), + topN, + limit, + shouldExecuteFilter(), + () -> + createDataReader( + queryAuthContext.split(), + queryAuthContext.authResult()), + this::createBlobViewPrescanRead); + } else { + reader = createDataReader(queryAuthContext.split(), queryAuthContext.authResult()); + } + } else { + reader = createDataReader(queryAuthContext.split(), queryAuthContext.authResult()); + } + return LimitRecordReader.limit(reader, limit); + } + + private InnerTableRead createBlobViewPrescanRead() { + KeyValueTableRead read = + new KeyValueTableRead(mergeReadSupplier, batchRawReadSupplier, schema(), null); + if (ioManager != null) { + read.withIOManager(ioManager); + } + if (forceKeepDelete) { + read.forceKeepDelete(); + } + if (shouldExecuteFilter()) { + read.executeFilter(); + } + return read; } @Override diff --git a/paimon-core/src/test/java/org/apache/paimon/io/PrimaryKeyBlobFileWriterTest.java b/paimon-core/src/test/java/org/apache/paimon/io/PrimaryKeyBlobFileWriterTest.java index 22c549bf3f10..d7f4d2ee0e54 100644 --- a/paimon-core/src/test/java/org/apache/paimon/io/PrimaryKeyBlobFileWriterTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/io/PrimaryKeyBlobFileWriterTest.java @@ -22,6 +22,7 @@ import org.apache.paimon.KeyValue; import org.apache.paimon.blob.ManagedBlobReferenceFile; import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.Blob; import org.apache.paimon.data.BlobRef; import org.apache.paimon.data.GenericRow; @@ -145,4 +146,101 @@ void testAttachReferenceSidecarToDataFile() throws Exception { Path emptySidecar = pathFactory.toAlignedPath(emptyMeta.extraFiles().get(0), emptyMeta); assertThat(ManagedBlobReferenceFile.read(fileIO, emptySidecar)).isEmpty(); } + + @Test + void testPartialUpdateManagedBlobSidecarCollectsIntermediateReferences() throws Exception { + FileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + RowType keyType = + new RowType( + Collections.singletonList( + new DataField( + SpecialFields.KEY_FIELD_ID_START, + SpecialFields.KEY_FIELD_PREFIX + "id", + DataTypes.INT()))); + RowType valueType = + new RowType( + java.util.Arrays.asList( + new DataField(0, "id", DataTypes.INT()), + new DataField(1, "name", DataTypes.STRING()), + new DataField(2, "payload", DataTypes.BLOB()))); + Function pathFactories = + format -> + new FileStorePathFactory( + tablePath, + RowType.of(), + CoreOptions.PARTITION_DEFAULT_NAME.defaultValue(), + format, + CoreOptions.DATA_FILE_PREFIX.defaultValue(), + CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue(), + CoreOptions.PARTITION_GENERATE_LEGACY_NAME.defaultValue(), + CoreOptions.FILE_SUFFIX_INCLUDE_COMPRESSION.defaultValue(), + CoreOptions.FILE_COMPRESSION.defaultValue(), + null, + null, + CoreOptions.ExternalPathStrategy.NONE, + null, + false, + null); + Options options = new Options(); + options.set(CoreOptions.BLOB_FIELD, "payload"); + options.set(CoreOptions.MERGE_ENGINE, CoreOptions.MergeEngine.PARTIAL_UPDATE); + options.set(CoreOptions.CHANGELOG_PRODUCER, CoreOptions.ChangelogProducer.NONE); + KeyValueFileWriterFactory factory = + KeyValueFileWriterFactory.builder( + fileIO, + 0, + keyType, + valueType, + new FlushingFileFormat("avro"), + pathFactories, + 1024 * 1024) + .build(BinaryRow.EMPTY_ROW, 0, new CoreOptions(options)); + DataFilePathFactory pathFactory = factory.pathFactory(0); + + RollingFileWriter writer = + factory.createRollingMergeTreeFileWriter(0, FileSource.APPEND); + + Path firstPack = + pathFactory.newPathFromExtension(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX); + InternalRow first = + GenericRow.of( + 1, + BinaryString.fromString("a"), + Blob.fromFile(fileIO, firstPack.toString(), 0, 3)); + + writer.write(new KeyValue().replace(GenericRow.of(1), 0, RowKind.INSERT, first)); + + Path secondPack = + pathFactory.newPathFromExtension(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX); + InternalRow second = + GenericRow.of( + 1, + BinaryString.fromString("b"), + Blob.fromFile(fileIO, secondPack.toString(), 0, 3)); + assertThat(secondPack).isNotEqualTo(firstPack); + + writer.write(new KeyValue().replace(GenericRow.of(1), 1, RowKind.INSERT, second)); + + writer.write( + new KeyValue() + .replace( + GenericRow.of(1), + 2, + RowKind.INSERT, + GenericRow.of(1, BinaryString.fromString("c"), null))); + + writer.close(); + factory.prepareCommit(); + + DataFileMeta meta = writer.result().get(0); + assertThat(meta.extraFiles()).singleElement().asString().endsWith(".blobref"); + Path sidecar = pathFactory.toAlignedPath(meta.extraFiles().get(0), meta); + assertThat(ManagedBlobReferenceFile.read(fileIO, sidecar)) + .containsExactlyInAnyOrder( + new ManagedBlobReferenceFile.Reference( + sidecar.getParent().toString(), firstPack.getName()), + new ManagedBlobReferenceFile.Reference( + sidecar.getParent().toString(), secondPack.getName())); + } } 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..8adb20abb084 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,8 @@ public void testPrimaryKeyBlobFileField() { @Test public void testPrimaryKeyInlineBlobDoesNotTriggerManagedRestrictions() { + // Partial-update supports scalar blob-descriptor-field, managed blob-field, and + // blob-view-field on primary-key tables. List fields = Arrays.asList( new DataField(0, "id", DataTypes.INT()), @@ -423,6 +425,59 @@ public void testPrimaryKeyInlineBlobDoesNotTriggerManagedRestrictions() { assertThatCode(() -> validateTableSchema(schema)).doesNotThrowAnyException(); } + @Test + public void testPartialUpdateAllowsBlobViewField() { + 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, ""); + + assertThatCode(() -> validateTableSchema(schema)).doesNotThrowAnyException(); + } + + @Test + public void testPartialUpdateAllowsDescriptorWithBlobViewField() { + 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, ""); + + assertThatCode(() -> validateTableSchema(schema)).doesNotThrowAnyException(); + } + + @Test + public void testPartialUpdateAllowsManagedBlobField() { + List fields = + Arrays.asList( + new DataField(0, "id", DataTypes.INT()), + new DataField(1, "payload", DataTypes.BLOB())); + Map options = new HashMap<>(); + options.put(BUCKET.key(), "1"); + options.put(CoreOptions.BLOB_FIELD.key(), "payload"); + options.put(CoreOptions.MERGE_ENGINE.key(), "partial-update"); + + TableSchema schema = + new TableSchema(1, fields, 10, emptyList(), singletonList("id"), options, ""); + + assertThatCode(() -> validateTableSchema(schema)).doesNotThrowAnyException(); + } + @Test public void testPrimaryKeyBlobViewCoexistsWithManagedBlob() { List fields = @@ -466,12 +521,31 @@ 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]."); + + assertThatCode( () -> validateTableSchema( primaryKeyBlobSchema( mergeOptions, singletonList("id"), emptyList()))) - .hasMessage( - "Primary-key managed BLOB tables only support the deduplicate merge engine."); + .doesNotThrowAnyException(); Map changelogOptions = new HashMap<>(options); changelogOptions.put(CoreOptions.CHANGELOG_PRODUCER.key(), "input"); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyPartialUpdateBlobTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyPartialUpdateBlobTest.java new file mode 100644 index 000000000000..30cc53518c81 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyPartialUpdateBlobTest.java @@ -0,0 +1,310 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.Blob; +import org.apache.paimon.data.BlobData; +import org.apache.paimon.data.BlobView; +import org.apache.paimon.data.BlobViewStruct; +import org.apache.paimon.data.GenericArray; +import org.apache.paimon.data.GenericMap; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalArray; +import org.apache.paimon.data.InternalMap; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.sink.StreamTableCommit; +import org.apache.paimon.table.sink.StreamTableWrite; +import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.InnerTableRead; +import org.apache.paimon.table.source.ReadBuilder; +import org.apache.paimon.table.system.RowTrackingTable; +import org.apache.paimon.types.DataTypes; + +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for primary-key partial-update tables with managed and view BLOB fields. */ +public class PrimaryKeyPartialUpdateBlobTest extends TableTestBase { + + @Test + public void testPartialUpdateManagedBlobMergeAndCompact() throws Exception { + String tableName = "pk_pu_managed_blob"; + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("name", DataTypes.STRING()) + .column("payload", DataTypes.BLOB()) + .primaryKey("id") + .option(CoreOptions.MERGE_ENGINE.key(), "partial-update") + .option(CoreOptions.BLOB_FIELD.key(), "payload") + .option(CoreOptions.CHANGELOG_PRODUCER.key(), "none") + .option(CoreOptions.BUCKET.key(), "1") + .build(); + catalog.createTable(identifier(tableName), schema, true); + FileStoreTable table = getTable(identifier(tableName)); + + byte[] first = new byte[] {1, 2, 3}; + byte[] second = new byte[] {4, 5, 6}; + + try (StreamTableWrite write = table.newWrite(commitUser); + StreamTableCommit commit = table.newCommit(commitUser)) { + write.write(GenericRow.of(1, BinaryString.fromString("a"), new BlobData(first))); + commit.commit(0, write.prepareCommit(false, 0)); + + write.write(GenericRow.of(1, BinaryString.fromString("b"), null)); + commit.commit(1, write.prepareCommit(false, 1)); + + write.write(GenericRow.of(1, null, new BlobData(second))); + commit.commit(2, write.prepareCommit(false, 2)); + } + + ReadBuilder readBuilder = table.newReadBuilder(); + List rows = readRows(readBuilder); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getString(1).toString()).isEqualTo("b"); + assertThat(rows.get(0).getBlob(2).toData()).isEqualTo(second); + + List dataFiles = listDataFiles(table); + assertThat(dataFiles).isNotEmpty(); + assertThat(dataFiles.get(0).extraFiles()) + .anyMatch(extraFile -> extraFile.endsWith(".blobref")); + + compact(table, BinaryRow.EMPTY_ROW, 0); + rows = readRows(table.newReadBuilder()); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getBlob(2).toData()).isEqualTo(second); + assertThat(listDataFiles(table)).hasSize(1); + } + + @Test + public void testPartialUpdateManagedBlobCollections() throws Exception { + String tableName = "pk_pu_managed_blob_collections"; + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("name", DataTypes.STRING()) + .column("payloads", DataTypes.ARRAY(DataTypes.BLOB())) + .column("assets", DataTypes.MAP(DataTypes.STRING(), DataTypes.BLOB())) + .primaryKey("id") + .option(CoreOptions.MERGE_ENGINE.key(), "partial-update") + .option(CoreOptions.BLOB_FIELD.key(), "payloads,assets") + .option(CoreOptions.CHANGELOG_PRODUCER.key(), "none") + .option(CoreOptions.BUCKET.key(), "1") + .build(); + catalog.createTable(identifier(tableName), schema, true); + FileStoreTable table = getTable(identifier(tableName)); + + byte[] first = new byte[] {1, 2, 3}; + byte[] second = new byte[] {4, 5, 6}; + try (StreamTableWrite write = table.newWrite(commitUser); + StreamTableCommit commit = table.newCommit(commitUser)) { + write.write( + GenericRow.of( + 1, + BinaryString.fromString("a"), + new GenericArray(new Object[] {new BlobData(first), null}), + blobMap("first", first))); + commit.commit(0, write.prepareCommit(false, 0)); + + write.write(GenericRow.of(1, BinaryString.fromString("b"), null, null)); + commit.commit(1, write.prepareCommit(false, 1)); + } + + assertManagedCollectionRow(table, "b", first, "first", first, 2); + + try (StreamTableWrite write = table.newWrite(commitUser); + StreamTableCommit commit = table.newCommit(commitUser)) { + write.write( + GenericRow.of( + 1, + null, + new GenericArray(new Object[] {new BlobData(second)}), + blobMap("second", second))); + commit.commit(2, write.prepareCommit(false, 2)); + } + + assertManagedCollectionRow(table, "b", second, "second", second, 1); + compact(table, BinaryRow.EMPTY_ROW, 0); + assertManagedCollectionRow(table, "b", second, "second", second, 1); + } + + @Test + public void testPartialUpdateBlobViewResolvesOnRead() throws Exception { + String upstreamName = "pk_pu_upstream_blob"; + Schema upstreamSchema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("image", DataTypes.BLOB()) + .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true") + .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true") + .option(CoreOptions.BLOB_FIELD.key(), "image") + .build(); + catalog.createTable(identifier(upstreamName), upstreamSchema, true); + FileStoreTable upstreamTable = getTable(identifier(upstreamName)); + + byte[] imageBytes = new byte[] {72, 101, 108, 108, 111}; + byte[] secondImageBytes = new byte[] {87, 111, 114, 108, 100}; + write( + upstreamTable, + GenericRow.of(1, new BlobData(imageBytes)), + GenericRow.of(2, new BlobData(secondImageBytes))); + + int imageFieldId = upstreamTable.rowType().getField("image").id(); + RowTrackingTable upstreamRowTracking = new RowTrackingTable(upstreamTable); + ReadBuilder rowIdReader = + upstreamRowTracking.newReadBuilder().withProjection(new int[] {0, 2}); + Map idToRowId = new HashMap<>(); + rowIdReader + .newRead() + .createReader(rowIdReader.newScan().plan()) + .forEachRemaining(row -> idToRowId.put(row.getInt(0), row.getLong(1))); + assertThat(idToRowId).containsKeys(1, 2); + + String downstreamName = "pk_pu_blob_view"; + Schema downstreamSchema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("label", DataTypes.STRING()) + .column("image_ref", DataTypes.BLOB()) + .primaryKey("id") + .option(CoreOptions.MERGE_ENGINE.key(), "partial-update") + .option(CoreOptions.BLOB_VIEW_FIELD.key(), "image_ref") + .option(CoreOptions.BUCKET.key(), "1") + .build(); + catalog.createTable(identifier(downstreamName), downstreamSchema, true); + FileStoreTable downstreamTable = getTable(identifier(downstreamName)); + + String upstreamFullName = database + "." + upstreamName; + write( + downstreamTable, + GenericRow.of( + 1, + BinaryString.fromString("label1"), + Blob.fromView( + new BlobViewStruct( + Identifier.fromString(upstreamFullName), + imageFieldId, + idToRowId.get(1)))), + GenericRow.of( + 2, + BinaryString.fromString("label2"), + Blob.fromView( + new BlobViewStruct( + Identifier.fromString(upstreamFullName), + imageFieldId, + idToRowId.get(2))))); + + try (StreamTableWrite write = downstreamTable.newWrite(commitUser); + StreamTableCommit commit = downstreamTable.newCommit(commitUser)) { + write.write(GenericRow.of(1, BinaryString.fromString("updated"), null)); + commit.commit(0, write.prepareCommit(false, 0)); + } + + ReadBuilder readBuilder = downstreamTable.newReadBuilder(); + List rows = readRows(readBuilder); + assertThat(rows).hasSize(2); + InternalRow updatedRow = + rows.stream() + .filter(row -> row.getInt(0) == 1) + .findFirst() + .orElseThrow(AssertionError::new); + assertThat(updatedRow.getString(1).toString()).isEqualTo("updated"); + Blob blob = updatedRow.getBlob(2); + assertThat(blob).isInstanceOf(BlobView.class); + assertThat(((BlobView) blob).isResolved()).isTrue(); + assertThat(blob.toData()).isEqualTo(imageBytes); + + PredicateBuilder predicateBuilder = new PredicateBuilder(downstreamTable.rowType()); + ReadBuilder filteredReadBuilder = + downstreamTable.newReadBuilder().withFilter(predicateBuilder.equal(0, 2)); + InnerTableRead filteredRead = (InnerTableRead) filteredReadBuilder.newRead(); + filteredRead.withLimit(1); + filteredRead.executeFilter(); + rows = read(filteredRead.createReader(filteredReadBuilder.newScan().plan())); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getInt(0)).isEqualTo(2); + assertThat(rows.get(0).getBlob(2).toData()).isEqualTo(secondImageBytes); + } + + private List readRows(ReadBuilder readBuilder) throws Exception { + RecordReader reader = + readBuilder.newRead().createReader(readBuilder.newScan().plan()); + return read(reader); + } + + private void assertManagedCollectionRow( + FileStoreTable table, + String expectedName, + byte[] expectedArrayValue, + String expectedMapKey, + byte[] expectedMapValue, + int expectedArraySize) + throws Exception { + List rows = readRows(table.newReadBuilder()); + assertThat(rows).hasSize(1); + InternalRow row = rows.get(0); + assertThat(row.getString(1).toString()).isEqualTo(expectedName); + + InternalArray payloads = row.getArray(2); + assertThat(payloads.size()).isEqualTo(expectedArraySize); + assertThat(payloads.getBlob(0).toData()).isEqualTo(expectedArrayValue); + if (expectedArraySize > 1) { + assertThat(payloads.isNullAt(1)).isTrue(); + } + + InternalMap assets = row.getMap(3); + assertThat(assets.size()).isEqualTo(1); + assertThat(assets.keyArray().getString(0).toString()).isEqualTo(expectedMapKey); + assertThat(assets.valueArray().getBlob(0).toData()).isEqualTo(expectedMapValue); + } + + private GenericMap blobMap(String key, byte[] value) { + Map blobs = new LinkedHashMap<>(); + blobs.put(BinaryString.fromString(key), new BlobData(value)); + return new GenericMap(blobs); + } + + private List listDataFiles(FileStoreTable table) { + return table.newSnapshotReader().read().dataSplits().stream() + .flatMap(split -> ((DataSplit) split).dataFiles().stream()) + .collect(Collectors.toList()); + } + + private List read(RecordReader reader) throws Exception { + List rows = new java.util.ArrayList<>(); + reader.forEachRemaining(rows::add); + reader.close(); + return rows; + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/BlobViewTableReadSupportTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/BlobViewTableReadSupportTest.java new file mode 100644 index 000000000000..6dbd65574067 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/BlobViewTableReadSupportTest.java @@ -0,0 +1,75 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table.source; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.options.Options; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link BlobViewTableReadSupport}. */ +class BlobViewTableReadSupportTest { + + @Test + void testBlobViewFieldIndexesWhenResolveDisabled() { + RowType rowType = + RowType.of( + new DataField(0, "id", DataTypes.INT()), + new DataField(1, "view", DataTypes.BLOB())); + Options options = new Options(); + options.set(CoreOptions.BLOB_VIEW_FIELD, "view"); + options.set(CoreOptions.BLOB_VIEW_RESOLVE_ENABLED, false); + + assertThat(BlobViewTableReadSupport.blobViewFieldIndexes(rowType, new CoreOptions(options))) + .isEmpty(); + } + + @Test + void testBlobViewFieldIndexesReturnsProjectedIndexes() { + RowType rowType = + RowType.of( + new DataField(0, "id", DataTypes.INT()), + new DataField(1, "label", DataTypes.STRING()), + new DataField(2, "view", DataTypes.BLOB())); + Options options = new Options(); + options.set(CoreOptions.BLOB_VIEW_FIELD, "view"); + + assertThat(BlobViewTableReadSupport.blobViewFieldIndexes(rowType, new CoreOptions(options))) + .containsExactly(2); + } + + @Test + void testBlobViewFieldIndexesIgnoresNonConfiguredFields() { + RowType rowType = + RowType.of( + new DataField(0, "id", DataTypes.INT()), + new DataField(1, "payload", DataTypes.BLOB()), + new DataField(2, "view", DataTypes.BLOB())); + Options options = new Options(); + options.set(CoreOptions.BLOB_VIEW_FIELD, "view"); + + assertThat(BlobViewTableReadSupport.blobViewFieldIndexes(rowType, new CoreOptions(options))) + .containsExactly(2); + } +} 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..681821f19032 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,17 @@ package org.apache.paimon.flink; +import org.apache.paimon.Snapshot; +import org.apache.paimon.data.BlobDescriptor; +import org.apache.paimon.data.BlobViewStruct; +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 +38,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 +779,366 @@ 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 testManagedBlobPartialUpdate() throws Exception { + byte[] first = "blob-v1".getBytes(); + byte[] second = "blob-v2".getBytes(); + + sql( + "CREATE TABLE pu_managed_blob (" + + " id INT PRIMARY KEY NOT ENFORCED," + + " name STRING," + + " payload BYTES" + + ") WITH (" + + " 'merge-engine'='partial-update'," + + " 'blob-field'='payload'," + + " 'changelog-producer'='none'," + + " 'bucket'='1'," + + " 'write-only'='true'," + + " 'num-sorted-run.compaction-trigger'='100'" + + ")"); + + sql("INSERT INTO pu_managed_blob VALUES (1, 'a', " + toHexLiteral(first) + ")"); + sql("INSERT INTO pu_managed_blob VALUES (1, 'b', CAST(NULL AS BYTES))"); + assertThat(sql("SELECT id, name, payload FROM pu_managed_blob")) + .containsExactly(Row.of(1, "b", first)); + + sql( + "INSERT INTO pu_managed_blob VALUES (1, CAST(NULL AS STRING), " + + toHexLiteral(second) + + ")"); + assertThat(sql("SELECT id, name, payload FROM pu_managed_blob")) + .containsExactly(Row.of(1, "b", second)); + + assertThat((long) sql("SELECT COUNT(*) FROM `pu_managed_blob$files`").get(0).getField(0)) + .isGreaterThan(1); + tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true); + sql("CALL sys.compact(`table` => 'default.pu_managed_blob', compact_strategy => 'full')"); + Snapshot snapshot = findLatestSnapshot("pu_managed_blob"); + assertThat(snapshot).isNotNull(); + assertThat(snapshot.commitKind()).isEqualTo(Snapshot.CommitKind.COMPACT); + assertThat(sql("SELECT COUNT(*) FROM `pu_managed_blob$files`")).containsExactly(Row.of(1L)); + assertThat(sql("SELECT id, name, payload FROM pu_managed_blob")) + .containsExactly(Row.of(1, "b", second)); + } + + @Test + public void testManagedBlobPartialUpdateSequenceGroup() throws Exception { + byte[] first = "blob-seq-v1".getBytes(); + byte[] second = "blob-seq-v2".getBytes(); + sql( + "CREATE TABLE pu_managed_blob_seq (" + + " id INT PRIMARY KEY NOT ENFORCED," + + " name STRING," + + " payload BYTES," + + " ts INT" + + ") WITH (" + + " 'merge-engine'='partial-update'," + + " 'blob-field'='payload'," + + " 'changelog-producer'='none'," + + " 'fields.ts.sequence-group'='name,payload'," + + " 'bucket'='1'," + + " 'write-only'='true'," + + " 'num-sorted-run.compaction-trigger'='100'" + + ")"); + + sql("INSERT INTO pu_managed_blob_seq VALUES (1, 'first', " + toHexLiteral(first) + ", 2)"); + sql("INSERT INTO pu_managed_blob_seq VALUES (1, 'older', " + toHexLiteral(second) + ", 1)"); + assertThat(sql("SELECT id, name, payload, ts FROM pu_managed_blob_seq")) + .containsExactly(Row.of(1, "first", first, 2)); + + sql("INSERT INTO pu_managed_blob_seq VALUES (1, 'cleared', CAST(NULL AS BYTES), 3)"); + assertThat(sql("SELECT id, name, payload, ts FROM pu_managed_blob_seq")) + .containsExactly(Row.of(1, "cleared", null, 3)); + + tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true); + sql( + "CALL sys.compact(" + + "`table` => 'default.pu_managed_blob_seq', " + + "compact_strategy => 'full')"); + assertThat(sql("SELECT id, name, payload, ts FROM pu_managed_blob_seq")) + .containsExactly(Row.of(1, "cleared", null, 3)); + } + + @Test + public void testBlobViewPartialUpdate() throws Exception { + createBlobViewPartialUpdateTables(); + + assertThat(sql("SELECT id, label, image_ref FROM pu_blob_view ORDER BY id")) + .containsExactly(Row.of(1, "row1", new byte[] {72, 101, 108, 108, 111})); + + sql("INSERT INTO pu_blob_view VALUES (1, 'updated', CAST(NULL AS BYTES))"); + assertThat(sql("SELECT id, label, image_ref FROM pu_blob_view")) + .containsExactly(Row.of(1, "updated", new byte[] {72, 101, 108, 108, 111})); + + assertThat((long) sql("SELECT COUNT(*) FROM `pu_blob_view$files`").get(0).getField(0)) + .isGreaterThan(1); + tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true); + sql("CALL sys.compact(`table` => 'default.pu_blob_view', compact_strategy => 'full')"); + Snapshot snapshot = findLatestSnapshot("pu_blob_view"); + assertThat(snapshot).isNotNull(); + assertThat(snapshot.commitKind()).isEqualTo(Snapshot.CommitKind.COMPACT); + assertThat(sql("SELECT COUNT(*) FROM `pu_blob_view$files`")).containsExactly(Row.of(1L)); + assertThat(sql("SELECT id, label, image_ref FROM pu_blob_view")) + .containsExactly(Row.of(1, "updated", new byte[] {72, 101, 108, 108, 111})); + } + + @Test + public void testBlobViewPartialUpdateForwardReference() throws Exception { + String fullTableName = createBlobViewPartialUpdateTables(); + sql("INSERT INTO pu_blob_view VALUES (1, 'updated', CAST(NULL AS BYTES))"); + sql( + "CREATE TABLE pu_blob_view_forward (" + + " id INT PRIMARY KEY NOT ENFORCED," + + " label STRING," + + " image_ref BYTES" + + ") WITH (" + + " 'merge-engine'='partial-update'," + + " 'blob-view-field'='image_ref'" + + ")"); + sql( + "INSERT INTO pu_blob_view_forward" + + " SELECT id, label, image_ref" + + " FROM pu_blob_view" + + " /*+ OPTIONS('blob-view.resolve.enabled'='false') */"); + + assertThat(sql("SELECT id, label, image_ref FROM pu_blob_view_forward")) + .containsExactly(Row.of(1, "updated", new byte[] {72, 101, 108, 108, 111})); + + byte[] originalReference = + (byte[]) + sql("SELECT image_ref FROM pu_blob_view" + + " /*+ OPTIONS('blob-view.resolve.enabled'='false') */") + .get(0) + .getField(0); + byte[] forwardedReference = + (byte[]) + sql("SELECT image_ref FROM pu_blob_view_forward" + + " /*+ OPTIONS('blob-view.resolve.enabled'='false') */") + .get(0) + .getField(0); + assertThat(forwardedReference).isEqualTo(originalReference); + assertThat(BlobViewStruct.deserialize(forwardedReference).identifier().getFullName()) + .isEqualTo(fullTableName); + } + + @Test + public void testBlobViewPartialUpdateSequenceGroup() throws Exception { + String fullTableName = createBlobViewUpstream(); + sql( + "CREATE TABLE pu_blob_view_seq (" + + " id INT PRIMARY KEY NOT ENFORCED," + + " label STRING," + + " image_ref BYTES," + + " ts INT" + + ") WITH (" + + " 'merge-engine'='partial-update'," + + " 'blob-view-field'='image_ref'," + + " 'fields.ts.sequence-group'='label,image_ref'," + + " 'bucket'='1'," + + " 'write-only'='true'," + + " 'num-sorted-run.compaction-trigger'='100'" + + ")"); + sql( + String.format( + "INSERT INTO pu_blob_view_seq" + + " SELECT id, name, sys.blob_view('%s', 'picture', _ROW_ID), 2" + + " FROM `pu_upstream_blob$row_tracking`", + fullTableName)); + + sql("INSERT INTO pu_blob_view_seq VALUES (1, 'older', CAST(NULL AS BYTES), 1)"); + assertThat(sql("SELECT id, label, image_ref, ts FROM pu_blob_view_seq")) + .containsExactly(Row.of(1, "row1", new byte[] {72, 101, 108, 108, 111}, 2)); + + sql("INSERT INTO pu_blob_view_seq VALUES (1, 'cleared', CAST(NULL AS BYTES), 3)"); + assertThat(sql("SELECT id, label, image_ref, ts FROM pu_blob_view_seq")) + .containsExactly(Row.of(1, "cleared", null, 3)); + + tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true); + sql( + "CALL sys.compact(" + + "`table` => 'default.pu_blob_view_seq', " + + "compact_strategy => 'full')"); + assertThat(sql("SELECT id, label, image_ref, ts FROM pu_blob_view_seq")) + .containsExactly(Row.of(1, "cleared", null, 3)); + } + + private String createBlobViewPartialUpdateTables() { + String fullTableName = createBlobViewUpstream(); + sql( + "CREATE TABLE pu_blob_view (" + + " id INT PRIMARY KEY NOT ENFORCED," + + " label STRING," + + " image_ref BYTES" + + ") WITH (" + + " 'merge-engine'='partial-update'," + + " 'blob-view-field'='image_ref'," + + " 'bucket'='1'," + + " 'write-only'='true'," + + " 'num-sorted-run.compaction-trigger'='100'" + + ")"); + sql( + String.format( + "INSERT INTO pu_blob_view" + + " SELECT id, name, sys.blob_view('%s', 'picture', _ROW_ID)" + + " FROM `pu_upstream_blob$row_tracking`", + fullTableName)); + return fullTableName; + } + + private String createBlobViewUpstream() { + sql( + "CREATE TABLE pu_upstream_blob (" + + " id INT, name STRING, picture BYTES" + + ") WITH (" + + " 'row-tracking.enabled'='true'," + + " 'data-evolution.enabled'='true'," + + " 'blob-field'='picture'" + + ")"); + sql("INSERT INTO pu_upstream_blob VALUES (1, 'row1', X'48656C6C6F')"); + + String fullTableName = tEnv.getCurrentDatabase() + ".pu_upstream_blob"; + return fullTableName; + } + + private static String toHexLiteral(byte[] data) { + StringBuilder builder = new StringBuilder("X'"); + for (byte value : data) { + builder.append(String.format("%02X", value)); + } + builder.append("'"); + return builder.toString(); + } + + 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; + } }