update: 新增对达梦数据库的支持 #553 #417 #633

datax-admin/pom.xml

@@ -195,6 +195,14 @@
             <version>${mysql-connector.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.dameng</groupId>
+            <artifactId>Dm8JdbcDriver18</artifactId>
+            <version>8.1.1.193</version>
+            <scope>system</scope>
+            <systemPath>${basedir}/src/main/lib/DmJdbcDriver18-8.1.1.193.jar</systemPath>
+        </dependency>
+
         <!-- datax-core -->
         <dependency>
             <groupId>com.wugui</groupId>

datax-admin/src/main/java/com/wugui/datax/admin/tool/datax/DataxJsonHelper.java

@@ -127,9 +127,13 @@ public void initReader(DataXJsonBuildDto dataxJsonDto, JobDatasource readerDatas
         } else if (MONGODB.equals(datasource)) {
             readerPlugin = new MongoDBReader();
             buildReader = buildMongoDBReader();
+        } else if (DM.equals(datasource)) {
+            readerPlugin = new RdbmsReader();
+            buildReader = buildReader();
         }
     }
 
+
     public void initWriter(DataXJsonBuildDto dataxJsonDto, JobDatasource readerDatasource) {
         this.writerDatasource = readerDatasource;
         this.writerTables = dataxJsonDto.getWriterTables();
@@ -165,6 +169,9 @@ public void initWriter(DataXJsonBuildDto dataxJsonDto, JobDatasource readerDatas
         } else if (JdbcConstants.MONGODB.equals(datasource)) {
             writerPlugin = new MongoDBWriter();
             buildWriter = this.buildMongoDBWriter();
+        } else if (DM.equals(datasource)) {
+            writerPlugin = new RdbmsWriter();
+            buildWriter = this.buildWriter();
         }
     }
 

datax-admin/src/main/java/com/wugui/datax/admin/tool/datax/reader/RdbmsReader.java

@@ -0,0 +1,24 @@
+package com.wugui.datax.admin.tool.datax.reader;
+
+import java.util.Collections;
+import java.util.Map;
+
+/**
+ * <p>
+ * rdbms 构建类
+ * </p>
+ *
+ * @author Listenabe
+ * @since 2024/06/02
+ */
+public class RdbmsReader extends BaseReaderPlugin implements DataxReaderInterface{
+    @Override
+    public String getName() {
+        return "rdbmsreader";
+    }
+
+    @Override
+    public Map<String, Object> sample() {
+        return Collections.emptyMap();
+    }
+}

datax-admin/src/main/java/com/wugui/datax/admin/tool/datax/writer/RdbmsWriter.java

@@ -0,0 +1,24 @@
+package com.wugui.datax.admin.tool.datax.writer;
+
+import java.util.Collections;
+import java.util.Map;
+
+/**
+ * <p>
+ * rdbms构建类
+ * </p>
+ *
+ * @author Listenabe
+ * @since 2024/06/02
+ */
+public class RdbmsWriter extends BaseWriterPlugin implements DataxWriterInterface{
+    @Override
+    public String getName() {
+        return "rdbmswriter";
+    }
+
+    @Override
+    public Map<String, Object> sample() {
+        return Collections.emptyMap();
+    }
+}

datax-admin/src/main/java/com/wugui/datax/admin/tool/meta/DatabaseMetaFactory.java

@@ -28,6 +28,8 @@ public static DatabaseInterface getByDbType(String dbType) {
             return ClickHouseDataBaseMeta.getInstance();
         } else if(JdbcConstants.HBASE20XSQL.equals(dbType)) {
             return Hbase20xsqlMeta.getInstance();
+        }else if(JdbcConstants.DM.equals(dbType)) {
+            return DmDatabaseMeta.getInstance();
         } else {
             throw new UnsupportedOperationException("暂不支持的类型：".concat(dbType));
         }

datax-admin/src/main/java/com/wugui/datax/admin/tool/meta/DmDatabaseMeta.java

@@ -0,0 +1,56 @@
+package com.wugui.datax.admin.tool.meta;
+
+import cn.hutool.core.util.StrUtil;
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * <p>
+ * dm数据库meta信息查询
+ * </p>
+ *
+ * @author Listenabe
+ * @since 2024/06/02
+ */
+public class DmDatabaseMeta extends BaseDatabaseMeta implements DatabaseInterface{
+
+    private volatile static DmDatabaseMeta single;
+
+    public static DmDatabaseMeta getInstance() {
+        if (single == null) {
+            synchronized (DmDatabaseMeta.class) {
+                if (single == null) {
+                    single = new DmDatabaseMeta();
+                }
+            }
+        }
+        return single;
+    }
+
+    /**
+     * 根据条件返回表字段注释
+     * @param schemaName schemaName
+     * @param tableName 达梦数据库目前返回的是schema.tableName， 所以需要截取tableName
+     * @param columnName 表字段名
+     * @return
+     */
+    @Override
+    public String getSQLQueryComment(String schemaName, String tableName, String columnName) {
+        tableName = StringUtils.substringAfterLast(tableName, StrUtil.DOT).replace("\"", "");
+        return String.format("SELECT comments FROM USER_COL_COMMENTS where OWNER = '%s' and TABLE_NAME = '%s' and COLUMN_NAME = '%s'", schemaName, tableName, columnName);
+    }
+
+    public static void main(String[] args) {
+        String tableName = "TEST_SYNC.aaaa.TEST_USER";
+        System.out.println(StringUtils.substringAfterLast(tableName, StrUtil.DOT).replace("\"", ""));
+    }
+
+    @Override
+    public String getSQLQueryTables() {
+        return "select CONCAT(OWNER, '.\"', table_name, '\"') from all_tables";
+    }
+
+    @Override
+    public String getSQLQueryTables(String... tableSchema) {
+        return "select CONCAT(OWNER, '.\"', table_name, '\"') from all_tables where OWNER = '%s'";
+    }
+}

datax-admin/src/main/java/com/wugui/datax/admin/tool/query/DmQueryTool.java

@@ -0,0 +1,23 @@
+package com.wugui.datax.admin.tool.query;
+
+
+import com.wugui.datax.admin.entity.JobDatasource;
+
+import java.sql.SQLException;
+
+
+/**
+ * <p>
+ * 达梦数据库使用的查询工具
+ * </p>
+ *
+ * @author Listenabe
+ * @since 2024/06/02
+ */
+public class DmQueryTool extends BaseQueryTool implements QueryToolInterface {
+
+
+    DmQueryTool(JobDatasource jobDatasource) throws SQLException {
+        super(jobDatasource);
+    }
+}

datax-admin/src/main/java/com/wugui/datax/admin/tool/query/QueryToolFactory.java

@@ -33,6 +33,8 @@ public static BaseQueryTool getByDbType(JobDatasource jobDatasource) {
             return getClickHouseQueryToolInstance(jobDatasource);
         }else if (JdbcConstants.HBASE20XSQL.equals(datasource)) {
             return getHbase20XsqlQueryToolQueryToolInstance(jobDatasource);
+        } else if (JdbcConstants.DM.equals(datasource)) {
+            return getDmQueryToolInstance(jobDatasource);
         }
         throw new UnsupportedOperationException("找不到该类型: ".concat(datasource));
     }
@@ -73,6 +75,15 @@ private static BaseQueryTool getSqlserverQueryToolInstance(JobDatasource jdbcDat
         }
     }
 
+    private static BaseQueryTool getDmQueryToolInstance(JobDatasource jdbcDatasource) {
+        try {
+            return new DmQueryTool(jdbcDatasource);
+        } catch (SQLException e) {
+            throw RdbmsException.asConnException(JdbcConstants.DM,
+                    e,jdbcDatasource.getJdbcUsername(),jdbcDatasource.getDatasourceName());
+        }
+    }
+
     private static BaseQueryTool getHiveQueryToolInstance(JobDatasource jdbcDatasource) {
         try {
             return new HiveQueryTool(jdbcDatasource);