Merge remote-tracking branch 'origin/V_1.0.0_Beta' into yjwang

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/AbstractChecker.java

@@ -4,7 +4,7 @@ import org.dbsyncer.biz.BizException;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.config.Filter;
+import org.dbsyncer.connector.model.Filter;
 import org.dbsyncer.manager.Manager;
 import org.dbsyncer.parser.model.AbstractConfigModel;
 import org.dbsyncer.parser.model.ConfigModel;

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

@@ -6,7 +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.model.Table;
 import org.dbsyncer.connector.enums.ConnectorEnum;
 import org.dbsyncer.manager.Manager;
 import org.dbsyncer.parser.logger.LogService;

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

@@ -3,9 +3,9 @@ package org.dbsyncer.biz.checker.impl.connector;
 import org.dbsyncer.biz.enums.OracleIncrementEnum;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.config.Field;
-import org.dbsyncer.connector.config.MetaInfo;
-import org.dbsyncer.connector.config.Table;
+import org.dbsyncer.connector.model.Field;
+import org.dbsyncer.connector.model.MetaInfo;
+import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.manager.Manager;
 import org.dbsyncer.parser.model.FieldMapping;
 import org.dbsyncer.parser.model.Mapping;

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

@@ -5,9 +5,9 @@ import org.dbsyncer.biz.checker.AbstractChecker;
 import org.dbsyncer.biz.checker.ConnectorConfigChecker;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.config.Field;
-import org.dbsyncer.connector.config.MetaInfo;
-import org.dbsyncer.connector.config.Table;
+import org.dbsyncer.connector.model.Field;
+import org.dbsyncer.connector.model.MetaInfo;
+import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.manager.Manager;
 import org.dbsyncer.parser.enums.ModelEnum;
 import org.dbsyncer.parser.model.ConfigModel;

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

@@ -4,7 +4,7 @@ import org.dbsyncer.biz.TableGroupService;
 import org.dbsyncer.biz.checker.Checker;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.config.Field;
+import org.dbsyncer.connector.model.Field;
 import org.dbsyncer.parser.logger.LogType;
 import org.dbsyncer.parser.model.Mapping;
 import org.dbsyncer.parser.model.TableGroup;

dbsyncer-common/src/main/java/org/dbsyncer/common/util/DateFormatUtil.java

@@ -1,10 +1,11 @@
 package org.dbsyncer.common.util;
 
+import java.sql.Date;
+import java.sql.Timestamp;
 import java.time.*;
 import java.time.format.*;
 import java.time.temporal.ChronoField;
 import java.time.temporal.TemporalAccessor;
-import java.util.Date;
 
 public abstract class DateFormatUtil {
 
@@ -79,19 +80,17 @@ public abstract class DateFormatUtil {
     }
 
     public static Date stringToDate(String s) {
-        LocalDate localDate = LocalDate.parse(s, DATE_FORMATTER);
-        Instant instant = localDate.atStartOfDay().atZone(zoneId).toInstant();
-        return Date.from(instant);
-    }
-
-    public static LocalDate stringToLocalDate(String s) {
-        return LocalDate.parse(s, DATE_FORMATTER);
+        return Date.valueOf(LocalDate.parse(s, DATE_FORMATTER));
     }
 
     public static LocalTime stringToLocalTime(String s) {
         return LocalTime.parse(s, CHINESE_STANDARD_TIME_FORMATTER);
     }
 
+    public static Timestamp stringToTimestamp(String s) {
+        return Timestamp.valueOf(LocalDateTime.from(TS_FORMAT.parse(s)));
+    }
+
     public static OffsetTime timeWithTimeZone(String s) {
         return OffsetTime.parse(s, TIME_TZ_FORMAT).withOffsetSameInstant(ZoneOffset.UTC);
     }
@@ -106,7 +105,4 @@ public abstract class DateFormatUtil {
         return OffsetDateTime.from(parsedTimestamp).withOffsetSameInstant(ZoneOffset.UTC);
     }
 
-    public static Instant timestampToInstant(String s) {
-        return LocalDateTime.from(TS_FORMAT.parse(s)).toInstant(ZoneOffset.UTC);
-    }
 }

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

@@ -1,7 +1,7 @@
 package org.dbsyncer.connector;
 
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.config.Field;
+import org.dbsyncer.connector.model.Field;
 import org.dbsyncer.connector.constant.ConnectorConstant;
 
 import java.util.List;

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

@@ -2,6 +2,8 @@ package org.dbsyncer.connector;
 
 import org.dbsyncer.common.model.Result;
 import org.dbsyncer.connector.config.*;
+import org.dbsyncer.connector.model.MetaInfo;
+import org.dbsyncer.connector.model.Table;
 
 import java.util.List;
 import java.util.Map;

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

@@ -3,6 +3,8 @@ package org.dbsyncer.connector;
 import org.dbsyncer.common.model.Result;
 import org.dbsyncer.connector.config.*;
 import org.dbsyncer.connector.enums.ConnectorEnum;
+import org.dbsyncer.connector.model.MetaInfo;
+import org.dbsyncer.connector.model.Table;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.util.Assert;
 

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

@@ -23,9 +23,7 @@ public interface ConnectorMapper<K, V> {
 
     K getConfig();
 
-    default V getConnection() throws Exception {
-        throw new ConnectorException("Unsupported method.");
-    }
+    V getConnection() throws Exception;
 
-    default void close() {}
+    void close();
 }

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

@@ -1,5 +1,8 @@
 package org.dbsyncer.connector.config;
 
+import org.dbsyncer.connector.model.Filter;
+import org.dbsyncer.connector.model.Table;
+
 import java.util.List;
 
 /**

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

@@ -0,0 +1,35 @@
+package org.dbsyncer.connector.config;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/5/5 23:19
+ */
+public class FileConfig extends ConnectorConfig {
+
+    /**
+     * 文件目录
+     */
+    private String fileDir;
+
+    /**
+     * 文件描述信息
+     */
+    private String schema;
+
+    public String getFileDir() {
+        return fileDir;
+    }
+
+    public void setFileDir(String fileDir) {
+        this.fileDir = fileDir;
+    }
+
+    public String getSchema() {
+        return schema;
+    }
+
+    public void setSchema(String schema) {
+        this.schema = schema;
+    }
+}

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

@@ -1,22 +1,23 @@
 package org.dbsyncer.connector.config;
 
 import org.dbsyncer.connector.database.Database;
+import org.dbsyncer.connector.model.Field;
 
 import java.util.List;
 
 public class SqlBuilderConfig {
 
-    private Database    database;
+    private Database database;
     // 表名
-    private String      tableName;
+    private String tableName;
     // 主键
-    private String      pk;
+    private String pk;
     // 字段
     private List<Field> fields;
     // 过滤条件
-    private String      queryFilter;
+    private String queryFilter;
     // 引号
-    private String      quotation;
+    private String quotation;
 
     public SqlBuilderConfig(String name, String pk) {
         this.tableName = name;

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

@@ -1,5 +1,7 @@
 package org.dbsyncer.connector.config;
 
+import org.dbsyncer.connector.model.Field;
+
 import java.util.List;
 import java.util.Map;
 

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

@@ -13,6 +13,10 @@ import org.dbsyncer.connector.enums.OperationEnum;
 import org.dbsyncer.connector.enums.SetterEnum;
 import org.dbsyncer.connector.enums.SqlBuilderEnum;
 import org.dbsyncer.connector.enums.TableTypeEnum;
+import org.dbsyncer.connector.model.Field;
+import org.dbsyncer.connector.model.Filter;
+import org.dbsyncer.connector.model.MetaInfo;
+import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.connector.util.DatabaseUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -85,8 +89,7 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector
         Collections.addAll(config.getArgs(), getPageArgs(config.getPageIndex(), config.getPageSize()));
 
         // 3、执行SQL
-        List<Map<String, Object>> list = connectorMapper.execute(
-                databaseTemplate -> databaseTemplate.queryForList(querySql, config.getArgs().toArray()));
+        List<Map<String, Object>> list = connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForList(querySql, config.getArgs().toArray()));
 
         // 4、返回结果集
         return new Result(list);

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

@@ -1,6 +1,6 @@
 package org.dbsyncer.connector.database;
 
-import org.dbsyncer.connector.config.PageSqlConfig;
+import org.dbsyncer.connector.model.PageSql;
 
 public interface Database {
 
@@ -10,7 +10,7 @@ public interface Database {
      * @param config
      * @return
      */
-    String getPageSql(PageSqlConfig config);
+    String getPageSql(PageSql config);
 
     /**
      * 获取分页SQL

dbsyncer-connector/src/main/java/org/dbsyncer/connector/database/setter/DateSetter.java

@@ -1,10 +1,12 @@
 package org.dbsyncer.connector.database.setter;
 
+import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.connector.database.AbstractSetter;
 
 import java.sql.Date;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
+import java.sql.Timestamp;
 
 public class DateSetter extends AbstractSetter<Date> {
 
@@ -13,4 +15,13 @@ public class DateSetter extends AbstractSetter<Date> {
         ps.setDate(i, val);
     }
 
+    @Override
+    protected void setIfValueTypeNotMatch(PreparedFieldMapper mapper, PreparedStatement ps, int i, int type, Object val) throws SQLException {
+        if (val instanceof Timestamp) {
+            Timestamp timestamp = (Timestamp) val;
+            ps.setDate(i, Date.valueOf(timestamp.toLocalDateTime().toLocalDate()));
+            return;
+        }
+        throw new ConnectorException(String.format("DateSetter can not find type [%s], val [%s]", type, val));
+    }
 }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/database/setter/TimestampSetter.java

@@ -2,7 +2,7 @@ package org.dbsyncer.connector.database.setter;
 
 import org.dbsyncer.connector.database.AbstractSetter;
 
-import java.util.Date;
+import java.sql.Date;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
 import java.sql.Timestamp;

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

@@ -1,6 +1,6 @@
 package org.dbsyncer.connector.database.sqlbuilder;
 
-import org.dbsyncer.connector.config.Field;
+import org.dbsyncer.connector.model.Field;
 import org.dbsyncer.connector.config.SqlBuilderConfig;
 import org.dbsyncer.connector.database.AbstractSqlBuilder;
 

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

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.database.sqlbuilder;
 
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.config.Field;
-import org.dbsyncer.connector.config.PageSqlConfig;
+import org.dbsyncer.connector.model.Field;
+import org.dbsyncer.connector.model.PageSql;
 import org.dbsyncer.connector.config.SqlBuilderConfig;
 import org.dbsyncer.connector.database.AbstractSqlBuilder;
 import org.dbsyncer.connector.database.Database;
@@ -20,7 +20,7 @@ public class SqlBuilderQuery extends AbstractSqlBuilder {
     public String buildSql(SqlBuilderConfig config) {
         // 分页语句
         Database database = config.getDatabase();
-        return database.getPageSql(new PageSqlConfig(buildQuerySql(config), config.getPk()));
+        return database.getPageSql(new PageSql(buildQuerySql(config), config.getPk()));
     }
 
     @Override

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

@@ -1,6 +1,6 @@
 package org.dbsyncer.connector.database.sqlbuilder;
 
-import org.dbsyncer.connector.config.Field;
+import org.dbsyncer.connector.model.Field;
 import org.dbsyncer.connector.config.SqlBuilderConfig;
 import org.dbsyncer.connector.database.AbstractSqlBuilder;
 import org.slf4j.Logger;

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

@@ -5,8 +5,10 @@ import org.dbsyncer.connector.Connector;
 import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.connector.config.DatabaseConfig;
 import org.dbsyncer.connector.config.ESConfig;
+import org.dbsyncer.connector.config.FileConfig;
 import org.dbsyncer.connector.config.KafkaConfig;
 import org.dbsyncer.connector.es.ESConnector;
+import org.dbsyncer.connector.file.FileConnector;
 import org.dbsyncer.connector.kafka.KafkaConnector;
 import org.dbsyncer.connector.mysql.MysqlConnector;
 import org.dbsyncer.connector.oracle.OracleConnector;
@@ -50,6 +52,10 @@ public enum ConnectorEnum {
      * Kafka 连接器
      */
     KAFKA("Kafka", new KafkaConnector(), KafkaConfig.class),
+    /**
+     * File 连接器
+     */
+    FILE("File", new FileConnector(), FileConfig.class),
     /**
      * DqlMysql 连接器
      */

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

@@ -3,10 +3,10 @@ package org.dbsyncer.connector.enums;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.ConnectorException;
 
+import java.sql.Date;
 import java.sql.Time;
 import java.sql.Timestamp;
 import java.sql.Types;
-import java.util.Date;
 
 /**
  * Kafka字段类型

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

@@ -13,6 +13,10 @@ import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.connector.enums.ESFieldTypeEnum;
 import org.dbsyncer.connector.enums.FilterEnum;
 import org.dbsyncer.connector.enums.OperationEnum;
+import org.dbsyncer.connector.model.Field;
+import org.dbsyncer.connector.model.Filter;
+import org.dbsyncer.connector.model.MetaInfo;
+import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.connector.util.ESUtil;
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkResponse;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/file/FileConnector.java

@@ -0,0 +1,88 @@
+package org.dbsyncer.connector.file;
+
+import org.dbsyncer.common.model.Result;
+import org.dbsyncer.connector.AbstractConnector;
+import org.dbsyncer.connector.Connector;
+import org.dbsyncer.connector.ConnectorMapper;
+import org.dbsyncer.connector.config.*;
+import org.dbsyncer.connector.model.MetaInfo;
+import org.dbsyncer.connector.model.Table;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/5/5 23:19
+ */
+public final class FileConnector extends AbstractConnector implements Connector<FileConnectorMapper, FileConfig> {
+
+    private final Logger logger = LoggerFactory.getLogger(getClass());
+
+    @Override
+    public ConnectorMapper connect(FileConfig config) {
+        return new FileConnectorMapper(config);
+    }
+
+    @Override
+    public void disconnect(FileConnectorMapper connectorMapper) {
+
+    }
+
+    @Override
+    public boolean isAlive(FileConnectorMapper connectorMapper) {
+        return false;
+    }
+
+    @Override
+    public String getConnectorMapperCacheKey(FileConfig config) {
+        String localIP;
+        try {
+            localIP = InetAddress.getLocalHost().getHostAddress();
+        } catch (UnknownHostException e) {
+            logger.error(e.getMessage());
+            localIP = "127.0.0.1";
+        }
+        return String.format("%s-%s", config.getConnectorType(), localIP, config.getFileDir());
+    }
+
+    @Override
+    public List<Table> getTable(FileConnectorMapper connectorMapper) {
+        return null;
+    }
+
+    @Override
+    public MetaInfo getMetaInfo(FileConnectorMapper connectorMapper, String tableName) {
+        return null;
+    }
+
+    @Override
+    public long getCount(FileConnectorMapper connectorMapper, Map<String, String> command) {
+        return 0;
+    }
+
+    @Override
+    public Result reader(FileConnectorMapper connectorMapper, ReaderConfig config) {
+        return null;
+    }
+
+    @Override
+    public Result writer(FileConnectorMapper connectorMapper, WriterBatchConfig config) {
+        return null;
+    }
+
+    @Override
+    public Map<String, String> getSourceCommand(CommandConfig commandConfig) {
+        return null;
+    }
+
+    @Override
+    public Map<String, String> getTargetCommand(CommandConfig commandConfig) {
+        return null;
+    }
+}

dbsyncer-connector/src/main/java/org/dbsyncer/connector/file/FileConnectorMapper.java

@@ -0,0 +1,35 @@
+package org.dbsyncer.connector.file;
+
+import org.dbsyncer.connector.ConnectorMapper;
+import org.dbsyncer.connector.config.FileConfig;
+
+import java.io.File;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/5/5 23:19
+ */
+public final class FileConnectorMapper implements ConnectorMapper<FileConfig, File> {
+    private FileConfig config;
+    private File file;
+
+    public FileConnectorMapper(FileConfig config) {
+        this.config = config;
+        file = new File(config.getFileDir());
+    }
+
+    public FileConfig getConfig() {
+        return config;
+    }
+
+    @Override
+    public File getConnection() {
+        return file;
+    }
+
+    @Override
+    public void close() {
+
+    }
+}

dbsyncer-connector/src/main/java/org/dbsyncer/connector/kafka/KafkaConnector.java

@@ -8,6 +8,9 @@ import org.dbsyncer.connector.Connector;
 import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.config.*;
+import org.dbsyncer.connector.model.Field;
+import org.dbsyncer.connector.model.MetaInfo;
+import org.dbsyncer.connector.model.Table;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

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

@@ -1,4 +1,4 @@
-package org.dbsyncer.connector.config;
+package org.dbsyncer.connector.model;
 
 import org.dbsyncer.common.util.JsonUtil;
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/config/Filter.java → dbsyncer-connector/src/main/java/org/dbsyncer/connector/model/Filter.java

@@ -1,4 +1,4 @@
-package org.dbsyncer.connector.config;
+package org.dbsyncer.connector.model;
 
 import org.dbsyncer.connector.enums.FilterEnum;
 import org.dbsyncer.connector.enums.OperationEnum;

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

@@ -1,4 +1,4 @@
-package org.dbsyncer.connector.config;
+package org.dbsyncer.connector.model;
 
 import java.util.List;
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/config/PageSqlConfig.java → dbsyncer-connector/src/main/java/org/dbsyncer/connector/model/PageSql.java

@@ -1,12 +1,12 @@
-package org.dbsyncer.connector.config;
+package org.dbsyncer.connector.model;
 
-public class PageSqlConfig {
+public class PageSql {
 
     private String querySql;
 
     private String pk;
 
-    public PageSqlConfig(String querySql, String pk) {
+    public PageSql(String querySql, String pk) {
         this.querySql = querySql;
         this.pk = pk;
     }

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

@@ -1,4 +1,4 @@
-package org.dbsyncer.connector.config;
+package org.dbsyncer.connector.model;
 
 import org.dbsyncer.connector.enums.TableTypeEnum;
 

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

@@ -1,7 +1,7 @@
 package org.dbsyncer.connector.mysql;
 
-import org.dbsyncer.connector.config.PageSqlConfig;
-import org.dbsyncer.connector.config.Table;
+import org.dbsyncer.connector.model.PageSql;
+import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.connector.constant.DatabaseConstant;
 import org.dbsyncer.connector.database.AbstractDatabaseConnector;
 import org.dbsyncer.connector.database.DatabaseConnectorMapper;
@@ -11,7 +11,7 @@ import java.util.List;
 public final class MysqlConnector extends AbstractDatabaseConnector {
 
     @Override
-    public String getPageSql(PageSqlConfig config) {
+    public String getPageSql(PageSql config) {
         return config.getQuerySql() + DatabaseConstant.MYSQL_PAGE_SQL;
     }
 

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

@@ -1,8 +1,8 @@
 package org.dbsyncer.connector.oracle;
 
 import org.dbsyncer.common.util.CollectionUtils;
-import org.dbsyncer.connector.config.PageSqlConfig;
-import org.dbsyncer.connector.config.Table;
+import org.dbsyncer.connector.model.PageSql;
+import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.connector.constant.DatabaseConstant;
 import org.dbsyncer.connector.database.AbstractDatabaseConnector;
 import org.dbsyncer.connector.database.DatabaseConnectorMapper;
@@ -25,7 +25,7 @@ public final class OracleConnector extends AbstractDatabaseConnector {
     }
 
     @Override
-    public String getPageSql(PageSqlConfig config) {
+    public String getPageSql(PageSql config) {
         return DatabaseConstant.ORACLE_PAGE_SQL_START + config.getQuerySql() + DatabaseConstant.ORACLE_PAGE_SQL_END;
     }
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/postgresql/PostgreSQLConnector.java

@@ -2,8 +2,8 @@ package org.dbsyncer.connector.postgresql;
 
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.connector.config.DatabaseConfig;
-import org.dbsyncer.connector.config.PageSqlConfig;
-import org.dbsyncer.connector.config.Table;
+import org.dbsyncer.connector.model.PageSql;
+import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.connector.constant.DatabaseConstant;
 import org.dbsyncer.connector.database.AbstractDatabaseConnector;
 import org.dbsyncer.connector.database.DatabaseConnectorMapper;
@@ -33,7 +33,7 @@ public final class PostgreSQLConnector extends AbstractDatabaseConnector {
     }
 
     @Override
-    public String getPageSql(PageSqlConfig config) {
+    public String getPageSql(PageSql config) {
         return config.getQuerySql() + DatabaseConstant.POSTGRESQL_PAGE_SQL;
     }
 

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

@@ -6,6 +6,10 @@ import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.connector.database.AbstractDatabaseConnector;
 import org.dbsyncer.connector.database.DatabaseConnectorMapper;
 import org.dbsyncer.connector.enums.SqlBuilderEnum;
+import org.dbsyncer.connector.model.Filter;
+import org.dbsyncer.connector.model.MetaInfo;
+import org.dbsyncer.connector.model.PageSql;
+import org.dbsyncer.connector.model.Table;
 
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -66,7 +70,7 @@ public abstract class AbstractDQLConnector extends AbstractDatabaseConnector {
         }
         String quotation = buildSqlWithQuotation();
         String pk = findTablePrimaryKey(commandConfig.getOriginalTable(), quotation);
-        map.put(SqlBuilderEnum.QUERY.getName(), getPageSql(new PageSqlConfig(querySql, pk)));
+        map.put(SqlBuilderEnum.QUERY.getName(), getPageSql(new PageSql(querySql, pk)));
 
         // 获取查询总数SQL
         StringBuilder queryCount = new StringBuilder();

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

@@ -1,7 +1,7 @@
 package org.dbsyncer.connector.sql;
 
 import org.dbsyncer.connector.config.CommandConfig;
-import org.dbsyncer.connector.config.PageSqlConfig;
+import org.dbsyncer.connector.model.PageSql;
 import org.dbsyncer.connector.constant.DatabaseConstant;
 
 import java.util.Map;
@@ -9,7 +9,7 @@ import java.util.Map;
 public final class DQLMysqlConnector extends AbstractDQLConnector {
 
     @Override
-    public String getPageSql(PageSqlConfig config) {
+    public String getPageSql(PageSql config) {
         return config.getQuerySql() + DatabaseConstant.MYSQL_PAGE_SQL;
     }
 

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

@@ -1,12 +1,12 @@
 package org.dbsyncer.connector.sql;
 
-import org.dbsyncer.connector.config.PageSqlConfig;
+import org.dbsyncer.connector.model.PageSql;
 import org.dbsyncer.connector.constant.DatabaseConstant;
 
 public final class DQLOracleConnector extends AbstractDQLConnector {
 
     @Override
-    public String getPageSql(PageSqlConfig config) {
+    public String getPageSql(PageSql config) {
         return DatabaseConstant.ORACLE_PAGE_SQL_START + config.getQuerySql() + DatabaseConstant.ORACLE_PAGE_SQL_END;
     }
 

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

@@ -1,12 +1,12 @@
 package org.dbsyncer.connector.sql;
 
-import org.dbsyncer.connector.config.PageSqlConfig;
+import org.dbsyncer.connector.model.PageSql;
 import org.dbsyncer.connector.constant.DatabaseConstant;
 
 public final class DQLPostgreSQLConnector extends AbstractDQLConnector {
 
     @Override
-    public String getPageSql(PageSqlConfig config) {
+    public String getPageSql(PageSql config) {
         return config.getQuerySql() + DatabaseConstant.POSTGRESQL_PAGE_SQL;
     }
 

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

@@ -2,7 +2,7 @@ package org.dbsyncer.connector.sql;
 
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.config.PageSqlConfig;
+import org.dbsyncer.connector.model.PageSql;
 import org.dbsyncer.connector.constant.DatabaseConstant;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -12,7 +12,7 @@ public final class DQLSqlServerConnector extends AbstractDQLConnector {
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
     @Override
-    public String getPageSql(PageSqlConfig config) {
+    public String getPageSql(PageSql config) {
         if (StringUtil.isBlank(config.getPk())) {
             logger.error("Table primary key can not be empty.");
             throw new ConnectorException("Table primary key can not be empty.");

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

@@ -4,8 +4,8 @@ import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.ConnectorException;
 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.model.PageSql;
+import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.connector.constant.DatabaseConstant;
 import org.dbsyncer.connector.database.AbstractDatabaseConnector;
@@ -28,7 +28,7 @@ public final class SqlServerConnector extends AbstractDatabaseConnector {
     }
 
     @Override
-    public String getPageSql(PageSqlConfig config) {
+    public String getPageSql(PageSql config) {
         if (StringUtil.isBlank(config.getPk())) {
             logger.error("Table primary key can not be empty.");
             throw new ConnectorException("Table primary key can not be empty.");

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

@@ -77,7 +77,7 @@ public abstract class AbstractExtractor implements Extractor {
         try {
             TimeUnit.MILLISECONDS.sleep(timeout);
         } catch (InterruptedException e) {
-            logger.error(e.getMessage());
+            logger.info(e.getMessage());
         }
     }
 

dbsyncer-listener/src/main/java/org/dbsyncer/listener/postgresql/AbstractMessageDecoder.java

@@ -1,8 +1,11 @@
 package org.dbsyncer.listener.postgresql;
 
 import org.dbsyncer.connector.config.DatabaseConfig;
+import org.dbsyncer.listener.postgresql.column.ColumnValue;
+import org.dbsyncer.listener.postgresql.column.PgColumnValue;
 import org.dbsyncer.listener.postgresql.enums.MessageTypeEnum;
 import org.postgresql.replication.LogSequenceNumber;
+import org.postgresql.util.PGmoney;
 
 import java.nio.ByteBuffer;
 
@@ -15,6 +18,8 @@ public abstract class AbstractMessageDecoder implements MessageDecoder {
 
     protected DatabaseConfig config;
 
+    private ColumnValue value = new PgColumnValue();
+
     @Override
     public boolean skipMessage(ByteBuffer buffer, LogSequenceNumber startLsn, LogSequenceNumber lastReceiveLsn) {
         if (null == lastReceiveLsn || lastReceiveLsn.asLong() == 0 || startLsn.equals(lastReceiveLsn)) {
@@ -51,4 +56,146 @@ public abstract class AbstractMessageDecoder implements MessageDecoder {
     public void setConfig(DatabaseConfig config) {
         this.config = config;
     }
+
+    /**
+     * Resolve the value of a {@link ColumnValue}.
+     *
+     * @param typeName
+     * @param columnValue
+     * @return
+     */
+    protected Object resolveValue(String typeName, String columnValue) {
+        value.setValue(columnValue);
+
+        if (value.isNull()) {
+            // nulls are null
+            return null;
+        }
+
+        switch (typeName) {
+            // include all types from https://www.postgresql.org/docs/current/static/datatype.html#DATATYPE-TABLE
+            case "boolean":
+            case "bool":
+                return value.asBoolean();
+
+            case "integer":
+            case "int":
+            case "int4":
+            case "smallint":
+            case "int2":
+            case "smallserial":
+            case "serial":
+            case "serial2":
+            case "serial4":
+                return value.asInteger();
+
+            case "bigint":
+            case "bigserial":
+            case "int8":
+            case "oid":
+                return value.asLong();
+
+            case "real":
+            case "float4":
+                return value.asFloat();
+
+            case "double precision":
+            case "float8":
+                return value.asDouble();
+
+            case "numeric":
+            case "decimal":
+                return value.asDecimal();
+
+            case "character":
+            case "char":
+            case "character varying":
+            case "varchar":
+            case "bpchar":
+            case "text":
+            case "hstore":
+                return value.asString();
+
+            case "date":
+                return value.asDate();
+
+            case "timestamp with time zone":
+            case "timestamptz":
+                return value.asOffsetDateTimeAtUtc();
+
+            case "timestamp":
+            case "timestamp without time zone":
+                return value.asTimestamp();
+
+            case "time":
+                return value.asTime();
+
+            case "time without time zone":
+                return value.asLocalTime();
+
+            case "time with time zone":
+            case "timetz":
+                return value.asOffsetTimeUtc();
+
+            case "bytea":
+                return value.asByteArray();
+
+            // these are all PG-specific types and we use the JDBC representations
+            // note that, with the exception of point, no converters for these types are implemented yet,
+            // i.e. those values won't actually be propagated to the outbound message until that's the case
+            case "box":
+                return value.asBox();
+            case "circle":
+                return value.asCircle();
+            case "interval":
+                return value.asInterval();
+            case "line":
+                return value.asLine();
+            case "lseg":
+                return value.asLseg();
+            case "money":
+                final Object v = value.asMoney();
+                return (v instanceof PGmoney) ? ((PGmoney) v).val : v;
+            case "path":
+                return value.asPath();
+            case "point":
+                return value.asPoint();
+            case "polygon":
+                return value.asPolygon();
+
+            // PostGIS types are HexEWKB strings
+            // ValueConverter turns them into the correct types
+            case "geometry":
+            case "geography":
+            case "citext":
+            case "bit":
+            case "bit varying":
+            case "varbit":
+            case "json":
+            case "jsonb":
+            case "xml":
+            case "uuid":
+            case "tsrange":
+            case "tstzrange":
+            case "daterange":
+            case "inet":
+            case "cidr":
+            case "macaddr":
+            case "macaddr8":
+            case "int4range":
+            case "numrange":
+            case "int8range":
+                return value.asString();
+
+            // catch-all for other known/builtin PG types
+            case "pg_lsn":
+            case "tsquery":
+            case "tsvector":
+            case "txid_snapshot":
+                // catch-all for unknown (extension module/custom) types
+            default:
+                return null;
+        }
+
+    }
 }

dbsyncer-listener/src/main/java/org/dbsyncer/listener/postgresql/MessageDecoder.java

@@ -1,6 +1,7 @@
 package org.dbsyncer.listener.postgresql;
 
 import org.dbsyncer.common.event.RowChangedEvent;
+import org.dbsyncer.connector.ConnectorFactory;
 import org.dbsyncer.connector.config.DatabaseConfig;
 import org.dbsyncer.connector.database.DatabaseConnectorMapper;
 import org.postgresql.replication.LogSequenceNumber;
@@ -15,7 +16,7 @@ import java.nio.ByteBuffer;
  */
 public interface MessageDecoder {
 
-    default void postProcessBeforeInitialization(DatabaseConnectorMapper connectorMapper) {
+    default void postProcessBeforeInitialization(ConnectorFactory connectorFactory, DatabaseConnectorMapper connectorMapper) {
     }
 
     boolean skipMessage(ByteBuffer buffer, LogSequenceNumber startLsn, LogSequenceNumber lastReceiveLsn);

dbsyncer-listener/src/main/java/org/dbsyncer/listener/postgresql/PostgreSQLExtractor.java

@@ -40,6 +40,7 @@ public class PostgreSQLExtractor extends AbstractExtractor {
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
     private static final String GET_SLOT = "select count(1) from pg_replication_slots where database = ? and slot_name = ? and plugin = ?";
+    private static final String GET_RESTART_LSN = "select restart_lsn from pg_replication_slots where database = ? and slot_name = ? and plugin = ?";
     private static final String GET_ROLE = "SELECT r.rolcanlogin AS login, r.rolreplication AS replication, CAST(array_position(ARRAY(SELECT b.rolname FROM pg_catalog.pg_auth_members m JOIN pg_catalog.pg_roles b ON (m.roleid = b.oid) WHERE m.member = r.oid), 'rds_superuser') AS BOOL) IS TRUE AS superuser, CAST(array_position(ARRAY(SELECT b.rolname FROM pg_catalog.pg_auth_members m JOIN pg_catalog.pg_roles b ON (m.roleid = b.oid) WHERE m.member = r.oid), 'rdsadmin') AS BOOL) IS TRUE AS admin, CAST(array_position(ARRAY(SELECT b.rolname FROM pg_catalog.pg_auth_members m JOIN pg_catalog.pg_roles b ON (m.roleid = b.oid) WHERE m.member = r.oid), 'rdsrepladmin') AS BOOL) IS TRUE AS rep_admin FROM pg_roles r WHERE r.rolname = current_user";
     private static final String GET_DATABASE = "SELECT current_database()";
     private static final String GET_WAL_LEVEL = "SHOW WAL_LEVEL";
@@ -57,6 +58,7 @@ public class PostgreSQLExtractor extends AbstractExtractor {
     private MessageDecoder messageDecoder;
     private Worker worker;
     private LogSequenceNumber startLsn;
+    private String database;
 
     @Override
     public void start() {
@@ -88,9 +90,10 @@ public class PostgreSQLExtractor extends AbstractExtractor {
                 throw new ListenerException(String.format("Postgres roles LOGIN and REPLICATION are not assigned to user: %s", config.getUsername()));
             }
 
+            database = connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForObject(GET_DATABASE, String.class));
             messageDecoder = MessageDecoderEnum.getMessageDecoder(config.getProperty(PLUGIN_NAME));
             messageDecoder.setConfig(config);
-            messageDecoder.postProcessBeforeInitialization(connectorMapper);
+            messageDecoder.postProcessBeforeInitialization(connectorFactory, connectorMapper);
             dropSlotOnClose = BooleanUtil.toBoolean(config.getProperty(DROP_SLOT_ON_CLOSE, "true"));
 
             connect();
@@ -144,20 +147,7 @@ public class PostgreSQLExtractor extends AbstractExtractor {
         sleepInMills(10L);
     }
 
-    private LogSequenceNumber readLastLsn() throws SQLException {
-        if (!snapshot.containsKey(LSN_POSITION)) {
-            LogSequenceNumber lsn = currentXLogLocation();
-            if (null == lsn || lsn.asLong() == 0) {
-                throw new ListenerException("No maximum LSN recorded in the database");
-            }
-            snapshot.put(LSN_POSITION, lsn.asString());
-        }
-
-        return LogSequenceNumber.valueOf(snapshot.get(LSN_POSITION));
-    }
-
     private void createReplicationStream(PGConnection pgConnection) throws SQLException {
-        this.startLsn = readLastLsn();
         ChainedLogicalStreamBuilder streamBuilder = pgConnection
                 .getReplicationAPI()
                 .replicationStream()
@@ -171,7 +161,6 @@ public class PostgreSQLExtractor extends AbstractExtractor {
     }
 
     private void createReplicationSlot(PGConnection pgConnection) throws SQLException {
-        String database = connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForObject(GET_DATABASE, String.class));
         String slotName = messageDecoder.getSlotName();
         String plugin = messageDecoder.getOutputPlugin();
         boolean existSlot = connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForObject(GET_SLOT, new Object[]{database, slotName, plugin}, Integer.class) > 0);
@@ -182,7 +171,20 @@ public class PostgreSQLExtractor extends AbstractExtractor {
                     .withSlotName(slotName)
                     .withOutputPlugin(plugin)
                     .make();
+
+            // wait for create replication slot to have finished
+            sleepInMills(300);
         }
+
+        if (!snapshot.containsKey(LSN_POSITION)) {
+            LogSequenceNumber lsn = connectorMapper.execute(databaseTemplate -> LogSequenceNumber.valueOf(databaseTemplate.queryForObject(GET_RESTART_LSN, new Object[] {database, slotName, plugin}, String.class)));
+            if (null == lsn || lsn.asLong() == 0) {
+                throw new ListenerException("No maximum LSN recorded in the database");
+            }
+            snapshot.put(LSN_POSITION, lsn.asString());
+        }
+
+        this.startLsn = LogSequenceNumber.valueOf(snapshot.get(LSN_POSITION));
     }
 
     private void dropReplicationSlot() {
@@ -222,12 +224,6 @@ public class PostgreSQLExtractor extends AbstractExtractor {
         }
     }
 
-    private LogSequenceNumber currentXLogLocation() throws SQLException {
-        int majorVersion = connection.getMetaData().getDatabaseMajorVersion();
-        String sql = majorVersion >= 10 ? "select * from pg_current_wal_lsn()" : "select * from pg_current_xlog_location()";
-        return connectorMapper.execute(databaseTemplate -> LogSequenceNumber.valueOf(databaseTemplate.queryForObject(sql, String.class)));
-    }
-
     private void recover() {
         connectLock.lock();
         try {
@@ -283,13 +279,12 @@ public class PostgreSQLExtractor extends AbstractExtractor {
                     flushLsn(lsn);
                     // process decoder
                     changedEvent(messageDecoder.processMessage(msg));
-                    forceFlushEvent();
 
                     // feedback
                     stream.setAppliedLSN(lsn);
                     stream.setFlushedLSN(lsn);
                     stream.forceUpdateStatus();
-                } catch (IllegalStateException e) {
+                } catch (IllegalStateException | ListenerException e) {
                     logger.error(e.getMessage());
                 } catch (Exception e) {
                     logger.error(e.getMessage());

dbsyncer-listener/src/main/java/org/dbsyncer/listener/postgresql/column/AbstractColumnValue.java

@@ -10,7 +10,12 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.sql.SQLException;
-import java.time.*;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.OffsetDateTime;
+import java.time.OffsetTime;
+import java.time.ZoneOffset;
 import java.util.concurrent.TimeUnit;
 
 public abstract class AbstractColumnValue implements ColumnValue {
@@ -18,8 +23,8 @@ public abstract class AbstractColumnValue implements ColumnValue {
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
     @Override
-    public LocalDate asLocalDate() {
-        return DateFormatUtil.stringToLocalDate(asString());
+    public Date asDate() {
+        return DateFormatUtil.stringToDate(asString());
     }
 
     @Override
@@ -48,13 +53,13 @@ public abstract class AbstractColumnValue implements ColumnValue {
     }
 
     @Override
-    public Instant asInstant() {
+    public Timestamp asTimestamp() {
         if ("infinity".equals(asString())) {
-            return toInstantFromMicros(PGStatement.DATE_POSITIVE_INFINITY);
+            return Timestamp.from(toInstantFromMicros(PGStatement.DATE_POSITIVE_INFINITY));
         } else if ("-infinity".equals(asString())) {
-            return toInstantFromMicros(PGStatement.DATE_NEGATIVE_INFINITY);
+            return Timestamp.from(toInstantFromMicros(PGStatement.DATE_NEGATIVE_INFINITY));
         }
-        return DateFormatUtil.timestampToInstant(asString());
+        return DateFormatUtil.stringToTimestamp(asString());
     }
 
     @Override

dbsyncer-listener/src/main/java/org/dbsyncer/listener/postgresql/column/ColumnValue.java

@@ -3,8 +3,8 @@ package org.dbsyncer.listener.postgresql.column;
 import org.postgresql.geometric.*;
 import org.postgresql.util.PGmoney;
 
-import java.time.Instant;
-import java.time.LocalDate;
+import java.sql.Date;
+import java.sql.Timestamp;
 import java.time.OffsetDateTime;
 import java.time.OffsetTime;
 
@@ -16,6 +16,8 @@ import java.time.OffsetTime;
  */
 public interface ColumnValue {
 
+    void setValue(String value);
+
     boolean isNull();
 
     String asString();
@@ -32,11 +34,11 @@ public interface ColumnValue {
 
     Object asDecimal();
 
-    LocalDate asLocalDate();
+    Date asDate();
 
     OffsetDateTime asOffsetDateTimeAtUtc();
 
-    Instant asInstant();
+    Timestamp asTimestamp();
 
     Object asTime();
 

dbsyncer-listener/src/main/java/org/dbsyncer/listener/postgresql/column/ColumnValueResolver.java

@@ -1,158 +0,0 @@
-package org.dbsyncer.listener.postgresql.column;
-
-import org.postgresql.util.PGmoney;
-
-/**
- * @author AE86
- * @version 1.0.0
- * @date 2022/4/23 22:45
- */
-public class ColumnValueResolver {
-
-    /**
-     * Resolve the value of a {@link ColumnValue}.
-     *
-     * @param type
-     * @param value
-     * @return
-     */
-    public Object resolveValue(String type, ColumnValue value) {
-        if (value.isNull()) {
-            // nulls are null
-            return null;
-        }
-
-        switch (type) {
-            // include all types from https://www.postgresql.org/docs/current/static/datatype.html#DATATYPE-TABLE
-            // plus aliases from the shorter names produced by older wal2json
-            case "boolean":
-            case "bool":
-                return value.asBoolean();
-
-            case "hstore":
-                return value.asString();
-
-            case "integer":
-            case "int":
-            case "int4":
-            case "smallint":
-            case "int2":
-            case "smallserial":
-            case "serial":
-            case "serial2":
-            case "serial4":
-                return value.asInteger();
-
-            case "bigint":
-            case "bigserial":
-            case "int8":
-            case "oid":
-                return value.asLong();
-
-            case "real":
-            case "float4":
-                return value.asFloat();
-
-            case "double precision":
-            case "float8":
-                return value.asDouble();
-
-            case "numeric":
-            case "decimal":
-                return value.asDecimal();
-
-            case "character":
-            case "char":
-            case "character varying":
-            case "varchar":
-            case "bpchar":
-            case "text":
-                return value.asString();
-
-            case "date":
-                return value.asLocalDate();
-
-            case "timestamp with time zone":
-            case "timestamptz":
-                return value.asOffsetDateTimeAtUtc();
-
-            case "timestamp":
-            case "timestamp without time zone":
-                return value.asInstant();
-
-            case "time":
-                return value.asTime();
-
-            case "time without time zone":
-                return value.asLocalTime();
-
-            case "time with time zone":
-            case "timetz":
-                return value.asOffsetTimeUtc();
-
-            case "bytea":
-                return value.asByteArray();
-
-            // these are all PG-specific types and we use the JDBC representations
-            // note that, with the exception of point, no converters for these types are implemented yet,
-            // i.e. those values won't actually be propagated to the outbound message until that's the case
-            case "box":
-                return value.asBox();
-            case "circle":
-                return value.asCircle();
-            case "interval":
-                return value.asInterval();
-            case "line":
-                return value.asLine();
-            case "lseg":
-                return value.asLseg();
-            case "money":
-                final Object v = value.asMoney();
-                return (v instanceof PGmoney) ? ((PGmoney) v).val : v;
-            case "path":
-                return value.asPath();
-            case "point":
-                return value.asPoint();
-            case "polygon":
-                return value.asPolygon();
-
-            // PostGIS types are HexEWKB strings
-            // ValueConverter turns them into the correct types
-            case "geometry":
-            case "geography":
-                return value.asString();
-
-            case "citext":
-            case "bit":
-            case "bit varying":
-            case "varbit":
-            case "json":
-            case "jsonb":
-            case "xml":
-            case "uuid":
-            case "tsrange":
-            case "tstzrange":
-            case "daterange":
-            case "inet":
-            case "cidr":
-            case "macaddr":
-            case "macaddr8":
-            case "int4range":
-            case "numrange":
-            case "int8range":
-                return value.asString();
-
-            // catch-all for other known/builtin PG types
-            // TODO: improve with more specific/useful classes here?
-            case "pg_lsn":
-            case "tsquery":
-            case "tsvector":
-            case "txid_snapshot":
-                // catch-all for unknown (extension module/custom) types
-            default:
-                return null;
-        }
-
-    }
-
-}

dbsyncer-listener/src/main/java/org/dbsyncer/listener/postgresql/column/TestDecodingColumnValue.java → dbsyncer-listener/src/main/java/org/dbsyncer/listener/postgresql/column/PgColumnValue.java

@@ -1,59 +1,59 @@
-package org.dbsyncer.listener.postgresql.column;
-
-import org.dbsyncer.common.util.StringUtil;
-
-import java.math.BigDecimal;
-
-public final class TestDecodingColumnValue extends AbstractColumnValue {
-
-    private String value;
-
-    public TestDecodingColumnValue(String value) {
-        this.value = value;
-    }
-
-    @Override
-    public boolean isNull() {
-        return value == null;
-    }
-
-    @Override
-    public String asString() {
-        return value;
-    }
-
-    @Override
-    public Boolean asBoolean() {
-        return "t".equalsIgnoreCase(value);
-    }
-
-    @Override
-    public Integer asInteger() {
-        return Integer.valueOf(value);
-    }
-
-    @Override
-    public Long asLong() {
-        return Long.valueOf(value);
-    }
-
-    @Override
-    public Float asFloat() {
-        return Float.valueOf(value);
-    }
-
-    @Override
-    public Double asDouble() {
-        return Double.valueOf(value);
-    }
-
-    @Override
-    public Object asDecimal() {
-        return new BigDecimal(value);
-    }
-
-    @Override
-    public byte[] asByteArray() {
-        return StringUtil.hexStringToByteArray(value.substring(2));
-    }
+package org.dbsyncer.listener.postgresql.column;
+
+import org.dbsyncer.common.util.StringUtil;
+
+import java.math.BigDecimal;
+
+public final class PgColumnValue extends AbstractColumnValue {
+
+    private String value;
+
+    public void setValue(String value) {
+        this.value = value;
+    }
+
+    @Override
+    public boolean isNull() {
+        return value == null;
+    }
+
+    @Override
+    public String asString() {
+        return value;
+    }
+
+    @Override
+    public Boolean asBoolean() {
+        return "t".equalsIgnoreCase(value);
+    }
+
+    @Override
+    public Integer asInteger() {
+        return Integer.valueOf(value);
+    }
+
+    @Override
+    public Long asLong() {
+        return Long.valueOf(value);
+    }
+
+    @Override
+    public Float asFloat() {
+        return Float.valueOf(value);
+    }
+
+    @Override
+    public Double asDouble() {
+        return Double.valueOf(value);
+    }
+
+    @Override
+    public Object asDecimal() {
+        return new BigDecimal(value);
+    }
+
+    @Override
+    public byte[] asByteArray() {
+        return StringUtil.hexStringToByteArray(value.substring(2));
+    }
 }

dbsyncer-listener/src/main/java/org/dbsyncer/listener/postgresql/decoder/PgOutputMessageDecoder.java

@@ -2,21 +2,20 @@ package org.dbsyncer.listener.postgresql.decoder;
 
 import org.dbsyncer.common.event.RowChangedEvent;
 import org.dbsyncer.common.util.CollectionUtils;
-import org.dbsyncer.connector.constant.ConnectorConstant;
+import org.dbsyncer.connector.ConnectorFactory;
+import org.dbsyncer.connector.model.Field;
+import org.dbsyncer.connector.model.MetaInfo;
 import org.dbsyncer.connector.database.DatabaseConnectorMapper;
 import org.dbsyncer.listener.ListenerException;
 import org.dbsyncer.listener.postgresql.AbstractMessageDecoder;
 import org.dbsyncer.listener.postgresql.enums.MessageDecoderEnum;
 import org.dbsyncer.listener.postgresql.enums.MessageTypeEnum;
-import org.postgresql.jdbc.PgResultSet;
 import org.postgresql.replication.fluent.logical.ChainedLogicalStreamBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.util.Assert;
 
 import java.nio.ByteBuffer;
-import java.sql.Connection;
-import java.sql.DatabaseMetaData;
-import java.sql.ResultSet;
 import java.time.LocalDateTime;
 import java.util.*;
 
@@ -30,15 +29,17 @@ public class PgOutputMessageDecoder extends AbstractMessageDecoder {
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
     private static final LocalDateTime PG_EPOCH = LocalDateTime.of(2000, 1, 1, 0, 0, 0);
-
-    private static final String GET_TABLE_SCHEMA = "select oid,relname as tableName from pg_class t inner join (select ns.oid as nspoid, ns.nspname from pg_namespace ns where ns.nspname = (select (current_schemas(false))[s.r] from generate_series(1, array_upper(current_schemas(false), 1)) as s(r))) as n on n.nspoid = t.relnamespace";
-
+    private static final String GET_TABLE_SCHEMA = "select oid,relname as tableName from pg_class t inner join (select ns.oid as nspoid, ns.nspname from pg_namespace ns where ns.nspname = (select (current_schemas(false))[s.r] from generate_series(1, array_upper(current_schemas(false), 1)) as s(r))) as n on n.nspoid = t.relnamespace where relkind = 'r'";
     private static final Map<Integer, TableId> tables = new LinkedHashMap<>();
+    private ConnectorFactory connectorFactory;
+    private DatabaseConnectorMapper connectorMapper;
 
     @Override
-    public void postProcessBeforeInitialization(DatabaseConnectorMapper connectorMapper) {
-        initPublication(connectorMapper);
-        readSchema(connectorMapper);
+    public void postProcessBeforeInitialization(ConnectorFactory connectorFactory, DatabaseConnectorMapper connectorMapper) {
+        this.connectorFactory = connectorFactory;
+        this.connectorMapper = connectorMapper;
+        initPublication();
+        readSchema();
     }
 
     @Override
@@ -47,20 +48,12 @@ public class PgOutputMessageDecoder extends AbstractMessageDecoder {
             throw new IllegalStateException("Invalid buffer received from PG server during streaming replication");
         }
 
-        RowChangedEvent event = null;
         MessageTypeEnum type = MessageTypeEnum.getType((char) buffer.get());
         switch (type) {
             case UPDATE:
-                event = parseUpdate(buffer);
-                break;
-
             case INSERT:
-                event = parseInsert(buffer);
-                break;
-
             case DELETE:
-                event = parseDelete(buffer);
-                break;
+                return parseData(type, buffer);
 
             case BEGIN:
                 long beginLsn = buffer.getLong();
@@ -81,10 +74,6 @@ public class PgOutputMessageDecoder extends AbstractMessageDecoder {
                 logger.info("Type {} not implemented", type.name());
         }
 
-        if (null != event) {
-            logger.info(event.toString());
-        }
-
         return null;
     }
 
@@ -103,7 +92,7 @@ public class PgOutputMessageDecoder extends AbstractMessageDecoder {
         return String.format("dbs_pub_%s_%s", config.getSchema(), config.getUsername());
     }
 
-    private void initPublication(DatabaseConnectorMapper connectorMapper) {
+    private void initPublication() {
         String pubName = getPubName();
         String selectPublication = String.format("SELECT COUNT(1) FROM pg_publication WHERE pubname = '%s'", pubName);
         Integer count = connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForObject(selectPublication, Integer.class));
@@ -124,116 +113,77 @@ public class PgOutputMessageDecoder extends AbstractMessageDecoder {
         }
     }
 
-    private void readSchema(DatabaseConnectorMapper connectorMapper) {
+    private void readSchema() {
         List<Map> schemas = connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForList(GET_TABLE_SCHEMA));
         if (!CollectionUtils.isEmpty(schemas)) {
             schemas.forEach(map -> {
                 Long oid = (Long) map.get("oid");
                 String tableName = (String) map.get("tableName");
-                tables.put(oid.intValue(), new TableId(oid.intValue(), tableName));
+                MetaInfo metaInfo = connectorFactory.getMetaInfo(connectorMapper, tableName);
+                Assert.notEmpty(metaInfo.getColumn(), String.format("The table column for '%s' must not be empty.", tableName));
+                tables.put(oid.intValue(), new TableId(oid.intValue(), tableName, metaInfo.getColumn()));
             });
         }
-
-        try {
-            Connection connection = connectorMapper.getConnection();
-            DatabaseMetaData metaData = connection.getMetaData();
-
-            // read column
-            for (TableId tableId : tables.values()) {
-                ResultSet rs = metaData.getColumns("", config.getSchema(), tableId.tableName, null);
-                PgResultSet pgResultSet = (PgResultSet) rs;
-                while (pgResultSet.next()) {
-                    pgResultSet.getRow();
-                }
-            }
-        } catch (Exception e) {
-            throw new ListenerException(e.getCause());
-        }
-    }
-
-    private RowChangedEvent parseDelete(ByteBuffer buffer) {
-        int relationId = buffer.getInt();
-        logger.info("Delete table {}", tables.get(relationId).tableName);
-
-        List<Object> data = new ArrayList<>();
-        String newTuple = new String(new byte[]{buffer.get()}, 0, 1);
-
-        switch (newTuple) {
-            case "K":
-                readTupleData(buffer, data);
-                break;
-            default:
-                logger.info("K not set, got instead {}", newTuple);
-        }
-        return new RowChangedEvent(tables.get(relationId).tableName, ConnectorConstant.OPERTION_INSERT, data, Collections.EMPTY_LIST);
-    }
-
-    private RowChangedEvent parseInsert(ByteBuffer buffer) {
-        int relationId = buffer.getInt();
-        logger.info("Insert table {}", tables.get(relationId).tableName);
-
-        List<Object> data = new ArrayList<>();
-        String newTuple = new String(new byte[]{buffer.get()}, 0, 1);
-        switch (newTuple) {
-            case "N":
-                readTupleData(buffer, data);
-                break;
-            default:
-                logger.info("N not set, got instead {}", newTuple);
-        }
-        return new RowChangedEvent(tables.get(relationId).tableName, ConnectorConstant.OPERTION_INSERT, Collections.EMPTY_LIST, data);
     }
 
-    private RowChangedEvent parseUpdate(ByteBuffer buffer) {
-        int relationId = buffer.getInt();
-        logger.info("Update table {}", tables.get(relationId).tableName);
-
-        List<Object> data = new ArrayList<>();
-        String newTuple = new String(new byte[]{buffer.get()}, 0, 1);
-        switch (newTuple) {
-            case "K":
-                logger.info("Key update");
-                logger.info("Old Key");
-                readTupleData(buffer, data);
-                break;
-            case "O":
-                logger.info("Value update");
-                logger.info("Old Value");
-                readTupleData(buffer, data);
-                break;
-            case "N":
-                readTupleData(buffer, data);
-                break;
-            default:
-                logger.info("K or O Byte1 not set, got instead {}", newTuple);
+    private RowChangedEvent parseData(MessageTypeEnum type, ByteBuffer buffer) {
+        final int relationId = buffer.getInt();
+        final TableId tableId = tables.get(relationId);
+        if (null != tableId) {
+            String newTuple = new String(new byte[]{buffer.get()}, 0, 1);
+            switch (newTuple) {
+                case "N":
+                case "K":
+                case "O":
+                    List<Object> data = new ArrayList<>();
+                    readTupleData(tableId, buffer, data);
+                    if (MessageTypeEnum.DELETE == type) {
+                        return new RowChangedEvent(tableId.tableName, type.name(), data, Collections.EMPTY_LIST);
+                    }
+                    return new RowChangedEvent(tableId.tableName, type.name(), Collections.EMPTY_LIST, data);
+
+                default:
+                    logger.info("N, K, O not set, got instead {}", newTuple);
+            }
         }
-
-        return new RowChangedEvent(tables.get(relationId).tableName, ConnectorConstant.OPERTION_UPDATE, Collections.EMPTY_LIST, data);
+        return null;
     }
 
-    private void readTupleData(ByteBuffer msg, List<Object> data) {
+    private void readTupleData(TableId tableId, ByteBuffer msg, List<Object> data) {
         short nColumn = msg.getShort();
-        for (int n = 0; n < nColumn; n++) {
-            String tupleContentType = new String(new byte[]{msg.get()}, 0, 1);
-            if (tupleContentType.equals("t")) {
-                int size = msg.getInt();
-                byte[] text = new byte[size];
-
-                for (int z = 0; z < size; z++) {
-                    text[z] = msg.get();
-                }
-                String content = new String(text, 0, size);
-                data.add(content);
-                continue;
-            }
+        if (nColumn != tableId.fields.size()) {
+            logger.warn("The column size of table '{}' is {}, but we has been received column size is {}.", tableId.tableName, tableId.fields.size(), nColumn);
 
-            if (tupleContentType.equals("n")) {
-                data.add(null);
-                continue;
+            // The table schema has been changed, we should be get a new table schema from db.
+            MetaInfo metaInfo = connectorFactory.getMetaInfo(connectorMapper, tableId.tableName);
+            if (CollectionUtils.isEmpty(metaInfo.getColumn())) {
+                throw new ListenerException(String.format("The table column for '%s' is empty.", tableId.tableName));
             }
+            tableId.fields = metaInfo.getColumn();
+            return;
+        }
 
-            if (tupleContentType.equals("u")) {
-                data.add("TOASTED");
+        for (int n = 0; n < nColumn; n++) {
+            String type = new String(new byte[]{msg.get()}, 0, 1);
+            switch (type) {
+                case "t":
+                    int size = msg.getInt();
+                    byte[] text = new byte[size];
+                    for (int z = 0; z < size; z++) {
+                        text[z] = msg.get();
+                    }
+                    data.add(resolveValue(tableId.fields.get(n).getTypeName(), new String(text, 0, size)));
+                    break;
+
+                case "n":
+                    data.add(null);
+                    break;
+
+                case "u":
+                    data.add("TOASTED");
+                    break;
+                default:
+                    logger.info("t, n, u not set, got instead {}", type);
             }
         }
     }
@@ -241,10 +191,12 @@ public class PgOutputMessageDecoder extends AbstractMessageDecoder {
     final class TableId {
         Integer oid;
         String tableName;
+        List<Field> fields;
 
-        public TableId(Integer oid, String tableName) {
+        public TableId(Integer oid, String tableName, List<Field> fields) {
             this.oid = oid;
             this.tableName = tableName;
+            this.fields = fields;
         }
     }
 

dbsyncer-listener/src/main/java/org/dbsyncer/listener/postgresql/decoder/TestDecodingMessageDecoder.java

@@ -3,9 +3,7 @@ package org.dbsyncer.listener.postgresql.decoder;
 import org.dbsyncer.common.event.RowChangedEvent;
 import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.listener.postgresql.AbstractMessageDecoder;
-import org.dbsyncer.listener.postgresql.column.ColumnValueResolver;
 import org.dbsyncer.listener.postgresql.column.Lexer;
-import org.dbsyncer.listener.postgresql.column.TestDecodingColumnValue;
 import org.dbsyncer.listener.postgresql.enums.MessageDecoderEnum;
 import org.dbsyncer.listener.postgresql.enums.MessageTypeEnum;
 import org.postgresql.replication.fluent.logical.ChainedLogicalStreamBuilder;
@@ -25,7 +23,6 @@ import java.util.List;
 public class TestDecodingMessageDecoder extends AbstractMessageDecoder {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
-    private static final ColumnValueResolver resolver = new ColumnValueResolver();
 
     @Override
     public RowChangedEvent processMessage(ByteBuffer buffer) {
@@ -77,7 +74,7 @@ public class TestDecodingMessageDecoder extends AbstractMessageDecoder {
             String type = parseType(lexer);
             lexer.skip(1);
             String value = parseValue(lexer);
-            data.add(resolver.resolveValue(type, new TestDecodingColumnValue(value)));
+            data.add(resolveValue(type, value));
         }
 
         RowChangedEvent event = null;

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

@@ -3,7 +3,7 @@ package org.dbsyncer.listener.quartz;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.config.Filter;
+import org.dbsyncer.connector.model.Filter;
 import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.listener.ListenerException;
 import org.dbsyncer.listener.enums.QuartzFilterEnum;

dbsyncer-listener/src/main/java/org/dbsyncer/listener/quartz/filter/DateFilter.java

@@ -3,7 +3,8 @@ package org.dbsyncer.listener.quartz.filter;
 import org.dbsyncer.common.util.DateFormatUtil;
 import org.dbsyncer.listener.quartz.QuartzFilter;
 
-import java.util.Date;
+import java.sql.Date;
+import java.time.LocalDate;
 
 public class DateFilter implements QuartzFilter {
 
@@ -15,7 +16,7 @@ public class DateFilter implements QuartzFilter {
 
     @Override
     public Object getObject() {
-        return new Date();
+        return Date.valueOf(LocalDate.now());
     }
 
     @Override

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

@@ -3,8 +3,8 @@ package org.dbsyncer.manager;
 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.model.MetaInfo;
+import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.connector.enums.ConnectorEnum;
 import org.dbsyncer.connector.enums.FilterEnum;
 import org.dbsyncer.connector.enums.OperationEnum;

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

@@ -5,8 +5,8 @@ import org.dbsyncer.manager.event.ClosedEvent;
 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.model.MetaInfo;
+import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.connector.enums.ConnectorEnum;
 import org.dbsyncer.connector.enums.FilterEnum;
 import org.dbsyncer.connector.enums.OperationEnum;

dbsyncer-manager/src/main/java/org/dbsyncer/manager/config/FieldPicker.java

@@ -3,8 +3,8 @@ package org.dbsyncer.manager.config;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.CompareFilter;
-import org.dbsyncer.connector.config.Field;
-import org.dbsyncer.connector.config.Filter;
+import org.dbsyncer.connector.model.Field;
+import org.dbsyncer.connector.model.Filter;
 import org.dbsyncer.connector.enums.FilterEnum;
 import org.dbsyncer.connector.enums.OperationEnum;
 import org.dbsyncer.parser.model.FieldMapping;

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

@@ -8,8 +8,8 @@ import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.ConnectorFactory;
 import org.dbsyncer.connector.config.ConnectorConfig;
-import org.dbsyncer.connector.config.Field;
-import org.dbsyncer.connector.config.Table;
+import org.dbsyncer.connector.model.Field;
+import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.listener.AbstractExtractor;
 import org.dbsyncer.listener.Extractor;

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

@@ -4,8 +4,8 @@ import org.dbsyncer.common.event.RowChangedEvent;
 import org.dbsyncer.common.model.Result;
 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.model.MetaInfo;
+import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.connector.enums.ConnectorEnum;
 import org.dbsyncer.connector.enums.FilterEnum;
 import org.dbsyncer.connector.enums.OperationEnum;

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

@@ -13,6 +13,9 @@ import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.connector.enums.ConnectorEnum;
 import org.dbsyncer.connector.enums.FilterEnum;
 import org.dbsyncer.connector.enums.OperationEnum;
+import org.dbsyncer.connector.model.Field;
+import org.dbsyncer.connector.model.MetaInfo;
+import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.listener.enums.QuartzFilterEnum;
 import org.dbsyncer.parser.enums.ConvertEnum;
 import org.dbsyncer.parser.enums.ParserEnum;

dbsyncer-parser/src/main/java/org/dbsyncer/parser/convert/handler/DateHandler.java

@@ -3,7 +3,8 @@ package org.dbsyncer.parser.convert.handler;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.parser.convert.Handler;
 
-import java.util.Date;
+import java.sql.Date;
+import java.time.LocalDate;
 
 /**
  * 系统日期
@@ -16,6 +17,6 @@ public class DateHandler implements Handler {
 
     @Override
     public Object handle(String args, Object value) {
-        return null == value || StringUtil.isBlank(String.valueOf(value)) ? new Date() : value;
+        return null == value || StringUtil.isBlank(String.valueOf(value)) ? Date.valueOf(LocalDate.now()) : value;
     }
 }

dbsyncer-parser/src/main/java/org/dbsyncer/parser/convert/handler/DateToChineseStandardTimeHandler.java

@@ -3,7 +3,7 @@ package org.dbsyncer.parser.convert.handler;
 import org.dbsyncer.common.util.DateFormatUtil;
 import org.dbsyncer.parser.convert.AbstractHandler;
 
-import java.util.Date;
+import java.sql.Date;
 
 /**
  * Date转中国标准时间

dbsyncer-parser/src/main/java/org/dbsyncer/parser/convert/handler/TimestampToDateHandler.java

@@ -2,9 +2,9 @@ package org.dbsyncer.parser.convert.handler;
 
 import org.dbsyncer.parser.convert.AbstractHandler;
 
+import java.sql.Date;
 import java.sql.Timestamp;
 import java.time.ZoneId;
-import java.util.Date;
 
 /**
  * 时间戳转日期

dbsyncer-parser/src/main/java/org/dbsyncer/parser/flush/model/AbstractWriter.java

@@ -1,6 +1,6 @@
 package org.dbsyncer.parser.flush.model;
 
-import org.dbsyncer.connector.config.Field;
+import org.dbsyncer.connector.model.Field;
 
 import java.util.List;
 import java.util.Map;

dbsyncer-parser/src/main/java/org/dbsyncer/parser/flush/model/WriterRequest.java

@@ -1,6 +1,6 @@
 package org.dbsyncer.parser.flush.model;
 
-import org.dbsyncer.connector.config.Field;
+import org.dbsyncer.connector.model.Field;
 import org.dbsyncer.parser.flush.BufferRequest;
 
 import java.util.List;

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

@@ -1,6 +1,6 @@
 package org.dbsyncer.parser.model;
 
-import org.dbsyncer.connector.config.Filter;
+import org.dbsyncer.connector.model.Filter;
 import org.dbsyncer.plugin.config.Plugin;
 
 import java.util.List;

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

@@ -1,7 +1,7 @@
 package org.dbsyncer.parser.model;
 
 import org.dbsyncer.connector.ConnectorMapper;
-import org.dbsyncer.connector.config.Field;
+import org.dbsyncer.connector.model.Field;
 
 import java.util.List;
 import java.util.Map;

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

@@ -1,7 +1,7 @@
 package org.dbsyncer.parser.model;
 
 import org.dbsyncer.connector.config.ConnectorConfig;
-import org.dbsyncer.connector.config.Table;
+import org.dbsyncer.connector.model.Table;
 
 import java.util.List;
 

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

@@ -1,6 +1,6 @@
 package org.dbsyncer.parser.model;
 
-import org.dbsyncer.connector.config.Field;
+import org.dbsyncer.connector.model.Field;
 
 /**
  * 字段映射关系

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

@@ -1,6 +1,6 @@
 package org.dbsyncer.parser.model;
 
-import org.dbsyncer.connector.config.Field;
+import org.dbsyncer.connector.model.Field;
 import org.dbsyncer.listener.config.ListenerConfig;
 import org.dbsyncer.parser.enums.ModelEnum;
 

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

@@ -1,7 +1,7 @@
 package org.dbsyncer.parser.model;
 
 import org.dbsyncer.common.util.CollectionUtils;
-import org.dbsyncer.connector.config.Field;
+import org.dbsyncer.connector.model.Field;
 
 import java.util.*;
 

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

@@ -1,6 +1,6 @@
 package org.dbsyncer.parser.model;
 
-import org.dbsyncer.connector.config.Table;
+import org.dbsyncer.connector.model.Table;
 
 import java.util.List;
 import java.util.Map;

dbsyncer-parser/src/main/java/org/dbsyncer/parser/util/PickerUtil.java

@@ -2,8 +2,8 @@ package org.dbsyncer.parser.util;
 
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.config.Field;
-import org.dbsyncer.connector.config.Filter;
+import org.dbsyncer.connector.model.Field;
+import org.dbsyncer.connector.model.Filter;
 import org.dbsyncer.parser.model.Convert;
 import org.dbsyncer.parser.model.FieldMapping;
 import org.dbsyncer.parser.model.Mapping;

dbsyncer-storage/src/main/java/org/dbsyncer/storage/support/MysqlStorageServiceImpl.java

@@ -6,7 +6,7 @@ import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.ConnectorFactory;
 import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.config.DatabaseConfig;
-import org.dbsyncer.connector.config.Field;
+import org.dbsyncer.connector.model.Field;
 import org.dbsyncer.connector.config.SqlBuilderConfig;
 import org.dbsyncer.connector.config.WriterBatchConfig;
 import org.dbsyncer.connector.constant.ConnectorConstant;

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

@@ -56,14 +56,11 @@
                    type="checkbox">
         </div>
         <label class="col-sm-2 control-label">插件</label>
-        <div class="col-sm-4"
-             th:if="${#maps.isEmpty(connector?.config?.properties) or not #maps.containsKey(connector?.config?.properties, 'pluginName')}">
-            <input class="form-control" maxlength="32" name="pluginName" th:value="'test_decoding'" type="text"/>
-        </div>
-        <div class="col-sm-4"
-             th:if="${not #maps.isEmpty(connector?.config?.properties) and #maps.containsKey(connector?.config?.properties, 'pluginName')}">
-            <input class="form-control" maxlength="32" name="pluginName"
-                   th:value="${connector?.config?.properties?.pluginName}" type="text"/>
+        <div class="col-sm-4">
+            <select class="form-control select-control" name="pluginName">
+                <option value="pgoutput" th:selected="${connector?.config?.properties?.pluginName eq 'pgoutput'}">pgoutput</option>
+                <option value="test_decoding" th:selected="${connector?.config?.properties?.pluginName eq 'test_decoding'}">test_decoding</option>
+            </select>
         </div>
     </div>
     <div class="form-group">
@@ -83,6 +80,8 @@
                 offColor: "info",
                 size: "normal"
             });
+            // 初始化select插件
+            initSelectIndex($(".select-control"), 1);
         })
     </script>
 </div>

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

@@ -0,0 +1,61 @@
+<!DOCTYPE html>
+<html xmlns="http://www.w3.org/1999/xhtml"
+      xmlns:th="http://www.thymeleaf.org" lang="zh-CN">
+
+<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="监听的文件路径"></i> <strong
+                class="driverVerifcateRequired">*</strong></label>
+        <div class="col-sm-4">
+            <input class="form-control" name="fileDir" type="text" maxlength="512" dbsyncer-valid="require"
+                   th:value="${connector?.config?.fileDir}?:'/soft'"/>
+        </div>
+        <div class="col-sm-6"></div>
+    </div>
+
+    <div class="form-group">
+        <label class="col-sm-2 control-label">schema <i class="fa fa-question-circle fa_gray" aria-hidden="true"
+                                                        title="支持11种字段类型。name字段名, typeName类型名称, type类型编码, pk是否为主键"></i><strong
+                class="driverVerifcateRequired">*</strong></label>
+        <div class="col-sm-9">
+            <textarea id="schema" name="schema" class="form-control dbsyncer_textarea_resize_none" maxlength="4096"
+                      dbsyncer-valid="require" rows="20" th:text="${connector?.config?.schema}"></textarea>
+        </div>
+        <div class="col-sm-1">
+            <button type="button" class="btn btn-default" onclick="format()">
+                <span class="fa fa-magic"></span>美化
+            </button>
+        </div>
+    </div>
+
+    <script type="text/javascript">
+        function format() {
+            const $text = $("#schema");
+            if ("" == $text.text()) {
+                const data = [
+                    {"name": "id", "typeName": "String", "type": 12, "pk": true},
+                    {"name": "age", "typeName": "Integer", "type": 4},
+                    {"name": "count", "typeName": "Long", "type": -5},
+                    {"name": "type", "typeName": "Short", "type": 5},
+                    {"name": "money", "typeName": "Float", "type": 6},
+                    {"name": "score", "typeName": "Double", "type": 8},
+                    {"name": "status", "typeName": "Boolean", "type": -7},
+                    // {"name":"photo","typeName":"byte[]","type":-2},
+                    {"name": "create_date", "typeName": "Date", "type": 91},
+                    {"name": "time", "typeName": "Time", "type": 92},
+                    {"name": "update_time", "typeName": "Timestamp", "type": 93}
+                ];
+                $text.val(JSON.stringify(data, null, 4));
+                return;
+            }
+            $text.val(JSON.stringify(JSON.parse($text.text()), null, 4));
+        }
+
+        $(function () {
+            format();
+        })
+    </script>
+</div>
+
+</html>

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

@@ -32,29 +32,24 @@
         <div class="col-sm-6"></div>
     </div>
     <div class="form-group">
-        <label class="col-sm-2 control-label">删除Slot <i aria-hidden="true" class="fa fa-question-circle fa_gray"
-                                                        title="增量同步，停止驱动自动删除Slot"></i></label>
+        <label class="col-sm-2 control-label">删除Slot <i aria-hidden="true" class="fa fa-question-circle fa_gray" title="增量同步，停止驱动自动删除Slot"></i></label>
         <div class="col-sm-4">
             <input id="dropSlotOnCloseSwitch" name="dropSlotOnClose"
                    th:checked="${#maps.isEmpty(connector?.config?.properties) or connector?.config?.properties?.dropSlotOnClose eq 'true'}"
                    type="checkbox">
         </div>
         <label class="col-sm-2 control-label">插件</label>
-        <div class="col-sm-4"
-             th:if="${#maps.isEmpty(connector?.config?.properties) or not #maps.containsKey(connector?.config?.properties, 'pluginName')}">
-            <input class="form-control" maxlength="32" name="pluginName" th:value="'test_decoding'" type="text"/>
-        </div>
-        <div class="col-sm-4"
-             th:if="${not #maps.isEmpty(connector?.config?.properties) and #maps.containsKey(connector?.config?.properties, 'pluginName')}">
-            <input class="form-control" maxlength="32" name="pluginName"
-                   th:value="${connector?.config?.properties?.pluginName}" type="text"/>
+        <div class="col-sm-4">
+            <select class="form-control select-control" name="pluginName">
+                <option value="pgoutput" th:selected="${connector?.config?.properties?.pluginName eq 'pgoutput'}">pgoutput</option>
+                <option value="test_decoding" th:selected="${connector?.config?.properties?.pluginName eq 'test_decoding'}">test_decoding</option>
+            </select>
         </div>
     </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">驱动 </label>
         <div class="col-sm-10">
-            <input class="form-control" name="driverClassName" readonly="true"
-                   th:value="${connector?.config?.driverClassName} ?: 'org.postgresql.Driver'" type="text"/>
+            <input class="form-control" name="driverClassName" readonly="true" th:value="${connector?.config?.driverClassName} ?: 'org.postgresql.Driver'" type="text"/>
         </div>
     </div>
 
@@ -67,6 +62,8 @@
                 offColor: "info",
                 size: "normal"
             });
+            // 初始化select插件
+            initSelectIndex($(".select-control"), 1);
         })
     </script>
 </div>