Stephen0421 commented on code in PR #10514:
URL: https://github.com/apache/paimon/pull/10514#discussion_r4255803265
##########
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:
Fixed. dataEvolutionWriteOptions() now sets write-buffer-for-append=false,
and table.copy() overrides the user setting together with the existing
target-file-size and target-file-row-num overrides. Placeholders are no longer
serialized by BufferedSinkWriter, and a buffer flush cannot close the file
before the row-id range is finished.
forceBufferSpill() returns immediately when blobContext != null, so a
managed BLOB column does not take that path either.
testUpdateRawBlobColumnWithAppendWriteBuffer sets the table option to true,
updates only the first row, and checks that the unchanged row keeps its payload.
##########
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:
Fixed for the sort path. The pre-sort wrapper carries the stored descriptor
bytes and does not open a URI reader. After the sort, BinaryRow.getBlob and
BinaryArray.getBlob call Blob.fromBytes, which rebuilds a BlobRef when the
bytes are a serialized descriptor. BlobDescriptorResolvingRow then attaches the
catalog reader, and UriReaderFactory.create() runs inside
AbstractBlobElementWriter.getBlob(), so an invalid URI is written as NULL when
blob-write-null-on-fetch-failure=true.
ARRAY<BLOB> and MAP<K,BLOB> use the same reconstruction.
BlobDescriptorResolvingArray and BlobDescriptorResolvingMap wrap the post-sort
BinaryArray, so the writer does not see the pre-sort BlobData or copy the
descriptor bytes as payload.
testUpdateRawBlobColumnNullOnInvalidDescriptorUri covers the reported URI
and an unchanged row. testUpdateRawBlobColumnFromDescriptor still reads back
the local file payload.
--
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]
