update read table column meta

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/tablegroup/TableGroupChecker.java

@@ -86,6 +86,11 @@ public class TableGroupChecker extends AbstractChecker {
         Assert.hasText(fieldMappingJson, "TableGroupChecker check params fieldMapping is empty");
         setFieldMapping(tableGroup, fieldMappingJson);
 
+        // 生成command
+        Mapping mapping = manager.getMapping(tableGroup.getMappingId());
+        Assert.notNull(mapping, "mapping can not be null.");
+        setCommand(mapping, tableGroup);
+
         // 修改高级配置：过滤条件/转换配置/插件配置
         this.modifySuperConfigModel(tableGroup, params);
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/config/Field.java

@@ -38,6 +38,13 @@ public class Field {
         this.type = type;
     }
 
+    public Field(String name, String typeName, int type, boolean pk) {
+        this.name = name;
+        this.typeName = typeName;
+        this.type = type;
+        this.pk = pk;
+    }
+
     public String getName() {
         return name;
     }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/database/AbstractDatabaseConnector.java

@@ -2,11 +2,11 @@ package org.dbsyncer.connector.database;
 
 import org.apache.commons.lang.StringUtils;
 import org.dbsyncer.common.util.CollectionUtils;
-import org.dbsyncer.connector.template.CommandTemplate;
 import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.connector.config.*;
 import org.dbsyncer.connector.enums.OperationEnum;
 import org.dbsyncer.connector.enums.SqlBuilderEnum;
+import org.dbsyncer.connector.template.CommandTemplate;
 import org.dbsyncer.connector.util.DatabaseUtil;
 import org.dbsyncer.connector.util.JDBCUtil;
 import org.slf4j.Logger;
@@ -17,10 +17,7 @@ import org.springframework.util.Assert;
 import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.stream.Collectors;
 
 public abstract class AbstractDatabaseConnector implements Database {
@@ -78,7 +75,7 @@ public abstract class AbstractDatabaseConnector implements Database {
             String metaSql = getMetaSql(cfg, tableName);
             metaInfo = DatabaseUtil.getMetaInfo(jdbcTemplate, metaSql);
         } catch (Exception e) {
-            logger.error("getMetaInfo failed", e);
+            logger.error(e.getMessage());
         } finally {
             // 释放连接
             this.close(jdbcTemplate);
@@ -88,12 +85,34 @@ public abstract class AbstractDatabaseConnector implements Database {
 
     @Override
     public Map<String, String> getSourceCommand(CommandTemplate commandTemplate) {
-        return null;
+        // 获取过滤SQL
+        List<Filter> filter = commandTemplate.getFilter();
+        String queryFilterSql = getQueryFilterSql(filter);
+
+        // 获取查询SQL
+        Table table = commandTemplate.getTable();
+        String type = SqlBuilderEnum.QUERY.getName();
+        String querySql = getQuerySql(type, table, queryFilterSql);
+        Map<String, String> map = new HashMap<>();
+        map.put(type, querySql);
+        return map;
     }
 
     @Override
     public Map<String, String> getTargetCommand(CommandTemplate commandTemplate) {
-        return null;
+        // 获取增删改SQL
+        Map<String, String> map = new HashMap<>();
+        Table table = commandTemplate.getTable();
+
+        String insert = SqlBuilderEnum.INSERT.getName();
+        map.put(insert, getQuerySql(insert, table, null));
+
+        String update = SqlBuilderEnum.UPDATE.getName();
+        map.put(update, getQuerySql(update, table, null));
+
+        String delete = SqlBuilderEnum.DELETE.getName();
+        map.put(delete, getQuerySql(delete, table, null));
+        return map;
     }
 
     @Override
@@ -110,40 +129,51 @@ public abstract class AbstractDatabaseConnector implements Database {
         }
     }
 
-    @Override
-    public String getQueryFilterSql(List<Filter> filter) {
-        if (CollectionUtils.isEmpty(filter)) {
-            return "";
-        }
-        // 过滤条件SQL
-        StringBuilder condition = new StringBuilder();
-
-        // 拼接并且SQL
-        String addSql = getFilterSql(OperationEnum.AND.getName(), filter);
-        // 如果Add条件存在
-        if (StringUtils.isNotBlank(addSql)) {
-            condition.append(addSql);
-        }
-
-        // 拼接或者SQL
-        String orSql = getFilterSql(OperationEnum.OR.getName(), filter);
-        // 如果Or条件和Add条件都存在
-        if (StringUtils.isNotBlank(orSql) && StringUtils.isNotBlank(addSql)) {
-            condition.append(" OR ").append(orSql);
-        }
+    /**
+     * 获取DQL表信息
+     *
+     * @param config
+     * @return
+     */
+    protected List<String> getDqlTable(ConnectorConfig config) {
+        MetaInfo metaInfo = getDqlMetaInfo(config);
+        Assert.notNull(metaInfo, "SQL解析异常.");
+        DatabaseConfig cfg = (DatabaseConfig) config;
+        return Arrays.asList(cfg.getSql());
+    }
 
-        // 如果有条件加上 WHERE
-        StringBuilder queryFilterSql = new StringBuilder();
-        if (StringUtils.isNotBlank(condition.toString())) {
-            // WHERE (USER.USERNAME = 'zhangsan' AND USER.AGE='20') OR (USER.TEL='18299996666')
-            queryFilterSql.insert(0, " WHERE ").append(condition);
+    /**
+     * 获取DQl元信息
+     *
+     * @param config
+     * @return
+     */
+    protected MetaInfo getDqlMetaInfo(ConnectorConfig config) {
+        DatabaseConfig cfg = (DatabaseConfig) config;
+        JdbcTemplate jdbcTemplate = null;
+        MetaInfo metaInfo = null;
+        try {
+            jdbcTemplate = getJdbcTemplate(cfg);
+            metaInfo = DatabaseUtil.getMetaInfo(jdbcTemplate, cfg.getSql());
+        } catch (Exception e) {
+            logger.error(e.getMessage());
+        } finally {
+            // 释放连接
+            this.close(jdbcTemplate);
         }
-        return queryFilterSql.toString();
+        return metaInfo;
     }
 
-    @Override
-    public String getJdbcSql(String opertion, DatabaseConfig config, Table table, String queryFilterSQL) {
-        if(null == table){
+    /**
+     * 获取查询SQL
+     *
+     * @param type           {@link SqlBuilderEnum}
+     * @param table
+     * @param queryFilterSQL
+     * @return
+     */
+    private String getQuerySql(String type, Table table, String queryFilterSQL) {
+        if (null == table) {
             logger.error("Table can not be null.");
             throw new ConnectorException("Table can not be null.");
         }
@@ -155,9 +185,9 @@ public abstract class AbstractDatabaseConnector implements Database {
         // 获取主键
         String pk = null;
         // 去掉重复的查询字段
-        List<String> filedNames = new ArrayList<String>();
+        List<String> filedNames = new ArrayList<>();
         for (Field c : column) {
-            if(c.isPk()){
+            if (c.isPk()) {
                 pk = c.getName();
             }
             String name = c.getName();
@@ -166,10 +196,6 @@ public abstract class AbstractDatabaseConnector implements Database {
                 filedNames.add(name);
             }
         }
-        if(StringUtils.isBlank(pk)){
-            logger.error("Table primary key can not be empty.");
-            throw new ConnectorException("Table primary key can not be empty.");
-        }
         if (CollectionUtils.isEmpty(filedNames)) {
             logger.error("The filedNames can not be empty.");
             throw new ConnectorException("The filedNames can not be empty.");
@@ -179,92 +205,43 @@ public abstract class AbstractDatabaseConnector implements Database {
             logger.error("Table name can not be empty.");
             throw new ConnectorException("Table name can not be empty.");
         }
-        return SqlBuilderEnum.getSqlBuilder(opertion).buildSql(config, tableName, pk, filedNames, queryFilterSQL, this);
-    }
-
-    @Override
-    public String getJdbcSqlQuartzRange(String tableName, String quartzFiled, String queryFilterSQL) {
-        quartzFiled = tableName + "." + quartzFiled;
-        StringBuilder f = new StringBuilder();
-        // 如果没有加过滤条件就拼接WHERE语法
-        // TB_USER.LASTDATE > ? AND TB_USER.LASTDATE <= ?
-        f.append(StringUtils.isBlank(queryFilterSQL) ? " WHERE " : " AND ");
-        // LASTDATE > '2017-11-10 11:07:41' AND LASTDATE <= '2017-11-10 11:30:01' ORDER BY LASTDATE
-        f.append(quartzFiled).append(" > ?").append(" AND ").append(quartzFiled).append(" <= ?").append(" ORDER BY ").append(quartzFiled);
-        return f.toString();
-    }
-
-    @Override
-    public String getJdbcSqlQuartzAll(String tableName, String quartzFiled, String queryFilterSQL) {
-        StringBuilder f = new StringBuilder();
-        // 如果没有加过滤条件就拼接WHERE语法
-        f.append(StringUtils.isBlank(queryFilterSQL) ? " WHERE " : " AND ");
-        // TB_USER.LASTDATE <= ?
-        f.append(tableName).append(".").append(quartzFiled).append(" <= ?").append(" ORDER BY ").append(tableName).append(".").append(quartzFiled);
-        // TB_USER.LASTDATE <= '2017-11-10 11:07:41'
-        return f.toString();
-    }
-
-    @Override
-    public String getJdbcSqlQuartzMax(String tableName, String quartzFiled) {
-        StringBuilder f = new StringBuilder();
-        // SELECT MAX(USER.LASTDATE) FROM TB_USER
-        f.append("SELECT MAX(").append(tableName).append(".").append(quartzFiled).append(") FROM ").append(tableName);
-        return f.toString();
-    }
-
-    @Override
-    public void batchRowsSetter(PreparedStatement ps, List<Field> fields, Map<String, Object> row) {
-        if (CollectionUtils.isEmpty(fields)) {
-            logger.error("Rows fields can not be empty.");
-            throw new ConnectorException(String.format("Rows fields can not be empty."));
-        }
-        int fieldSize = fields.size();
-        Field f = null;
-        int type;
-        Object val = null;
-        for (int i = 0; i < fieldSize; i++) {
-            // 取出字段和对应值
-            f = fields.get(i);
-            type = f.getType();
-            val = row.get(f.getName());
-            DatabaseUtil.preparedStatementSetter(ps, i + 1, type, val);
-        }
+        return SqlBuilderEnum.getSqlBuilder(type).buildSql(tableName, pk, filedNames, queryFilterSQL, this);
     }
 
     /**
-     * 获取DQL表信息
+     * 获取查询条件SQL
      *
-     * @param config
+     * @param filter
      * @return
      */
-    protected List<String> getDqlTable(ConnectorConfig config) {
-        MetaInfo metaInfo = getDqlMetaInfo(config);
-        Assert.notNull(metaInfo, "SQL解析异常.");
-        DatabaseConfig cfg = (DatabaseConfig) config;
-        return Arrays.asList(cfg.getSql());
-    }
+    private String getQueryFilterSql(List<Filter> filter) {
+        if (CollectionUtils.isEmpty(filter)) {
+            return "";
+        }
+        // 过滤条件SQL
+        StringBuilder condition = new StringBuilder();
 
-    /**
-     * 获取DQl元信息
-     *
-     * @param config
-     * @return
-     */
-    protected MetaInfo getDqlMetaInfo(ConnectorConfig config) {
-        DatabaseConfig cfg = (DatabaseConfig) config;
-        JdbcTemplate jdbcTemplate = null;
-        MetaInfo metaInfo = null;
-        try {
-            jdbcTemplate = getJdbcTemplate(cfg);
-            metaInfo = DatabaseUtil.getMetaInfo(jdbcTemplate, cfg.getSql());
-        } catch (Exception e) {
-            logger.error(e.getMessage());
-        } finally {
-            // 释放连接
-            this.close(jdbcTemplate);
+        // 拼接并且SQL
+        String addSql = getFilterSql(OperationEnum.AND.getName(), filter);
+        // 如果Add条件存在
+        if (StringUtils.isNotBlank(addSql)) {
+            condition.append(addSql);
         }
-        return metaInfo;
+
+        // 拼接或者SQL
+        String orSql = getFilterSql(OperationEnum.OR.getName(), filter);
+        // 如果Or条件和Add条件都存在
+        if (StringUtils.isNotBlank(orSql) && StringUtils.isNotBlank(addSql)) {
+            condition.append(" OR ").append(orSql);
+        }
+
+        // 如果有条件加上 WHERE
+        StringBuilder sql = new StringBuilder();
+        if (StringUtils.isNotBlank(condition.toString())) {
+            // WHERE (USER.USERNAME = 'zhangsan' AND USER.AGE='20') OR (USER.TEL='18299996666')
+            sql.insert(0, " WHERE ").append(condition);
+        }
+        return sql.toString();
     }
 
     /**
@@ -276,7 +253,7 @@ public abstract class AbstractDatabaseConnector implements Database {
      */
     private String getFilterSql(String queryOperator, List<Filter> filter) {
         List<Filter> list = filter.stream().filter(f -> StringUtils.equals(f.getOperation(), queryOperator)).collect(Collectors.toList());
-        if(CollectionUtils.isEmpty(list)){
+        if (CollectionUtils.isEmpty(list)) {
             return "";
         }
 
@@ -297,4 +274,27 @@ public abstract class AbstractDatabaseConnector implements Database {
         return sql.toString();
     }
 
+    /**
+     * @param ps     参数构造器
+     * @param fields 同步字段，例如[{name=ID, type=4}, {name=NAME, type=12}]
+     * @param row    同步字段对应的值，例如{ID=123, NAME=张三11}
+     */
+    private void batchRowsSetter(PreparedStatement ps, List<Field> fields, Map<String, Object> row) {
+        if (CollectionUtils.isEmpty(fields)) {
+            logger.error("Rows fields can not be empty.");
+            throw new ConnectorException(String.format("Rows fields can not be empty."));
+        }
+        int fieldSize = fields.size();
+        Field f = null;
+        int type;
+        Object val = null;
+        for (int i = 0; i < fieldSize; i++) {
+            // 取出字段和对应值
+            f = fields.get(i);
+            type = f.getType();
+            val = row.get(f.getName());
+            DatabaseUtil.preparedStatementSetter(ps, i + 1, type, val);
+        }
+    }
+
 }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/database/Database.java

@@ -2,41 +2,14 @@ package org.dbsyncer.connector.database;
 
 import org.dbsyncer.connector.Connector;
 import org.dbsyncer.connector.config.DatabaseConfig;
-import org.dbsyncer.connector.config.Field;
-import org.dbsyncer.connector.config.Filter;
-import org.dbsyncer.connector.config.Table;
 import org.springframework.jdbc.core.JdbcTemplate;
 
-import java.sql.PreparedStatement;
-import java.sql.SQLException;
-import java.util.List;
-import java.util.Map;
-
 public interface Database extends Connector {
 
+    String getPageSql(String tableName, String pk, String querySQL);
+
     JdbcTemplate getJdbcTemplate(DatabaseConfig config);
 
     void close(JdbcTemplate jdbcTemplate);
 
-    String getQueryFilterSql(List<Filter> filter);
-
-    String getPageSql(DatabaseConfig config, String tableName, String pk, String querySQL);
-
-    String getJdbcSql(String opertion, DatabaseConfig config, Table table, String queryFilterSQL);
-
-    String getJdbcSqlQuartzRange(String tableName, String quartzFiled, String queryFilterSQL);
-
-    String getJdbcSqlQuartzAll(String tableName, String quartzFiled, String queryFilterSQL);
-
-    String getJdbcSqlQuartzMax(String tableName, String quartzFiled);
-
-    /**
-     * @param ps     参数构造器
-     * @param fields 同步字段，例如[{name=ID, type=4}, {name=NAME, type=12}]
-     * @param row    同步字段对应的值，例如{ID=123, NAME=张三11}
-     * @throws NumberFormatException
-     * @throws SQLException
-     */
-    void batchRowsSetter(PreparedStatement ps, List<Field> fields, Map<String, Object> row) throws Exception;
-
 }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/database/sqlbuilder/SqlBuilder.java

@@ -1,7 +1,5 @@
 package org.dbsyncer.connector.database.sqlbuilder;
 
-import org.dbsyncer.connector.config.DatabaseConfig;
-import org.dbsyncer.connector.config.Table;
 import org.dbsyncer.connector.database.Database;
 
 import java.util.List;
@@ -17,7 +15,6 @@ public interface SqlBuilder {
 
     /**
      * 生成SQL
-     * @param config
      * @param tableName
      * @param pk
      * @param filedNames
@@ -25,6 +22,6 @@ public interface SqlBuilder {
      * @param database
      * @return
      */
-    String buildSql(DatabaseConfig config, String tableName, String pk, List<String> filedNames, String queryFilter, Database database);
+    String buildSql(String tableName, String pk, List<String> filedNames, String queryFilter, Database database);
 
 }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/database/sqlbuilder/SqlBuilderDelete.java

@@ -1,7 +1,10 @@
 package org.dbsyncer.connector.database.sqlbuilder;
 
-import org.dbsyncer.connector.config.DatabaseConfig;
+import org.apache.commons.lang.StringUtils;
+import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.connector.database.Database;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.List;
 
@@ -12,9 +15,14 @@ import java.util.List;
  */
 public class SqlBuilderDelete implements SqlBuilder {
 
+    private final Logger logger = LoggerFactory.getLogger(getClass());
+
     @Override
-    public String buildSql(DatabaseConfig config, String tableName, String pk, List<String> filedNames, String queryFilter,
-                           Database database) {
+    public String buildSql(String tableName, String pk, List<String> filedNames, String queryFilter, Database database) {
+        if(StringUtils.isBlank(pk)){
+            logger.error("Table primary key can not be empty.");
+            throw new ConnectorException("Table primary key can not be empty.");
+        }
         // DELETE FROM USER WHERE USER.ID=?
         return new StringBuilder().append("DELETE FROM ").append(tableName).append(" WHERE ").append(tableName).append(".").append(pk)
                 .append("=?").toString();

dbsyncer-connector/src/main/java/org/dbsyncer/connector/database/sqlbuilder/SqlBuilderInsert.java

@@ -1,7 +1,5 @@
 package org.dbsyncer.connector.database.sqlbuilder;
 
-import org.dbsyncer.connector.config.DatabaseConfig;
-import org.dbsyncer.connector.config.Table;
 import org.dbsyncer.connector.database.Database;
 
 import java.util.List;
@@ -14,8 +12,7 @@ import java.util.List;
 public class SqlBuilderInsert implements SqlBuilder {
 
     @Override
-    public String buildSql(DatabaseConfig config, String tableName, String pk, List<String> filedNames, String queryFilter,
-                           Database database) {
+    public String buildSql(String tableName, String pk, List<String> filedNames, String queryFilter, Database database) {
         StringBuilder sql = new StringBuilder();
         StringBuilder fs = new StringBuilder();
         StringBuilder vs = new StringBuilder();

dbsyncer-connector/src/main/java/org/dbsyncer/connector/database/sqlbuilder/SqlBuilderQuery.java

@@ -15,8 +15,7 @@ import java.util.List;
 public class SqlBuilderQuery implements SqlBuilder {
 
     @Override
-    public String buildSql(DatabaseConfig config, String tableName, String pk, List<String> filedNames, String queryFilter,
-                           Database database) {
+    public String buildSql(String tableName, String pk, List<String> filedNames, String queryFilter, Database database) {
         StringBuilder sql = new StringBuilder();
         int size = filedNames.size();
         int end = size - 1;
@@ -35,7 +34,7 @@ public class SqlBuilderQuery implements SqlBuilder {
             sql.append(queryFilter);
         }
         // 分页语句
-        return database.getPageSql(config, tableName, pk, sql.toString());
+        return database.getPageSql(tableName, pk, sql.toString());
     }
 
 }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/database/sqlbuilder/SqlBuilderUpdate.java

@@ -1,7 +1,10 @@
 package org.dbsyncer.connector.database.sqlbuilder;
 
-import org.dbsyncer.connector.config.DatabaseConfig;
+import org.apache.commons.lang.StringUtils;
+import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.connector.database.Database;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.List;
 
@@ -12,9 +15,14 @@ import java.util.List;
  */
 public class SqlBuilderUpdate implements SqlBuilder {
 
+    private final Logger logger = LoggerFactory.getLogger(getClass());
+
     @Override
-    public String buildSql(DatabaseConfig config, String tableName, String pk, List<String> filedNames, String queryFilter,
-                           Database database) {
+    public String buildSql(String tableName, String pk, List<String> filedNames, String queryFilter, Database database) {
+        if(StringUtils.isBlank(pk)){
+            logger.error("Table primary key can not be empty.");
+            throw new ConnectorException("Table primary key can not be empty.");
+        }
         StringBuilder sql = new StringBuilder();
         int size = filedNames.size();
         int end = size - 1;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/enums/SqlBuilderEnum.java

@@ -28,11 +28,7 @@ public enum SqlBuilderEnum {
     /**
      * 查询SQL生成器
      */
-    QUERY(ConnectorConstant.OPERTION_QUERY, new SqlBuilderQuery()),
-    /**
-     * 查询SQL最新记录点
-     */
-    QUERY_MAX(ConnectorConstant.OPERTION_QUERY_MAX, new SqlBuilderQueryMax());
+    QUERY(ConnectorConstant.OPERTION_QUERY, new SqlBuilderQuery());
 
     // SQL构造器名称
     private String name;
@@ -62,4 +58,4 @@ public enum SqlBuilderEnum {
         return sqlBuilder;
     }
 
-}
+}

dbsyncer-connector/src/main/java/org/dbsyncer/connector/mysql/MysqlConnector.java

@@ -12,7 +12,7 @@ public final class MysqlConnector extends AbstractDatabaseConnector {
     }
 
     @Override
-    public String getPageSql(DatabaseConfig config, String tableName, String pk, String querySQL) {
+    public String getPageSql(String tableName, String pk, String querySQL) {
         // Mysql 分页查询
         return querySQL + DatabaseConstant.MYSQL_PAGE_SQL;
     }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/oracle/OracleConnector.java

@@ -12,7 +12,7 @@ public final class OracleConnector extends AbstractDatabaseConnector {
     }
 
     @Override
-    public String getPageSql(DatabaseConfig config, String tableName, String pk, String querySQL) {
+    public String getPageSql(String tableName, String pk, String querySQL) {
         // Oracle 分页查询
         return DatabaseConstant.ORACLE_PAGE_SQL_START + querySQL + DatabaseConstant.ORACLE_PAGE_SQL_END;
     }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sql/DQLMysqlConnector.java

@@ -17,7 +17,7 @@ public final class DQLMysqlConnector extends AbstractDatabaseConnector {
     }
 
     @Override
-    public String getPageSql(DatabaseConfig config, String tableName, String pk, String querySQL) {
+    public String getPageSql(String tableName, String pk, String querySQL) {
         // Mysql 分页查询
         return querySQL + DatabaseConstant.MYSQL_PAGE_SQL;
     }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sql/DQLOracleConnector.java

@@ -17,7 +17,7 @@ public final class DQLOracleConnector extends AbstractDatabaseConnector {
     }
 
     @Override
-    public String getPageSql(DatabaseConfig config, String tableName, String pk, String querySQL) {
+    public String getPageSql(String tableName, String pk, String querySQL) {
         // Oracle 分页查询
         return DatabaseConstant.ORACLE_PAGE_SQL_START + querySQL + DatabaseConstant.ORACLE_PAGE_SQL_END;
     }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sqlserver/SqlServerConnector.java

@@ -1,17 +1,27 @@
 package org.dbsyncer.connector.sqlserver;
 
+import org.apache.commons.lang.StringUtils;
+import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.connector.config.DatabaseConfig;
 import org.dbsyncer.connector.database.AbstractDatabaseConnector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public final class SqlServerConnector extends AbstractDatabaseConnector implements SqlServer {
 
+    private final Logger logger = LoggerFactory.getLogger(getClass());
+
     @Override
     public String getMetaSql(DatabaseConfig config, String tableName) {
         return new StringBuilder().append("SELECT * FROM ").append(tableName).toString();
     }
 
     @Override
-    public String getPageSql(DatabaseConfig config, String tableName, String pk, String querySQL) {
+    public String getPageSql(String tableName, String pk, String querySQL) {
+        if(StringUtils.isBlank(pk)){
+            logger.error("Table primary key can not be empty.");
+            throw new ConnectorException("Table primary key can not be empty.");
+        }
         // SqlServer 分页查询
         // sql> SELECT * FROM SD_USER
         // sql> SELECT * FROM SD_USER ORDER BY USER.ID OFFSET(3-1) * 1000 ROWS FETCH NEXT 1000 ROWS ONLY

dbsyncer-connector/src/main/java/org/dbsyncer/connector/template/CommandTemplate.java

@@ -6,7 +6,7 @@ import org.dbsyncer.connector.config.Table;
 import java.util.List;
 
 /**
- * 查询模板
+ * 查询同步参数模板
  *
  * @author AE86
  * @version 1.0.0

dbsyncer-connector/src/main/java/org/dbsyncer/connector/util/DatabaseUtil.java

@@ -1,23 +1,24 @@
 package org.dbsyncer.connector.util;
 
+import com.sun.rowset.CachedRowSetImpl;
 import org.apache.commons.dbcp.BasicDataSource;
+import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.connector.config.DatabaseConfig;
 import org.dbsyncer.connector.config.Field;
 import org.dbsyncer.connector.config.MetaInfo;
 import org.springframework.jdbc.core.JdbcTemplate;
-import org.springframework.jdbc.core.RowCountCallbackHandler;
+import org.springframework.jdbc.support.rowset.ResultSetWrappingSqlRowSet;
 import org.springframework.jdbc.support.rowset.SqlRowSet;
 import org.springframework.jdbc.support.rowset.SqlRowSetMetaData;
 
 import javax.sql.DataSource;
-import java.sql.PreparedStatement;
-import java.sql.SQLException;
-import java.sql.Timestamp;
-import java.sql.Types;
+import java.sql.*;
 import java.time.LocalDate;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 public abstract class DatabaseUtil {
 
@@ -52,25 +53,57 @@ public abstract class DatabaseUtil {
      * @param metaSql      查询元数据
      * @return
      */
-    public static MetaInfo getMetaInfo(JdbcTemplate jdbcTemplate, String metaSql) {
-        // 查询表字段信息和总行数
-        RowCountCallbackHandler handler = new RowCountCallbackHandler();
-        jdbcTemplate.query(metaSql, handler);
-        String[] columnNames = handler.getColumnNames();
-        int[] columnTypes = handler.getColumnTypes();
-        if(null == columnNames || null == columnTypes || columnNames.length != columnTypes.length){
-            throw new ConnectorException("无法查询表字段信息.");
+    public static MetaInfo getMetaInfo(JdbcTemplate jdbcTemplate, String metaSql) throws SQLException {
+        SqlRowSet sqlRowSet = jdbcTemplate.queryForRowSet(metaSql);
+        ResultSetWrappingSqlRowSet rowSet = (ResultSetWrappingSqlRowSet) sqlRowSet;
+        CachedRowSetImpl resultSet = (CachedRowSetImpl) rowSet.getResultSet();
+        SqlRowSetMetaData metaData = rowSet.getMetaData();
+
+        // 查询表字段信息
+        int columnCount = metaData.getColumnCount();
+        if (1 > columnCount) {
+            throw new ConnectorException("查询表字段不能为空.");
         }
-        int len = columnNames.length;
-        List<Field> fields = new ArrayList<>(len);
+        Connection connection = jdbcTemplate.getDataSource().getConnection();
+        DatabaseMetaData md = connection.getMetaData();
+
+        List<Field> fields = new ArrayList<>(columnCount);
+        // <表名,[主键, ...]>
+        Map<String, List<String>> tables = new HashMap<>();
         String name = null;
+        String label = null;
         String typeName = null;
-        for (int i = 0; i < len; i++) {
-            name = columnNames[i];
-            typeName = getTypeName(columnTypes[i]);
-            fields.add(new Field(name, typeName, columnTypes[i]));
+        String tableName = null;
+        int columnType;
+        boolean pk;
+        for (int i = 1; i <= columnCount; i++) {
+            tableName = metaData.getTableName(i);
+            if (null == tables.get(tableName)) {
+                tables.putIfAbsent(tableName, findTablePrimaryKeys(md, tableName));
+            }
+            name = metaData.getColumnName(i);
+            label = metaData.getColumnLabel(i);
+            typeName = metaData.getColumnTypeName(i);
+            columnType = metaData.getColumnType(i);
+            pk = isPk(tables, tableName, name);
+            fields.add(new Field(label, typeName, columnType, pk));
+        }
+        return new MetaInfo(fields, resultSet.size());
+    }
+
+    private static boolean isPk(Map<String, List<String>> tables, String tableName, String name) {
+        List<String> pk = tables.get(tableName);
+        return !CollectionUtils.isEmpty(pk) && pk.contains(name);
+    }
+
+    private static List<String> findTablePrimaryKeys(DatabaseMetaData md, String tableName) throws SQLException {
+        //根据表名获得主键结果集
+        ResultSet rs = md.getPrimaryKeys(null, null, tableName);
+        List<String> primaryKeys = new ArrayList<>();
+        while (rs.next()) {
+            primaryKeys.add(rs.getString("COLUMN_NAME"));
         }
-        return new MetaInfo(fields, handler.getRowCount());
+        return primaryKeys;
     }
 
     /**
@@ -255,105 +288,4 @@ public abstract class DatabaseUtil {
         }
     }
 
-    /**
-     * 获取数据库表元数据信息
-     *
-     * @param columnTypes
-     * @return
-     */
-    public static List<String> getColumnTypeNames(int[] columnTypes) {
-        if (columnTypes != null && columnTypes.length > 0) {
-            List<String> columnTypeNames = new ArrayList<String>(columnTypes.length);
-            for (int t : columnTypes) {
-                columnTypeNames.add(getTypeName(t));
-            }
-            return columnTypeNames;
-        }
-        return null;
-    }
-
-    private static String getTypeName(int type) {
-        switch (type) {
-            case Types.BIT:
-                return "BIT";
-            case Types.TINYINT:
-                return "TINYINT";
-            case Types.SMALLINT:
-                return "SMALLINT";
-            case Types.INTEGER:
-                return "INTEGER";
-            case Types.BIGINT:
-                return "BIGINT";
-            case Types.FLOAT:
-                return "FLOAT";
-            case Types.REAL:
-                return "REAL";
-            case Types.DOUBLE:
-                return "DOUBLE";
-            case Types.NUMERIC:
-                return "NUMERIC";
-            case Types.DECIMAL:
-                return "DECIMAL";
-            case Types.CHAR:
-                return "CHAR";
-            case Types.VARCHAR:
-                return "VARCHAR";
-            case Types.LONGVARCHAR:
-                return "LONGVARCHAR";
-            case Types.DATE:
-                return "DATE";
-            case Types.TIME:
-                return "TIME";
-            case Types.TIMESTAMP:
-                return "TIMESTAMP";
-            case Types.BINARY:
-                return "BINARY";
-            case Types.VARBINARY:
-                return "VARBINARY";
-            case Types.LONGVARBINARY:
-                return "LONGVARBINARY";
-            case Types.NULL:
-                return "NULL";
-            case Types.OTHER:
-                return "OTHER";
-            case Types.JAVA_OBJECT:
-                return "JAVA_OBJECT";
-            case Types.DISTINCT:
-                return "DISTINCT";
-            case Types.STRUCT:
-                return "STRUCT";
-            case Types.ARRAY:
-                return "ARRAY";
-            case Types.BLOB:
-                return "BLOB";
-            case Types.CLOB:
-                return "CLOB";
-            case Types.REF:
-                return "REF";
-            case Types.DATALINK:
-                return "DATALINK";
-            case Types.BOOLEAN:
-                return "BOOLEAN";
-            case Types.ROWID:
-                return "ROWID";
-            case Types.NCHAR:
-                return "NCHAR";
-            case Types.NVARCHAR:
-                return "NVARCHAR";
-            case Types.LONGNVARCHAR:
-                return "LONGNVARCHAR";
-            case Types.NCLOB:
-                return "NCLOB";
-            case Types.SQLXML:
-                return "SQLXML";
-            case Types.REF_CURSOR:
-                return "REF_CURSOR";
-            case Types.TIME_WITH_TIMEZONE:
-                return "TIME_WITH_TIMEZONE";
-            case Types.TIMESTAMP_WITH_TIMEZONE:
-                return "TIMESTAMP_WITH_TIMEZONE";
-        }
-        return null;
-    }
-
 }

dbsyncer-parser/src/main/resources/TableGroup.json

@@ -47,8 +47,7 @@
     "QUERY":"SELECT * FROM MY_USER where name = ?",
     "INSERT":"INSERT INTO MY_USER(id,name)values(?,?)",
     "UPDATE":"UPDATE MY_USER set name=? where MY_USER.id=?",
-    "DELETE":"DELETE MY_USER WHERE MY_USER.id=?",
-    "QUERY_MAX":"select xxx"
+    "DELETE":"DELETE MY_USER WHERE MY_USER.id=?"
   },
   "filter": [
     {