You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "zuston (via GitHub)" <gi...@apache.org> on 2026/05/08 08:02:22 UTC
[PR] [flink] add split assignment mechanism w/o partitionId [fluss]
zuston opened a new pull request, #3272:
URL: https://github.com/apache/fluss/pull/3272
### Purpose
Linked issue: close #3269
The motivation of this PR is to align with the sink channel selector to acheive the all bucket data could be routed into the same subtasks if having the partitionIds when `numBucket % concurrency = 0` , that could emlinate the shuffle to improve the performance to reduce backfill time for large-scale data.
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] add split assignment mechanism w/o partitionId [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #3272:
URL: https://github.com/apache/fluss/pull/3272#issuecomment-4534239089
cc @loserwang1024
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] add split assignment mechanism w/o partitionId [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3272:
URL: https://github.com/apache/fluss/pull/3272#discussion_r3207333674
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java:
##########
@@ -942,21 +943,23 @@ private void assignPendingSplits(Set<Integer> pendingReaders) {
@VisibleForTesting
protected int getSplitOwner(SourceSplitBase split) {
TableBucket tableBucket = split.getTableBucket();
- int startIndex =
- tableBucket.getPartitionId() == null
- ? 0
- : ((tableBucket.getPartitionId().hashCode() * 31) & 0x7FFFFFFF)
- % context.currentParallelism();
+ int numChannels = context.currentParallelism();
// super hack logic, if the bucket is -1, it means the split is
// for bucket unaware, like paimon unaware bucket log table,
// we use hash split id to get the split owner
// todo: refactor the split assign logic
if (split.isLakeSplit() && tableBucket.getBucket() == -1) {
- return (split.splitId().hashCode() & 0x7FFFFFFF) % context.currentParallelism();
+ return (split.splitId().hashCode() & 0x7FFFFFFF) % numChannels;
}
- return (startIndex + tableBucket.getBucket()) % context.currentParallelism();
+ Long partitionId = tableBucket.getPartitionId();
+ int bucketId = tableBucket.getBucket();
+ if (ChannelComputer.shouldCombinePartitionInSharding(
+ partitionId != null, tableInfo.getNumBuckets(), numChannels)) {
+ return ChannelComputer.select(partitionId, bucketId, numChannels);
+ }
+ return ChannelComputer.select(bucketId, numChannels);
Review Comment:
This changes split ownership for partitioned tables when numBuckets % parallelism == 0 (it becomes bucket-only). The existing unit test FlinkSourceEnumeratorTest#testGetSplitOwner currently asserts different owners for different partitionIds with the same bucket (e.g., partition 1 vs 2, bucket 0) and will fail under the new behavior. Please update/add assertions to cover both cases: (1) divisible -> same owner for same bucket across partitions, and (2) non-divisible -> include partition in sharding to avoid skew.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java:
##########
@@ -942,21 +943,23 @@ private void assignPendingSplits(Set<Integer> pendingReaders) {
@VisibleForTesting
protected int getSplitOwner(SourceSplitBase split) {
TableBucket tableBucket = split.getTableBucket();
- int startIndex =
- tableBucket.getPartitionId() == null
- ? 0
- : ((tableBucket.getPartitionId().hashCode() * 31) & 0x7FFFFFFF)
- % context.currentParallelism();
+ int numChannels = context.currentParallelism();
// super hack logic, if the bucket is -1, it means the split is
// for bucket unaware, like paimon unaware bucket log table,
// we use hash split id to get the split owner
// todo: refactor the split assign logic
if (split.isLakeSplit() && tableBucket.getBucket() == -1) {
- return (split.splitId().hashCode() & 0x7FFFFFFF) % context.currentParallelism();
+ return (split.splitId().hashCode() & 0x7FFFFFFF) % numChannels;
}
- return (startIndex + tableBucket.getBucket()) % context.currentParallelism();
+ Long partitionId = tableBucket.getPartitionId();
+ int bucketId = tableBucket.getBucket();
+ if (ChannelComputer.shouldCombinePartitionInSharding(
+ partitionId != null, tableInfo.getNumBuckets(), numChannels)) {
+ return ChannelComputer.select(partitionId, bucketId, numChannels);
Review Comment:
shouldCombinePartitionInSharding is intended to be driven by whether the *table* is partitioned; passing (partitionId != null) couples the decision to per-split state and can hide bugs where a partitioned split accidentally has a null partitionId. Consider using the existing isPartitioned flag here, and if the table is partitioned and the combine-with-partition path is chosen, fail fast with a clear null check before calling select(partitionId, ...).
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java:
##########
@@ -32,6 +32,7 @@
import org.apache.fluss.flink.lake.LakeSplitGenerator;
import org.apache.fluss.flink.lake.split.LakeSnapshotAndFlussLogSplit;
import org.apache.fluss.flink.lake.split.LakeSnapshotSplit;
+import org.apache.fluss.flink.sink.ChannelComputer;
Review Comment:
FlinkSourceEnumerator now depends on org.apache.fluss.flink.sink.ChannelComputer just to reuse static sharding helpers. This introduces a source->sink package dependency that can be confusing and makes future refactors harder. Consider extracting the static sharding utilities (shouldCombinePartitionInSharding/select) into a neutral flink-common utility class/package so both source and sink can depend on it without crossing layers.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java:
##########
@@ -942,21 +943,23 @@ private void assignPendingSplits(Set<Integer> pendingReaders) {
@VisibleForTesting
protected int getSplitOwner(SourceSplitBase split) {
TableBucket tableBucket = split.getTableBucket();
Review Comment:
The Javadoc for getSplitOwner() (immediately above this method) still states that, for partitioned tables, bucket assignment is always offset/round-robin based on partition id. With the new conditional routing (bucket-only when numBuckets % parallelism == 0), that contract is no longer accurate—please update the Javadoc to reflect when partition is (and isn’t) part of the channel computation.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/ChannelComputer.java:
##########
@@ -59,6 +59,11 @@ static int select(String partitionName, int bucket, int numChannels) {
return (startChannel + bucket) % numChannels;
}
+ static int select(Long partitionId, int bucket, int numChannels) {
+ int startChannel = ((partitionId.hashCode() * 31) & 0x7FFFFFFF) % numChannels;
+ return (startChannel + bucket) % numChannels;
+ }
Review Comment:
select(Long partitionId, ...) accepts a boxed Long but immediately dereferences it; passing null will NPE. Since null isn’t supported, consider either switching the parameter type to primitive long (and unbox at the call site with an explicit null check) or adding an explicit non-null check here to fail with a clearer message.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java:
##########
@@ -942,21 +943,23 @@ private void assignPendingSplits(Set<Integer> pendingReaders) {
@VisibleForTesting
protected int getSplitOwner(SourceSplitBase split) {
TableBucket tableBucket = split.getTableBucket();
- int startIndex =
- tableBucket.getPartitionId() == null
- ? 0
- : ((tableBucket.getPartitionId().hashCode() * 31) & 0x7FFFFFFF)
- % context.currentParallelism();
+ int numChannels = context.currentParallelism();
// super hack logic, if the bucket is -1, it means the split is
// for bucket unaware, like paimon unaware bucket log table,
// we use hash split id to get the split owner
// todo: refactor the split assign logic
if (split.isLakeSplit() && tableBucket.getBucket() == -1) {
- return (split.splitId().hashCode() & 0x7FFFFFFF) % context.currentParallelism();
+ return (split.splitId().hashCode() & 0x7FFFFFFF) % numChannels;
}
- return (startIndex + tableBucket.getBucket()) % context.currentParallelism();
+ Long partitionId = tableBucket.getPartitionId();
+ int bucketId = tableBucket.getBucket();
+ if (ChannelComputer.shouldCombinePartitionInSharding(
+ partitionId != null, tableInfo.getNumBuckets(), numChannels)) {
+ return ChannelComputer.select(partitionId, bucketId, numChannels);
+ }
+ return ChannelComputer.select(bucketId, numChannels);
Review Comment:
getSplitOwner() now dereferences tableInfo.getNumBuckets(), but tableInfo is initialized lazily in start(). Calling getSplitOwner before start() (as FlinkSourceEnumeratorTest#testGetSplitOwner currently does) will throw an NPE with an unclear message. Please add an explicit precondition (e.g., checkState/tableInfo non-null) or restructure so numBuckets is available without relying on start(), to avoid surprising NPEs in tests and future call sites.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org