fix(flink): use partition-aware file groups for Flink NBCC bulk inserts - #19518
fix(flink): use partition-aware file groups for Flink NBCC bulk inserts#19518cshuo wants to merge 2 commits into
Conversation
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! The PR makes Flink bucket bulk inserts under NBCC partition-aware by prepending the partition path to the internal sort row and keying write handles by HoodieFileGroupId, while leaving the non-NBCC path (which already gets unique per-partition file IDs) unchanged. The scoping is sound and the sorted/unsorted handle logic traces through correctly. One coupling worth double-checking noted inline. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. Two small readability nits below.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! The PR fixes Flink bucket bulk inserts under NBCC where fixed bucket file IDs can repeat across partitions, by making sort rows and write-handle caching partition-aware only when NBCC is enabled. The scoping to NBCC looks correct — non-NBCC file IDs are UUID-based and already unique per partition — and the producer (row builders/sorters) and consumer (writer decode) both derive the flag from the same config, keeping the in-flight sort-row layouts consistent. No new issues flagged from this automated pass; the round-1 concerns appear addressed. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. One naming concern where a needFixedFileIdSuffix value is passed into an isNonBlockingConcurrencyControl parameter.
cc @yihua
| RowType sortRowType = | ||
| LsmBucketBulkInsertWriterHelper.rowTypeWithFileIdAndKey(rowType); | ||
| LsmBucketBulkInsertWriterHelper.rowTypeWithFileIdAndKey( | ||
| rowType, needFixedFileIdSuffix); |
There was a problem hiding this comment.
🤖 nit: these helpers take a param named isNonBlockingConcurrencyControl, but here you're passing needFixedFileIdSuffix. Even if the two coincide today, conflating them makes this harder to reason about later — could you align the naming (or pass the NBCC flag explicitly)?
| HoodieRowDataCreateHandle rowCreateHandle = new HoodieRowDataCreateHandle(hoodieTable, writeConfig, partitionPath, fileId, | ||
| instantTime, taskPartitionId, totalSubtaskNum, taskEpochId, writerSchema, preserveHoodieMetadata, isAppendMode && !populateMetaFields); | ||
| handles.put(fileId, rowCreateHandle); | ||
| handles.put(handleKey, rowCreateHandle); |
There was a problem hiding this comment.
can we just sort the inputs with _fg_id and partiiton fields instead?
There was a problem hiding this 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.
Address review comments 3717996733 and 3717996744.
ffd19fd to
87c32b1
Compare
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR fixes Flink NBCC bucket bulk inserts so that fixed bucket file IDs repeating across partitions no longer mix records or reuse the wrong write handle, by making the sort rows and handle cache partition-aware. I traced the row encode/decode paths and confirmed the NBCC flag is derived from the same config everywhere (Pipelines' needFixedFileIdSuffix and the helper's isNonBlockingConcurrencyControl both come from OptionsResolver.isNonBlockingConcurrencyControl), so encoding and decoding stay consistent. No new correctness issues flagged from this automated pass beyond what prior rounds raised — a Hudi committer or PMC member can take it from here for a final review. One naming inconsistency worth addressing, otherwise the code is clean and well-commented.
cc @yihua
| needFixedFileIdSuffix); | ||
| return GenericRowData.of(StringData.fromString(fileId), record); | ||
| return needFixedFileIdSuffix | ||
| ? GenericRowData.of( |
There was a problem hiding this comment.
🤖 nit: needFixedFileIdSuffix is used here (and in LsmBucketBulkInsertWriterHelper.rowWithFileIdAndKey) as a stand-in for the NBCC flag, while every other new method in this PR names the same concept isNonBlockingConcurrencyControl. A future reader has to independently verify these are always equivalent. Could you rename the parameter to isNonBlockingConcurrencyControl (or at least add a brief inline comment) to make the connection explicit?
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19518 +/- ##
============================================
- Coverage 76.60% 76.59% -0.01%
- Complexity 32226 32230 +4
============================================
Files 2520 2520
Lines 139014 139050 +36
Branches 16699 16706 +7
============================================
+ Hits 106492 106511 +19
- Misses 24909 24919 +10
- Partials 7613 7620 +7
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
5 participants
Describe the issue this Pull Request addresses
Closes #19517.
With Flink bucket bulk inserts under non-blocking concurrency control (NBCC), fixed bucket file IDs can repeat across partitions. Sorting and caching write handles only by file ID can therefore mix records from different partitions or reuse the wrong handle.
Summary and Changelog
HoodieFileGroupIdwhile retaining the existing file-ID-only path for non-NBCC writes.Impact
This change affects only Flink bucket bulk inserts using NBCC. It introduces no public API or configuration changes. Non-NBCC row schemas, sort keys, and handle lookup behavior remain unchanged, avoiding additional overhead on existing paths; NBCC sort rows carry one additional partition-path field.
Risk Level
Low. The behavior change is scoped to NBCC bucket bulk inserts and is covered by a parameterized regression test for sorted and unsorted input.
Documentation Update
None.
Contributor's checklist