hudi-agent commented on code in PR #19518:
URL: https://github.com/apache/hudi/pull/19518#discussion_r4043922959
##########
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();
}
Review Comment:
🤖 nit: building the handle map key via `partitionPath + "/" + fileId` is a
bit fragile/opaque — could you extract a small helper (e.g.
`getHandleKey(partitionPath, fileId)`) so the key format is defined in one
place and self-documenting?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/BucketBulkInsertWriterHelper.java:
##########
@@ -46,23 +47,30 @@
@Slf4j
public class BucketBulkInsertWriterHelper extends BulkInsertWriterHelper {
public static final String FILE_GROUP_META_FIELD = "_fg";
+ public static final String PARTITION_PATH_META_FIELD = "_partition_path";
protected final int recordArity;
+ protected final boolean isNonBlockingConcurrencyControl;
private String lastFileId; // for efficient code path
+ private String lastPartitionPath; // only used by NBCC where file IDs repeat
across partitions
public BucketBulkInsertWriterHelper(Configuration conf, HoodieTable<?, ?, ?,
?> hoodieTable, HoodieWriteConfig writeConfig,
String instantTime, int taskPartitionId,
long taskId, long taskEpochId, RowType rowType) {
super(conf, hoodieTable, writeConfig, instantTime, taskPartitionId,
taskId, taskEpochId, rowType);
this.recordArity = rowType.getFieldCount();
+ this.isNonBlockingConcurrencyControl =
OptionsResolver.isNonBlockingConcurrencyControl(conf);
}
public void write(RowData tuple) throws IOException {
try {
Review Comment:
🤖 nit: `fileIdOrdinal` is recomputed from `isNonBlockingConcurrencyControl`
on every `write()` call even though it's constant for the lifetime of the
helper — consider computing it once in the constructor and storing it as a
field.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
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 confirm from the code: `SortOperatorGen` resolves sort fields via
`rowType.getFieldIndex(...)` on the wrapper row only, and `ROW` isn't in
`isSortable`, so the nested `record.<partition_field>` can't be used as a sort
key without extending the generator. There's also a subtle reason to sort on
the derived partition path rather than the raw partition columns: the sorted
write path closes all handles whenever the `(partition, fileId)` key changes,
so it relies on each file group being contiguous — that only holds if the sort
order matches the key generator's output, which raw columns don't guarantee for
key gens that transform values (e.g. timestamp-based partitioning).
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/BulkInsertFunctionWrapper.java:
##########
@@ -235,6 +239,8 @@ private void setupMapFunction() {
}
private void setupSortOperator() throws Exception {
+ boolean isNonBlockingConcurrencyControl =
Review Comment:
🤖 nit: `isNonBlockingConcurrencyControl` is now computed independently in
both the constructor and `setupSortOperator()` — could this be stored as a
field instead so it's computed once and reused?
<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]
