添加目标表不区分大小写的支持，并可开关（默认为区分大小写） (#2329)

* 添加目标表不区分大小写的支持，并可开关（默认为区分大小写）

* 添加目标表不区分大小写的支持，并可开关（默认为区分大小写）-format

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java

@@ -92,19 +92,21 @@ public class MappingConfig implements AdapterConfig {
 
     public static class DbMapping implements AdapterMapping {
 
-        private boolean             mirrorDb    = false;                 // 是否镜像库
-        private String              database;                            // 数据库名或schema名
-        private String              table;                               // 表名
-        private Map<String, String> targetPk    = new LinkedHashMap<>(); // 目标表主键字段
-        private boolean             mapAll      = false;                 // 映射所有字段
-        private String              targetDb;                            // 目标库名
-        private String              targetTable;                         // 目标表名
-        private Map<String, String> targetColumns;                       // 目标表字段映射
+        private boolean             mirrorDb        = false;                 // 是否镜像库
+        private String              database;                                // 数据库名或schema名
+        private String              table;                                   // 表名
+        private Map<String, String> targetPk        = new LinkedHashMap<>(); // 目标表主键字段
+        private boolean             mapAll          = false;                 // 映射所有字段
+        private String              targetDb;                                // 目标库名
+        private String              targetTable;                             // 目标表名
+        private Map<String, String> targetColumns;                           // 目标表字段映射
 
-        private String              etlCondition;                        // etl条件sql
+        private boolean             caseInsensitive = false;                 // 目标表不区分大小写，默认是否
 
-        private int                 readBatch   = 5000;
-        private int                 commitBatch = 5000;                  // etl等批量提交大小
+        private String              etlCondition;                            // etl条件sql
+
+        private int                 readBatch       = 5000;
+        private int                 commitBatch     = 5000;                  // etl等批量提交大小
 
         private Map<String, String> allMapColumns;
 
@@ -179,6 +181,14 @@ public class MappingConfig implements AdapterConfig {
             this.targetColumns = targetColumns;
         }
 
+        public boolean isCaseInsensitive() {
+            return caseInsensitive;
+        }
+
+        public void setCaseInsensitive(boolean caseInsensitive) {
+            this.caseInsensitive = caseInsensitive;
+        }
+
         public String getEtlCondition() {
             return etlCondition;
         }

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java

@@ -91,9 +91,10 @@ public class RdbMirrorDbSyncService {
                 if (config == null) {
                     return false;
                 }
-
+                // 是否区分大小写
+                boolean caseInsensitive = config.getDbMapping().isCaseInsensitive();
                 if (config.getConcurrent()) {
-                    List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml);
+                    List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml, caseInsensitive);
                     singleDmls.forEach(singleDml -> {
                         int hash = rdbSyncService.pkHash(config.getDbMapping(), singleDml.getData());
                         RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml);
@@ -101,7 +102,7 @@ public class RdbMirrorDbSyncService {
                     });
                 } else {
                     int hash = 0;
-                    List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml);
+                    List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml, caseInsensitive);
                     singleDmls.forEach(singleDml -> {
                         RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml);
                         rdbSyncService.getDmlsPartition()[hash].add(syncItem);

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java

@@ -112,9 +112,8 @@ public class RdbSyncService {
 
                     futures.add(executorThreads[i].submit(() -> {
                         try {
-                            dmlsPartition[j].forEach(syncItem -> sync(batchExecutors[j],
-                                syncItem.config,
-                                syncItem.singleDml));
+                            dmlsPartition[j]
+                                .forEach(syncItem -> sync(batchExecutors[j], syncItem.config, syncItem.singleDml));
                             dmlsPartition[j].clear();
                             batchExecutors[j].commit();
                             return true;
@@ -152,49 +151,50 @@ public class RdbSyncService {
         sync(dmls, dml -> {
             if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) {
                 // DDL
-            columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable());
-            return false;
-        } else {
-            // DML
-            String destination = StringUtils.trimToEmpty(dml.getDestination());
-            String groupId = StringUtils.trimToEmpty(dml.getGroupId());
-            String database = dml.getDatabase();
-            String table = dml.getTable();
-            Map<String, MappingConfig> configMap;
-            if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
-                configMap = mappingConfig.get(destination + "-" + groupId + "_" + database + "-" + table);
+                columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable());
+                return false;
             } else {
-                configMap = mappingConfig.get(destination + "_" + database + "-" + table);
-            }
+                // DML
+                String destination = StringUtils.trimToEmpty(dml.getDestination());
+                String groupId = StringUtils.trimToEmpty(dml.getGroupId());
+                String database = dml.getDatabase();
+                String table = dml.getTable();
+                Map<String, MappingConfig> configMap;
+                if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
+                    configMap = mappingConfig.get(destination + "-" + groupId + "_" + database + "-" + table);
+                } else {
+                    configMap = mappingConfig.get(destination + "_" + database + "-" + table);
+                }
 
-            if (configMap == null) {
-                return false;
-            }
+                if (configMap == null) {
+                    return false;
+                }
 
-            if (configMap.values().isEmpty()) {
-                return false;
-            }
+                if (configMap.values().isEmpty()) {
+                    return false;
+                }
 
-            for (MappingConfig config : configMap.values()) {
-                if (config.getConcurrent()) {
-                    List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml);
-                    singleDmls.forEach(singleDml -> {
-                        int hash = pkHash(config.getDbMapping(), singleDml.getData());
-                        SyncItem syncItem = new SyncItem(config, singleDml);
-                        dmlsPartition[hash].add(syncItem);
-                    });
-                } else {
-                    int hash = 0;
-                    List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml);
-                    singleDmls.forEach(singleDml -> {
-                        SyncItem syncItem = new SyncItem(config, singleDml);
-                        dmlsPartition[hash].add(syncItem);
-                    });
+                for (MappingConfig config : configMap.values()) {
+                    boolean caseInsensitive = config.getDbMapping().isCaseInsensitive();
+                    if (config.getConcurrent()) {
+                        List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml, caseInsensitive);
+                        singleDmls.forEach(singleDml -> {
+                            int hash = pkHash(config.getDbMapping(), singleDml.getData());
+                            SyncItem syncItem = new SyncItem(config, singleDml);
+                            dmlsPartition[hash].add(syncItem);
+                        });
+                    } else {
+                        int hash = 0;
+                        List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml, caseInsensitive);
+                        singleDmls.forEach(singleDml -> {
+                            SyncItem syncItem = new SyncItem(config, singleDml);
+                            dmlsPartition[hash].add(syncItem);
+                        });
+                    }
                 }
+                return true;
             }
-            return true;
-        }
-    }   );
+        });
     }
 
     /**

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SingleDml.java

@@ -5,6 +5,7 @@ import java.util.List;
 import java.util.Map;
 
 import com.alibaba.otter.canal.client.adapter.support.Dml;
+import org.springframework.util.LinkedCaseInsensitiveMap;
 
 public class SingleDml {
 
@@ -63,7 +64,7 @@ public class SingleDml {
         this.old = old;
     }
 
-    public static List<SingleDml> dml2SingleDmls(Dml dml) {
+    public static List<SingleDml> dml2SingleDmls(Dml dml, boolean caseInsensitive) {
         List<SingleDml> singleDmls = new ArrayList<>();
         if (dml.getData() != null) {
             int size = dml.getData().size();
@@ -73,9 +74,17 @@ public class SingleDml {
                 singleDml.setDatabase(dml.getDatabase());
                 singleDml.setTable(dml.getTable());
                 singleDml.setType(dml.getType());
-                singleDml.setData(dml.getData().get(i));
+                Map<String, Object> data = dml.getData().get(i);
+                if (caseInsensitive) {
+                    data = toCaseInsensitiveMap(data);
+                }
+                singleDml.setData(data);
                 if (dml.getOld() != null) {
-                    singleDml.setOld(dml.getOld().get(i));
+                    Map<String, Object> oldData = dml.getOld().get(i);
+                    if (caseInsensitive) {
+                        oldData = toCaseInsensitiveMap(oldData);
+                    }
+                    singleDml.setOld(oldData);
                 }
                 singleDmls.add(singleDml);
             }
@@ -89,4 +98,14 @@ public class SingleDml {
         }
         return singleDmls;
     }
+
+    public static List<SingleDml> dml2SingleDmls(Dml dml) {
+        return dml2SingleDmls(dml, false);
+    }
+
+    private static <V> LinkedCaseInsensitiveMap<V> toCaseInsensitiveMap(Map<String, V> data) {
+        LinkedCaseInsensitiveMap map = new LinkedCaseInsensitiveMap();
+        map.putAll(data);
+        return map;
+    }
 }