[FLINK-37634][pipeline-connector/mysql]When the table name of mysql supports size sensitive, fix the error that table name not being found. (#3996)

Co-authored-by: linjc13 <linjc13@chinatelecom.cn>

Author: linjianchang

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSource.java

@@ -29,6 +29,7 @@
 import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
 import org.apache.flink.cdc.connectors.mysql.source.reader.MySqlPipelineRecordEmitter;
 import org.apache.flink.cdc.connectors.mysql.table.MySqlReadableMetadata;
+import org.apache.flink.cdc.connectors.mysql.utils.MySqlSchemaUtils;
 import org.apache.flink.cdc.debezium.table.DebeziumChangelogMode;
 
 import io.debezium.relational.RelationalDatabaseConnectorConfig;
@@ -72,7 +73,8 @@ public EventSourceProvider getEventSourceProvider() {
                         sourceConfig.isIncludeSchemaChanges(),
                         readableMetadataList,
                         includeComments,
-                        sourceConfig.isTreatTinyInt1AsBoolean());
+                        sourceConfig.isTreatTinyInt1AsBoolean(),
+                        MySqlSchemaUtils.isTableIdCaseInsensitive(sourceConfig));
 
         MySqlSource<Event> source =
                 new MySqlSource<>(

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializer.java

@@ -69,37 +69,45 @@ public class MySqlEventDeserializer extends DebeziumEventDeserializationSchema {
     private transient CustomMySqlAntlrDdlParser customParser;
 
     private List<MySqlReadableMetadata> readableMetadataList;
+    private boolean isTableIdCaseInsensitive;
 
     public MySqlEventDeserializer(
             DebeziumChangelogMode changelogMode,
             boolean includeSchemaChanges,
-            boolean tinyInt1isBit) {
+            boolean tinyInt1isBit,
+            boolean isTableIdCaseInsensitive) {
         this(
                 changelogMode,
                 includeSchemaChanges,
                 new ArrayList<>(),
                 includeSchemaChanges,
-                tinyInt1isBit);
+                tinyInt1isBit,
+                isTableIdCaseInsensitive);
+        this.isTableIdCaseInsensitive = isTableIdCaseInsensitive;
     }
 
     public MySqlEventDeserializer(
             DebeziumChangelogMode changelogMode,
             boolean includeSchemaChanges,
             List<MySqlReadableMetadata> readableMetadataList,
             boolean includeComments,
-            boolean tinyInt1isBit) {
+            boolean tinyInt1isBit,
+            boolean isTableIdCaseInsensitive) {
         super(new MySqlSchemaDataTypeInference(), changelogMode);
         this.includeSchemaChanges = includeSchemaChanges;
         this.readableMetadataList = readableMetadataList;
         this.includeComments = includeComments;
         this.tinyInt1isBit = tinyInt1isBit;
+        this.isTableIdCaseInsensitive = isTableIdCaseInsensitive;
     }
 
     @Override
     protected List<SchemaChangeEvent> deserializeSchemaChangeRecord(SourceRecord record) {
         if (includeSchemaChanges) {
             if (customParser == null) {
-                customParser = new CustomMySqlAntlrDdlParser(includeComments, tinyInt1isBit);
+                customParser =
+                        new CustomMySqlAntlrDdlParser(
+                                includeComments, tinyInt1isBit, isTableIdCaseInsensitive);
                 tables = new Tables();
             }
 

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/parser/CustomAlterTableParserListener.java

@@ -45,6 +45,7 @@
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.stream.Collectors;
 
@@ -65,18 +66,20 @@ public class CustomAlterTableParserListener extends MySqlParserBaseListener {
     private List<ColumnEditor> columnEditors;
     private CustomColumnDefinitionParserListener columnDefinitionListener;
     private TableEditor tableEditor;
-
+    private boolean isTableIdCaseInsensitive;
     private int parsingColumnIndex = STARTING_INDEX;
 
     public CustomAlterTableParserListener(
             MySqlAntlrDdlParser parser,
             List<ParseTreeListener> listeners,
             LinkedList<SchemaChangeEvent> changes,
-            boolean tinyInt1isBit) {
+            boolean tinyInt1isBit,
+            boolean isTableIdCaseInsensitive) {
         this.parser = parser;
         this.listeners = listeners;
         this.changes = changes;
         this.tinyInt1isBit = tinyInt1isBit;
+        this.isTableIdCaseInsensitive = isTableIdCaseInsensitive;
     }
 
     @Override
@@ -438,7 +441,13 @@ private org.apache.flink.cdc.common.schema.Column toCdcColumn(Column dbzColumn)
     }
 
     private org.apache.flink.cdc.common.event.TableId toCdcTableId(TableId dbzTableId) {
-        return org.apache.flink.cdc.common.event.TableId.tableId(
-                dbzTableId.catalog(), dbzTableId.table());
+        if (isTableIdCaseInsensitive) {
+            return org.apache.flink.cdc.common.event.TableId.tableId(
+                    dbzTableId.catalog().toLowerCase(Locale.ROOT),
+                    dbzTableId.table().toLowerCase(Locale.ROOT));
+        } else {
+            return org.apache.flink.cdc.common.event.TableId.tableId(
+                    dbzTableId.catalog(), dbzTableId.table());
+        }
     }
 }

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/parser/CustomMySqlAntlrDdlParser.java

@@ -36,11 +36,14 @@ public class CustomMySqlAntlrDdlParser extends MySqlAntlrDdlParser {
 
     private final LinkedList<SchemaChangeEvent> parsedEvents;
     private final boolean tinyInt1isBit;
+    private boolean isTableIdCaseInsensitive;
 
-    public CustomMySqlAntlrDdlParser(boolean includeComments, boolean tinyInt1isBit) {
+    public CustomMySqlAntlrDdlParser(
+            boolean includeComments, boolean tinyInt1isBit, boolean isTableIdCaseInsensitive) {
         super(true, false, includeComments, null, Tables.TableFilter.includeAll());
         this.parsedEvents = new LinkedList<>();
         this.tinyInt1isBit = tinyInt1isBit;
+        this.isTableIdCaseInsensitive = isTableIdCaseInsensitive;
     }
 
     // Overriding this method because the BIT type requires default length dimension of 1.
@@ -280,7 +283,8 @@ protected DataTypeResolver initializeDataTypeResolver() {
 
     @Override
     protected AntlrDdlParserListener createParseTreeWalkerListener() {
-        return new CustomMySqlAntlrDdlParserListener(this, parsedEvents, tinyInt1isBit);
+        return new CustomMySqlAntlrDdlParserListener(
+                this, parsedEvents, tinyInt1isBit, isTableIdCaseInsensitive);
     }
 
     public List<SchemaChangeEvent> getAndClearParsedEvents() {

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/parser/CustomMySqlAntlrDdlParserListener.java

@@ -76,13 +76,15 @@ public class CustomMySqlAntlrDdlParserListener extends MySqlParserBaseListener
     public CustomMySqlAntlrDdlParserListener(
             MySqlAntlrDdlParser parser,
             LinkedList<SchemaChangeEvent> parsedEvents,
-            boolean tinyInt1isBit) {
+            boolean tinyInt1isBit,
+            boolean isTableIdCaseInsensitive) {
         // 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, tinyInt1isBit));
+                new CustomAlterTableParserListener(
+                        parser, listeners, parsedEvents, tinyInt1isBit, isTableIdCaseInsensitive));
         listeners.add(new DropTableParserListener(parser));
         listeners.add(new RenameTableParserListener(parser));
         listeners.add(new TruncateTableParserListener(parser));

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/utils/MySqlSchemaUtils.java

@@ -158,5 +158,13 @@ public static io.debezium.relational.TableId toDbzTableId(TableId tableId) {
                 tableId.getSchemaName(), null, tableId.getTableName());
     }
 
+    public static boolean isTableIdCaseInsensitive(MySqlSourceConfig sourceConfig) {
+        try (MySqlConnection jdbc = createMySqlConnection(sourceConfig)) {
+            return jdbc.isTableIdCaseSensitive();
+        } catch (Exception e) {
+            throw new RuntimeException("Error to get table id caseSensitive: " + e.getMessage(), e);
+        }
+    }
+
     private MySqlSchemaUtils() {}
 }