You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "loserwang1024 (via GitHub)" <gi...@apache.org> on 2026/09/21 02:20:04 UTC
[PR] [flink] Support BUCKET_LOAD_BALANCE shuffle strategy for table with bucket kyes. [fluss]
loserwang1024 opened a new pull request, #4450:
URL: https://github.com/apache/fluss/pull/4450
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink], [rust], [python], [c++], [elixir]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
- **Generative AI disclosure:** Indicate whether generative AI tools were used in authoring this PR. If yes, specify the tool below.
- [ ] No generative AI tools used
- [ ] Yes (please specify the tool below)
**(The sections below can be removed for hotfixes or typos)**
-->
<!--
Generated-by: [Tool Name and Version] following [the guidelines](https://github.com/apache/fluss/blob/main/AGENTS.md)
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #4449
<!-- What is the purpose of the change -->
### 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] Support BUCKET_BALANCE shuffle strategy for table with bucket kyes. [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 merged PR #4450:
URL: https://github.com/apache/fluss/pull/4450
--
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] Support BUCKET_LOAD_BALANCE shuffle strategy for table with bucket kyes. [fluss]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4450:
URL: https://github.com/apache/fluss/pull/4450#discussion_r4070657370
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtils.java:
##########
@@ -91,12 +91,13 @@ public static void validateDistributionModeForMergeEngine(
@Nullable MergeEngineType mergeEngineType, DistributionMode distributionMode) {
if (mergeEngineType != null
&& distributionMode != DistributionMode.BUCKET
+ && distributionMode != DistributionMode.BUCKET_LOAD_BALANCE
Review Comment:
Should `BUCKET_LOAD_BALANCE` be opened up to PK tables here? As written it is allowed for every merge engine and `FlinkSink` adds no extra gate on the PK path (`FlinkSink.java:357`), so a config like `bucket.num = 100` + sink parallelism 8 is accepted today.
The part that may not be handled yet is undo recovery, which `FlinkTableSink.java:207` enables automatically for AGGREGATION and users cannot opt out of: it assumes one writer per bucket — `WriterState` tracks "the last written offset for each bucket that this writer is responsible for", and `putMergedOffset` (`RecoveryOffsetManager.java:387-395`) throws `Conflicting checkpoint offsets` once the union state carries two different offsets for the same bucket. This mode does keep "same bucket key -> same subtask", but in the non-divisible branches one physical bucket fans out to several subtasks (per the routing formulas, 100 buckets x 8 parallelism shares buckets 12/37/62/87, 2 subtasks each), so the first recovery after a checkpoint may fail in `initializeState()` on every subtask. `isAssignedToSubtask` (`RecoveryOffsetManager.java:536-570`) also still derives the owner from the old `bucket % parallelism` mapping.
Would it make sense to either gate `AGGREGATION + BUCKET_LOAD_BALANCE` for now with an explicit message, and narrow the "safe for Primary Key tables and merge-engine tables" wording (`DistributionMode.java:79`, `website/docs/engine-flink/options.md:193`) to deduplication only; or align the undo model with the new routing (e.g. allow several writer offsets per bucket and take the min, or use the registered producer offsets as the baseline)? A restore test with a non-divisible parallelism may also be helpful to pin the behaviour down.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/BucketLoadBalanceChannelComputer.java:
##########
@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.flink.sink;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.bucketing.BucketingFunction;
+import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.flink.row.RowWithOp;
+import org.apache.fluss.flink.sink.serializer.FlussSerializationSchema;
+import org.apache.fluss.flink.sink.serializer.SerializerInitContextImpl;
+import org.apache.fluss.metadata.DataLakeFormat;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.encode.KeyEncoder;
+import org.apache.fluss.types.RowType;
+import org.apache.fluss.utils.MathUtils;
+
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * {@link ChannelComputer} for {@link
+ * org.apache.fluss.flink.sink.shuffle.DistributionMode#BUCKET_LOAD_BALANCE}.
+ *
+ * <p>Distributes records by bucket key with even load across all downstream channels. Unlike {@link
+ * FlinkRowDataChannelComputer} (BUCKET mode) which maps each bucket to a single channel, this
+ * computer uses an LCM-based logical slot assignment so that every channel receives traffic from
+ * every bucket.
+ *
+ * <p>The core routing algorithm ({@link #routeByBucket}) is shared between the sink (this class)
+ * and the source-side lookup-join partitioner ({@code FlussLookupInputPartitioner}).
+ *
+ * <p>Records with the same bucket key always route to the same channel.
+ *
+ * @param <InputT> the type of records
+ */
+@Internal
+public class BucketLoadBalanceChannelComputer<InputT> implements ChannelComputer<InputT> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final @Nullable DataLakeFormat lakeFormat;
+ private final int numBucket;
+ private final RowType flussRowType;
+ private final List<String> bucketKeys;
+ private final FlussSerializationSchema<InputT> serializationSchema;
+
+ private transient int numChannels;
+ private transient BucketingFunction bucketingFunction;
+ private transient KeyEncoder bucketKeyEncoder;
+
+ public BucketLoadBalanceChannelComputer(
+ RowType flussRowType,
+ List<String> bucketKeys,
+ @Nullable DataLakeFormat lakeFormat,
+ int numBucket,
+ FlussSerializationSchema<InputT> serializationSchema) {
+ this.flussRowType = flussRowType;
+ this.bucketKeys = bucketKeys;
+ this.lakeFormat = lakeFormat;
+ this.numBucket = numBucket;
+ this.serializationSchema = serializationSchema;
+ }
+
+ @Override
+ public void setup(int numChannels) {
+ this.numChannels = numChannels;
+ this.bucketingFunction = BucketingFunction.of(lakeFormat);
+ this.bucketKeyEncoder = KeyEncoder.ofBucketKeyEncoder(flussRowType, bucketKeys, lakeFormat);
+
+ try {
+ this.serializationSchema.open(new SerializerInitContextImpl(flussRowType, false));
+ } catch (Exception e) {
+ throw new FlussRuntimeException(e);
+ }
+ }
+
+ @Override
+ public int channel(InputT record) {
+ try {
+ RowWithOp rowWithOp = serializationSchema.serialize(record);
+ InternalRow row = rowWithOp.getRow();
+ byte[] bucketKeyBytes = bucketKeyEncoder.encodeKey(row);
+ return routeByBucket(
+ bucketKeyBytes, bucketKeyBytes, numBucket, numChannels, bucketingFunction);
+ } catch (Exception e) {
+ throw new FlussRuntimeException(
+ String.format(
+ "Failed to serialize record of type '%s'"
+ + " in BucketLoadBalanceChannelComputer: %s",
+ record != null ? record.getClass().getName() : "null", e.getMessage()),
+ e);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "BUCKET_LOAD_BALANCE";
+ }
+
+ // ------------------------------------------------------------------------------------------
+ // Shared routing logic
+ // ------------------------------------------------------------------------------------------
+
+ /**
+ * Routes a record to a downstream channel using bucket-ID + LCM-based slot assignment.
+ *
+ * <p>This is the core routing algorithm used by both the sink-side {@code
+ * BucketLoadBalanceChannelComputer} and the source-side lookup-join partitioner ({@code
+ * FlussLookupInputPartitioner}). Callers provide two separate byte arrays so that different
+ * hash keys can be used for the intra-bucket slot selection without affecting bucket-ID
+ * computation.
+ *
+ * <ul>
+ * <li>Sink usage: both {@code bucketKeyBytes} and {@code hashKeyBytes} are the encoded bucket
+ * key — records with the same bucket key always land on the same subtask.
+ * <li>Lookup-join usage: {@code bucketKeyBytes} comes from the bucket-key encoder while
+ * {@code hashKeyBytes} comes from the full lookup-key encoder, giving a different hash
+ * domain within each bucket so that different lookup keys in the same bucket are spread
+ * across the bucket's subtask range.
+ * </ul>
+ *
+ * <p>When the bucket count is evenly divisible by the channel count the hash is skipped
+ * entirely and the result is simply {@code bucketId % numChannels}, matching the behaviour of
+ * the original {@code BUCKET} distribution mode.
+ *
+ * @param bucketKeyBytes encoded bucket key (determines bucket ID)
+ * @param hashKeyBytes encoded key used for the intra-bucket slot hash
+ * @param numBuckets total number of buckets
+ * @param numChannels number of downstream channels (subtask parallelism)
+ * @param bucketingFunction the bucketing function for computing the bucket ID
+ * @return channel index in {@code [0, numChannels)}
+ */
+ public static int routeByBucket(
Review 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.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/ChannelComputer.java:
##########
@@ -62,4 +62,55 @@ static int select(String partitionName, int bucket, int numChannels) {
static int select(int bucket, int numChannels) {
return bucket % numChannels;
}
+
+ /**
+ * Selects a channel using LCM-based logical slot assignment, ensuring even load distribution
+ * across all channels regardless of whether the bucket count and channel count are evenly
+ * divisible.
+ *
+ * <p>Three cases are handled:
+ *
+ * <ul>
+ * <li>{@code numBuckets >= numChannels && numBuckets % numChannels == 0}: simple {@code
+ * bucketId % numChannels}.
+ * <li>{@code numChannels > numBuckets && numChannels % numBuckets == 0}: disjoint
+ * round-robin, preserving the original assignment shape.
+ * <li>Otherwise: LCM-based logical slots — each bucket owns {@code numChannels / gcd}
+ * consecutive slots and each channel owns {@code numBuckets / gcd} consecutive slots. A
+ * hash within the bucket distributes records evenly across channels whose slot ranges
+ * overlap the bucket's range.
+ * </ul>
+ *
+ * @param bucketId the bucket ID computed by {@code BucketingFunction}
+ * @param numBuckets total number of buckets in the table
+ * @param numChannels number of downstream channels (subtask parallelism)
+ * @param recordHash hash of the record (e.g. murmur hash of the bucket key bytes)
+ * @return the target channel index in {@code [0, numChannels)}
+ */
+ static int select(int bucketId, int numBuckets, int numChannels, int recordHash) {
+ if (numBuckets >= numChannels && numBuckets % numChannels == 0) {
Review Comment:
`numChannels` here is the runtime sink parallelism (injected through `FlinkStreamPartitioner#setup`, `FlinkStreamPartitioner.java:39-42`), which means "one physical bucket, one sink subtask" stops being a table property and becomes a function of the current parallelism. Per these formulas — verified for all `numBuckets, numChannels` in [1, 40] — a bucket is shared by several subtasks exactly when `bucket.num % parallelism != 0`, whereas `BUCKET` mode never shares a bucket at any parallelism. For `bucket.num = 12`, only `P in {1,2,3,4,6,12}` is share-free, so `6 -> 8` or `12 -> 13` moves the job across that boundary.
Since users may adjust the sink parallelism of a running job without touching the table or the connector config, this may be worth a thought: a `Conflicting checkpoint offsets` failure during undo recovery (aggregation tables auto-enable it at `FlinkTableSink.java:207`) could stay dormant under one parallelism and only show up on the first recovery after a rescale, and nothing currently detects the shape change — `WriterState` records no parallelism or distribution mode (`WriterState.java:57-69`), while `isAssignedToSubtask` recomputes owners from the new parallelism with the old `bucket % parallelism` formula.
Would it make sense to derive the fan-out from a stable slot count (or record the routing shape in the state as a V3 field), or to use the cluster-authoritative registered producer offsets as the undo baseline, which would drop the single-writer assumption and be rescale-safe by construction? If neither is planned for now, it may help to document that aggregation tables need `bucket.num % sink.parallelism == 0` to keep holding after a parallelism change, and to add rescale tests for divisible -> non-divisible and for scale-up beyond the bucket count.
--
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] Support BUCKET_LOAD_BALANCE shuffle strategy for table with bucket kyes. [fluss]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4450:
URL: https://github.com/apache/fluss/pull/4450#discussion_r4217692864
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/FlinkConnectorOptions.java:
##########
@@ -232,6 +232,7 @@ public class FlinkConnectorOptions {
.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"
Review Comment:
Could we make the guarantee of `BUCKET_LOAD_BALANCE` slightly more precise here?
BUCKET_LOAD_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.
--
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] Support BUCKET_LOAD_BALANCE shuffle strategy for table with bucket kyes. [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #4450:
URL: https://github.com/apache/fluss/pull/4450#issuecomment-5790602466
@leonardBang , I have rebase this PR.
--
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] Support BUCKET_LOAD_BALANCE shuffle strategy for table with bucket kyes. [fluss]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4450:
URL: https://github.com/apache/fluss/pull/4450#discussion_r4217692864
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/FlinkConnectorOptions.java:
##########
@@ -232,6 +232,7 @@ public class FlinkConnectorOptions {
.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"
Review 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.
--
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