You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by di...@apache.org on 2024/03/22 09:32:41 UTC
(doris-flink-connector) branch master updated: [improve](cdc) support changing column names using the 'CHANGE' statement (#349)
This is an automated email from the ASF dual-hosted git repository.
diwu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-flink-connector.git
The following commit(s) were added to refs/heads/master by this push:
new a7ad42c [improve](cdc) support changing column names using the 'CHANGE' statement (#349)
a7ad42c is described below
commit a7ad42c5d27fa94ca03db544832ca7012e1366fd
Author: Petrichor <31...@users.noreply.github.com>
AuthorDate: Fri Mar 22 17:32:37 2024 +0800
[improve](cdc) support changing column names using the 'CHANGE' statement (#349)
---
.../JsonDebeziumSchemaChangeImplV2.java | 20 ++++++++----
.../TestJsonDebeziumSchemaChangeImplV2.java | 36 ++++++++++++++++++++++
2 files changed, 50 insertions(+), 6 deletions(-)
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/serializer/jsondebezium/JsonDebeziumSchemaChangeImplV2.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/serializer/jsondebezium/JsonDebeziumSchemaChangeImplV2.java
index 993b822..7973212 100644
--- a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/serializer/jsondebezium/JsonDebeziumSchemaChangeImplV2.java
+++ b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/serializer/jsondebezium/JsonDebeziumSchemaChangeImplV2.java
@@ -65,7 +65,9 @@ public class JsonDebeziumSchemaChangeImplV2 extends JsonDebeziumSchemaChange {
private static final Logger LOG = LoggerFactory.getLogger(JsonDebeziumSchemaChangeImplV2.class);
private static final Pattern renameDDLPattern =
Pattern.compile(
- "ALTER\\s+TABLE\\s+(\\w+)\\s+RENAME\\s+COLUMN\\s+(\\w+)\\s+TO\\s+(\\w+)",
+ "ALTER\\s+TABLE\\s+\\w+\\s+"
+ + "(RENAME\\s+COLUMN\\s+(\\w+)\\s+TO\\s+(\\w+)|"
+ + "CHANGE\\s+(?:column\\s+)?(\\w+)\\s+(\\w+)\\s+(\\w+))",
Pattern.CASE_INSENSITIVE);
// schemaChange saves table names, field, and field column information
private Map<String, Map<String, FieldSchema>> originFieldSchemaMap = new LinkedHashMap<>();
@@ -196,15 +198,21 @@ public class JsonDebeziumSchemaChangeImplV2 extends JsonDebeziumSchemaChange {
}
Map<String, FieldSchema> fieldSchemaMap = originFieldSchemaMap.get(dorisTable);
+ // remove backtick
+ ddl = ddl.replace("`", "");
// rename ddl
- Matcher renameMatcher = renameDDLPattern.matcher(ddl);
- if (renameMatcher.find()) {
- String oldColumnName = renameMatcher.group(2);
- String newColumnName = renameMatcher.group(3);
+ Matcher renameDdlMatcher = renameDDLPattern.matcher(ddl);
+ if (renameDdlMatcher.find()) {
+ String oldColumnName = renameDdlMatcher.group(2);
+ String newColumnName = renameDdlMatcher.group(3);
+ // Change operation
+ if (oldColumnName == null) {
+ oldColumnName = renameDdlMatcher.group(4);
+ newColumnName = renameDdlMatcher.group(5);
+ }
return SchemaChangeHelper.generateRenameDDLSql(
dorisTable, oldColumnName, newColumnName, fieldSchemaMap);
}
-
// add/drop ddl
Map<String, FieldSchema> updateFiledSchema = new LinkedHashMap<>();
for (JsonNode column : columns) {
diff --git a/flink-doris-connector/src/test/java/org/apache/doris/flink/sink/writer/serializer/jsondebezium/TestJsonDebeziumSchemaChangeImplV2.java b/flink-doris-connector/src/test/java/org/apache/doris/flink/sink/writer/serializer/jsondebezium/TestJsonDebeziumSchemaChangeImplV2.java
index 4801e88..89d1ffc 100644
--- a/flink-doris-connector/src/test/java/org/apache/doris/flink/sink/writer/serializer/jsondebezium/TestJsonDebeziumSchemaChangeImplV2.java
+++ b/flink-doris-connector/src/test/java/org/apache/doris/flink/sink/writer/serializer/jsondebezium/TestJsonDebeziumSchemaChangeImplV2.java
@@ -320,6 +320,42 @@ public class TestJsonDebeziumSchemaChangeImplV2 extends TestJsonDebeziumChangeBa
Assert.assertEquals("ALTER TABLE `test`.`t1` RENAME COLUMN `c3` `c333`", ddlList.get(0));
}
+ @Test
+ public void testExtractDDlListChangeName() throws IOException {
+ String columnInfo =
+ "{\"source\":{\"version\":\"1.9.7.Final\",\"connector\":\"mysql\",\"name\":\"mysql_binlog_source\",\"ts_ms\":1710925209991,\"snapshot\":\"false\",\"db\":\"test\",\"sequence\":null,\"table\":\"t1\",\"server_id\":1,\"gtid\":null,\"file\":\"mysql-bin.000288\",\"pos\":81654,\"row\":0,\"thread\":null,\"query\":null},\"historyRecord\":\"{\\\"source\\\":{\\\"file\\\":\\\"mysql-bin.000288\\\",\\\"pos\\\":81654,\\\"server_id\\\":1},\\\"position\\\":{\\\"transaction_id\\\":null,\\\ [...]
+ Map<String, Map<String, FieldSchema>> originFieldSchemaHashMap = new LinkedHashMap<>();
+ Map<String, FieldSchema> fieldSchemaHashMap = Maps.newHashMap();
+ JsonNode record = objectMapper.readTree(columnInfo);
+ schemaChange.setSourceConnector("mysql");
+
+ fieldSchemaHashMap.put("name", new FieldSchema("name", "VARCHAR(256)", "", ""));
+ fieldSchemaHashMap.put("age", new FieldSchema("age", "INT", "", ""));
+ originFieldSchemaHashMap.put("test.t1", fieldSchemaHashMap);
+ schemaChange.setOriginFieldSchemaMap(originFieldSchemaHashMap);
+ List<String> changeNameList = schemaChange.extractDDLList(record);
+ Assert.assertEquals(
+ "ALTER TABLE `test`.`t1` RENAME COLUMN `age` `age1`", changeNameList.get(0));
+ }
+
+ @Test
+ public void testExtractDDlListChangeNameWithColumn() throws IOException {
+ String columnInfo =
+ "{\"source\":{\"version\":\"1.9.7.Final\",\"connector\":\"mysql\",\"name\":\"mysql_binlog_source\",\"ts_ms\":1711088321412,\"snapshot\":\"false\",\"db\":\"doris_test\",\"sequence\":null,\"table\":\"t1\",\"server_id\":1,\"gtid\":null,\"file\":\"mysql-bin.000292\",\"pos\":55695,\"row\":0,\"thread\":null,\"query\":null},\"historyRecord\":\"{\\\"source\\\":{\\\"file\\\":\\\"mysql-bin.000292\\\",\\\"pos\\\":55695,\\\"server_id\\\":1},\\\"position\\\":{\\\"transaction_id\\\":nu [...]
+ Map<String, Map<String, FieldSchema>> originFieldSchemaHashMap = new LinkedHashMap<>();
+ Map<String, FieldSchema> fieldSchemaHashMap = Maps.newHashMap();
+ JsonNode record = objectMapper.readTree(columnInfo);
+ schemaChange.setSourceConnector("mysql");
+
+ fieldSchemaHashMap.put("id", new FieldSchema("id", "int", "", ""));
+ fieldSchemaHashMap.put("key", new FieldSchema("key", "int", "", ""));
+ originFieldSchemaHashMap.put("test.t1", fieldSchemaHashMap);
+ schemaChange.setOriginFieldSchemaMap(originFieldSchemaHashMap);
+ List<String> changeNameList = schemaChange.extractDDLList(record);
+ Assert.assertEquals(
+ "ALTER TABLE `test`.`t1` RENAME COLUMN `key` `key_word`", changeNameList.get(0));
+ }
+
@Test
public void testGetCdcTableIdentifier() throws Exception {
String insert =
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org