cshuo commented on code in PR #19518:
URL: https://github.com/apache/hudi/pull/19518#discussion_r3718523628
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/BucketBulkInsertWriterHelper.java:
##########
@@ -125,12 +143,24 @@ public static RowData rowWithFileId(Map<String, String>
bucketIdToFileId, RowDat
indexKeyFields,
numBucketsFunction,
needFixedFileIdSuffix);
- return GenericRowData.of(StringData.fromString(fileId), record);
+ return needFixedFileIdSuffix
Review Comment:
By design: all production sort-row builders, row types, and sorters derive
this flag from the same `Configuration`, and the writer decodes the row from
that configuration as well. The direct test caller also enables NBCC when
passing `true`. Keeping this invariant at pipeline construction avoids adding
validation to the per-record write path.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriterHelper.java:
##########
@@ -80,7 +80,8 @@ public class BulkInsertWriterHelper implements AutoCloseable {
private String lastKnownPartitionPath = null;
private final String fileIdPrefix;
private int numFilesWritten = 0;
- protected final Map<String, HoodieRowDataCreateHandle> handles = new
HashMap<>();
+ // String keys are used by existing paths; NBCC bucket writes use
HoodieFileGroupId keys.
+ protected final Map<Object, HoodieRowDataCreateHandle> handles = new
HashMap<>();
Review Comment:
Fixed in ffd19fd18620. NBCC handle keys now use `partitionPath + "/" +
fileId`, allowing the base map to remain `Map<String,
HoodieRowDataCreateHandle>` while leaving the non-NBCC key path unchanged.
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/BulkInsertFunctionWrapper.java:
##########
@@ -245,8 +249,10 @@ private void setupSortOperator() throws Exception {
.setExecutionConfig(new ExecutionConfig().enableObjectReuse())
.build();
SortOperatorGen sortOperatorGen = lsmSortInput
- ?
LsmBucketBulkInsertWriterHelper.getFileIdAndKeySorterGen(sortInputRowType)
- : BucketBulkInsertWriterHelper.getFileIdSorterGen(rowTypeWithFileId);
+ ? LsmBucketBulkInsertWriterHelper.getFileIdAndKeySorterGen(
+ sortInputRowType,
OptionsResolver.isNonBlockingConcurrencyControl(conf))
+ : BucketBulkInsertWriterHelper.getFileIdSorterGen(
Review Comment:
Fixed in ffd19fd18620. `setupSortOperator` now resolves the NBCC flag once
and passes the same local value to both sorter branches.
--
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]
