Repository navigation
[flink] Support BUCKET_BALANCE shuffle strategy for table with bucket kyes. - #4450
Conversation
| * @param bucketingFunction the bucketing function for computing the bucket ID | ||
| * @return channel index in {@code [0, numChannels)} | ||
| */ | ||
| public static int routeByBucket( |
There was a problem hiding this comment.
Sharing one routing implementation between the sink shuffle and the lookup partitioner looks like the right direction — it removes the duplicated gcd/slot logic and stops the two algorithms from drifting apart. The owner feels a bit off, though: this static helper lives on a ChannelComputer implementation in the sink package, while FlussLookupInputPartitioner is in source.lookup and reaches it through a fully qualified name (FlussLookupInputPartitioner.java:133) without ever constructing a ChannelComputer.
Would it make sense to move the shared algorithm to a neutral internal home, e.g. flink.shuffle.BucketLoadBalanceRouter#route(...) (clearest ownership), or at least into flink.sink.shuffle next to DistributionMode, so both callers can use a regular import? If more callers show up later, static helpers would otherwise keep piling onto this class.
Two details that could ride along, if useful: the "evenly divisible" fast path is currently encoded twice (routeByBucket:160 and ChannelComputer#select:91), and on the lookup side lookupKeyEncoder.encodeKey(...) now runs before that check (FlussLookupInputPartitioner.java:128), so the divisible case pays one extra key encoding per probe key where it used to return early.
|
@leonardBang , I have rebase this PR. |
leonardBang
left a comment
leonardBang
left a comment
There was a problem hiding this comment.
Thanks @loserwang1024 for the update, I only left one comment about docs. Could you update and rebase this PR again as the main branch has merged several commits recently ？
| .defaultValue(DistributionMode.AUTO) | ||
| .withDescription( | ||
| "Defines the distribution mode for writing data to the sink. Available options are:\n" | ||
| + "- BUCKET_LOAD_BALANCE: Shuffle data by bucket key with even load distribution across all downstream subtasks using LCM-based logical slot assignment. Records with the same bucket key always route to the same subtask. Requires 'bucket.key' to be defined. Suitable for tables where intra-bucket ordering is not required but even load distribution matters. Not supported for tables with the aggregation merge engine, because Undo Recovery requires each bucket to be written by exactly one subtask.\n" |
There was a problem hiding this comment.
Could we make the guarantee of BUCKET_LOAD_BALANCE slightly more precise here?
BUCKET_BALANCE: Shuffles records by bucket key using an LCM-based logical-slot assignment. This improves sink-subtask utilization and expected load distribution, particularly when the bucket count and sink parallelism are not evenly divisible. Records with the same bucket key always route to the same subtask, while different bucket keys in the same physical bucket may be processed by different subtasks whose logical-slot ranges overlap that bucket. This mode does not guarantee that every physical bucket is distributed across all subtasks; when the bucket count is an exact multiple of the sink parallelism, it behaves like BUCKET mode. It requires bucket keys and does not preserve intra-bucket ordering. It is not supported for tables using the aggregation merge engine because Undo Recovery requires each physical bucket to be written by exactly one subtask.
…ucket kyes. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6-astra AI-Contributed/Feature: 252/338 AI-Contributed/UT: 0/200
Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6-astra AI-Contributed/Feature: 259/358 AI-Contributed/UT: 0/112
Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6-astra AI-Contributed/Feature: 73/349 AI-Contributed/UT: 0/411
949c17c to
818fb08
Compare
loserwang1024
changed the title
2 participants
Purpose
Linked issue: close #4449
Brief change log
Tests
API and Format
Documentation