add module dbsyncer-connector-sqlserver

dbsyncer-connector/dbsyncer-connector-oracle/pom.xml

@@ -26,11 +26,6 @@
             <artifactId>ojdbc6</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>com.github.jsqlparser</groupId>
-            <artifactId>jsqlparser</artifactId>
-        </dependency>
-
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-log4j2</artifactId>

dbsyncer-connector/dbsyncer-connector-sqlserver/pom.xml

@@ -0,0 +1,42 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>dbsyncer-connector</artifactId>
+        <groupId>org.ghi</groupId>
+        <version>2.0.0</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>dbsyncer-connector-sqlserver</artifactId>
+
+    <dependencies>
+        <!-- sdk -->
+        <dependency>
+            <groupId>org.ghi</groupId>
+            <artifactId>dbsyncer-sdk</artifactId>
+            <version>${project.parent.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- sqlserver-driver -->
+        <dependency>
+            <groupId>com.microsoft.sqlserver</groupId>
+            <artifactId>mssql-jdbc</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-log4j2</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+    </dependencies>
+</project>

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

@@ -1,51 +1,67 @@
-package org.dbsyncer.connector.sql;
-
-import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.sdk.listener.DatabaseQuartzListener;
-import org.dbsyncer.connector.sqlserver.DqlSqlServerListener;
-import org.dbsyncer.sdk.config.ReaderConfig;
-import org.dbsyncer.sdk.connector.database.AbstractDQLConnector;
-import org.dbsyncer.sdk.constant.DatabaseConstant;
-import org.dbsyncer.sdk.enums.ListenerTypeEnum;
-import org.dbsyncer.sdk.listener.Listener;
-import org.dbsyncer.sdk.model.PageSql;
-import org.springframework.stereotype.Component;
-
-import java.util.List;
-
-@Component
-public final class DQLSqlServerConnector extends AbstractDQLConnector {
-
-    private final String TYPE = "DqlSqlServer";
-
-    @Override
-    public String getConnectorType() {
-        return TYPE;
-    }
-
-    @Override
-    public Listener getListener(String listenerType) {
-        if (ListenerTypeEnum.isTiming(listenerType)) {
-            return new DatabaseQuartzListener();
-        }
-
-        if (ListenerTypeEnum.isLog(listenerType)) {
-            return new DqlSqlServerListener();
-        }
-        return null;
-    }
-
-    @Override
-    public String getPageSql(PageSql config) {
-        List<String> primaryKeys = config.getPrimaryKeys();
-        String orderBy = StringUtil.join(primaryKeys, ",");
-        return String.format(DatabaseConstant.SQLSERVER_PAGE_SQL, orderBy, config.getQuerySql());
-    }
-
-    @Override
-    public Object[] getPageArgs(ReaderConfig config) {
-        int pageSize = config.getPageSize();
-        int pageIndex = config.getPageIndex();
-        return new Object[]{(pageIndex - 1) * pageSize + 1, pageIndex * pageSize};
-    }
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.sqlserver;
+
+import org.dbsyncer.common.util.StringUtil;
+import org.dbsyncer.connector.sqlserver.cdc.DqlSqlServerListener;
+import org.dbsyncer.connector.sqlserver.validator.DqlSqlServerConfigValidator;
+import org.dbsyncer.sdk.config.ReaderConfig;
+import org.dbsyncer.sdk.connector.ConfigValidator;
+import org.dbsyncer.sdk.connector.database.AbstractDQLConnector;
+import org.dbsyncer.sdk.constant.DatabaseConstant;
+import org.dbsyncer.sdk.enums.ListenerTypeEnum;
+import org.dbsyncer.sdk.listener.DatabaseQuartzListener;
+import org.dbsyncer.sdk.listener.Listener;
+import org.dbsyncer.sdk.model.PageSql;
+
+import java.util.List;
+
+/**
+ * DQLSqlServer连接器实现
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2022-05-22 22:56
+ */
+public final class DQLSqlServerConnector extends AbstractDQLConnector {
+
+    private final String TYPE = "DqlSqlServer";
+    private final DqlSqlServerConfigValidator configValidator = new DqlSqlServerConfigValidator();
+
+    @Override
+    public String getConnectorType() {
+        return TYPE;
+    }
+
+    @Override
+    public ConfigValidator getConfigValidator() {
+        return null;
+    }
+
+    @Override
+    public Listener getListener(String listenerType) {
+        if (ListenerTypeEnum.isTiming(listenerType)) {
+            return new DatabaseQuartzListener();
+        }
+
+        if (ListenerTypeEnum.isLog(listenerType)) {
+            return new DqlSqlServerListener();
+        }
+        return null;
+    }
+
+    @Override
+    public String getPageSql(PageSql config) {
+        List<String> primaryKeys = config.getPrimaryKeys();
+        String orderBy = StringUtil.join(primaryKeys, ",");
+        return String.format(DatabaseConstant.SQLSERVER_PAGE_SQL, orderBy, config.getQuerySql());
+    }
+
+    @Override
+    public Object[] getPageArgs(ReaderConfig config) {
+        int pageSize = config.getPageSize();
+        int pageIndex = config.getPageIndex();
+        return new Object[]{(pageIndex - 1) * pageSize + 1, pageIndex * pageSize};
+    }
 }

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

@@ -1,40 +1,56 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 package org.dbsyncer.connector.sqlserver;
 
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.sdk.listener.DatabaseQuartzListener;
+import org.dbsyncer.connector.sqlserver.cdc.SqlServerListener;
+import org.dbsyncer.connector.sqlserver.validator.SqlServerConfigValidator;
 import org.dbsyncer.sdk.config.CommandConfig;
 import org.dbsyncer.sdk.config.DatabaseConfig;
 import org.dbsyncer.sdk.config.ReaderConfig;
+import org.dbsyncer.sdk.connector.ConfigValidator;
 import org.dbsyncer.sdk.connector.database.AbstractDatabaseConnector;
 import org.dbsyncer.sdk.connector.database.DatabaseConnectorInstance;
 import org.dbsyncer.sdk.constant.DatabaseConstant;
 import org.dbsyncer.sdk.enums.ListenerTypeEnum;
 import org.dbsyncer.sdk.enums.TableTypeEnum;
+import org.dbsyncer.sdk.listener.DatabaseQuartzListener;
 import org.dbsyncer.sdk.listener.Listener;
 import org.dbsyncer.sdk.model.Field;
 import org.dbsyncer.sdk.model.PageSql;
 import org.dbsyncer.sdk.model.Table;
-import org.springframework.stereotype.Component;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.stream.Collectors;
 
-@Component
+/**
+ * SqlServer连接器实现
+ *
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2022-05-22 22:56
+ */
 public final class SqlServerConnector extends AbstractDatabaseConnector {
 
     private final String QUERY_VIEW = "select name from sysobjects where xtype in('v')";
-
     private final String QUERY_TABLE = "select name from sys.tables where schema_id = schema_id('%s') and is_ms_shipped = 0";
 
     private final String TYPE = "SqlServer";
+    private final SqlServerConfigValidator configValidator = new SqlServerConfigValidator();
 
     @Override
     public String getConnectorType() {
         return TYPE;
     }
 
+    @Override
+    public ConfigValidator getConfigValidator() {
+        return configValidator;
+    }
+
     @Override
     public List<Table> getTable(DatabaseConnectorInstance connectorInstance) {
         DatabaseConfig config = connectorInstance.getConfig();

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

@@ -0,0 +1,30 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.sqlserver;
+
+/**
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2020-01-08 15:17
+ */
+public class SqlServerException extends RuntimeException {
+
+	private static final long serialVersionUID = 1L;
+
+	public SqlServerException(String message) {
+        super(message);
+    }
+
+    public SqlServerException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public SqlServerException(Throwable cause) {
+        super(cause);
+    }
+
+    protected SqlServerException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+        super(message, cause, enableSuppression, writableStackTrace);
+    }
+}

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sqlserver/DqlSqlServerListener.java → dbsyncer-connector/dbsyncer-connector-sqlserver/src/main/java/org/dbsyncer/connector/sqlserver/cdc/DqlSqlServerListener.java

@@ -1,11 +1,14 @@
-package org.dbsyncer.connector.sqlserver;
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.sqlserver.cdc;
 
 import org.dbsyncer.sdk.listener.ChangedEvent;
 
 /**
- * @author AE86
- * @version 1.0.0
- * @date 2022/5/22 22:56
+ * @Author AE86
+ * @Version 1.0.0
+ * @Date 2022-05-22 22:56
  */
 public class DqlSqlServerListener extends SqlServerListener {
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sqlserver/Lsn.java → dbsyncer-connector/dbsyncer-connector-sqlserver/src/main/java/org/dbsyncer/connector/sqlserver/cdc/Lsn.java

@@ -1,4 +1,7 @@
-package org.dbsyncer.connector.sqlserver;
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.sqlserver.cdc;
 
 import java.util.Arrays;
 
@@ -8,7 +11,8 @@ import org.dbsyncer.common.util.StringUtil;
  * SQL Server LSN（日志序列号）位置的逻辑表示, LSN不可用时为NULL。
  *
  * @Author AE86
- * @Date 2021-06-04 22:25
+ * @Version 1.0.0
+ * @Date 2022-05-22 22:56
  */
 public class Lsn implements Comparable<Lsn> {
     private static final String NULL_STRING = "NULL";

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sqlserver/LsnPuller.java → dbsyncer-connector/dbsyncer-connector-sqlserver/src/main/java/org/dbsyncer/connector/sqlserver/cdc/LsnPuller.java

@@ -1,4 +1,7 @@
-package org.dbsyncer.connector.sqlserver;
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.sqlserver.cdc;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -8,10 +11,9 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 
 /**
- * @author Xinpeng.Fu
- * @version V1.0
- * @description
- * @date 2022/8/30 10:04
+ * @Author Xinpeng.Fu
+ * @Version 1.0.0
+ * @Date 2022-05-22 22:56
  */
 public class LsnPuller {
     private final Logger logger = LoggerFactory.getLogger(getClass());

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sqlserver/SqlServerListener.java → dbsyncer-connector/dbsyncer-connector-sqlserver/src/main/java/org/dbsyncer/connector/sqlserver/cdc/SqlServerListener.java

@@ -1,16 +1,21 @@
-package org.dbsyncer.connector.sqlserver;
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.sqlserver.cdc;
 
 import com.microsoft.sqlserver.jdbc.SQLServerException;
-import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.enums.TableOperationEnum;
-import org.dbsyncer.sdk.connector.database.AbstractDatabaseConnector;
-import org.dbsyncer.sdk.model.ChangedOffset;
-import org.dbsyncer.sdk.listener.event.RowChangedEvent;
 import org.dbsyncer.common.util.CollectionUtils;
+import org.dbsyncer.connector.sqlserver.model.SqlServerChangeTable;
+import org.dbsyncer.connector.sqlserver.SqlServerException;
+import org.dbsyncer.connector.sqlserver.enums.TableOperationEnum;
+import org.dbsyncer.connector.sqlserver.model.CDCEvent;
 import org.dbsyncer.sdk.config.DatabaseConfig;
-import org.dbsyncer.sdk.constant.ConnectorConstant;
+import org.dbsyncer.sdk.connector.database.AbstractDatabaseConnector;
 import org.dbsyncer.sdk.connector.database.DatabaseConnectorInstance;
+import org.dbsyncer.sdk.constant.ConnectorConstant;
 import org.dbsyncer.sdk.listener.AbstractDatabaseListener;
+import org.dbsyncer.sdk.listener.event.RowChangedEvent;
+import org.dbsyncer.sdk.model.ChangedOffset;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.util.Assert;
@@ -33,9 +38,9 @@ import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
 /**
- * @version 1.0.0
  * @Author AE86
- * @Date 2021-06-18 01:20
+ * @Version 1.0.0
+ * @Date 2022-05-22 22:56
  */
 public class SqlServerListener extends AbstractDatabaseListener {
 
@@ -100,7 +105,7 @@ public class SqlServerListener extends AbstractDatabaseListener {
         } catch (Exception e) {
             close();
             logger.error("启动失败:{}", e.getMessage());
-            throw new ConnectorException(e);
+            throw new SqlServerException(e);
         } finally {
             connectLock.unlock();
         }
@@ -154,7 +159,7 @@ public class SqlServerListener extends AbstractDatabaseListener {
                 return;
             }
             // Shouldn't happen if the agent is running, but it is better to guard against such situation
-            throw new ConnectorException("No maximum LSN recorded in the database");
+            throw new SqlServerException("No maximum LSN recorded in the database");
         }
         lastLsn = Lsn.valueOf(snapshot.get(LSN_POSITION));
     }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/enums/TableOperationEnum.java → dbsyncer-connector/dbsyncer-connector-sqlserver/src/main/java/org/dbsyncer/connector/sqlserver/enums/TableOperationEnum.java

@@ -1,4 +1,7 @@
-package org.dbsyncer.connector.enums;
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.sqlserver.enums;
 
 public enum TableOperationEnum {
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sqlserver/CDCEvent.java → dbsyncer-connector/dbsyncer-connector-sqlserver/src/main/java/org/dbsyncer/connector/sqlserver/model/CDCEvent.java

@@ -1,7 +1,7 @@
 /**
- * DBSyncer Copyright 2019-2024 All Rights Reserved.
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
  */
-package org.dbsyncer.connector.sqlserver;
+package org.dbsyncer.connector.sqlserver.model;
 
 import java.util.List;
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sqlserver/SqlServerChangeTable.java → dbsyncer-connector/dbsyncer-connector-sqlserver/src/main/java/org/dbsyncer/connector/sqlserver/model/SqlServerChangeTable.java

@@ -1,4 +1,7 @@
-package org.dbsyncer.connector.sqlserver;
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
+package org.dbsyncer.connector.sqlserver.model;
 
 import org.dbsyncer.common.util.JsonUtil;
 

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/DqlSqlServerConfigValidator.java → dbsyncer-connector/dbsyncer-connector-sqlserver/src/main/java/org/dbsyncer/connector/sqlserver/validator/DqlSqlServerConfigValidator.java

@@ -1,11 +1,10 @@
 /**
  * DBSyncer Copyright 2020-2023 All Rights Reserved.
  */
-package org.dbsyncer.biz.checker.impl.connector;
+package org.dbsyncer.connector.sqlserver.validator;
 
 import org.dbsyncer.sdk.config.DatabaseConfig;
 import org.dbsyncer.sdk.connector.AbstractDataBaseConfigValidator;
-import org.springframework.stereotype.Component;
 
 import java.util.Map;
 
@@ -16,7 +15,6 @@ import java.util.Map;
  * @Version 1.0.0
  * @Date 2022-05-18 20:17
  */
-@Component
 public class DqlSqlServerConfigValidator extends AbstractDataBaseConfigValidator {
 
     @Override

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/SqlServerConfigValidator.java → dbsyncer-connector/dbsyncer-connector-sqlserver/src/main/java/org/dbsyncer/connector/sqlserver/validator/SqlServerConfigValidator.java

@@ -1,20 +1,20 @@
 /**
  * DBSyncer Copyright 2020-2023 All Rights Reserved.
  */
-package org.dbsyncer.biz.checker.impl.connector;
+package org.dbsyncer.connector.sqlserver.validator;
 
 import org.dbsyncer.sdk.config.DatabaseConfig;
 import org.dbsyncer.sdk.connector.AbstractDataBaseConfigValidator;
-import org.springframework.stereotype.Component;
 
 import java.util.Map;
 
 /**
+ * SqlServer连接配置校验器实现
+ *
  * @Author AE86
  * @Version 1.0.0
  * @Date 2023-11-25 23:10
  */
-@Component
 public class SqlServerConfigValidator extends AbstractDataBaseConfigValidator {
     @Override
     public void modify(DatabaseConfig connectorConfig, Map<String, String> params) {

dbsyncer-connector/dbsyncer-connector-sqlserver/src/main/resources/META-INF/services/org.dbsyncer.sdk.spi.ConnectorService

@@ -0,0 +1,2 @@
+org.dbsyncer.connector.sqlserver.SqlServerConnector
+org.dbsyncer.connector.sqlserver.DQLSqlServerConnector

dbsyncer-connector/src/main/test/ChangeDataCaptureTest.java → dbsyncer-connector/dbsyncer-connector-sqlserver/src/test/java/ChangeDataCaptureTest.java

@@ -1,7 +1,10 @@
+/**
+ * DBSyncer Copyright 2020-2023 All Rights Reserved.
+ */
 import com.microsoft.sqlserver.jdbc.SQLServerException;
 import org.dbsyncer.common.util.CollectionUtils;
-import org.dbsyncer.connector.sqlserver.Lsn;
-import org.dbsyncer.connector.sqlserver.SqlServerChangeTable;
+import org.dbsyncer.connector.sqlserver.model.SqlServerChangeTable;
+import org.dbsyncer.connector.sqlserver.cdc.Lsn;
 import org.junit.Assert;
 import org.junit.Test;
 import org.slf4j.Logger;
@@ -14,9 +17,9 @@ import java.util.concurrent.atomic.AtomicInteger;
 import java.util.regex.Matcher;
 
 /**
- * @version 1.0.0
  * @Author AE86
- * @Date 2021-06-14 01:55
+ * @Version 1.0.0
+ * @Date 2022-05-22 22:56
  * @Link https://www.red-gate.com/simple-talk/sql/learn-sql-server/introduction-to-change-data-capture-cdc-in-sql-server-2008/
  */
 public class ChangeDataCaptureTest {

dbsyncer-connector/pom.xml

@@ -16,6 +16,7 @@
         <module>dbsyncer-connector-kafka</module>
         <module>dbsyncer-connector-mysql</module>
         <module>dbsyncer-connector-oracle</module>
+        <module>dbsyncer-connector-sqlserver</module>
     </modules>
 
     <dependencies>
@@ -54,11 +55,18 @@
             <version>${project.parent.version}</version>
         </dependency> -->
 
-        <!-- dbsyncer-connector-oracle -->
+        <!-- dbsyncer-connector-oracle
         <dependency>
             <groupId>org.ghi</groupId>
             <artifactId>dbsyncer-connector-oracle</artifactId>
             <version>${project.parent.version}</version>
+        </dependency> -->
+
+        <!-- dbsyncer-connector-sqlserver -->
+        <dependency>
+            <groupId>org.ghi</groupId>
+            <artifactId>dbsyncer-connector-sqlserver</artifactId>
+            <version>${project.parent.version}</version>
         </dependency>
 
         <!-- Spring-JDBC -->
@@ -67,12 +75,6 @@
             <artifactId>spring-jdbc</artifactId>
         </dependency>
 
-        <!-- sqlserver-driver -->
-        <dependency>
-            <groupId>com.microsoft.sqlserver</groupId>
-            <artifactId>mssql-jdbc</artifactId>
-        </dependency>
-
         <dependency>
             <groupId>org.antlr</groupId>
             <artifactId>antlr4-runtime</artifactId>

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

@@ -1,6 +1,7 @@
 package org.dbsyncer.connector.postgresql;
 
 import org.dbsyncer.common.util.StringUtil;
+import org.dbsyncer.sdk.connector.ConfigValidator;
 import org.dbsyncer.sdk.listener.DatabaseQuartzListener;
 import org.dbsyncer.sdk.config.CommandConfig;
 import org.dbsyncer.sdk.config.DatabaseConfig;
@@ -33,6 +34,11 @@ public final class PostgreSQLConnector extends AbstractDatabaseConnector {
         return TYPE;
     }
 
+    @Override
+    public ConfigValidator getConfigValidator() {
+        return null;
+    }
+
     @Override
     public Listener getListener(String listenerType) {
         if (ListenerTypeEnum.isTiming(listenerType)) {

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

@@ -1,6 +1,7 @@
 package org.dbsyncer.connector.sql;
 
 import org.dbsyncer.connector.postgresql.DqlPostgreSQLListener;
+import org.dbsyncer.sdk.connector.ConfigValidator;
 import org.dbsyncer.sdk.listener.DatabaseQuartzListener;
 import org.dbsyncer.sdk.config.ReaderConfig;
 import org.dbsyncer.sdk.connector.database.AbstractDQLConnector;
@@ -20,6 +21,11 @@ public final class DQLPostgreSQLConnector extends AbstractDQLConnector {
         return TYPE;
     }
 
+    @Override
+    public ConfigValidator getConfigValidator() {
+        return null;
+    }
+
     @Override
     public Listener getListener(String listenerType) {
         if (ListenerTypeEnum.isTiming(listenerType)) {

dbsyncer-connector/src/main/test/ConnectionTest.java

@@ -1,360 +0,0 @@
-import oracle.jdbc.OracleConnection;
-import org.dbsyncer.common.util.CollectionUtils;
-import org.dbsyncer.common.util.RandomUtil;
-import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.sdk.config.DatabaseConfig;
-import org.dbsyncer.sdk.connector.database.DatabaseConnectorInstance;
-import org.dbsyncer.sdk.connector.database.ds.SimpleConnection;
-import org.dbsyncer.sdk.enums.TableTypeEnum;
-import org.dbsyncer.sdk.model.Field;
-import org.dbsyncer.sdk.model.Table;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.jdbc.core.BatchPreparedStatementSetter;
-
-import java.nio.charset.Charset;
-import java.sql.*;
-import java.time.Instant;
-import java.time.LocalDateTime;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.*;
-
-/**
- * @author AE86
- * @version 1.0.0
- * @date 2022/4/11 20:19
- */
-public class ConnectionTest {
-
-    private final Logger logger = LoggerFactory.getLogger(getClass());
-
-    @Test
-    public void testByte() {
-        final DatabaseConnectorInstance connectorInstance = new DatabaseConnectorInstance(createOracleConfig());
-
-        String executeSql = "UPDATE \"my_user\" SET \"name\"=?,\"clo\"=? WHERE \"id\"=?";
-        int[] execute = connectorInstance.execute(databaseTemplate ->
-                databaseTemplate.batchUpdate(executeSql, new BatchPreparedStatementSetter() {
-                    @Override
-                    public void setValues(PreparedStatement ps, int i) {
-                        try {
-                            SimpleConnection connection = databaseTemplate.getSimpleConnection();
-                            OracleConnection conn = (OracleConnection) connection.getConnection();
-                            Clob clob = conn.createClob();
-                            clob.setString(1, new String("中文888".getBytes(Charset.defaultCharset())));
-
-                            ps.setString(1, "hello888");
-                            ps.setClob(2, clob);
-                            ps.setInt(3, 2);
-                        } catch (SQLException e) {
-                            e.printStackTrace();
-                        }
-                    }
-
-                    @Override
-                    public int getBatchSize() {
-                        return 1;
-                    }
-                })
-        );
-        logger.info("execute:{}", execute);
-    }
-
-    @Test
-    public void testConnection() throws InterruptedException {
-        final DatabaseConnectorInstance connectorInstance = new DatabaseConnectorInstance(createSqlServerConfig());
-
-        // 模拟并发
-        final int threadSize = 100;
-        final ExecutorService pool = Executors.newFixedThreadPool(threadSize);
-        final CyclicBarrier barrier = new CyclicBarrier(threadSize);
-        final CountDownLatch latch = new CountDownLatch(threadSize);
-        for (int i = 0; i < threadSize; i++) {
-            final int k = i + 3;
-            pool.submit(() -> {
-                try {
-                    barrier.await();
-
-                    // 模拟操作
-                    System.out.println(String.format("%s %s:%s", LocalDateTime.now(), Thread.currentThread().getName(), k));
-
-                    Object execute = connectorInstance.execute(tem -> tem.queryForObject("select 1", Integer.class));
-                    System.out.println(String.format("%s %s:%s execute=>%s", LocalDateTime.now(), Thread.currentThread().getName(), k, execute));
-
-                } catch (InterruptedException e) {
-                    logger.error(e.getMessage());
-                } catch (BrokenBarrierException e) {
-                    logger.error(e.getMessage());
-                } catch (Exception e) {
-                    logger.error(e.getMessage());
-                } finally {
-                    latch.countDown();
-                }
-            });
-        }
-
-        try {
-            latch.await();
-            logger.info("try to shutdown");
-            pool.shutdown();
-        } catch (InterruptedException e) {
-            logger.error(e.getMessage());
-        }
-
-        TimeUnit.SECONDS.sleep(3);
-        logger.info("test end");
-    }
-
-    @Test
-    public void testBatchInsert() {
-        final DatabaseConnectorInstance connectorInstance = new DatabaseConnectorInstance(createMysqlConfig());
-
-        long begin = Instant.now().toEpochMilli();
-        final int threadSize = 10;
-        final ExecutorService pool = Executors.newFixedThreadPool(threadSize);
-        final String sql = "INSERT INTO `vote_records_copy` (`id`, `user_id`, `vote_num`, `group_id`, `status`, `create_time`) VALUES (?, ?, ?, ?, ?, ?)";
-
-        // 模拟1000w条数据
-        List<Object[]> dataList = new ArrayList<>();
-        for (int i = 1; i <= 200001; i++) {
-            // 442001, 'dA8LeJLtX9MgQgDe7H1O', 9620, 1, 2, '2022-11-17 16:35:21'
-            Object[] args = new Object[6];
-            args[0] = i;
-            args[1] = randomUserId(20);
-            args[2] = RandomUtil.nextInt(1, 9999);
-            args[3] = RandomUtil.nextInt(0, 3);
-            args[4] = RandomUtil.nextInt(1, 3);
-            args[5] = Timestamp.valueOf(LocalDateTime.now());
-            dataList.add(args);
-
-            if (i % 10000 == 0) {
-                System.out.println(i + "-----------------正在处理");
-                batchUpdate(connectorInstance, pool, sql, dataList, 1000);
-                dataList.clear();
-            }
-        }
-
-        if(!CollectionUtils.isEmpty(dataList)){
-            System.out.println("-----------------正在处理剩余数据");
-            batchUpdate(connectorInstance, pool, sql, dataList, 1000);
-        }
-
-        pool.shutdown();
-        logger.info("总共耗时：{}秒", (Instant.now().toEpochMilli() - begin) / 1000);
-    }
-
-    @Test
-    public void testBatchUpdate() {
-        final DatabaseConnectorInstance connectorInstance = new DatabaseConnectorInstance(createMysqlConfig());
-
-        long begin = Instant.now().toEpochMilli();
-        final int threadSize = 10;
-        final ExecutorService pool = Executors.newFixedThreadPool(threadSize);
-        final String sql = "UPDATE `test`.`vote_records` SET `user_id` = ?, `create_time` = now() WHERE `id` = ?";
-
-        // 模拟100w条数据
-        int k = 10;
-        while (k > 0) {
-            List<Object[]> dataList = new ArrayList<>();
-            for (int i = 1; i <= 100000; i++) {
-                // 'dA8LeJLtX9MgQgDe7H1O', '2022-11-17 16:35:21', 1
-                Object[] args = new Object[2];
-                args[0] = randomUserId(20);
-                args[1] = i;
-                dataList.add(args);
-
-                if (i % 10000 == 0) {
-                    System.out.println(i + "-----------------正在处理");
-                    batchUpdate(connectorInstance, pool, sql, dataList, 1000);
-                    dataList.clear();
-                }
-            }
-
-            if (!CollectionUtils.isEmpty(dataList)) {
-                System.out.println("-----------------正在处理剩余数据");
-                batchUpdate(connectorInstance, pool, sql, dataList, 1000);
-            }
-            k--;
-        }
-
-        pool.shutdown();
-        logger.info("总共耗时：{}秒", (Instant.now().toEpochMilli() - begin) / 1000);
-    }
-
-    @Test
-    public void testBatchDelete() {
-        final DatabaseConnectorInstance connectorInstance = new DatabaseConnectorInstance(createMysqlConfig());
-
-        long begin = Instant.now().toEpochMilli();
-        final int threadSize = 10;
-        final ExecutorService pool = Executors.newFixedThreadPool(threadSize);
-        final String sql = "delete from `test`.`vote_records` WHERE `id` = ?";
-
-        List<Object[]> dataList = new ArrayList<>();
-        for (int i = 1; i <= 3259000; i++) {
-            // 'dA8LeJLtX9MgQgDe7H1O', '2022-11-17 16:35:21', 1
-            Object[] args = new Object[1];
-            args[0] = i;
-            dataList.add(args);
-
-            if (i % 10000 == 0) {
-                System.out.println(i + "-----------------正在处理");
-                batchUpdate(connectorInstance, pool, sql, dataList, 1000);
-                dataList.clear();
-            }
-        }
-
-        if (!CollectionUtils.isEmpty(dataList)) {
-            System.out.println("-----------------正在处理剩余数据");
-            batchUpdate(connectorInstance, pool, sql, dataList, 1000);
-        }
-
-        pool.shutdown();
-        logger.info("总共耗时：{}秒", (Instant.now().toEpochMilli() - begin) / 1000);
-    }
-
-    private final static String STR = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
-
-    private String randomUserId(int i) {
-        StringBuilder s = new StringBuilder();
-        for (int j = 0; j < i; j++) {
-            int r = RandomUtil.nextInt(0, 62);
-            s.append(StringUtil.substring(STR, r, r + 1));
-        }
-        return s.toString();
-    }
-
-    private void batchUpdate(DatabaseConnectorInstance connectorInstance, ExecutorService pool, String sql, List<Object[]> dataList, int batchSize) {
-        int total = dataList.size();
-        int taskSize = total % batchSize == 0 ? total / batchSize : total / batchSize + 1;
-        final CountDownLatch latch = new CountDownLatch(taskSize);
-        int fromIndex = 0;
-        int toIndex = batchSize;
-        for (int i = 0; i < taskSize; i++) {
-            final List<Object[]> data;
-            if (toIndex > total) {
-                toIndex = fromIndex + (total % batchSize);
-                data = dataList.subList(fromIndex, toIndex);
-            } else {
-                data = dataList.subList(fromIndex, toIndex);
-                fromIndex += batchSize;
-                toIndex += batchSize;
-            }
-
-            pool.submit(() -> {
-                try {
-                    connectorInstance.execute(databaseTemplate -> databaseTemplate.batchUpdate(sql, data));
-                } catch (Exception e) {
-                    logger.error(e.getMessage());
-                } finally {
-                    latch.countDown();
-                }
-            });
-        }
-        try {
-            latch.await();
-        } catch (InterruptedException e) {
-            logger.error(e.getMessage());
-        }
-    }
-
-    @Test
-    public void testReadSchema() {
-        getTables(createOracleConfig(), "test", "AE86", "MY_ORG");
-        getTables(createOracleConfig(), "test", "AE86", null);
-
-        getTables(createMysqlConfig(), "test", "root", "MY_ORG");
-        getTables(createMysqlConfig(), "test", "root", null);
-
-        getTables(createSqlServerConfig(), "test", "dbo", "MY_ORG");
-        getTables(createSqlServerConfig(), "test", "dbo", null);
-
-        getTables(createPostgresConfig(), "postgres", "public", "MY_ORG");
-        getTables(createPostgresConfig(), "postgres", "public", null);
-    }
-
-    @Test
-    public void testGetColumnsDetails() {
-        final String schema = "root";
-        final String tableNamePattern = "sw_test";
-        final DatabaseConnectorInstance connectorInstance = new DatabaseConnectorInstance(createMysqlConfig());
-        connectorInstance.execute(databaseTemplate -> {
-            SimpleConnection connection = databaseTemplate.getSimpleConnection();
-            Connection conn = connection.getConnection();
-            String databaseCatalog = conn.getCatalog();
-            String schemaNamePattern = null == schema ? conn.getSchema() : schema;
-            List<Field> fields = new ArrayList<>();
-            DatabaseMetaData metaData = conn.getMetaData();
-            ResultSet columnMetadata = metaData.getColumns(databaseCatalog, schemaNamePattern, tableNamePattern, null);
-            while (columnMetadata.next()) {
-                String columnName = columnMetadata.getString(4);
-                int columnType = columnMetadata.getInt(5);
-                String typeName = columnMetadata.getString(6);
-                fields.add(new Field(columnName, typeName, columnType));
-            }
-            return fields;
-        });
-    }
-
-    private List<Table> getTables(DatabaseConfig config, final String catalog, final String schema, final String tableNamePattern) {
-        final DatabaseConnectorInstance connectorInstance = new DatabaseConnectorInstance(config);
-        List<Table> tables = new ArrayList<>();
-        connectorInstance.execute(databaseTemplate -> {
-            SimpleConnection connection = databaseTemplate.getSimpleConnection();
-            Connection conn = connection.getConnection();
-            String databaseCatalog = null == catalog ? conn.getCatalog() : catalog;
-            String schemaNamePattern = null == schema ? conn.getSchema() : schema;
-            String[] types = {TableTypeEnum.TABLE.getCode(), TableTypeEnum.VIEW.getCode(), TableTypeEnum.MATERIALIZED_VIEW.getCode()};
-            final ResultSet rs = conn.getMetaData().getTables(databaseCatalog, schemaNamePattern, tableNamePattern, types);
-            while (rs.next()) {
-                final String tableName = rs.getString("TABLE_NAME");
-                final String tableType = rs.getString("TABLE_TYPE");
-                tables.add(new Table(tableName, tableType));
-            }
-            return tables;
-        });
-
-        logger.info("\r 表总数{}", tables.size());
-        tables.forEach(t -> logger.info("{} {}", t.getName(), t.getType()));
-
-        return tables;
-    }
-
-    private DatabaseConfig createSqlServerConfig() {
-        DatabaseConfig config = new DatabaseConfig();
-        config.setUrl("jdbc:sqlserver://127.0.0.1:1433;DatabaseName=test");
-        config.setUsername("sa");
-        config.setPassword("123");
-        config.setDriverClassName("com.microsoft.sqlserver.jdbc.SQLServerDriver");
-        return config;
-    }
-
-    private DatabaseConfig createOracleConfig() {
-        DatabaseConfig config = new DatabaseConfig();
-        config.setUrl("jdbc:oracle:thin:@127.0.0.1:1521:ORCL");
-        config.setUsername("ae86");
-        config.setPassword("123");
-        config.setDriverClassName("oracle.jdbc.OracleDriver");
-        return config;
-    }
-
-    private DatabaseConfig createMysqlConfig() {
-        DatabaseConfig config = new DatabaseConfig();
-        config.setUrl("jdbc:mysql://127.0.0.1:3305/test?rewriteBatchedStatements=true&useUnicode=true&characterEncoding=UTF8&serverTimezone=Asia/Shanghai&useSSL=false&verifyServerCertificate=false&autoReconnect=true&failOverReadOnly=false");
-        config.setUsername("root");
-        config.setPassword("123");
-        config.setDriverClassName("com.mysql.cj.jdbc.Driver");
-        return config;
-    }
-
-    private DatabaseConfig createPostgresConfig() {
-        DatabaseConfig config = new DatabaseConfig();
-        config.setUrl("jdbc:postgresql://127.0.0.1:5432/postgres");
-        config.setUsername("postgres");
-        config.setPassword("123456");
-        config.setDriverClassName("org.postgresql.Driver");
-        return config;
-    }
-}

dbsyncer-parser/pom.xml

@@ -25,6 +25,11 @@
             <version>${project.parent.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.github.jsqlparser</groupId>
+            <artifactId>jsqlparser</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>

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

@@ -14,7 +14,6 @@ import org.dbsyncer.sdk.config.ReaderConfig;
 import org.dbsyncer.sdk.config.SqlBuilderConfig;
 import org.dbsyncer.sdk.config.WriterBatchConfig;
 import org.dbsyncer.sdk.connector.AbstractConnector;
-import org.dbsyncer.sdk.connector.ConfigValidator;
 import org.dbsyncer.sdk.connector.ConnectorInstance;
 import org.dbsyncer.sdk.connector.database.ds.SimpleConnection;
 import org.dbsyncer.sdk.constant.ConnectorConstant;
@@ -88,11 +87,6 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector implem
         return new DatabaseConnectorInstance(config);
     }
 
-    @Override
-    public ConfigValidator getConfigValidator() {
-        return null;
-    }
-
     @Override
     public void disconnect(DatabaseConnectorInstance connectorInstance) {
         connectorInstance.close();

dbsyncer-storage/pom.xml

@@ -24,6 +24,12 @@
             <artifactId>protobuf-java</artifactId>
         </dependency>
 
+        <!-- oracle-driver -->
+        <dependency>
+            <groupId>com.oracle</groupId>
+            <artifactId>ojdbc6</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-log4j2</artifactId>