hudi-agent commented on code in PR #19518:
URL: https://github.com/apache/hudi/pull/19518#discussion_r4127102216
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/BucketBulkInsertWriterHelper.java:
##########
@@ -76,29 +85,38 @@ protected void writeRecord(
String partitionPath,
String fileId,
RowData record) throws IOException {
- if ((lastFileId == null) || !lastFileId.equals(fileId)) {
- log.info("Creating new file for partition path {}", partitionPath);
+ if ((lastFileId == null)
+ || !lastFileId.equals(fileId)
+ || (isNonBlockingConcurrencyControl &&
!partitionPath.equals(lastPartitionPath))) {
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;
+ HoodieRowDataCreateHandle rowCreateHandle = handles.get(handleKey);
+ if (rowCreateHandle == null) { // if there is no handle corresponding to
the file group
if (this.isInputSorted) {
// if records are sorted, we can close all existing handles
close();
Review Comment:
🤖 nit: building the handle key via `partitionPath + "/" + fileId` could
collide if a partition path itself contains "/" (common with multi-level
partitioning). Might be safer to key by a `Pair<String, String>` or similar
instead of string concatenation.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
