cshuo commented on code in PR #19518:
URL: https://github.com/apache/hudi/pull/19518#discussion_r3725497731
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/BucketBulkInsertWriterHelper.java:
##########
@@ -76,29 +84,38 @@ protected void writeRecord(
String partitionPath,
String fileId,
RowData record) throws IOException {
- if ((lastFileId == null) || !lastFileId.equals(fileId)) {
+ if ((lastFileId == null)
+ || !lastFileId.equals(fileId)
+ || (isNonBlockingConcurrencyControl &&
!partitionPath.equals(lastPartitionPath))) {
log.info("Creating new file for partition path {}", partitionPath);
handle = getRowCreateHandle(partitionPath, fileId);
lastFileId = fileId;
+ lastPartitionPath = partitionPath;
}
handle.write(recordKey, partitionPath, record);
}
private HoodieRowDataCreateHandle getRowCreateHandle(String partitionPath,
String fileId) throws IOException {
- if (!handles.containsKey(fileId)) { // if there is no handle corresponding
to the fileId
+ String handleKey = isNonBlockingConcurrencyControl
+ ? partitionPath + "/" + fileId
+ : fileId;
+ if (!handles.containsKey(handleKey)) { // if there is no handle
corresponding to the file group
if (this.isInputSorted) {
// if records are sorted, we can close all existing handles
close();
}
HoodieRowDataCreateHandle rowCreateHandle = new
HoodieRowDataCreateHandle(hoodieTable, writeConfig, partitionPath, fileId,
instantTime, taskPartitionId, totalSubtaskNum, taskEpochId,
writerSchema, preserveHoodieMetadata, isAppendMode && !populateMetaFields);
- handles.put(fileId, rowCreateHandle);
+ handles.put(handleKey, rowCreateHandle);
Review Comment:
Adding the `_fg` field wraps the input into a nested row, so the original
partition field becomes `record.partition_field`. Reusing it directly would
require enhancing the sorter to support nested fields, including comparator and
normalized-key generation, since it currently supports only top-level fields.
Adding the partition path to the wrapper row has limited overhead: it was
already computed in the writer helper, and now the same computation is simply
moved earlier and reused for sorting and writing.
--
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]
