refactor(flink): simplify streaming write pipeline construction - #19579
refactor(flink): simplify streaming write pipeline construction#19579danny0405 wants to merge 1 commit into
Conversation
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR refactors the Flink streaming write pipeline construction into bucketStreamWrite and addIndexWrite, shares the final bucket-write operator between the simple and consistent-hashing engines, and converts nested branches into guard clauses. I traced both bucket engine branches and the stream/index write path — operator names, UIDs, parallelism, type information, partitioner selection, the insert-overwrite guard, and managed-memory declarations all appear preserved, so the generated operator graph looks unchanged. 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. A few minor naming nits below, but the refactor is clean and readable overall.
cc @yihua
| declareManagedMemoryIfNecessary( | ||
| conf, writeDataStream, () -> OptionsResolver.getWriteBufferSizeInBytes(conf)); | ||
|
|
||
| return isStreamingIndexWriteEnabled |
There was a problem hiding this comment.
🤖 nit: addIndexWrite uses a verb prefix while its sibling bucketStreamWrite reads as a noun/description — could you align them? Something like indexWritePipeline or buildIndexWriteStream would match the style of bucketStreamWrite and make it clearer these are both pipeline-builder helpers.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #19579 +/- ##
=========================================
Coverage 77.49% 77.50%
- Complexity 32784 32790 +6
=========================================
Files 2522 2522
Lines 139140 139145 +5
Branches 16723 16723
=========================================
+ Hits 107832 107842 +10
Misses 23742 23742
+ Partials 7566 7561 -5
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
Pipelines.hoodieStreamWritecurrently combines bucket-engine selection, data-write construction, and optional index-write construction in one nested method. This makes the pipeline topology harder to follow and duplicates the final bucket-write stage across simple and consistent-hashing bucket engines.Summary and Changelog
This change makes the Flink streaming write pipeline construction easier to follow without changing its topology or behavior.
bucketStreamWrite.addIndexWrite.No code was copied from another project.
Impact
No public API, user-facing behavior, configuration, storage format, or performance behavior changes. Operator names, UIDs, partitioning, parallelism, type information, and managed-memory declarations remain unchanged.
Risk Level
low. The change refactors pipeline construction while preserving the generated Flink operator graph. Focused graph tests cover normal writes, simple bucket writes, consistent-hashing bucket writes, record-level index routing, bulk inserts, and table-service pipelines.
Documentation Update
none. This is an internal refactor with Javadocs added for the extracted methods.
Testing
mvn -pl hudi-flink-datasource/hudi-flink -am -Drat.skip=true -DskipITs -Dtest=org.apache.hudi.sink.utils.TestPipelines -Dsurefire.failIfNoSpecifiedTests=false test(10 tests passed)mvn -pl hudi-flink-datasource/hudi-flink -Drat.skip=true -DskipTests checkstyle:checkgit diff --checkContributor's checklist