This is an automated email from the ASF dual-hosted git repository.
yuxiqian 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 67ba6b1c1 [hotfix][transform] Preserve table comments through
transform (#4537)
67ba6b1c1 is described below
commit 67ba6b1c174d1c2a9f736be0ab728b502f2d8741
Author: yuxiqian <[email protected]>
AuthorDate: Thu Sep 17 16:01:22 2026 +0800
[hotfix][transform] Preserve table comments through transform (#4537)
---
.../cdc/pipeline/tests/stage1/MysqlE2eITCase.java | 51 ++++++++++++++++++++++
.../operators/transform/PreTransformOperator.java | 3 +-
.../transform/PreTransformOperatorTest.java | 2 +
3 files changed, 55 insertions(+), 1 deletion(-)
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/MysqlE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/MysqlE2eITCase.java
index b7583c53f..5b42de0dd 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/MysqlE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage1/MysqlE2eITCase.java
@@ -21,6 +21,7 @@ import org.apache.flink.cdc.common.utils.Preconditions;
import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
+import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -160,6 +161,56 @@ class MysqlE2eITCase extends PipelineTestEnvironment {
"DataChangeEvent{tableId=%s.products, before=[111, scooter,
Big 2-wheel scooter , 5.17, null, null, null, 1], after=[], op=DELETE,
meta=()}");
}
+ @Test
+ void testTableCommentWithTransform() throws Exception {
+ String mysqlJdbcUrl =
+ String.format(
+ "jdbc:mysql://%s:%s/%s",
+ MYSQL.getHost(),
+ MYSQL.getDatabasePort(),
+ mysqlInventoryDatabase.getDatabaseName());
+ try (Connection conn =
+ DriverManager.getConnection(
+ mysqlJdbcUrl, MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD);
+ Statement stat = conn.createStatement()) {
+ stat.execute("ALTER TABLE products COMMENT = 'mysql table comment
through transform';");
+ }
+
+ String pipelineJob =
+ String.format(
+ "source:\n"
+ + " type: mysql\n"
+ + " hostname: %s\n"
+ + " port: 3306\n"
+ + " username: %s\n"
+ + " password: %s\n"
+ + " tables: %s.products\n"
+ + " server-id: 5400-5404\n"
+ + " server-time-zone: UTC\n"
+ + " include-comments.enabled: true\n"
+ + "\n"
+ + "transform:\n"
+ + " - source-table: \\.*.\\.*\n"
+ + " projection: \\*\n"
+ + "\n"
+ + "sink:\n"
+ + " type: values",
+ INTER_CONTAINER_MYSQL_ALIAS,
+ MYSQL_TEST_USER,
+ MYSQL_TEST_PASSWORD,
+ mysqlInventoryDatabase.getDatabaseName());
+
+ submitPipelineJob(pipelineJob);
+ waitUntilJobRunning(Duration.ofSeconds(30));
+ waitUntilLogContains(
+ taskManagerConsumer,
+ String.format(
+ "CreateTableEvent{tableId=%s.products",
+ mysqlInventoryDatabase.getDatabaseName()));
+ Assertions.assertThat(taskManagerConsumer.toUtf8String())
+ .contains("comment=mysql table comment through transform");
+ }
+
@Test
void testSyncWholeDatabaseInBatchMode() throws Exception {
String pipelineJob =
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java
index 7c6db1735..b1cd4e220 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java
@@ -362,7 +362,8 @@ public class PreTransformOperator extends
AbstractStreamOperatorAdapter<Event>
private Schema transformSchemaMetaData(
Schema schema, SchemaMetadataTransform schemaMetadataTransform) {
- Schema.Builder schemaBuilder =
Schema.newBuilder().setColumns(schema.getColumns());
+ Schema.Builder schemaBuilder =
+
Schema.newBuilder().setColumns(schema.getColumns()).comment(schema.comment());
if (!schemaMetadataTransform.getPrimaryKeys().isEmpty()) {
schemaBuilder.primaryKey(schemaMetadataTransform.getPrimaryKeys());
} else {
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperatorTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperatorTest.java
index 0425751d7..a1271a216 100644
---
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperatorTest.java
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperatorTest.java
@@ -143,6 +143,7 @@ class PreTransformOperatorTest {
.physicalColumn("age", DataTypes.INT())
.physicalColumn("name", DataTypes.STRING())
.primaryKey("id")
+ .comment("metadata table comment")
.options(ImmutableMap.of("key1", "value1", "key2",
"value2"))
.build();
@@ -152,6 +153,7 @@ class PreTransformOperatorTest {
.physicalColumn("age", DataTypes.INT())
.physicalColumn("name", DataTypes.STRING())
.primaryKey("id")
+ .comment("metadata table comment")
.options(ImmutableMap.of("key1", "value1", "key2",
"value2"))
.build();
