feat(flink): support remote partitioner for simple bucket index#18897
feat(flink): support remote partitioner for simple bucket index#18897fhan688 wants to merge 3 commits into
Conversation
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.
Thanks for the contribution! This PR adds remote partitioner support for Flink simple bucket index, gated behind a new option that requires simple bucket index + embedded timeline server, and propagates the setting into HoodieWriteConfig, EmbeddedTimelineService reuse identity, and both the bulk insert / streaming write pipelines. The implementation mirrors the existing Spark BucketPartitionUtils pattern, and the lazy load of RemotePartitionHelper on the task side is safe given the coordinator writes ViewStorageProperties during start() before subtasks process records. As a bonus, the rewritten TimelineServiceIdentifier.equals fixes a latent bug where two identifiers with null hostAddr would compare equal without checking the other fields. 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 naming nits below — method predicate convention, an ambiguous parameter name, and a missing is prefix on a boolean field.
cc @yihua
Can you explain a bit more about this part, the record keys are routed based on bucket ids instead of tasks, even if the task parallelism changes, the same key should still route to the same bucket? |
Thanks for the correction. I agree that the current local simple bucket logic keeps the key-to-bucket mapping stable. A parallelism change only affects the bucket-to-task mapping, and both I updated the motivation to be more precise: this change centralizes bucket-to-task assignment through the timeline service by reusing the existing remote partitioner capability. This makes the routing side and bucket-loading side use the same remote assignment source, instead of maintaining separate local assignment calculations. |
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.
Thanks for the contribution! This PR wires the Flink simple bucket index path to the existing remote partitioner via the embedded timeline service, and also fixes a latent equals/hashCode bug in TimelineServiceIdentifier. One thing worth a second look is the withProps(...) reordering in FlinkWriteClients.getHoodieClientConfig, which silently flips override precedence for multiple settings (not just the timeline server flag this feature needs). Please take a look at the inline comment, and this should be ready for a Hudi committer or PMC member to take it from here. A couple of small readability suggestions around the remote partitioner's error handling.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #18897 +/- ##
============================================
+ Coverage 67.01% 68.80% +1.79%
- Complexity 28461 29187 +726
============================================
Files 2520 2521 +1
Lines 140046 140097 +51
Branches 17197 17210 +13
============================================
+ Hits 93850 96399 +2549
+ Misses 38529 35925 -2604
- Partials 7667 7773 +106
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.
Thanks for the contribution! This PR wires Flink simple bucket index writes to the existing remote partitioner path, and fixes a latent equals/hashCode bug in TimelineServiceIdentifier along the way. No new issues flagged from this automated pass beyond what previous review rounds have already raised — a Hudi committer or PMC member can take it from here for a final review. A couple of minor readability nits below; otherwise the code is clean.
cc @yihua
| } | ||
|
|
||
| private void setRemotePartitionHelper( | ||
| BucketIndexRemotePartitioner<HoodieKey> partitioner, |
There was a problem hiding this comment.
🤖 nit: using reflection to inject the mock is brittle — could remotePartitionHelper be exposed via a package-private setter or test-only constructor so the test doesn't need setAccessible(true)?
- AI-generated; verify before applying. React 👍/👎 to flag quality.
| * Bucket index input partitioner backed by the embedded timeline service. | ||
| * | ||
| * @param <T> The type of object to hash | ||
| */ |
There was a problem hiding this comment.
🤖 nit: the javadoc says @param <T> The type of object to hash, but T is bounded to HoodieKey and the class doesn't really hash — could you reword this (e.g. "the key type to partition by") to avoid confusion?
- AI-generated; verify before applying. React 👍/👎 to flag quality.
| @Override | ||
| public int partition(T key, int numPartitions) { | ||
| String partitionPath = normalizePartitionPath(key.getPartitionPath()); | ||
| int numBuckets = numBucketsFunction.getNumBuckets(partitionPath); |
There was a problem hiding this comment.
🤖 nit: the two getRemotePartition overloads have very similar signatures (one takes NumBucketsFunction, the other int numBuckets) — could the private one be renamed (e.g. doGetRemotePartition) to make it obvious at call sites which is which?
- AI-generated; verify before applying. React 👍/👎 to flag quality.
6 participants
Describe the issue this Pull Request addresses
This PR wires Flink simple bucket index writes to the existing timeline-service-backed remote partitioner path.
Today Flink simple bucket index routes records with local bucket-to-task mapping. When writer parallelism changes, bucket ownership can shift across subtasks, which can make the writer route records and load bucket file groups inconsistently. Hudi already has remote partition helper support in the timeline service and write config, but the Flink simple bucket index write pipeline does not use it.
This PR enables Flink writers to use remote bucket partition assignment when
hoodie.bucket.index.remote.partitioner.enable=true.Summary and Changelog
This PR adds remote partitioner support for Flink simple bucket index.
Changes:
hoodie.bucket.index.remote.partitioner.enable, backed by the existing core bucket remote partitioner config key.BucketIndexRemotePartitionerfor FlinkpartitionCustom, usingRemotePartitionHelperandNumBucketsFunction.BucketStreamWriteFunctionuse remote partition assignment when deciding whether a bucket belongs to the current writer subtask.HoodieWriteConfig.EmbeddedTimelineServicereuse identity to avoid sharing a timeline service across incompatible writerconfigs.
No code was copied from external sources.
Impact
This is a user-facing Flink write feature, disabled by default.
When
hoodie.bucket.index.remote.partitioner.enable=true, Flink simple bucket index writers use the timeline service to determine bucket-to-task assignment. This change centralizes bucket-to-task assignment through the timeline service by reusing the existing remote partitioner capability. This makes the routing side and bucket-loading side use the same remote assignment source, instead of maintaining separate local assignment calculations.Public API/config impact:
hoodie.bucket.index.remote.partitioner.enable.Storage format impact:
Performance impact:
Risk Level
medium
This changes Flink bucket index routing behavior when the new option is enabled. The default behavior is unchanged because the option defaults to
false.Risk mitigation:
Verification:
mvn -pl hudi-flink-datasource/hudi-flink -am -Dcheckstyle.skip=true -DskipITs -DfailIfNoTests=false -Dsurefire.failIfNoSpecifiedTests=false -Dtest=TestOptionsResolver,TestStreamerUtil,TestFlinkWriteClients,TestBucketIndexRemotePartitioner testmvn -pl hudi-flink-datasource/hudi-flink -am -Dcheckstyle.skip=true -DfailIfNoTests=false -Dsurefire.failIfNoSpecifiedTests=false -Dtest=ITTestBucketStreamWrite#testRemotePartitioner testDocumentation Update
Required.
The Hudi website/config documentation should be updated to describe Flink support for:
hoodie.bucket.index.remote.partitioner.enableThe documentation should mention that this option applies to Flink simple bucket index writes, requires embedded timeline server, and defaults to
false.Contributor's checklist