增加oracle视图

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

@@ -6,6 +6,7 @@ import org.dbsyncer.biz.checker.ConnectorConfigChecker;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.config.ConnectorConfig;
+import org.dbsyncer.connector.config.Table;
 import org.dbsyncer.connector.enums.ConnectorEnum;
 import org.dbsyncer.manager.Manager;
 import org.dbsyncer.parser.logger.LogService;
@@ -115,7 +116,7 @@ public class ConnectorChecker extends AbstractChecker {
         Assert.isTrue(isAlive, "无法连接.");
         // 获取表信息
         ConnectorMapper connectorMapper = manager.connect(connector.getConfig());
-        List<String> table = manager.getTable(connectorMapper);
+        List<Table> table = manager.getTable(connectorMapper);
         connector.setTable(table);
     }
 

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/mapping/TimingConfigChecker.java

@@ -35,9 +35,7 @@ public class TimingConfigChecker implements MappingConfigChecker {
         if (StringUtil.isNotBlank(period)) {
             config.setPeriod(NumberUtil.toLong(period, 30));
         }
-        if (StringUtil.isNotBlank(eventFieldName)) {
-            config.setEventFieldName(eventFieldName);
-        }
+        config.setEventFieldName(eventFieldName);
         if (StringUtil.isNotBlank(insert)) {
             config.setInsert(insert);
         }

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

@@ -133,7 +133,7 @@ public class TableGroupChecker extends AbstractChecker {
     private Table getTable(String connectorId, String tableName) {
         MetaInfo metaInfo = manager.getMetaInfo(connectorId, tableName);
         Assert.notNull(metaInfo, "无法获取连接器表信息.");
-        return new Table().setName(tableName).setColumn(metaInfo.getColumn());
+        return new Table(tableName, metaInfo.getTableType(), metaInfo.getColumn());
     }
 
     private void checkRepeatedTable(String mappingId, String sourceTable, String targetTable) {

dbsyncer-biz/src/main/java/org/dbsyncer/biz/impl/MappingServiceImpl.java

@@ -201,8 +201,8 @@ public class MappingServiceImpl extends BaseServiceImpl implements MappingServic
         }
 
         // 存在交集
-        List<String> sTables = new LinkedList<>(s.getTable());
-        List<String> tTables = new LinkedList<>(t.getTable());
+        List<String> sTables = s.getTable().stream().map(table -> table.getName()).collect(Collectors.toList());
+        List<String> tTables = t.getTable().stream().map(table -> table.getName()).collect(Collectors.toList());
         sTables.retainAll(tTables);
         if (!CollectionUtils.isEmpty(sTables)) {
             Map<String, String> params = new HashMap<>();

dbsyncer-common/src/main/java/org/dbsyncer/common/event/RowChangedEvent.java

@@ -17,13 +17,14 @@ import java.util.Map;
  */
 public class RowChangedEvent {
 
-    private int                 tableGroupIndex;
-    private String              tableName;
-    private String              event;
-    private List<Object>        beforeData;
-    private List<Object>        afterData;
+    private int tableGroupIndex;
+    private String tableName;
+    private String event;
+    private List<Object> beforeData;
+    private List<Object> afterData;
     private Map<String, Object> before;
     private Map<String, Object> after;
+    private boolean updateRowIfInsertFailed;
 
     public RowChangedEvent(int tableGroupIndex, String event, Map<String, Object> before, Map<String, Object> after) {
         this.tableGroupIndex = tableGroupIndex;
@@ -32,6 +33,14 @@ public class RowChangedEvent {
         this.after = after;
     }
 
+    public RowChangedEvent(int tableGroupIndex, String event, Map<String, Object> before, Map<String, Object> after, boolean updateRowIfInsertFailed) {
+        this.tableGroupIndex = tableGroupIndex;
+        this.event = event;
+        this.before = before;
+        this.after = after;
+        this.updateRowIfInsertFailed = updateRowIfInsertFailed;
+    }
+
     public RowChangedEvent(String tableName, String event, List<Object> beforeData, List<Object> afterData) {
         this.tableName = tableName;
         this.event = event;
@@ -79,6 +88,15 @@ public class RowChangedEvent {
         this.after = after;
     }
 
+    public boolean isUpdateRowIfInsertFailed() {
+        return updateRowIfInsertFailed;
+    }
+
+    public RowChangedEvent setUpdateRowIfInsertFailed(boolean updateRowIfInsertFailed) {
+        this.updateRowIfInsertFailed = updateRowIfInsertFailed;
+        return this;
+    }
+
     @Override
     public String toString() {
         return JsonUtil.objToJson(this);

dbsyncer-connector/src/main/java/org/dbsyncer/connector/Connector.java

@@ -54,7 +54,7 @@ public interface Connector<M, C> {
      * @param connectorMapper
      * @return
      */
-    List<String> getTable(M connectorMapper);
+    List<Table> getTable(M connectorMapper);
 
     /**
      * 获取表元信息

dbsyncer-connector/src/main/java/org/dbsyncer/connector/ConnectorFactory.java

@@ -84,7 +84,7 @@ public class ConnectorFactory implements DisposableBean {
      *
      * @return
      */
-    public List<String> getTable(ConnectorMapper config) {
+    public List<Table> getTable(ConnectorMapper config) {
         Assert.notNull(config, "ConnectorMapper can not be null.");
         String type = config.getOriginalConfig().getConnectorType();
         return getConnector(type).getTable(config);

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

@@ -12,12 +12,25 @@ import java.util.List;
  */
 public class MetaInfo {
 
+    /**
+     * 表类型
+     */
+    private String tableType;
     /**
      * 属性字段
      * 格式：[{"name":"ID","typeName":"INT","type":"4"},{"name":"NAME","typeName":"VARCHAR","type":"12"}]
      */
     private List<Field> column;
 
+    public String getTableType() {
+        return tableType;
+    }
+
+    public MetaInfo setTableType(String tableType) {
+        this.tableType = tableType;
+        return this;
+    }
+
     public List<Field> getColumn() {
         return column;
     }
@@ -28,6 +41,6 @@ public class MetaInfo {
     }
     @Override
     public String toString() {
-        return new StringBuilder().append("MetaInfo{").append("column=").append(column).append('}').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/config/Table.java

@@ -1,7 +1,8 @@
 package org.dbsyncer.connector.config;
 
+import org.dbsyncer.connector.enums.TableTypeEnum;
+
 import java.util.List;
-import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * @author AE86
@@ -10,9 +11,16 @@ import java.util.concurrent.atomic.AtomicLong;
  */
 public class Table {
 
-    // 表名
+    /**
+     * 表名
+     */
     private String name;
 
+    /**
+     * 表类型[TABLE、VIEW]
+     */
+    private String type;
+
     /**
      * 属性字段
      * 格式：[{"name":"ID","typeName":"INT","type":"4"},{"name":"NAME","typeName":"VARCHAR","type":"12"}]
@@ -22,6 +30,25 @@ public class Table {
     // 总数
     private long count;
 
+    public Table() {
+    }
+
+    public Table(String name) {
+        this.name = name;
+        this.type = TableTypeEnum.TABLE.getCode();
+    }
+
+    public Table(String name, String type) {
+        this.name = name;
+        this.type = type;
+    }
+
+    public Table(String name, String type, List<Field> column) {
+        this.name = name;
+        this.type = type;
+        this.column = column;
+    }
+
     public String getName() {
         return name;
     }
@@ -31,6 +58,15 @@ public class Table {
         return this;
     }
 
+    public String getType() {
+        return type;
+    }
+
+    public Table setType(String type) {
+        this.type = type;
+        return this;
+    }
+
     public List<Field> getColumn() {
         return column;
     }

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

@@ -21,16 +21,22 @@ public class WriterSingleConfig extends WriterConfig {
     private String table;
 
     /**
-     * 更新失败转insert
+     * 重试标记
      */
     private boolean retry;
 
-    public WriterSingleConfig(List<Field> fields, Map<String, String> command, String event, Map<String, Object> data, String table) {
+    /**
+     * 插入失败转更新
+     */
+    private boolean updateRowIfInsertFailed;
+
+    public WriterSingleConfig(List<Field> fields, Map<String, String> command, String event, Map<String, Object> data, String table, boolean updateRowIfInsertFailed) {
         setCommand(command);
         setFields(fields);
         setData(data);
         setEvent(event);
         setTable(table);
+        this.updateRowIfInsertFailed = updateRowIfInsertFailed;
     }
 
     public Map<String, Object> getData() {
@@ -67,4 +73,13 @@ public class WriterSingleConfig extends WriterConfig {
     public void setRetry(boolean retry) {
         this.retry = retry;
     }
+
+    public boolean isUpdateRowIfInsertFailed() {
+        return updateRowIfInsertFailed;
+    }
+
+    public WriterSingleConfig setUpdateRowIfInsertFailed(boolean updateRowIfInsertFailed) {
+        this.updateRowIfInsertFailed = updateRowIfInsertFailed;
+        return this;
+    }
 }

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

@@ -27,7 +27,7 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector implem
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
-    protected abstract String getTableSql(DatabaseConfig config);
+    protected abstract String getTableSql();
 
     @Override
     public ConnectorMapper connect(DatabaseConfig config) {
@@ -56,9 +56,13 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector implem
     }
 
     @Override
-    public List<String> getTable(DatabaseConnectorMapper connectorMapper) {
-        String sql = getTableSql(connectorMapper.getConfig());
-        return connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForList(sql, String.class));
+    public List<Table> getTable(DatabaseConnectorMapper connectorMapper) {
+        String sql = getTableSql();
+        List<String> tableNames = connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForList(sql, String.class));
+        if(!CollectionUtils.isEmpty(tableNames)){
+            return tableNames.stream().map(name -> new Table(name)).collect(Collectors.toList());
+        }
+        return Collections.EMPTY_LIST;
     }
 
     @Override
@@ -153,10 +157,9 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector implem
             throw new ConnectorException("writer data can not be empty.");
         }
 
-        Field pkField = null;
+        Field pkField = getPrimaryKeyField(fields);
         // Update / Delete
         if (isUpdate(event) || isDelete(event)) {
-            pkField = getPrimaryKeyField(fields);
             if (isDelete(event)) {
                 fields.clear();
             }
@@ -165,10 +168,10 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector implem
 
         int size = fields.size();
         Result result = new Result();
-        int update = 0;
+        int execute = 0;
         try {
             // 2、设置参数
-            update = connectorMapper.execute(databaseTemplate ->
+            execute = connectorMapper.execute(databaseTemplate ->
                     databaseTemplate.update(sql, (ps) -> {
                         Field f = null;
                         for (int i = 0; i < size; i++) {
@@ -179,25 +182,36 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector implem
             );
         } catch (Exception e) {
             // 记录错误数据
-            result.getFailData().add(data);
-            result.getFail().set(1);
-            result.getError().append("SQL:").append(sql).append(System.lineSeparator())
-                    .append("DATA:").append(data).append(System.lineSeparator())
-                    .append("ERROR:").append(e.getMessage()).append(System.lineSeparator());
-            logger.error("SQL:{}, DATA:{}, ERROR:{}", sql, data, e.getMessage());
+            if(!config.isUpdateRowIfInsertFailed()){
+                result.getFailData().add(data);
+                result.getFail().set(1);
+                result.getError().append("SQL:").append(sql).append(System.lineSeparator())
+                        .append("DATA:").append(data).append(System.lineSeparator())
+                        .append("ERROR:").append(e.getMessage()).append(System.lineSeparator());
+                logger.error("SQL:{}, DATA:{}, ERROR:{}", sql, data, e.getMessage());
+            }
         }
 
-        // 更新失败尝试插入
-        if (0 == update && isUpdate(event) && null != pkField && !config.isRetry()) {
-            // 插入前检查有无数据
-            String queryCount = config.getCommand().get(ConnectorConstant.OPERTION_QUERY_COUNT_EXIST);
-            if (!existRow(connectorMapper, queryCount, data.get(pkField.getName()))) {
-                fields.remove(fields.size() - 1);
-                config.setEvent(ConnectorConstant.OPERTION_INSERT);
+        if (0 == execute && !config.isRetry() && null != pkField) {
+            // 不存在转insert
+            if(isUpdate(event)){
+                String queryCount = config.getCommand().get(ConnectorConstant.OPERTION_QUERY_COUNT_EXIST);
+                if(!existRow(connectorMapper, queryCount, data.get(pkField.getName()))){
+                    fields.remove(fields.size() - 1);
+                    config.setEvent(ConnectorConstant.OPERTION_INSERT);
+                    config.setRetry(true);
+                    logger.warn("{}表执行{}失败, 尝试执行{}", config.getTable(), event, config.getEvent());
+                    return writer(connectorMapper, config);
+                }
+                return result;
+            }
+            // 存在转update
+            if(isInsert(event)){
+                config.setEvent(ConnectorConstant.OPERTION_UPDATE);
                 config.setRetry(true);
-                logger.warn("{}表执行{}失败, 尝试执行{}", config.getTable(), event, config.getEvent());
-                result = writer(connectorMapper, config);
+                return writer(connectorMapper, config);
             }
+
         }
         return result;
     }
@@ -223,7 +237,10 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector implem
         if (StringUtil.isNotBlank(queryFilterSql)) {
             queryCount.append(queryFilterSql);
         }
-        queryCount.append(" GROUP BY ").append(pk).append(") DBSYNCER_T");
+        if(!StringUtil.isBlank(pk)){
+            queryCount.append(" GROUP BY ").append(pk);
+        }
+        queryCount.append(") DBSYNCER_T");
         map.put(ConnectorConstant.OPERTION_QUERY_COUNT, queryCount.toString());
         return map;
     }
@@ -260,12 +277,12 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector implem
      * @param config
      * @return
      */
-    protected List<String> getDqlTable(DatabaseConnectorMapper config) {
+    protected List<Table> getDqlTable(DatabaseConnectorMapper config) {
         MetaInfo metaInfo = getDqlMetaInfo(config);
         Assert.notNull(metaInfo, "SQL解析异常.");
         DatabaseConfig cfg = config.getConfig();
-        List<String> tables = new ArrayList<>();
-        tables.add(cfg.getSql());
+        List<Table> tables = new ArrayList<>();
+        tables.add(new Table(cfg.getSql()));
         return tables;
     }
 

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

@@ -0,0 +1,41 @@
+package org.dbsyncer.connector.enums;
+
+/**
+ * 表类型
+ *
+ * @author AE86
+ * @version 1.0.0
+ * @date 2021/08/26 21:13
+ */
+public enum TableTypeEnum {
+
+    /**
+     * 表
+     */
+    TABLE("TABLE"),
+    /**
+     * 视图
+     */
+    VIEW("VIEW");
+
+    private String code;
+
+    TableTypeEnum(String code) {
+        this.code = code;
+    }
+
+    /**
+     * 是否视图类型
+     *
+     * @param type
+     * @return
+     */
+    public static boolean isView(String type) {
+        return VIEW.getCode().equals(type);
+    }
+
+    public String getCode() {
+        return code;
+    }
+
+}

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

@@ -85,14 +85,14 @@ public final class ESConnector extends AbstractConnector implements Connector<ES
     }
 
     @Override
-    public List<String> getTable(ESConnectorMapper connectorMapper) {
+    public List<Table> getTable(ESConnectorMapper connectorMapper) {
         try {
             ESConfig config = connectorMapper.getConfig();
             GetIndexRequest request = new GetIndexRequest(config.getIndex());
             GetIndexResponse indexResponse = connectorMapper.getConnection().indices().get(request, RequestOptions.DEFAULT);
             MappingMetaData mappingMetaData = indexResponse.getMappings().get(config.getIndex());
-            List<String> tables = new ArrayList<>();
-            tables.add(mappingMetaData.type());
+            List<Table> tables = new ArrayList<>();
+            tables.add(new Table(mappingMetaData.type()));
             return tables;
         } catch (IOException e) {
             logger.error(e.getMessage());

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

@@ -1,6 +1,5 @@
 package org.dbsyncer.connector.mysql;
 
-import org.dbsyncer.connector.config.DatabaseConfig;
 import org.dbsyncer.connector.config.PageSqlConfig;
 import org.dbsyncer.connector.constant.DatabaseConstant;
 import org.dbsyncer.connector.database.AbstractDatabaseConnector;
@@ -8,7 +7,7 @@ import org.dbsyncer.connector.database.AbstractDatabaseConnector;
 public final class MysqlConnector extends AbstractDatabaseConnector {
 
     @Override
-    protected String getTableSql(DatabaseConfig config) {
+    protected String getTableSql() {
         return "show tables";
     }
 

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

@@ -1,15 +1,32 @@
 package org.dbsyncer.connector.oracle;
 
-import org.dbsyncer.connector.config.DatabaseConfig;
+import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.connector.config.PageSqlConfig;
+import org.dbsyncer.connector.config.Table;
 import org.dbsyncer.connector.constant.DatabaseConstant;
 import org.dbsyncer.connector.database.AbstractDatabaseConnector;
+import org.dbsyncer.connector.database.DatabaseConnectorMapper;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
 
 public final class OracleConnector extends AbstractDatabaseConnector {
 
     @Override
-    protected String getTableSql(DatabaseConfig config) {
-        return String.format("SELECT TABLE_NAME FROM ALL_TABLES WHERE OWNER='%s'", config.getUsername()).toUpperCase();
+    protected String getTableSql() {
+        return "SELECT TABLE_NAME,TABLE_TYPE FROM USER_TAB_COMMENTS";
+    }
+
+    @Override
+    public List<Table> getTable(DatabaseConnectorMapper connectorMapper) {
+        String sql = getTableSql();
+        List<Map<String, Object>> list = connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForList(sql));
+        if (!CollectionUtils.isEmpty(list)) {
+            return list.stream().map(r -> new Table(r.get("TABLE_NAME").toString(), r.get("TABLE_TYPE").toString())).collect(Collectors.toList());
+        }
+        return Collections.EMPTY_LIST;
     }
 
     @Override
@@ -19,7 +36,7 @@ public final class OracleConnector extends AbstractDatabaseConnector {
 
     @Override
     public Object[] getPageArgs(int pageIndex, int pageSize) {
-        return new Object[] {pageIndex * pageSize, (pageIndex - 1) * pageSize};
+        return new Object[]{pageIndex * pageSize, (pageIndex - 1) * pageSize};
     }
 
     @Override

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

@@ -1,9 +1,6 @@
 package org.dbsyncer.connector.sql;
 
-import org.dbsyncer.connector.config.CommandConfig;
-import org.dbsyncer.connector.config.DatabaseConfig;
-import org.dbsyncer.connector.config.MetaInfo;
-import org.dbsyncer.connector.config.PageSqlConfig;
+import org.dbsyncer.connector.config.*;
 import org.dbsyncer.connector.constant.DatabaseConstant;
 import org.dbsyncer.connector.database.AbstractDatabaseConnector;
 import org.dbsyncer.connector.database.DatabaseConnectorMapper;
@@ -14,7 +11,7 @@ import java.util.Map;
 public final class DQLMysqlConnector extends AbstractDatabaseConnector {
 
     @Override
-    protected String getTableSql(DatabaseConfig config) {
+    protected String getTableSql() {
         return "show tables";
     }
 
@@ -29,7 +26,7 @@ public final class DQLMysqlConnector extends AbstractDatabaseConnector {
     }
 
     @Override
-    public List<String> getTable(DatabaseConnectorMapper config) {
+    public List<Table> getTable(DatabaseConnectorMapper config) {
         return super.getDqlTable(config);
     }
 

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

@@ -11,8 +11,8 @@ import java.util.Map;
 public final class DQLOracleConnector extends AbstractDatabaseConnector {
 
     @Override
-    protected String getTableSql(DatabaseConfig config) {
-        return String.format("SELECT TABLE_NAME FROM ALL_TABLES WHERE OWNER='%s'", config.getUsername()).toUpperCase();
+    protected String getTableSql() {
+        return "SELECT TABLE_NAME,TABLE_TYPE FROM USER_TAB_COMMENTS";
     }
 
     @Override
@@ -26,7 +26,7 @@ public final class DQLOracleConnector extends AbstractDatabaseConnector {
     }
 
     @Override
-    public List<String> getTable(DatabaseConnectorMapper config) {
+    public List<Table> getTable(DatabaseConnectorMapper config) {
         return super.getDqlTable(config);
     }
 

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

@@ -30,7 +30,7 @@ public final class DQLSqlServerConnector extends AbstractDatabaseConnector {
     }
 
     @Override
-    protected String getTableSql(DatabaseConfig config) {
+    protected String getTableSql() {
         return "SELECT NAME FROM SYS.TABLES WHERE SCHEMA_ID = SCHEMA_ID('DBO')";
     }
 
@@ -49,7 +49,7 @@ public final class DQLSqlServerConnector extends AbstractDatabaseConnector {
     }
 
     @Override
-    public List<String> getTable(DatabaseConnectorMapper config) {
+    public List<Table> getTable(DatabaseConnectorMapper config) {
         return super.getDqlTable(config);
     }
 

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

@@ -3,7 +3,10 @@ package org.dbsyncer.connector.sqlserver;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.connector.ConnectorMapper;
-import org.dbsyncer.connector.config.*;
+import org.dbsyncer.connector.config.CommandConfig;
+import org.dbsyncer.connector.config.DatabaseConfig;
+import org.dbsyncer.connector.config.PageSqlConfig;
+import org.dbsyncer.connector.config.Table;
 import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.connector.constant.DatabaseConstant;
 import org.dbsyncer.connector.database.AbstractDatabaseConnector;
@@ -29,7 +32,7 @@ public final class SqlServerConnector extends AbstractDatabaseConnector implemen
     }
 
     @Override
-    protected String getTableSql(DatabaseConfig config) {
+    protected String getTableSql() {
         return "SELECT NAME FROM SYS.TABLES WHERE SCHEMA_ID = SCHEMA_ID('DBO') AND IS_MS_SHIPPED = 0";
     }
 

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

@@ -8,6 +8,7 @@ import org.dbsyncer.connector.config.Field;
 import org.dbsyncer.connector.config.MetaInfo;
 import org.dbsyncer.connector.config.Table;
 import org.dbsyncer.connector.database.DatabaseTemplate;
+import org.dbsyncer.connector.enums.TableTypeEnum;
 import org.springframework.jdbc.support.rowset.ResultSetWrappingSqlRowSet;
 import org.springframework.jdbc.support.rowset.SqlRowSet;
 import org.springframework.jdbc.support.rowset.SqlRowSetMetaData;
@@ -101,7 +102,10 @@ public abstract class DatabaseUtil {
                 }
             }
         }
-        throw new ConnectorException("Table primary key can not be empty.");
+        if(!TableTypeEnum.isView(table.getType())){
+            throw new ConnectorException("Table primary key can not be empty.");
+        }
+        return "";
     }
 
     private static boolean isPk(Map<String, List<String>> tables, String tableName, String name) {

dbsyncer-listener/src/main/java/org/dbsyncer/listener/quartz/AbstractQuartzExtractor.java

@@ -3,6 +3,7 @@ package org.dbsyncer.listener.quartz;
 import org.dbsyncer.common.event.RowChangedEvent;
 import org.dbsyncer.common.model.Result;
 import org.dbsyncer.common.util.CollectionUtils;
+import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.common.util.UUIDUtil;
 import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.config.ReaderConfig;
@@ -105,6 +106,11 @@ public abstract class AbstractQuartzExtractor extends AbstractExtractor implemen
 
             Object event = null;
             for (Map<String, Object> row : data) {
+                if(StringUtil.isBlank(eventFieldName)){
+                    changedEvent(new RowChangedEvent(index, ConnectorConstant.OPERTION_INSERT, Collections.EMPTY_MAP, row, true));
+                    continue;
+                }
+
                 event = row.get(eventFieldName);
                 if (update.contains(event)) {
                     changedEvent(new RowChangedEvent(index, ConnectorConstant.OPERTION_UPDATE, Collections.EMPTY_MAP, row));

dbsyncer-manager/src/main/java/org/dbsyncer/manager/Manager.java

@@ -4,6 +4,7 @@ import org.dbsyncer.common.model.Paging;
 import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.config.ConnectorConfig;
 import org.dbsyncer.connector.config.MetaInfo;
+import org.dbsyncer.connector.config.Table;
 import org.dbsyncer.connector.enums.ConnectorEnum;
 import org.dbsyncer.connector.enums.FilterEnum;
 import org.dbsyncer.connector.enums.OperationEnum;
@@ -33,7 +34,7 @@ public interface Manager extends Executor {
 
     boolean isAliveConnectorConfig(ConnectorConfig config);
 
-    List<String> getTable(ConnectorMapper config);
+    List<Table> getTable(ConnectorMapper config);
 
     MetaInfo getMetaInfo(String connectorId, String tableName);
 

dbsyncer-manager/src/main/java/org/dbsyncer/manager/ManagerFactory.java

@@ -5,6 +5,7 @@ import org.dbsyncer.common.model.Paging;
 import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.config.ConnectorConfig;
 import org.dbsyncer.connector.config.MetaInfo;
+import org.dbsyncer.connector.config.Table;
 import org.dbsyncer.connector.enums.ConnectorEnum;
 import org.dbsyncer.connector.enums.FilterEnum;
 import org.dbsyncer.connector.enums.OperationEnum;
@@ -80,7 +81,7 @@ public class ManagerFactory implements Manager, ApplicationListener<ClosedEvent>
     }
 
     @Override
-    public List<String> getTable(ConnectorMapper config) {
+    public List<Table> getTable(ConnectorMapper config) {
         return parser.getTable(config);
     }
 

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

@@ -5,6 +5,7 @@ import org.dbsyncer.common.model.Task;
 import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.config.ConnectorConfig;
 import org.dbsyncer.connector.config.MetaInfo;
+import org.dbsyncer.connector.config.Table;
 import org.dbsyncer.connector.enums.ConnectorEnum;
 import org.dbsyncer.connector.enums.FilterEnum;
 import org.dbsyncer.connector.enums.OperationEnum;
@@ -54,7 +55,7 @@ public interface Parser {
      * @param config
      * @return
      */
-    List<String> getTable(ConnectorMapper config);
+    List<Table> getTable(ConnectorMapper config);
 
     /**
      * 获取表元信息

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

@@ -25,6 +25,7 @@ import org.dbsyncer.parser.util.ConvertUtil;
 import org.dbsyncer.parser.util.PickerUtil;
 import org.dbsyncer.plugin.PluginFactory;
 import org.dbsyncer.storage.enums.StorageDataStatusEnum;
+import org.json.JSONArray;
 import org.json.JSONException;
 import org.json.JSONObject;
 import org.slf4j.Logger;
@@ -105,14 +106,24 @@ public class ParserFactory implements Parser {
     }
 
     @Override
-    public List<String> getTable(ConnectorMapper config) {
+    public List<Table> getTable(ConnectorMapper config) {
         return connectorFactory.getTable(config);
     }
 
     @Override
     public MetaInfo getMetaInfo(String connectorId, String tableName) {
-        ConnectorMapper connectorMapper = connectorFactory.connect(getConnectorConfig(connectorId));
-        return connectorFactory.getMetaInfo(connectorMapper, tableName);
+        Connector connector = getConnector(connectorId);
+        ConnectorMapper connectorMapper = connectorFactory.connect(connector.getConfig());
+        MetaInfo metaInfo = connectorFactory.getMetaInfo(connectorMapper, tableName);
+        if(!CollectionUtils.isEmpty(connector.getTable())){
+            for(Table t :connector.getTable()){
+                if(t.getName().equals(tableName)){
+                    metaInfo.setTableType(t.getType());
+                    break;
+                }
+            }
+        }
+        return metaInfo;
     }
 
     @Override
@@ -121,8 +132,8 @@ public class ParserFactory implements Parser {
         String tType = getConnectorConfig(mapping.getTargetConnectorId()).getConnectorType();
         Table sourceTable = tableGroup.getSourceTable();
         Table targetTable = tableGroup.getTargetTable();
-        Table sTable = new Table().setName(sourceTable.getName()).setColumn(new ArrayList<>());
-        Table tTable = new Table().setName(targetTable.getName()).setColumn(new ArrayList<>());
+        Table sTable = new Table(sourceTable.getName(), sourceTable.getType(), new ArrayList<>());
+        Table tTable = new Table(targetTable.getName(), targetTable.getType(), new ArrayList<>());
         List<FieldMapping> fieldMapping = tableGroup.getFieldMapping();
         if (!CollectionUtils.isEmpty(fieldMapping)) {
             fieldMapping.forEach(m -> {
@@ -152,12 +163,27 @@ public class ParserFactory implements Parser {
         try {
             JSONObject conn = new JSONObject(json);
             JSONObject config = (JSONObject) conn.remove("config");
+            JSONArray table = (JSONArray) conn.remove("table");
             Connector connector = JsonUtil.jsonToObj(conn.toString(), Connector.class);
             Assert.notNull(connector, "Connector can not be null.");
             String connectorType = config.getString("connectorType");
             Class<?> configClass = ConnectorEnum.getConfigClass(connectorType);
             ConnectorConfig obj = (ConnectorConfig) JsonUtil.jsonToObj(config.toString(), configClass);
             connector.setConfig(obj);
+
+            List<Table> tableList = new ArrayList<>();
+            boolean exist = false;
+            for (int i = 0; i < table.length(); i++) {
+                if(table.get(i) instanceof String){
+                    tableList.add(new Table(table.getString(i)));
+                    exist = true;
+                }
+            }
+            if(!exist){
+                tableList = JsonUtil.jsonToArray(table.toString(), Table.class);
+            }
+            connector.setTable(tableList);
+
             return connector;
         } catch (JSONException e) {
             logger.error(e.getMessage());
@@ -293,7 +319,7 @@ public class ParserFactory implements Parser {
         pluginFactory.convert(tableGroup.getPlugin(), event, data, target);
 
         // 4、写入目标源
-        Result writer = connectorFactory.writer(tConnectorMapper, new WriterSingleConfig(picker.getTargetFields(), tableGroup.getCommand(), event, target, rowChangedEvent.getTableName()));
+        Result writer = connectorFactory.writer(tConnectorMapper, new WriterSingleConfig(picker.getTargetFields(), tableGroup.getCommand(), event, target, rowChangedEvent.getTableName(), rowChangedEvent.isUpdateRowIfInsertFailed()));
 
         // 5、更新结果
         flush(metaId, writer, event, picker.getTargetMapList());
@@ -347,17 +373,28 @@ public class ParserFactory implements Parser {
     }
 
     /**
-     * 获取连接配置
+     * 获取连接器
      *
      * @param connectorId
      * @return
      */
-    private ConnectorConfig getConnectorConfig(String connectorId) {
+    private Connector getConnector(String connectorId) {
         Assert.hasText(connectorId, "Connector id can not be empty.");
         Connector conn = cacheService.get(connectorId, Connector.class);
         Assert.notNull(conn, "Connector can not be null.");
         Connector connector = new Connector();
         BeanUtils.copyProperties(conn, connector);
+        return connector;
+    }
+
+    /**
+     * 获取连接配置
+     *
+     * @param connectorId
+     * @return
+     */
+    private ConnectorConfig getConnectorConfig(String connectorId) {
+        Connector connector = getConnector(connectorId);
         return connector.getConfig();
     }
 

dbsyncer-parser/src/main/java/org/dbsyncer/parser/model/Connector.java

@@ -1,6 +1,7 @@
 package org.dbsyncer.parser.model;
 
 import org.dbsyncer.connector.config.ConnectorConfig;
+import org.dbsyncer.connector.config.Table;
 
 import java.util.List;
 
@@ -14,18 +15,18 @@ public class Connector extends ConfigModel{
     /**
      * 表名,["MY_USER", "T_MY_USER", "table_999"]
      */
-    private List<String> table;
+    private List<Table> table;
 
     /**
      * 连接器配置
      */
     private ConnectorConfig config;
 
-    public List<String> getTable() {
+    public List<Table> getTable() {
         return table;
     }
 
-    public Connector setTable(List<String> table) {
+    public Connector setTable(List<Table> table) {
         this.table = table;
         return this;
     }

dbsyncer-web/src/main/resources/public/mapping/editIncrementQuartz.html

@@ -14,8 +14,8 @@
             <div class="col-md-4">
                 <label class="col-sm-3 control-label text-right">事件</label>
                 <div class="col-sm-9" title="区分增删改">
-                    <select name="incrementStrategyTimingEventFieldName" class="form-control select-control">
-                        <!-- 数据源表公共字段 -->
+                    <select name="incrementStrategyTimingEventFieldName" class="form-control select-control-default">
+                        <option value="" selected="selected">无</option>
                         <option th:each="c,s:${mapping?.sourceColumn}" th:value="${c?.name}" th:text="${c?.name} +' (' + ${c?.typeName} +')'" th:selected="${c.name eq mapping?.listener?.eventFieldName}"/>
                     </select>
                 </div>

dbsyncer-web/src/main/resources/public/mapping/editTable.html

@@ -11,7 +11,7 @@
             <label class="col-sm-3 control-label text-right">数据源表</label>
             <div class="col-sm-9">
                 <select id="sourceTable" class="form-control select-control-table" multiple="multiple">
-                    <option th:each="t,s:${mapping?.sourceConnector?.table}" th:value="${t}" th:text="${t}" />
+                    <option th:each="t,s:${mapping?.sourceConnector?.table}" th:value="${t?.name}" th:text="${t?.name}" />
                 </select>
             </div>
         </div>
@@ -25,7 +25,7 @@
                 <label class="col-sm-3 control-label text-right">目标源表</label>
                 <div class="col-sm-9">
                     <select id="targetTable" class="form-control select-control-table" multiple="multiple">
-                        <option th:each="t,s:${mapping?.targetConnector?.table}" th:value="${t}" th:text="${t}" />
+                        <option th:each="t,s:${mapping?.targetConnector?.table}" th:value="${t?.name}" th:text="${t?.name}" />
                     </select>
                 </div>
             </div>