[flink] Support BUCKET_LOAD_BALANCE shuffle strategy for table with bucket kyes. - #4450
loserwang1024 wants to merge 2 commits into
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. |
2 participants
Purpose
Linked issue: close #4449
Brief change log
Tests
API and Format
Documentation