复合主键

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/AbstractDataBaseConfigChecker.java

@@ -2,8 +2,11 @@ package org.dbsyncer.biz.checker.impl.connector;
 
 import org.dbsyncer.biz.checker.ConnectorConfigChecker;
 import org.dbsyncer.connector.config.DatabaseConfig;
+import org.dbsyncer.connector.model.SqlTable;
 import org.springframework.util.Assert;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -19,7 +22,6 @@ public abstract class AbstractDataBaseConfigChecker implements ConnectorConfigCh
         String password = params.get("password");
         String url = params.get("url");
         String driverClassName = params.get("driverClassName");
-        String primaryKey = params.get("primaryKey");
         Assert.hasText(username, "Username is empty.");
         Assert.hasText(password, "Password is empty.");
         Assert.hasText(url, "Url is empty.");
@@ -28,19 +30,17 @@ public abstract class AbstractDataBaseConfigChecker implements ConnectorConfigCh
         connectorConfig.setPassword(password);
         connectorConfig.setUrl(url);
         connectorConfig.setDriverClassName(driverClassName);
-        connectorConfig.setPrimaryKey(primaryKey);
     }
 
     protected void modifyDql(DatabaseConfig connectorConfig, Map<String, String> params) {
         String sql = params.get("sql");
         String table = params.get("table");
-        String primaryKey = params.get("primaryKey");
         Assert.hasText(sql, "Sql is empty.");
         Assert.hasText(table, "Table is empty.");
-        Assert.hasText(primaryKey, "PrimaryKey is empty.");
-        connectorConfig.setSql(sql);
-        connectorConfig.setTable(table);
-        connectorConfig.setPrimaryKey(primaryKey);
+
+        List<SqlTable> list = new ArrayList<>();
+        list.add(new SqlTable(sql, table));
+        connectorConfig.setSqlTables(list);
     }
 
     protected void modifySchema(DatabaseConfig connectorConfig, Map<String, String> params) {

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/ElasticsearchConfigChecker.java

@@ -21,7 +21,6 @@ public class ElasticsearchConfigChecker implements ConnectorConfigChecker<ESConf
         String password = params.get("password");
         String index = params.get("index");
         String url = params.get("url");
-        String primaryKey = params.get("primaryKey");
         Assert.hasText(username, "Username is empty.");
         Assert.hasText(password, "Password is empty.");
         Assert.hasText(index, "Index is empty.");
@@ -31,6 +30,5 @@ public class ElasticsearchConfigChecker implements ConnectorConfigChecker<ESConf
         connectorConfig.setPassword(password);
         connectorConfig.setIndex(index);
         connectorConfig.setUrl(url);
-        connectorConfig.setPrimaryKey(primaryKey);
     }
 }

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

@@ -134,7 +134,7 @@ public class TableGroupChecker extends AbstractChecker {
 
     private Table getTable(String connectorId, String tableName, String primaryKey) {
         MetaInfo metaInfo = manager.getMetaInfo(connectorId, tableName);
-        Assert.notNull(metaInfo, "无法获取连接器表信息.");
+        Assert.notNull(metaInfo, "无法获取连接器表信息:" + tableName);
         // 自定义主键
         if (StringUtil.isNotBlank(primaryKey) && !CollectionUtils.isEmpty(metaInfo.getColumn())) {
             for (Field field : metaInfo.getColumn()) {
@@ -144,7 +144,7 @@ public class TableGroupChecker extends AbstractChecker {
                 }
             }
         }
-        return new Table(tableName, metaInfo.getTableType(), primaryKey, metaInfo.getColumn());
+        return new Table(tableName, metaInfo.getTableType(), primaryKey, metaInfo.getColumn(), metaInfo.getSql());
     }
 
     private void checkRepeatedTable(String mappingId, String sourceTable, String targetTable) {

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

@@ -22,8 +22,4 @@ public abstract class AbstractConnectorConfig {
         return this;
     }
 
-    public String getPrimaryKey() {
-        return "";
-    }
-
 }

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

@@ -1,8 +1,10 @@
 package org.dbsyncer.connector.config;
 
 import org.dbsyncer.common.model.AbstractConnectorConfig;
+import org.dbsyncer.connector.model.SqlTable;
 
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -34,24 +36,14 @@ public class DatabaseConfig extends AbstractConnectorConfig {
     private String password;
 
     /**
-     * 主表
-     */
-    private String table;
-
-    /**
-     * 主键
-     */
-    private String primaryKey;
-
-    /**
-     * SQL
+     * 构架名
      */
-    private String sql;
+    private String schema;
 
     /**
-     * 构架名
+     * sql
      */
-    private String schema;
+    private List<SqlTable> sqlTables;
 
     /**
      * 参数配置
@@ -98,30 +90,6 @@ public class DatabaseConfig extends AbstractConnectorConfig {
         this.password = password;
     }
 
-    public String getTable() {
-        return table;
-    }
-
-    public void setTable(String table) {
-        this.table = table;
-    }
-
-    public String getPrimaryKey() {
-        return primaryKey;
-    }
-
-    public void setPrimaryKey(String primaryKey) {
-        this.primaryKey = primaryKey;
-    }
-
-    public String getSql() {
-        return sql;
-    }
-
-    public void setSql(String sql) {
-        this.sql = sql;
-    }
-
     public String getSchema() {
         return schema;
     }
@@ -130,6 +98,14 @@ public class DatabaseConfig extends AbstractConnectorConfig {
         this.schema = schema;
     }
 
+    public List<SqlTable> getSqlTables() {
+        return sqlTables;
+    }
+
+    public void setSqlTables(List<SqlTable> sqlTables) {
+        this.sqlTables = sqlTables;
+    }
+
     public Map<String, String> getProperties() {
         return properties;
     }

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

@@ -35,11 +35,6 @@ public class ESConfig extends AbstractConnectorConfig {
      */
     private String type = "_doc";
 
-    /**
-     * 主键
-     */
-    private String primaryKey;
-
     public String getUrl() {
         return url;
     }
@@ -80,11 +75,4 @@ public class ESConfig extends AbstractConnectorConfig {
         this.type = type;
     }
 
-    public String getPrimaryKey() {
-        return primaryKey;
-    }
-
-    public void setPrimaryKey(String primaryKey) {
-        this.primaryKey = primaryKey;
-    }
 }

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

@@ -223,7 +223,7 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector implem
 
         // 获取查询数据行是否存在
         String tableName = commandConfig.getTable().getName();
-        String pk = PrimaryKeyUtil.findOriginalTablePrimaryKey(commandConfig.getTable());
+        String pk = PrimaryKeyUtil.findOriginalTablePrimaryKey(commandConfig.getOriginalTable());
         StringBuilder queryCount = new StringBuilder("SELECT COUNT(1) FROM ").append(schema).append(quotation).append(tableName).append(quotation)
                 .append(" WHERE ").append(quotation).append(pk).append(quotation).append(" = ?");
         String queryCountExist = ConnectorConstant.OPERTION_QUERY_COUNT_EXIST;
@@ -278,33 +278,6 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector implem
         return config.getSchema();
     }
 
-    /**
-     * 获取表列表
-     *
-     * @param connectorMapper
-     * @param catalog
-     * @param schema
-     * @param tableNamePattern
-     * @return
-     */
-    protected List<Table> getTable(DatabaseConnectorMapper connectorMapper, String catalog, String schema, String tableNamePattern) {
-        return connectorMapper.execute(databaseTemplate -> {
-            List<Table> tables = new ArrayList<>();
-            SimpleConnection connection = (SimpleConnection) databaseTemplate.getConnection();
-            Connection conn = connection.getConnection();
-            String databaseCatalog = null == catalog ? conn.getCatalog() : catalog;
-            String schemaNamePattern = null == schema ? conn.getSchema() : schema;
-            String[] types = {TableTypeEnum.TABLE.getCode(), TableTypeEnum.VIEW.getCode(), TableTypeEnum.MATERIALIZED_VIEW.getCode()};
-            final ResultSet rs = conn.getMetaData().getTables(databaseCatalog, schemaNamePattern, tableNamePattern, types);
-            while (rs.next()) {
-                final String tableName = rs.getString("TABLE_NAME");
-                final String tableType = rs.getString("TABLE_TYPE");
-                tables.add(new Table(tableName, tableType));
-            }
-            return tables;
-        });
-    }
-
     /**
      * 获取数据库表元数据信息
      *
@@ -366,7 +339,7 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector implem
      */
     protected String getQueryCountSql(CommandConfig commandConfig, String schema, String quotation, String queryFilterSql) {
         String table = commandConfig.getTable().getName();
-        String pk = PrimaryKeyUtil.findOriginalTablePrimaryKey(commandConfig.getTable());
+        String pk = PrimaryKeyUtil.findOriginalTablePrimaryKey(commandConfig.getOriginalTable());
         StringBuilder queryCount = new StringBuilder();
         queryCount.append("SELECT COUNT(1) FROM (SELECT 1 FROM ").append(schema).append(quotation).append(table).append(quotation);
         if (StringUtil.isNotBlank(queryFilterSql)) {
@@ -412,6 +385,33 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector implem
         return sql.toString();
     }
 
+    /**
+     * 获取表列表
+     *
+     * @param connectorMapper
+     * @param catalog
+     * @param schema
+     * @param tableNamePattern
+     * @return
+     */
+    private List<Table> getTable(DatabaseConnectorMapper connectorMapper, String catalog, String schema, String tableNamePattern) {
+        return connectorMapper.execute(databaseTemplate -> {
+            List<Table> tables = new ArrayList<>();
+            SimpleConnection connection = (SimpleConnection) databaseTemplate.getConnection();
+            Connection conn = connection.getConnection();
+            String databaseCatalog = null == catalog ? conn.getCatalog() : catalog;
+            String schemaNamePattern = null == schema ? conn.getSchema() : schema;
+            String[] types = {TableTypeEnum.TABLE.getCode(), TableTypeEnum.VIEW.getCode(), TableTypeEnum.MATERIALIZED_VIEW.getCode()};
+            final ResultSet rs = conn.getMetaData().getTables(databaseCatalog, schemaNamePattern, tableNamePattern, types);
+            while (rs.next()) {
+                final String tableName = rs.getString("TABLE_NAME");
+                final String tableType = rs.getString("TABLE_TYPE");
+                tables.add(new Table(tableName, tableType));
+            }
+            return tables;
+        });
+    }
+
     /**
      * 根据过滤条件获取查询SQL
      *
@@ -466,7 +466,6 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector implem
         if (CollectionUtils.isEmpty(column)) {
             return null;
         }
-        String pk = null;
         Set<String> mark = new HashSet<>();
         List<Field> fields = new ArrayList<>();
         for (Field c : column) {
@@ -474,9 +473,6 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector implem
             if (StringUtil.isBlank(name)) {
                 throw new ConnectorException("The field name can not be empty.");
             }
-            if (c.isPk()) {
-                pk = name;
-            }
             if (!mark.contains(name)) {
                 fields.add(c);
                 mark.add(name);
@@ -491,9 +487,7 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector implem
             logger.error("Table name can not be empty.");
             throw new ConnectorException("Table name can not be empty.");
         }
-        if (StringUtil.isBlank(pk)) {
-            pk = PrimaryKeyUtil.findOriginalTablePrimaryKey(table);
-        }
+        String pk = PrimaryKeyUtil.findOriginalTablePrimaryKey(commandConfig.getOriginalTable());
 
         SqlBuilderConfig config = new SqlBuilderConfig(this, schema, tableName, pk, fields, queryFilterSQL, buildSqlWithQuotation());
         return SqlBuilderEnum.getSqlBuilder(type).buildSql(config);

dbsyncer-connector/src/main/java/org/dbsyncer/connector/es/ESConnector.java

@@ -126,7 +126,7 @@ public final class ESConnector extends AbstractConnector implements Connector<ES
             }
             properties.forEach((k, v) -> {
                 String columnType = (String) v.get("type");
-                fields.add(new Field(k, columnType, ESFieldTypeEnum.getType(columnType), StringUtil.equals(config.getPrimaryKey(), k)));
+                fields.add(new Field(k, columnType, ESFieldTypeEnum.getType(columnType)));
             });
         } catch (IOException e) {
             logger.error(e.getMessage());

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

@@ -44,11 +44,6 @@ public class Field {
     public Field() {
     }
 
-    public Field(String name, String labelName) {
-        this.name = name;
-        this.labelName = labelName;
-    }
-
     public Field(String name, String typeName, int type) {
         this.name = name;
         this.typeName = typeName;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/model/MetaInfo.java

@@ -16,12 +16,18 @@ public class MetaInfo {
      * 表类型
      */
     private String tableType;
+
     /**
      * 属性字段
      * 格式：[{"name":"ID","typeName":"INT","type":"4"},{"name":"NAME","typeName":"VARCHAR","type":"12"}]
      */
     private List<Field> column;
 
+    /**
+     * sql
+     */
+    private String sql;
+
     public String getTableType() {
         return tableType;
     }
@@ -39,6 +45,16 @@ public class MetaInfo {
         this.column = column;
         return this;
     }
+
+    public String getSql() {
+        return sql;
+    }
+
+    public MetaInfo setSql(String sql) {
+        this.sql = sql;
+        return this;
+    }
+
     @Override
     public String toString() {
         return new StringBuilder().append("MetaInfo{").append("tableType=").append(tableType).append(", ").append("column=").append(column).append('}').toString();

dbsyncer-connector/src/main/java/org/dbsyncer/connector/model/SqlTable.java

@@ -0,0 +1,32 @@
+package org.dbsyncer.connector.model;
+
+public class SqlTable {
+
+    private String sql;
+
+    private String table;
+
+    public SqlTable() {
+    }
+
+    public SqlTable(String sql, String table) {
+        this.sql = sql;
+        this.table = table;
+    }
+
+    public String getSql() {
+        return sql;
+    }
+
+    public void setSql(String sql) {
+        this.sql = sql;
+    }
+
+    public String getTable() {
+        return table;
+    }
+
+    public void setTable(String table) {
+        this.table = table;
+    }
+}

dbsyncer-connector/src/main/java/org/dbsyncer/connector/model/Table.java

@@ -32,6 +32,11 @@ public class Table {
      */
     private List<Field> column;
 
+    /**
+     * sql
+     */
+    private String sql;
+
     // 总数
     private long count;
 
@@ -43,14 +48,15 @@ public class Table {
     }
 
     public Table(String name, String type) {
-        this(name, type, null, null);
+        this(name, type, null, null, null);
     }
 
-    public Table(String name, String type, String primaryKey, List<Field> column) {
+    public Table(String name, String type, String primaryKey, List<Field> column, String sql) {
         this.name = name;
         this.type = type;
         this.primaryKey = primaryKey;
         this.column = column;
+        this.sql = sql;
     }
 
     public String getName() {
@@ -88,6 +94,15 @@ public class Table {
         return this;
     }
 
+    public String getSql() {
+        return sql;
+    }
+
+    public Table setSql(String sql) {
+        this.sql = sql;
+        return this;
+    }
+
     public long getCount() {
         return count;
     }

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

@@ -1,5 +1,6 @@
 package org.dbsyncer.connector.sql;
 
+import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.config.CommandConfig;
 import org.dbsyncer.connector.config.DatabaseConfig;
@@ -9,8 +10,11 @@ import org.dbsyncer.connector.database.DatabaseConnectorMapper;
 import org.dbsyncer.connector.enums.SqlBuilderEnum;
 import org.dbsyncer.connector.model.MetaInfo;
 import org.dbsyncer.connector.model.PageSql;
+import org.dbsyncer.connector.model.SqlTable;
 import org.dbsyncer.connector.model.Table;
+import org.dbsyncer.connector.util.PrimaryKeyUtil;
 
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -23,9 +27,18 @@ import java.util.Map;
 public abstract class AbstractDQLConnector extends AbstractDatabaseConnector {
 
     @Override
-    public List<Table> getTable(DatabaseConnectorMapper config) {
-        DatabaseConfig cfg = config.getConfig();
-        return super.getTable(config, null, getSchema(cfg), cfg.getTable());
+    public List<Table> getTable(DatabaseConnectorMapper connectorMapper) {
+        DatabaseConfig cfg = connectorMapper.getConfig();
+        List<SqlTable> sqlTables = cfg.getSqlTables();
+        List<Table> tables = new ArrayList<>();
+        if (!CollectionUtils.isEmpty(sqlTables)) {
+            sqlTables.forEach(s -> {
+                Table table = new Table(s.getTable());
+                table.setSql(s.getSql());
+                tables.add(table);
+            });
+        }
+        return tables;
     }
 
     @Override
@@ -36,12 +49,18 @@ public abstract class AbstractDQLConnector extends AbstractDatabaseConnector {
     @Override
     public MetaInfo getMetaInfo(DatabaseConnectorMapper connectorMapper, String tableName) {
         DatabaseConfig cfg = connectorMapper.getConfig();
-        String sql = cfg.getSql().toUpperCase();
-        sql = sql.replace("\t", " ");
-        sql = sql.replace("\r", " ");
-        sql = sql.replace("\n", " ");
-        String queryMetaSql = StringUtil.contains(sql, " WHERE ") ? cfg.getSql() + " AND 1!=1 " : cfg.getSql() + " WHERE 1!=1 ";
-        return connectorMapper.execute(databaseTemplate -> super.getMetaInfo(databaseTemplate, queryMetaSql, getSchema(cfg), cfg.getTable()));
+        List<SqlTable> sqlTables = cfg.getSqlTables();
+        for (SqlTable s : sqlTables) {
+            if (s.getTable().equals(tableName)) {
+                String sql = s.getSql().toUpperCase();
+                sql = sql.replace("\t", " ");
+                sql = sql.replace("\r", " ");
+                sql = sql.replace("\n", " ");
+                String queryMetaSql = StringUtil.contains(sql, " WHERE ") ? s.getSql() + " AND 1!=1 " : s.getSql() + " WHERE 1!=1 ";
+                return connectorMapper.execute(databaseTemplate -> super.getMetaInfo(databaseTemplate, queryMetaSql, getSchema(cfg), s.getTable()));
+            }
+        }
+        return null;
     }
 
     /**
@@ -54,18 +73,19 @@ public abstract class AbstractDQLConnector extends AbstractDatabaseConnector {
     protected Map<String, String> getDqlSourceCommand(CommandConfig commandConfig, boolean groupByPK) {
         // 获取过滤SQL
         String queryFilterSql = getQueryFilterSql(commandConfig.getFilter());
-        DatabaseConfig cfg = (DatabaseConfig) commandConfig.getConnectorConfig();
+        Table table = commandConfig.getTable();
+        String primaryKey = PrimaryKeyUtil.findOriginalTablePrimaryKey(commandConfig.getOriginalTable());
 
         // 获取查询SQL
         Map<String, String> map = new HashMap<>();
-        String querySql = cfg.getSql();
+        String querySql = table.getSql();
 
         // 存在条件
         if (StringUtil.isNotBlank(queryFilterSql)) {
             querySql += queryFilterSql;
         }
         String quotation = buildSqlWithQuotation();
-        String pk = new StringBuilder(quotation).append(cfg.getPrimaryKey()).append(quotation).toString();
+        String pk = new StringBuilder(quotation).append(primaryKey).append(quotation).toString();
         map.put(SqlBuilderEnum.QUERY.getName(), getPageSql(new PageSql(querySql, pk)));
 
         // 获取查询总数SQL

dbsyncer-listener/src/main/java/org/dbsyncer/listener/AbstractDatabaseExtractor.java

@@ -3,15 +3,17 @@ package org.dbsyncer.listener;
 import org.dbsyncer.common.event.RowChangedEvent;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.config.DatabaseConfig;
 import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.connector.database.DatabaseConnectorMapper;
 import org.dbsyncer.connector.model.Field;
 import org.dbsyncer.connector.model.MetaInfo;
+import org.dbsyncer.connector.model.Table;
+import org.dbsyncer.connector.util.PrimaryKeyUtil;
 import org.springframework.util.Assert;
 
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * @author AE86
@@ -20,14 +22,14 @@ import java.util.Map;
  */
 public abstract class AbstractDatabaseExtractor extends AbstractExtractor {
 
-    private DqlMapper dqlMapper;
+    private Map<String, DqlMapper> dqlMap = new ConcurrentHashMap<>();
 
     /**
      * 发送增量事件
      *
      * @param event
      */
-    protected void sendChangedEvent(RowChangedEvent event){
+    protected void sendChangedEvent(RowChangedEvent event) {
         changedEvent(event);
     }
 
@@ -37,16 +39,19 @@ public abstract class AbstractDatabaseExtractor extends AbstractExtractor {
      * @param event
      */
     protected void sendDqlChangedEvent(RowChangedEvent event) {
-        if (null != event && event.getSourceTableName().equals(dqlMapper.tableName)) {
-            switch (event.getEvent()) {
-                case ConnectorConstant.OPERTION_UPDATE:
-                case ConnectorConstant.OPERTION_INSERT:
-                    event.setDataList(queryDqlData(event.getDataList()));
-                    break;
-                default:
-                    break;
+        if (null != event) {
+            DqlMapper dqlMapper = dqlMap.get(event.getSourceTableName());
+            if (null != dqlMapper) {
+                switch (event.getEvent()) {
+                    case ConnectorConstant.OPERTION_UPDATE:
+                    case ConnectorConstant.OPERTION_INSERT:
+                        event.setDataList(queryDqlData(dqlMapper, event.getDataList()));
+                        break;
+                    default:
+                        break;
+                }
+                changedEvent(event);
             }
-            changedEvent(event);
         }
     }
 
@@ -55,27 +60,31 @@ public abstract class AbstractDatabaseExtractor extends AbstractExtractor {
      */
     protected void postProcessDqlBeforeInitialization() {
         DatabaseConnectorMapper mapper = (DatabaseConnectorMapper) connectorFactory.connect(connectorConfig);
-        DatabaseConfig cfg = mapper.getConfig();
-        final String tableName = cfg.getTable();
-        final String primaryKey = cfg.getPrimaryKey();
-        Assert.hasText(tableName, String.format("The table name '%s' is null.", tableName));
-        Assert.hasText(primaryKey, "The primaryKey is null.");
-        MetaInfo metaInfo = connectorFactory.getMetaInfo(mapper, tableName);
-        final List<Field> column = metaInfo.getColumn();
-        Assert.notEmpty(column, String.format("The column of table name '%s' is empty.", tableName));
+        for (Table table : sourceTable) {
+            String sql = table.getSql();
+            String tableName = table.getName();
+            String primaryKey = PrimaryKeyUtil.findOriginalTablePrimaryKey(table);
+            Assert.hasText(sql, "The sql is null.");
+            Assert.hasText(tableName, "The tableName is null.");
+            Assert.hasText(primaryKey, "The primaryKey is null.");
 
-        String sql = cfg.getSql().toUpperCase();
-        sql = sql.replace("\t", " ");
-        sql = sql.replace("\r", " ");
-        sql = sql.replace("\n", " ");
-        StringBuilder querySql = new StringBuilder(cfg.getSql());
-        if(StringUtil.contains(sql, " WHERE ")){
-            querySql.append(" AND ");
-        }else{
-            querySql.append(" WHERE ");
+            MetaInfo metaInfo = connectorFactory.getMetaInfo(mapper, tableName);
+            final List<Field> column = metaInfo.getColumn();
+            Assert.notEmpty(column, String.format("The column of table name '%s' is empty.", tableName));
+
+            sql = sql.toUpperCase().replace("\t", " ");
+            sql = sql.replace("\r", " ");
+            sql = sql.replace("\n", " ");
+            StringBuilder querySql = new StringBuilder(table.getSql());
+            if (StringUtil.contains(sql, " WHERE ")) {
+                querySql.append(" AND ");
+            } else {
+                querySql.append(" WHERE ");
+            }
+            querySql.append(primaryKey).append("=?");
+            DqlMapper dqlMapper = new DqlMapper(mapper, querySql.toString(), tableName, column, getPKIndex(column, primaryKey));
+            dqlMap.putIfAbsent(tableName, dqlMapper);
         }
-        querySql.append(primaryKey).append("=?");
-        dqlMapper = new DqlMapper(mapper, querySql.toString(), tableName, column, getPKIndex(column, primaryKey));
     }
 
     /**
@@ -99,7 +108,7 @@ public abstract class AbstractDatabaseExtractor extends AbstractExtractor {
         return pkIndex;
     }
 
-    private List<Object> queryDqlData(List<Object> data) {
+    private List<Object> queryDqlData(DqlMapper dqlMapper, List<Object> data) {
         if (!CollectionUtils.isEmpty(data)) {
             Map<String, Object> row = dqlMapper.mapper.execute(databaseTemplate -> databaseTemplate.queryForMap(dqlMapper.sql, data.get(dqlMapper.pkIndex)));
             if (!CollectionUtils.isEmpty(row)) {

dbsyncer-listener/src/main/java/org/dbsyncer/listener/AbstractExtractor.java

@@ -6,6 +6,7 @@ import org.dbsyncer.common.model.AbstractConnectorConfig;
 import org.dbsyncer.common.scheduled.ScheduledTaskService;
 import org.dbsyncer.connector.ConnectorFactory;
 import org.dbsyncer.connector.constant.ConnectorConstant;
+import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.listener.config.ListenerConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -13,7 +14,6 @@ import org.slf4j.LoggerFactory;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -24,25 +24,19 @@ import java.util.concurrent.TimeUnit;
 public abstract class AbstractExtractor implements Extractor {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
-    protected String metaId;
     protected ConnectorFactory connectorFactory;
     protected ScheduledTaskService scheduledTaskService;
     protected AbstractConnectorConfig connectorConfig;
     protected ListenerConfig listenerConfig;
-    protected Map<String, String> snapshot;
     protected Set<String> filterTable;
-    private final List<Event> watcher = new CopyOnWriteArrayList<>();
-
-    @Override
-    public void addListener(Event event) {
-        if (null != event) {
-            watcher.add(event);
-        }
-    }
+    protected List<Table> sourceTable;
+    protected Map<String, String> snapshot;
+    protected String metaId;
+    protected Event watcher;
 
     @Override
-    public void clearAllListener() {
-        watcher.clear();
+    public void register(Event event) {
+        watcher = event;
     }
 
     @Override
@@ -69,23 +63,23 @@ public abstract class AbstractExtractor implements Extractor {
 
     @Override
     public void flushEvent() {
-        watcher.forEach(w -> w.flushEvent(snapshot));
+        watcher.flushEvent(snapshot);
     }
 
     @Override
     public void forceFlushEvent() {
         logger.info("Force flush:{}", snapshot);
-        watcher.forEach(w -> w.forceFlushEvent(snapshot));
+        watcher.forceFlushEvent(snapshot);
     }
 
     @Override
     public void errorEvent(Exception e) {
-        watcher.forEach(w -> w.errorEvent(e));
+        watcher.errorEvent(e);
     }
 
     @Override
     public void interruptException(Exception e) {
-        watcher.forEach(w -> w.interruptException(e));
+        watcher.interruptException(e);
     }
 
     protected void sleepInMills(long timeout) {
@@ -104,14 +98,10 @@ public abstract class AbstractExtractor implements Extractor {
      */
     private void processEvent(boolean permitEvent, RowChangedEvent event) {
         if (permitEvent) {
-            watcher.forEach(w -> w.changedEvent(event));
+            watcher.changedEvent(event);
         }
     }
 
-    public void setMetaId(String metaId) {
-        this.metaId = metaId;
-    }
-
     public void setConnectorFactory(ConnectorFactory connectorFactory) {
         this.connectorFactory = connectorFactory;
     }
@@ -128,11 +118,20 @@ public abstract class AbstractExtractor implements Extractor {
         this.listenerConfig = listenerConfig;
     }
 
+    public void setFilterTable(Set<String> filterTable) {
+        this.filterTable = filterTable;
+    }
+
+    public AbstractExtractor setSourceTable(List<Table> sourceTable) {
+        this.sourceTable = sourceTable;
+        return this;
+    }
+
     public void setSnapshot(Map<String, String> snapshot) {
         this.snapshot = snapshot;
     }
 
-    public void setFilterTable(Set<String> filterTable) {
-        this.filterTable = filterTable;
+    public void setMetaId(String metaId) {
+        this.metaId = metaId;
     }
 }

dbsyncer-listener/src/main/java/org/dbsyncer/listener/Extractor.java

@@ -16,16 +16,11 @@ public interface Extractor {
     void close();
 
     /**
-     * 添加监听器（获取增量数据）
+     * 注册监听事件（获取增量数据）
      *
      * @param event
      */
-    void addListener(Event event);
-
-    /**
-     * 清空监听器
-     */
-    void clearAllListener();
+    void register(Event event);
 
     /**
      * 数据变更事件

dbsyncer-manager/src/main/java/org/dbsyncer/manager/puller/IncrementPuller.java

@@ -30,7 +30,6 @@ import org.dbsyncer.parser.model.TableGroup;
 import org.dbsyncer.parser.util.PickerUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Component;
 import org.springframework.util.Assert;
 
@@ -47,7 +46,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
@@ -81,10 +79,6 @@ public class IncrementPuller extends AbstractPuller implements ScheduledTaskJob
     @Resource
     private ConnectorFactory connectorFactory;
 
-    @Qualifier("taskExecutor")
-    @Resource
-    private Executor taskExecutor;
-
     private Map<String, Extractor> map = new ConcurrentHashMap<>();
 
     @PostConstruct
@@ -127,7 +121,6 @@ public class IncrementPuller extends AbstractPuller implements ScheduledTaskJob
     public void close(String metaId) {
         Extractor extractor = map.get(metaId);
         if (null != extractor) {
-            extractor.clearAllListener();
             extractor.close();
         }
         map.remove(metaId);
@@ -141,50 +134,56 @@ public class IncrementPuller extends AbstractPuller implements ScheduledTaskJob
         map.forEach((k, v) -> v.flushEvent());
     }
 
-    private AbstractExtractor getExtractor(Mapping mapping, Connector connector, List<TableGroup> list, Meta meta)
-            throws InstantiationException, IllegalAccessException {
+    private AbstractExtractor getExtractor(Mapping mapping, Connector connector, List<TableGroup> list, Meta meta) throws InstantiationException, IllegalAccessException {
         AbstractConnectorConfig connectorConfig = connector.getConfig();
         ListenerConfig listenerConfig = mapping.getListener();
 
         // timing/log
         final String listenerType = listenerConfig.getListenerType();
 
+        AbstractExtractor extractor = null;
         // 默认定时抽取
         if (ListenerTypeEnum.isTiming(listenerType)) {
-            AbstractQuartzExtractor extractor = listener.getExtractor(ListenerTypeEnum.TIMING, connectorConfig.getConnectorType(), AbstractQuartzExtractor.class);
-            extractor.setCommands(list.stream().map(t -> {
+            AbstractQuartzExtractor quartzExtractor = listener.getExtractor(ListenerTypeEnum.TIMING, connectorConfig.getConnectorType(), AbstractQuartzExtractor.class);
+            quartzExtractor.setCommands(list.stream().map(t -> {
                 String pk = PrimaryKeyUtil.findOriginalTablePrimaryKey(t.getSourceTable());
                 return new TableGroupCommand(pk, t.getCommand());
             }).collect(Collectors.toList()));
-            setExtractorConfig(extractor, connectorConfig, listenerConfig, meta.getSnapshot(), new QuartzListener(mapping, list));
-            return extractor;
+            quartzExtractor.register(new QuartzListener(mapping, list));
+            extractor = quartzExtractor;
         }
 
         // 基于日志抽取
         if (ListenerTypeEnum.isLog(listenerType)) {
-            AbstractExtractor extractor = listener.getExtractor(ListenerTypeEnum.LOG, connectorConfig.getConnectorType(), AbstractExtractor.class);
+            extractor = listener.getExtractor(ListenerTypeEnum.LOG, connectorConfig.getConnectorType(), AbstractExtractor.class);
+            extractor.register(new LogListener(mapping, list, extractor));
+        }
+
+        if (null != extractor) {
             Set<String> filterTable = new HashSet<>();
-            LogListener logListener = new LogListener(mapping, list, extractor);
-            logListener.getTablePicker().forEach((k, fieldPickers) -> filterTable.add(k));
+            List<Table> sourceTable = new ArrayList<>();
+            list.forEach(t -> {
+                Table table = t.getSourceTable();
+                if (!filterTable.contains(t.getName())) {
+                    sourceTable.add(table);
+                }
+                filterTable.add(table.getName());
+            });
+
+            extractor.setConnectorFactory(connectorFactory);
+            extractor.setScheduledTaskService(scheduledTaskService);
+            extractor.setConnectorConfig(connectorConfig);
+            extractor.setListenerConfig(listenerConfig);
             extractor.setFilterTable(filterTable);
-            setExtractorConfig(extractor, connectorConfig, listenerConfig, meta.getSnapshot(), logListener);
+            extractor.setSourceTable(sourceTable);
+            extractor.setSnapshot(meta.getSnapshot());
+            extractor.setMetaId(meta.getId());
             return extractor;
         }
 
         throw new ManagerException("未知的监听配置.");
     }
 
-    private void setExtractorConfig(AbstractExtractor extractor, AbstractConnectorConfig connector, ListenerConfig listener,
-                                    Map<String, String> snapshot, AbstractListener event) {
-        extractor.setConnectorFactory(connectorFactory);
-        extractor.setScheduledTaskService(scheduledTaskService);
-        extractor.setConnectorConfig(connector);
-        extractor.setListenerConfig(listener);
-        extractor.setSnapshot(snapshot);
-        extractor.addListener(event);
-        extractor.setMetaId(event.metaId);
-    }
-
     abstract class AbstractListener implements Event {
         private static final int FLUSH_DELAYED_SECONDS = 30;
         protected Mapping mapping;
@@ -245,11 +244,11 @@ public class IncrementPuller extends AbstractPuller implements ScheduledTaskJob
 
         private List<FieldPicker> tablePicker;
 
-        public QuartzListener(Mapping mapping, List<TableGroup> list) {
+        public QuartzListener(Mapping mapping, List<TableGroup> tableGroups) {
             this.mapping = mapping;
             this.metaId = mapping.getMetaId();
             this.tablePicker = new LinkedList<>();
-            list.forEach(t -> tablePicker.add(new FieldPicker(PickerUtil.mergeTableGroupConfig(mapping, t))));
+            tableGroups.forEach(t -> tablePicker.add(new FieldPicker(PickerUtil.mergeTableGroupConfig(mapping, t))));
         }
 
         @Override
@@ -290,13 +289,13 @@ public class IncrementPuller extends AbstractPuller implements ScheduledTaskJob
         private AtomicInteger eventCounter;
         private static final int MAX_LOG_CACHE_SIZE = 128;
 
-        public LogListener(Mapping mapping, List<TableGroup> list, Extractor extractor) {
+        public LogListener(Mapping mapping, List<TableGroup> tableGroups, Extractor extractor) {
             this.mapping = mapping;
             this.metaId = mapping.getMetaId();
             this.extractor = extractor;
             this.tablePicker = new LinkedHashMap<>();
             this.eventCounter = new AtomicInteger();
-            list.forEach(t -> {
+            tableGroups.forEach(t -> {
                 final Table table = t.getSourceTable();
                 final String tableName = table.getName();
                 List<Field> pkList = t.getTargetTable().getColumn().stream().filter(field -> field.isPk()).collect(Collectors.toList());
@@ -328,10 +327,6 @@ public class IncrementPuller extends AbstractPuller implements ScheduledTaskJob
                 eventCounter.set(0);
             }
         }
-
-        public Map<String, List<FieldPicker>> getTablePicker() {
-            return tablePicker;
-        }
     }
 
 }

dbsyncer-parser/src/main/java/org/dbsyncer/parser/ParserFactory.java

@@ -139,6 +139,7 @@ public class ParserFactory implements Parser {
             for (Table t : connector.getTable()) {
                 if (t.getName().equals(tableName)) {
                     metaInfo.setTableType(t.getType());
+                    metaInfo.setSql(t.getSql());
                     break;
                 }
             }
@@ -152,8 +153,8 @@ public class ParserFactory implements Parser {
         AbstractConnectorConfig tConnConfig = getConnectorConfig(mapping.getTargetConnectorId());
         Table sourceTable = tableGroup.getSourceTable();
         Table targetTable = tableGroup.getTargetTable();
-        Table sTable = new Table(sourceTable.getName(), sourceTable.getType(), sourceTable.getPrimaryKey(), new ArrayList<>());
-        Table tTable = new Table(targetTable.getName(), targetTable.getType(), targetTable.getPrimaryKey(), new ArrayList<>());
+        Table sTable = new Table(sourceTable.getName(), sourceTable.getType(), sourceTable.getPrimaryKey(), new ArrayList<>(), sourceTable.getSql());
+        Table tTable = new Table(targetTable.getName(), targetTable.getType(), targetTable.getPrimaryKey(), new ArrayList<>(), sourceTable.getSql());
         List<FieldMapping> fieldMapping = tableGroup.getFieldMapping();
         if (!CollectionUtils.isEmpty(fieldMapping)) {
             fieldMapping.forEach(m -> {

dbsyncer-web/src/main/resources/application.properties

@@ -20,9 +20,9 @@ dbsyncer.parser.flush.buffer.actuator.queue-capacity=100000
 dbsyncer.parser.flush.buffer.actuator.period-millisecond=300
 
 #plugin
-dbsyncer.plugin.notify.mail.enabled=false
-dbsyncer.plugin.notify.mail.username=your mail username
-dbsyncer.plugin.notify.mail.password=your mail authorization code
+dbsyncer.plugin.notify.mail.enabled=true
+dbsyncer.plugin.notify.mail.username=836391306@qq.com
+dbsyncer.plugin.notify.mail.password=wqosskohcvmlbbig
 
 #storage
 dbsyncer.storage.binlog.recorder.batch-count=3000
@@ -30,8 +30,8 @@ dbsyncer.storage.binlog.recorder.max-processing-seconds=120
 dbsyncer.storage.binlog.recorder.queue-capacity=10000
 dbsyncer.storage.binlog.recorder.writer-period-millisecond=500
 dbsyncer.storage.binlog.recorder.reader-period-millisecond=2000
-dbsyncer.storage.support.mysql.enabled=false
-dbsyncer.storage.support.mysql.config.url=jdbc:mysql://127.0.0.1:3306/dbsyncer?rewriteBatchedStatements=true&seUnicode=true&characterEncoding=UTF8&serverTimezone=Asia/Shanghai&useSSL=false&verifyServerCertificate=false
+dbsyncer.storage.support.mysql.enabled=true
+dbsyncer.storage.support.mysql.config.url=jdbc:mysql://127.0.0.1:3305/dbsyncer?rewriteBatchedStatements=true&seUnicode=true&characterEncoding=UTF8&serverTimezone=Asia/Shanghai&useSSL=false&verifyServerCertificate=false
 dbsyncer.storage.support.mysql.config.username=root
 dbsyncer.storage.support.mysql.config.password=123
 

dbsyncer-web/src/main/resources/public/connector/addDqlMysql.html

@@ -17,8 +17,8 @@
     <div class="form-group">
         <label class="col-sm-2 control-label">SQL <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-10">
-            <textarea id="sql" name="sql" class="form-control" maxlength="8192" dbsyncer-valid="require" rows="10"
-                      th:text="${connector?.config?.sql}?:'SELECT T1.* FROM USER T1'"></textarea>
+            <textarea name="sql" class="sql form-control" maxlength="8192" dbsyncer-valid="require" rows="10"
+                      th:text="${connector?.config?.sqlTables[0]?.sql}?:'SELECT T1.* FROM USER T1'"></textarea>
         </div>
     </div>
     <div class="form-group">
@@ -32,13 +32,9 @@
         <label class="col-sm-2 control-label">主表 <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-4">
             <input class="form-control" name="table" type="text" maxlength="32" dbsyncer-valid="require"
-                   placeholder="USER" th:value="${connector?.config?.table}"/>
-        </div>
-        <label class="col-sm-2 control-label">主键 <i class="fa fa-question-circle fa_gray" aria-hidden="true" title="如果使用的表没有主键，可以自定义主键（大小写必须一致，添加表映射关系时，主键不能为空）。"></i><strong class="driverVerifcateRequired">*</strong></label>
-        <div class="col-sm-4">
-            <input class="form-control" name="primaryKey" type="text" maxlength="32" dbsyncer-valid="require"
-                   placeholder="ID" th:value="${connector?.config?.primaryKey}"/>
+                   placeholder="USER" th:value="${connector?.config?.sqlTables[0]?.table}"/>
         </div>
+        <div class="col-sm-6"></div>
     </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">URL <strong class="driverVerifcateRequired">*</strong></label>

dbsyncer-web/src/main/resources/public/connector/addDqlOracle.html

@@ -17,9 +17,9 @@
     <div class="form-group">
         <label class="col-sm-2 control-label">SQL <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-10">
-            <textarea id="sql" name="sql" class="form-control dbsyncer_textarea_resize_none" maxlength="8192"
+            <textarea name="sql" class="sql form-control dbsyncer_textarea_resize_none" maxlength="8192"
                       dbsyncer-valid="require" rows="10"
-                      th:text="${connector?.config?.sql}?:'SELECT T1.*,ROWIDTOCHAR(ROWID) as RID FROM &quot;USER&quot; T1'"></textarea>
+                      th:text="${connector?.config?.sqlTables[0]?.sql}?:'SELECT T1.*,ROWIDTOCHAR(ROWID) as RID FROM &quot;USER&quot; T1'"></textarea>
         </div>
     </div>
     <div class="form-group">
@@ -33,13 +33,9 @@
         <label class="col-sm-2 control-label">主表 <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-4">
             <input class="form-control" name="table" type="text" maxlength="32" dbsyncer-valid="require"
-                   placeholder="USER" th:value="${connector?.config?.table}"/>
-        </div>
-        <label class="col-sm-2 control-label">主键 <i class="fa fa-question-circle fa_gray" aria-hidden="true" title="如果使用的表没有主键，可以自定义主键（大小写必须一致，添加表映射关系时，主键不能为空）。"></i><strong class="driverVerifcateRequired">*</strong></label>
-        <div class="col-sm-4">
-            <input class="form-control" name="primaryKey" type="text" maxlength="32" dbsyncer-valid="require"
-                   placeholder="ID" th:value="${connector?.config?.primaryKey}"/>
+                   placeholder="USER" th:value="${connector?.config?.sqlTables[0]?.table}"/>
         </div>
+        <div class="col-sm-6"></div>
     </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">URL <strong class="driverVerifcateRequired">*</strong></label>

dbsyncer-web/src/main/resources/public/connector/addDqlPostgreSQL.html

@@ -18,9 +18,9 @@
     <div class="form-group">
         <label class="col-sm-2 control-label">SQL <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-10">
-            <textarea class="form-control dbsyncer_textarea_resize_none" maxlength="8192" dbsyncer-valid="require"
-                      id="sql" name="sql" rows="10"
-                      th:text="${connector?.config?.sql}?:'SELECT T1.* FROM &quot;USER&quot; T1'"></textarea>
+            <textarea name="sql" class="sql form-control dbsyncer_textarea_resize_none" maxlength="8192"
+                      dbsyncer-valid="require" rows="10"
+                      th:text="${connector?.config?.sqlTables[0]?.sql}?:'SELECT T1.* FROM &quot;USER&quot; T1'"></textarea>
         </div>
     </div>
     <div class="form-group">
@@ -34,13 +34,9 @@
         <label class="col-sm-2 control-label">主表 <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-4">
             <input class="form-control" name="table" type="text" maxlength="32" dbsyncer-valid="require"
-                   placeholder="USER" th:value="${connector?.config?.table}"/>
-        </div>
-        <label class="col-sm-2 control-label">主键 <i class="fa fa-question-circle fa_gray" aria-hidden="true" title="如果使用的表没有主键，可以自定义主键（大小写必须一致，添加表映射关系时，主键不能为空）。"></i><strong class="driverVerifcateRequired">*</strong></label>
-        <div class="col-sm-4">
-            <input class="form-control" name="primaryKey" type="text" maxlength="32" dbsyncer-valid="require"
-                   placeholder="ID" th:value="${connector?.config?.primaryKey}"/>
+                   placeholder="USER" th:value="${connector?.config?.sqlTables[0]?.table}"/>
         </div>
+        <div class="col-sm-6"></div>
     </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">URL <strong class="driverVerifcateRequired">*</strong></label>

dbsyncer-web/src/main/resources/public/connector/addDqlSqlServer.html

@@ -17,9 +17,9 @@
     <div class="form-group">
         <label class="col-sm-2 control-label">SQL <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-10">
-            <textarea id="sql" name="sql" class="form-control dbsyncer_textarea_resize_none" maxlength="1024"
+            <textarea name="sql" class="sql form-control dbsyncer_textarea_resize_none" maxlength="1024"
                       dbsyncer-valid="require" rows="10"
-                      th:text="${connector?.config?.sql}?:'SELECT T1.* FROM USER T1'"></textarea>
+                      th:text="${connector?.config?.sqlTables[0]?.sql}?:'SELECT T1.* FROM USER T1'"></textarea>
         </div>
     </div>
     <div class="form-group">
@@ -33,13 +33,9 @@
         <label class="col-sm-2 control-label">主表 <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-4">
             <input class="form-control" name="table" type="text" maxlength="32" dbsyncer-valid="require"
-                   placeholder="USER" th:value="${connector?.config?.table}"/>
-        </div>
-        <label class="col-sm-2 control-label">主键 <i class="fa fa-question-circle fa_gray" aria-hidden="true" title="如果使用的表没有主键，可以自定义主键（大小写必须一致，添加表映射关系时，主键不能为空）。"></i><strong class="driverVerifcateRequired">*</strong></label>
-        <div class="col-sm-4">
-            <input class="form-control" name="primaryKey" type="text" maxlength="32" dbsyncer-valid="require"
-                   placeholder="ID" th:value="${connector?.config?.primaryKey}"/>
+                   placeholder="USER" th:value="${connector?.config?.sqlTables[0]?.table}"/>
         </div>
+        <div class="col-sm-6"></div>
     </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">URL <strong class="driverVerifcateRequired">*</strong></label>

dbsyncer-web/src/main/resources/public/connector/addElasticsearch.html

@@ -14,21 +14,14 @@
         </div>
     </div>
     <div class="form-group">
-        <label class="col-sm-2 control-label">index <strong class="driverVerifcateRequired">*</strong></label>
-        <div class="col-sm-4">
-            <input class="form-control" name="index" type="text" maxlength="32" dbsyncer-valid="require" placeholder="test" th:value="${connector?.config?.index}?:'test'"/>
-        </div>
-        <label class="col-sm-2 control-label">主键 <i class="fa fa-question-circle fa_gray" aria-hidden="true" title="如果使用的表没有主键，可以自定义主键（大小写必须一致，添加表映射关系时，主键不能为空）。"></i><strong class="driverVerifcateRequired">*</strong></label>
+        <label class="col-sm-2 control-label">URL <i class="fa fa-question-circle fa_gray" aria-hidden="true" title="多个使用英文逗号，例如：http://192.168.1.100:9200,http://192.168.1.200:9200"></i> <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-4">
-            <input class="form-control" name="primaryKey" type="text" maxlength="32" dbsyncer-valid="require" placeholder="id" th:value="${connector?.config?.primaryKey}?:'id'"/>
+            <input class="form-control" name="url" type="text" maxlength="1024" dbsyncer-valid="require" th:value="${connector?.config?.url}?:'http://127.0.0.1:9200'"/>
         </div>
-    </div>
-    <div class="form-group">
-        <label class="col-sm-2 control-label">地址 <i class="fa fa-question-circle fa_gray" aria-hidden="true" title="多个使用英文逗号，例如：http://192.168.1.100:9200,http://192.168.1.200:9200"></i> <strong class="driverVerifcateRequired">*</strong></label>
+        <label class="col-sm-2 control-label">index <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-4">
-            <textarea id="sql" name="url" class="form-control dbsyncer_textarea_resize_none" maxlength="1024" dbsyncer-valid="require" rows="2" th:text="${connector?.config?.url}?:'http://127.0.0.1:9200'"></textarea>
+            <input class="form-control" name="index" type="text" maxlength="32" dbsyncer-valid="require" placeholder="test" th:value="${connector?.config?.index}?:'test'"/>
         </div>
-        <div class="col-sm-6"></div>
     </div>
 
     <script type="text/javascript">

dbsyncer-web/src/main/resources/public/connector/addKafka.html

@@ -4,9 +4,9 @@
 
 <div th:fragment="content">
     <div class="form-group">
-        <label class="col-sm-2 control-label">地址 <i class="fa fa-question-circle fa_gray" aria-hidden="true" title="多个使用英文逗号，例如：192.168.1.100:9092,192.168.1.200:9092"></i> <strong class="driverVerifcateRequired">*</strong></label>
+        <label class="col-sm-2 control-label">URL <i class="fa fa-question-circle fa_gray" aria-hidden="true" title="多个使用英文逗号，例如：192.168.1.100:9092,192.168.1.200:9092"></i> <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-4">
-            <textarea name="bootstrapServers" class="form-control dbsyncer_textarea_resize_none" maxlength="1024" dbsyncer-valid="require" rows="2" th:text="${connector?.config?.bootstrapServers}?:'127.0.0.1:9092'"></textarea>
+            <input class="form-control" name="bootstrapServers" type="text" maxlength="1024" dbsyncer-valid="require" th:value="${connector?.config?.bootstrapServers}?:'127.0.0.1:9092'"/>
         </div>
         <label class="col-sm-2 control-label">Topic <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-4">

dbsyncer-web/src/main/resources/static/js/common.js

@@ -38,7 +38,7 @@ function backIndexPage(projectGroupId) {
 
 // 美化SQL
 function beautifySql(){
-    var $sql = $("#sql");
+    var $sql = $(".sql");
     var $tmp = $sql.attr('tmp');
     if(null == $tmp){
         $sql.attr('tmp', $sql.val());