兼容mysql字段映射 modify语句修改替换字段名

dbsyncer-common/pom.xml

@@ -57,6 +57,11 @@
             <artifactId>fastjson2</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>com.github.jsqlparser</groupId>
+            <artifactId>jsqlparser</artifactId>
+        </dependency>
+
     </dependencies>
 
 </project>

dbsyncer-common/src/main/java/org/dbsyncer/common/model/DDLConvertContext.java

@@ -1,7 +1,5 @@
 package org.dbsyncer.common.model;
 
-import java.util.List;
-import java.util.Map;
 import org.dbsyncer.common.spi.ConnectorMapper;
 
 public class DDLConvertContext extends AbstractConvertContext {
@@ -10,6 +8,16 @@ public class DDLConvertContext extends AbstractConvertContext {
 
     String targetSql;
 
+    String tableName;
+
+    String originFiledName;
+
+    String targetFileName;
+
+    String originType;
+
+    String targetType;
+
     public DDLConvertContext(ConnectorMapper sourceConnectorMapper, ConnectorMapper targetConnectorMapper, String sourceTableName, String targetTableName, String event,String sourceSql) {
         super.init(sourceConnectorMapper, targetConnectorMapper, sourceTableName, targetTableName, event, null, null);
         this.sourceSql = sourceSql;
@@ -36,6 +44,8 @@ public class DDLConvertContext extends AbstractConvertContext {
      */
     public void convertSql(){
         //TODO 从源sql转化为目标源sql
+        //获取目标源数据库类型
+
         setTargetSql(sourceSql);
     }
 }

dbsyncer-parser/src/main/java/org/dbsyncer/parser/flush/impl/GeneralBufferActuator.java

@@ -1,5 +1,9 @@
 package org.dbsyncer.parser.flush.impl;
 
+import net.sf.jsqlparser.JSQLParserException;
+import net.sf.jsqlparser.parser.CCJSqlParserUtil;
+import net.sf.jsqlparser.statement.alter.Alter;
+import net.sf.jsqlparser.statement.alter.AlterExpression;
 import org.dbsyncer.cache.CacheService;
 import org.dbsyncer.common.config.GeneralBufferConfig;
 import org.dbsyncer.common.event.RefreshOffsetEvent;
@@ -12,10 +16,12 @@ import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.ConnectorFactory;
 import org.dbsyncer.connector.constant.ConnectorConstant;
+import org.dbsyncer.connector.database.DatabaseConnectorMapper;
 import org.dbsyncer.parser.ParserFactory;
 import org.dbsyncer.parser.flush.AbstractBufferActuator;
 import org.dbsyncer.parser.model.BatchWriter;
 import org.dbsyncer.parser.model.Connector;
+import org.dbsyncer.parser.model.FieldMapping;
 import org.dbsyncer.parser.model.Mapping;
 import org.dbsyncer.parser.model.Picker;
 import org.dbsyncer.parser.model.TableGroup;
@@ -121,7 +127,32 @@ public class GeneralBufferActuator extends AbstractBufferActuator<WriterRequest,
             //转换到目标数据源的sql语句
             String sourceSql = response.getSql();
             DDLConvertContext ddlConvertContext = new DDLConvertContext(sConnectorMapper, tConnectorMapper, sourceTableName, targetTableName, event,sourceSql);
-            ddlConvertContext.convertSql();
+//            ddlConvertContext.convertSql();
+            //先确定转换的sql,源数据库类型的sql-》目标数据库类型的sql
+            //如果都是db，那么db只是对应的字段进行映射，意思是只替换映射字段
+            if (sConnectorMapper instanceof DatabaseConnectorMapper && tConnectorMapper instanceof  DatabaseConnectorMapper){
+                try {
+                    Alter alter = (Alter) CCJSqlParserUtil.parse(sourceSql);
+                    for (AlterExpression alterExpression:alter.getAlterExpressions()) {
+                        for (AlterExpression.ColumnDataType columnDataType:alterExpression.getColDataTypeList()) {
+                            String columName = columnDataType.getColumnName();
+                            columName = columName.replaceAll("`","");
+                            for (FieldMapping fieldMapping :group.getFieldMapping()) {
+                                if (fieldMapping.getSource().getName().equals(columName)){
+                                    columnDataType.setColumnName(fieldMapping.getTarget().getName());
+                                    break;
+                                }
+                            }
+                        }
+                        ddlConvertContext.setTargetSql(alter.toString());
+                    }
+                } catch (JSQLParserException e) {
+//                    throw new RuntimeException(e);
+                    ddlConvertContext.convertSql();
+                }
+            }else{//如果有一方不是db，解析sql，进行填充
+                ddlConvertContext.convertSql();
+            }
             //进行sql执行
             Result result = parserFactory.writeSql(ddlConvertContext, generalExecutor);
             flushStrategy.flushIncrementData(mapping.getMetaId(), result, event);

pom.xml

@@ -54,6 +54,7 @@
         <protobuf.version>3.21.1</protobuf.version>
         <log4j2.version>2.17.1</log4j2.version>
         <junit.version>4.12</junit.version>
+        <jsql-parse.version>4.7</jsql-parse.version>
     </properties>
 
     <!-- 镜像仓库地址 -->
@@ -228,6 +229,12 @@
                 <version>${junit.version}</version>
             </dependency>
 
+            <dependency>
+                <groupId>com.github.jsqlparser</groupId>
+                <artifactId>jsqlparser</artifactId>
+                <version>${jsql-parse.version}</version>
+            </dependency>
+
         </dependencies>
     </dependencyManagement>