This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 35d32cb61f [core] Support BLOB fields in partial-update tables. (#8928)
35d32cb61f is described below
commit 35d32cb61fd9812b02f8f7b023132ca64034db3c
Author: Wenchao Wu <[email protected]>
AuthorDate: Wed Aug 5 15:03:26 2026 +0800
[core] Support BLOB fields in partial-update tables. (#8928)
---
docs/docs/multimodal-table/blob.mdx | 6 +-
docs/docs/primary-key-table/blob-storage.md | 86 +++-
.../paimon/blob/ManagedBlobReferenceCollector.java | 2 +-
.../compact/PartialUpdateMergeFunction.java | 36 +-
.../org/apache/paimon/schema/SchemaValidation.java | 106 ++++-
.../paimon/table/PrimaryKeyFileStoreTable.java | 5 +-
.../paimon/table/source/AbstractDataTableRead.java | 4 +
...ableRead.java => BlobViewTableReadSupport.java} | 99 ++---
.../table/source/DataEvolutionTableRead.java | 119 ++----
.../paimon/table/source/KeyValueTableRead.java | 61 ++-
.../paimon/io/PrimaryKeyBlobFileWriterTest.java | 98 +++++
.../compact/PartialUpdateMergeFunctionTest.java | 59 +++
.../apache/paimon/schema/SchemaValidationTest.java | 192 ++++++++-
.../table/PrimaryKeyPartialUpdateBlobTest.java | 441 +++++++++++++++++++++
.../table/source/BlobViewTableReadSupportTest.java | 75 ++++
.../apache/paimon/flink/PartialUpdateITCase.java | 369 +++++++++++++++++
16 files changed, 1567 insertions(+), 191 deletions(-)
diff --git a/docs/docs/multimodal-table/blob.mdx
b/docs/docs/multimodal-table/blob.mdx
index 3ee89a0910..064ea567a5 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<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). 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 3eff81bb83..05acbaf3f6 100644
--- a/docs/docs/primary-key-table/blob-storage.md
+++ b/docs/docs/primary-key-table/blob-storage.md
@@ -102,9 +102,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.
:::
@@ -115,7 +115,7 @@ Primary-key managed BLOB storage has the following
requirements:
| Item | Requirement |
|------|-------------|
| Managed BLOB declaration | `BLOB`, `ARRAY<BLOB>`, and `MAP<K, BLOB>` 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 |
@@ -123,11 +123,81 @@ 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<BLOB>`, or `MAP<K, BLOB>`), 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 |
+
+The ordering fields on the left-hand side of
+`fields.<ordering-field[,ordering-field...]>.sequence-group` cannot contain
BLOB values. This
+includes `BLOB`, `ARRAY<BLOB>`, and `MAP<K, BLOB>` fields in any storage mode.
BLOB fields are
+supported on the right-hand side as fields protected by the sequence group.
+
+```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).
+Managed BLOB retract records do not retain payload values. When a managed BLOB
field is protected
+by a sequence group and retract records are processed, it therefore cannot use
an aggregate
+function that depends on the original retract payload. `last_value` is
supported because it always
+clears the field on retract; other aggregate functions require
+`fields.<field-name>.ignore-retract=true`. This restriction does not apply
when `ignore-delete=true`
+or when the managed BLOB field is not protected by a sequence group.
+
+```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. Without aggregate functions, an
+incoming sequence value that is newer or equal replaces every field in the
group, including replacing a BLOB value
+with null; an older record is ignored. With aggregate functions, every record
with a non-null sequence value
+participates in aggregation or retraction, even when its sequence value is
older. For example, `last_value` clears
+the field for both newer and older retract records.
+
+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 529b077888..841d3d1eed 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 java.util.Set;
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/mergetree/compact/PartialUpdateMergeFunction.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
index 6cc551baec..00bc83b152 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
@@ -69,6 +69,30 @@ public class PartialUpdateMergeFunction implements
MergeFunction<KeyValue> {
"The sequence-group '%s' contains primary key field '%s', "
+ "which is not allowed. Primary key columns cannot be put
in sequence-group.";
+ public static boolean isSequenceGroupOption(String optionKey) {
+ return optionKey.startsWith(FIELDS_PREFIX + ".")
+ && optionKey.endsWith("." + SEQUENCE_GROUP);
+ }
+
+ public static boolean isSequenceGroupOptionCandidate(String optionKey) {
+ return optionKey.startsWith(FIELDS_PREFIX) &&
optionKey.endsWith(SEQUENCE_GROUP);
+ }
+
+ public static List<String> sequenceGroupOrderingFields(String optionKey) {
+ checkArgument(
+ isSequenceGroupOption(optionKey), "Invalid sequence-group
option: %s", optionKey);
+ return Arrays.asList(
+ optionKey
+ .substring(
+ FIELDS_PREFIX.length() + 1,
+ optionKey.length() - SEQUENCE_GROUP.length() -
1)
+ .split(FIELDS_SEPARATOR));
+ }
+
+ public static List<String> sequenceGroupProtectedFields(String
optionValue) {
+ return Arrays.asList(optionValue.split(FIELDS_SEPARATOR));
+ }
+
private final InternalRow.FieldGetter[] getters;
private final boolean ignoreDelete;
private final List<WrapperWithFieldIndex<FieldsComparator>>
fieldSeqComparators;
@@ -413,21 +437,15 @@ public class PartialUpdateMergeFunction implements
MergeFunction<KeyValue> {
for (Map.Entry<String, String> entry : options.toMap().entrySet())
{
String k = entry.getKey();
String v = entry.getValue();
- if (k.startsWith(FIELDS_PREFIX) && k.endsWith(SEQUENCE_GROUP))
{
+ if (isSequenceGroupOptionCandidate(k)) {
int[] sequenceFields =
- Arrays.stream(
- k.substring(
-
FIELDS_PREFIX.length() + 1,
- k.length()
- -
SEQUENCE_GROUP.length()
- - 1)
- .split(FIELDS_SEPARATOR))
+ sequenceGroupOrderingFields(k).stream()
.mapToInt(fieldName ->
requireField(fieldName, fieldNames))
.toArray();
Supplier<FieldsComparator> userDefinedSeqComparator =
() -> UserDefinedSeqComparator.create(rowType,
sequenceFields, true);
- Arrays.stream(v.split(FIELDS_SEPARATOR))
+ sequenceGroupProtectedFields(v).stream()
.map(fieldName -> requireField(fieldName,
fieldNames))
.forEach(
field -> {
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 7b9dead77c..0e57e364b5 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
@@ -34,6 +34,7 @@ import
org.apache.paimon.globalindex.bitmap.BitmapGlobalIndexerFactory;
import org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory;
import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
import
org.apache.paimon.mergetree.compact.aggregate.factory.FieldAggregatorFactory;
+import
org.apache.paimon.mergetree.compact.aggregate.factory.FieldLastValueAggFactory;
import org.apache.paimon.options.ConfigOption;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.BucketMode;
@@ -90,6 +91,10 @@ import static
org.apache.paimon.CoreOptions.SNAPSHOT_NUM_RETAINED_MAX;
import static org.apache.paimon.CoreOptions.SNAPSHOT_NUM_RETAINED_MIN;
import static org.apache.paimon.CoreOptions.STREAMING_READ_OVERWRITE;
import static org.apache.paimon.format.FileFormat.vectorFileFormat;
+import static
org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction.isSequenceGroupOption;
+import static
org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction.isSequenceGroupOptionCandidate;
+import static
org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction.sequenceGroupOrderingFields;
+import static
org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction.sequenceGroupProtectedFields;
import static org.apache.paimon.schema.TableSchema.PAIMON_07_VERSION;
import static
org.apache.paimon.table.PrimaryKeyTableUtils.createMergeFunctionFactory;
import static org.apache.paimon.table.SpecialFields.KEY_FIELD_PREFIX;
@@ -156,6 +161,8 @@ public class SchemaValidation {
validateSequenceField(schema, options);
+ validateSequenceGroupOrderingFields(schema, options);
+
validateMergeFunction(schema);
ChangelogProducer changelogProducer = options.changelogProducer();
@@ -220,6 +227,7 @@ public class SchemaValidation {
Set<String> blobDescriptorFields =
validateBlobDescriptorFields(tableRowType, options);
Set<String> blobViewFields =
validateBlobViewFields(tableRowType, options,
blobDescriptorFields);
+ validatePrimaryKeyBlobKeyConfiguration(schema, options);
validatePrimaryKeyBlobConfiguration(schema, options);
Set<String> blobInlineFields = new HashSet<>(blobDescriptorFields);
blobInlineFields.addAll(blobViewFields);
@@ -1423,6 +1431,89 @@ public class SchemaValidation {
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());
+
+ if (options.mergeEngine() == MergeEngine.PARTIAL_UPDATE &&
!options.ignoreDelete()) {
+ Set<String> fieldsProtectedBySequenceGroup =
+ options.toMap().entrySet().stream()
+ .filter(entry ->
isSequenceGroupOption(entry.getKey()))
+ .flatMap(
+ entry ->
+
sequenceGroupProtectedFields(entry.getValue()).stream())
+ .collect(Collectors.toSet());
+ for (String field : managedBlobFields) {
+ if (!fieldsProtectedBySequenceGroup.contains(field)) {
+ continue;
+ }
+ String aggregateFunction = options.fieldAggFunc(field);
+ if (aggregateFunction == null) {
+ aggregateFunction = options.fieldsDefaultFunc();
+ }
+ checkArgument(
+ aggregateFunction == null
+ ||
FieldLastValueAggFactory.NAME.equals(aggregateFunction)
+ || options.fieldAggIgnoreRetract(field),
+ "Managed BLOB field '%s' cannot use aggregate function
'%s' because "
+ + "managed BLOB payloads are not retained in
retract messages. "
+ + "Set 'fields.%s.ignore-retract' to true to
ignore retract messages.",
+ field,
+ aggregateFunction,
+ field);
+ }
+ }
+ }
+
+ private static void validateSequenceGroupOrderingFields(
+ TableSchema schema, CoreOptions options) {
+ if (options.mergeEngine() != MergeEngine.PARTIAL_UPDATE) {
+ return;
+ }
+
+ RowType rowType = new RowType(schema.fields());
+ for (String optionKey : options.toMap().keySet()) {
+ if (!isSequenceGroupOptionCandidate(optionKey)) {
+ continue;
+ }
+ for (String fieldName : sequenceGroupOrderingFields(optionKey)) {
+ DataField field = rowType.getField(fieldName);
+ checkArgument(
+ !containsType(field.type(), type ->
type.is(DataTypeRoot.BLOB)),
+ "Field '%s' with type %s cannot be used as a
sequence-group ordering "
+ + "field in option '%s'.",
+ fieldName,
+ field.type(),
+ optionKey);
+ }
+ }
+ }
+
+ 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)
@@ -1449,21 +1540,6 @@ public class SchemaValidation {
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 5e48557d76..ca482ca095 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
@@ -150,7 +150,10 @@ public class PrimaryKeyFileStoreTable extends
AbstractFileStoreTable {
@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 f349047fe8..c46e1f299f 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 @@ public abstract class AbstractDataTableRead implements
InnerTableRead {
return predicate;
}
+ protected boolean shouldExecuteFilter() {
+ return executeFilter;
+ }
+
@Override
public RecordReader<InternalRow> createReader(Split split) throws
IOException {
QueryAuthContext queryAuthContext = unwrapQueryAuthSplit(split);
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/BlobViewTableReadSupport.java
similarity index 60%
copy from
paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionTableRead.java
copy to
paimon-core/src/main/java/org/apache/paimon/table/source/BlobViewTableReadSupport.java
index bf41ff70b6..6602d40a6e 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/BlobViewTableReadSupport.java
@@ -27,10 +27,8 @@ 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.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;
@@ -41,46 +39,15 @@ 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;
-/** A {@link TableRead} for data-evolution enabled append-only tables. */
-public class DataEvolutionTableRead extends AppendTableRead {
+/** Shared helpers for resolving {@link
org.apache.paimon.CoreOptions#BLOB_VIEW_FIELD} on read. */
+final class BlobViewTableReadSupport {
- @Nullable private final CatalogContext catalogContext;
- @Nullable private final Supplier<InnerTableRead> readFactory;
+ private BlobViewTableReadSupport() {}
- public DataEvolutionTableRead(
- List<Function<SplitReadConfig, SplitReadProvider>>
providerFactories,
- TableSchema schema,
- @Nullable CatalogContext catalogContext,
- @Nullable Supplier<InnerTableRead> readFactory) {
- super(providerFactories, schema);
- this.catalogContext = catalogContext;
- this.readFactory = readFactory;
- }
-
- @Override
- public RecordReader<InternalRow> 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());
+ static int[] blobViewFieldIndexes(RowType rowType, CoreOptions options) {
if (!options.blobViewResolveEnabled()) {
return new int[0];
}
@@ -99,28 +66,51 @@ public class DataEvolutionTableRead extends AppendTableRead
{
.toArray();
}
- private RecordReader<InternalRow> createBlobViewReader(
- Split split, @Nullable TableQueryAuthResult authResult, int[]
blobViewFields)
+ static RecordReader<InternalRow> 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<InnerTableRead> prescanReadSupplier)
throws IOException {
- RowType blobViewOnlyType = currentReadType().project(blobViewFields);
- InnerTableRead prescanRead = readFactory.get();
- prescanRead.withReadType(blobViewOnlyType);
- Predicate predicate = predicate();
+ 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);
}
- configureBlobViewPrescanRead(prescanRead);
+ if (topN != null) {
+ prescanRead.withTopN(topN);
+ }
+ if (limit != null) {
+ prescanRead.withLimit(limit);
+ }
+
Split prescanSplit = authResult != null ? new QueryAuthSplit(split,
authResult) : split;
LinkedHashSet<BlobViewStruct> viewStructs = new LinkedHashSet<>();
RecordReader<InternalRow> prescanReader =
prescanRead.createReader(prescanSplit);
try {
prescanReader.forEachRemaining(
row -> {
- for (int i = 0; i < blobViewFields.length; i++) {
- if (row.isNullAt(i)) {
+ for (int field : prescanBlobViewFields) {
+ if (row.isNullAt(field)) {
continue;
}
- Blob blob = row.getBlob(i);
+ Blob blob = row.getBlob(field);
if (!(blob instanceof BlobView)) {
throw new IllegalArgumentException(
"blob-view-field requires blob field
value to be a "
@@ -136,7 +126,7 @@ public class DataEvolutionTableRead extends AppendTableRead
{
BlobViewResolver resolver =
BlobViewLookup.createResolver(catalogContext, new
ArrayList<>(viewStructs));
- RecordReader<InternalRow> reader = createDataReader(split, authResult);
+ RecordReader<InternalRow> reader = dataReaderSupplier.get();
Set<Integer> blobViewFieldSet = new HashSet<>();
for (int field : blobViewFields) {
blobViewFieldSet.add(field);
@@ -144,12 +134,9 @@ public class DataEvolutionTableRead extends
AppendTableRead {
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);
- }
+ @FunctionalInterface
+ interface RecordReaderSupplier {
+
+ RecordReader<InternalRow> 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 bf41ff70b6..60bd2b99fe 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 @@ 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.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 class DataEvolutionTableRead extends AppendTableRead
{
@Override
public RecordReader<InternalRow> 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<String> 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<InternalRow> 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<BlobViewStruct> viewStructs = new LinkedHashSet<>();
- RecordReader<InternalRow> 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<InternalRow> reader = createDataReader(split, authResult);
- Set<Integer> 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 0b19df5c0c..9c02bb470f 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 @@ package org.apache.paimon.table.source;
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 @@ import java.util.function.Supplier;
*/
public final class KeyValueTableRead extends AbstractDataTableRead {
+ private final Supplier<MergeFileSplitRead> mergeReadSupplier;
+ private final Supplier<RawFileSplitRead> batchRawReadSupplier;
private final List<SplitReadProvider> readProviders;
+ @Nullable private final CatalogContext catalogContext;
@Nullable private RowType readType = null;
private boolean forceKeepDelete = false;
@@ -66,7 +70,18 @@ public final class KeyValueTableRead extends
AbstractDataTableRead {
Supplier<MergeFileSplitRead> mergeReadSupplier,
Supplier<RawFileSplitRead> batchRawReadSupplier,
TableSchema schema) {
+ this(mergeReadSupplier, batchRawReadSupplier, schema, null);
+ }
+
+ public KeyValueTableRead(
+ Supplier<MergeFileSplitRead> mergeReadSupplier,
+ Supplier<RawFileSplitRead> 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 final class KeyValueTableRead extends
AbstractDataTableRead {
@Override
public RecordReader<InternalRow> createReader(Split split) throws
IOException {
- return LimitRecordReader.limit(super.createReader(split), limit);
+ QueryAuthContext queryAuthContext = unwrapQueryAuthSplit(split);
+ RecordReader<InternalRow> 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 22c549bf3f..d7f4d2ee0e 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.CoreOptions;
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 @@ class PrimaryKeyBlobFileWriterTest {
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<String, FileStorePathFactory> 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<KeyValue, DataFileMeta> 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 050f4b855b..9de9d22957 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 class PartialUpdateMergeFunctionTest {
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 @@ public class PartialUpdateMergeFunctionTest {
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));
}
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 26294112f6..4d3a71a4f2 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 @@ class SchemaValidationTest {
@Test
public void testPrimaryKeyInlineBlobDoesNotTriggerManagedRestrictions() {
+ // Partial-update supports scalar blob-descriptor-field, managed
blob-field, and
+ // blob-view-field on primary-key tables.
List<DataField> fields =
Arrays.asList(
new DataField(0, "id", DataTypes.INT()),
@@ -423,6 +425,173 @@ class SchemaValidationTest {
assertThatCode(() ->
validateTableSchema(schema)).doesNotThrowAnyException();
}
+ @Test
+ public void testPartialUpdateAllowsBlobViewField() {
+ 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, "");
+
+ assertThatCode(() ->
validateTableSchema(schema)).doesNotThrowAnyException();
+ }
+
+ @Test
+ public void testPartialUpdateAllowsDescriptorWithBlobViewField() {
+ 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, "");
+
+ assertThatCode(() ->
validateTableSchema(schema)).doesNotThrowAnyException();
+ }
+
+ @Test
+ public void testPartialUpdateAllowsManagedBlobField() {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "payload", DataTypes.BLOB()));
+ Map<String, String> 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 testPartialUpdateRejectsManagedBlobRetractAggregation() {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "payload", DataTypes.BLOB()),
+ new DataField(2, "ts", DataTypes.INT()));
+ Map<String, String> options = new HashMap<>();
+ options.put(BUCKET.key(), "1");
+ options.put(CoreOptions.BLOB_FIELD.key(), "payload");
+ options.put(CoreOptions.MERGE_ENGINE.key(), "partial-update");
+ options.put("fields.ts.sequence-group", "payload");
+ options.put("fields.payload.aggregate-function",
"last_non_null_value");
+
+ TableSchema schema =
+ new TableSchema(1, fields, 10, emptyList(),
singletonList("id"), options, "");
+ assertThatThrownBy(() -> validateTableSchema(schema))
+ .hasMessageContaining(
+ "Managed BLOB field 'payload' cannot use aggregate
function "
+ + "'last_non_null_value'")
+ .hasMessageContaining("fields.payload.ignore-retract");
+
+ options.put("fields.payload.ignore-retract", "true");
+ assertThatCode(() ->
validateTableSchema(schema)).doesNotThrowAnyException();
+
+ options.remove("fields.payload.ignore-retract");
+ options.put("fields.payload.aggregate-function", "last_value");
+ assertThatCode(() ->
validateTableSchema(schema)).doesNotThrowAnyException();
+
+ options.put("fields.payload.aggregate-function",
"last_non_null_value");
+ options.put(CoreOptions.IGNORE_DELETE.key(), "true");
+ assertThatCode(() ->
validateTableSchema(schema)).doesNotThrowAnyException();
+
+ options.remove(CoreOptions.IGNORE_DELETE.key());
+ options.remove("fields.ts.sequence-group");
+ assertThatCode(() ->
validateTableSchema(schema)).doesNotThrowAnyException();
+
+ options.remove("fields.payload.aggregate-function");
+ options.put("fields.ts.sequence-group", "payload");
+ options.put("fields.default-aggregate-function",
"last_non_null_value");
+ assertThatThrownBy(() -> validateTableSchema(schema))
+ .hasMessageContaining(
+ "Managed BLOB field 'payload' cannot use aggregate
function "
+ + "'last_non_null_value'");
+ }
+
+ @Test
+ public void testPartialUpdateRejectsBlobSequenceGroupOrderingFields() {
+ List<DataType> unsupportedTypes =
+ Arrays.asList(
+ DataTypes.BLOB(),
+ DataTypes.ARRAY(DataTypes.BLOB()),
+ DataTypes.MAP(DataTypes.STRING(), DataTypes.BLOB()));
+ for (DataType unsupportedType : unsupportedTypes) {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "ordering", unsupportedType),
+ new DataField(2, "payload", DataTypes.INT()),
+ new DataField(3, "ts", DataTypes.INT()));
+ Map<String, String> options = new HashMap<>();
+ options.put(BUCKET.key(), "1");
+ options.put(CoreOptions.BLOB_FIELD.key(), "ordering");
+ options.put(CoreOptions.MERGE_ENGINE.key(), "partial-update");
+ options.put("fields.ordering.sequence-group", "payload");
+ TableSchema schema =
+ new TableSchema(1, fields, 10, emptyList(),
singletonList("id"), options, "");
+
+ assertThatThrownBy(() -> validateTableSchema(schema))
+ .as("ordering type %s", unsupportedType)
+ .hasMessageContaining("Field 'ordering' with type")
+ .hasMessageContaining("cannot be used as a sequence-group
ordering field")
+ .hasMessageContaining("fields.ordering.sequence-group");
+
+ options.remove("fields.ordering.sequence-group");
+ options.put("fields.ts,ordering.sequence-group", "payload");
+ assertThatThrownBy(() -> validateTableSchema(schema))
+ .as("multi-field ordering type %s", unsupportedType)
+ .hasMessageContaining("Field 'ordering' with type")
+ .hasMessageContaining("fields.ts,ordering.sequence-group");
+
+ options.remove("fields.ts,ordering.sequence-group");
+ options.put("fields.ts.sequence-group", "ordering,payload");
+ assertThatCode(() -> validateTableSchema(schema))
+ .as("protected type %s", unsupportedType)
+ .doesNotThrowAnyException();
+
+ options.remove("fields.ts.sequence-group");
+ options.put("fields.ordering.sequence-group", "payload");
+ options.put(CoreOptions.MERGE_ENGINE.key(), "deduplicate");
+ assertThatCode(() -> validateTableSchema(schema))
+ .as("dormant ordering option with type %s",
unsupportedType)
+ .doesNotThrowAnyException();
+ }
+ }
+
+ @Test
+ public void testPartialUpdateRejectsMalformedSequenceGroupOption() {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "payload", DataTypes.INT()),
+ new DataField(2, "ts", DataTypes.INT()));
+ Map<String, String> options = new HashMap<>();
+ options.put(BUCKET.key(), "1");
+ options.put(CoreOptions.MERGE_ENGINE.key(), "partial-update");
+ options.put("fields.ts.xsequence-group", "payload");
+ TableSchema schema =
+ new TableSchema(1, fields, 10, emptyList(),
singletonList("id"), options, "");
+
+ assertThatThrownBy(() -> validateTableSchema(schema))
+ .hasMessageContaining("Invalid sequence-group option:
fields.ts.xsequence-group");
+ }
+
@Test
public void testPrimaryKeyBlobViewCoexistsWithManagedBlob() {
List<DataField> fields =
@@ -466,12 +635,31 @@ class SchemaValidationTest {
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].");
+
+ assertThatCode(
() ->
validateTableSchema(
primaryKeyBlobSchema(
mergeOptions,
singletonList("id"), emptyList())))
- .hasMessage(
- "Primary-key managed BLOB tables only support the
deduplicate merge engine.");
+ .doesNotThrowAnyException();
Map<String, String> 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 0000000000..c5ef165282
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyPartialUpdateBlobTest.java
@@ -0,0 +1,441 @@
+/*
+ * 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.blob.ManagedBlobReferenceFile;
+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.io.DataFilePathFactory;
+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.apache.paimon.types.RowKind;
+
+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<InternalRow> 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<DataFileMeta> 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 testPartialUpdateManagedBlobLastValueRetract() throws
Exception {
+ String tableName = "pk_pu_managed_blob_retract";
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("payload", DataTypes.BLOB())
+ .column("ts", DataTypes.INT())
+ .primaryKey("id")
+ .option(CoreOptions.MERGE_ENGINE.key(),
"partial-update")
+ .option(CoreOptions.BLOB_FIELD.key(), "payload")
+ .option(CoreOptions.CHANGELOG_PRODUCER.key(), "none")
+ .option("fields.ts.sequence-group", "payload")
+ .option("fields.payload.aggregate-function",
"last_value")
+ .option(CoreOptions.BUCKET.key(), "1")
+ .build();
+ catalog.createTable(identifier(tableName), schema, true);
+ FileStoreTable table = getTable(identifier(tableName));
+
+ byte[] payload = new byte[] {1, 2, 3};
+ try (StreamTableWrite write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser)) {
+ write.write(GenericRow.of(1, new BlobData(payload), 1));
+ write.write(GenericRow.of(2, new BlobData(payload), 2));
+ commit.commit(0, write.prepareCommit(false, 0));
+
+ write.write(GenericRow.ofKind(RowKind.DELETE, 1, new
BlobData(payload), 2));
+ write.write(GenericRow.ofKind(RowKind.DELETE, 2, new
BlobData(payload), 1));
+ commit.commit(1, write.prepareCommit(false, 1));
+ }
+
+ List<InternalRow> rows = readRows(table.newReadBuilder());
+ assertLastValueRetractedRows(rows);
+
+ compact(table, BinaryRow.EMPTY_ROW, 0);
+ rows = readRows(table.newReadBuilder());
+ assertLastValueRetractedRows(rows);
+ assertThat(managedBlobReferences(table)).isEmpty();
+ }
+
+ @Test
+ public void testPartialUpdateManagedBlobCollectionsIgnoreRetract() throws
Exception {
+ String tableName = "pk_pu_managed_blob_collections_ignore_retract";
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("payload", DataTypes.BLOB())
+ .column("payloads", DataTypes.ARRAY(DataTypes.BLOB()))
+ .column("assets", DataTypes.MAP(DataTypes.STRING(),
DataTypes.BLOB()))
+ .column("ts", DataTypes.INT())
+ .primaryKey("id")
+ .option(CoreOptions.MERGE_ENGINE.key(),
"partial-update")
+ .option(CoreOptions.BLOB_FIELD.key(),
"payload,payloads,assets")
+ .option(CoreOptions.CHANGELOG_PRODUCER.key(), "none")
+ .option("fields.ts.sequence-group",
"payload,payloads,assets")
+ .option("fields.default-aggregate-function",
"last_non_null_value")
+ .option("fields.payload.ignore-retract", "true")
+ .option("fields.payloads.ignore-retract", "true")
+ .option("fields.assets.ignore-retract", "true")
+ .option(CoreOptions.BUCKET.key(), "1")
+ .build();
+ catalog.createTable(identifier(tableName), schema, true);
+ FileStoreTable table = getTable(identifier(tableName));
+
+ byte[] payload = new byte[] {1, 2, 3};
+ try (StreamTableWrite write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser)) {
+ write.write(
+ GenericRow.of(
+ 1,
+ new BlobData(payload),
+ new GenericArray(new Object[] {new
BlobData(payload)}),
+ blobMap("payload", payload),
+ 1));
+ commit.commit(0, write.prepareCommit(false, 0));
+
+ write.write(
+ GenericRow.ofKind(
+ RowKind.DELETE,
+ 1,
+ new BlobData(payload),
+ new GenericArray(new Object[] {new
BlobData(payload)}),
+ blobMap("payload", payload),
+ 2));
+ commit.commit(1, write.prepareCommit(false, 1));
+ }
+
+ assertManagedIgnoreRetractRow(table, payload);
+ compact(table, BinaryRow.EMPTY_ROW, 0);
+ assertManagedIgnoreRetractRow(table, payload);
+ }
+
+ @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<Integer, Long> 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<InternalRow> 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<InternalRow> readRows(ReadBuilder readBuilder) throws
Exception {
+ RecordReader<InternalRow> 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<InternalRow> 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<BinaryString, Blob> blobs = new LinkedHashMap<>();
+ blobs.put(BinaryString.fromString(key), new BlobData(value));
+ return new GenericMap(blobs);
+ }
+
+ private void assertManagedIgnoreRetractRow(FileStoreTable table, byte[]
expected)
+ throws Exception {
+ List<InternalRow> rows = readRows(table.newReadBuilder());
+ assertThat(rows).hasSize(1);
+ InternalRow row = rows.get(0);
+ assertThat(row.getBlob(1).toData()).isEqualTo(expected);
+ assertThat(row.getArray(2).getBlob(0).toData()).isEqualTo(expected);
+
assertThat(row.getMap(3).valueArray().getBlob(0).toData()).isEqualTo(expected);
+ assertThat(row.getInt(4)).isEqualTo(2);
+ }
+
+ private void assertLastValueRetractedRows(List<InternalRow> rows) {
+ assertThat(rows).hasSize(2);
+ assertThat(rows).allMatch(row -> row.isNullAt(1));
+ assertThat(rows).allMatch(row -> row.getInt(2) == 2);
+ }
+
+ private List<ManagedBlobReferenceFile.Reference>
managedBlobReferences(FileStoreTable table)
+ throws Exception {
+ List<DataFileMeta> dataFiles = listDataFiles(table);
+ assertThat(dataFiles).hasSize(1);
+ DataFileMeta dataFile = dataFiles.get(0);
+ String referenceFile =
+ dataFile.extraFiles().stream()
+ .filter(
+ file ->
+ file.endsWith(
+
ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("Missing managed
BLOB sidecar."));
+ DataFilePathFactory pathFactory =
+
table.store().pathFactory().createDataFilePathFactory(BinaryRow.EMPTY_ROW, 0);
+ return ManagedBlobReferenceFile.read(
+ table.fileIO(), pathFactory.toAlignedPath(referenceFile,
dataFile));
+ }
+
+ private List<DataFileMeta> listDataFiles(FileStoreTable table) {
+ return table.newSnapshotReader().read().dataSplits().stream()
+ .flatMap(split -> ((DataSplit) split).dataFiles().stream())
+ .collect(Collectors.toList());
+ }
+
+ private List<InternalRow> read(RecordReader<InternalRow> reader) throws
Exception {
+ List<InternalRow> 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 0000000000..6dbd655740
--- /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 c1fcc2ca09..681821f190 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.api.Test;
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 class PartialUpdateITCase extends
CatalogITCaseBase {
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;
+ }
}
