fix FLINK-38244. During the full snapshot phase, the field case is adjusted based on isTableIdCaseInsensitive.

Signed-off-by: peiyu <125331682@qq.com>

Author: sd4324530

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

@@ -66,23 +66,26 @@ public EventSourceProvider getEventSourceProvider() {
                         .getBoolean(
                                 RelationalDatabaseConnectorConfig.INCLUDE_SCHEMA_COMMENTS.name(),
                                 false);
-
+        boolean isTableIdCaseInsensitive = MySqlSchemaUtils.isTableIdCaseInsensitive(sourceConfig);
         MySqlEventDeserializer deserializer =
                 new MySqlEventDeserializer(
                         DebeziumChangelogMode.ALL,
                         sourceConfig.isIncludeSchemaChanges(),
                         readableMetadataList,
                         includeComments,
                         sourceConfig.isTreatTinyInt1AsBoolean(),
-                        MySqlSchemaUtils.isTableIdCaseInsensitive(sourceConfig));
+                        isTableIdCaseInsensitive);
 
         MySqlSource<Event> source =
                 new MySqlSource<>(
                         configFactory,
                         deserializer,
                         (sourceReaderMetrics, sourceConfig) ->
                                 new MySqlPipelineRecordEmitter(
-                                        deserializer, sourceReaderMetrics, sourceConfig));
+                                        deserializer,
+                                        sourceReaderMetrics,
+                                        sourceConfig,
+                                        isTableIdCaseInsensitive));
 
         return FlinkSourceProvider.of(source);
     }

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

@@ -55,9 +55,11 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 import static org.apache.flink.cdc.connectors.mysql.debezium.DebeziumUtils.openJdbcConnection;
 import static org.apache.flink.cdc.connectors.mysql.source.utils.RecordUtils.getTableId;
@@ -80,6 +82,7 @@ public class MySqlPipelineRecordEmitter extends MySqlRecordEmitter<Event> {
     // Used when startup mode is snapshot
     private boolean shouldEmitAllCreateTableEventsInSnapshotMode = true;
     private boolean isBounded = false;
+    private final boolean isTableIdCaseInsensitive;
 
     private final DebeziumDeserializationSchema<Event> debeziumDeserializationSchema;
 
@@ -88,7 +91,8 @@ public class MySqlPipelineRecordEmitter extends MySqlRecordEmitter<Event> {
     public MySqlPipelineRecordEmitter(
             DebeziumDeserializationSchema<Event> debeziumDeserializationSchema,
             MySqlSourceReaderMetrics sourceReaderMetrics,
-            MySqlSourceConfig sourceConfig) {
+            MySqlSourceConfig sourceConfig,
+            boolean isTableIdCaseInsensitive) {
         super(
                 debeziumDeserializationSchema,
                 sourceReaderMetrics,
@@ -102,6 +106,7 @@ public MySqlPipelineRecordEmitter(
                 ((DebeziumEventDeserializationSchema) debeziumDeserializationSchema)
                         .getCreateTableEventCache();
         this.isBounded = StartupOptions.snapshot().equals(sourceConfig.getStartupOptions());
+        this.isTableIdCaseInsensitive = isTableIdCaseInsensitive;
     }
 
     @Override
@@ -261,7 +266,10 @@ private Schema buildSchemaFromTable(Table table) {
         for (int i = 0; i < columns.size(); i++) {
             Column column = columns.get(i);
 
-            String colName = column.name();
+            String colName =
+                    this.isTableIdCaseInsensitive
+                            ? column.name().toLowerCase(Locale.ROOT)
+                            : column.name();
             DataType dataType =
                     MySqlTypeUtils.fromDbzColumn(column, sourceConfig.isTreatTinyInt1AsBoolean());
             if (!column.isOptional()) {
@@ -277,6 +285,12 @@ private Schema buildSchemaFromTable(Table table) {
 
         List<String> primaryKey = table.primaryKeyColumnNames();
         if (Objects.nonNull(primaryKey) && !primaryKey.isEmpty()) {
+            if (this.isTableIdCaseInsensitive) {
+                primaryKey =
+                        primaryKey.stream()
+                                .map(key -> key.toLowerCase(Locale.ROOT))
+                                .collect(Collectors.toList());
+            }
             tableBuilder.primaryKey(primaryKey);
         }
         return tableBuilder.build();