danny0405 commented on code in PR #19518:
URL: https://github.com/apache/hudi/pull/19518#discussion_r3725127226
##########
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:
can we just sort the inputs with _fg_id and partiiton fields instead?
--
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]
