Merge pull request #513 from tapdata/TAP-6443-develop

TAP-6443：fix mysql no pk AUTO_INCREMENT filed no sync

Author: jarad0628

connectors-common/sql-core/src/main/java/io/tapdata/common/CommonDbConnector.java

@@ -446,7 +446,18 @@ protected void createIndex(TapConnectorContext connectorContext, TapTable tapTab
                     }
                 }
             });
+            List<String> afterUniqueAutoIncrementSql =getAfterUniqueAutoIncrementFields(tapTable, indexList);
+            if(EmptyKit.isNotEmpty(afterUniqueAutoIncrementSql)){
+                afterUniqueAutoIncrementSql.forEach(sql -> {
+                    try {
+                        jdbcContext.execute(sql);
+                    } catch (SQLException e) {
+                        tapLogger.warn("Failed to update auto-increment column {}, please execute it manually [{}]", e.getMessage(), sql);
+                    }
+                });
+            }
         }
+
     }
 
     protected void createConstraint(TapConnectorContext connectorContext, TapTable tapTable, TapCreateConstraintEvent createConstraintEvent, boolean create) {
@@ -964,4 +975,8 @@ protected long countByAdvanceFilterV2(TapConnectorContext connectorContext, TapT
         });
         return count.get();
     }
+    protected List<String> getAfterUniqueAutoIncrementFields(TapTable tapTable,List<TapIndex> indexList) {
+        return new ArrayList<>();
+    }
+
 }

connectors/mysql-connector/src/main/java/io/tapdata/connector/mysql/MysqlConnector.java

@@ -1030,4 +1030,23 @@ protected String getCreateIndexSql(TapTable tapTable, TapIndex tapIndex) {
         return sb.toString();
     }
 
+    protected List<String> getAfterUniqueAutoIncrementFields(TapTable tapTable,List<TapIndex> indexList) {
+        if(!mysqlConfig.getCreateAutoInc())return new ArrayList<>();
+        String sql = "ALTER TABLE `%s` MODIFY COLUMN `%s` %s AUTO_INCREMENT";
+        List<String> uniqueFields = new ArrayList<>();
+        List<String> uniqueAutoIncrementFields = new ArrayList<>();
+        indexList.forEach(index -> {
+            if(index.isUnique()){
+                uniqueFields.addAll(index.getIndexFields().stream().map(TapIndexField::getName).collect(Collectors.toList()));
+            }
+        });
+        tapTable.getNameFieldMap().values().forEach(f -> {
+            if(f.getAutoInc() && !f.getPrimaryKey() && uniqueFields.contains(f.getName())){
+                uniqueAutoIncrementFields.add(f.getName());
+            }
+        });
+
+        return uniqueAutoIncrementFields.stream().map(f -> String.format(sql, tapTable.getId(), f, tapTable.getNameFieldMap().get(f).getDataType())).collect(Collectors.toList());
+    }
+
 }