feat(flink): Sort bulk insert records by record key for LSM layout - #19390
feat(flink): Sort bulk insert records by record key for LSM layout#19390cshuo wants to merge 2 commits into
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19390 +/- ##
============================================
- Coverage 74.83% 74.82% -0.01%
+ Complexity 32332 32322 -10
============================================
Files 2574 2577 +3
Lines 142978 143065 +87
Branches 17527 17535 +8
============================================
+ Hits 106992 107055 +63
- Misses 27921 27950 +29
+ Partials 8065 8060 -5
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.
| return HoodieTableConfig.TableStorageLayout.fromConfigValue(conf.getString( | ||
| HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(), | ||
| (isInsertOperation(conf) || isBulkInsertOperation(conf)) | ||
| isInsertOperation(conf) |
There was a problem hiding this comment.
🤖 Bulk insert now defaults to LSM, but the LSM reader collapses all versions of a key within a sorted run (LsmFileGroupRecordIterator#nextMergedRecord pops consecutive same-key records and merges them), so duplicate keys that bulk insert writes into a base file get merged into one on snapshot read. checkStorageLayout still blocks insert "because they allow duplicate record keys" — bulk insert has the same property. @danny0405 is silently collapsing bulk-insert duplicates on read the intended behavior? The new test only checks the sort row, not an end-to-end read.
| } else { | ||
| return OptionsResolver.isBucketIndexType(conf) | ||
| ? new BucketBulkInsertWriterHelper( | ||
| conf, hoodieTable, writeConfig, instantTime, taskPartitionId, taskId, taskEpochId, rowType) |
There was a problem hiding this comment.
🤖 nit: the constructor arguments for BucketBulkInsertWriterHelper are indented at the ? level rather than four spaces past it, unlike every other constructor in this block — could you align it with the LsmBucketBulkInsertWriterHelper args above (i.e. add four more spaces)?
4 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