(paimon) branch master updated: [cdc] add integer type for mysql cdc (#3506)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new fb43b65b9 [cdc] add integer type for mysql cdc (#3506)
fb43b65b9 is described below

commit fb43b65b9d3801f2299bc7511e1f83f87be2e576
Author: MOBIN <18814118...@163.com>
AuthorDate: Wed Jun 12 10:55:09 2024 +0800

    [cdc] add integer type for mysql cdc (#3506)
---
 .../paimon/flink/action/cdc/mysql/MySqlTypeUtils.java     |  6 ++++++
 .../action/cdc/mysql/MySqlSyncTableActionITCase.java      | 15 +++++++++------
 2 files changed, 15 insertions(+), 6 deletions(-)

diff --git 
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTypeUtils.java
 
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTypeUtils.java
index 5b5efa1cf..2bc916671 100644
--- 
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTypeUtils.java
+++ 
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTypeUtils.java
@@ -72,6 +72,9 @@ public class MySqlTypeUtils {
     private static final String INT = "INT";
     private static final String INT_UNSIGNED = "INT UNSIGNED";
     private static final String INT_UNSIGNED_ZEROFILL = "INT UNSIGNED 
ZEROFILL";
+    private static final String INTEGER = "INTEGER";
+    private static final String INTEGER_UNSIGNED = "INTEGER UNSIGNED";
+    private static final String INTEGER_UNSIGNED_ZEROFILL = "INTEGER UNSIGNED 
ZEROFILL";
     private static final String BIGINT = "BIGINT";
     private static final String SERIAL = "SERIAL";
     private static final String BIGINT_UNSIGNED = "BIGINT UNSIGNED";
@@ -197,11 +200,14 @@ public class MySqlTypeUtils {
             case SMALLINT_UNSIGNED:
             case SMALLINT_UNSIGNED_ZEROFILL:
             case INT:
+            case INTEGER:
             case MEDIUMINT:
             case YEAR:
                 return DataTypes.INT();
             case INT_UNSIGNED:
+            case INTEGER_UNSIGNED:
             case INT_UNSIGNED_ZEROFILL:
+            case INTEGER_UNSIGNED_ZEROFILL:
             case MEDIUMINT_UNSIGNED:
             case MEDIUMINT_UNSIGNED_ZEROFILL:
             case BIGINT:
diff --git 
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
 
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
index fb78d16eb..db8e1744b 100644
--- 
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
+++ 
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
@@ -297,11 +297,11 @@ public class MySqlSyncTableActionITCase extends 
MySqlActionITCaseBase {
                         + "MODIFY COLUMN v1 VARCHAR(20), "
                         // I'd love to change COMMENT to DEFAULT
                         // however debezium parser seems to have a bug here
-                        + "ADD COLUMN (v5 DOUBLE, v6 DECIMAL(5, 3), `$% ^,& 
*(` VARCHAR(10) COMMENT 'Hi, v700 DOUBLE \\', v701 INT a test'), "
+                        + "ADD COLUMN (v5 DOUBLE, v6 DECIMAL(5, 3), `$% ^,& 
*(` VARCHAR(10), v7 INTEGER COMMENT 'Hi, v700 DOUBLE \\', v701 INT a test'), "
                         + "MODIFY v2 BIGINT");
         statement.executeUpdate(
                 "INSERT INTO schema_evolution_multiple VALUES "
-                        + "(2, 'long_string_two', 2000000000000, 'string_2', 
20, 20.5, 20.002, 'test_2')");
+                        + "(2, 'long_string_two', 2000000000000, 'string_2', 
20, 20.5, 20.002, 'test_2', 200)");
         rowType =
                 RowType.of(
                         new DataType[] {
@@ -312,13 +312,16 @@ public class MySqlSyncTableActionITCase extends 
MySqlActionITCaseBase {
                             DataTypes.INT(),
                             DataTypes.DOUBLE(),
                             DataTypes.DECIMAL(5, 3),
-                            DataTypes.VARCHAR(10)
+                            DataTypes.VARCHAR(10),
+                            DataTypes.INT(),
                         },
-                        new String[] {"_id", "v1", "v2", "v3", "v4", "v5", 
"v6", "$% ^,& *("});
+                        new String[] {
+                            "_id", "v1", "v2", "v3", "v4", "v5", "v6", "$% ^,& 
*(", "v7"
+                        });
         expected =
                 Arrays.asList(
-                        "+I[1, one, 10, string_1, NULL, NULL, NULL, NULL]",
-                        "+I[2, long_string_two, 2000000000000, string_2, 20, 
20.5, 20.002, test_2]");
+                        "+I[1, one, 10, string_1, NULL, NULL, NULL, NULL, 
NULL]",
+                        "+I[2, long_string_two, 2000000000000, string_2, 20, 
20.5, 20.002, test_2, 200]");
         waitForResult(expected, table, rowType, primaryKeys);
     }