This is an automated email from the ASF dual-hosted git repository.
lvyanquan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new b18f95e67 [FLINK-39777][common] Add TableId partitioning strategy
configuration (#4423)
b18f95e67 is described below
commit b18f95e67293475eafb51247a182f5e661e2a4ca
Author: haruki <[email protected]>
AuthorDate: Tue Jul 14 13:57:50 2026 +0800
[FLINK-39777][common] Add TableId partitioning strategy configuration
(#4423)
---
docs/content.zh/docs/core-concept/data-pipeline.md | 1 +
docs/content/docs/core-concept/data-pipeline.md | 1 +
.../parser/YamlPipelineDefinitionParserTest.java | 4 +
.../pipeline-definition-full-with-repsym.yaml | 3 +-
.../definitions/pipeline-definition-full.yaml | 1 +
.../cdc/common/pipeline/HashFunctionStrategy.java | 77 ++++++++++
.../flink/cdc/common/pipeline/PipelineOptions.java | 20 +++
.../common/sink/TableIdHashFunctionProvider.java | 58 ++++++++
.../sink/TableIdHashFunctionProviderTest.java | 158 +++++++++++++++++++++
.../cdc/composer/flink/FlinkPipelineComposer.java | 90 +++++++++---
.../composer/flink/FlinkPipelineComposerTest.java | 106 +++++++++++++-
.../cdc/pipeline/tests/MySqlToDorisE2eITCase.java | 69 +++++----
.../partitioning/PrePartitionOperatorTest.java | 73 ++++++++++
13 files changed, 612 insertions(+), 49 deletions(-)
diff --git a/docs/content.zh/docs/core-concept/data-pipeline.md
b/docs/content.zh/docs/core-concept/data-pipeline.md
index 6bde6e462..928f6880c 100644
--- a/docs/content.zh/docs/core-concept/data-pipeline.md
+++ b/docs/content.zh/docs/core-concept/data-pipeline.md
@@ -123,5 +123,6 @@ under the License.
| `schema.operator.uid` | Schema 算子的唯一 ID。此 ID
用于算子间通信，必须在所有算子中保持唯一。**已废弃**：请使用 `operator.uid.prefix` 代替。
[...]
| `schema-operator.rpc-timeout` | SchemaOperator 等待下游 SchemaChangeEvent
应用完成的超时时间，默认值是 3 分钟。
[...]
| `operator.uid.prefix` | Pipeline 中算子 UID 的前缀。如果不设置，Flink 会为每个算子生成唯一的
UID。 建议设置这个参数以提供稳定和可识别的算子 ID，这有助于有状态升级、问题排查和在 Flink UI 上的诊断。
[...]
+| `sink.partitioning.strategy` | Sink
写入数据时使用的分区策略。数据类型：String。默认值：`SINK_DEFINED`。备注：可配置的值如下：`SINK_DEFINED`：使用 Sink
定义的分区策略；`PRIMARY_KEY`：按表 ID 和主键分区；`TABLE_ID`：仅按表 ID 分区。
| optional |
注意：虽然上述参数都是可选的，但至少需要指定其中一个。`pipeline` 部分是必需的，不能为空。
diff --git a/docs/content/docs/core-concept/data-pipeline.md
b/docs/content/docs/core-concept/data-pipeline.md
index 2a9a27b50..68f3ca212 100644
--- a/docs/content/docs/core-concept/data-pipeline.md
+++ b/docs/content/docs/core-concept/data-pipeline.md
@@ -125,6 +125,7 @@ Note that whilst the parameters are each individually
optional, at least one of
| `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.
[...]
| `schema-operator.rpc-timeout` | The timeout time for SchemaOperator to wait
downstream SchemaChangeEvent applying finished, the default value is 3 minutes.
[...]
| `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.
[...]
+| `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.
[...]
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.
diff --git
a/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParserTest.java
b/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParserTest.java
index c8c10db76..4ecc72d14 100644
---
a/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParserTest.java
+++
b/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParserTest.java
@@ -438,6 +438,7 @@ class YamlPipelineDefinitionParserTest {
ImmutableMap.<String, String>builder()
.put("name", "source-database-sync-pipe")
.put("parallelism", "4")
+ .put("sink.partitioning.strategy",
"TABLE_ID")
.put("execution.runtime-mode", "STREAMING")
.put("schema.change.behavior", "evolve")
.put("schema-operator.rpc-timeout", "1 h")
@@ -488,6 +489,7 @@ class YamlPipelineDefinitionParserTest {
+ "pipeline:\n"
+ " name: source-database-sync-pipe\n"
+ " parallelism: 4\n"
+ + " sink.partitioning.strategy: TABLE_ID\n"
+ " schema.change.behavior: evolve\n"
+ " schema-operator.rpc-timeout: 1 h\n"
+ " execution.runtime-mode: STREAMING\n"
@@ -576,6 +578,7 @@ class YamlPipelineDefinitionParserTest {
ImmutableMap.<String, String>builder()
.put("name", "source-database-sync-pipe")
.put("parallelism", "4")
+ .put("sink.partitioning.strategy",
"TABLE_ID")
.put("schema.change.behavior", "evolve")
.put("schema-operator.rpc-timeout", "1 h")
.put("execution.runtime-mode", "STREAMING")
@@ -706,6 +709,7 @@ class YamlPipelineDefinitionParserTest {
ImmutableMap.<String, String>builder()
.put("name", "source-database-sync-pipe")
.put("parallelism", "4")
+ .put("sink.partitioning.strategy",
"TABLE_ID")
.put("schema.change.behavior", "evolve")
.put("schema-operator.rpc-timeout", "1 h")
.build()));
diff --git
a/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full-with-repsym.yaml
b/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full-with-repsym.yaml
index 10c6f6e0c..a00cfda2f 100644
---
a/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full-with-repsym.yaml
+++
b/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full-with-repsym.yaml
@@ -58,5 +58,6 @@ transform:
pipeline:
name: source-database-sync-pipe
parallelism: 4
+ sink.partitioning.strategy: TABLE_ID
schema.change.behavior: evolve
- schema-operator.rpc-timeout: 1 h
\ No newline at end of file
+ schema-operator.rpc-timeout: 1 h
diff --git
a/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full.yaml
b/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full.yaml
index dec2c25dc..d1a694d70 100644
--- a/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full.yaml
+++ b/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full.yaml
@@ -56,6 +56,7 @@ transform:
pipeline:
name: source-database-sync-pipe
parallelism: 4
+ sink.partitioning.strategy: TABLE_ID
schema.change.behavior: evolve
schema-operator.rpc-timeout: 1 h
execution.runtime-mode: STREAMING
diff --git
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/HashFunctionStrategy.java
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/HashFunctionStrategy.java
new file mode 100644
index 000000000..06898e839
--- /dev/null
+++
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/HashFunctionStrategy.java
@@ -0,0 +1,77 @@
+/*
+ * 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.flink.cdc.common.pipeline;
+
+import org.apache.flink.cdc.common.annotation.PublicEvolving;
+
+/**
+ * Partitioning strategy for {@link
org.apache.flink.cdc.common.event.DataChangeEvent} in Flink CDC
+ * pipeline.
+ *
+ * <p>This enum defines different strategies used by {@code
PrePartitionOperator} to determine which
+ * downstream subtask an event should be routed to.
+ */
+@PublicEvolving
+public enum HashFunctionStrategy {
+
+ /**
+ * Use the HashFunctionProvider defined by the sink.
+ *
+ * <p>This is the default behavior when {@code sink.partitioning.strategy}
is not explicitly
+ * set. The provider returned by the sink's {@code
getDataChangeEventHashFunctionProvider}
+ * method will be used, preserving backward compatibility with existing
pipeline configurations.
+ *
+ * <p>This is the only supported strategy for sinks that require
sink-defined routing, such as
+ * paimon, fluss, and maxcompute.
+ */
+ SINK_DEFINED,
+
+ /**
+ * Hash by TableId and primary keys.
+ *
+ * <p>This strategy computes hash based on TableId (namespace, schema,
table) combined with
+ * primary key column values. Events with the same TableId and primary key
will be routed to the
+ * same subtask, while events from the same table with different primary
keys may be spread
+ * across multiple subtasks for load balancing.
+ *
+ * <p>This is suitable when you want good distribution of data across
subtasks and don't require
+ * strict ordering within a table.
+ *
+ * <p>This strategy is not supported for paimon, fluss, or maxcompute
sinks.
+ */
+ PRIMARY_KEY,
+
+ /**
+ * Hash by TableId only.
+ *
+ * <p>This strategy computes hash based solely on TableId (namespace,
schema, table), ignoring
+ * the record payload entirely. All events from the same table will be
routed to the same
+ * subtask, ensuring per-table ordering semantics.
+ *
+ * <p>This is suitable when:
+ *
+ * <ul>
+ * <li>The sink requires strict single-writer-per-table semantics.
+ * <li>The table has no primary key or has a changing primary key set.
+ * <li>You want to avoid cross-subtask reordering within the same table.
+ * </ul>
+ *
+ * <p>This strategy is not supported for paimon, fluss, or maxcompute
sinks.
+ */
+ TABLE_ID
+}
diff --git
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java
index 443d6145b..7c6e49b9e 100644
---
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java
+++
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java
@@ -136,5 +136,25 @@ public class PipelineOptions {
.withDescription(
"The timeout time for SchemaOperator to wait
downstream SchemaChangeEvent applying finished, the default value is 3
minutes.");
+ public static final ConfigOption<HashFunctionStrategy>
PIPELINE_PARTITIONING_STRATEGY =
+ ConfigOptions.key("sink.partitioning.strategy")
+ .enumType(HashFunctionStrategy.class)
+ .defaultValue(HashFunctionStrategy.SINK_DEFINED)
+ .withDescription(
+ Description.builder()
+ .text(
+ "Partitioning strategy for
DataChangeEvent routing. Defaults to SINK_DEFINED. "
+ + "Paimon, Fluss, and
MaxCompute sinks only support SINK_DEFINED.")
+ .linebreak()
+ .add(
+ ListElement.list(
+ text(
+ "SINK_DEFINED: Use
the HashFunctionProvider defined by the sink (default). This is the only
supported strategy for paimon, fluss, and maxcompute sinks."),
+ text(
+ "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."),
+ text(
+ "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.")))
+ .build());
+
private PipelineOptions() {}
}
diff --git
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/sink/TableIdHashFunctionProvider.java
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/sink/TableIdHashFunctionProvider.java
new file mode 100644
index 000000000..436f84288
--- /dev/null
+++
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/sink/TableIdHashFunctionProvider.java
@@ -0,0 +1,58 @@
+/*
+ * 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.flink.cdc.common.sink;
+
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.function.HashFunction;
+import org.apache.flink.cdc.common.function.HashFunctionProvider;
+import org.apache.flink.cdc.common.schema.Schema;
+
+import javax.annotation.Nullable;
+
+/**
+ * A {@link HashFunctionProvider} implementation that hashes events based
solely on {@link TableId}.
+ *
+ * <p>This provider ensures all events from the same table are routed to the
same downstream
+ * subtask, regardless of their primary key values or record payload. The hash
is computed once when
+ * the HashFunction is created and cached for all subsequent events.
+ */
+public class TableIdHashFunctionProvider implements
HashFunctionProvider<DataChangeEvent> {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public HashFunction<DataChangeEvent> getHashFunction(@Nullable TableId
tableId, Schema schema) {
+ return new TableIdHashFunction(tableId);
+ }
+
+ /** A {@link HashFunction} that computes hash based solely on TableId,
cached at creation. */
+ static class TableIdHashFunction implements HashFunction<DataChangeEvent> {
+
+ private final int cachedHash;
+
+ TableIdHashFunction(TableId tableId) {
+ this.cachedHash = (tableId.hashCode() * 31) & 0x7FFFFFFF;
+ }
+
+ @Override
+ public int hashcode(DataChangeEvent event) {
+ return cachedHash;
+ }
+ }
+}
diff --git
a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/sink/TableIdHashFunctionProviderTest.java
b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/sink/TableIdHashFunctionProviderTest.java
new file mode 100644
index 000000000..1e8195fc2
--- /dev/null
+++
b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/sink/TableIdHashFunctionProviderTest.java
@@ -0,0 +1,158 @@
+/*
+ * 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.flink.cdc.common.sink;
+
+import org.apache.flink.cdc.common.data.GenericRecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.function.HashFunction;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit test for {@link TableIdHashFunctionProvider}. */
+class TableIdHashFunctionProviderTest {
+
+ private static final TableId TABLE_A = TableId.tableId("namespace",
"schema", "table_a");
+ private static final TableId TABLE_B = TableId.tableId("namespace",
"schema", "table_b");
+ private static final TableId TABLE_C = TableId.tableId("other_namespace",
"schema", "table_a");
+
+ private static final Schema SCHEMA =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT())
+ .physicalColumn("name", DataTypes.STRING())
+ .primaryKey("id")
+ .build();
+
+ private final TableIdHashFunctionProvider provider = new
TableIdHashFunctionProvider();
+
+ @Test
+ void testSameTableSameHash() {
+ HashFunction<DataChangeEvent> hashFunction =
provider.getHashFunction(TABLE_A, SCHEMA);
+
+ DataChangeEvent event1 = createInsertEvent(TABLE_A, 1, "Alice");
+ DataChangeEvent event2 = createInsertEvent(TABLE_A, 2, "Bob");
+ DataChangeEvent event3 = createInsertEvent(TABLE_A, 3, "Charlie");
+
+ int hash1 = hashFunction.hashcode(event1);
+ int hash2 = hashFunction.hashcode(event2);
+ int hash3 = hashFunction.hashcode(event3);
+
+ // All events from the same table should have the same hash
+ assertThat(hash1).isEqualTo(hash2);
+ assertThat(hash2).isEqualTo(hash3);
+ }
+
+ @Test
+ void testDifferentTableDifferentHash() {
+ HashFunction<DataChangeEvent> hashFunctionA =
provider.getHashFunction(TABLE_A, SCHEMA);
+ HashFunction<DataChangeEvent> hashFunctionB =
provider.getHashFunction(TABLE_B, SCHEMA);
+
+ DataChangeEvent eventA = createInsertEvent(TABLE_A, 1, "Alice");
+ DataChangeEvent eventB = createInsertEvent(TABLE_B, 1, "Alice");
+
+ int hashA = hashFunctionA.hashcode(eventA);
+ int hashB = hashFunctionB.hashcode(eventB);
+
+ // Events from different tables should have different hashes
+ assertThat(hashA).isNotEqualTo(hashB);
+ }
+
+ @Test
+ void testDifferentNamespaceDifferentHash() {
+ HashFunction<DataChangeEvent> hashFunctionA =
provider.getHashFunction(TABLE_A, SCHEMA);
+ HashFunction<DataChangeEvent> hashFunctionC =
provider.getHashFunction(TABLE_C, SCHEMA);
+
+ DataChangeEvent eventA = createInsertEvent(TABLE_A, 1, "Alice");
+ DataChangeEvent eventC = createInsertEvent(TABLE_C, 1, "Alice");
+
+ int hashA = hashFunctionA.hashcode(eventA);
+ int hashC = hashFunctionC.hashcode(eventC);
+
+ // Events from different namespaces should have different hashes
+ assertThat(hashA).isNotEqualTo(hashC);
+ }
+
+ @Test
+ void testHashIsPositive() {
+ HashFunction<DataChangeEvent> hashFunction =
provider.getHashFunction(TABLE_A, SCHEMA);
+
+ DataChangeEvent event = createInsertEvent(TABLE_A, 1, "Alice");
+ int hash = hashFunction.hashcode(event);
+
+ // Hash should always be non-negative
+ assertThat(hash).isGreaterThanOrEqualTo(0);
+ }
+
+ @Test
+ void testHashIgnoresPrimaryKeyValue() {
+ HashFunction<DataChangeEvent> hashFunction =
provider.getHashFunction(TABLE_A, SCHEMA);
+
+ // Create events with different primary key values
+ DataChangeEvent eventPk1 = createInsertEvent(TABLE_A, 1, "Alice");
+ DataChangeEvent eventPk2 = createInsertEvent(TABLE_A, 2, "Bob");
+ DataChangeEvent eventPk3 = createInsertEvent(TABLE_A, 100, "Charlie");
+
+ int hash1 = hashFunction.hashcode(eventPk1);
+ int hash2 = hashFunction.hashcode(eventPk2);
+ int hash3 = hashFunction.hashcode(eventPk3);
+
+ // Hash should be the same regardless of primary key values
+ assertThat(hash1).isEqualTo(hash2);
+ assertThat(hash2).isEqualTo(hash3);
+ }
+
+ @Test
+ void testHashIgnoresOperationType() {
+ HashFunction<DataChangeEvent> hashFunction =
provider.getHashFunction(TABLE_A, SCHEMA);
+
+ DataChangeEvent insertEvent = createInsertEvent(TABLE_A, 1, "Alice");
+ DataChangeEvent updateEvent = createUpdateEvent(TABLE_A, 1, "Alice",
"Alice_Updated");
+ DataChangeEvent deleteEvent = createDeleteEvent(TABLE_A, 1, "Alice");
+
+ int insertHash = hashFunction.hashcode(insertEvent);
+ int updateHash = hashFunction.hashcode(updateEvent);
+ int deleteHash = hashFunction.hashcode(deleteEvent);
+
+ // Hash should be the same regardless of operation type
+ assertThat(insertHash).isEqualTo(updateHash);
+ assertThat(updateHash).isEqualTo(deleteHash);
+ }
+
+ private DataChangeEvent createInsertEvent(TableId tableId, int id, String
name) {
+ return DataChangeEvent.insertEvent(
+ tableId, GenericRecordData.of(id, new BinaryStringData(name)));
+ }
+
+ private DataChangeEvent createUpdateEvent(
+ TableId tableId, int id, String oldName, String newName) {
+ return DataChangeEvent.updateEvent(
+ tableId,
+ GenericRecordData.of(id, new BinaryStringData(oldName)),
+ GenericRecordData.of(id, new BinaryStringData(newName)));
+ }
+
+ private DataChangeEvent createDeleteEvent(TableId tableId, int id, String
name) {
+ return DataChangeEvent.deleteEvent(
+ tableId, GenericRecordData.of(id, new BinaryStringData(name)));
+ }
+}
diff --git
a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
index 1854b0503..4d843ac87 100644
---
a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
+++
b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
@@ -20,11 +20,16 @@ package org.apache.flink.cdc.composer.flink;
import org.apache.flink.cdc.common.annotation.Internal;
import org.apache.flink.cdc.common.annotation.VisibleForTesting;
import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.function.HashFunctionProvider;
+import org.apache.flink.cdc.common.pipeline.HashFunctionStrategy;
import org.apache.flink.cdc.common.pipeline.PipelineOptions;
import org.apache.flink.cdc.common.pipeline.RuntimeExecutionMode;
import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
import org.apache.flink.cdc.common.sink.DataSink;
+import
org.apache.flink.cdc.common.sink.DefaultDataChangeEventHashFunctionProvider;
+import org.apache.flink.cdc.common.sink.TableIdHashFunctionProvider;
import org.apache.flink.cdc.common.source.DataSource;
import org.apache.flink.cdc.composer.PipelineComposer;
import org.apache.flink.cdc.composer.PipelineExecution;
@@ -128,22 +133,13 @@ public class FlinkPipelineComposer implements
PipelineComposer {
}
// Validate configuration
+ validatePipelineConfiguration(pipelineDef.getConfig());
+
String schemaOperatorUid =
pipelineDefConfig.get(PipelineOptions.PIPELINE_SCHEMA_OPERATOR_UID);
@Nullable
String operatorUidPrefix =
pipelineDefConfig.get(PipelineOptions.PIPELINE_OPERATOR_UID_PREFIX);
- if (!Objects.equals(
- schemaOperatorUid,
-
PipelineOptions.PIPELINE_SCHEMA_OPERATOR_UID.defaultValue())
- && operatorUidPrefix != null) {
- throw new IllegalArgumentException(
- String.format(
- "Only one of the %s and %s pipeline options can be
set.",
- PipelineOptions.PIPELINE_OPERATOR_UID_PREFIX.key(),
-
PipelineOptions.PIPELINE_SCHEMA_OPERATOR_UID.key()));
- }
-
OperatorUidGenerator operatorUidGenerator = new
OperatorUidGenerator(operatorUidPrefix);
if (operatorUidPrefix != null) {
@@ -168,6 +164,12 @@ public class FlinkPipelineComposer implements
PipelineComposer {
sourceTranslator.createDataSource(pipelineDef.getSource(),
pipelineDefConfig, env);
DataSink dataSink =
sinkTranslator.createDataSink(pipelineDef.getSink(),
pipelineDefConfig, env);
+ HashFunctionProvider<DataChangeEvent> sinkDefinedHashFunctionProvider =
+ dataSink.getDataChangeEventHashFunctionProvider(parallelism);
+ validatePartitioningStrategyCompatibility(pipelineDef,
sinkDefinedHashFunctionProvider);
+
+ HashFunctionProvider<DataChangeEvent> hashFunctionProvider =
+ resolveHashFunctionProvider(pipelineDefConfig,
sinkDefinedHashFunctionProvider);
boolean isParallelMetadataSource =
dataSource.isParallelMetadataSource();
@@ -205,10 +207,7 @@ public class FlinkPipelineComposer implements
PipelineComposer {
// PostTransform -> Partitioning
DataStream<PartitioningEvent> partitionedStream =
partitioningTranslator.translateDistributed(
- stream,
- parallelism,
- parallelism,
-
dataSink.getDataChangeEventHashFunctionProvider(parallelism));
+ stream, parallelism, parallelism,
hashFunctionProvider);
// Partitioning -> Schema Operator
stream =
@@ -247,7 +246,7 @@ public class FlinkPipelineComposer implements
PipelineComposer {
parallelism,
isBatchMode,
schemaOperatorIDGenerator.generate(),
-
dataSink.getDataChangeEventHashFunctionProvider(parallelism),
+ hashFunctionProvider,
operatorUidGenerator);
}
@@ -292,6 +291,65 @@ public class FlinkPipelineComposer implements
PipelineComposer {
return Optional.of(container);
}
+ private void validatePipelineConfiguration(Configuration pipelineConfig) {
+ String schemaOperatorUid =
pipelineConfig.get(PipelineOptions.PIPELINE_SCHEMA_OPERATOR_UID);
+ @Nullable
+ String operatorUidPrefix =
pipelineConfig.get(PipelineOptions.PIPELINE_OPERATOR_UID_PREFIX);
+ if (!Objects.equals(
+ schemaOperatorUid,
+
PipelineOptions.PIPELINE_SCHEMA_OPERATOR_UID.defaultValue())
+ && operatorUidPrefix != null) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Only one of the %s and %s pipeline options can be
set.",
+ PipelineOptions.PIPELINE_OPERATOR_UID_PREFIX.key(),
+
PipelineOptions.PIPELINE_SCHEMA_OPERATOR_UID.key()));
+ }
+ }
+
+ private void validatePartitioningStrategyCompatibility(
+ PipelineDef pipelineDef,
+ HashFunctionProvider<DataChangeEvent>
sinkDefinedHashFunctionProvider) {
+ HashFunctionStrategy partitioningStrategy =
+
pipelineDef.getConfig().get(PipelineOptions.PIPELINE_PARTITIONING_STRATEGY);
+ if (!(sinkDefinedHashFunctionProvider instanceof
DefaultDataChangeEventHashFunctionProvider)
+ && partitioningStrategy != HashFunctionStrategy.SINK_DEFINED) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Sink type '%s' only supports %s=%s when
DataSink.getDataChangeEventHashFunctionProvider(...) returns a custom
HashFunctionProvider, but found %s.",
+ pipelineDef.getSink().getType(),
+
PipelineOptions.PIPELINE_PARTITIONING_STRATEGY.key(),
+ HashFunctionStrategy.SINK_DEFINED,
+ partitioningStrategy));
+ }
+ }
+
+ /**
+ * Resolves the {@link HashFunctionProvider} to use for partitioning based
on pipeline
+ * configuration.
+ *
+ * @param pipelineConfig the pipeline configuration
+ * @param sinkDefinedHashFunctionProvider the provider returned by the sink
+ * @return the resolved HashFunctionProvider
+ */
+ private HashFunctionProvider<DataChangeEvent> resolveHashFunctionProvider(
+ Configuration pipelineConfig,
+ HashFunctionProvider<DataChangeEvent>
sinkDefinedHashFunctionProvider) {
+ HashFunctionStrategy strategy =
+
pipelineConfig.get(PipelineOptions.PIPELINE_PARTITIONING_STRATEGY);
+
+ switch (strategy) {
+ case SINK_DEFINED:
+ return sinkDefinedHashFunctionProvider;
+ case PRIMARY_KEY:
+ return new DefaultDataChangeEventHashFunctionProvider();
+ case TABLE_ID:
+ return new TableIdHashFunctionProvider();
+ }
+
+ throw new IllegalStateException("Unexpected hash function strategy: "
+ strategy);
+ }
+
@VisibleForTesting
public StreamExecutionEnvironment getEnv() {
return env;
diff --git
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerTest.java
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerTest.java
index 15bcd0495..a9da9c398 100644
---
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerTest.java
+++
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerTest.java
@@ -18,11 +18,17 @@
package org.apache.flink.cdc.composer.flink;
import org.apache.flink.cdc.common.configuration.ConfigOption;
+import org.apache.flink.cdc.common.configuration.ConfigOptions;
import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.factories.DataSinkFactory;
import org.apache.flink.cdc.common.factories.DataSourceFactory;
import org.apache.flink.cdc.common.factories.FactoryHelper;
+import org.apache.flink.cdc.common.function.HashFunctionProvider;
+import org.apache.flink.cdc.common.pipeline.HashFunctionStrategy;
import org.apache.flink.cdc.common.pipeline.PipelineOptions;
+import org.apache.flink.cdc.common.schema.Schema;
import org.apache.flink.cdc.common.sink.DataSink;
import org.apache.flink.cdc.common.sink.EventSinkProvider;
import org.apache.flink.cdc.common.sink.MetadataApplier;
@@ -102,15 +108,29 @@ class FlinkPipelineComposerTest {
public static class TestDataSinkFactory implements DataSinkFactory {
public static final String IDENTIFIER = "test-sink-factory";
+ public static final ConfigOption<Boolean> CUSTOM_HASH_PROVIDER =
+
ConfigOptions.key("custom-hash-provider").booleanType().defaultValue(false);
@Override
public DataSink createDataSink(Context context) {
- // This option has no default value.
String target =
context.getFlinkConf().get(DeploymentOptions.TARGET);
if (!"local".equals(target)) {
throw new IllegalArgumentException(
"The flink configuration is invalid. Please check the
pipeline configuration.");
}
+ if (!context.getFactoryConfiguration().get(CUSTOM_HASH_PROVIDER)) {
+ return new DataSink() {
+ @Override
+ public EventSinkProvider getEventSinkProvider() {
+ return null;
+ }
+
+ @Override
+ public MetadataApplier getMetadataApplier() {
+ return schemaChangeEvent -> {};
+ }
+ };
+ }
return new DataSink() {
@Override
public EventSinkProvider getEventSinkProvider() {
@@ -121,6 +141,12 @@ class FlinkPipelineComposerTest {
public MetadataApplier getMetadataApplier() {
return schemaChangeEvent -> {};
}
+
+ @Override
+ public HashFunctionProvider<DataChangeEvent>
getDataChangeEventHashFunctionProvider(
+ int parallelism) {
+ return new TestHashFunctionProvider();
+ }
};
}
@@ -136,7 +162,17 @@ class FlinkPipelineComposerTest {
@Override
public Set<ConfigOption<?>> optionalOptions() {
- return new HashSet<>();
+ Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(CUSTOM_HASH_PROVIDER);
+ return options;
+ }
+ }
+
+ private static class TestHashFunctionProvider implements
HashFunctionProvider<DataChangeEvent> {
+ @Override
+ public
org.apache.flink.cdc.common.function.HashFunction<DataChangeEvent>
getHashFunction(
+ TableId tableId, Schema schema) {
+ return event -> 0;
}
}
@@ -191,6 +227,45 @@ class FlinkPipelineComposerTest {
return Stream.of(Arguments.of(configuration1));
}
+ @ParameterizedTest
+ @MethodSource
+ void testSinkDefinedOnlyPartitioningStrategyConfiguration(
+ HashFunctionStrategy partitioningStrategy) {
+ FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
+ Configuration pipelineConfig = new Configuration();
+ pipelineConfig.set(PipelineOptions.PIPELINE_PARTITIONING_STRATEGY,
partitioningStrategy);
+ Configuration sinkConfig = new Configuration();
+ sinkConfig.set(TestDataSinkFactory.CUSTOM_HASH_PROVIDER, true);
+ PipelineDef pipelineDef =
+ buildPipelineDefinitionFromConfiguration(
+ TestDataSinkFactory.IDENTIFIER, sinkConfig,
pipelineConfig);
+
+ assertThatThrownBy(() -> composer.compose(pipelineDef))
+ .isExactlyInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Sink type '" +
TestDataSinkFactory.IDENTIFIER + "'")
+
.hasMessageContaining(PipelineOptions.PIPELINE_PARTITIONING_STRATEGY.key())
+ .hasMessageContaining(HashFunctionStrategy.SINK_DEFINED.name())
+ .hasMessageContaining(partitioningStrategy.name());
+ }
+
+ static Stream<Arguments>
testSinkDefinedOnlyPartitioningStrategyConfiguration() {
+ return Stream.of(
+ Arguments.of(HashFunctionStrategy.PRIMARY_KEY),
+ Arguments.of(HashFunctionStrategy.TABLE_ID));
+ }
+
+ @Test
+ void testSinkDefinedOnlyPartitioningStrategyDefaultConfiguration() {
+ FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
+ Configuration sinkConfig = new Configuration();
+ sinkConfig.set(TestDataSinkFactory.CUSTOM_HASH_PROVIDER, true);
+ PipelineDef pipelineDef =
+ buildPipelineDefinitionFromConfiguration(
+ TestDataSinkFactory.IDENTIFIER, sinkConfig, new
Configuration());
+
+ assertThatCode(() ->
composer.compose(pipelineDef)).doesNotThrowAnyException();
+ }
+
@ParameterizedTest
@MethodSource
void testValidPipelineConfiguration(Configuration pipelineConfig) {
@@ -208,16 +283,39 @@ class FlinkPipelineComposerTest {
Configuration configuration3 = new Configuration();
configuration3.set(PipelineOptions.PIPELINE_OPERATOR_UID_PREFIX,
"junit");
+ // Test HashFunctionStrategy configurations
+ Configuration configuration4 = new Configuration();
+ configuration4.set(
+ PipelineOptions.PIPELINE_PARTITIONING_STRATEGY,
HashFunctionStrategy.PRIMARY_KEY);
+
+ Configuration configuration5 = new Configuration();
+ configuration5.set(
+ PipelineOptions.PIPELINE_PARTITIONING_STRATEGY,
HashFunctionStrategy.TABLE_ID);
+
return Stream.of(
Arguments.of(configuration1),
Arguments.of(configuration2),
- Arguments.of(configuration3));
+ Arguments.of(configuration3),
+ Arguments.of(configuration4),
+ Arguments.of(configuration5));
}
PipelineDef buildPipelineDefinitionFromConfiguration(Configuration
pipelineConfig) {
+ return buildPipelineDefinitionFromConfiguration(
+ ValuesDataFactory.IDENTIFIER, new Configuration(),
pipelineConfig);
+ }
+
+ PipelineDef buildPipelineDefinitionFromConfiguration(
+ String sinkType, Configuration pipelineConfig) {
+ return buildPipelineDefinitionFromConfiguration(
+ sinkType, new Configuration(), pipelineConfig);
+ }
+
+ PipelineDef buildPipelineDefinitionFromConfiguration(
+ String sinkType, Configuration sinkConfig, Configuration
pipelineConfig) {
return new PipelineDef(
new SourceDef(ValuesDataFactory.IDENTIFIER, null, new
Configuration()),
- new SinkDef(ValuesDataFactory.IDENTIFIER, null, new
Configuration()),
+ new SinkDef(sinkType, null, sinkConfig),
Collections.emptyList(),
Collections.emptyList(),
Collections.emptyList(),
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToDorisE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToDorisE2eITCase.java
index 64cd0e183..b9b8aaccb 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToDorisE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToDorisE2eITCase.java
@@ -137,35 +137,17 @@ class MySqlToDorisE2eITCase extends
PipelineTestEnvironment {
@Test
void testSyncWholeDatabase() throws Exception {
+ testSyncWholeDatabase("");
+ }
+
+ @Test
+ void testSyncWholeDatabaseWithTableIdPartitioning() throws Exception {
+ testSyncWholeDatabase("\n sink.partitioning.strategy: TABLE_ID");
+ }
+
+ private void testSyncWholeDatabase(String extraPipelineConfig) throws
Exception {
String databaseName = mysqlInventoryDatabase.getDatabaseName();
- String pipelineJob =
- String.format(
- "source:\n"
- + " type: mysql\n"
- + " hostname: mysql\n"
- + " port: 3306\n"
- + " username: %s\n"
- + " password: %s\n"
- + " tables: %s.\\.*\n"
- + " server-id: 5400-5404\n"
- + " server-time-zone: UTC\n"
- + "\n"
- + "sink:\n"
- + " type: doris\n"
- + " fenodes: doris:8030\n"
- + " benodes: doris:8040\n"
- + " username: %s\n"
- + " password: \"%s\"\n"
- + " table.create.properties.replication_num:
1\n"
- + "\n"
- + "pipeline:\n"
- + " parallelism: %d",
- MYSQL_TEST_USER,
- MYSQL_TEST_PASSWORD,
- databaseName,
- DORIS.getUsername(),
- DORIS.getPassword(),
- parallelism);
+ String pipelineJob = buildWholeDatabasePipelineJob(databaseName,
extraPipelineConfig);
Path dorisCdcConnector =
TestUtils.getResource("doris-cdc-pipeline-connector.jar");
submitPipelineJob(pipelineJob, dorisCdcConnector);
waitUntilJobRunning(Duration.ofSeconds(30));
@@ -303,6 +285,37 @@ class MySqlToDorisE2eITCase extends
PipelineTestEnvironment {
}
}
+ private String buildWholeDatabasePipelineJob(String databaseName, String
extraPipelineConfig) {
+ return String.format(
+ "source:\n"
+ + " type: mysql\n"
+ + " hostname: mysql\n"
+ + " port: 3306\n"
+ + " username: %s\n"
+ + " password: %s\n"
+ + " tables: %s.\\.*\n"
+ + " server-id: 5400-5404\n"
+ + " server-time-zone: UTC\n"
+ + "\n"
+ + "sink:\n"
+ + " type: doris\n"
+ + " fenodes: doris:8030\n"
+ + " benodes: doris:8040\n"
+ + " username: %s\n"
+ + " password: \"%s\"\n"
+ + " table.create.properties.replication_num: 1\n"
+ + "\n"
+ + "pipeline:\n"
+ + " parallelism: %d%s",
+ MYSQL_TEST_USER,
+ MYSQL_TEST_PASSWORD,
+ databaseName,
+ DORIS.getUsername(),
+ DORIS.getPassword(),
+ parallelism,
+ extraPipelineConfig);
+ }
+
@Test
void testSyncWholeDatabaseInBatchMode() throws Exception {
String databaseName = mysqlInventoryDatabase.getDatabaseName();
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/partitioning/PrePartitionOperatorTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/partitioning/PrePartitionOperatorTest.java
index 2f0e55abe..4273f97cd 100644
---
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/partitioning/PrePartitionOperatorTest.java
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/partitioning/PrePartitionOperatorTest.java
@@ -25,6 +25,7 @@ import
org.apache.flink.cdc.common.event.SchemaChangeEventType;
import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.schema.Schema;
import
org.apache.flink.cdc.common.sink.DefaultDataChangeEventHashFunctionProvider;
+import org.apache.flink.cdc.common.sink.TableIdHashFunctionProvider;
import org.apache.flink.cdc.common.types.DataTypes;
import org.apache.flink.cdc.common.types.RowType;
import
org.apache.flink.cdc.runtime.testutils.operators.RegularEventOperatorTestHarness;
@@ -146,6 +147,78 @@ class PrePartitionOperatorTest {
% DOWNSTREAM_PARALLELISM;
}
+ @Test
+ void testPartitioningDataChangeEventWithTableIdStrategy() throws Exception
{
+ try (RegularEventOperatorTestHarness<RegularPrePartitionOperator,
PartitioningEvent>
+ testHarness = createTableIdStrategyTestHarness()) {
+ // Initialization
+ testHarness.open();
+ testHarness.registerTableSchema(CUSTOMERS, CUSTOMERS_SCHEMA);
+
+ // DataChangeEvent with different primary key values
+ RegularPrePartitionOperator operator = testHarness.getOperator();
+ BinaryRecordDataGenerator recordDataGenerator =
+ new BinaryRecordDataGenerator(((RowType)
CUSTOMERS_SCHEMA.toRowDataType()));
+ DataChangeEvent eventA =
+ DataChangeEvent.insertEvent(
+ CUSTOMERS,
+ recordDataGenerator.generate(
+ new Object[] {1, new
BinaryStringData("Alice"), 12345678L}));
+ DataChangeEvent eventB =
+ DataChangeEvent.insertEvent(
+ CUSTOMERS,
+ recordDataGenerator.generate(
+ new Object[] {2, new
BinaryStringData("Bob"), 12345689L}));
+ DataChangeEvent eventC =
+ DataChangeEvent.insertEvent(
+ CUSTOMERS,
+ recordDataGenerator.generate(
+ new Object[] {
+ 100, new BinaryStringData("Charlie"),
99999999L
+ }));
+ operator.processElement(new StreamRecord<>(eventA));
+ operator.processElement(new StreamRecord<>(eventB));
+ operator.processElement(new StreamRecord<>(eventC));
+
+ // All events from the same table should be routed to the same
subtask
+ StreamRecord<?> recordA = testHarness.getOutputRecords().poll();
+ StreamRecord<?> recordB = testHarness.getOutputRecords().poll();
+ StreamRecord<?> recordC = testHarness.getOutputRecords().poll();
+
+ int targetA = ((PartitioningEvent)
recordA.getValue()).getTargetPartition();
+ int targetB = ((PartitioningEvent)
recordB.getValue()).getTargetPartition();
+ int targetC = ((PartitioningEvent)
recordC.getValue()).getTargetPartition();
+
+ // Verify all events land on the same subtask
+ assertThat(targetA).isEqualTo(targetB);
+ assertThat(targetB).isEqualTo(targetC);
+
+ // Verify the target is calculated based on TableId only
+ int expectedTarget = getTableIdPartitioningTarget(CUSTOMERS);
+ assertThat(targetA).isEqualTo(expectedTarget);
+ }
+ }
+
+ private int getTableIdPartitioningTarget(TableId tableId) {
+ return new TableIdHashFunctionProvider()
+ .getHashFunction(tableId, CUSTOMERS_SCHEMA)
+ .hashcode(
+ DataChangeEvent.insertEvent(
+ tableId,
+
org.apache.flink.cdc.common.data.GenericRecordData.of()))
+ % DOWNSTREAM_PARALLELISM;
+ }
+
+ private RegularEventOperatorTestHarness<RegularPrePartitionOperator,
PartitioningEvent>
+ createTableIdStrategyTestHarness() {
+ RegularPrePartitionOperator operator =
+ new RegularPrePartitionOperator(
+ TestingSchemaRegistryGateway.SCHEMA_OPERATOR_ID,
+ DOWNSTREAM_PARALLELISM,
+ new TableIdHashFunctionProvider());
+ return RegularEventOperatorTestHarness.with(operator,
DOWNSTREAM_PARALLELISM);
+ }
+
private RegularEventOperatorTestHarness<RegularPrePartitionOperator,
PartitioningEvent>
createTestHarness() {
RegularPrePartitionOperator operator =
