Skip to content

Commit b18f95e

Browse files
haruki-830haruki-830
authored
[FLINK-39777][common] Add TableId partitioning strategy configuration (#4423)
1 parent cd8b8a9 commit b18f95e

13 files changed

Lines changed: 612 additions & 49 deletions

File tree

‎docs/content.zh/docs/core-concept/data-pipeline.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -123,5 +123,6 @@ under the License.
123123
| `schema.operator.uid` | Schema 算子的唯一 ID。此 ID 用于算子间通信，必须在所有算子中保持唯一。**已废弃**：请使用 `operator.uid.prefix` 代替。 | optional |
124124
| `schema-operator.rpc-timeout` | SchemaOperator 等待下游 SchemaChangeEvent 应用完成的超时时间，默认值是 3 分钟。 | optional |
125125
| `operator.uid.prefix` | Pipeline 中算子 UID 的前缀。如果不设置，Flink 会为每个算子生成唯一的 UID。 建议设置这个参数以提供稳定和可识别的算子 ID，这有助于有状态升级、问题排查和在 Flink UI 上的诊断。 | optional |
126+
| `sink.partitioning.strategy` | Sink 写入数据时使用的分区策略。数据类型：String。默认值：`SINK_DEFINED`。备注：可配置的值如下：`SINK_DEFINED`：使用 Sink 定义的分区策略；`PRIMARY_KEY`：按表 ID 和主键分区；`TABLE_ID`：仅按表 ID 分区。 | optional |
126127

127128
注意：虽然上述参数都是可选的，但至少需要指定其中一个。`pipeline` 部分是必需的，不能为空。

‎docs/content/docs/core-concept/data-pipeline.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -125,6 +125,7 @@ Note that whilst the parameters are each individually optional, at least one of
125125
| `schema.operator.uid` | The unique ID for schema operator. This ID will be used for inter-operator communications and must be unique across operators. **Deprecated**: use `operator.uid.prefix` instead. | optional |
126126
| `schema-operator.rpc-timeout` | The timeout time for SchemaOperator to wait downstream SchemaChangeEvent applying finished, the default value is 3 minutes. | optional |
127127
| `operator.uid.prefix` | The prefix to use for all pipeline operator UIDs. If not set, all pipeline operator UIDs will be generated by Flink. It is recommended to set this parameter to ensure stable and recognizable operator UIDs, which can help with stateful upgrades, troubleshooting, and Flink UI diagnostics. | optional |
128+
| `sink.partitioning.strategy` | The partitioning strategy used when writing data to the sink. Data type: String. Default value: `SINK_DEFINED`. Available values: `SINK_DEFINED`: uses the partitioning strategy defined by the sink; `PRIMARY_KEY`: partitions by table ID and primary key; `TABLE_ID`: partitions only by table ID. | optional |
128129

129130
NOTE: Whilst the above parameters are each individually optional, at least one of them must be specified. The `pipeline` section is mandatory and cannot be empty.
130131

‎flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParserTest.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -438,6 +438,7 @@ private void testSchemaEvolutionTypesParsing(
438438
ImmutableMap.<String, String>builder()
439439
.put("name", "source-database-sync-pipe")
440440
.put("parallelism", "4")
441+
.put("sink.partitioning.strategy", "TABLE_ID")
441442
.put("execution.runtime-mode", "STREAMING")
442443
.put("schema.change.behavior", "evolve")
443444
.put("schema-operator.rpc-timeout", "1 h")
@@ -488,6 +489,7 @@ void testParsingFullDefinitionFromString() throws Exception {
488489
+ "pipeline:\n"
489490
+ " name: source-database-sync-pipe\n"
490491
+ " parallelism: 4\n"
492+
+ " sink.partitioning.strategy: TABLE_ID\n"
491493
+ " schema.change.behavior: evolve\n"
492494
+ " schema-operator.rpc-timeout: 1 h\n"
493495
+ " execution.runtime-mode: STREAMING\n"
@@ -576,6 +578,7 @@ void testParsingFullDefinitionFromString() throws Exception {
576578
ImmutableMap.<String, String>builder()
577579
.put("name", "source-database-sync-pipe")
578580
.put("parallelism", "4")
581+
.put("sink.partitioning.strategy", "TABLE_ID")
579582
.put("schema.change.behavior", "evolve")
580583
.put("schema-operator.rpc-timeout", "1 h")
581584
.put("execution.runtime-mode", "STREAMING")
@@ -706,6 +709,7 @@ void testParsingFullDefinitionFromString() throws Exception {
706709
ImmutableMap.<String, String>builder()
707710
.put("name", "source-database-sync-pipe")
708711
.put("parallelism", "4")
712+
.put("sink.partitioning.strategy", "TABLE_ID")
709713
.put("schema.change.behavior", "evolve")
710714
.put("schema-operator.rpc-timeout", "1 h")
711715
.build()));

‎flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full-with-repsym.yaml‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -58,5 +58,6 @@ transform:
5858
pipeline:
5959
name: source-database-sync-pipe
6060
parallelism: 4
61+
sink.partitioning.strategy: TABLE_ID
6162
schema.change.behavior: evolve
62-
schema-operator.rpc-timeout: 1 h
63+
schema-operator.rpc-timeout: 1 h

‎flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full.yaml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@ transform:
5656
pipeline:
5757
name: source-database-sync-pipe
5858
parallelism: 4
59+
sink.partitioning.strategy: TABLE_ID
5960
schema.change.behavior: evolve
6061
schema-operator.rpc-timeout: 1 h
6162
execution.runtime-mode: STREAMING

‎flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/HashFunctionStrategy.java‎

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,77 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.flink.cdc.common.pipeline;
19+
20+
import org.apache.flink.cdc.common.annotation.PublicEvolving;
21+
22+
/**
23+
* Partitioning strategy for {@link org.apache.flink.cdc.common.event.DataChangeEvent} in Flink CDC
24+
* pipeline.
25+
*
26+
* <p>This enum defines different strategies used by {@code PrePartitionOperator} to determine which
27+
* downstream subtask an event should be routed to.
28+
*/
29+
@PublicEvolving
30+
public enum HashFunctionStrategy {
31+
32+
/**
33+
* Use the HashFunctionProvider defined by the sink.
34+
*
35+
* <p>This is the default behavior when {@code sink.partitioning.strategy} is not explicitly
36+
* set. The provider returned by the sink's {@code getDataChangeEventHashFunctionProvider}
37+
* method will be used, preserving backward compatibility with existing pipeline configurations.
38+
*
39+
* <p>This is the only supported strategy for sinks that require sink-defined routing, such as
40+
* paimon, fluss, and maxcompute.
41+
*/
42+
SINK_DEFINED,
43+
44+
/**
45+
* Hash by TableId and primary keys.
46+
*
47+
* <p>This strategy computes hash based on TableId (namespace, schema, table) combined with
48+
* primary key column values. Events with the same TableId and primary key will be routed to the
49+
* same subtask, while events from the same table with different primary keys may be spread
50+
* across multiple subtasks for load balancing.
51+
*
52+
* <p>This is suitable when you want good distribution of data across subtasks and don't require
53+
* strict ordering within a table.
54+
*
55+
* <p>This strategy is not supported for paimon, fluss, or maxcompute sinks.
56+
*/
57+
PRIMARY_KEY,
58+
59+
/**
60+
* Hash by TableId only.
61+
*
62+
* <p>This strategy computes hash based solely on TableId (namespace, schema, table), ignoring
63+
* the record payload entirely. All events from the same table will be routed to the same
64+
* subtask, ensuring per-table ordering semantics.
65+
*
66+
* <p>This is suitable when:
67+
*
68+
* <ul>
69+
* <li>The sink requires strict single-writer-per-table semantics.
70+
* <li>The table has no primary key or has a changing primary key set.
71+
* <li>You want to avoid cross-subtask reordering within the same table.
72+
* </ul>
73+
*
74+
* <p>This strategy is not supported for paimon, fluss, or maxcompute sinks.
75+
*/
76+
TABLE_ID
77+
}

‎flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -136,5 +136,25 @@ public class PipelineOptions {
136136
.withDescription(
137137
"The timeout time for SchemaOperator to wait downstream SchemaChangeEvent applying finished, the default value is 3 minutes.");
138138

139+
public static final ConfigOption<HashFunctionStrategy> PIPELINE_PARTITIONING_STRATEGY =
140+
ConfigOptions.key("sink.partitioning.strategy")
141+
.enumType(HashFunctionStrategy.class)
142+
.defaultValue(HashFunctionStrategy.SINK_DEFINED)
143+
.withDescription(
144+
Description.builder()
145+
.text(
146+
"Partitioning strategy for DataChangeEvent routing. Defaults to SINK_DEFINED. "
147+
+ "Paimon, Fluss, and MaxCompute sinks only support SINK_DEFINED.")
148+
.linebreak()
149+
.add(
150+
ListElement.list(
151+
text(
152+
"SINK_DEFINED: Use the HashFunctionProvider defined by the sink (default). This is the only supported strategy for paimon, fluss, and maxcompute sinks."),
153+
text(
154+
"PRIMARY_KEY: Hash by TableId and primary keys. Events from the same table may be distributed across multiple subtasks for load balancing. This strategy is not supported for paimon, fluss, or maxcompute sinks."),
155+
text(
156+
"TABLE_ID: Hash by TableId only. All events from the same table will land on the same subtask, ensuring per-table ordering semantics. This strategy is not supported for paimon, fluss, or maxcompute sinks.")))
157+
.build());
158+
139159
private PipelineOptions() {}
140160
}

‎flink-cdc-common/src/main/java/org/apache/flink/cdc/common/sink/TableIdHashFunctionProvider.java‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.flink.cdc.common.sink;
19+
20+
import org.apache.flink.cdc.common.event.DataChangeEvent;
21+
import org.apache.flink.cdc.common.event.TableId;
22+
import org.apache.flink.cdc.common.function.HashFunction;
23+
import org.apache.flink.cdc.common.function.HashFunctionProvider;
24+
import org.apache.flink.cdc.common.schema.Schema;
25+
26+
import javax.annotation.Nullable;
27+
28+
/**
29+
* A {@link HashFunctionProvider} implementation that hashes events based solely on {@link TableId}.
30+
*
31+
* <p>This provider ensures all events from the same table are routed to the same downstream
32+
* subtask, regardless of their primary key values or record payload. The hash is computed once when
33+
* the HashFunction is created and cached for all subsequent events.
34+
*/
35+
public class TableIdHashFunctionProvider implements HashFunctionProvider<DataChangeEvent> {
36+
37+
private static final long serialVersionUID = 1L;
38+
39+
@Override
40+
public HashFunction<DataChangeEvent> getHashFunction(@Nullable TableId tableId, Schema schema) {
41+
return new TableIdHashFunction(tableId);
42+
}
43+
44+
/** A {@link HashFunction} that computes hash based solely on TableId, cached at creation. */
45+
static class TableIdHashFunction implements HashFunction<DataChangeEvent> {
46+
47+
private final int cachedHash;
48+
49+
TableIdHashFunction(TableId tableId) {
50+
this.cachedHash = (tableId.hashCode() * 31) & 0x7FFFFFFF;
51+
}
52+
53+
@Override
54+
public int hashcode(DataChangeEvent event) {
55+
return cachedHash;
56+
}
57+
}
58+
}

0 commit comments

Comments
 (0)