JingsongLi commented on code in PR #10514:
URL: https://github.com/apache/paimon/pull/10514#discussion_r4253470987
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java:
##########
@@ -680,8 +684,27 @@ public DataStream<Committable> writePartialColumns(
// 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);
DataStream<InternalRow> sorted =
- shuffled.map(t -> new FlinkRowWrapper(t.f1), typeInfo)
+ shuffled.map(
+ t ->
+ FlinkRowWrapper.fromUriReaderFactory(
+ t.f1,
+ blobDescriptorReaderFactory,
+ nullOnMissingFile,
+ nullOnFetchFailure,
Review Comment:
[P2] Preserve the fetch-failure NULL policy through sorting
With `blob-write-null-on-fetch-failure=true`, a BYTES source containing
`sys.path_to_descriptor('https://img.alicdn.com/imgextra/##1304008055350781673')`
aborts the MERGE with `IllegalArgumentException: Invalid URI` instead of
storing NULL. I reproduced this with a managed BLOB target and one unchanged
row. During sort serialization, this wrapper's `getBlob()` calls
`Blob.fromBytes` and eagerly creates the URI reader; the exception escapes
before `AbstractBlobElementWriter` can apply its fetch-failure fallback.
Existing
`BlobFormatWriterTest#testWriteNullOnFetchFailureForInvalidUriDescriptor`
establishes the expected NULL behavior for these descriptor bytes. Carry
descriptor bytes through sorting without resolving the reader, then resolve
inside the writer's policy boundary (or apply the same fallback before
serializing). Add an end-to-end regression test for the option.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionPartialWriteOperator.java:
##########
@@ -304,8 +411,7 @@ private Committable finish() throws Exception {
InternalRow row;
while (reader.hasNext()) {
row = reader.next();
- reusedRow.replaceRow(row);
- writer.write(reusedRow);
+ writer.write(rowForOriginal(row));
Review Comment:
[P2] Disable append buffering before writing BLOB gap placeholders
The copied table retains `write-buffer-for-append=true`. With that supported
option, these unchanged-row placeholders enter `BufferedSinkWriter` rather than
the BLOB writer; `InternalRowSerializer` attempts to serialize them and
`BlobPlaceholder.toDescriptor()` throws `UnsupportedOperationException: Should
never call this method for placeholder blob.` I reproduced this by inserting
two managed BLOB rows, updating only the first through this procedure, and
leaving the second unchanged. The entire MERGE fails at this line when
finishing the remaining row. Force `WRITE_BUFFER_FOR_APPEND=false` in the
partial writer's copied options, or explicitly support placeholder
serialization, and cover this configuration in an integration test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
