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 ad2e76668f [flink] Support managed BLOB updates in data-evolution
MERGE INTO (#10514)
ad2e76668f is described below
commit ad2e76668f61f484358965730d75fac56b56f2f6
Author: Wenchao Wu <[email protected]>
AuthorDate: Fri Oct 16 13:17:25 2026 +0800
[flink] Support managed BLOB updates in data-evolution MERGE INTO (#10514)
---
docs/docs/multimodal-table/blob.mdx | 3 +-
docs/docs/multimodal-table/data-evolution.mdx | 10 +-
.../main/java/org/apache/paimon/CoreOptions.java | 11 -
.../org/apache/paimon/flink/FlinkRowWrapper.java | 84 +++-
.../flink/action/DataEvolutionMergeIntoAction.java | 127 ++++-
.../DataEvolutionPartialWriteOperator.java | 324 +++++++++++-
.../action/DataEvolutionMergeIntoActionITCase.java | 554 +++++++++++++++++++--
7 files changed, 1006 insertions(+), 107 deletions(-)
diff --git a/docs/docs/multimodal-table/blob.mdx
b/docs/docs/multimodal-table/blob.mdx
index 1809bb95fa..138624ef8a 100644
--- a/docs/docs/multimodal-table/blob.mdx
+++ b/docs/docs/multimodal-table/blob.mdx
@@ -510,7 +510,7 @@ Partial-column BLOB updates differ by engine:
| Engine | Managed BLOB fields | Descriptor and view fields |
| --- | --- | --- |
| Spark `MERGE INTO` | Supports scalar BLOBs, BLOB arrays, and BLOB maps,
including updates to `NULL`. | Supports scalar fields. |
-| Flink `sys.merge_into` | Rejects updates to raw-data BLOB fields, including
arrays and maps. Other columns in the same table can still be updated. |
Supports scalar fields. |
+| Flink `sys.data_evolution_merge_into` | Supports scalar BLOBs, BLOB arrays,
and BLOB maps, including updates to `NULL`. Unchanged rows keep their previous
payload. | Supports scalar fields. |
For example, Spark can replace the payload for an existing row from the earlier
`image_table` example:
@@ -539,6 +539,7 @@ for Python partial writes.
6. **Collection Storage Mode**: `ARRAY<BLOB>` and `MAP<K, BLOB>` are supported
only for raw managed blob storage (`__BLOB_FIELD`), not for
`__BLOB_DESCRIPTOR_FIELD` or `__BLOB_VIEW_FIELD`.
7. **Append Write Buffer**: Writers that write BLOB payload files ignore
`write-buffer-for-append`
and write directly, because the buffer cannot preserve descriptor source
readers.
+8. **Flink Merge Sort Buffer**: `sys.data_evolution_merge_into` sorts updated
rows before writing. A managed BLOB value is inlined into that sort buffer. A
single row larger than `write-buffer-size` (256 MB by default) fails the job.
Raise `write-buffer-size` when updating large payloads. Video-frame fields
cannot be updated by this procedure.
## Best Practices
diff --git a/docs/docs/multimodal-table/data-evolution.mdx
b/docs/docs/multimodal-table/data-evolution.mdx
index 2a3eca3f3a..c57b95a065 100644
--- a/docs/docs/multimodal-table/data-evolution.mdx
+++ b/docs/docs/multimodal-table/data-evolution.mdx
@@ -234,9 +234,13 @@ CALL sys.data_evolution_merge_into(
```
Use `''` for unused optional positional arguments. Updating partition columns
-is not supported. For BLOB updates, this Flink procedure supports
-[descriptor-only](./blob-references#descriptor-only-storage) and [BLOB
view](./blob-references#blob-view)
-fields; it rejects raw-data BLOB fields, including BLOB arrays and maps.
+is not supported. For BLOB updates, this Flink procedure supports managed
+raw-data fields (`BLOB`, `ARRAY<BLOB>`, and `MAP<K, BLOB>`, including updates
+to `NULL`), [descriptor-only](./blob-references#descriptor-only-storage)
fields,
+and scalar [BLOB view](./blob-references#blob-view) fields. Unchanged managed
+BLOB payloads stay in the previous blob file. Video-frame fields cannot be
+updated. A managed BLOB value is inlined into the merge sort buffer, so a
+single row larger than `write-buffer-size` (256 MB by default) fails the job.
### Python API
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index b350e3b2b2..3bca19e557 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -3858,17 +3858,6 @@ public class CoreOptions implements Serializable {
return fields;
}
- /**
- * Returns the set of BLOB fields that support partial updates (e.g. via
MERGE INTO).
- *
- * <p>Currently, only descriptor-based BLOB fields (configured via {@link
- * #BLOB_DESCRIPTOR_FIELD}) are updatable. Raw-data BLOB fields are not
updatable because the
- * update cost is too high.
- */
- public Set<String> updatableBlobFields() {
- return blobInlineField();
- }
-
private Set<String> parseCommaSeparatedSet(ConfigOption<String> option) {
return options.getOptional(option)
.map(
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowWrapper.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowWrapper.java
index 74487659a0..5cd2ca41ac 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowWrapper.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowWrapper.java
@@ -21,6 +21,7 @@ package org.apache.paimon.flink;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.Blob;
+import org.apache.paimon.data.BlobData;
import org.apache.paimon.data.BlobDescriptor;
import org.apache.paimon.data.Decimal;
import org.apache.paimon.data.InternalArray;
@@ -64,6 +65,8 @@ public class FlinkRowWrapper implements InternalRow {
private final boolean writeNullOnFetchFailure;
private final Set<Integer> blobFields;
private final Set<Integer> materializedBlobFields;
+ /** When true, blob getters return the stored bytes and do not open a URI
reader. */
+ private final boolean carryBlobBytes;
public FlinkRowWrapper(org.apache.flink.table.data.RowData row) {
this(row, null);
@@ -113,7 +116,8 @@ public class FlinkRowWrapper implements InternalRow {
checkBlobDescriptorExists,
writeNullOnFetchFailure,
blobFields,
- Collections.emptySet());
+ Collections.emptySet(),
+ false);
}
public static FlinkRowWrapper fromUriReaderFactory(
@@ -138,13 +142,32 @@ public class FlinkRowWrapper implements InternalRow {
boolean writeNullOnFetchFailure,
Set<Integer> blobFields,
Set<Integer> materializedBlobFields) {
+ return fromUriReaderFactory(
+ row,
+ uriReaderFactory,
+ checkBlobDescriptorExists,
+ writeNullOnFetchFailure,
+ blobFields,
+ materializedBlobFields,
+ false);
+ }
+
+ public static FlinkRowWrapper fromUriReaderFactory(
+ org.apache.flink.table.data.RowData row,
+ UriReaderFactory uriReaderFactory,
+ boolean checkBlobDescriptorExists,
+ boolean writeNullOnFetchFailure,
+ Set<Integer> blobFields,
+ Set<Integer> materializedBlobFields,
+ boolean carryBlobBytes) {
return new FlinkRowWrapper(
row,
uriReaderFactory,
checkBlobDescriptorExists,
writeNullOnFetchFailure,
blobFields,
- materializedBlobFields);
+ materializedBlobFields,
+ carryBlobBytes);
}
private FlinkRowWrapper(
@@ -153,13 +176,15 @@ public class FlinkRowWrapper implements InternalRow {
boolean checkBlobDescriptorExists,
boolean writeNullOnFetchFailure,
Set<Integer> blobFields,
- Set<Integer> materializedBlobFields) {
+ Set<Integer> materializedBlobFields,
+ boolean carryBlobBytes) {
this.row = row;
this.uriReaderFactory = uriReaderFactory;
this.checkBlobDescriptorExists = checkBlobDescriptorExists;
this.writeNullOnFetchFailure = writeNullOnFetchFailure;
this.blobFields = blobFields;
this.materializedBlobFields = materializedBlobFields;
+ this.carryBlobBytes = carryBlobBytes;
}
public static Set<Integer>
blobFieldIndexes(org.apache.paimon.types.RowType rowType) {
@@ -262,7 +287,21 @@ public class FlinkRowWrapper implements InternalRow {
@Override
public Blob getBlob(int pos) {
- byte[] bytes = row.getBinary(pos);
+ return blobFromStoredBytes(row.getBinary(pos), uriReaderFactory,
carryBlobBytes);
+ }
+
+ /**
+ * Sort inlines whatever {@link #getBlob} returns. Returning the stored
bytes leaves descriptor
+ * resolution to the blob writer, which applies the fetch-failure NULL
policy.
+ */
+ private static Blob blobFromStoredBytes(
+ byte[] bytes, @Nullable UriReaderFactory uriReaderFactory, boolean
carryBlobBytes) {
+ if (bytes == null) {
+ return null;
+ }
+ if (carryBlobBytes) {
+ return new BlobData(bytes);
+ }
return Blob.fromBytes(bytes, uriReaderFactory, null);
}
@@ -344,17 +383,17 @@ public class FlinkRowWrapper implements InternalRow {
@Override
public InternalArray getArray(int pos) {
- return new FlinkArrayWrapper(row.getArray(pos), uriReaderFactory);
+ return new FlinkArrayWrapper(row.getArray(pos), uriReaderFactory,
carryBlobBytes);
}
@Override
public InternalVector getVector(int pos) {
- return new FlinkVectorWrapper(row.getArray(pos), uriReaderFactory);
+ return new FlinkVectorWrapper(row.getArray(pos), uriReaderFactory,
carryBlobBytes);
}
@Override
public InternalMap getMap(int pos) {
- return new FlinkMapWrapper(row.getMap(pos), uriReaderFactory);
+ return new FlinkMapWrapper(row.getMap(pos), uriReaderFactory,
carryBlobBytes);
}
@Override
@@ -366,16 +405,19 @@ public class FlinkRowWrapper implements InternalRow {
private final org.apache.flink.table.data.ArrayData array;
@Nullable private final UriReaderFactory uriReaderFactory;
+ private final boolean carryBlobBytes;
private FlinkArrayWrapper(org.apache.flink.table.data.ArrayData array)
{
- this(array, null);
+ this(array, null, false);
}
private FlinkArrayWrapper(
org.apache.flink.table.data.ArrayData array,
- @Nullable UriReaderFactory uriReaderFactory) {
+ @Nullable UriReaderFactory uriReaderFactory,
+ boolean carryBlobBytes) {
this.array = array;
this.uriReaderFactory = uriReaderFactory;
+ this.carryBlobBytes = carryBlobBytes;
}
@Override
@@ -452,22 +494,22 @@ public class FlinkRowWrapper implements InternalRow {
@Override
public Blob getBlob(int pos) {
- return Blob.fromBytes(array.getBinary(pos), uriReaderFactory,
null);
+ return blobFromStoredBytes(array.getBinary(pos), uriReaderFactory,
carryBlobBytes);
}
@Override
public InternalArray getArray(int pos) {
- return new FlinkArrayWrapper(array.getArray(pos),
uriReaderFactory);
+ return new FlinkArrayWrapper(array.getArray(pos),
uriReaderFactory, carryBlobBytes);
}
@Override
public InternalVector getVector(int pos) {
- return new FlinkVectorWrapper(array.getArray(pos),
uriReaderFactory);
+ return new FlinkVectorWrapper(array.getArray(pos),
uriReaderFactory, carryBlobBytes);
}
@Override
public InternalMap getMap(int pos) {
- return new FlinkMapWrapper(array.getMap(pos), uriReaderFactory);
+ return new FlinkMapWrapper(array.getMap(pos), uriReaderFactory,
carryBlobBytes);
}
@Override
@@ -518,8 +560,9 @@ public class FlinkRowWrapper implements InternalRow {
private FlinkVectorWrapper(
org.apache.flink.table.data.ArrayData array,
- @Nullable UriReaderFactory uriReaderFactory) {
- super(array, uriReaderFactory);
+ @Nullable UriReaderFactory uriReaderFactory,
+ boolean carryBlobBytes) {
+ super(array, uriReaderFactory, carryBlobBytes);
}
}
@@ -527,16 +570,19 @@ public class FlinkRowWrapper implements InternalRow {
private final org.apache.flink.table.data.MapData map;
@Nullable private final UriReaderFactory uriReaderFactory;
+ private final boolean carryBlobBytes;
private FlinkMapWrapper(org.apache.flink.table.data.MapData map) {
- this(map, null);
+ this(map, null, false);
}
private FlinkMapWrapper(
org.apache.flink.table.data.MapData map,
- @Nullable UriReaderFactory uriReaderFactory) {
+ @Nullable UriReaderFactory uriReaderFactory,
+ boolean carryBlobBytes) {
this.map = map;
this.uriReaderFactory = uriReaderFactory;
+ this.carryBlobBytes = carryBlobBytes;
}
@Override
@@ -546,12 +592,12 @@ public class FlinkRowWrapper implements InternalRow {
@Override
public InternalArray keyArray() {
- return new FlinkArrayWrapper(map.keyArray(), uriReaderFactory);
+ return new FlinkArrayWrapper(map.keyArray(), uriReaderFactory,
carryBlobBytes);
}
@Override
public InternalArray valueArray() {
- return new FlinkArrayWrapper(map.valueArray(), uriReaderFactory);
+ return new FlinkArrayWrapper(map.valueArray(), uriReaderFactory,
carryBlobBytes);
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java
index da95b4f36e..46b2b5dff3 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java
@@ -35,16 +35,20 @@ import org.apache.paimon.flink.sorter.SortOperator;
import org.apache.paimon.flink.utils.FlinkCalciteClasses;
import org.apache.paimon.flink.utils.InternalTypeInfo;
import org.apache.paimon.manifest.ManifestCommittable;
+import org.apache.paimon.table.BlobDescriptorReaderFactory;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.SpecialFields;
+import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.BlobType;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypeCasts;
import org.apache.paimon.types.DataTypeFamily;
import org.apache.paimon.types.DataTypeRoot;
+import org.apache.paimon.types.MapType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Preconditions;
+import org.apache.paimon.utils.UriReaderFactory;
import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.functions.RichMapFunction;
@@ -680,8 +684,30 @@ public class DataEvolutionMergeIntoAction extends
TableActionBase {
// 1. sort data by row id
InternalTypeInfo<InternalRow> typeInfo =
InternalTypeInfo.fromRowType(rowType);
RowType sortType = rowType.project(SpecialFields.ROW_ID.name());
+ FileStoreTable storeTable = (FileStoreTable) table;
+ UriReaderFactory blobDescriptorReaderFactory =
+ BlobDescriptorReaderFactory.create(storeTable);
+ boolean nullOnMissingFile = coreOptions.blobWriteNullOnMissingFile();
+ boolean nullOnFetchFailure = coreOptions.blobWriteNullOnFetchFailure();
+ Set<Integer> blobFields =
+ nullOnMissingFile
+ ? FlinkRowWrapper.blobFieldIndexes(rowType)
+ : Collections.emptySet();
+ Set<Integer> materializedBlobFields =
materializedBlobFieldIndexes(rowType);
+ // Carry descriptor bytes through the sort. Resolving the URI here
throws before the blob
+ // writer can apply blob-write-null-on-fetch-failure.
DataStream<InternalRow> sorted =
- shuffled.map(t -> new FlinkRowWrapper(t.f1), typeInfo)
+ shuffled.map(
+ t ->
+ FlinkRowWrapper.fromUriReaderFactory(
+ t.f1,
+ blobDescriptorReaderFactory,
+ nullOnMissingFile,
+ nullOnFetchFailure,
+ blobFields,
+ materializedBlobFields,
+ true),
+ typeInfo)
.setParallelism(sinkParallelism)
.transform(
"SORT BY _ROW_ID",
@@ -702,7 +728,11 @@ public class DataEvolutionMergeIntoAction extends
TableActionBase {
"PARTIAL WRITE COLUMNS",
new CommittableTypeInfo(),
new DataEvolutionPartialWriteOperator(
- (FileStoreTable) table, rowType, writePaths,
baseSnapshotId))
+ storeTable,
+ rowType,
+ writePaths,
+ baseSnapshotId,
+ blobDescriptorReaderFactory))
.setParallelism(sinkParallelism);
}
@@ -811,8 +841,10 @@ public class DataEvolutionMergeIntoAction extends
TableActionBase {
.collect(Collectors.toMap(DataField::name,
Function.identity()));
List<String> partitionKeys = ((FileStoreTable)
table).schema().partitionKeys();
- // Get updatable BLOB fields (descriptor-based fields that support
partial updates)
- Set<String> updatableBlobFields = coreOptions.updatableBlobFields();
+ // Inline descriptor and view fields are rewritten as ordinary bytes.
Managed raw BLOB
+ // fields are rewritten into a blob file, with placeholders for
unchanged rows.
+ Set<String> inlineBlobFields = coreOptions.blobInlineField();
+ Set<String> videoFrameFields = coreOptions.videoFrameFields();
List<Column> flinkColumns = source.getResolvedSchema().getColumns();
boolean foundRowIdColumn = false;
@@ -833,31 +865,36 @@ public class DataEvolutionMergeIntoAction extends
TableActionBase {
throw new IllegalStateException(
"Column not found in target table: " +
flinkColumn.getName());
}
- if (BlobType.isBlobFileField(targetField.type())
- &&
!updatableBlobFields.contains(flinkColumn.getName())) {
+ if (videoFrameFields.contains(flinkColumn.getName())) {
throw new IllegalStateException(
- "Should not append/update raw-data BLOB,
ARRAY<BLOB> or MAP<X, BLOB> column '"
- + flinkColumn.getName()
- + "' through MERGE INTO. "
- + "Only descriptor-based BLOB columns
(configured via '"
- + CoreOptions.BLOB_DESCRIPTOR_FIELD.key()
- + "' or '"
- + CoreOptions.BLOB_VIEW_FIELD.key()
- + "') can be updated.");
+ String.format(
+ "Cannot update video-frame field '%s'
through data-evolution MERGE INTO. "
+ + "Video fields require a
VideoFrameDescriptor, and BYTES input is not a video frame. "
+ + "The field is configured by
'%s'.",
+ flinkColumn.getName(),
CoreOptions.VIDEO_FRAME_FIELD.key()));
}
-
DataType paimonType =
LogicalTypeConversion.toDataType(
flinkColumn.getDataType().getLogicalType());
- // For descriptor-based BLOB fields, allow BYTES
(VARBINARY/BINARY) source
- // since BLOB is represented as BYTES in Flink SQL
- boolean blobCompatible =
- targetField.type().getTypeRoot() == DataTypeRoot.BLOB
- &&
updatableBlobFields.contains(flinkColumn.getName())
- && paimonType
- .getTypeRoot()
- .getFamilies()
-
.contains(DataTypeFamily.BINARY_STRING);
+ boolean inlineBlob =
inlineBlobFields.contains(flinkColumn.getName());
+ boolean rawBlob = BlobType.isBlobFileField(targetField.type())
&& !inlineBlob;
+ // Flink SQL exposes BLOB as BYTES. Descriptor and view fields
accept binary
+ // input. Managed raw fields also accept BYTES, ARRAY<BYTES>,
and MAP<K, BYTES>.
+ boolean blobCompatible = false;
+ if (inlineBlob && targetField.type().getTypeRoot() ==
DataTypeRoot.BLOB) {
+ blobCompatible = isBinaryType(paimonType);
+ } else if (rawBlob) {
+ blobCompatible = isRawBlobSourceCompatible(paimonType,
targetField.type());
+ if (!blobCompatible) {
+ throw new IllegalStateException(
+ String.format(
+ "Cannot update managed BLOB column
'%s' through MERGE INTO: "
+ + "source type %s is not
compatible with %s. "
+ + "Use BYTES for BLOB,
ARRAY<BYTES> for ARRAY<BLOB>, "
+ + "or MAP<K, BYTES> for MAP<K,
BLOB>.",
+ flinkColumn.getName(), paimonType,
targetField.type()));
+ }
+ }
// Struct columns need a structural compatibility check:
DataTypeCasts does not
// support ROW-to-ROW casts. For a sub-field write (dotted
paths like nest.a) the
// source is a partial (subset) struct carrying only the
updated sub-fields, so a
@@ -890,6 +927,48 @@ public class DataEvolutionMergeIntoAction extends
TableActionBase {
}
}
+ /**
+ * Managed raw BLOB columns copy descriptor payloads in the blob writer.
Skip the wrapper's
+ * existence preflight for those columns so an HTTP descriptor is fetched
once, by the writer.
+ * Inline descriptor and view columns stay in the preflight.
+ */
+ private Set<Integer> materializedBlobFieldIndexes(RowType rowType) {
+ Set<String> inlineBlobFields = coreOptions.blobInlineField();
+ Set<Integer> materializedBlobFields =
FlinkRowWrapper.blobFieldIndexes(rowType);
+ materializedBlobFields.removeIf(
+ pos ->
inlineBlobFields.contains(rowType.getFields().get(pos).name()));
+ return materializedBlobFields;
+ }
+
+ private static boolean isBinaryType(DataType type) {
+ return
type.getTypeRoot().getFamilies().contains(DataTypeFamily.BINARY_STRING);
+ }
+
+ /**
+ * Whether a Flink source value can be written into a managed raw BLOB
column. Flink SQL has no
+ * native BLOB type, so payloads arrive as BYTES, ARRAY<BYTES>, or
MAP<K, BYTES>.
+ */
+ private static boolean isRawBlobSourceCompatible(DataType source, DataType
target) {
+ if (target.getTypeRoot() == DataTypeRoot.BLOB) {
+ return isBinaryType(source);
+ }
+ if (target.getTypeRoot() == DataTypeRoot.ARRAY) {
+ return source.getTypeRoot() == DataTypeRoot.ARRAY
+ && isBinaryType(((ArrayType) source).getElementType());
+ }
+ if (target.getTypeRoot() == DataTypeRoot.MAP) {
+ if (source.getTypeRoot() != DataTypeRoot.MAP) {
+ return false;
+ }
+ MapType sourceMap = (MapType) source;
+ MapType targetMap = (MapType) target;
+ return DataTypeCasts.supportsCompatibleCast(
+ sourceMap.getKeyType(), targetMap.getKeyType())
+ && isBinaryType(sourceMap.getValueType());
+ }
+ return false;
+ }
+
/**
* Whether {@code part} is a valid partial (subset) of the full struct
{@code full}: every
* sub-field of {@code part} must exist in {@code full} (by name) with a
compatible cast.
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionPartialWriteOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionPartialWriteOperator.java
index 7b1e79604f..c44a450bb9 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionPartialWriteOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionPartialWriteOperator.java
@@ -20,7 +20,18 @@ package org.apache.paimon.flink.dataevolution;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Blob;
+import org.apache.paimon.data.BlobArrayPlaceholder;
+import org.apache.paimon.data.BlobMapPlaceholder;
+import org.apache.paimon.data.BlobPlaceholder;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.data.InternalMap;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.InternalVector;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.data.variant.Variant;
import org.apache.paimon.flink.sink.Committable;
import org.apache.paimon.flink.utils.BoundedOneInputOperator;
import org.apache.paimon.io.CompactIncrement;
@@ -36,12 +47,19 @@ import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.sink.TableWriteImpl;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.InnerTableRead;
+import org.apache.paimon.types.BlobType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeRoot;
+import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.BlobDescriptorResolvingRow;
import org.apache.paimon.utils.CloseableIterator;
import org.apache.paimon.utils.CommitIncrement;
import org.apache.paimon.utils.Preconditions;
import org.apache.paimon.utils.ProjectedRow;
import org.apache.paimon.utils.RecordWriter;
+import org.apache.paimon.utils.UriReaderFactory;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.slf4j.Logger;
@@ -51,8 +69,10 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.TreeSet;
import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
@@ -70,6 +90,7 @@ public class DataEvolutionPartialWriteOperator
private final FileStoreTable table;
private final Long baseSnapshotId;
+ private final UriReaderFactory blobDescriptorReaderFactory;
private final boolean nestedFieldEnabled;
// dataType
@@ -79,6 +100,14 @@ public class DataEvolutionPartialWriteOperator
// data type excludes of _ROW_ID field.
private final RowType writeType;
+ // Original-file read type. Raw blob columns are omitted so gap rows do
not pull payloads.
+ private final RowType originalReadType;
+ private final int originalRowIdIndex;
+ private final boolean hasRawBlobFields;
+ // writeType index -> originalReadType index, or -1 when the field is a
raw blob placeholder.
+ private final int[] rawBlobGapMapping;
+ private final Object[] rawBlobGapPlaceholders;
+
// --------------------- transient fields ---------------------------
private transient List<Committable> committables;
@@ -94,9 +123,12 @@ public class DataEvolutionPartialWriteOperator
private static Map<String, String> dataEvolutionWriteOptions() {
// Data evolution requires a single output file: disable both size and
row rolling.
+ // Append buffering serializes the row before the blob writer sees it.
Placeholders cannot
+ // be serialized, and a buffer flush can close the file before this
range is finished.
Map<String, String> options = new HashMap<>();
options.put(CoreOptions.TARGET_FILE_SIZE.key(), "99999 G");
options.put(CoreOptions.TARGET_FILE_ROW_NUM.key(),
String.valueOf(Long.MAX_VALUE));
+ options.put(CoreOptions.WRITE_BUFFER_FOR_APPEND.key(), "false");
return options;
}
@@ -104,9 +136,11 @@ public class DataEvolutionPartialWriteOperator
FileStoreTable table,
RowType sourceType,
List<String> writePaths,
- Long baseSnapshotId) {
+ Long baseSnapshotId,
+ UriReaderFactory blobDescriptorReaderFactory) {
this.table = table.copy(dataEvolutionWriteOptions());
this.baseSnapshotId = baseSnapshotId;
+ this.blobDescriptorReaderFactory = blobDescriptorReaderFactory;
this.nestedFieldEnabled =
this.table.coreOptions().dataEvolutionNestedFieldEnabled();
this.writeType =
nestedFieldEnabled
@@ -116,6 +150,61 @@ public class DataEvolutionPartialWriteOperator
// carries the table's field ids, so it is used directly as the
read/data type.
this.dataType = sourceType;
this.rowIdIndex =
this.dataType.getFieldIndex(SpecialFields.ROW_ID.name());
+ Set<String> rawBlobFields = rawBlobFieldNames(this.writeType,
this.table.coreOptions());
+ this.hasRawBlobFields = !rawBlobFields.isEmpty();
+ this.originalReadType = readTypeWithoutRawBlobs(this.dataType,
rawBlobFields);
+ this.originalRowIdIndex =
this.originalReadType.getFieldIndex(SpecialFields.ROW_ID.name());
+ this.rawBlobGapMapping = new int[this.writeType.getFieldCount()];
+ this.rawBlobGapPlaceholders = new
Object[this.writeType.getFieldCount()];
+ for (int i = 0; i < this.writeType.getFieldCount(); i++) {
+ DataField field = this.writeType.getFields().get(i);
+ if (rawBlobFields.contains(field.name())) {
+ this.rawBlobGapMapping[i] = -1;
+ this.rawBlobGapPlaceholders[i] =
rawBlobPlaceholder(field.type());
+ } else {
+ int originalIndex =
this.originalReadType.getFieldIndex(field.name());
+ Preconditions.checkState(
+ originalIndex >= 0,
+ "Partial write field '%s' is missing from the original
read type.",
+ field.name());
+ this.rawBlobGapMapping[i] = originalIndex;
+ this.rawBlobGapPlaceholders[i] = null;
+ }
+ }
+ }
+
+ private static Set<String> rawBlobFieldNames(RowType writeType,
CoreOptions options) {
+ Set<String> inlineFields = options.blobInlineField();
+ Set<String> rawBlobFields = new HashSet<>();
+ for (DataField field : writeType.getFields()) {
+ if (BlobType.isBlobFileField(field.type()) &&
!inlineFields.contains(field.name())) {
+ rawBlobFields.add(field.name());
+ }
+ }
+ return rawBlobFields;
+ }
+
+ private static RowType readTypeWithoutRawBlobs(RowType dataType,
Set<String> rawBlobFields) {
+ if (rawBlobFields.isEmpty()) {
+ return dataType;
+ }
+ List<DataField> fields = new ArrayList<>();
+ for (DataField field : dataType.getFields()) {
+ if (!rawBlobFields.contains(field.name())) {
+ fields.add(field);
+ }
+ }
+ return new RowType(fields);
+ }
+
+ private static Object rawBlobPlaceholder(DataType type) {
+ if (type.getTypeRoot() == DataTypeRoot.ARRAY) {
+ return BlobArrayPlaceholder.INSTANCE;
+ }
+ if (type.getTypeRoot() == DataTypeRoot.MAP) {
+ return BlobMapPlaceholder.INSTANCE;
+ }
+ return BlobPlaceholder.INSTANCE;
}
@Override
@@ -150,7 +239,10 @@ public class DataEvolutionPartialWriteOperator
firstRowIdLookup = new FirstRowIdLookup(new ArrayList<>(rowIdSet));
// initialize table read & table write
- tableRead = table.newRead().withReadType(dataType);
+ // The split built below contains only non-blob files. Raw blob
columns must be removed
+ // from the read type; leaving them in would read null or fail.
Unchanged rows are written
+ // as placeholders and resolved from the previous blob file.
+ tableRead = table.newRead().withReadType(originalReadType);
@SuppressWarnings("unchecked")
TableWriteImpl<InternalRow> writeImpl =
(TableWriteImpl<InternalRow>)
@@ -240,7 +332,9 @@ public class DataEvolutionPartialWriteOperator
// writer to write new columns
private final RecordWriter<InternalRow> writer;
- private final ProjectedRow reusedRow;
+ private final ProjectedRow updatedRow;
+ private final BlobDescriptorResolvingRow descriptorRow;
+ private final RawBlobGapRow gapRow;
private final long firstRowId;
private final long rowCount;
private final BinaryRow partition;
@@ -255,12 +349,28 @@ public class DataEvolutionPartialWriteOperator
long rowCount) {
this.reader = reader.toCloseableIterator();
this.writer = writer;
- this.reusedRow = ProjectedRow.from(writeType, dataType);
+ this.updatedRow = ProjectedRow.from(writeType, dataType);
+ // Sort serializes descriptor BLOBs as bytes and BinaryRow.getBlob
reopens them with a
+ // null FileIO. Reattach the catalog-aware reader before the blob
writer copies bytes.
+ // The null index set is intentional: resolve every column in
source index space.
+ // ProjectedRow wraps this row, so its projected indexes must not
be passed here.
+ this.descriptorRow = new
BlobDescriptorResolvingRow(blobDescriptorReaderFactory, null);
+ this.gapRow =
+ hasRawBlobFields
+ ? new RawBlobGapRow(rawBlobGapMapping,
rawBlobGapPlaceholders)
+ : null;
this.firstRowId = firstRowId;
this.rowCount = rowCount;
this.partition = firstIdToPartition.get(firstRowId);
}
+ private InternalRow rowForOriginal(InternalRow originalRow) {
+ if (gapRow == null) {
+ return updatedRow.replaceRow(originalRow);
+ }
+ return gapRow.replace(originalRow);
+ }
+
void write(InternalRow row) throws Exception {
long currentRowId = row.getLong(rowIdIndex);
@@ -277,17 +387,17 @@ public class DataEvolutionPartialWriteOperator
InternalRow originalRow;
while (reader.hasNext()) {
originalRow = reader.next();
- long originalRowId = originalRow.getLong(rowIdIndex);
+ long originalRowId = originalRow.getLong(originalRowIdIndex);
if (originalRowId < currentRowId) {
- // new row is absent, we should use the original row
- reusedRow.replaceRow(originalRow);
- writer.write(reusedRow);
+ // Unchanged row. Raw blob fields are placeholders, not
copied payloads.
+ writer.write(rowForOriginal(originalRow));
writtenNum++;
} else if (originalRowId == currentRowId) {
- // new row is present, we should use the new row
- reusedRow.replaceRow(row);
- writer.write(reusedRow);
+ // new row is present, we should use the new row.
+ // descriptorRow is inside ProjectedRow and sees
unprojected source indexes.
+ updatedRow.replaceRow(descriptorRow.replace(row));
+ writer.write(updatedRow);
writtenNum++;
break;
} else {
@@ -304,8 +414,7 @@ public class DataEvolutionPartialWriteOperator
InternalRow row;
while (reader.hasNext()) {
row = reader.next();
- reusedRow.replaceRow(row);
- writer.write(reusedRow);
+ writer.write(rowForOriginal(row));
writtenNum++;
}
} finally {
@@ -321,10 +430,8 @@ public class DataEvolutionPartialWriteOperator
// 2. finish writer
try {
CommitIncrement written = writer.prepareCommit(false);
- List<DataFileMeta> fileMetas =
written.newFilesIncrement().newFiles();
- Preconditions.checkState(
- fileMetas.size() == 1, "This is a bug, Writer could
only produce one file");
- DataFileMeta fileMeta =
fileMetas.get(0).assignFirstRowId(firstRowId);
+ List<DataFileMeta> fileMetas =
+
assignFirstRowIds(written.newFilesIncrement().newFiles(), firstRowId);
CommitMessage commitMessage =
new CommitMessageImpl(
@@ -332,7 +439,7 @@ public class DataEvolutionPartialWriteOperator
0,
null,
new DataIncrement(
-
Collections.singletonList(fileMeta),
+ fileMetas,
Collections.emptyList(),
Collections.emptyList()),
CompactIncrement.emptyIncrement())
@@ -356,4 +463,185 @@ public class DataEvolutionPartialWriteOperator
return false;
}
}
+
+ /**
+ * Assigns first row ids for a partial write. A blob-only update produces
blob files and no
+ * normal file. A mixed update produces at most one normal file plus the
dedicated blob files.
+ * Blob files for the same column are ordered and cover the target row-id
range in sequence.
+ */
+ private static List<DataFileMeta> assignFirstRowIds(
+ List<DataFileMeta> dataFiles, long firstRowId) {
+ if (dataFiles.isEmpty()) {
+ throw new IllegalStateException(
+ "This is a bug, data-evolution partial write produced no
files.");
+ }
+ List<DataFileMeta> assigned = new ArrayList<>(dataFiles.size());
+ Map<String, Long> blobFieldStarts = new HashMap<>();
+ int normalFileCount = 0;
+ long vectorStoreStart = firstRowId;
+ for (DataFileMeta file : dataFiles) {
+ if (isBlobFile(file.fileName())) {
+ String blobFieldName = file.writeCols().get(0);
+ long blobStart =
+ blobFieldStarts.containsKey(blobFieldName)
+ ? blobFieldStarts.get(blobFieldName)
+ : firstRowId;
+ assigned.add(file.assignFirstRowId(blobStart));
+ blobFieldStarts.put(blobFieldName, blobStart +
file.rowCount());
+ } else if (isVectorStoreFile(file.fileName())) {
+ assigned.add(file.assignFirstRowId(vectorStoreStart));
+ vectorStoreStart += file.rowCount();
+ } else {
+ normalFileCount++;
+ if (normalFileCount > 1) {
+ throw new IllegalStateException(
+ "This is a bug: data-evolution partial write
should produce at most one "
+ + "normal file, but produced "
+ + normalFileCount
+ + " files. Files: "
+ + dataFiles);
+ }
+ assigned.add(file.assignFirstRowId(firstRowId));
+ }
+ }
+ return assigned;
+ }
+
+ /**
+ * Gap row for a managed BLOB partial write. Non-blob fields are copied
from the original file.
+ * Raw blob fields are placeholders so unchanged payloads stay in the
previous blob file.
+ */
+ private static final class RawBlobGapRow implements InternalRow {
+
+ private final int[] indexMapping;
+ private final Object[] placeholders;
+ private InternalRow row;
+
+ private RawBlobGapRow(int[] indexMapping, Object[] placeholders) {
+ this.indexMapping = indexMapping;
+ this.placeholders = placeholders;
+ }
+
+ private RawBlobGapRow replace(InternalRow row) {
+ this.row = row;
+ return this;
+ }
+
+ @Override
+ public int getFieldCount() {
+ return indexMapping.length;
+ }
+
+ @Override
+ public RowKind getRowKind() {
+ return row.getRowKind();
+ }
+
+ @Override
+ public void setRowKind(RowKind kind) {
+ row.setRowKind(kind);
+ }
+
+ @Override
+ public boolean isNullAt(int pos) {
+ if (indexMapping[pos] < 0) {
+ return false;
+ }
+ return row.isNullAt(indexMapping[pos]);
+ }
+
+ @Override
+ public boolean getBoolean(int pos) {
+ return row.getBoolean(indexMapping[pos]);
+ }
+
+ @Override
+ public byte getByte(int pos) {
+ return row.getByte(indexMapping[pos]);
+ }
+
+ @Override
+ public short getShort(int pos) {
+ return row.getShort(indexMapping[pos]);
+ }
+
+ @Override
+ public int getInt(int pos) {
+ return row.getInt(indexMapping[pos]);
+ }
+
+ @Override
+ public long getLong(int pos) {
+ return row.getLong(indexMapping[pos]);
+ }
+
+ @Override
+ public float getFloat(int pos) {
+ return row.getFloat(indexMapping[pos]);
+ }
+
+ @Override
+ public double getDouble(int pos) {
+ return row.getDouble(indexMapping[pos]);
+ }
+
+ @Override
+ public BinaryString getString(int pos) {
+ return row.getString(indexMapping[pos]);
+ }
+
+ @Override
+ public Decimal getDecimal(int pos, int precision, int scale) {
+ return row.getDecimal(indexMapping[pos], precision, scale);
+ }
+
+ @Override
+ public Timestamp getTimestamp(int pos, int precision) {
+ return row.getTimestamp(indexMapping[pos], precision);
+ }
+
+ @Override
+ public byte[] getBinary(int pos) {
+ return row.getBinary(indexMapping[pos]);
+ }
+
+ @Override
+ public Variant getVariant(int pos) {
+ return row.getVariant(indexMapping[pos]);
+ }
+
+ @Override
+ public Blob getBlob(int pos) {
+ if (indexMapping[pos] < 0) {
+ return (Blob) placeholders[pos];
+ }
+ return row.getBlob(indexMapping[pos]);
+ }
+
+ @Override
+ public InternalArray getArray(int pos) {
+ if (indexMapping[pos] < 0) {
+ return (InternalArray) placeholders[pos];
+ }
+ return row.getArray(indexMapping[pos]);
+ }
+
+ @Override
+ public InternalVector getVector(int pos) {
+ return row.getVector(indexMapping[pos]);
+ }
+
+ @Override
+ public InternalMap getMap(int pos) {
+ if (indexMapping[pos] < 0) {
+ return (InternalMap) placeholders[pos];
+ }
+ return row.getMap(indexMapping[pos]);
+ }
+
+ @Override
+ public InternalRow getRow(int pos, int numFields) {
+ return row.getRow(indexMapping[pos], numFields);
+ }
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java
index bcc25cc1bf..ff1480cd0c 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java
@@ -26,6 +26,7 @@ import org.apache.paimon.table.FileStoreTable;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.planner.factories.TestValuesTableFactory;
import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -35,6 +36,9 @@ import org.junit.jupiter.params.provider.MethodSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -715,8 +719,7 @@ public class DataEvolutionMergeIntoActionITCase extends
ActionITCaseBase {
}
@Test
- public void testUpdateRawBlobColumnThrowsError() throws Exception {
- // Create a table with raw-data BLOB column
+ public void testUpdateRawBlobColumn() throws Exception {
sEnv.executeSql(
buildDdl(
"RAW_BLOB_T",
@@ -730,12 +733,43 @@ public class DataEvolutionMergeIntoActionITCase extends
ActionITCaseBase {
put("blob-field", "picture");
}
}));
- insertInto("RAW_BLOB_T", "(1, 'name1', X'48656C6C6F')");
+ insertInto(
+ "RAW_BLOB_T",
+ "(1, 'name1', X'48656C6C6F')",
+ "(2, 'name2', X'5945')",
+ "(3, 'name3', X'414243')");
- // Create source table
sEnv.executeSql(
buildDdl(
"RAW_BLOB_S",
+ Arrays.asList("id INT", "name STRING", "picture
BYTES"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyMap()));
+ insertInto("RAW_BLOB_S", "(1, 'updated_name1', X'4E4557')");
+
+ builder(warehouse, database, "RAW_BLOB_T")
+ .withMergeCondition("RAW_BLOB_T.id=RAW_BLOB_S.id")
+ .withMatchedUpdateSet(
+ "RAW_BLOB_T.name=RAW_BLOB_S.name,
RAW_BLOB_T.picture=RAW_BLOB_S.picture")
+ .withSourceTable("RAW_BLOB_S")
+ .withSinkParallelism(1)
+ .build()
+ .run();
+
+ testBatchRead(
+ "SELECT id, name, picture FROM RAW_BLOB_T ORDER BY id",
+ Arrays.asList(
+ changelogRow("+I", 1, "updated_name1", new byte[] {78,
69, 87}),
+ changelogRow("+I", 2, "name2", new byte[] {89, 69}),
+ changelogRow("+I", 3, "name3", new byte[] {65, 66,
67})));
+ }
+
+ @Test
+ public void testUpdateRawBlobColumnToNull() throws Exception {
+ sEnv.executeSql(
+ buildDdl(
+ "RAW_BLOB_NULL_T",
Arrays.asList("id INT", "picture BYTES"),
Collections.emptyList(),
Collections.emptyList(),
@@ -746,25 +780,107 @@ public class DataEvolutionMergeIntoActionITCase extends
ActionITCaseBase {
put("blob-field", "picture");
}
}));
- insertInto("RAW_BLOB_S", "(1, X'4E4557')");
+ insertInto("RAW_BLOB_NULL_T", "(1, X'48656C6C6F')", "(2, X'5945')");
- DataEvolutionMergeIntoAction action =
- builder(warehouse, database, "RAW_BLOB_T")
- .withMergeCondition("RAW_BLOB_T.id=RAW_BLOB_S.id")
-
.withMatchedUpdateSet("RAW_BLOB_T.picture=RAW_BLOB_S.picture")
- .withSourceTable("RAW_BLOB_S")
- .withSinkParallelism(1)
- .build();
+ sEnv.executeSql(
+ buildDdl(
+ "RAW_BLOB_NULL_S",
+ Arrays.asList("id INT", "picture BYTES"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyMap()));
+ insertInto("RAW_BLOB_NULL_S", "(1, CAST(NULL AS BYTES))");
- Throwable t = Assertions.assertThrows(IllegalStateException.class, ()
-> action.run());
- Assertions.assertTrue(
- t.getMessage().contains("raw-data BLOB, ARRAY<BLOB> or MAP<X,
BLOB> column"),
- "Expected error about raw-data BLOB, ARRAY<BLOB> or MAP<X,
BLOB> column but got: "
- + t.getMessage());
+ builder(warehouse, database, "RAW_BLOB_NULL_T")
+ .withMergeCondition("RAW_BLOB_NULL_T.id=RAW_BLOB_NULL_S.id")
+
.withMatchedUpdateSet("RAW_BLOB_NULL_T.picture=RAW_BLOB_NULL_S.picture")
+ .withSourceTable("RAW_BLOB_NULL_S")
+ .withSinkParallelism(1)
+ .build()
+ .run();
+
+ testBatchRead(
+ "SELECT id, picture FROM RAW_BLOB_NULL_T ORDER BY id",
+ Arrays.asList(
+ changelogRow("+I", 1, null), changelogRow("+I", 2, new
byte[] {89, 69})));
}
@Test
- public void testUpdateRawMapBlobColumnThrowsError() throws Exception {
+ public void testUpdateRawArrayBlobColumn() throws Exception {
+ sEnv.executeSql(
+ buildDdl(
+ "RAW_ARRAY_BLOB_T",
+ Arrays.asList("id INT", "pictures ARRAY<BYTES>"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ new HashMap<String, String>() {
+ {
+ put(ROW_TRACKING_ENABLED.key(), "true");
+ put(DATA_EVOLUTION_ENABLED.key(), "true");
+ put("blob-field", "pictures");
+ }
+ }));
+ String targetDataId =
+ TestValuesTableFactory.registerData(
+ Arrays.asList(
+ Row.of(1, new byte[][] {new byte[] {72, 101,
108, 108, 111}}),
+ Row.of(2, new byte[][] {new byte[] {65, 66,
67}}),
+ Row.of(3, null)));
+ sEnv.executeSql(
+ String.format(
+ "CREATE TEMPORARY TABLE
RAW_ARRAY_BLOB_TARGET_SOURCE "
+ + "(id INT, pictures ARRAY<BYTES>) "
+ + "WITH ('connector'='values',
'bounded'='true', 'data-id'='%s')",
+ targetDataId))
+ .await();
+ sEnv.executeSql(
+ "INSERT INTO RAW_ARRAY_BLOB_T "
+ + "SELECT * FROM RAW_ARRAY_BLOB_TARGET_SOURCE")
+ .await();
+
+ sEnv.executeSql(
+ buildDdl(
+ "RAW_ARRAY_BLOB_S",
+ Arrays.asList("id INT", "pictures ARRAY<BYTES>"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyMap()));
+ String sourceDataId =
+ TestValuesTableFactory.registerData(
+ Arrays.asList(
+ Row.of(1, new byte[][] {new byte[] {78, 69,
87}, null}),
+ Row.of(3, new byte[0][])));
+ sEnv.executeSql(
+ String.format(
+ "CREATE TEMPORARY TABLE
RAW_ARRAY_BLOB_UPDATE_SOURCE "
+ + "(id INT, pictures ARRAY<BYTES>) "
+ + "WITH ('connector'='values',
'bounded'='true', 'data-id'='%s')",
+ sourceDataId))
+ .await();
+ sEnv.executeSql(
+ "INSERT INTO RAW_ARRAY_BLOB_S "
+ + "SELECT * FROM RAW_ARRAY_BLOB_UPDATE_SOURCE")
+ .await();
+
+ builder(warehouse, database, "RAW_ARRAY_BLOB_T")
+ .withMergeCondition("RAW_ARRAY_BLOB_T.id=RAW_ARRAY_BLOB_S.id")
+
.withMatchedUpdateSet("RAW_ARRAY_BLOB_T.pictures=RAW_ARRAY_BLOB_S.pictures")
+ .withSourceTable("RAW_ARRAY_BLOB_S")
+ .withSinkParallelism(1)
+ .build()
+ .run();
+
+ testBatchRead(
+ "SELECT id, CARDINALITY(pictures), pictures[1], pictures[2] "
+ + "FROM RAW_ARRAY_BLOB_T ORDER BY id",
+ Arrays.asList(
+ changelogRow("+I", 1, 2, new byte[] {78, 69, 87},
null),
+ changelogRow("+I", 2, 1, new byte[] {65, 66, 67},
null),
+ changelogRow("+I", 3, 0, null, null)));
+ }
+
+ @Test
+ public void testUpdateRawMapBlobColumn() throws Exception {
sEnv.executeSql(
buildDdl(
"RAW_MAP_BLOB_T",
@@ -778,7 +894,23 @@ public class DataEvolutionMergeIntoActionITCase extends
ActionITCaseBase {
put("blob-field", "pictures");
}
}));
- insertInto("RAW_MAP_BLOB_T", "(1, MAP['old', X'4F4C44'])");
+ Map<String, byte[]> oldPictures = new LinkedHashMap<>();
+ oldPictures.put("old", new byte[] {72, 101, 108, 108, 111});
+ Map<String, byte[]> keptPictures = new LinkedHashMap<>();
+ keptPictures.put("keep", new byte[] {65, 66, 67});
+ String targetDataId =
+ TestValuesTableFactory.registerData(
+ Arrays.asList(
+ Row.of(1, oldPictures), Row.of(2,
keptPictures), Row.of(3, null)));
+ sEnv.executeSql(
+ String.format(
+ "CREATE TEMPORARY TABLE
RAW_MAP_BLOB_TARGET_SOURCE "
+ + "(id INT, pictures MAP<STRING,
BYTES>) "
+ + "WITH ('connector'='values',
'bounded'='true', 'data-id'='%s')",
+ targetDataId))
+ .await();
+ sEnv.executeSql("INSERT INTO RAW_MAP_BLOB_T SELECT * FROM
RAW_MAP_BLOB_TARGET_SOURCE")
+ .await();
sEnv.executeSql(
buildDdl(
"RAW_MAP_BLOB_S",
@@ -786,20 +918,36 @@ public class DataEvolutionMergeIntoActionITCase extends
ActionITCaseBase {
Collections.emptyList(),
Collections.emptyList(),
Collections.emptyMap()));
- insertInto("RAW_MAP_BLOB_S", "(1, MAP['new', X'4E4557'])");
+ Map<String, byte[]> newPictures = new LinkedHashMap<>();
+ newPictures.put("new", new byte[] {78, 69, 87});
+ String sourceDataId =
+ TestValuesTableFactory.registerData(
+ Arrays.asList(Row.of(1, newPictures), Row.of(3, new
LinkedHashMap<>())));
+ sEnv.executeSql(
+ String.format(
+ "CREATE TEMPORARY TABLE
RAW_MAP_BLOB_UPDATE_SOURCE "
+ + "(id INT, pictures MAP<STRING,
BYTES>) "
+ + "WITH ('connector'='values',
'bounded'='true', 'data-id'='%s')",
+ sourceDataId))
+ .await();
+ sEnv.executeSql("INSERT INTO RAW_MAP_BLOB_S SELECT * FROM
RAW_MAP_BLOB_UPDATE_SOURCE")
+ .await();
- DataEvolutionMergeIntoAction action =
- builder(warehouse, database, "RAW_MAP_BLOB_T")
-
.withMergeCondition("RAW_MAP_BLOB_T.id=RAW_MAP_BLOB_S.id")
-
.withMatchedUpdateSet("RAW_MAP_BLOB_T.pictures=RAW_MAP_BLOB_S.pictures")
- .withSourceTable("RAW_MAP_BLOB_S")
- .withSinkParallelism(1)
- .build();
+ builder(warehouse, database, "RAW_MAP_BLOB_T")
+ .withMergeCondition("RAW_MAP_BLOB_T.id=RAW_MAP_BLOB_S.id")
+
.withMatchedUpdateSet("RAW_MAP_BLOB_T.pictures=RAW_MAP_BLOB_S.pictures")
+ .withSourceTable("RAW_MAP_BLOB_S")
+ .withSinkParallelism(1)
+ .build()
+ .run();
- Throwable t = Assertions.assertThrows(IllegalStateException.class, ()
-> action.run());
- Assertions.assertTrue(
- t.getMessage().contains("raw-data BLOB, ARRAY<BLOB> or MAP<X,
BLOB> column"),
- "Expected error about raw-data MAP<X, BLOB> column but got: "
+ t.getMessage());
+ testBatchRead(
+ "SELECT id, CARDINALITY(pictures), pictures['old'],
pictures['keep'], pictures['new'] "
+ + "FROM RAW_MAP_BLOB_T ORDER BY id",
+ Arrays.asList(
+ changelogRow("+I", 1, 1, null, null, new byte[] {78,
69, 87}),
+ changelogRow("+I", 2, 1, null, new byte[] {65, 66,
67}, null),
+ changelogRow("+I", 3, 0, null, null, null)));
}
@Test
@@ -1070,6 +1218,350 @@ public class DataEvolutionMergeIntoActionITCase extends
ActionITCaseBase {
testBatchRead("SELECT id, name FROM DESC_BLOB_T ORDER BY id",
expected);
}
+ @Test
+ public void testUpdateRawBlobColumnSplitsFilesAndKeepsUnchangedPayload()
throws Exception {
+ byte[] large = new byte[64 * 1024];
+ Arrays.fill(large, (byte) 7);
+ sEnv.executeSql(
+ buildDdl(
+ "BLOB_SPLIT_UPDATE_T",
+ Arrays.asList("id INT", "picture BYTES"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ new HashMap<String, String>() {
+ {
+ put(ROW_TRACKING_ENABLED.key(), "true");
+ put(DATA_EVOLUTION_ENABLED.key(), "true");
+ put("blob-field", "picture");
+ put(CoreOptions.BLOB_TARGET_FILE_SIZE.key(),
"1 b");
+ put("sink.parallelism", "1");
+ }
+ }));
+ String targetDataId =
+ TestValuesTableFactory.registerData(
+ Arrays.asList(
+ Row.of(1, new byte[] {1}),
+ Row.of(2, large),
+ Row.of(3, new byte[] {3})));
+ sEnv.executeSql(
+ String.format(
+ "CREATE TEMPORARY TABLE
BLOB_SPLIT_UPDATE_TARGET_SOURCE "
+ + "(id INT, picture BYTES) "
+ + "WITH ('connector'='values',
'bounded'='true', 'data-id'='%s')",
+ targetDataId))
+ .await();
+ sEnv.executeSql(
+ "INSERT INTO BLOB_SPLIT_UPDATE_T "
+ + "SELECT * FROM
BLOB_SPLIT_UPDATE_TARGET_SOURCE")
+ .await();
+ long sequenceBeforeUpdate = maxSequence("BLOB_SPLIT_UPDATE_T");
+
+ sEnv.executeSql(
+ buildDdl(
+ "BLOB_SPLIT_UPDATE_S",
+ Arrays.asList("id INT", "picture BYTES"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyMap()));
+ insertInto("BLOB_SPLIT_UPDATE_S", "(1, X'02')");
+
+ builder(warehouse, database, "BLOB_SPLIT_UPDATE_T")
+
.withMergeCondition("BLOB_SPLIT_UPDATE_T.id=BLOB_SPLIT_UPDATE_S.id")
+
.withMatchedUpdateSet("BLOB_SPLIT_UPDATE_T.picture=BLOB_SPLIT_UPDATE_S.picture")
+ .withSourceTable("BLOB_SPLIT_UPDATE_S")
+ .withSinkParallelism(1)
+ .build()
+ .run();
+
+ testBatchRead(
+ "SELECT id, picture FROM BLOB_SPLIT_UPDATE_T ORDER BY id",
+ Arrays.asList(
+ changelogRow("+I", 1, new byte[] {2}),
+ changelogRow("+I", 2, large),
+ changelogRow("+I", 3, new byte[] {3})));
+
+ List<Row> newBlobFiles = new ArrayList<>();
+ boolean sawOriginalLargeFile = false;
+ for (Row file :
+ collect(
+ "SELECT first_row_id, record_count,
file_size_in_bytes, max_sequence_number "
+ + "FROM `BLOB_SPLIT_UPDATE_T$files` WHERE
file_path LIKE '%.blob'")) {
+ long sequence = ((Number) file.getField(3)).longValue();
+ long size = ((Number) file.getField(2)).longValue();
+ if (sequence <= sequenceBeforeUpdate) {
+ if (size >= large.length) {
+ sawOriginalLargeFile = true;
+ }
+ } else {
+ newBlobFiles.add(file);
+ }
+ }
+ assertTrue(sawOriginalLargeFile);
+ newBlobFiles.sort(
+ (left, right) ->
+ Long.compare(
+ ((Number) left.getField(0)).longValue(),
+ ((Number) right.getField(0)).longValue()));
+ assertTrue(newBlobFiles.size() > 1);
+ long expectedRowId = 0;
+ long totalRecords = 0;
+ for (Row file : newBlobFiles) {
+ long firstRowId = ((Number) file.getField(0)).longValue();
+ long recordCount = ((Number) file.getField(1)).longValue();
+ long size = ((Number) file.getField(2)).longValue();
+ assertEquals(expectedRowId, firstRowId);
+ assertTrue(size < 4096);
+ expectedRowId += recordCount;
+ totalRecords += recordCount;
+ }
+ assertEquals(3L, totalRecords);
+ }
+
+ @Test
+ public void testUpdateRawBlobColumnTwice() throws Exception {
+ sEnv.executeSql(
+ buildDdl(
+ "BLOB_TWICE_T",
+ Arrays.asList("id INT", "picture BYTES"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ new HashMap<String, String>() {
+ {
+ put(ROW_TRACKING_ENABLED.key(), "true");
+ put(DATA_EVOLUTION_ENABLED.key(), "true");
+ put("blob-field", "picture");
+ }
+ }));
+ insertInto("BLOB_TWICE_T", "(1, X'01')", "(2, X'02')", "(3, X'03')");
+ sEnv.executeSql(
+ buildDdl(
+ "BLOB_TWICE_S1",
+ Arrays.asList("id INT", "picture BYTES"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyMap()));
+ insertInto("BLOB_TWICE_S1", "(1, X'11')");
+ builder(warehouse, database, "BLOB_TWICE_T")
+ .withMergeCondition("BLOB_TWICE_T.id=BLOB_TWICE_S1.id")
+
.withMatchedUpdateSet("BLOB_TWICE_T.picture=BLOB_TWICE_S1.picture")
+ .withSourceTable("BLOB_TWICE_S1")
+ .withSinkParallelism(1)
+ .build()
+ .run();
+
+ sEnv.executeSql(
+ buildDdl(
+ "BLOB_TWICE_S2",
+ Arrays.asList("id INT", "picture BYTES"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyMap()));
+ insertInto("BLOB_TWICE_S2", "(1, X'21')", "(3, X'23')");
+ builder(warehouse, database, "BLOB_TWICE_T")
+ .withMergeCondition("BLOB_TWICE_T.id=BLOB_TWICE_S2.id")
+
.withMatchedUpdateSet("BLOB_TWICE_T.picture=BLOB_TWICE_S2.picture")
+ .withSourceTable("BLOB_TWICE_S2")
+ .withSinkParallelism(1)
+ .build()
+ .run();
+
+ testBatchRead(
+ "SELECT id, picture FROM BLOB_TWICE_T ORDER BY id",
+ Arrays.asList(
+ changelogRow("+I", 1, new byte[] {0x21}),
+ changelogRow("+I", 2, new byte[] {2}),
+ changelogRow("+I", 3, new byte[] {0x23})));
+ }
+
+ @Test
+ public void testUpdateRawBlobColumnFromDescriptor() throws Exception {
+ File external = new File(warehouse, "external-blob.bin");
+ byte[] payload = "from-descriptor".getBytes(StandardCharsets.UTF_8);
+ Files.write(external.toPath(), payload);
+ String uri = external.toURI().toString();
+
+ sEnv.executeSql(
+ buildDdl(
+ "BLOB_DESC_MERGE_T",
+ Arrays.asList("id INT", "picture BYTES"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ new HashMap<String, String>() {
+ {
+ put(ROW_TRACKING_ENABLED.key(), "true");
+ put(DATA_EVOLUTION_ENABLED.key(), "true");
+ put("blob-field", "picture");
+ }
+ }));
+ insertInto("BLOB_DESC_MERGE_T", "(1, X'00')", "(2, X'01')");
+ sEnv.executeSql(
+ buildDdl(
+ "BLOB_DESC_MERGE_S",
+ Arrays.asList("id INT", "picture BYTES"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyMap()));
+ sEnv.executeSql(
+ "INSERT INTO BLOB_DESC_MERGE_S "
+ + "SELECT 1, sys.path_to_descriptor('"
+ + uri
+ + "')")
+ .await();
+
+ builder(warehouse, database, "BLOB_DESC_MERGE_T")
+
.withMergeCondition("BLOB_DESC_MERGE_T.id=BLOB_DESC_MERGE_S.id")
+
.withMatchedUpdateSet("BLOB_DESC_MERGE_T.picture=BLOB_DESC_MERGE_S.picture")
+ .withSourceTable("BLOB_DESC_MERGE_S")
+ .withSinkParallelism(1)
+ .build()
+ .run();
+
+ testBatchRead(
+ "SELECT id, picture FROM BLOB_DESC_MERGE_T ORDER BY id",
+ Arrays.asList(
+ changelogRow("+I", 1, payload), changelogRow("+I", 2,
new byte[] {1})));
+ }
+
+ @Test
+ public void testUpdateRawBlobColumnWithAppendWriteBuffer() throws
Exception {
+ sEnv.executeSql(
+ buildDdl(
+ "BLOB_BUFFER_T",
+ Arrays.asList("id INT", "picture BYTES"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ new HashMap<String, String>() {
+ {
+ put(ROW_TRACKING_ENABLED.key(), "true");
+ put(DATA_EVOLUTION_ENABLED.key(), "true");
+ put("blob-field", "picture");
+ put(CoreOptions.WRITE_BUFFER_FOR_APPEND.key(),
"true");
+ }
+ }));
+ insertInto("BLOB_BUFFER_T", "(1, X'01')", "(2, X'02')");
+ sEnv.executeSql(
+ buildDdl(
+ "BLOB_BUFFER_S",
+ Arrays.asList("id INT", "picture BYTES"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyMap()));
+ insertInto("BLOB_BUFFER_S", "(1, X'11')");
+
+ builder(warehouse, database, "BLOB_BUFFER_T")
+ .withMergeCondition("BLOB_BUFFER_T.id=BLOB_BUFFER_S.id")
+
.withMatchedUpdateSet("BLOB_BUFFER_T.picture=BLOB_BUFFER_S.picture")
+ .withSourceTable("BLOB_BUFFER_S")
+ .withSinkParallelism(1)
+ .build()
+ .run();
+
+ testBatchRead(
+ "SELECT id, picture FROM BLOB_BUFFER_T ORDER BY id",
+ Arrays.asList(
+ changelogRow("+I", 1, new byte[] {0x11}),
+ changelogRow("+I", 2, new byte[] {2})));
+ }
+
+ @Test
+ public void testUpdateRawBlobColumnNullOnInvalidDescriptorUri() throws
Exception {
+ sEnv.executeSql(
+ buildDdl(
+ "BLOB_FETCH_FAIL_T",
+ Arrays.asList("id INT", "picture BYTES"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ new HashMap<String, String>() {
+ {
+ put(ROW_TRACKING_ENABLED.key(), "true");
+ put(DATA_EVOLUTION_ENABLED.key(), "true");
+ put("blob-field", "picture");
+
put(CoreOptions.BLOB_WRITE_NULL_ON_FETCH_FAILURE.key(), "true");
+ }
+ }));
+ insertInto("BLOB_FETCH_FAIL_T", "(1, X'01')", "(2, X'02')");
+ sEnv.executeSql(
+ buildDdl(
+ "BLOB_FETCH_FAIL_S",
+ Arrays.asList("id INT", "picture BYTES"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyMap()));
+ sEnv.executeSql(
+ "INSERT INTO BLOB_FETCH_FAIL_S SELECT 1,
sys.path_to_descriptor("
+ +
"'https://img.alicdn.com/imgextra/##1304008055350781673')")
+ .await();
+
+ builder(warehouse, database, "BLOB_FETCH_FAIL_T")
+
.withMergeCondition("BLOB_FETCH_FAIL_T.id=BLOB_FETCH_FAIL_S.id")
+
.withMatchedUpdateSet("BLOB_FETCH_FAIL_T.picture=BLOB_FETCH_FAIL_S.picture")
+ .withSourceTable("BLOB_FETCH_FAIL_S")
+ .withSinkParallelism(1)
+ .build()
+ .run();
+
+ testBatchRead(
+ "SELECT id, picture FROM BLOB_FETCH_FAIL_T ORDER BY id",
+ Arrays.asList(changelogRow("+I", 1, null), changelogRow("+I",
2, new byte[] {2})));
+ }
+
+ @Test
+ public void testUpdateVideoFrameFieldThrowsError() throws Exception {
+ sEnv.executeSql(
+ buildDdl(
+ "VIDEO_FRAME_T",
+ Arrays.asList("id INT", "frame BYTES"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ new HashMap<String, String>() {
+ {
+ put(ROW_TRACKING_ENABLED.key(), "true");
+ put(DATA_EVOLUTION_ENABLED.key(), "true");
+ put("video-frame-field", "frame");
+ }
+ }));
+ insertInto("VIDEO_FRAME_T", "(1, CAST(NULL AS BYTES))");
+ sEnv.executeSql(
+ buildDdl(
+ "VIDEO_FRAME_S",
+ Arrays.asList("id INT", "frame BYTES"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyMap()));
+ insertInto("VIDEO_FRAME_S", "(1, X'01')");
+
+ IllegalStateException thrown =
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () ->
+ builder(warehouse, database, "VIDEO_FRAME_T")
+
.withMergeCondition("VIDEO_FRAME_T.id=VIDEO_FRAME_S.id")
+ .withMatchedUpdateSet(
+
"VIDEO_FRAME_T.frame=VIDEO_FRAME_S.frame")
+ .withSourceTable("VIDEO_FRAME_S")
+ .withSinkParallelism(1)
+ .build()
+ .run());
+ assertTrue(thrown.getMessage().contains("video-frame field"));
+ assertTrue(thrown.getMessage().contains("frame"));
+ }
+
+ private List<Row> collect(String query) throws Exception {
+ List<Row> rows = new ArrayList<>();
+ try (CloseableIterator<Row> iterator =
bEnv.executeSql(query).collect()) {
+ while (iterator.hasNext()) {
+ rows.add(iterator.next());
+ }
+ }
+ return rows;
+ }
+
+ private long maxSequence(String table) throws Exception {
+ List<Row> rows = collect("SELECT MAX(max_sequence_number) FROM `" +
table + "$files`");
+ Object value = rows.get(0).getField(0);
+ return value == null ? -1L : ((Number) value).longValue();
+ }
+
private void prepareTargetTable() throws Exception {
sEnv.executeSql(
buildDdl(
