github/DataX
5
0
Fork 0
mirror of https://github.com/alibaba/DataX.git synced 2025-05-02 11:11:08 +08:00
Code Issues Packages Projects Releases Wiki Activity

Support OceanBase 4.x and fix OceanBase writer.

Browse Source
This commit is contained in:
dingxiaobo 2022-10-26 14:46:06 +08:00
parent c91dc5faaf
commit aa1c822edc
4 changed files with 8 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/util/ObVersion.java
View File

@ -16,6 +16,8 @@ public class ObVersion implements Comparable<ObVersion> {
private int patchNumber;
public static final ObVersion V2276 = valueOf("2.2.76");
public static final ObVersion V4000 = valueOf("4.0.0.0");
private static final ObVersion DEFAULT_VERSION =
valueOf(System.getProperty("defaultObVersion","3.2.3.0"));

3
oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/util/PartitionSplitUtil.java
View File

@ -157,7 +157,8 @@ public class PartitionSplitUtil {
conn = DBUtil.getConnection(DataBaseType.OceanBase, jdbcUrl, username, password);
ObVersion obVersion = ObReaderUtils.getObVersion(conn);
if (obVersion.compareTo(ObVersion.V2276) >= 0) {
if (obVersion.compareTo(ObVersion.V2276) >= 0 &&
obVersion.compareTo(ObVersion.V4000) < 0) {
allTable = "__all_table_v2";
}

9
oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/ConcurrentTableWriterTask.java
View File

@ -76,7 +76,6 @@ public class ConcurrentTableWriterTask extends CommonRdbmsWriter.Task {
private Condition condition = lock.newCondition();
private long startTime;
private boolean isOb2 = false;
private String obWriteMode = "update";
private boolean isOracleCompatibleMode = false;
private String obUpdateColumns = null;
@ -130,12 +129,6 @@ public class ConcurrentTableWriterTask extends CommonRdbmsWriter.Task {
concurrentWriter = new ConcurrentTableWriter(config, connectInfo, writeRecordSql);
allTaskInQueue = false;
}
String version = config.getString(Config.OB_VERSION);
int pIdx = version.lastIndexOf('.');
if ((Float.valueOf(version.substring(0, pIdx)) >= 2.1f)) {
isOb2 = true;
}
}
private void initPartCalculator(ServerConnectInfo connectInfo) {
@ -313,7 +306,7 @@ public class ConcurrentTableWriterTask extends CommonRdbmsWriter.Task {
LOG.warn("fail to get partition id: " + e1.getMessage() + ", record: " + record);
}
if (partId == null && isOb2) {
if (partId == null) {
LOG.debug("fail to calculate parition id, just put into the default buffer.");
partId = Long.MAX_VALUE;
}

3
plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/util/DBUtil.java
View File

@ -380,6 +380,9 @@ public final class DBUtil {
// unit ms
prop.put("oracle.jdbc.ReadTimeout", socketTimeout);
}
if (dataBaseType == DataBaseType.OceanBase) {
url = url.replace("jdbc:mysql:", "jdbc:oceanbase:");
}
return connect(dataBaseType, url, prop);
}