[FLINK-36282][pipeline-connector][cdc-connector][mysql]fix incorrect …

…data type of TINYINT(1) in mysql pipeline connector
apache · Sep 14, 2024 · 1dd101d · 1dd101d
1 parent 4b4b8ea
commit 1dd101d
Show file tree

Hide file tree

Showing 15 changed files with 202 additions and 64 deletions.
diff --git a/...tor-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSource.java b/...tor-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSource.java
@@ -45,7 +45,9 @@ public MySqlDataSource(MySqlSourceConfigFactory configFactory) {
     public EventSourceProvider getEventSourceProvider() {
         MySqlEventDeserializer deserializer =
                 new MySqlEventDeserializer(
-                        DebeziumChangelogMode.ALL, sourceConfig.isIncludeSchemaChanges());
+                        DebeziumChangelogMode.ALL,
+                        sourceConfig.isIncludeSchemaChanges(),
+                        sourceConfig.getJdbcProperties());
 
         MySqlSource<Event> source =
                 new MySqlSource<>(

diff --git a/...ql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializer.java b/...ql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializer.java
@@ -44,6 +44,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Properties;
 
 import static org.apache.flink.cdc.connectors.mysql.source.utils.RecordUtils.getHistoryRecord;
 
@@ -59,21 +60,25 @@ public class MySqlEventDeserializer extends DebeziumEventDeserializationSchema {
     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
     private final boolean includeSchemaChanges;
+    private final Properties jdbcProperties;
 
     private transient Tables tables;
     private transient CustomMySqlAntlrDdlParser customParser;
 
     public MySqlEventDeserializer(
-            DebeziumChangelogMode changelogMode, boolean includeSchemaChanges) {
+            DebeziumChangelogMode changelogMode,
+            boolean includeSchemaChanges,
+            Properties jdbcProperties) {
         super(new MySqlSchemaDataTypeInference(), changelogMode);
         this.includeSchemaChanges = includeSchemaChanges;
+        this.jdbcProperties = jdbcProperties;
     }
 
     @Override
     protected List<SchemaChangeEvent> deserializeSchemaChangeRecord(SourceRecord record) {
         if (includeSchemaChanges) {
             if (customParser == null) {
-                customParser = new CustomMySqlAntlrDdlParser();
+                customParser = new CustomMySqlAntlrDdlParser(jdbcProperties);
                 tables = new Tables();
             }
 

diff --git a/...a/org/apache/flink/cdc/connectors/mysql/source/parser/CustomAlterTableParserListener.java b/...a/org/apache/flink/cdc/connectors/mysql/source/parser/CustomAlterTableParserListener.java
@@ -46,6 +46,7 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import java.util.stream.Collectors;
 
 import static org.apache.flink.cdc.connectors.mysql.utils.MySqlTypeUtils.fromDbzColumn;
@@ -60,6 +61,7 @@ public class CustomAlterTableParserListener extends MySqlParserBaseListener {
     private final MySqlAntlrDdlParser parser;
     private final List<ParseTreeListener> listeners;
     private final LinkedList<SchemaChangeEvent> changes;
+    private final Properties jdbcProperties;
     private org.apache.flink.cdc.common.event.TableId currentTable;
     private List<ColumnEditor> columnEditors;
     private CustomColumnDefinitionParserListener columnDefinitionListener;
@@ -70,10 +72,12 @@ public class CustomAlterTableParserListener extends MySqlParserBaseListener {
     public CustomAlterTableParserListener(
             MySqlAntlrDdlParser parser,
             List<ParseTreeListener> listeners,
-            LinkedList<SchemaChangeEvent> changes) {
+            LinkedList<SchemaChangeEvent> changes,
+            Properties jdbcProperties) {
         this.parser = parser;
         this.listeners = listeners;
         this.changes = changes;
+        this.jdbcProperties = jdbcProperties;
     }
 
     @Override
@@ -315,7 +319,7 @@ public void exitAlterByChangeColumn(MySqlParser.AlterByChangeColumnContext ctx)
                     String newColumnName = parser.parseName(ctx.newColumn);
 
                     Map<String, DataType> typeMapping = new HashMap<>();
-                    typeMapping.put(column.name(), fromDbzColumn(column));
+                    typeMapping.put(column.name(), fromDbzColumn(column, jdbcProperties));
                     changes.add(new AlterColumnTypeEvent(currentTable, typeMapping));
 
                     if (newColumnName != null && !column.name().equalsIgnoreCase(newColumnName)) {
@@ -366,7 +370,7 @@ public void exitAlterByModifyColumn(MySqlParser.AlterByModifyColumnContext ctx)
                 () -> {
                     Column column = columnDefinitionListener.getColumn();
                     Map<String, DataType> typeMapping = new HashMap<>();
-                    typeMapping.put(column.name(), fromDbzColumn(column));
+                    typeMapping.put(column.name(), fromDbzColumn(column, jdbcProperties));
                     changes.add(new AlterColumnTypeEvent(currentTable, typeMapping));
                     listeners.remove(columnDefinitionListener);
                 },
@@ -413,7 +417,7 @@ public void exitDropTable(MySqlParser.DropTableContext ctx) {
     private org.apache.flink.cdc.common.schema.Column toCdcColumn(Column dbzColumn) {
         return org.apache.flink.cdc.common.schema.Column.physicalColumn(
                 dbzColumn.name(),
-                fromDbzColumn(dbzColumn),
+                fromDbzColumn(dbzColumn, jdbcProperties),
                 dbzColumn.comment(),
                 dbzColumn.defaultValueExpression().orElse(null));
     }

diff --git a/...n/java/org/apache/flink/cdc/connectors/mysql/source/parser/CustomMySqlAntlrDdlParser.java b/...n/java/org/apache/flink/cdc/connectors/mysql/source/parser/CustomMySqlAntlrDdlParser.java
@@ -29,15 +29,18 @@
 import java.util.Arrays;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Properties;
 
 /** A ddl parser that will use custom listener. */
 public class CustomMySqlAntlrDdlParser extends MySqlAntlrDdlParser {
 
     private final LinkedList<SchemaChangeEvent> parsedEvents;
+    private final Properties jdbcProperties;
 
-    public CustomMySqlAntlrDdlParser() {
+    public CustomMySqlAntlrDdlParser(Properties jdbcProperties) {
         super();
         this.parsedEvents = new LinkedList<>();
+        this.jdbcProperties = jdbcProperties;
     }
 
     // Overriding this method because the BIT type requires default length dimension of 1.
@@ -277,7 +280,7 @@ protected DataTypeResolver initializeDataTypeResolver() {
 
     @Override
     protected AntlrDdlParserListener createParseTreeWalkerListener() {
-        return new CustomMySqlAntlrDdlParserListener(this, parsedEvents);
+        return new CustomMySqlAntlrDdlParserListener(this, parsedEvents, jdbcProperties);
     }
 
     public List<SchemaChangeEvent> getAndClearParsedEvents() {

diff --git a/...rg/apache/flink/cdc/connectors/mysql/source/parser/CustomMySqlAntlrDdlParserListener.java b/...rg/apache/flink/cdc/connectors/mysql/source/parser/CustomMySqlAntlrDdlParserListener.java
@@ -47,6 +47,7 @@
 import java.util.Collection;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Properties;
 import java.util.concurrent.CopyOnWriteArrayList;
 
 /**
@@ -74,12 +75,16 @@ public class CustomMySqlAntlrDdlParserListener extends MySqlParserBaseListener
     private final Collection<ParsingException> errors = new ArrayList<>();
 
     public CustomMySqlAntlrDdlParserListener(
-            MySqlAntlrDdlParser parser, LinkedList<SchemaChangeEvent> parsedEvents) {
+            MySqlAntlrDdlParser parser,
+            LinkedList<SchemaChangeEvent> parsedEvents,
+            Properties jdbcProperties) {
         // initialize listeners
         listeners.add(new CreateAndAlterDatabaseParserListener(parser));
         listeners.add(new DropDatabaseParserListener(parser));
         listeners.add(new CreateTableParserListener(parser, listeners));
-        listeners.add(new CustomAlterTableParserListener(parser, listeners, parsedEvents));
+        listeners.add(
+                new CustomAlterTableParserListener(
+                        parser, listeners, parsedEvents, jdbcProperties));
         listeners.add(new DropTableParserListener(parser));
         listeners.add(new RenameTableParserListener(parser));
         listeners.add(new TruncateTableParserListener(parser));

diff --git a/.../java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java b/.../java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java
@@ -201,7 +201,8 @@ private Schema parseDDL(String ddlStatement, TableId tableId) {
             Column column = columns.get(i);
 
             String colName = column.name();
-            DataType dataType = MySqlTypeUtils.fromDbzColumn(column);
+            DataType dataType =
+                    MySqlTypeUtils.fromDbzColumn(column, sourceConfig.getJdbcProperties());
             if (!column.isOptional()) {
                 dataType = dataType.notNull();
             }

diff --git a/...tor-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/utils/MySqlSchemaUtils.java b/...tor-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/utils/MySqlSchemaUtils.java
@@ -37,6 +37,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Properties;
 import java.util.stream.Collectors;
 
 import static org.apache.flink.cdc.connectors.mysql.debezium.DebeziumUtils.createMySqlConnection;
@@ -129,14 +130,14 @@ public static Schema getTableSchema(
                 new MySqlSchema(sourceConfig, jdbc.isTableIdCaseSensitive())) {
             TableChanges.TableChange tableSchema =
                     mySqlSchema.getTableSchema(partition, jdbc, toDbzTableId(tableId));
-            return toSchema(tableSchema.getTable());
+            return toSchema(tableSchema.getTable(), sourceConfig.getJdbcProperties());
         }
     }
 
-    public static Schema toSchema(Table table) {
+    public static Schema toSchema(Table table, Properties jdbcProperties) {
         List<Column> columns =
                 table.columns().stream()
-                        .map(MySqlSchemaUtils::toColumn)
+                        .map(column -> toColumn(column, jdbcProperties))
                         .collect(Collectors.toList());
 
         return Schema.newBuilder()
@@ -146,9 +147,11 @@ public static Schema toSchema(Table table) {
                 .build();
     }
 
-    public static Column toColumn(io.debezium.relational.Column column) {
+    public static Column toColumn(io.debezium.relational.Column column, Properties jdbcProperties) {
         return Column.physicalColumn(
-                column.name(), MySqlTypeUtils.fromDbzColumn(column), column.comment());
+                column.name(),
+                MySqlTypeUtils.fromDbzColumn(column, jdbcProperties),
+                column.comment());
     }
 
     public static io.debezium.relational.TableId toDbzTableId(TableId tableId) {

diff --git a/...ector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/utils/MySqlTypeUtils.java b/...ector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/utils/MySqlTypeUtils.java
@@ -22,8 +22,11 @@
 import org.apache.flink.cdc.common.types.DataType;
 import org.apache.flink.cdc.common.types.DataTypes;
 
+import com.mysql.cj.conf.PropertyKey;
 import io.debezium.relational.Column;
 
+import java.util.Properties;
+
 /** Utilities for converting from MySQL types to {@link DataType}s. */
 public class MySqlTypeUtils {
 
@@ -109,8 +112,8 @@ public class MySqlTypeUtils {
     private static final String UNKNOWN = "UNKNOWN";
 
     /** Returns a corresponding Flink data type from a debezium {@link Column}. */
-    public static DataType fromDbzColumn(Column column) {
-        DataType dataType = convertFromColumn(column);
+    public static DataType fromDbzColumn(Column column, Properties jdbcProperties) {
+        DataType dataType = convertFromColumn(column, jdbcProperties);
         if (column.isOptional()) {
             return dataType;
         } else {
@@ -122,7 +125,7 @@ public static DataType fromDbzColumn(Column column) {
      * Returns a corresponding Flink data type from a debezium {@link Column} with nullable always
      * be true.
      */
-    private static DataType convertFromColumn(Column column) {
+    private static DataType convertFromColumn(Column column, Properties jdbcProperties) {
         String typeName = column.typeName();
         switch (typeName) {
             case BIT:
@@ -137,7 +140,13 @@ private static DataType convertFromColumn(Column column) {
                 // user should not use tinyint(1) to store number although jdbc url parameter
                 // tinyInt1isBit=false can help change the return value, it's not a general way
                 // btw: mybatis and mysql-connector-java map tinyint(1) to boolean by default
-                return column.length() == 1 ? DataTypes.BOOLEAN() : DataTypes.TINYINT();
+                boolean tinyInt1isBit =
+                        Boolean.parseBoolean(
+                                jdbcProperties.getProperty(
+                                        PropertyKey.tinyInt1isBit.getKeyName(), "true"));
+                return (column.length() == 1 && tinyInt1isBit)
+                        ? DataTypes.BOOLEAN()
+                        : DataTypes.TINYINT();
             case TINYINT_UNSIGNED:
             case TINYINT_UNSIGNED_ZEROFILL:
             case SMALLINT:

diff --git a/...c/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlMetadataAccessorITCase.java b/...c/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlMetadataAccessorITCase.java
@@ -32,6 +32,7 @@
 import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
+import com.mysql.cj.conf.PropertyKey;
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
@@ -41,6 +42,7 @@
 import java.time.ZoneId;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Properties;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -102,13 +104,23 @@ public void testMysql8AccessDatabaseAndTable() {
     }
 
     @Test
-    public void testMysql57AccessCommonTypesSchema() {
-        testAccessCommonTypesSchema(fullTypesMySql57Database);
+    public void testMysql57AccessCommonTypesSchemaTinyintIsBit() {
+        testAccessCommonTypesSchema(fullTypesMySql57Database, true);
     }
 
     @Test
-    public void testMysql8AccessCommonTypesSchema() {
-        testAccessCommonTypesSchema(fullTypesMySql8Database);
+    public void testMysql57AccessCommonTypesSchemaTinyintIsNotBit() {
+        testAccessCommonTypesSchema(fullTypesMySql57Database, false);
+    }
+
+    @Test
+    public void testMysql8AccessCommonTypesSchemaTinyintIsBit() {
+        testAccessCommonTypesSchema(fullTypesMySql8Database, true);
+    }
+
+    @Test
+    public void testMysql8AccessCommonTypesSchemaTinyintIsNotBit() {
+        testAccessCommonTypesSchema(fullTypesMySql8Database, false);
     }
 
     @Test
@@ -117,7 +129,7 @@ public void testMysql57AccessTimeTypesSchema() {
 
         String[] tables = new String[] {"time_types"};
         MySqlMetadataAccessor metadataAccessor =
-                getMetadataAccessor(tables, fullTypesMySql57Database);
+                getMetadataAccessor(tables, fullTypesMySql57Database, true);
 
         Schema actualSchema =
                 metadataAccessor.getTableSchema(
@@ -163,7 +175,7 @@ public void testMysql8AccessTimeTypesSchema() {
 
         String[] tables = new String[] {"time_types"};
         MySqlMetadataAccessor metadataAccessor =
-                getMetadataAccessor(tables, fullTypesMySql8Database);
+                getMetadataAccessor(tables, fullTypesMySql8Database, true);
 
         Schema actualSchema =
                 metadataAccessor.getTableSchema(
@@ -211,7 +223,7 @@ private void testAccessDatabaseAndTable(UniqueDatabase database) {
         database.createAndInitialize();
 
         String[] tables = new String[] {"common_types", "time_types", "precision_types"};
-        MySqlMetadataAccessor metadataAccessor = getMetadataAccessor(tables, database);
+        MySqlMetadataAccessor metadataAccessor = getMetadataAccessor(tables, database, true);
 
         assertThatThrownBy(metadataAccessor::listNamespaces)
                 .isInstanceOf(UnsupportedOperationException.class);
@@ -227,11 +239,12 @@ private void testAccessDatabaseAndTable(UniqueDatabase database) {
         assertThat(actualTables).containsExactlyInAnyOrderElementsOf(expectedTables);
     }
 
-    private void testAccessCommonTypesSchema(UniqueDatabase database) {
+    private void testAccessCommonTypesSchema(UniqueDatabase database, boolean tinyint1IsBit) {
         database.createAndInitialize();
 
         String[] tables = new String[] {"common_types"};
-        MySqlMetadataAccessor metadataAccessor = getMetadataAccessor(tables, database);
+        MySqlMetadataAccessor metadataAccessor =
+                getMetadataAccessor(tables, database, tinyint1IsBit);
 
         Schema actualSchema =
                 metadataAccessor.getTableSchema(
@@ -277,8 +290,12 @@ private void testAccessCommonTypesSchema(UniqueDatabase database) {
                                             DataTypes.STRING(),
                                             DataTypes.BOOLEAN(),
                                             DataTypes.BINARY(1),
-                                            DataTypes.BOOLEAN(),
-                                            DataTypes.BOOLEAN(),
+                                            tinyint1IsBit
+                                                    ? DataTypes.BOOLEAN()
+                                                    : DataTypes.TINYINT(),
+                                            tinyint1IsBit
+                                                    ? DataTypes.BOOLEAN()
+                                                    : DataTypes.TINYINT(),
                                             DataTypes.BINARY(16),
                                             DataTypes.BINARY(8),
                                             DataTypes.STRING(),
@@ -357,17 +374,22 @@ private void testAccessCommonTypesSchema(UniqueDatabase database) {
         assertThat(actualSchema).isEqualTo(expectedSchema);
     }
 
-    private MySqlMetadataAccessor getMetadataAccessor(String[] tables, UniqueDatabase database) {
-        MySqlSourceConfig sourceConfig = getConfig(tables, database);
+    private MySqlMetadataAccessor getMetadataAccessor(
+            String[] tables, UniqueDatabase database, boolean tinyint1IsBit) {
+        MySqlSourceConfig sourceConfig = getConfig(tables, database, tinyint1IsBit);
         return new MySqlMetadataAccessor(sourceConfig);
     }
 
-    private MySqlSourceConfig getConfig(String[] captureTables, UniqueDatabase database) {
+    private MySqlSourceConfig getConfig(
+            String[] captureTables, UniqueDatabase database, boolean tinyint1IsBit) {
         String[] captureTableIds =
                 Arrays.stream(captureTables)
                         .map(tableName -> database.getDatabaseName() + "." + tableName)
                         .toArray(String[]::new);
 
+        Properties jdbcProperties = new Properties();
+        jdbcProperties.put(PropertyKey.tinyInt1isBit.getKeyName(), String.valueOf(tinyint1IsBit));
+
         return new MySqlSourceConfigFactory()
                 .startupOptions(StartupOptions.latest())
                 .databaseList(database.getDatabaseName())
@@ -380,6 +402,7 @@ private MySqlSourceConfig getConfig(String[] captureTables, UniqueDatabase datab
                 .username(database.getUsername())
                 .password(database.getPassword())
                 .serverTimeZone(ZoneId.of("UTC").toString())
+                .jdbcProperties(jdbcProperties)
                 .createConfig(0);
     }
 }