feat(flink): Sort bulk insert records by record key for LSM layout - #19390
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19390 +/- ##
============================================
+ Coverage 74.83% 75.55% +0.72%
- Complexity 32332 32679 +347
============================================
Files 2574 2576 +2
Lines 142978 143089 +111
Branches 17527 17582 +55
============================================
+ Hits 106992 108115 +1123
+ Misses 27921 26939 -982
+ Partials 8065 8035 -30
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
09cdda4 to
1b8b304
Compare
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for the contribution! This PR enables the LSM storage layout as the default for Flink bulk insert and physically orders records by encoded Hudi record key (partition/file-group, then key) via dedicated LSM sort-row and writer helpers, while preserving existing default-layout behavior. I traced the pipeline routing, writer handle management, sort-row decoding, and backward compatibility for existing default-layout tables, and did not find correctness issues in this automated pass. No correctness issues found. A few style/readability suggestions in the inline comments. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here. One indentation inconsistency in WriterHelpers — the rest of the changes are clean and well-structured.
cc @yihua
| } else { | ||
| return OptionsResolver.isBucketIndexType(conf) | ||
| ? new BucketBulkInsertWriterHelper( | ||
| conf, hoodieTable, writeConfig, instantTime, taskPartitionId, taskId, taskEpochId, rowType) |
There was a problem hiding this comment.
🤖 nit: the conf argument here is indented to the ? level rather than the constructor argument level — could you align it with the conf on line 39/47 ( conf, ...) to match the other three constructor calls in this method?
1b8b304 to
6c24577
Compare
6c24577 to
51b5e60
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 enables the LSM storage layout for Flink bulk insert — sorting records by encoded record key and adding dedicated LSM sort/writer helpers. The routing, sorting, and writer refactor look consistent, and backward compatibility for existing tables is preserved (the persisted hoodie.table.storage.layout is inherited on reopen). The one item worth a closer look is how duplicate record keys behave on read now that bulk insert defaults to LSM — see the inline comment. 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 indentation inconsistency worth a quick fix, otherwise the code is clean and well-structured.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR enables the LSM storage layout as the default for Flink bulk insert and sorts records by the encoded Hudi record key, isolating the layout-specific sorting/writing into dedicated helpers. The refactor and backward-compatibility handling for existing tables look careful and well-tested; one cross-operation compatibility edge case is worth double-checking in the inline comment. 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.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR enables the LSM storage layout as the default for Flink bulk_insert and sorts records by encoded record key (per partition/file-group), adding dedicated LSM sort-row and writer helpers plus routing changes in Pipelines. No new issues flagged from this automated pass — the previously-raised duplicate-key/LSM-merge-collapse concern is already tracked in the inline thread — a Hudi committer or PMC member can take it from here for a final review. A small naming-consistency suggestion below; otherwise the refactor reads cleanly.
cc @yihua
|
|
||
| public static final String FILE_GROUP_META_FIELD = "_fg"; | ||
| public static final String PARTITION_META_FIELD = "_partition"; | ||
| public static final String RECORD_KEY_FIELD = "_record_key"; |
There was a problem hiding this comment.
🤖 nit: the other internal field-name constants use an underscore prefix (_fg, _partition, _record_key) but RECORD_FIELD is just "record" — could you make it "_record" for consistency so all internal sort-row fields follow the same convention?
5 participants
Describe the issue this Pull Request addresses
Flink uses the LSM tree storage layout by default, but
bulk_insertpreviously fell back to the default layout because its input was not guaranteed to satisfy the LSM sorted-run invariant.LSM bulk insert must order records by the encoded Hudi record key rather than the key field’s native Flink type. This PR enables LSM bulk insert and ensures the required physical ordering.
Fixes #19325.
Summary and Changelog
Impact
hoodie.table.storage.layout=defaultretains its existing behavior.Risk Level
Medium. The change affects Flink bulk-insert routing, sorting, and writer selection. Unit and integration tests cover bucket and non-bucket indexes, COW/MOR tables, partitioned and non-partitioned tables, duplicate keys, and subsequent updates.
Documentation Update
none. No new user-facing configuration is introduced.
Contributor's checklist