项目初始化

datax-common/datax-common-database/pom.xml

@@ -16,6 +16,31 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-jdbc</artifactId>
         </dependency>
+        <dependency>
+            <groupId>mysql</groupId>
+            <artifactId>mysql-connector-java</artifactId>
+            <version>${mysql.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.mariadb.jdbc</groupId>
+            <artifactId>mariadb-java-client</artifactId>
+            <version>${mariadb.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.oracle.ojdbc</groupId>
+            <artifactId>ojdbc8</artifactId>
+            <version>${oracle.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.postgresql</groupId>
+            <artifactId>postgresql</artifactId>
+            <version>${postgresql.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.microsoft.sqlserver</groupId>
+            <artifactId>mssql-jdbc</artifactId>
+            <version>${sqlserver.version}</version>
+        </dependency>
     </dependencies>
 
 </project>
\ No newline at end of file

datax-common/datax-common-database/src/main/java/cn/datax/common/database/constants/DbQueryProperty.java

@@ -3,7 +3,7 @@ package cn.datax.common.database.constants;
 import cn.datax.common.database.exception.DataQueryException;
 import lombok.AllArgsConstructor;
 import lombok.Data;
-import org.apache.commons.lang.StringUtils;
+import org.springframework.util.StringUtils;
 
 import java.io.Serializable;
 
@@ -14,15 +14,19 @@ public class DbQueryProperty implements Serializable {
     private static final long serialVersionUID = 1L;
 
     private String dbType;
-    private String jdbcUrl;
+    private String host;
     private String username;
     private String password;
+    private Integer port;
+    private String dbName;
 
     /**
      * 参数合法性校验
      */
     public void viald() {
-        if (StringUtils.isEmpty(jdbcUrl) || StringUtils.isEmpty(username) || StringUtils.isEmpty(password)) {
+        if (StringUtils.isEmpty(dbType) || StringUtils.isEmpty(host) ||
+                StringUtils.isEmpty(username) || StringUtils.isEmpty(password) ||
+                StringUtils.isEmpty(port) || StringUtils.isEmpty(dbName)) {
             throw new DataQueryException("参数不完整");
         }
         if (DbType.OTHER.getDb().equals(dbType)) {

datax-common/datax-common-database/src/main/java/cn/datax/common/database/constants/DbType.java

@@ -11,35 +11,35 @@ public enum DbType {
     /**
      * MYSQL
      */
-    MYSQL("mysql", "MySql数据库"),
+    MYSQL("mysql", "MySql数据库", "jdbc:mysql://${host}:${port}/${dbName}?serverTimezone=GMT%2B8&characterEncoding=UTF-8&useUnicode=true&useSSL=false"),
     /**
      * MARIADB
      */
-    MARIADB("mariadb", "MariaDB数据库"),
+    MARIADB("mariadb", "MariaDB数据库", "jdbc:mariadb://${host}:${port}/${dbName}"),
     /**
      * ORACLE
      */
-    ORACLE("oracle", "Oracle11g及以下数据库"),
+    ORACLE("oracle", "Oracle11g及以下数据库", "jdbc:oracle:thin:@${host}:${port}:${dbName}"),
     /**
      * oracle12c new pagination
      */
-    ORACLE_12C("oracle12c", "Oracle12c+数据库"),
+    ORACLE_12C("oracle12c", "Oracle12c+数据库", "jdbc:oracle:thin:@${host}:${port}:${dbName}"),
     /**
      * POSTGRE
      */
-    POSTGRE_SQL("postgresql", "Postgre数据库"),
+    POSTGRE_SQL("postgresql", "Postgre数据库", "jdbc:postgresql://${host}:${port}/${dbName}"),
     /**
      * SQLSERVER2005
      */
-    SQL_SERVER2005("sqlserver2005", "SQLServer2005数据库"),
+    SQL_SERVER2008("sqlserver2008", "SQLServer2008及以下数据库", "jdbc:sqlserver://${host}:${port};DatabaseName=${dbName}"),
     /**
      * SQLSERVER
      */
-    SQL_SERVER("sqlserver", "SQLServer2012+数据库"),
+    SQL_SERVER("sqlserver", "SQLServer2012+数据库", "jdbc:sqlserver://${host}:${port};DatabaseName=${dbName}"),
     /**
      * UNKONWN DB
      */
-    OTHER("other", "其他数据库");
+    OTHER("other", "其他数据库", "");
 
     /**
      * 数据库名称
@@ -51,6 +51,11 @@ public enum DbType {
      */
     private final String desc;
 
+    /**
+     * url
+     */
+    private final String url;
+
     public String getDb() {
         return this.db;
     }
@@ -59,9 +64,14 @@ public enum DbType {
         return this.desc;
     }
 
-    DbType(String db, String desc) {
+    public String getUrl() {
+        return this.url;
+    }
+
+    DbType(String db, String desc, String url) {
         this.db = db;
         this.desc = desc;
+        this.url = url;
     }
 
     /**

datax-common/datax-common-database/src/main/java/cn/datax/common/database/core/PageResult.java

@@ -17,9 +17,7 @@ public class PageResult<T> implements Serializable {
     private long total;
     private List<T> data;
 
-    public PageResult(long pageNum, long pageSize, long total, List<T> data) {
-        this.pageNum = pageNum;
-        this.pageSize = pageSize;
+    public PageResult(long total, List<T> data) {
         this.total = total;
         this.data = data;
     }

datax-common/datax-common-database/src/main/java/cn/datax/common/database/datasource/AbstractDataSourceFactory.java

@@ -7,6 +7,7 @@ import cn.datax.common.database.exception.DataQueryException;
 import cn.datax.common.database.query.AbstractDbQuery;
 import com.zaxxer.hikari.HikariDataSource;
 import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.util.StringUtils;
 
 import javax.sql.DataSource;
 
@@ -35,9 +36,23 @@ public abstract class AbstractDataSourceFactory implements DataSourceFactory {
 
     public DataSource createDataSource(DbQueryProperty property) {
         HikariDataSource dataSource = new HikariDataSource();
-        dataSource.setJdbcUrl(property.getJdbcUrl());
+        dataSource.setJdbcUrl(trainToJdbcUrl(property));
         dataSource.setUsername(property.getUsername());
         dataSource.setPassword(property.getPassword());
         return dataSource;
     }
+
+    protected String trainToJdbcUrl(DbQueryProperty property) {
+        String url = DbType.getDbType(property.getDbType()).getUrl();
+        if (StringUtils.isEmpty(url)) {
+            throw new DataQueryException("无效数据库类型!");
+        }
+        url = url.replace("${host}", property.getHost());
+        url = url.replace("${port}", String.valueOf(property.getPort()));
+        if (null == property.getDbName()){
+            property.setDbName("");
+        }
+        url = url.replace("${dbName}", property.getDbName());
+        return url;
+    }
 }

datax-common/datax-common-database/src/main/java/cn/datax/common/database/datasource/CacheDataSourceFactoryBean.java

@@ -17,7 +17,7 @@ public class CacheDataSourceFactoryBean extends AbstractDataSourceFactory {
 
     @Override
     public DataSource createDataSource(DbQueryProperty property) {
-        String key = property.getJdbcUrl() + ":" + property.getUsername();
+        String key = property.getHost() + ":" + property.getPort() + ":" + property.getUsername()+ ":" + property.getDbName();
         String s = compress(key);
         DataSource dataSource = dataSourceMap.get(s);
         if (null == dataSource) {
@@ -45,7 +45,7 @@ public class CacheDataSourceFactoryBean extends AbstractDataSourceFactory {
         md.update(str.getBytes());
         byte b[] = md.digest();
         int i;
-        StringBuffer buf = new StringBuffer("");
+        StringBuffer buf = new StringBuffer();
         for (int offset = 0; offset < b.length; offset++) {
             i = b[offset];
             if (i < 0)

datax-common/datax-common-database/src/main/java/cn/datax/common/database/dialect/DialectRegistry.java

@@ -16,7 +16,7 @@ public class DialectRegistry {
         dialect_enum_map.put(DbType.ORACLE_12C, new Oracle12cDialect());
         dialect_enum_map.put(DbType.ORACLE, new OracleDialect());
         dialect_enum_map.put(DbType.POSTGRE_SQL, new PostgreDialect());
-        dialect_enum_map.put(DbType.SQL_SERVER2005, new SQLServer2005Dialect());
+        dialect_enum_map.put(DbType.SQL_SERVER2008, new SQLServer2008Dialect());
         dialect_enum_map.put(DbType.SQL_SERVER, new SQLServerDialect());
         dialect_enum_map.put(DbType.OTHER, new UnknownDialect());
     }

datax-common/datax-common-database/src/main/java/cn/datax/common/database/dialect/SQLServer2005Dialect.java → datax-common/datax-common-database/src/main/java/cn/datax/common/database/dialect/SQLServer2008Dialect.java

@@ -12,7 +12,7 @@ import java.sql.ResultSet;
  * @author yuwei
  * @since 2020-03-14
  */
-public class SQLServer2005Dialect extends AbstractDbDialect {
+public class SQLServer2008Dialect extends AbstractDbDialect {
 
     @Override
     public String columns(String tableName) {

datax-common/datax-common-database/src/main/java/cn/datax/common/database/dialect/SQLServerDialect.java

@@ -6,7 +6,7 @@ package cn.datax.common.database.dialect;
  * @author yuwei
  * @since 2020-03-14
  */
-public class SQLServerDialect extends SQLServer2005Dialect {
+public class SQLServerDialect extends SQLServer2008Dialect {
 
     @Override
     public String buildPaginationSql(String originalSql, long offset, long count) {

datax-common/datax-common-database/src/main/java/cn/datax/common/database/query/AbstractDbQuery.java

@@ -69,6 +69,6 @@ public abstract class AbstractDbQuery implements DbQuery {
         int total = count(sql);
         String pageSql = dbDialect.buildPaginationSql(sql, offset, size);
         List<Map<String, Object>> records = jdbcTemplate.queryForList(pageSql);
-        return new PageResult<>(offset, size, total, records);
+        return new PageResult<>(total, records);
     }
 }

datax-common/datax-common-database/src/main/java/cn/datax/common/database/query/QueryRegistry.java

@@ -16,7 +16,7 @@ public class QueryRegistry {
         query_enum_map.put(DbType.ORACLE_12C, new Oracle12cQuery());
         query_enum_map.put(DbType.ORACLE, new OracleQuery());
         query_enum_map.put(DbType.POSTGRE_SQL, new PostgreQuery());
-        query_enum_map.put(DbType.SQL_SERVER2005, new SQLServer2005Query());
+        query_enum_map.put(DbType.SQL_SERVER2008, new SQLServer2008Query());
         query_enum_map.put(DbType.SQL_SERVER, new SQLServerQuery());
         query_enum_map.put(DbType.OTHER, new UnknownQuery());
     }

datax-common/datax-common-database/src/main/java/cn/datax/common/database/query/SQLServer2005Query.java → datax-common/datax-common-database/src/main/java/cn/datax/common/database/query/SQLServer2008Query.java

 package cn.datax.common.database.query;
 
-public class SQLServer2005Query extends AbstractDbQuery {
+public class SQLServer2008Query extends AbstractDbQuery {
 }

datax-modules/data-factory-service-parent/data-factory-service-api/src/main/java/cn/datax/service/data/factory/api/dto/DbSchemaDto.java

@@ -6,6 +6,7 @@ import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 
 import javax.validation.constraints.NotBlank;
+import javax.validation.constraints.NotNull;
 import java.io.Serializable;
 
 @ApiModel(value = "数据源连接信息Model")
@@ -14,9 +15,15 @@ public class DbSchemaDto implements Serializable {
 
     private static final long serialVersionUID=1L;
 
-    @ApiModelProperty(value = "JDBC URL")
-    @NotBlank(message = "JDBC URL不能为空", groups = {ValidationGroups.Insert.class, ValidationGroups.Update.class})
-    private String jdbcUrl;
+    @ApiModelProperty(value = "主机")
+    @NotBlank(message = "主机不能为空", groups = {ValidationGroups.Insert.class, ValidationGroups.Update.class})
+    private String host;
+    @ApiModelProperty(value = "端口")
+    @NotNull(message = "端口不能为空", groups = {ValidationGroups.Insert.class, ValidationGroups.Update.class})
+    private Integer port;
+    @ApiModelProperty(value = "数据库")
+    @NotBlank(message = "数据库不能为空", groups = {ValidationGroups.Insert.class, ValidationGroups.Update.class})
+    private String dbName;
     @ApiModelProperty(value = "用户名")
     @NotBlank(message = "用户名不能为空", groups = {ValidationGroups.Insert.class, ValidationGroups.Update.class})
     private String username;

datax-modules/data-factory-service-parent/data-factory-service-api/src/main/java/cn/datax/service/data/factory/api/entity/DbSchema.java

@@ -9,7 +9,9 @@ public class DbSchema implements Serializable {
 
     private static final long serialVersionUID = 1L;
 
-    private String jdbcUrl;
+    private String host;
     private String username;
     private String password;
+    private Integer port;
+    private String dbName;
 }

datax-modules/data-factory-service-parent/data-factory-service-api/src/main/java/cn/datax/service/data/factory/api/query/DataSourceQuery.java

@@ -17,4 +17,6 @@ import lombok.EqualsAndHashCode;
 public class DataSourceQuery extends BaseQueryParams {
 
     private static final long serialVersionUID=1L;
+
+    private String sourceName;
 }

datax-modules/data-factory-service-parent/data-factory-service-api/src/main/java/cn/datax/service/data/factory/api/query/DbDataQuery.java

+package cn.datax.service.data.factory.api.query;
+
+import lombok.Data;
+
+import javax.validation.constraints.NotBlank;
+import java.io.Serializable;
+
+/**
+ * <p>
+ * 数据查询 查询实体
+ * </p>
+ *
+ * @author yuwei
+ * @since 2020-03-14
+ */
+@Data
+public class DbDataQuery implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    @NotBlank(message = "数据源不能为空")
+    private String dataSourceId;
+    @NotBlank(message = "查询sql不能为空")
+    private String sql;
+    // 当前页码
+    private Integer pageNum = 1;
+    // 分页条数
+    private Integer pageSize = 20;
+
+    public Integer getOffset() {
+        pageSize = pageSize == null ? 20 : pageSize;
+        pageNum = pageNum == null ? 1 : pageNum;
+        int offset = pageNum > 0 ? (pageNum - 1) * pageSize : 0;
+        return offset;
+    }
+}

datax-modules/data-factory-service-parent/data-factory-service/src/main/java/cn/datax/service/data/factory/controller/DataSourceController.java

@@ -3,10 +3,13 @@ package cn.datax.service.data.factory.controller;
 import cn.datax.common.core.JsonPage;
 import cn.datax.common.core.R;
 import cn.datax.common.database.DbQuery;
+import cn.datax.common.database.core.DbColumn;
+import cn.datax.common.database.core.PageResult;
 import cn.datax.common.validate.ValidationGroups;
 import cn.datax.service.data.factory.api.dto.DataSourceDto;
 import cn.datax.service.data.factory.api.entity.DataSourceEntity;
 import cn.datax.service.data.factory.api.query.DataSourceQuery;
+import cn.datax.service.data.factory.api.query.DbDataQuery;
 import cn.datax.service.data.factory.api.vo.DataSourceVo;
 import cn.datax.service.data.factory.mapstruct.DataSourceMapper;
 import cn.datax.service.data.factory.service.DataSourceService;
@@ -148,7 +151,62 @@ public class DataSourceController extends BaseController {
             valid = dbQuery.valid();
         } catch (SQLException e) {
             e.printStackTrace();
+            return R.error(e.getMessage());
         }
         return valid ? R.ok() : R.error("数据库连接有误，请检查数据库配置是否正确");
     }
+
+    /**
+     * 数据库表
+     * @param id
+     * @return
+     */
+    @ApiOperation(value = "数据库表", notes = "根据数据源的id来获取指定数据库表")
+    @ApiImplicitParams({
+            @ApiImplicitParam(name = "id", value = "ID", required = true, dataType = "String", paramType = "path"),
+            @ApiImplicitParam(name = "dbName", value = "数据库表名", required = true, dataType = "String", paramType = "path")
+    })
+    @GetMapping("/{id}/{dbName}/tables")
+    public R getDbTables(@PathVariable String id, @PathVariable String dbName) {
+        DbQuery dbQuery = dataSourceService.getDbTables(id);
+        List<String> tables = dbQuery.getTables(dbName);
+        return R.ok().setData(tables);
+    }
+
+    /**
+     * 数据库表结构
+     * @param id
+     * @return
+     */
+    @ApiOperation(value = "数据库表结构", notes = "根据数据源的id来获取指定数据库表的表结构")
+    @ApiImplicitParams({
+            @ApiImplicitParam(name = "id", value = "ID", required = true, dataType = "String", paramType = "path"),
+            @ApiImplicitParam(name = "dbName", value = "数据库名", required = true, dataType = "String", paramType = "path"),
+            @ApiImplicitParam(name = "tableName", value = "数据表名", required = true, dataType = "String", paramType = "path")
+    })
+    @GetMapping("/{id}/{dbName}/{tableName}/columns")
+    public R getDbTableColumns(@PathVariable String id, @PathVariable String dbName, @PathVariable String tableName) {
+        DbQuery dbQuery = dataSourceService.getDbTables(id);
+        List<DbColumn> columns = dbQuery.getTableColumns(dbName, tableName);
+        return R.ok().setData(columns);
+    }
+
+    @ApiOperation(value = "获取SQL结果", notes = "根据数据源的id来获取SQL结果")
+    @ApiImplicitParam(name = "dbDataQuery", value = "详细实体dbDataQuery", required = true, dataType = "DbDataQuery")
+    @GetMapping("/queryList")
+    public R queryList(@RequestBody @Validated DbDataQuery dbDataQuery) {
+        DbQuery dbQuery = dataSourceService.getDbTables(dbDataQuery.getDataSourceId());
+        List<Map<String, Object>> list = dbQuery.queryList(dbDataQuery.getSql());
+        return R.ok().setData(list);
+    }
+
+    @ApiOperation(value = "分页获取SQL结果", notes = "根据数据源的id来分页获取SQL结果")
+    @ApiImplicitParam(name = "dbDataQuery", value = "详细实体dbDataQuery", required = true, dataType = "DbDataQuery")
+    @GetMapping("/queryByPage")
+    public R queryByPage(@RequestBody @Validated DbDataQuery dbDataQuery) {
+        DbQuery dbQuery = dataSourceService.getDbTables(dbDataQuery.getDataSourceId());
+        PageResult<Map<String, Object>> page = dbQuery.queryByPage(dbDataQuery.getSql(), dbDataQuery.getOffset(), dbDataQuery.getPageSize());
+        page.setPageNum(dbDataQuery.getPageNum()).setPageSize(dbDataQuery.getPageSize());
+        return R.ok().setData(page);
+    }
 }

datax-modules/data-factory-service-parent/data-factory-service/src/main/java/cn/datax/service/data/factory/service/DataSourceService.java

@@ -27,4 +27,6 @@ public interface DataSourceService extends BaseService<DataSourceEntity> {
     DbQuery checkConnection(DataSourceDto dataSource);
 
     List<Map<String, Object>> getDbTypes();
+
+    DbQuery getDbTables(String id);
 }

datax-modules/data-factory-service-parent/data-factory-service/src/main/java/cn/datax/service/data/factory/service/impl/DataSourceServiceImpl.java

@@ -67,7 +67,8 @@ public class DataSourceServiceImpl extends BaseServiceImpl<DataSourceDao, DataSo
     public DbQuery checkConnection(DataSourceDto dataSourceDto) {
         DataSourceEntity dataSource = dataSourceMapper.toEntity(dataSourceDto);
         DbSchema dbSchema = dataSource.getDbSchema();
-        DbQueryProperty dbQueryProperty = new DbQueryProperty(dataSource.getDbType(), dbSchema.getJdbcUrl(), dbSchema.getUsername(), dbSchema.getPassword());
+        DbQueryProperty dbQueryProperty = new DbQueryProperty(dataSource.getDbType(), dbSchema.getHost(),
+                dbSchema.getUsername(), dbSchema.getPassword(), dbSchema.getPort(), dbSchema.getDbName());
         DbQuery dbQuery = dataSourceFactory.createDbQuery(dbQueryProperty);
         return dbQuery;
     }
@@ -81,4 +82,14 @@ public class DataSourceServiceImpl extends BaseServiceImpl<DataSourceDao, DataSo
             return map;
         }).collect(Collectors.toList());
     }
+
+    @Override
+    public DbQuery getDbTables(String id) {
+        DataSourceEntity dataSource = getById(id);
+        DbSchema dbSchema = dataSource.getDbSchema();
+        DbQueryProperty dbQueryProperty = new DbQueryProperty(dataSource.getDbType(), dbSchema.getHost(),
+                dbSchema.getUsername(), dbSchema.getPassword(), dbSchema.getPort(), dbSchema.getDbName());
+        DbQuery dbQuery = dataSourceFactory.createDbQuery(dbQueryProperty);
+        return dbQuery;
+    }
 }

pom.xml

@@ -42,6 +42,11 @@
     <aliyun-sdk-oss.version>3.6.0</aliyun-sdk-oss.version>
     <qiniu-java-sdk.version>7.2.25</qiniu-java-sdk.version>
     <fastdfs-client.version>1.26.7</fastdfs-client.version>
+    <mysql.version>8.0.19</mysql.version>
+    <mariadb.version>2.5.4</mariadb.version>
+    <oracle.version>19.3.0.0</oracle.version>
+    <postgresql.version>42.2.11</postgresql.version>
+    <sqlserver.version>8.2.1.jre8</sqlserver.version>
   </properties>
 
   <modules>