dromara/dbswitch
1
0
Fork 0
mirror of https://gitee.com/dromara/dbswitch.git synced 2025-08-29 00:44:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

代码优化与错误拼写修正

Browse Source
This commit is contained in:
inrgihc
2021-06-01 23:35:00 +08:00
parent 425183fe73
commit 044f97d5c3
113 changed files with 2338 additions and 2340 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
LICENSE
View File

@@ -1,5 +1,5 @@
// dbswitch - 一个Java语言编写的异构数据库迁移工具
// Copyright (c) 2019, tang. All rights reserved.
// Copyright (c) 2020, tang(inrgihc@126.com). All rights reserved.
// Use of this source code is governed by a BSD-style license
// https://gitee.com/inrgihc/dbswitch
//

3
build.sh
View File

@@ -1,4 +1,3 @@
#!/bin/sh
mvn clean -f pom.xml
mvn package -f pom.xml -Dmaven.test.skip=true
mvn clean package -Dmaven.test.skip=true

2
dbswitch-common/src/main/java/com/gitee/dbswitch/common/constant/DatabaseTypeEnum.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.common.constant;

108
dbswitch-common/src/main/java/com/gitee/dbswitch/common/util/CommonUtils.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.common.util;
@@ -14,63 +14,71 @@ import com.gitee.dbswitch.common.constant.DatabaseTypeEnum;
/**
* 普通工具类
*
* @author tang
*
* @author tang
*/
public final class CommonUtils {
private CommonUtils() {
private CommonUtils() {
}
}
/**
* 根据数据库类型获取表的全名：schema.table
*
* @param dbType 数据库类型
* @param schema schema名
* @param table table名
* @return 表的全名字符串
*/
public static String getTableFullNameByDatabase(DatabaseTypeEnum dbType, String schema, String table) {
if (dbType == DatabaseTypeEnum.MYSQL || dbType == DatabaseTypeEnum.MARIADB) {
return String.format("`%s`.`%s`", schema, table);
} else if (dbType == DatabaseTypeEnum.SQLSERVER || dbType == DatabaseTypeEnum.SQLSERVER2000) {
return String.format("[%s].[%s]", schema, table);
} else {
return String.format("\"%s\".\"%s\"", schema, table);
}
}
public static String getTableFullNameByDatabase(DatabaseTypeEnum dbtype, String schemaName, String tableName) {
if (dbtype == DatabaseTypeEnum.MYSQL) {
return String.format("`%s`.`%s`", schemaName, tableName);
} else if (dbtype == DatabaseTypeEnum.SQLSERVER) {
return String.format("[%s].[%s]", schemaName, tableName);
} else {
return String.format("\"%s\".\"%s\"", schemaName, tableName);
}
}
/**
* 拼接SELECT查询指定字段的SQL语句
*
* @param dbType 数据库类型
* @param schema schema名
* @param table table名
* @param columns 列名列表
* @return SQL语句字符串
*/
public static String getSelectColumnsSQL(DatabaseTypeEnum dbType, String schema, String table, List<String> columns) {
StringBuilder sb = new StringBuilder();
sb.append(" SELECT ");
for (int i = 0; i < columns.size(); ++i) {
String field = columns.get(i);
String quoteField = quoteString(dbType, field);
sb.append(quoteField);
public static String getSelectColumnsSQL(DatabaseTypeEnum dbtype, String schema, String table, List<String> columns) {
StringBuilder sb = new StringBuilder();
sb.append(" SELECT ");
for (int i = 0; i < columns.size(); ++i) {
String field = columns.get(i);
String quoteField = quoteString(dbtype, field);
sb.append(quoteField);
if (i < columns.size() - 1) {
sb.append(",");
}
}
sb.append(" FROM ");
if (null != schema && !schema.isEmpty()) {
sb.append(quoteString(dbType, schema));
sb.append(".");
}
sb.append(quoteString(dbType, table));
if (i < columns.size() - 1) {
sb.append(",");
}
}
sb.append(" FROM ");
if (null != schema && !schema.isEmpty()) {
sb.append(quoteString(dbtype, schema));
sb.append(".");
}
sb.append(quoteString(dbtype, table));
return sb.toString();
}
return sb.toString();
}
private static String quoteString(DatabaseTypeEnum dbType, String keyName) {
if (dbType == DatabaseTypeEnum.MYSQL || dbType == DatabaseTypeEnum.MARIADB) {
return String.format("`%s`", keyName);
} else if (dbType == DatabaseTypeEnum.SQLSERVER || dbType == DatabaseTypeEnum.SQLSERVER2000) {
return String.format("[%s]", keyName);
} else {
return String.format("\"%s\"", keyName);
}
}
private static String quoteString(DatabaseTypeEnum dbtype, String keyName) {
if (dbtype == DatabaseTypeEnum.MYSQL) {
return String.format("`%s`", keyName);
} else if (dbtype == DatabaseTypeEnum.SQLSERVER) {
return String.format("[%s]", keyName);
} else {
return String.format("\"%s\"", keyName);
}
}
public static String getQuotationChar(DatabaseTypeEnum dbtype) {
if (dbtype == DatabaseTypeEnum.MYSQL) {
return "`";
} else {
return "\"";
}
}
}

3
dbswitch-core/src/main/java/com/gitee/dbswitch/core/constant/Const.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.core.constant;
@@ -71,7 +71,6 @@ public final class Const {
* Constructor Function
*/
private Const() {
}
}

2
dbswitch-core/src/main/java/com/gitee/dbswitch/core/constant/DBTableType.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.core.constant;

2
dbswitch-core/src/main/java/com/gitee/dbswitch/core/database/AbstractDatabase.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.core.database;

2
dbswitch-core/src/main/java/com/gitee/dbswitch/core/database/DatabaseFactory.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.core.database;

2
dbswitch-core/src/main/java/com/gitee/dbswitch/core/database/IDatabaseInterface.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.core.database;

5
dbswitch-core/src/main/java/com/gitee/dbswitch/core/database/impl/DatabaseDB2Impl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.core.database.impl;
@@ -16,6 +16,7 @@ import com.gitee.dbswitch.core.database.AbstractDatabase;
import com.gitee.dbswitch.core.database.IDatabaseInterface;
import com.gitee.dbswitch.core.model.ColumnDescription;
import com.gitee.dbswitch.core.model.ColumnMetaData;
import org.springframework.util.CollectionUtils;
/**
* 支持DB2数据库的元信息实现
@@ -117,7 +118,7 @@ public class DatabaseDB2Impl extends AbstractDatabase implements IDatabaseInterf
retval += ")";
}
if (null != pks && pks.contains(fieldname)) {
if (!CollectionUtils.isEmpty(pks) && pks.contains(fieldname)) {
retval += " NOT NULL";
}

2
dbswitch-core/src/main/java/com/gitee/dbswitch/core/database/impl/DatabaseDmImpl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.core.database.impl;

6
dbswitch-core/src/main/java/com/gitee/dbswitch/core/database/impl/DatabaseGreenplumImpl.java
View File

@@ -4,11 +4,12 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.core.database.impl;
import java.util.Collections;
import java.util.List;
import com.alibaba.druid.sql.SQLUtils;
import com.gitee.dbswitch.core.constant.Const;
@@ -17,6 +18,7 @@ import com.gitee.dbswitch.core.database.AbstractDatabase;
import com.gitee.dbswitch.core.database.IDatabaseInterface;
import com.gitee.dbswitch.core.model.ColumnDescription;
import com.gitee.dbswitch.core.model.ColumnMetaData;
import org.springframework.util.CollectionUtils;
/**
* 支持Greenplum数据库的元信息实现
@@ -76,7 +78,7 @@ public class DatabaseGreenplumImpl extends AbstractDatabase implements IDatabase
case ColumnMetaData.TYPE_NUMBER:
case ColumnMetaData.TYPE_INTEGER:
case ColumnMetaData.TYPE_BIGNUMBER:
if (null!=pks && pks.contains(fieldname)) {
if (!CollectionUtils.isEmpty(pks) && pks.contains(fieldname)) {
if (useAutoInc) {
retval += "BIGSERIAL";
} else {

5
dbswitch-core/src/main/java/com/gitee/dbswitch/core/database/impl/DatabaseKingbaseImpl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.core.database.impl;
@@ -17,6 +17,7 @@ import com.gitee.dbswitch.core.database.AbstractDatabase;
import com.gitee.dbswitch.core.database.IDatabaseInterface;
import com.gitee.dbswitch.core.model.ColumnDescription;
import com.gitee.dbswitch.core.model.ColumnMetaData;
import org.springframework.util.CollectionUtils;
/**
* 支持Kingbase数据库的元信息实现
@@ -76,7 +77,7 @@ public class DatabaseKingbaseImpl extends AbstractDatabase implements IDatabaseI
case ColumnMetaData.TYPE_NUMBER:
case ColumnMetaData.TYPE_INTEGER:
case ColumnMetaData.TYPE_BIGNUMBER:
if (null != pks && pks.contains(fieldname)) {
if (!CollectionUtils.isEmpty(pks) && pks.contains(fieldname)) {
if (useAutoInc) {
retval += "BIGSERIAL";
} else {

2
dbswitch-core/src/main/java/com/gitee/dbswitch/core/database/impl/DatabaseMariaDBImpl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.core.database.impl;

9
dbswitch-core/src/main/java/com/gitee/dbswitch/core/database/impl/DatabaseMysqlImpl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.core.database.impl;
@@ -26,6 +26,7 @@ import com.gitee.dbswitch.core.model.ColumnMetaData;
import com.gitee.dbswitch.core.model.TableDescription;
import com.gitee.dbswitch.core.util.JdbcOperatorUtils;
import com.gitee.dbswitch.core.util.JdbcUrlUtils;
import org.springframework.util.CollectionUtils;
/**
* 支持MySQL数据库的元信息实现
@@ -145,7 +146,7 @@ public class DatabaseMysqlImpl extends AbstractDatabase implements IDatabaseInte
case ColumnMetaData.TYPE_NUMBER:
case ColumnMetaData.TYPE_INTEGER:
case ColumnMetaData.TYPE_BIGNUMBER:
if (null!=pks && pks.contains(fieldname)) {
if (!CollectionUtils.isEmpty(pks) && pks.contains(fieldname)) {
if (useAutoInc) {
retval += "BIGINT AUTO_INCREMENT NOT NULL";
} else {
@@ -188,7 +189,7 @@ public class DatabaseMysqlImpl extends AbstractDatabase implements IDatabaseInte
retval += "CHAR(1)";
} else if (length < 256) {
retval += "VARCHAR(" + length + ")";
}else if (null!=pks && pks.contains(fieldname)) {
}else if (!CollectionUtils.isEmpty(pks) && pks.contains(fieldname)) {
/*
* MySQL5.6中varchar字段为主键时最大长度为254,例如如下的建表语句在MySQL5.7下能通过，但在MySQL5.6下无法通过：
* create table `t_test`(
@@ -226,7 +227,7 @@ public class DatabaseMysqlImpl extends AbstractDatabase implements IDatabaseInte
@Override
public String getPrimaryKeyAsString(List<String> pks) {
if (!pks.isEmpty()) {
if (!CollectionUtils.isEmpty(pks)) {
StringBuilder sb = new StringBuilder();
sb.append("`");
sb.append(StringUtils.join(pks, "` , `"));

8
dbswitch-core/src/main/java/com/gitee/dbswitch/core/database/impl/DatabaseOracleImpl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.core.database.impl;
@@ -45,7 +45,7 @@ public class DatabaseOracleImpl extends AbstractDatabase implements IDatabaseInt
@Override
public List<TableDescription> queryTableList(String schemaName) {
List<TableDescription> ret = new ArrayList<TableDescription>();
List<TableDescription> ret = new ArrayList<>();
String sql = String.format(
"SELECT \"OWNER\",\"TABLE_NAME\",\"TABLE_TYPE\",\"COMMENTS\" from all_tab_comments where \"OWNER\"='%s'",
schemaName);
@@ -84,7 +84,7 @@ public class DatabaseOracleImpl extends AbstractDatabase implements IDatabaseInt
// Oracle表的主键可以使用如下命令设置主键是否生效
// 使主键失效：alter table tableName disable primary key;
// 使主键恢复：alter table tableName enable primary key;
Set<String> ret = new HashSet<String>();
Set<String> ret = new HashSet<>();
String sql = String.format("SELECT COLUMN_NAME FROM user_cons_columns WHERE owner='%s' and constraint_name = "
+ "(SELECT constraint_name FROM user_constraints WHERE table_name = '%s' AND constraint_type = 'P' and STATUS='ENABLED') ",
schemaName, tableName);
@@ -173,7 +173,7 @@ public class DatabaseOracleImpl extends AbstractDatabase implements IDatabaseInt
} else {
if (length == 1) {
retval.append("NVARCHAR2(1)");
} else if (length > 0) {
} else if (length > 0 && length < 2000) {
// VARCHAR2(size)，size最大值为4000，单位是字节；而NVARCHAR2(size)，size最大值为2000，单位是字符
retval.append("NVARCHAR2(").append(length).append(')');
} else {

7
dbswitch-core/src/main/java/com/gitee/dbswitch/core/database/impl/DatabasePostgresImpl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.core.database.impl;
@@ -17,6 +17,7 @@ import com.gitee.dbswitch.core.database.AbstractDatabase;
import com.gitee.dbswitch.core.database.IDatabaseInterface;
import com.gitee.dbswitch.core.model.ColumnDescription;
import com.gitee.dbswitch.core.model.ColumnMetaData;
import org.springframework.util.CollectionUtils;
/**
* 支持PostgreSQL数据库的元信息实现
@@ -76,7 +77,7 @@ public class DatabasePostgresImpl extends AbstractDatabase implements IDatabaseI
case ColumnMetaData.TYPE_NUMBER:
case ColumnMetaData.TYPE_INTEGER:
case ColumnMetaData.TYPE_BIGNUMBER:
if (null != pks && pks.contains(fieldname)) {
if (!CollectionUtils.isEmpty(pks) && pks.contains(fieldname)) {
if (useAutoInc) {
retval += "BIGSERIAL";
} else {
@@ -112,7 +113,7 @@ public class DatabasePostgresImpl extends AbstractDatabase implements IDatabaseI
if (length < 1 || length >= AbstractDatabase.CLOB_LENGTH) {
retval += "TEXT";
} else {
if (null != pks && pks.contains(fieldname)) {
if (!CollectionUtils.isEmpty(pks) && pks.contains(fieldname)) {
retval += "VARCHAR(" + length + ")";
} else {
retval += "TEXT";

2
dbswitch-core/src/main/java/com/gitee/dbswitch/core/database/impl/DatabaseSqlserver2000Impl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.core.database.impl;

8
dbswitch-core/src/main/java/com/gitee/dbswitch/core/database/impl/DatabaseSqlserverImpl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.core.database.impl;
@@ -26,6 +26,7 @@ import com.gitee.dbswitch.core.model.ColumnDescription;
import com.gitee.dbswitch.core.model.ColumnMetaData;
import com.gitee.dbswitch.core.model.TableDescription;
import com.gitee.dbswitch.core.util.JdbcOperatorUtils;
import org.springframework.util.CollectionUtils;
/**
* 支持SQLServer数据库的元信息实现
@@ -87,7 +88,6 @@ public class DatabaseSqlserverImpl extends AbstractDatabase implements IDatabase
try {
if (null != schemas) {
schemas.close();
schemas = null;
}
} catch (SQLException e) {
}
@@ -228,7 +228,7 @@ public class DatabaseSqlserverImpl extends AbstractDatabase implements IDatabase
case ColumnMetaData.TYPE_NUMBER:
case ColumnMetaData.TYPE_INTEGER:
case ColumnMetaData.TYPE_BIGNUMBER:
if (null != pks && pks.contains(fieldname)) {
if (!CollectionUtils.isEmpty(pks) && pks.contains(fieldname)) {
if (useAutoInc) {
retval += "BIGINT IDENTITY(0,1)";
} else {
@@ -288,7 +288,7 @@ public class DatabaseSqlserverImpl extends AbstractDatabase implements IDatabase
@Override
public String getPrimaryKeyAsString(List<String> pks) {
if (!pks.isEmpty()) {
if (!CollectionUtils.isEmpty(pks)) {
StringBuilder sb = new StringBuilder();
sb.append("[");
sb.append(StringUtils.join(pks, "] , ["));

2
dbswitch-core/src/main/java/com/gitee/dbswitch/core/model/ColumnDescription.java
View File

@@ -13,8 +13,8 @@ import com.gitee.dbswitch.common.constant.DatabaseTypeEnum;
/**
* 数据库列描述符信息定义(Column Description)
* @author tang
*
* @author tang
*/
public class ColumnDescription {

8
dbswitch-core/src/main/java/com/gitee/dbswitch/core/model/ColumnMetaData.java
View File

@@ -14,6 +14,7 @@ import com.gitee.dbswitch.core.database.AbstractDatabase;
/**
* 数据库表列的元信息
*
* @author tang
*
*/
@@ -81,10 +82,7 @@ public class ColumnMetaData {
/**
* Constructor function
*
* @param name
* @param type
* @param length
* @param precision
* @param desc
*/
public ColumnMetaData(ColumnDescription desc) {
this.create(desc);
@@ -257,7 +255,7 @@ public class ColumnMetaData {
length = desc.getDisplaySize();
break;
case java.sql.Types.LONGVARCHAR: // Character Large Object
case java.sql.Types.LONGVARCHAR:
case java.sql.Types.CLOB:
case java.sql.Types.NCLOB:
valtype = ColumnMetaData.TYPE_STRING;

4
dbswitch-core/src/main/java/com/gitee/dbswitch/core/model/DatabaseDescription.java
View File

@@ -13,6 +13,7 @@ import com.gitee.dbswitch.common.constant.DatabaseTypeEnum;
/**
* 数据库连接描述符信息定义(Database Description)
*
* @author tang
*
*/
@@ -20,7 +21,8 @@ public class DatabaseDescription {
protected DatabaseTypeEnum type;
protected String host;
protected int port;
protected String mode;//对于Oracle数据库的模式，可取范围为：sid,servicename,tnsname三种
/** 对于Oracle数据库的模式，可取范围为：sid,serviceName,TNSName三种 */
protected String mode;
protected String dbname;
protected String charset;
protected String username;

1
dbswitch-core/src/main/java/com/gitee/dbswitch/core/model/TableDescription.java
View File

@@ -13,6 +13,7 @@ import com.gitee.dbswitch.core.constant.DBTableType;
/**
* 数据库表描述符信息定义(Table Description)
*
* @author tang
*
*/

7
dbswitch-core/src/main/java/com/gitee/dbswitch/core/util/GenerateSqlUtils.java
View File

@@ -11,7 +11,6 @@ package com.gitee.dbswitch.core.util;
import java.util.List;
import java.util.stream.Collectors;
import com.gitee.dbswitch.common.constant.DatabaseTypeEnum;
import com.gitee.dbswitch.core.constant.Const;
import com.gitee.dbswitch.core.database.AbstractDatabase;
@@ -30,8 +29,10 @@ public class GenerateSqlUtils {
public static String getDDLCreateTableSQL(DatabaseTypeEnum type, List<ColumnDescription> fieldNames,
List<String> primaryKeys, String schemaName, String tableName, boolean autoIncr) {
StringBuilder retval = new StringBuilder();
List<String> pks = fieldNames.stream().filter((cd) -> primaryKeys.contains(cd.getFieldName()))
.map((cd) -> cd.getFieldName()).collect(Collectors.toList());
List<String> pks = fieldNames.stream()
.filter((cd) -> primaryKeys.contains(cd.getFieldName()))
.map((cd) -> cd.getFieldName())
.collect(Collectors.toList());
AbstractDatabase db = DatabaseFactory.getDatabaseInstance(type);

6
dbswitch-data/pom.xml
View File

@@ -62,6 +62,12 @@
<artifactId>postgresql</artifactId>
</dependency>
<dependency>
<groupId>com.carrotsearch</groupId>
<artifactId>java-sizeof</artifactId>
<version>0.0.5</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>

2
dbswitch-data/src/main/java/com/gitee/dbswitch/data/DataSyncApplication.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.data;

2
dbswitch-data/src/main/java/com/gitee/dbswitch/data/config/DbswichProperties.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.data.config;

2
dbswitch-data/src/main/java/com/gitee/dbswitch/data/config/PropertiesConfig.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.data.config;

32
dbswitch-data/src/main/java/com/gitee/dbswitch/data/service/MainService.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.data.service;
@@ -17,8 +17,8 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.regex.Pattern;
import javax.sql.DataSource;
import com.carrotsearch.sizeof.RamUsageEstimator;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
@@ -58,6 +58,8 @@ import lombok.extern.slf4j.Slf4j;
@Service("MainService")
public class MainService {
private final long MAX_CACHE_BYTES_SIZE = 512 * 1024 * 1024;
private ObjectMapper jackson = new ObjectMapper();
@Autowired
@@ -264,7 +266,8 @@ public class MainService {
StatementResultSet srs = sourceOperator.queryTableData(tableDescription.getSchemaName(),
tableDescription.getTableName(), fields);
List<Object[]> cache = new LinkedList<Object[]>();
List<Object[]> cache = new LinkedList<>();
long cacheBytes=0;
long totalCount = 0;
try {
ResultSet rs = srs.getResultset();
@@ -280,12 +283,14 @@ public class MainService {
}
cache.add(record);
cacheBytes += RamUsageEstimator.shallowSizeOf(record);
++totalCount;
if (cache.size() >= BATCH_SIZE) {
if (cache.size() >= BATCH_SIZE || cacheBytes >= MAX_CACHE_BYTES_SIZE) {
long ret = writer.write(fields, cache);
log.info("[FullCoverSynch] handle table [{}] data count: {}", fullTableName, ret);
cache.clear();
cacheBytes = 0;
}
}
@@ -348,6 +353,7 @@ public class MainService {
private long countUpdate = 0;
private long countDelete = 0;
private long count = 0;
private long cacheBytes=0;
private List<Object[]> cacheInsert = new LinkedList<Object[]>();
private List<Object[]> cacheUpdate = new LinkedList<Object[]>();
private List<Object[]> cacheDelete = new LinkedList<Object[]>();
@@ -365,6 +371,7 @@ public class MainService {
countDelete++;
}
cacheBytes+=RamUsageEstimator.shallowSizeOf(record);
count++;
checkFull(fields);
}
@@ -376,7 +383,7 @@ public class MainService {
*/
private void checkFull(List<String> fields) {
if (cacheInsert.size() >= BATCH_SIZE || cacheUpdate.size() >= BATCH_SIZE
|| cacheDelete.size() >= BATCH_SIZE) {
|| cacheDelete.size() >= BATCH_SIZE || cacheBytes >= MAX_CACHE_BYTES_SIZE) {
if (cacheDelete.size() > 0) {
doDelete(fields);
}
@@ -388,6 +395,8 @@ public class MainService {
if (cacheUpdate.size() > 0) {
doUpdate(fields);
}
cacheBytes = 0;
}
}
@@ -433,7 +442,7 @@ public class MainService {
/**
* 创建于指定数据库连接描述符的连接池
*
* @param dbdesc 数据库连接描述符
* @param description 数据库连接描述符
* @return HikariDataSource连接池
*/
private HikariDataSource createSourceDataSource(DbswichProperties.SourceDataSourceProperties description) {
@@ -459,7 +468,7 @@ public class MainService {
/**
* 创建于指定数据库连接描述符的连接池
*
* @param dbdesc 数据库连接描述符
* @param description 数据库连接描述符
* @return HikariDataSource连接池
*/
private HikariDataSource createTargetDataSource(DbswichProperties.TargetDataSourceProperties description) {
@@ -499,15 +508,16 @@ public class MainService {
/**
* 检查MySQL数据库表的存储引擎是否为Innodb
*
*
* @param shemaName schema名
* @param tableName table名
* @param dataSource 数据源
* @param task 任务实体
* @return 为Innodb存储引擎时返回True,否在为false
* @return 为Innodb存储引擎时返回True, 否在为false
*/
private boolean isMysqlInodbStorageEngine(String shemaName, String tableName, DataSource dataSource) {
String sql = "SELECT count(*) as total FROM information_schema.tables WHERE table_schema=? AND table_name=? AND ENGINE='InnoDB'";
JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
return jdbcTemplate.queryForObject(sql, new Object[] { shemaName, tableName }, Integer.class) > 0;
return jdbcTemplate.queryForObject(sql, new Object[]{shemaName, tableName}, Integer.class) > 0;
}
/**

2
dbswitch-data/src/main/java/com/gitee/dbswitch/data/util/JdbcTemplateUtils.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.data.util;

2
dbswitch-dbchange/src/main/java/com/gitee/dbswitch/dbchange/ChangeCaculatorService.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbchange;

2
dbswitch-dbchange/src/main/java/com/gitee/dbswitch/dbchange/IDatabaseChangeCaculator.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbchange;

2
dbswitch-dbchange/src/main/java/com/gitee/dbswitch/dbchange/IDatabaseRowHandler.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbchange;

2
dbswitch-dbchange/src/main/java/com/gitee/dbswitch/dbchange/RecordChangeTypeEnum.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbchange;

2
dbswitch-dbchange/src/main/java/com/gitee/dbswitch/dbchange/pojo/TaskParamBean.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbchange.pojo;

2
dbswitch-dbchange/src/main/java/com/gitee/dbswitch/dbchange/util/JdbcTypesUtils.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbchange.util;

2
dbswitch-dbcommon/src/main/java/com/gitee/dbswitch/dbcommon/constant/Constants.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbcommon.constant;

145
dbswitch-dbcommon/src/main/java/com/gitee/dbswitch/dbcommon/database/AbstractDatabaseOperator.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbcommon.database;
@@ -15,6 +15,7 @@ import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Objects;
import javax.sql.DataSource;
import org.springframework.jdbc.support.JdbcUtils;
import com.gitee.dbswitch.dbcommon.constant.Constants;
import com.gitee.dbswitch.dbcommon.pojo.StatementResultSet;
@@ -22,94 +23,90 @@ import lombok.extern.slf4j.Slf4j;
/**
* 数据读取抽象基类
*
* @author tang
*
* @author tang
*/
@Slf4j
public abstract class AbstractDatabaseOperator implements IDatabaseOperator {
protected DataSource dataSource;
protected DataSource dataSource;
protected int fetchSize;
protected int fetchSize;
public AbstractDatabaseOperator(DataSource dataSource) {
this.dataSource = Objects.requireNonNull(dataSource, "数据源非法,为null");
this.fetchSize = Constants.DEFAULT_FETCH_SIZE;
}
public AbstractDatabaseOperator(DataSource dataSource) {
this.dataSource = Objects.requireNonNull(dataSource, "数据源非法,为null");
this.fetchSize = Constants.DEFAULT_FETCH_SIZE;
}
@Override
public DataSource getDataSource() {
return this.dataSource;
}
@Override
public DataSource getDataSource() {
return this.dataSource;
}
@Override
public int getFetchSize() {
return this.fetchSize;
}
@Override
public int getFetchSize() {
return this.fetchSize;
}
@Override
public void setFetchSize(int size) {
if (size < Constants.MINIMUM_FETCH_SIZE) {
throw new IllegalArgumentException("设置的批量处理行数的大小fetchSize不得小于" + Constants.MINIMUM_FETCH_SIZE);
}
@Override
public void setFetchSize(int size) {
if (size < Constants.MINIMUM_FETCH_SIZE) {
throw new IllegalArgumentException("设置的批量处理行数的大小fetchSize不得小于" + Constants.MINIMUM_FETCH_SIZE);
}
this.fetchSize = size;
}
this.fetchSize = size;
}
/**
* 已经指定的查询SQL语句查询数据结果集
*
* @param dataSource 数据源
* @param sql 查询的SQL语句
* @param fetchSize 批处理大小
* @return 结果集包装对象
*/
protected final StatementResultSet selectTableData(String sql, int fetchSize) {
if (log.isDebugEnabled()) {
log.debug("Query table data sql :{}", sql);
}
/**
* 已经指定的查询SQL语句查询数据结果集
*
* @param sql 查询的SQL语句
* @param fetchSize 批处理大小
* @return 结果集包装对象
*/
protected final StatementResultSet selectTableData(String sql, int fetchSize) {
if (log.isDebugEnabled()) {
log.debug("Query table data sql :{}", sql);
}
try {
StatementResultSet srs = new StatementResultSet();
srs.setConnection(dataSource.getConnection());
srs.setAutoCommit(srs.getConnection().getAutoCommit());
srs.getConnection().setAutoCommit(false);
srs.setStatement(
srs.getConnection().createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY));
srs.getStatement().setQueryTimeout(Constants.DEFAULT_QUERY_TIMEOUT_SECONDS);
srs.getStatement().setFetchSize(fetchSize);
srs.setResultset(srs.getStatement().executeQuery(sql));
return srs;
} catch (Throwable t) {
throw new RuntimeException(t);
}
}
try {
StatementResultSet srs = new StatementResultSet();
srs.setConnection(dataSource.getConnection());
srs.setAutoCommit(srs.getConnection().getAutoCommit());
srs.getConnection().setAutoCommit(false);
srs.setStatement(srs.getConnection().createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY));
srs.getStatement().setQueryTimeout(Constants.DEFAULT_QUERY_TIMEOUT_SECONDS);
srs.getStatement().setFetchSize(fetchSize);
srs.setResultset(srs.getStatement().executeQuery(sql));
return srs;
} catch (Throwable t) {
throw new RuntimeException(t);
}
}
/**
* 执行写SQL操作
*
* @param dataSource 数据源
* @param sql 写SQL语句
*/
protected final int executeSql(String sql) {
Connection connection = null;
PreparedStatement pstmt = null;
/**
* 执行写SQL操作
*
* @param sql 写SQL语句
*/
protected final int executeSql(String sql) {
Connection connection = null;
PreparedStatement pstmt = null;
if (log.isDebugEnabled()) {
log.debug("Execute sql :{}", sql);
}
if (log.isDebugEnabled()) {
log.debug("Execute sql :{}", sql);
}
try {
connection = dataSource.getConnection();
pstmt = connection.prepareStatement(sql);
return pstmt.executeUpdate();
} catch (SQLException e) {
throw new RuntimeException(e);
} finally {
JdbcUtils.closeStatement(pstmt);
JdbcUtils.closeConnection(connection);
}
}
try {
connection = dataSource.getConnection();
pstmt = connection.prepareStatement(sql);
return pstmt.executeUpdate();
} catch (SQLException e) {
throw new RuntimeException(e);
} finally {
JdbcUtils.closeStatement(pstmt);
JdbcUtils.closeConnection(connection);
}
}
}

68
dbswitch-dbcommon/src/main/java/com/gitee/dbswitch/dbcommon/database/DatabaseOperatorFactory.java
View File

@@ -4,14 +4,14 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbcommon.database;
import java.util.Map;
import java.lang.reflect.Constructor;
import java.util.HashMap;
import java.util.function.Function;
import javax.sql.DataSource;
import com.gitee.dbswitch.dbcommon.database.impl.DB2DatabaseOperator;
import com.gitee.dbswitch.dbcommon.database.impl.DmDatabaseOperator;
@@ -25,50 +25,40 @@ import com.gitee.dbswitch.dbcommon.util.DatabaseAwareUtils;
/**
* 数据库操作器构造工厂类
*
* @author tang
*
* @author tang
*/
public final class DatabaseOperatorFactory {
private static final Map<String, String> DATABASE_OPERATOR_MAPPER = new HashMap<String, String>() {
private static final Map<String, Function<DataSource, IDatabaseOperator>> DATABASE_OPERATOR_MAPPER = new HashMap<String, Function<DataSource, IDatabaseOperator>>() {
private static final long serialVersionUID = -5278835613240515265L;
private static final long serialVersionUID = -5278835613240515265L;
{
put("MYSQL", MysqlDatabaseOperator.class.getName());
put("ORACLE", OracleDatabaseOperator.class.getName());
put("SQLSERVER", SqlServerDatabaseOperator.class.getName());
put("POSTGRESQL", PostgreSqlDatabaseOperator.class.getName());
put("GREENPLUM", GreenplumDatabaseOperator.class.getName());
put("DB2", DB2DatabaseOperator.class.getName());
put("DM", DmDatabaseOperator.class.getName());
put("KINGBASE", KingbaseDatabaseOperator.class.getName());
}
};
{
put("MYSQL", MysqlDatabaseOperator::new);
put("ORACLE", OracleDatabaseOperator::new);
put("SQLSERVER", SqlServerDatabaseOperator::new);
put("POSTGRESQL", PostgreSqlDatabaseOperator::new);
put("GREENPLUM", GreenplumDatabaseOperator::new);
put("DB2", DB2DatabaseOperator::new);
put("DM", DmDatabaseOperator::new);
put("KINGBASE", KingbaseDatabaseOperator::new);
}
};
/**
* 根据数据源获取数据的读取操作器
*
* @param dataSource 数据库源
* @return 指定类型的数据库读取器
*/
public static IDatabaseOperator createDatabaseOperator(DataSource dataSource) {
String type = DatabaseAwareUtils.getDatabaseNameByDataSource(dataSource).toUpperCase();
if (DATABASE_OPERATOR_MAPPER.containsKey(type)) {
String className = DATABASE_OPERATOR_MAPPER.get(type);
try {
Class<?> clazz = Class.forName(className);
Class<?>[] paraTypes = { DataSource.class };
Object[] paraValues = { dataSource };
Constructor<?> cons = clazz.getConstructor(paraTypes);
return (IDatabaseOperator) cons.newInstance(paraValues);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
/**
* 根据数据源获取数据的读取操作器
*
* @param dataSource 数据库源
* @return 指定类型的数据库读取器
*/
public static IDatabaseOperator createDatabaseOperator(DataSource dataSource) {
String type = DatabaseAwareUtils.getDatabaseNameByDataSource(dataSource).toUpperCase();
if (!DATABASE_OPERATOR_MAPPER.containsKey(type)) {
throw new RuntimeException(String.format("[dbcommon] Unkown Supported database type (%s)", type));
}
throw new RuntimeException(String.format("[dbcommon] Unkown Supported database type (%s)", type));
}
return DATABASE_OPERATOR_MAPPER.get(type).apply(dataSource);
}
}

2
dbswitch-dbcommon/src/main/java/com/gitee/dbswitch/dbcommon/database/IDatabaseOperator.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbcommon.database;

2
dbswitch-dbcommon/src/main/java/com/gitee/dbswitch/dbcommon/database/impl/DB2DatabaseOperator.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbcommon.database.impl;

15
dbswitch-dbcommon/src/main/java/com/gitee/dbswitch/dbcommon/database/impl/DmDatabaseOperator.java
View File

@@ -1,8 +1,23 @@
// Copyright tang. All rights reserved.
// https://gitee.com/inrgihc/dbswitch
//
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbcommon.database.impl;
import javax.sql.DataSource;
import com.gitee.dbswitch.dbcommon.database.IDatabaseOperator;
/**
* DM数据库实现类
*
* @author tang
*
*/
public class DmDatabaseOperator extends OracleDatabaseOperator implements IDatabaseOperator {
public DmDatabaseOperator(DataSource dataSource) {

2
dbswitch-dbcommon/src/main/java/com/gitee/dbswitch/dbcommon/database/impl/GreenplumDatabaseOperator.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbcommon.database.impl;

15
dbswitch-dbcommon/src/main/java/com/gitee/dbswitch/dbcommon/database/impl/KingbaseDatabaseOperator.java
View File

@@ -1,5 +1,20 @@
// Copyright tang. All rights reserved.
// https://gitee.com/inrgihc/dbswitch
//
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbcommon.database.impl;
/**
* Kingbase8数据库实现类
*
* @author tang
*
*/
import javax.sql.DataSource;
import com.gitee.dbswitch.dbcommon.database.IDatabaseOperator;

2
dbswitch-dbcommon/src/main/java/com/gitee/dbswitch/dbcommon/database/impl/MysqlDatabaseOperator.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbcommon.database.impl;

2
dbswitch-dbcommon/src/main/java/com/gitee/dbswitch/dbcommon/database/impl/OracleDatabaseOperator.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbcommon.database.impl;

2
dbswitch-dbcommon/src/main/java/com/gitee/dbswitch/dbcommon/database/impl/PostgreSqlDatabaseOperator.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbcommon.database.impl;

2
dbswitch-dbcommon/src/main/java/com/gitee/dbswitch/dbcommon/database/impl/SqlServerDatabaseOperator.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbcommon.database.impl;

2
dbswitch-dbcommon/src/main/java/com/gitee/dbswitch/dbcommon/pojo/StatementResultSet.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbcommon.pojo;

2
dbswitch-dbcommon/src/main/java/com/gitee/dbswitch/dbcommon/util/DatabaseAwareUtils.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbcommon.util;

8
dbswitch-dbcommon/src/main/java/com/gitee/dbswitch/dbcommon/util/JdbcMetaDataUtils.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbcommon.util;
@@ -53,7 +53,7 @@ public class JdbcMetaDataUtils {
String columnName = columns.getString("COLUMN_NAME");
result.add(columnName);
}
return new ArrayList<String>(result);
return new ArrayList<>(result);
} catch (Throwable t) {
throw new RuntimeException(t);
} finally {
@@ -70,7 +70,7 @@ public class JdbcMetaDataUtils {
* @return 主键字段名称列表
*/
public List<String> queryTablePrimaryKeys(String schemaName, String tableName) {
Set<String> result = new HashSet<String>();
Set<String> result = new HashSet<>();
Connection connection = null;
try {
@@ -81,7 +81,7 @@ public class JdbcMetaDataUtils {
String columnName = columns.getString("COLUMN_NAME");
result.add(columnName);
}
return new ArrayList<String>(result);
return new ArrayList<>(result);
} catch (Throwable t) {
throw new RuntimeException(t);
} finally {

470
dbswitch-dbsynch/src/main/java/com/gitee/dbswitch/dbsynch/AbstractDatabaseSynchronize.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbsynch;
@@ -12,7 +12,6 @@ package com.gitee.dbswitch.dbsynch;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
@@ -20,8 +19,6 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import javax.sql.DataSource;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.ConnectionCallback;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.jdbc.support.JdbcUtils;
@@ -34,276 +31,255 @@ import lombok.extern.slf4j.Slf4j;
/**
* 数据同步抽象基类
*
* @author tang
*
* @author tang
*/
@Slf4j
public abstract class AbstractDatabaseSynchronize implements IDatabaseSynchronize {
private DefaultTransactionDefinition defination;
private JdbcTemplate jdbcTemplate;
private PlatformTransactionManager transactionManager;
//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~//
private Map<String, Integer> columnType;
protected List<String> fieldOrders;
protected List<String> pksOrders;
protected String insertStatementSql;
protected String updateStatementSql;
protected String deleteStatementSql;
protected int[] insertArgsType;
protected int[] updateArgsType;
protected int[] deleteArgsType;
public AbstractDatabaseSynchronize(DataSource ds) {
this.defination= new DefaultTransactionDefinition();
this.defination.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
this.defination.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
this.jdbcTemplate=new JdbcTemplate(ds);
this.transactionManager = new DataSourceTransactionManager(ds);
this.columnType=new HashMap<>();
}
@Override
public DataSource getDataSource() {
return this.jdbcTemplate.getDataSource();
}
/**
* 获取查询列元信息的SQL语句
*
* @param schemaName 模式名称
* @param tableName 表名称
* @return SQL语句
*/
public abstract String getColumMetaDataSql(String schemaName, String tableName);
private DefaultTransactionDefinition defination;
private JdbcTemplate jdbcTemplate;
private PlatformTransactionManager transactionManager;
private Map<String, Integer> columnType;
protected List<String> fieldOrders;
protected List<String> pksOrders;
protected String insertStatementSql;
protected String updateStatementSql;
protected String deleteStatementSql;
protected int[] insertArgsType;
protected int[] updateArgsType;
protected int[] deleteArgsType;
/**
* 生成Insert操作的SQL语句
*
* @param schemaName 模式名称
* @param tableName 表名称
* @param fieldNames 字段列表
* @return Insert操作的SQL语句
*/
public abstract String getInsertPrepareStatementSql(String schemaName, String tableName, List<String> fieldNames);
/**
* 生成Update操作的SQL语句
*
* @param schemaName 模式名称
* @param tableName 表名称
* @param fieldNames 字段列表
* @param pks 主键列表
* @return Update操作的SQL语句
*/
public abstract String getUpdatePrepareStatementSql(String schemaName, String tableName, List<String> fieldNames, List<String> pks);
public AbstractDatabaseSynchronize(DataSource ds) {
this.defination = new DefaultTransactionDefinition();
this.defination.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
this.defination.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
/**
* 生成Delete操作的SQL语句
*
* @param schemaName 模式名称
* @param tableName 表名称
* @param pks 主键列表
* @return Delete操作的SQL语句
*/
public abstract String getDeletePrepareStatementSql(String schemaName, String tableName, List<String> pks);
this.jdbcTemplate = new JdbcTemplate(ds);
this.transactionManager = new DataSourceTransactionManager(ds);
this.columnType = new HashMap<>();
}
@Override
public void prepare(String schemaName, String tableName, List<String> fieldNames, List<String> pks) {
if (fieldNames.isEmpty() || pks.isEmpty() || fieldNames.size() < pks.size()) {
throw new IllegalArgumentException("字段列表和主键列表不能为空，或者字段总个数应不小于主键总个数");
}
@Override
public DataSource getDataSource() {
return this.jdbcTemplate.getDataSource();
}
if (!fieldNames.containsAll(pks)) {
throw new IllegalArgumentException("字段列表必须包含主键列表");
}
/**
* 获取查询列元信息的SQL语句
*
* @param schemaName 模式名称
* @param tableName 表名称
* @return SQL语句
*/
public abstract String getColumnMetaDataSql(String schemaName, String tableName);
String sql = this.getColumMetaDataSql(schemaName, tableName);
columnType.clear();
/**
* 生成Insert操作的SQL语句
*
* @param schemaName 模式名称
* @param tableName 表名称
* @param fieldNames 字段列表
* @return Insert操作的SQL语句
*/
public abstract String getInsertPrepareStatementSql(String schemaName, String tableName, List<String> fieldNames);
this.jdbcTemplate.execute(new ConnectionCallback<Boolean>() {
/**
* 生成Update操作的SQL语句
*
* @param schemaName 模式名称
* @param tableName 表名称
* @param fieldNames 字段列表
* @param pks 主键列表
* @return Update操作的SQL语句
*/
public abstract String getUpdatePrepareStatementSql(String schemaName, String tableName, List<String> fieldNames, List<String> pks);
@Override
public Boolean doInConnection(Connection conn) throws SQLException, DataAccessException {
Statement stmt = null;
ResultSet rs = null;
try {
stmt = conn.createStatement();
rs = stmt.executeQuery(sql);
ResultSetMetaData rsMetaData = rs.getMetaData();
for (int i = 0, len = rsMetaData.getColumnCount(); i < len; i++) {
columnType.put(rsMetaData.getColumnName(i + 1), rsMetaData.getColumnType(i + 1));
}
/**
* 生成Delete操作的SQL语句
*
* @param schemaName 模式名称
* @param tableName 表名称
* @param pks 主键列表
* @return Delete操作的SQL语句
*/
public abstract String getDeletePrepareStatementSql(String schemaName, String tableName, List<String> pks);
return true;
} catch (Exception e) {
throw new RuntimeException(
String.format("获取表:%s.%s 的字段的元信息时失败. 请联系 DBA 核查该库、表信息.", schemaName, tableName), e);
} finally {
JdbcUtils.closeResultSet(rs);
JdbcUtils.closeStatement(stmt);
}
}
});
@Override
public void prepare(String schemaName, String tableName, List<String> fieldNames, List<String> pks) {
if (fieldNames.isEmpty() || pks.isEmpty() || fieldNames.size() < pks.size()) {
throw new IllegalArgumentException("字段列表和主键列表不能为空，或者字段总个数应不小于主键总个数");
}
this.fieldOrders = new ArrayList<String>(fieldNames);
this.pksOrders = new ArrayList<String>(pks);
if (!fieldNames.containsAll(pks)) {
throw new IllegalArgumentException("字段列表必须包含主键列表");
}
this.insertStatementSql = this.getInsertPrepareStatementSql(schemaName, tableName, fieldNames);
this.updateStatementSql = this.getUpdatePrepareStatementSql(schemaName, tableName, fieldNames, pks);
this.deleteStatementSql = this.getDeletePrepareStatementSql(schemaName, tableName, pks);
String sql = this.getColumnMetaDataSql(schemaName, tableName);
columnType.clear();
insertArgsType = new int[fieldNames.size()];
for (int k = 0; k < fieldNames.size(); ++k) {
String field = fieldNames.get(k);
insertArgsType[k] = this.columnType.get(field);
}
this.jdbcTemplate.execute((Connection conn) -> {
Statement stmt = null;
ResultSet rs = null;
try {
stmt = conn.createStatement();
rs = stmt.executeQuery(sql);
ResultSetMetaData rsMetaData = rs.getMetaData();
for (int i = 0, len = rsMetaData.getColumnCount(); i < len; i++) {
columnType.put(rsMetaData.getColumnName(i + 1), rsMetaData.getColumnType(i + 1));
}
updateArgsType = new int[fieldNames.size()];
int idx = 0;
for (int i = 0; i < fieldNames.size(); ++i) {
String field = fieldNames.get(i);
if (!pks.contains(field)) {
updateArgsType[idx++] = this.columnType.get(field);
}
}
for (String pk : pks) {
updateArgsType[idx++] = this.columnType.get(pk);
}
return true;
} catch (Exception e) {
throw new RuntimeException(String.format("获取表:%s.%s 的字段的元信息时失败. 请联系 DBA 核查该库、表信息.", schemaName, tableName), e);
} finally {
JdbcUtils.closeResultSet(rs);
JdbcUtils.closeStatement(stmt);
}
});
deleteArgsType = new int[pks.size()];
for (int j = 0; j < pks.size(); ++j) {
String pk = pks.get(j);
deleteArgsType[j] = this.columnType.get(pk);
}
}
@Override
public long executeInsert(List<Object[]> records) {
TransactionStatus status = transactionManager.getTransaction(defination);
if (log.isDebugEnabled()) {
log.debug("Execute Insert SQL : {}", this.insertStatementSql);
}
this.fieldOrders = new ArrayList<String>(fieldNames);
this.pksOrders = new ArrayList<String>(pks);
try {
int[] affects = jdbcTemplate.batchUpdate(this.insertStatementSql, records, this.insertArgsType);
int affectCount = 0;
for (int i : affects) {
affectCount += i;
}
this.insertStatementSql = this.getInsertPrepareStatementSql(schemaName, tableName, fieldNames);
this.updateStatementSql = this.getUpdatePrepareStatementSql(schemaName, tableName, fieldNames, pks);
this.deleteStatementSql = this.getDeletePrepareStatementSql(schemaName, tableName, pks);
transactionManager.commit(status);
return affectCount;
} catch (TransactionException e) {
transactionManager.rollback(status);
throw e;
} catch (Exception e) {
transactionManager.rollback(status);
throw e;
}
}
insertArgsType = new int[fieldNames.size()];
for (int k = 0; k < fieldNames.size(); ++k) {
String field = fieldNames.get(k);
insertArgsType[k] = this.columnType.get(field);
}
@Override
public long executeUpdate(List<Object[]> records) {
List<Object[]> datas = new LinkedList<Object[]>();
for (Object[] r : records) {
Object[] nr = new Object[this.fieldOrders.size()];
int idx=0;
for (int i = 0; i < this.fieldOrders.size(); ++i) {
String field = this.fieldOrders.get(i);
if(!this.pksOrders.contains(field)) {
int index = this.fieldOrders.indexOf(field);
nr[idx++] = r[index];
}
}
for(int j=0;j<this.pksOrders.size();++j) {
String pk = this.pksOrders.get(j);
int index = this.fieldOrders.indexOf(pk);
nr[idx++] = r[index];
}
updateArgsType = new int[fieldNames.size()];
int idx = 0;
for (int i = 0; i < fieldNames.size(); ++i) {
String field = fieldNames.get(i);
if (!pks.contains(field)) {
updateArgsType[idx++] = this.columnType.get(field);
}
}
for (String pk : pks) {
updateArgsType[idx++] = this.columnType.get(pk);
}
datas.add(nr);
}
TransactionStatus status = transactionManager.getTransaction(defination);
if (log.isDebugEnabled()) {
log.debug("Execute Update SQL : {}", this.updateStatementSql);
}
try {
int[] affects = jdbcTemplate.batchUpdate(this.updateStatementSql, datas, this.updateArgsType);
int affectCount = 0;
for (int i : affects) {
affectCount += i;
}
deleteArgsType = new int[pks.size()];
for (int j = 0; j < pks.size(); ++j) {
String pk = pks.get(j);
deleteArgsType[j] = this.columnType.get(pk);
}
}
transactionManager.commit(status);
return affectCount;
} catch (TransactionException e) {
transactionManager.rollback(status);
throw e;
} catch (Exception e) {
transactionManager.rollback(status);
throw e;
}
}
@Override
public long executeInsert(List<Object[]> records) {
TransactionStatus status = transactionManager.getTransaction(defination);
if (log.isDebugEnabled()) {
log.debug("Execute Insert SQL : {}", this.insertStatementSql);
}
@Override
public long executeDelete(List<Object[]> records) {
List<Object[]> datas = new LinkedList<Object[]>();
for (Object[] r : records) {
Object[] nr = new Object[this.pksOrders.size()];
for (int i = 0; i < this.pksOrders.size(); ++i) {
String pk = this.pksOrders.get(i);
int index = this.fieldOrders.indexOf(pk);
nr[i] = r[index];
}
try {
int[] affects = jdbcTemplate.batchUpdate(this.insertStatementSql, records, this.insertArgsType);
int affectCount = 0;
for (int i : affects) {
affectCount += i;
}
datas.add(nr);
}
transactionManager.commit(status);
return affectCount;
} catch (TransactionException e) {
transactionManager.rollback(status);
throw e;
} catch (Exception e) {
transactionManager.rollback(status);
throw e;
}
}
TransactionStatus status = transactionManager.getTransaction(defination);
if (log.isDebugEnabled()) {
log.debug("Execute Delete SQL : {}", this.deleteStatementSql);
}
@Override
public long executeUpdate(List<Object[]> records) {
List<Object[]> datas = new LinkedList<>();
for (Object[] r : records) {
try {
int[] affects = jdbcTemplate.batchUpdate(this.deleteStatementSql, datas, this.deleteArgsType);
int affectCount = 0;
for (int i : affects) {
affectCount += i;
}
Object[] nr = new Object[this.fieldOrders.size()];
int idx = 0;
transactionManager.commit(status);
return affectCount;
} catch (TransactionException e) {
transactionManager.rollback(status);
throw e;
} catch (Exception e) {
transactionManager.rollback(status);
throw e;
} finally {
datas.clear();
}
}
for (int i = 0; i < this.fieldOrders.size(); ++i) {
String field = this.fieldOrders.get(i);
if (!this.pksOrders.contains(field)) {
int index = this.fieldOrders.indexOf(field);
nr[idx++] = r[index];
}
}
for (int j = 0; j < this.pksOrders.size(); ++j) {
String pk = this.pksOrders.get(j);
int index = this.fieldOrders.indexOf(pk);
nr[idx++] = r[index];
}
datas.add(nr);
}
TransactionStatus status = transactionManager.getTransaction(defination);
if (log.isDebugEnabled()) {
log.debug("Execute Update SQL : {}", this.updateStatementSql);
}
try {
int[] affects = jdbcTemplate.batchUpdate(this.updateStatementSql, datas, this.updateArgsType);
int affectCount = 0;
for (int i : affects) {
affectCount += i;
}
transactionManager.commit(status);
return affectCount;
} catch (TransactionException e) {
transactionManager.rollback(status);
throw e;
} catch (Exception e) {
transactionManager.rollback(status);
throw e;
}
}
@Override
public long executeDelete(List<Object[]> records) {
List<Object[]> datas = new LinkedList<>();
for (Object[] r : records) {
Object[] nr = new Object[this.pksOrders.size()];
for (int i = 0; i < this.pksOrders.size(); ++i) {
String pk = this.pksOrders.get(i);
int index = this.fieldOrders.indexOf(pk);
nr[i] = r[index];
}
datas.add(nr);
}
TransactionStatus status = transactionManager.getTransaction(defination);
if (log.isDebugEnabled()) {
log.debug("Execute Delete SQL : {}", this.deleteStatementSql);
}
try {
int[] affects = jdbcTemplate.batchUpdate(this.deleteStatementSql, datas, this.deleteArgsType);
int affectCount = 0;
for (int i : affects) {
affectCount += i;
}
transactionManager.commit(status);
return affectCount;
} catch (TransactionException e) {
transactionManager.rollback(status);
throw e;
} catch (Exception e) {
transactionManager.rollback(status);
throw e;
} finally {
datas.clear();
}
}
}

68
dbswitch-dbsynch/src/main/java/com/gitee/dbswitch/dbsynch/DatabaseSynchronizeFactory.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbsynch;
@@ -21,54 +21,44 @@ import com.gitee.dbswitch.dbsynch.mysql.MySqlDatabaseSynchImpl;
import com.gitee.dbswitch.dbsynch.oracle.OracleDatabaseSynchImpl;
import com.gitee.dbswitch.dbsynch.pgsql.GreenplumDatabaseSynchImpl;
import com.gitee.dbswitch.dbsynch.pgsql.PostgresqlDatabaseSynchImpl;
import java.lang.reflect.Constructor;
import java.util.function.Function;
/**
* 数据库同步器构造工厂类
*
* @author tang
*
* @author tang
*/
public final class DatabaseSynchronizeFactory {
private static final Map<String, String> DATABASE_SYNCH_MAPPER = new HashMap<String, String>() {
private static final Map<String, Function<DataSource, IDatabaseSynchronize>> DATABASE_SYNCH_MAPPER = new HashMap<String, Function<DataSource, IDatabaseSynchronize>>() {
private static final long serialVersionUID = -2359773637275934408L;
private static final long serialVersionUID = -2359773637275934408L;
{
put("MYSQL", MySqlDatabaseSynchImpl.class.getName());
put("ORACLE", OracleDatabaseSynchImpl.class.getName());
put("SQLSERVER", SqlServerDatabaseSynchImpl.class.getName());
put("POSTGRESQL", PostgresqlDatabaseSynchImpl.class.getName());
put("GREENPLUM", GreenplumDatabaseSynchImpl.class.getName());
put("DB2",DB2DatabaseSynchImpl.class.getName());
put("DM",DmDatabaseSynchImpl.class.getName());
put("KINGBASE",KingbaseDatabaseSynchImpl.class.getName());
}
};
{
put("MYSQL", MySqlDatabaseSynchImpl::new);
put("ORACLE", OracleDatabaseSynchImpl::new);
put("SQLSERVER", SqlServerDatabaseSynchImpl::new);
put("POSTGRESQL", PostgresqlDatabaseSynchImpl::new);
put("GREENPLUM", GreenplumDatabaseSynchImpl::new);
put("DB2", DB2DatabaseSynchImpl::new);
put("DM", DmDatabaseSynchImpl::new);
put("KINGBASE", KingbaseDatabaseSynchImpl::new);
}
};
/**
* 获取指定数据源的同步器
*
* @param dataSource 数据源
* @return 同步器对象
*/
public static AbstractDatabaseSynchronize createDatabaseWriter(DataSource dataSource) {
String type = DatabaseAwareUtils.getDatabaseNameByDataSource(dataSource).toUpperCase();
/**
* 获取指定数据源的同步器
*
* @param dataSource 数据源
* @return 同步器对象
*/
public static IDatabaseSynchronize createDatabaseWriter(DataSource dataSource) {
String type = DatabaseAwareUtils.getDatabaseNameByDataSource(dataSource).toUpperCase();
if (DATABASE_SYNCH_MAPPER.containsKey(type)) {
String className = DATABASE_SYNCH_MAPPER.get(type);
try {
Class<?>[] paramTypes = { DataSource.class };
Object[] paramValues = { dataSource };
Class<?> clas = Class.forName(className);
Constructor<?> cons = clas.getConstructor(paramTypes);
return (AbstractDatabaseSynchronize) cons.newInstance(paramValues);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
if (!DATABASE_SYNCH_MAPPER.containsKey(type)) {
throw new RuntimeException(String.format("[dbsynch] Unkown Supported database type (%s)", type));
}
throw new RuntimeException(String.format("[dbsynch] Unkown Supported database type (%s)", type));
}
return DATABASE_SYNCH_MAPPER.get(type).apply(dataSource);
}
}

2
dbswitch-dbsynch/src/main/java/com/gitee/dbswitch/dbsynch/IDatabaseSynchronize.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbsynch;

83
dbswitch-dbsynch/src/main/java/com/gitee/dbswitch/dbsynch/db2/DB2DatabaseSynchImpl.java
View File

@@ -4,73 +4,64 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbsynch.db2;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
import javax.sql.DataSource;
import org.apache.commons.lang3.StringUtils;
import com.gitee.dbswitch.dbsynch.AbstractDatabaseSynchronize;
import com.gitee.dbswitch.dbsynch.IDatabaseSynchronize;
/**
* DB2数据库实现类
*
* @author tang
* DB2数据库DML同步实现类
*
* @author tang
*/
public class DB2DatabaseSynchImpl extends AbstractDatabaseSynchronize implements IDatabaseSynchronize {
public DB2DatabaseSynchImpl(DataSource ds) {
super(ds);
}
public DB2DatabaseSynchImpl(DataSource ds) {
super(ds);
}
@Override
public String getColumMetaDataSql(String schemaName, String tableName) {
return String.format("SELECT * FROM \"%s\".\"%s\" WHERE 1=2", schemaName, tableName);
}
@Override
public String getColumnMetaDataSql(String schemaName, String tableName) {
return String.format("SELECT * FROM \"%s\".\"%s\" WHERE 1=2", schemaName, tableName);
}
@Override
public String getInsertPrepareStatementSql(String schemaName, String tableName, List<String> fieldNames) {
List<String> placeHolders = new ArrayList<String>();
for (int i = 0; i < fieldNames.size(); ++i) {
placeHolders.add("?");
}
@Override
public String getInsertPrepareStatementSql(String schemaName, String tableName, List<String> fieldNames) {
List<String> placeHolders = Collections.nCopies(fieldNames.size(), "?");
return String.format("INSERT INTO \"%s\".\"%s\" ( \"%s\" ) VALUES ( %s )", schemaName, tableName,
StringUtils.join(fieldNames, "\",\""), StringUtils.join(placeHolders, ","));
}
return String.format("INSERT INTO \"%s\".\"%s\" ( \"%s\" ) VALUES ( %s )", schemaName, tableName,
StringUtils.join(fieldNames, "\",\""), StringUtils.join(placeHolders, ","));
}
@Override
public String getUpdatePrepareStatementSql(String schemaName, String tableName, List<String> fieldNames,
List<String> pks) {
List<String> uf = new ArrayList<String>();
for (String field : fieldNames) {
if (!pks.contains(field)) {
uf.add(String.format("\"%s\"=?", field));
}
}
@Override
public String getUpdatePrepareStatementSql(String schemaName, String tableName, List<String> fieldNames, List<String> pks) {
List<String> uf = fieldNames.stream()
.filter(field -> !pks.contains(field))
.map(field -> String.format("\"%s\"=?", field))
.collect(Collectors.toList());
List<String> uw = new ArrayList<String>();
for (String pk : pks) {
uw.add(String.format("\"%s\"=?", pk));
}
List<String> uw = pks.stream()
.map(pk -> String.format("\"%s\"=?", pk))
.collect(Collectors.toList());
return String.format("UPDATE \"%s\".\"%s\" SET %s WHERE %s", schemaName, tableName, StringUtils.join(uf, " , "),
StringUtils.join(uw, " AND "));
}
return String.format("UPDATE \"%s\".\"%s\" SET %s WHERE %s", schemaName, tableName, StringUtils.join(uf, " , "),
StringUtils.join(uw, " AND "));
}
@Override
public String getDeletePrepareStatementSql(String schemaName, String tableName, List<String> pks) {
List<String> uw = new ArrayList<String>();
for (String pk : pks) {
uw.add(String.format("\"%s\"=?", pk));
}
@Override
public String getDeletePrepareStatementSql(String schemaName, String tableName, List<String> pks) {
List<String> uw = pks.stream()
.map(pk -> String.format("\"%s\"=?", pk))
.collect(Collectors.toList());
return String.format("DELETE FROM \"%s\".\"%s\" WHERE %s ", schemaName, tableName, StringUtils.join(uw, " AND "));
}
return String.format("DELETE FROM \"%s\".\"%s\" WHERE %s ", schemaName, tableName, StringUtils.join(uw, " AND "));
}
}

8
dbswitch-dbsynch/src/main/java/com/gitee/dbswitch/dbsynch/dm/DmDatabaseSynchImpl.java
View File

@@ -4,15 +4,19 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbsynch.dm;
import javax.sql.DataSource;
import com.gitee.dbswitch.dbsynch.oracle.OracleDatabaseSynchImpl;
/**
* DM数据库DML同步实现类
*
* @author tang
*/
public class DmDatabaseSynchImpl extends OracleDatabaseSynchImpl {
public DmDatabaseSynchImpl(DataSource ds) {

9
dbswitch-dbsynch/src/main/java/com/gitee/dbswitch/dbsynch/kingbase/KingbaseDatabaseSynchImpl.java
View File

@@ -4,15 +4,20 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbsynch.kingbase;
import javax.sql.DataSource;
import com.gitee.dbswitch.dbsynch.pgsql.PostgresqlDatabaseSynchImpl;
/**
* kingbase8数据库DML同步实现类
*
* @author tang
*
*/
public class KingbaseDatabaseSynchImpl extends PostgresqlDatabaseSynchImpl {
public KingbaseDatabaseSynchImpl(DataSource ds) {

83
dbswitch-dbsynch/src/main/java/com/gitee/dbswitch/dbsynch/mssql/SqlServerDatabaseSynchImpl.java
View File

@@ -4,73 +4,64 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbsynch.mssql;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
import javax.sql.DataSource;
import org.apache.commons.lang3.StringUtils;
import com.gitee.dbswitch.dbsynch.AbstractDatabaseSynchronize;
import com.gitee.dbswitch.dbsynch.IDatabaseSynchronize;
/**
* SQLServer数据库实现类
*
* @author tang
* SQLServer数据库DML同步实现类
*
* @author tang
*/
public class SqlServerDatabaseSynchImpl extends AbstractDatabaseSynchronize implements IDatabaseSynchronize {
public SqlServerDatabaseSynchImpl(DataSource ds) {
super(ds);
}
public SqlServerDatabaseSynchImpl(DataSource ds) {
super(ds);
}
@Override
public String getColumMetaDataSql(String schemaName, String tableName) {
return String.format("SELECT * FROM [%s].[%s] WHERE 1=2", schemaName, tableName);
}
@Override
public String getColumnMetaDataSql(String schemaName, String tableName) {
return String.format("SELECT * FROM [%s].[%s] WHERE 1=2", schemaName, tableName);
}
@Override
public String getInsertPrepareStatementSql(String schemaName, String tableName, List<String> fieldNames) {
List<String> placeHolders = new ArrayList<String>();
for (int i = 0; i < fieldNames.size(); ++i) {
placeHolders.add("?");
}
@Override
public String getInsertPrepareStatementSql(String schemaName, String tableName, List<String> fieldNames) {
List<String> placeHolders = Collections.nCopies(fieldNames.size(), "?");
return String.format("INSERT INTO [%s].[%s] ( [%s] ) VALUES ( %s )", schemaName, tableName,
StringUtils.join(fieldNames, "],["), StringUtils.join(placeHolders, ","));
}
return String.format("INSERT INTO [%s].[%s] ( [%s] ) VALUES ( %s )", schemaName, tableName,
StringUtils.join(fieldNames, "],["), StringUtils.join(placeHolders, ","));
}
@Override
public String getUpdatePrepareStatementSql(String schemaName, String tableName, List<String> fieldNames, List<String> pks) {
List<String> uf = fieldNames.stream()
.filter(field -> !pks.contains(field))
.map(field -> String.format("[%s]=?", field))
.collect(Collectors.toList());
@Override
public String getUpdatePrepareStatementSql(String schemaName, String tableName, List<String> fieldNames,
List<String> pks) {
List<String> uf = new ArrayList<String>();
for (String field : fieldNames) {
if (!pks.contains(field)) {
uf.add(String.format("[%s]=?", field));
}
}
List<String> uw = pks.stream()
.map(pk -> String.format("[%s]=?", pk))
.collect(Collectors.toList());
List<String> uw = new ArrayList<String>();
for (String pk : pks) {
uw.add(String.format("[%s]=?", pk));
}
return String.format("UPDATE [%s].[%s] SET %s WHERE %s", schemaName, tableName, StringUtils.join(uf, " , "),
StringUtils.join(uw, " AND "));
}
return String.format("UPDATE [%s].[%s] SET %s WHERE %s", schemaName, tableName, StringUtils.join(uf, " , "),
StringUtils.join(uw, " AND "));
}
@Override
public String getDeletePrepareStatementSql(String schemaName, String tableName, List<String> pks) {
List<String> uw = pks.stream()
.map(pk -> String.format("[%s]=?", pk))
.collect(Collectors.toList());
@Override
public String getDeletePrepareStatementSql(String schemaName, String tableName, List<String> pks) {
List<String> uw = new ArrayList<String>();
for (String pk : pks) {
uw.add(String.format("[%s]=?", pk));
}
return String.format("DELETE FROM [%s].[%s] WHERE %s ", schemaName, tableName, StringUtils.join(uw, " AND "));
}
return String.format("DELETE FROM [%s].[%s] WHERE %s ", schemaName, tableName, StringUtils.join(uw, " AND "));
}
}

83
dbswitch-dbsynch/src/main/java/com/gitee/dbswitch/dbsynch/mysql/MySqlDatabaseSynchImpl.java
View File

@@ -4,73 +4,64 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbsynch.mysql;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
import javax.sql.DataSource;
import org.apache.commons.lang3.StringUtils;
import com.gitee.dbswitch.dbsynch.AbstractDatabaseSynchronize;
import com.gitee.dbswitch.dbsynch.IDatabaseSynchronize;
/**
* MySQL数据库实现类
*
* @author tang
* MySQL数据库DML同步实现类
*
* @author tang
*/
public class MySqlDatabaseSynchImpl extends AbstractDatabaseSynchronize implements IDatabaseSynchronize {
public MySqlDatabaseSynchImpl(DataSource ds) {
super(ds);
}
public MySqlDatabaseSynchImpl(DataSource ds) {
super(ds);
}
@Override
public String getColumMetaDataSql(String schemaName, String tableName) {
return String.format("SELECT * FROM `%s`.`%s` WHERE 1=2", schemaName, tableName);
}
@Override
public String getColumnMetaDataSql(String schemaName, String tableName) {
return String.format("SELECT * FROM `%s`.`%s` WHERE 1=2", schemaName, tableName);
}
@Override
public String getInsertPrepareStatementSql(String schemaName, String tableName, List<String> fieldNames) {
List<String> placeHolders = new ArrayList<String>();
for (int i = 0; i < fieldNames.size(); ++i) {
placeHolders.add("?");
}
@Override
public String getInsertPrepareStatementSql(String schemaName, String tableName, List<String> fieldNames) {
List<String> placeHolders = Collections.nCopies(fieldNames.size(), "?");
return String.format("INSERT INTO `%s`.`%s` ( `%s` ) VALUES ( %s )", schemaName, tableName,
StringUtils.join(fieldNames, "`,`"), StringUtils.join(placeHolders, ","));
}
return String.format("INSERT INTO `%s`.`%s` ( `%s` ) VALUES ( %s )", schemaName, tableName,
StringUtils.join(fieldNames, "`,`"), StringUtils.join(placeHolders, ","));
}
@Override
public String getUpdatePrepareStatementSql(String schemaName, String tableName, List<String> fieldNames,
List<String> pks) {
List<String> uf = new ArrayList<String>();
for (String field : fieldNames) {
if (!pks.contains(field)) {
uf.add(String.format("`%s`=?", field));
}
}
@Override
public String getUpdatePrepareStatementSql(String schemaName, String tableName, List<String> fieldNames, List<String> pks) {
List<String> uf = fieldNames.stream()
.filter(field -> !pks.contains(field))
.map(field -> String.format("`%s`=?", field))
.collect(Collectors.toList());
List<String> uw = new ArrayList<String>();
for (String pk : pks) {
uw.add(String.format("`%s`=?", pk));
}
List<String> uw = pks.stream()
.map(pk -> String.format("`%s`=?", pk))
.collect(Collectors.toList());
return String.format("UPDATE `%s`.`%s` SET %s WHERE %s", schemaName, tableName, StringUtils.join(uf, " , "),
StringUtils.join(uw, " AND "));
}
return String.format("UPDATE `%s`.`%s` SET %s WHERE %s", schemaName, tableName, StringUtils.join(uf, " , "),
StringUtils.join(uw, " AND "));
}
@Override
public String getDeletePrepareStatementSql(String schemaName, String tableName, List<String> pks) {
List<String> uw = new ArrayList<String>();
for (String pk : pks) {
uw.add(String.format("`%s`=?", pk));
}
@Override
public String getDeletePrepareStatementSql(String schemaName, String tableName, List<String> pks) {
List<String> uw = pks.stream()
.map(pk -> String.format("`%s`=?", pk))
.collect(Collectors.toList());
return String.format("DELETE FROM `%s`.`%s` WHERE %s ", schemaName, tableName, StringUtils.join(uw, " AND "));
}
return String.format("DELETE FROM `%s`.`%s` WHERE %s ", schemaName, tableName, StringUtils.join(uw, " AND "));
}
}

83
dbswitch-dbsynch/src/main/java/com/gitee/dbswitch/dbsynch/oracle/OracleDatabaseSynchImpl.java
View File

@@ -4,73 +4,64 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbsynch.oracle;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
import javax.sql.DataSource;
import org.apache.commons.lang3.StringUtils;
import com.gitee.dbswitch.dbsynch.AbstractDatabaseSynchronize;
import com.gitee.dbswitch.dbsynch.IDatabaseSynchronize;
/**
* Oracle数据库实现类
*
* @author tang
* Oracle数据库DML同步实现类
*
* @author tang
*/
public class OracleDatabaseSynchImpl extends AbstractDatabaseSynchronize implements IDatabaseSynchronize {
public OracleDatabaseSynchImpl(DataSource ds) {
super(ds);
}
public OracleDatabaseSynchImpl(DataSource ds) {
super(ds);
}
@Override
public String getColumMetaDataSql(String schemaName, String tableName) {
return String.format("SELECT * FROM \"%s\".\"%s\" WHERE 1=2", schemaName, tableName);
}
@Override
public String getColumnMetaDataSql(String schemaName, String tableName) {
return String.format("SELECT * FROM \"%s\".\"%s\" WHERE 1=2", schemaName, tableName);
}
@Override
public String getInsertPrepareStatementSql(String schemaName, String tableName, List<String> fieldNames) {
List<String> placeHolders = new ArrayList<String>();
for (int i = 0; i < fieldNames.size(); ++i) {
placeHolders.add("?");
}
@Override
public String getInsertPrepareStatementSql(String schemaName, String tableName, List<String> fieldNames) {
List<String> placeHolders = Collections.nCopies(fieldNames.size(), "?");
return String.format("INSERT INTO \"%s\".\"%s\" ( \"%s\" ) VALUES ( %s )", schemaName, tableName,
StringUtils.join(fieldNames, "\",\""), StringUtils.join(placeHolders, ","));
}
return String.format("INSERT INTO \"%s\".\"%s\" ( \"%s\" ) VALUES ( %s )", schemaName, tableName,
StringUtils.join(fieldNames, "\",\""), StringUtils.join(placeHolders, ","));
}
@Override
public String getUpdatePrepareStatementSql(String schemaName, String tableName, List<String> fieldNames,
List<String> pks) {
List<String> uf = new ArrayList<String>();
for (String field : fieldNames) {
if (!pks.contains(field)) {
uf.add(String.format("\"%s\"=?", field));
}
}
@Override
public String getUpdatePrepareStatementSql(String schemaName, String tableName, List<String> fieldNames, List<String> pks) {
List<String> uf = fieldNames.stream()
.filter(field -> !pks.contains(field))
.map(field -> String.format("\"%s\"=?", field))
.collect(Collectors.toList());
List<String> uw = new ArrayList<String>();
for (String pk : pks) {
uw.add(String.format("\"%s\"=?", pk));
}
List<String> uw = pks.stream()
.map(pk -> String.format("\"%s\"=?", pk))
.collect(Collectors.toList());
return String.format("UPDATE \"%s\".\"%s\" SET %s WHERE %s", schemaName, tableName, StringUtils.join(uf, " , "),
StringUtils.join(uw, " AND "));
}
return String.format("UPDATE \"%s\".\"%s\" SET %s WHERE %s", schemaName, tableName, StringUtils.join(uf, " , "),
StringUtils.join(uw, " AND "));
}
@Override
public String getDeletePrepareStatementSql(String schemaName, String tableName, List<String> pks) {
List<String> uw = new ArrayList<String>();
for (String pk : pks) {
uw.add(String.format("\"%s\"=?", pk));
}
@Override
public String getDeletePrepareStatementSql(String schemaName, String tableName, List<String> pks) {
List<String> uw = pks.stream()
.map(pk -> String.format("\"%s\"=?", pk))
.collect(Collectors.toList());
return String.format("DELETE FROM \"%s\".\"%s\" WHERE %s ", schemaName, tableName, StringUtils.join(uw, " AND "));
}
return String.format("DELETE FROM \"%s\".\"%s\" WHERE %s ", schemaName, tableName, StringUtils.join(uw, " AND "));
}
}

4
dbswitch-dbsynch/src/main/java/com/gitee/dbswitch/dbsynch/pgsql/GreenplumDatabaseSynchImpl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbsynch.pgsql;
@@ -13,7 +13,7 @@ import javax.sql.DataSource;
import com.gitee.dbswitch.dbsynch.IDatabaseSynchronize;
/**
* Greenplum数据库实现类
* Greenplum数据库DML同步实现类
*
* @author tang
*

83
dbswitch-dbsynch/src/main/java/com/gitee/dbswitch/dbsynch/pgsql/PostgresqlDatabaseSynchImpl.java
View File

@@ -4,73 +4,64 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbsynch.pgsql;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
import javax.sql.DataSource;
import org.apache.commons.lang3.StringUtils;
import com.gitee.dbswitch.dbsynch.AbstractDatabaseSynchronize;
import com.gitee.dbswitch.dbsynch.IDatabaseSynchronize;
/**
* PostgreSQL数据库实现类
*
* @author tang
* PostgreSQL数据库DML同步实现类
*
* @author tang
*/
public class PostgresqlDatabaseSynchImpl extends AbstractDatabaseSynchronize implements IDatabaseSynchronize {
public PostgresqlDatabaseSynchImpl(DataSource ds) {
super(ds);
}
public PostgresqlDatabaseSynchImpl(DataSource ds) {
super(ds);
}
@Override
public String getColumMetaDataSql(String schemaName, String tableName) {
return String.format("SELECT * FROM \"%s\".\"%s\" WHERE 1=2", schemaName, tableName);
}
@Override
public String getInsertPrepareStatementSql(String schemaName, String tableName, List<String> fieldNames) {
List<String> placeHolders = new ArrayList<String>();
for (int i = 0; i < fieldNames.size(); ++i) {
placeHolders.add("?");
}
@Override
public String getColumnMetaDataSql(String schemaName, String tableName) {
return String.format("SELECT * FROM \"%s\".\"%s\" WHERE 1=2", schemaName, tableName);
}
return String.format("INSERT INTO \"%s\".\"%s\" ( \"%s\" ) VALUES ( %s )", schemaName, tableName,
StringUtils.join(fieldNames, "\",\""), StringUtils.join(placeHolders, ","));
}
@Override
public String getInsertPrepareStatementSql(String schemaName, String tableName, List<String> fieldNames) {
List<String> placeHolders = Collections.nCopies(fieldNames.size(), "?");
return String.format("INSERT INTO \"%s\".\"%s\" ( \"%s\" ) VALUES ( %s )", schemaName, tableName,
StringUtils.join(fieldNames, "\",\""), StringUtils.join(placeHolders, ","));
}
@Override
public String getUpdatePrepareStatementSql(String schemaName, String tableName, List<String> fieldNames,
List<String> pks) {
List<String> uf = new ArrayList<String>();
for (String field : fieldNames) {
if (!pks.contains(field)) {
uf.add(String.format("\"%s\"=?", field));
}
}
@Override
public String getUpdatePrepareStatementSql(String schemaName, String tableName, List<String> fieldNames, List<String> pks) {
List<String> uf = fieldNames.stream()
.filter(field -> !pks.contains(field))
.map(field -> String.format("\"%s\"=?", field))
.collect(Collectors.toList());
List<String> uw = new ArrayList<String>();
for (String pk : pks) {
uw.add(String.format("\"%s\"=?", pk));
}
List<String> uw = pks.stream()
.map(pk -> String.format("\"%s\"=?", pk))
.collect(Collectors.toList());
return String.format("UPDATE \"%s\".\"%s\" SET %s WHERE %s", schemaName, tableName, StringUtils.join(uf, " , "),
StringUtils.join(uw, " AND "));
}
return String.format("UPDATE \"%s\".\"%s\" SET %s WHERE %s", schemaName, tableName, StringUtils.join(uf, " , "),
StringUtils.join(uw, " AND "));
}
@Override
public String getDeletePrepareStatementSql(String schemaName, String tableName, List<String> pks) {
List<String> uw = new ArrayList<String>();
for (String pk : pks) {
uw.add(String.format("\"%s\"=?", pk));
}
@Override
public String getDeletePrepareStatementSql(String schemaName, String tableName, List<String> pks) {
List<String> uw = pks.stream()
.map(pk -> String.format("\"%s\"=?", pk))
.collect(Collectors.toList());
return String.format("DELETE FROM \"%s\".\"%s\" WHERE %s ", schemaName, tableName, StringUtils.join(uw, " AND "));
}
return String.format("DELETE FROM \"%s\".\"%s\" WHERE %s ", schemaName, tableName, StringUtils.join(uw, " AND "));
}
}

2
dbswitch-dbwriter/src/main/java/com/gitee/dbswitch/dbwriter/AbstractDatabaseWriter.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbwriter;

46
dbswitch-dbwriter/src/main/java/com/gitee/dbswitch/dbwriter/DatabaseWriterFactory.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbwriter;
@@ -12,8 +12,8 @@ package com.gitee.dbswitch.dbwriter;
import java.util.Map;
import java.util.HashMap;
import javax.sql.DataSource;
import java.util.function.Function;
import com.gitee.dbswitch.dbcommon.util.DatabaseAwareUtils;
import java.lang.reflect.Constructor;
/**
* 数据库写入器构造工厂类
@@ -23,20 +23,20 @@ import java.lang.reflect.Constructor;
*/
public class DatabaseWriterFactory {
private static final Map<String, String> DATABASE_WRITER_MAPPER = new HashMap<String, String>() {
private static final Map<String, Function<DataSource, IDatabaseWriter>> DATABASE_WRITER_MAPPER = new HashMap<String, Function<DataSource, IDatabaseWriter>>() {
private static final long serialVersionUID = 3365136872693503697L;
{
put("MYSQL", "com.gitee.dbswitch.dbwriter.mysql.MySqlWriterImpl");
put("ORACLE", "com.gitee.dbswitch.dbwriter.oracle.OracleWriterImpl");
put("SQLSERVER", "com.gitee.dbswitch.dbwriter.mssql.SqlServerWriterImpl");
put("POSTGRESQL", "com.gitee.dbswitch.dbwriter.gpdb.GreenplumCopyWriterImpl");
put("GREENPLUM", "com.gitee.dbswitch.dbwriter.gpdb.GreenplumCopyWriterImpl");
put("DB2", "com.gitee.dbswitch.dbwriter.db2.DB2WriterImpl");
put("DM", "com.gitee.dbswitch.dbwriter.dm.DmWriterImpl");
put("MYSQL", com.gitee.dbswitch.dbwriter.mysql.MySqlWriterImpl::new);
put("ORACLE", com.gitee.dbswitch.dbwriter.oracle.OracleWriterImpl::new);
put("SQLSERVER", com.gitee.dbswitch.dbwriter.mssql.SqlServerWriterImpl::new);
put("POSTGRESQL", com.gitee.dbswitch.dbwriter.gpdb.GreenplumCopyWriterImpl::new);
put("GREENPLUM", com.gitee.dbswitch.dbwriter.gpdb.GreenplumCopyWriterImpl::new);
put("DB2", com.gitee.dbswitch.dbwriter.db2.DB2WriterImpl::new);
put("DM", com.gitee.dbswitch.dbwriter.dm.DmWriterImpl::new);
//对于kingbase当前只能使用insert模式
put("KINGBASE", "com.gitee.dbswitch.dbwriter.kingbase.KingbaseInsertWriterImpl");
put("KINGBASE", com.gitee.dbswitch.dbwriter.kingbase.KingbaseInsertWriterImpl::new);
}
};
@@ -46,7 +46,7 @@ public class DatabaseWriterFactory {
* @param dataSource 连接池数据源
* @return 写入器对象
*/
public static AbstractDatabaseWriter createDatabaseWriter(DataSource dataSource) {
public static IDatabaseWriter createDatabaseWriter(DataSource dataSource) {
return DatabaseWriterFactory.createDatabaseWriter(dataSource, false);
}
@@ -57,33 +57,19 @@ public class DatabaseWriterFactory {
* @param insert 对于GP/GP数据库来说是否使用insert引擎写入
* @return 写入器对象
*/
public static AbstractDatabaseWriter createDatabaseWriter(DataSource dataSource, boolean insert) {
public static IDatabaseWriter createDatabaseWriter(DataSource dataSource, boolean insert) {
String type = DatabaseAwareUtils.getDatabaseNameByDataSource(dataSource).toUpperCase();
if (insert) {
if ("POSTGRESQL".equalsIgnoreCase(type) || "GREENPLUM".equalsIgnoreCase(type)) {
return new com.gitee.dbswitch.dbwriter.gpdb.GreenplumInsertWriterImpl(dataSource);
}
if ("KINGBASE".equalsIgnoreCase(type)) {
return new com.gitee.dbswitch.dbwriter.kingbase.KingbaseInsertWriterImpl(dataSource);
}
}
if (DATABASE_WRITER_MAPPER.containsKey(type.trim())) {
String className = DATABASE_WRITER_MAPPER.get(type);
try {
Class<?>[] paraTypes = { DataSource.class };
Object[] paraValues = { dataSource };
Class<?> clas = Class.forName(className);
Constructor<?> cons = clas.getConstructor(paraTypes);
AbstractDatabaseWriter process = (AbstractDatabaseWriter) cons.newInstance(paraValues);
return process;
} catch (Exception e) {
throw new RuntimeException(e);
}
if (!DATABASE_WRITER_MAPPER.containsKey(type.trim())) {
throw new RuntimeException(String.format("[dbwrite] Unkown Supported database type (%s)", type));
}
throw new RuntimeException(String.format("[dbwrite] Unkown Supported database type (%s)", type));
return DATABASE_WRITER_MAPPER.get(type.trim()).apply(dataSource);
}
}

2
dbswitch-dbwriter/src/main/java/com/gitee/dbswitch/dbwriter/IDatabaseWriter.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbwriter;

12
dbswitch-dbwriter/src/main/java/com/gitee/dbswitch/dbwriter/db2/DB2WriterImpl.java
View File

@@ -4,12 +4,12 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbwriter.db2;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import javax.sql.DataSource;
@@ -38,13 +38,11 @@ public class DB2WriterImpl extends AbstractDatabaseWriter implements IDatabaseWr
@Override
public long write(List<String> fieldNames, List<Object[]> recordValues) {
List<String> placeHolders = new ArrayList<String>();
for (int i = 0; i < fieldNames.size(); ++i) {
placeHolders.add("?");
}
String schemaName = Objects.requireNonNull(this.schemaName, "schema-name名称为空，不合法!");
String tableName = Objects.requireNonNull(this.tableName, "table-name名称为空，不合法!");
List<String> placeHolders = Collections.nCopies(fieldNames.size(), "?");
String sqlInsert = String.format("INSERT INTO \"%s\".\"%s\" ( \"%s\" ) VALUES ( %s )", schemaName, tableName,
StringUtils.join(fieldNames, "\",\""), StringUtils.join(placeHolders, ","));

8
dbswitch-dbwriter/src/main/java/com/gitee/dbswitch/dbwriter/dm/DmWriterImpl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbwriter.dm;
@@ -13,6 +13,12 @@ import javax.sql.DataSource;
import com.gitee.dbswitch.dbwriter.oracle.OracleWriterImpl;
import com.gitee.dbswitch.dbwriter.IDatabaseWriter;
/**
* DM数据库写入实现类
*
* @author tang
*
*/
public class DmWriterImpl extends OracleWriterImpl implements IDatabaseWriter {
public DmWriterImpl(DataSource dataSource) {

2558
dbswitch-dbwriter/src/main/java/com/gitee/dbswitch/dbwriter/gpdb/GreenplumCopyWriterImpl.java
View File

File diff suppressed because it is too large Load Diff

86
dbswitch-dbwriter/src/main/java/com/gitee/dbswitch/dbwriter/gpdb/GreenplumInsertWriterImpl.java
View File

@@ -4,12 +4,12 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbwriter.gpdb;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import javax.sql.DataSource;
@@ -26,60 +26,56 @@ import lombok.extern.slf4j.Slf4j;
/**
* Greenplum数据库Insert写入实现类
*
* @author tang
*
* @author tang
*/
@Slf4j
public class GreenplumInsertWriterImpl extends AbstractDatabaseWriter implements IDatabaseWriter {
public GreenplumInsertWriterImpl(DataSource dataSource) {
super(dataSource);
}
public GreenplumInsertWriterImpl(DataSource dataSource) {
super(dataSource);
}
@Override
public long write(List<String> fieldNames, List<Object[]> recordValues) {
List<String> placeHolders = new ArrayList<String>();
for (int i = 0; i < fieldNames.size(); ++i) {
placeHolders.add("?");
}
@Override
public long write(List<String> fieldNames, List<Object[]> recordValues) {
String schemaName = Objects.requireNonNull(this.schemaName, "schema-name名称为空，不合法!");
String tableName = Objects.requireNonNull(this.tableName, "table-name名称为空，不合法!");
String schemaName = Objects.requireNonNull(this.schemaName, "schema-name名称为空，不合法!");
String tableName = Objects.requireNonNull(this.tableName, "table-name名称为空，不合法!");
String sqlInsert = String.format("INSERT INTO \"%s\".\"%s\" ( \"%s\" ) VALUES ( %s )", schemaName, tableName,
StringUtils.join(fieldNames, "\",\""), StringUtils.join(placeHolders, ","));
List<String> placeHolders = Collections.nCopies(fieldNames.size(), "?");
int[] argTypes = new int[fieldNames.size()];
for (int i = 0; i < fieldNames.size(); ++i) {
String col = fieldNames.get(i);
argTypes[i] = this.columnType.get(col);
}
String sqlInsert = String.format("INSERT INTO \"%s\".\"%s\" ( \"%s\" ) VALUES ( %s )", schemaName, tableName,
StringUtils.join(fieldNames, "\",\""), StringUtils.join(placeHolders, ","));
DefaultTransactionDefinition defination = new DefaultTransactionDefinition();
defination.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
defination.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
PlatformTransactionManager transactionManager = new DataSourceTransactionManager(
this.jdbcTemplate.getDataSource());
TransactionStatus status = transactionManager.getTransaction(defination);
int[] argTypes = new int[fieldNames.size()];
for (int i = 0; i < fieldNames.size(); ++i) {
String col = fieldNames.get(i);
argTypes[i] = this.columnType.get(col);
}
try {
int[] affects = jdbcTemplate.batchUpdate(sqlInsert, recordValues, argTypes);
int affectCount = 0;
for (int i : affects) {
affectCount += i;
}
DefaultTransactionDefinition defination = new DefaultTransactionDefinition();
defination.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
defination.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
PlatformTransactionManager transactionManager = new DataSourceTransactionManager(this.dataSource);
TransactionStatus status = transactionManager.getTransaction(defination);
recordValues.clear();
transactionManager.commit(status);
return affectCount;
} catch (TransactionException e) {
transactionManager.rollback(status);
throw e;
} catch (Exception e) {
transactionManager.rollback(status);
throw e;
}
try {
int[] affects = jdbcTemplate.batchUpdate(sqlInsert, recordValues, argTypes);
int affectCount = 0;
for (int i : affects) {
affectCount += i;
}
}
recordValues.clear();
transactionManager.commit(status);
return affectCount;
} catch (TransactionException e) {
transactionManager.rollback(status);
throw e;
} catch (Exception e) {
transactionManager.rollback(status);
throw e;
}
}
}

8
dbswitch-dbwriter/src/main/java/com/gitee/dbswitch/dbwriter/kingbase/KingbaseCopyWriterImpl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbwriter.kingbase;
@@ -13,6 +13,12 @@ import javax.sql.DataSource;
import com.gitee.dbswitch.dbwriter.IDatabaseWriter;
import com.gitee.dbswitch.dbwriter.gpdb.GreenplumCopyWriterImpl;
/**
* Kingbase8数据库Copy写入实现类
*
* @author tang
*
*/
public class KingbaseCopyWriterImpl extends GreenplumCopyWriterImpl implements IDatabaseWriter {
public KingbaseCopyWriterImpl(DataSource dataSource) {

8
dbswitch-dbwriter/src/main/java/com/gitee/dbswitch/dbwriter/kingbase/KingbaseInsertWriterImpl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbwriter.kingbase;
@@ -13,6 +13,12 @@ import javax.sql.DataSource;
import com.gitee.dbswitch.dbwriter.IDatabaseWriter;
import com.gitee.dbswitch.dbwriter.gpdb.GreenplumInsertWriterImpl;
/**
* Kingbase8数据库Insert写入实现类
*
* @author tang
*
*/
public class KingbaseInsertWriterImpl extends GreenplumInsertWriterImpl implements IDatabaseWriter {
public KingbaseInsertWriterImpl(DataSource dataSource) {

172
dbswitch-dbwriter/src/main/java/com/gitee/dbswitch/dbwriter/mssql/SqlServerWriterImpl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbwriter.mssql;
@@ -12,17 +12,10 @@ package com.gitee.dbswitch.dbwriter.mssql;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.*;
import javax.sql.DataSource;
import org.apache.commons.lang3.StringUtils;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.ConnectionCallback;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.jdbc.support.JdbcUtils;
import org.springframework.transaction.PlatformTransactionManager;
@@ -36,106 +29,97 @@ import lombok.extern.slf4j.Slf4j;
/**
* SQLServer批量写入实现类
*
* @author tang
*
* @author tang
*/
@Slf4j
public class SqlServerWriterImpl extends AbstractDatabaseWriter implements IDatabaseWriter {
public SqlServerWriterImpl(DataSource dataSource) {
super(dataSource);
}
public SqlServerWriterImpl(DataSource dataSource) {
super(dataSource);
}
@Override
public void prepareWrite(String schemaName, String tableName) {
String sql = String.format("SELECT * FROM [%s].[%s] WHERE 1=2", schemaName, tableName);
Map<String, Integer> columnMetaData = new HashMap<String, Integer>();
Boolean ret = this.jdbcTemplate.execute(new ConnectionCallback<Boolean>() {
@Override
public void prepareWrite(String schemaName, String tableName) {
String sql = String.format("SELECT * FROM [%s].[%s] WHERE 1=2", schemaName, tableName);
Map<String, Integer> columnMetaData = new HashMap<String, Integer>();
Boolean ret = this.jdbcTemplate.execute((Connection conn) -> {
Statement stmt = null;
ResultSet rs = null;
try {
stmt = conn.createStatement();
rs = stmt.executeQuery(sql);
ResultSetMetaData rsMetaData = rs.getMetaData();
for (int i = 0, len = rsMetaData.getColumnCount(); i < len; i++) {
columnMetaData.put(rsMetaData.getColumnName(i + 1), rsMetaData.getColumnType(i + 1));
}
@Override
public Boolean doInConnection(Connection conn) throws SQLException, DataAccessException {
Statement stmt = null;
ResultSet rs = null;
try {
stmt = conn.createStatement();
rs = stmt.executeQuery(sql);
ResultSetMetaData rsMetaData = rs.getMetaData();
for (int i = 0, len = rsMetaData.getColumnCount(); i < len; i++) {
columnMetaData.put(rsMetaData.getColumnName(i + 1), rsMetaData.getColumnType(i + 1));
}
return true;
} catch (Exception e) {
throw new RuntimeException(String.format("获取表:%s.%s 的字段的元信息时失败. 请联系 DBA 核查该库、表信息.", schemaName, tableName), e);
} finally {
JdbcUtils.closeResultSet(rs);
JdbcUtils.closeStatement(stmt);
}
});
return true;
} catch (Exception e) {
throw new RuntimeException(
String.format("获取表:%s.%s 的字段的元信息时失败. 请联系 DBA 核查该库、表信息.", schemaName, tableName), e);
} finally {
JdbcUtils.closeResultSet(rs);
JdbcUtils.closeStatement(stmt);
}
}
});
if (ret.booleanValue()) {
this.schemaName = schemaName;
this.tableName = tableName;
this.columnType = Objects.requireNonNull(columnMetaData);
if (ret) {
this.schemaName = schemaName;
this.tableName = tableName;
this.columnType = Objects.requireNonNull(columnMetaData);
if (this.columnType.isEmpty()) {
throw new RuntimeException(
String.format("获取表:%s.%s 的字段的元信息时失败. 请联系 DBA 核查该库、表信息.", schemaName, tableName));
}
} else {
throw new RuntimeException("内部代码出现错误，请开发人员排查！");
}
}
if (this.columnType.isEmpty()) {
throw new RuntimeException(
String.format("获取表:%s.%s 的字段的元信息时失败. 请联系 DBA 核查该库、表信息.", schemaName, tableName));
}
} else {
throw new RuntimeException("内部代码出现错误，请开发人员排查！");
}
}
@Override
public long write(List<String> fieldNames, List<Object[]> recordValues) {
List<String> placeHolders = new ArrayList<String>();
for (int i = 0; i < fieldNames.size(); ++i) {
placeHolders.add("?");
}
@Override
public long write(List<String> fieldNames, List<Object[]> recordValues) {
String schemaName = Objects.requireNonNull(this.schemaName, "schema名称为空，不合法!");
String tableName = Objects.requireNonNull(this.tableName, "table名称为空，不合法!");
String schemaName = Objects.requireNonNull(this.schemaName, "schema名称为空，不合法!");
String tableName = Objects.requireNonNull(this.tableName, "table名称为空，不合法!");
String sqlInsert = String.format("INSERT INTO [%s].[%s] ( [%s] ) VALUES ( %s )", schemaName, tableName,
StringUtils.join(fieldNames, "],["), StringUtils.join(placeHolders, ","));
List<String> placeHolders = Collections.nCopies(fieldNames.size(), "?");
int[] argTypes = new int[fieldNames.size()];
for (int i = 0; i < fieldNames.size(); ++i) {
String col = fieldNames.get(i);
argTypes[i] = this.columnType.get(col);
}
String sqlInsert = String.format("INSERT INTO [%s].[%s] ( [%s] ) VALUES ( %s )", schemaName, tableName,
StringUtils.join(fieldNames, "],["), StringUtils.join(placeHolders, ","));
DefaultTransactionDefinition defination = new DefaultTransactionDefinition();
defination.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
defination.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
PlatformTransactionManager transactionManager = new DataSourceTransactionManager(
this.jdbcTemplate.getDataSource());
TransactionStatus status = transactionManager.getTransaction(defination);
int[] argTypes = new int[fieldNames.size()];
for (int i = 0; i < fieldNames.size(); ++i) {
String col = fieldNames.get(i);
argTypes[i] = this.columnType.get(col);
}
try {
int[] affects = jdbcTemplate.batchUpdate(sqlInsert, recordValues, argTypes);
int affectCount = 0;
for (int i : affects) {
affectCount += i;
}
DefaultTransactionDefinition defination = new DefaultTransactionDefinition();
defination.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
defination.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
PlatformTransactionManager transactionManager = new DataSourceTransactionManager(this.dataSource);
TransactionStatus status = transactionManager.getTransaction(defination);
recordValues.clear();
transactionManager.commit(status);
if (log.isDebugEnabled()) {
log.debug("SQL Server insert data affect count: {}", affectCount);
}
return affectCount;
} catch (TransactionException e) {
transactionManager.rollback(status);
throw e;
} catch (Exception e) {
transactionManager.rollback(status);
throw e;
}
try {
int[] affects = jdbcTemplate.batchUpdate(sqlInsert, recordValues, argTypes);
int affectCount = 0;
for (int i : affects) {
affectCount += i;
}
}
recordValues.clear();
transactionManager.commit(status);
if (log.isDebugEnabled()) {
log.debug("SQL Server insert data affect count: {}", affectCount);
}
return affectCount;
} catch (TransactionException e) {
transactionManager.rollback(status);
throw e;
} catch (Exception e) {
transactionManager.rollback(status);
throw e;
}
}
}

85
dbswitch-dbwriter/src/main/java/com/gitee/dbswitch/dbwriter/mysql/MySqlWriterImpl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbwriter.mysql;
@@ -12,24 +12,16 @@ package com.gitee.dbswitch.dbwriter.mysql;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.*;
import javax.sql.DataSource;
import org.apache.commons.lang3.StringUtils;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.ConnectionCallback;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.jdbc.support.JdbcUtils;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import com.gitee.dbswitch.dbwriter.AbstractDatabaseWriter;
import com.gitee.dbswitch.dbwriter.IDatabaseWriter;
@@ -58,33 +50,28 @@ public class MySqlWriterImpl extends AbstractDatabaseWriter implements IDatabase
@Override
public void prepareWrite(String schemaName, String tableName) {
String sql = String.format("SELECT * FROM `%s`.`%s` WHERE 1=2", schemaName, tableName);
Map<String, Integer> columnMetaData = new HashMap<String, Integer>();
Boolean ret = this.jdbcTemplate.execute(new ConnectionCallback<Boolean>() {
@Override
public Boolean doInConnection(Connection conn) throws SQLException, DataAccessException {
Statement stmt = null;
ResultSet rs = null;
try {
stmt = conn.createStatement();
rs = stmt.executeQuery(sql);
ResultSetMetaData rsMetaData = rs.getMetaData();
for (int i = 0, len = rsMetaData.getColumnCount(); i < len; i++) {
columnMetaData.put(rsMetaData.getColumnName(i + 1), rsMetaData.getColumnType(i + 1));
}
return true;
} catch (Exception e) {
throw new RuntimeException(
String.format("获取表:%s.%s 的字段的元信息时失败. 请联系 DBA 核查该库、表信息.", schemaName, tableName), e);
} finally {
JdbcUtils.closeResultSet(rs);
JdbcUtils.closeStatement(stmt);
Map<String, Integer> columnMetaData = new HashMap<>();
Boolean ret = this.jdbcTemplate.execute((Connection conn) -> {
Statement stmt = null;
ResultSet rs = null;
try {
stmt = conn.createStatement();
rs = stmt.executeQuery(sql);
ResultSetMetaData rsMetaData = rs.getMetaData();
for (int i = 0, len = rsMetaData.getColumnCount(); i < len; i++) {
columnMetaData.put(rsMetaData.getColumnName(i + 1), rsMetaData.getColumnType(i + 1));
}
return true;
} catch (Exception e) {
throw new RuntimeException(String.format("获取表:%s.%s 的字段的元信息时失败. 请联系 DBA 核查该库、表信息.", schemaName, tableName), e);
} finally {
JdbcUtils.closeResultSet(rs);
JdbcUtils.closeStatement(stmt);
}
});
if (ret) {
if (ret.booleanValue()) {
this.schemaName = schemaName;
this.tableName = tableName;
this.columnType = Objects.requireNonNull(columnMetaData);
@@ -100,13 +87,11 @@ public class MySqlWriterImpl extends AbstractDatabaseWriter implements IDatabase
@Override
public long write(List<String> fieldNames, List<Object[]> recordValues) {
List<String> placeHolders = new ArrayList<String>();
for (int i = 0; i < fieldNames.size(); ++i) {
placeHolders.add("?");
}
String schemaName = Objects.requireNonNull(this.schemaName, "schema-name名称为空，不合法!");
String tableName = Objects.requireNonNull(this.tableName, "table-name名称为空，不合法!");
List<String> placeHolders = Collections.nCopies(fieldNames.size(), "?");
String sqlInsert = String.format("INSERT INTO `%s`.`%s` ( `%s` ) VALUES ( %s )", schemaName, tableName,
StringUtils.join(fieldNames, "`,`"), StringUtils.join(placeHolders, ","));
@@ -118,21 +103,17 @@ public class MySqlWriterImpl extends AbstractDatabaseWriter implements IDatabase
PlatformTransactionManager transactionManager = new DataSourceTransactionManager(this.dataSource);
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager, definition);
Integer ret = transactionTemplate.execute(new TransactionCallback<Integer>() {
@Override
public Integer doInTransaction(TransactionStatus transactionStatus) {
try {
int[] affects = jdbcTemplate.batchUpdate(sqlInsert, recordValues, argTypes);
int affectCount = 0;
for (int i : affects) {
affectCount += i;
}
return affectCount;
} catch (Throwable t) {
transactionStatus.setRollbackOnly();
throw t;
Integer ret = transactionTemplate.execute((TransactionStatus transactionStatus) -> {
try {
int[] affects = jdbcTemplate.batchUpdate(sqlInsert, recordValues, argTypes);
int affectCount = 0;
for (int i : affects) {
affectCount += i;
}
return affectCount;
} catch (Throwable t) {
transactionStatus.setRollbackOnly();
throw t;
}
});

72
dbswitch-dbwriter/src/main/java/com/gitee/dbswitch/dbwriter/oracle/OracleWriterImpl.java
View File

@@ -4,13 +4,17 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.dbwriter.oracle;
import java.io.BufferedReader;
import java.io.IOException;
import java.sql.Clob;
import java.sql.SQLException;
import java.util.Collections;
import java.util.List;
import java.util.ArrayList;
import java.util.Objects;
import javax.sql.DataSource;
import org.apache.commons.lang3.StringUtils;
@@ -38,13 +42,11 @@ public class OracleWriterImpl extends AbstractDatabaseWriter implements IDatabas
@Override
public long write(List<String> fieldNames, List<Object[]> recordValues) {
List<String> placeHolders = new ArrayList<String>();
for (int i = 0; i < fieldNames.size(); ++i) {
placeHolders.add("?");
}
String schemaName = Objects.requireNonNull(this.schemaName, "schema-name名称为空，不合法!");
String tableName = Objects.requireNonNull(this.tableName, "table-name名称为空，不合法!");
List<String> placeHolders = Collections.nCopies(fieldNames.size(), "?");
String sqlInsert = String.format("INSERT INTO \"%s\".\"%s\" ( \"%s\" ) VALUES ( %s )", schemaName, tableName,
StringUtils.join(fieldNames, "\",\""), StringUtils.join(placeHolders, ","));
@@ -54,6 +56,18 @@ public class OracleWriterImpl extends AbstractDatabaseWriter implements IDatabas
argTypes[i] = this.columnType.get(col);
}
/** 处理Oracle的Clob类型需写入String类型的数据的问题 */
recordValues.parallelStream().forEach((Object[] row) -> {
for (int i = 0; i < row.length; ++i) {
try {
row[i] = convertClobToString(row[i]);
} catch (Exception e) {
log.warn("Field Value convert from java.sql.Clob to java.lang.String failed, field name is : {} ", fieldNames.get(i));
row[i] = null;
}
}
});
DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
definition.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
definition.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
@@ -82,4 +96,48 @@ public class OracleWriterImpl extends AbstractDatabaseWriter implements IDatabas
}
/**
* 将java.sql.Clob 类型转换为java.lang.String
* @param o
* @return
* @throws SQLException
* @throws IOException
*/
private Object convertClobToString(Object o) throws SQLException, IOException {
if (null == o) {
return null;
}
if (o instanceof Clob) {
Clob clob = (Clob) o;
java.io.Reader is = null;
java.io.BufferedReader reader = null;
try {
is = clob.getCharacterStream();
reader = new java.io.BufferedReader(is);
String line = reader.readLine();
StringBuilder sb = new StringBuilder();
while (line != null) {
sb.append(line);
line = reader.readLine();
}
return sb.toString();
} catch (SQLException | java.io.IOException e) {
throw new RuntimeException(e);
} finally {
try {
if (null != reader) {
reader.close();
}
if (null != is) {
is.close();
}
} catch (Exception ex) {
}
}
}
return o;
}
}

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/calcite/TheMssqlSqlDialect.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.calcite;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/calcite/TheMysqlSqlDialect.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.calcite;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/calcite/TheOracleSqlDialect.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.calcite;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/calcite/ThePostgresqlSqlDialect.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.calcite;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/calcite/TheSqlOrderBy.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.calcite;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/calcite/TheSqlRowOperator.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.calcite;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/constant/Const.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.constant;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/ddl/AbstractDatabaseDialect.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.ddl;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/ddl/AbstractSqlDdlOperator.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.ddl;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/ddl/pojo/ColumnDefinition.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.ddl.pojo;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/ddl/pojo/TableDefinition.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.ddl.pojo;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/ddl/sql/DdlSqlAlterTable.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.ddl.sql;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/ddl/sql/DdlSqlCreateTable.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.ddl.sql;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/ddl/sql/DdlSqlDropTable.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.ddl.sql;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/ddl/sql/DdlSqlTruncateTable.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.ddl.sql;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/ddl/sql/impl/GreenplumDialectImpl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.ddl.sql.impl;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/ddl/sql/impl/MySqlDialectImpl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.ddl.sql.impl;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/ddl/sql/impl/OracleDialectImpl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.ddl.sql.impl;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/ddl/sql/impl/PostgresDialectImpl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.ddl.sql.impl;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/ddl/type/GreenplumDataTypeEnum.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.ddl.type;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/ddl/type/MySqlDataTypeEnum.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.ddl.type;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/ddl/type/OracleDataTypeEnum.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.ddl.type;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/ddl/type/PostgresDataTypeEnum.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.ddl.type;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/service/ISqlConvertService.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.service;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/service/ISqlGeneratorService.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.service;

2
dbswitch-sql/src/main/java/com/gitee/dbswitch/sql/service/impl/CalciteSqlConvertServiceImpl.java
View File

@@ -4,7 +4,7 @@
// Use of this source code is governed by a BSD-style license
//
// Author: tang (inrgihc@126.com)
// Data : 2020/1/2
// Date : 2020/1/2
// Location: beijing , china
/////////////////////////////////////////////////////////////
package com.gitee.dbswitch.sql.service.impl;

Some files were not shown because too many files have changed in this diff Show More