list = super.split(originalConfig, adviceNumber);
+ for (Configuration config : list) {
+ String jdbcUrl = config.getString(Key.JDBC_URL);
+ String obRegionName = getObRegionName(jdbcUrl);
+ config.set(CommonConstant.LOAD_BALANCE_RESOURCE_MARK, obRegionName);
+ }
+ return list;
+ }
+
+ private String getObRegionName(String jdbcUrl) {
+ if (jdbcUrl.startsWith(Constant.OB10_SPLIT_STRING)) {
+ String[] ss = jdbcUrl.split(Constant.OB10_SPLIT_STRING_PATTERN);
+ if (ss.length >= 2) {
+ String tenant = ss[1].trim();
+ String[] sss = tenant.split(":");
+ return sss[0];
+ }
+ }
+ return null;
+ }
+}
diff --git a/oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/ext/ReaderTask.java b/oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/ext/ReaderTask.java
new file mode 100644
index 00000000..073bb3cb
--- /dev/null
+++ b/oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/ext/ReaderTask.java
@@ -0,0 +1,301 @@
+package com.alibaba.datax.plugin.reader.oceanbasev10reader.ext;
+
+import java.sql.*;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.datax.common.element.Column;
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.plugin.RecordSender;
+import com.alibaba.datax.common.plugin.TaskPluginCollector;
+import com.alibaba.datax.common.statistics.PerfRecord;
+import com.alibaba.datax.common.statistics.PerfTrace;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.reader.CommonRdbmsReader;
+import com.alibaba.datax.plugin.rdbms.reader.Constant;
+import com.alibaba.datax.plugin.rdbms.reader.Key;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.rdbms.util.RdbmsException;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.Config;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.util.ObReaderUtils;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.util.TaskContext;
+
+public class ReaderTask extends CommonRdbmsReader.Task {
+ private static final Logger LOG = LoggerFactory.getLogger(ReaderTask.class);
+ private int taskGroupId = -1;
+ private int taskId = -1;
+
+ private String username;
+ private String password;
+ private String jdbcUrl;
+ private String mandatoryEncoding;
+ private int queryTimeoutSeconds;// 查询超时 默认48小时
+ private int readBatchSize;
+ private int retryLimit = 0;
+ private String compatibleMode = ObReaderUtils.OB_COMPATIBLE_MODE_MYSQL;
+ private boolean reuseConn = false;
+
+ public ReaderTask(int taskGroupId, int taskId) {
+ super(ObReaderUtils.DATABASE_TYPE, taskGroupId, taskId);
+ this.taskGroupId = taskGroupId;
+ this.taskId = taskId;
+ }
+
+ public void init(Configuration readerSliceConfig) {
+ /* for database connection */
+ username = readerSliceConfig.getString(Key.USERNAME);
+ password = readerSliceConfig.getString(Key.PASSWORD);
+ jdbcUrl = readerSliceConfig.getString(Key.JDBC_URL);
+ queryTimeoutSeconds = readerSliceConfig.getInt(Config.QUERY_TIMEOUT_SECOND,
+ Config.DEFAULT_QUERY_TIMEOUT_SECOND);
+ // ob10的处理
+ if(jdbcUrl.startsWith(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING)) {
+ String[] ss = jdbcUrl.split(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING_PATTERN);
+ if (ss.length == 3) {
+ LOG.info("this is ob1_0 jdbc url.");
+ username = ss[1].trim() + ":" + username;
+ jdbcUrl = ss[2];
+ }
+ }
+
+ if (ObReaderUtils.DATABASE_TYPE == DataBaseType.OceanBase) {
+ jdbcUrl = jdbcUrl.replace("jdbc:mysql:", "jdbc:oceanbase:") + "&socketTimeout=1800000&connectTimeout=60000"; //socketTimeout 半个小时
+ compatibleMode = ObReaderUtils.OB_COMPATIBLE_MODE_ORACLE;
+ } else {
+ jdbcUrl = jdbcUrl + "&socketTimeout=1800000&connectTimeout=60000"; //socketTimeout 半个小时
+ }
+ LOG.info("this is ob1_0 jdbc url. user=" + username + " :url=" + jdbcUrl);
+ mandatoryEncoding = readerSliceConfig.getString(Key.MANDATORY_ENCODING, "");
+ retryLimit = readerSliceConfig.getInt(Config.RETRY_LIMIT, Config.DEFAULT_RETRY_LIMIT);
+ LOG.info("retryLimit: "+ retryLimit);
+ }
+
+ private void buildSavePoint(TaskContext context) {
+ if (!ObReaderUtils.isUserSavePointValid(context)) {
+ LOG.info("user save point is not valid, set to null.");
+ context.setUserSavePoint(null);
+ }
+ }
+
+ /**
+ *
+ * 如果isTableMode && table有PK
+ *
+ * 则支持断点续读 (若pk不在原始的columns中,则追加到尾部,但不传给下游)
+ *
+ * 否则,则使用旧模式
+ */
+ @Override
+ public void startRead(Configuration readerSliceConfig, RecordSender recordSender,
+ TaskPluginCollector taskPluginCollector, int fetchSize) {
+ String querySql = readerSliceConfig.getString(Key.QUERY_SQL);
+ String table = readerSliceConfig.getString(Key.TABLE);
+ PerfTrace.getInstance().addTaskDetails(taskId, table + "," + jdbcUrl);
+ List columns = readerSliceConfig.getList(Key.COLUMN_LIST, String.class);
+ String where = readerSliceConfig.getString(Key.WHERE);
+ boolean weakRead = readerSliceConfig.getBool(Key.WEAK_READ, true); // default true, using weak read
+ String userSavePoint = readerSliceConfig.getString(Key.SAVE_POINT, null);
+ reuseConn = readerSliceConfig.getBool(Key.REUSE_CONN, false);
+ String partitionName = readerSliceConfig.getString(Key.PARTITION_NAME, null);
+ // 从配置文件中取readBatchSize,若无则用默认值
+ readBatchSize = readerSliceConfig.getInt(Config.READ_BATCH_SIZE, Config.DEFAULT_READ_BATCH_SIZE);
+ // 不能少于1万
+ if (readBatchSize < 10000) {
+ readBatchSize = 10000;
+ }
+ TaskContext context = new TaskContext(table, columns, where, fetchSize);
+ context.setQuerySql(querySql);
+ context.setWeakRead(weakRead);
+ context.setCompatibleMode(compatibleMode);
+ if (partitionName != null) {
+ context.setPartitionName(partitionName);
+ }
+ // Add the user save point into the context
+ context.setUserSavePoint(userSavePoint);
+ PerfRecord allPerf = new PerfRecord(taskGroupId, taskId, PerfRecord.PHASE.RESULT_NEXT_ALL);
+ allPerf.start();
+ boolean isTableMode = readerSliceConfig.getBool(Constant.IS_TABLE_MODE);
+ try {
+ startRead0(isTableMode, context, recordSender, taskPluginCollector);
+ } finally {
+ ObReaderUtils.close(null, null, context.getConn());
+ }
+ allPerf.end(context.getCost());
+ // 目前大盘是依赖这个打印，而之前这个Finish read record是包含了sql查询和result next的全部时间
+ LOG.info("finished read record by Sql: [{}\n] {}.", context.getQuerySql(), jdbcUrl);
+ }
+
+ private void startRead0(boolean isTableMode, TaskContext context, RecordSender recordSender,
+ TaskPluginCollector taskPluginCollector) {
+ // 不是table模式 直接使用原来的做法
+ if (!isTableMode) {
+ doRead(recordSender, taskPluginCollector, context);
+ return;
+ }
+ // check primary key index
+ Connection conn = DBUtil.getConnection(ObReaderUtils.DATABASE_TYPE, jdbcUrl, username, password);
+ ObReaderUtils.initConn4Reader(conn, queryTimeoutSeconds);
+ context.setConn(conn);
+ try {
+ ObReaderUtils.initIndex(conn, context);
+ ObReaderUtils.matchPkIndexs(conn, context);
+ } catch (Throwable e) {
+ LOG.warn("fetch PkIndexs fail,table=" + context.getTable(), e);
+ }
+ // 如果不是table 且 pk不存在 则仍然使用原来的做法
+ if (context.getPkIndexs() == null) {
+ doRead(recordSender, taskPluginCollector, context);
+ return;
+ }
+
+ // setup the user defined save point
+ buildSavePoint(context);
+
+ // 从这里开始就是 断点续读功能
+ // while(true) {
+ // 正常读 (需 order by pk asc)
+ // 如果遇到失败,分两种情况:
+ // a)已读出记录,则开始走增量读逻辑
+ // b)未读出记录,则走正常读逻辑(仍然需要order by pk asc)
+ // 正常结束 则 break
+ // }
+ context.setReadBatchSize(readBatchSize);
+ String getFirstQuerySql = ObReaderUtils.buildFirstQuerySql(context);
+ String appendQuerySql = ObReaderUtils.buildAppendQuerySql(conn, context);
+ LOG.warn("start table scan key : {}", context.getIndexName() == null ? "primary" : context.getIndexName());
+ context.setQuerySql(getFirstQuerySql);
+ boolean firstQuery = true;
+ // 原来打算firstQuery时 limit 1 减少
+ // 后来经过对比发现其实是多余的,因为:
+ // 1.假如走gmt_modified辅助索引,则直接索引扫描 不需要topN的order by
+ // 2.假如不走辅助索引,而是pk table scan,则减少排序规模并没有好处,因为下一次仍然要排序
+ // 减少这个多余的优化tip 可以让代码更易读
+ int retryCount = 0;
+ while (true) {
+ try {
+ boolean finish = doRead(recordSender, taskPluginCollector, context);
+ if (finish) {
+ break;
+ }
+ } catch (Throwable e) {
+ if (retryLimit == ++retryCount) {
+ throw RdbmsException.asQueryException(ObReaderUtils.DATABASE_TYPE, new Exception(e),
+ context.getQuerySql(), context.getTable(), username);
+ }
+ LOG.error("read fail, retry count " + retryCount + ", sleep 60 second, save point:" +
+ context.getSavePoint() + ", error: "+ e.getMessage());
+ ObReaderUtils.sleep(60000); // sleep 10s
+ }
+ // 假如原来的查询有查出数据,则改成增量查询
+ if (firstQuery && context.getPkIndexs() != null && context.getSavePoint() != null) {
+ context.setQuerySql(appendQuerySql);
+ firstQuery = false;
+ }
+ }
+ DBUtil.closeDBResources(null, context.getConn());
+ }
+
+ private boolean isConnectionAlive(Connection conn) {
+ if (conn == null) {
+ return false;
+ }
+ Statement stmt = null;
+ ResultSet rs = null;
+ String sql = "select 1" + (compatibleMode == ObReaderUtils.OB_COMPATIBLE_MODE_ORACLE ? " from dual" : "");
+ try {
+ stmt = conn.createStatement();
+ rs = stmt.executeQuery(sql);
+ rs.next();
+ } catch (Exception ex) {
+ LOG.info("connection is not alive: " + ex.getMessage());
+ return false;
+ } finally {
+ DBUtil.closeDBResources(rs, stmt, null);
+ }
+ return true;
+ }
+
+ private boolean doRead(RecordSender recordSender, TaskPluginCollector taskPluginCollector, TaskContext context) {
+ LOG.info("exe sql: {}", context.getQuerySql());
+ Connection conn = context.getConn();
+ if (reuseConn && isConnectionAlive(conn)) {
+ LOG.info("connection is alive, will reuse this connection.");
+ } else {
+ LOG.info("Create new connection for reader.");
+ conn = DBUtil.getConnection(ObReaderUtils.DATABASE_TYPE, jdbcUrl, username, password);
+ ObReaderUtils.initConn4Reader(conn, queryTimeoutSeconds);
+ context.setConn(conn);
+ }
+ PreparedStatement ps = null;
+ ResultSet rs = null;
+ PerfRecord perfRecord = new PerfRecord(taskGroupId, taskId, PerfRecord.PHASE.SQL_QUERY);
+ perfRecord.start();
+ try {
+ ps = conn.prepareStatement(context.getQuerySql(),
+ ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+ if (context.getPkIndexs() != null && context.getSavePoint() != null) {
+ Record savePoint = context.getSavePoint();
+ List point = ObReaderUtils.buildPoint(savePoint, context.getPkIndexs());
+ ObReaderUtils.binding(ps, point);
+ if (LOG.isWarnEnabled()) {
+ List pointForLog = new ArrayList();
+ for (Column c : point) {
+ pointForLog.add(c.asString());
+ }
+ LOG.warn("{} save point : {}", context.getTable(), StringUtils.join(pointForLog, ','));
+ }
+ }
+ // 打开流式接口
+ ps.setFetchSize(context.getFetchSize());
+ rs = ps.executeQuery();
+ ResultSetMetaData metaData = rs.getMetaData();
+ int columnNumber = metaData.getColumnCount();
+ long lastTime = System.nanoTime();
+ int count = 0;
+ for (; rs.next(); count++) {
+ context.addCost(System.nanoTime() - lastTime);
+ Record row = buildRecord(recordSender, rs, metaData, columnNumber, mandatoryEncoding,
+ taskPluginCollector);
+ // // 如果第一个record重复了,则不需要发送
+ // if (count == 0 &&
+ // ObReaderUtils.isPkEquals(context.getSavePoint(), row,
+ // context.getPkIndexs())) {
+ // continue;
+ // }
+ // 如果是querySql
+ if (context.getTransferColumnNumber() == -1
+ || row.getColumnNumber() == context.getTransferColumnNumber()) {
+ recordSender.sendToWriter(row);
+ } else {
+ Record newRow = recordSender.createRecord();
+ for (int i = 0; i < context.getTransferColumnNumber(); i++) {
+ newRow.addColumn(row.getColumn(i));
+ }
+ recordSender.sendToWriter(newRow);
+ }
+ context.setSavePoint(row);
+ lastTime = System.nanoTime();
+ }
+ LOG.info("end of sql: {}, " + count + "rows are read.", context.getQuerySql());
+ return context.getReadBatchSize() <= 0 || count < readBatchSize;
+ } catch (Exception e) {
+ ObReaderUtils.close(null, null, context.getConn());
+ context.setConn(null);
+ LOG.error("reader data fail", e);
+ throw RdbmsException.asQueryException(ObReaderUtils.DATABASE_TYPE, e, context.getQuerySql(),
+ context.getTable(), username);
+ } finally {
+ perfRecord.end();
+ if (reuseConn) {
+ ObReaderUtils.close(rs, ps, null);
+ } else {
+ ObReaderUtils.close(rs, ps, conn);
+ }
+ }
+ }
+}
diff --git a/oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/util/ObReaderUtils.java b/oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/util/ObReaderUtils.java
new file mode 100644
index 00000000..2290fb43
--- /dev/null
+++ b/oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/util/ObReaderUtils.java
@@ -0,0 +1,697 @@
+package com.alibaba.datax.plugin.reader.oceanbasev10reader.util;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.TreeMap;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.Set;
+import java.util.TreeSet;
+
+import org.apache.commons.lang3.ArrayUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.datax.common.element.BoolColumn;
+import com.alibaba.datax.common.element.BytesColumn;
+import com.alibaba.datax.common.element.Column;
+import com.alibaba.datax.common.element.DateColumn;
+import com.alibaba.datax.common.element.DoubleColumn;
+import com.alibaba.datax.common.element.LongColumn;
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.element.StringColumn;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.druid.sql.SQLUtils;
+import com.alibaba.druid.sql.ast.SQLExpr;
+import com.alibaba.druid.sql.ast.expr.SQLBinaryOpExpr;
+import com.alibaba.druid.sql.ast.expr.SQLBinaryOperator;
+
+public class ObReaderUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ObReaderUtils.class);
+
+ final static public String OB_COMPATIBLE_MODE = "obCompatibilityMode";
+ final static public String OB_COMPATIBLE_MODE_ORACLE = "ORACLE";
+ final static public String OB_COMPATIBLE_MODE_MYSQL = "MYSQL";
+
+ public static DataBaseType DATABASE_TYPE = DataBaseType.MySql;
+
+ public static void initConn4Reader(Connection conn, long queryTimeoutSeconds) {
+ String setQueryTimeout = "set ob_query_timeout=" + (queryTimeoutSeconds * 1000 * 1000L);
+ String setTrxTimeout = "set ob_trx_timeout=" + ((queryTimeoutSeconds + 5) * 1000 * 1000L);
+ Statement stmt = null;
+ try {
+ conn.setAutoCommit(true);
+ stmt = conn.createStatement();
+ stmt.execute(setQueryTimeout);
+ stmt.execute(setTrxTimeout);
+ LOG.warn("setAutoCommit=true;"+setQueryTimeout+";"+setTrxTimeout+";");
+ } catch (Throwable e) {
+ LOG.warn("initConn4Reader fail", e);
+ } finally {
+ DBUtil.closeDBResources(stmt, null);
+ }
+ }
+
+ public static void sleep(int ms) {
+ try {
+ Thread.sleep(ms);
+ } catch (InterruptedException e) {
+ }
+ }
+
+ /**
+ *
+ * @param conn
+ * @param context
+ */
+ public static void matchPkIndexs(Connection conn, TaskContext context) {
+ String[] pkColumns = getPkColumns(conn, context);
+ if (ArrayUtils.isEmpty(pkColumns)) {
+ LOG.warn("table=" + context.getTable() + " has no primary key");
+ return;
+ }
+ List columns = context.getColumns();
+ // 最后参与排序的索引列
+ context.setPkColumns(pkColumns);
+ int[] pkIndexs = new int[pkColumns.length];
+ for (int i = 0, n = pkColumns.length; i < n; i++) {
+ String pkc = pkColumns[i];
+ int j = 0;
+ for (int k = columns.size(); j < k; j++) {
+ // 如果用户定义的 columns中 带有 ``,也不影响,
+ // 最多只是在select里多加了几列PK column
+ if (StringUtils.equalsIgnoreCase(pkc, columns.get(j))) {
+ pkIndexs[i] = j;
+ break;
+ }
+ }
+ // 到这里 说明主键列不在columns中,则主动追加到尾部
+ if (j == columns.size()) {
+ columns.add(pkc);
+ pkIndexs[i] = columns.size() - 1;
+ }
+ }
+ context.setPkIndexs(pkIndexs);
+ }
+
+ private static String[] getPkColumns(Connection conn, TaskContext context) {
+ String tableName = context.getTable();
+ String sql = "show index from " + tableName + " where Key_name='PRIMARY'";
+ if (isOracleMode(context.getCompatibleMode())) {
+ tableName = tableName.toUpperCase();
+ sql = "SELECT cols.column_name Column_name "+
+ "FROM all_constraints cons, all_cons_columns cols " +
+ "WHERE cols.table_name = '" + tableName+ "' AND cons.constraint_type = 'P' " +
+ "AND cons.constraint_name = cols.constraint_name AND cons.owner = cols.owner";
+ }
+ LOG.info("get primary key by sql: " + sql);
+ Statement ps = null;
+ ResultSet rs = null;
+ List realIndex = new ArrayList();
+ realIndex.addAll(context.getSecondaryIndexColumns());
+ try {
+ ps = conn.createStatement();
+ rs = ps.executeQuery(sql);
+ while (rs.next()) {
+ String columnName = StringUtils.lowerCase(rs.getString("Column_name"));
+ if (!realIndex.contains(columnName)) {
+ realIndex.add(columnName);
+ }
+ }
+ String[] pks = new String[realIndex.size()];
+ realIndex.toArray(pks);
+ return pks;
+ } catch (Throwable e) {
+ LOG.error("show index from table fail :" + sql, e);
+ } finally {
+ close(rs, ps, null);
+ }
+ return null;
+ }
+
+ /**
+ * 首次查的SQL
+ *
+ * @param context
+ * @return
+ */
+ public static String buildFirstQuerySql(TaskContext context) {
+ String userSavePoint = context.getUserSavePoint();
+ String indexName = context.getIndexName();
+ String sql = "select ";
+ boolean weakRead = context.getWeakRead();
+ if (StringUtils.isNotEmpty(indexName)) {
+ String weakReadHint = weakRead ? "+READ_CONSISTENCY(WEAK)," : "+";
+ sql += " /*" + weakReadHint + "index(" + context.getTable() + " " + indexName + ")*/ ";
+ } else if (weakRead){
+ sql += " /*+READ_CONSISTENCY(WEAK)*/ ";
+ }
+ sql += StringUtils.join(context.getColumns(), ',');
+ sql += " from " + context.getTable();
+ if (context.getPartitionName() != null) {
+ sql += String.format(" partition(%s) ", context.getPartitionName());
+ }
+ if (StringUtils.isNotEmpty(context.getWhere())) {
+ sql += " where " + context.getWhere();
+ }
+
+ if (userSavePoint != null && userSavePoint.length() != 0) {
+ userSavePoint = userSavePoint.replace("=", ">");
+ sql += (StringUtils.isNotEmpty(context.getWhere()) ? " and " : " where ") + userSavePoint;
+ }
+
+ sql += " order by " + StringUtils.join(context.getPkColumns(), ',') + " asc";
+
+ // Using sub-query to apply rownum < readBatchSize since where has higher priority than order by
+ if (ObReaderUtils.isOracleMode(context.getCompatibleMode()) && context.getReadBatchSize() != -1) {
+ sql = String.format("select * from (%s) where rownum <= %d", sql, context.getReadBatchSize());
+ }
+
+ return sql;
+ }
+
+ /**
+ * 增量查的SQL
+ *
+ * @param conn
+ *
+ * @param context
+ * @return sql
+ */
+ public static String buildAppendQuerySql(Connection conn, TaskContext context) {
+ String indexName = context.getIndexName();
+ boolean weakRead = context.getWeakRead();
+ String sql = "select ";
+ if (StringUtils.isNotEmpty(indexName)) {
+ String weakReadHint = weakRead ? "+READ_CONSISTENCY(WEAK)," : "+";
+ sql += " /*"+ weakReadHint + "index(" + context.getTable() + " " + indexName + ")*/ ";
+ } else if (weakRead){
+ sql += " /*+READ_CONSISTENCY(WEAK)*/ ";
+ }
+ sql += StringUtils.join(context.getColumns(), ',') + " from " + context.getTable();
+
+ if (context.getPartitionName() != null) {
+ sql += String.format(" partition(%s) ", context.getPartitionName());
+ }
+
+ sql += " where ";
+ String append = "(" + StringUtils.join(context.getPkColumns(), ',') + ") > ("
+ + buildPlaceHolder(context.getPkColumns().length) + ")";
+
+ if (StringUtils.isNotEmpty(context.getWhere())) {
+ sql += "(" + context.getWhere() + ") and ";
+ }
+
+ sql = String.format("%s %s order by %s asc", sql, append, StringUtils.join(context.getPkColumns(), ','));
+
+ // Using sub-query to apply rownum < readBatchSize since where has higher priority than order by
+ if (ObReaderUtils.isOracleMode(context.getCompatibleMode()) && context.getReadBatchSize() != -1) {
+ sql = String.format("select * from (%s) where rownum <= %d", sql, context.getReadBatchSize());
+ }
+
+ return sql;
+ }
+
+ /**
+ * check if the userSavePoint is valid
+ *
+ * @param context
+ * @return true - valid, false - invalid
+ */
+ public static boolean isUserSavePointValid(TaskContext context) {
+ String userSavePoint = context.getUserSavePoint();
+ if (userSavePoint == null || userSavePoint.length() == 0) {
+ LOG.info("user save point is empty!");
+ return false;
+ }
+
+ LOG.info("validating user save point: " + userSavePoint);
+
+ final String patternString = "(.+)=(.+)";
+ Pattern parttern = Pattern.compile(patternString);
+ Matcher matcher = parttern.matcher(userSavePoint);
+ if (!matcher.find()) {
+ LOG.error("user save point format is not correct: " + userSavePoint);
+ return false;
+ }
+
+ List columnsInUserSavePoint = getColumnsFromUserSavePoint(userSavePoint);
+ List valuesInUserSavePoint = getValuesFromUserSavePoint(userSavePoint);
+ if (columnsInUserSavePoint.size() == 0 || valuesInUserSavePoint.size() == 0 ||
+ columnsInUserSavePoint.size() != valuesInUserSavePoint.size()) {
+ LOG.error("number of columns and values in user save point are different:" + userSavePoint);
+ return false;
+ }
+
+ String where = context.getWhere();
+ if (StringUtils.isNotEmpty(where)) {
+ for (String column : columnsInUserSavePoint) {
+ if (where.contains(column)) {
+ LOG.error("column " + column + " is conflict with where: " + where);
+ return false;
+ }
+ }
+ }
+
+ // Columns in userSavePoint must be the selected index.
+ String[] pkColumns = context.getPkColumns();
+ if (pkColumns.length != columnsInUserSavePoint.size()) {
+ LOG.error("user save point is not on the selected index.");
+ return false;
+ }
+
+ for (String column : columnsInUserSavePoint) {
+ boolean found = false;
+ for (String pkCol : pkColumns) {
+ if (pkCol.equals(column)) {
+ found = true;
+ break;
+ }
+ }
+ if (!found) {
+ LOG.error("column " + column + " is not on the selected index.");
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ private static String removeBracket(String str) {
+ final char leftBracket = '(';
+ final char rightBracket = ')';
+ if (str != null && str.contains(String.valueOf(leftBracket)) && str.contains(String.valueOf(rightBracket)) &&
+ str.indexOf(leftBracket) < str.indexOf(rightBracket)) {
+ return str.substring(str.indexOf(leftBracket)+1, str.indexOf(rightBracket));
+ }
+ return str;
+ }
+
+ private static List getColumnsFromUserSavePoint(String userSavePoint) {
+ return Arrays.asList(removeBracket(userSavePoint.split("=")[0]).split(","));
+ }
+
+ private static List getValuesFromUserSavePoint(String userSavePoint) {
+ return Arrays.asList(removeBracket(userSavePoint.split("=")[1]).split(","));
+ }
+
+ /**
+ * 先解析成where
+ *
+ * 再判断是否存在索引
+ *
+ * @param conn
+ * @param context
+ * @return
+ */
+ public static void initIndex(Connection conn, TaskContext context) {
+ if (StringUtils.isEmpty(context.getWhere())) {
+ return;
+ }
+ SQLExpr expr = SQLUtils.toSQLExpr(context.getWhere(), "mysql");
+ LOG.info("expr: " + expr);
+ List allColumnsInTab = getAllColumnFromTab(conn, context.getTable());
+ List allColNames = getColNames(allColumnsInTab, expr);
+
+ if (allColNames == null) {
+ return;
+ }
+
+ // Remove the duplicated column names
+ Set colNames = new TreeSet();
+ for (String colName : allColNames) {
+ if (!colNames.contains(colName)) {
+ colNames.add(colName);
+ }
+ }
+ List indexNames = getIndexName(conn, context.getTable(), colNames, context.getCompatibleMode());
+ findBestIndex(conn, indexNames, context.getTable(), context);
+ }
+
+ private static List getAllColumnFromTab(Connection conn, String tableName) {
+ String sql = "show columns from " + tableName;
+ Statement stmt = null;
+ ResultSet rs = null;
+ List allColumns = new ArrayList();
+ try {
+ stmt = conn.createStatement();
+ rs = stmt.executeQuery(sql);
+ while (rs.next()) {
+ allColumns.add(rs.getString("Field").toUpperCase());
+ }
+ } catch (Exception e) {
+ LOG.warn("fail to get all columns from table " + tableName, e);
+ } finally {
+ close(rs, stmt, null);
+ }
+
+ LOG.info("all columns in tab: " + String.join(",", allColumns));
+ return allColumns;
+ }
+
+ /**
+ * 找出where条件中的列名，目前仅支持全部为and条件，并且操作符为大于、大约等于、等于、小于、小于等于和不等于的表达式。
+ *
+ * test coverage: - c6 = 20180710 OR c4 = 320: no index selected - 20180710
+ * = c6: correct index selected - 20180710 = c6 and c4 = 320 or c2 < 100: no
+ * index selected
+ *
+ * @param expr
+ * @return
+ */
+ private static List getColNames(List allColInTab, SQLExpr expr) {
+ List colNames = new ArrayList();
+ if (expr instanceof SQLBinaryOpExpr) {
+ SQLBinaryOpExpr exp = (SQLBinaryOpExpr) expr;
+ if (exp.getOperator() == SQLBinaryOperator.BooleanAnd) {
+ List leftColumns = getColNames(allColInTab, exp.getLeft());
+ List rightColumns = getColNames(allColInTab, exp.getRight());
+ if (leftColumns == null || rightColumns == null) {
+ return null;
+ }
+ colNames.addAll(leftColumns);
+ colNames.addAll(rightColumns);
+ } else if (exp.getOperator() == SQLBinaryOperator.GreaterThan
+ || exp.getOperator() == SQLBinaryOperator.GreaterThanOrEqual
+ || exp.getOperator() == SQLBinaryOperator.Equality
+ || exp.getOperator() == SQLBinaryOperator.LessThan
+ || exp.getOperator() == SQLBinaryOperator.LessThanOrEqual
+ || exp.getOperator() == SQLBinaryOperator.NotEqual) {
+ // only support simple comparison operators
+ String left = SQLUtils.toMySqlString(exp.getLeft()).toUpperCase();
+ String right = SQLUtils.toMySqlString(exp.getRight()).toUpperCase();
+ LOG.debug("left: " + left + ", right: " + right);
+ if (allColInTab.contains(left)) {
+ colNames.add(left);
+ }
+
+ if (allColInTab.contains(right)) {
+ colNames.add(right);
+ }
+ } else {
+ // unsupported operators
+ return null;
+ }
+ }
+
+ return colNames;
+ }
+
+ private static Map> getAllIndex(Connection conn, String tableName, String compatibleMode) {
+ Map> allIndex = new HashMap>();
+ String sql = "show index from " + tableName;
+ if (isOracleMode(compatibleMode)) {
+ tableName = tableName.toUpperCase();
+ sql = "SELECT INDEX_NAME Key_name, COLUMN_NAME Column_name " +
+ "from dba_ind_columns where TABLE_NAME = '" + tableName +"' " +
+ " union all " +
+ "SELECT DISTINCT " +
+ "CASE " +
+ "WHEN cons.CONSTRAINT_TYPE = 'P' THEN 'PRIMARY' " +
+ "WHEN cons.CONSTRAINT_TYPE = 'U' THEN cons.CONSTRAINT_NAME " +
+ "ELSE '' " +
+ "END AS Key_name, " +
+ "cols.column_name Column_name " +
+ "FROM all_constraints cons, all_cons_columns cols " +
+ "WHERE cols.table_name = '" + tableName + "' AND cons.constraint_type in('P', 'U') " +
+ "AND cons.constraint_name = cols.constraint_name AND cons.owner = cols.owner";
+ }
+ Statement stmt = null;
+ ResultSet rs = null;
+
+ try {
+ LOG.info("running sql to get index: " + sql);
+ stmt = conn.createStatement();
+ rs = stmt.executeQuery(sql);
+ while (rs.next()) {
+ String keyName = rs.getString("Key_name");
+ String colName = rs.getString("Column_name").toUpperCase();
+ if (allIndex.containsKey(keyName)) {
+ allIndex.get(keyName).add(colName);
+ } else {
+ List allColumns = new ArrayList();
+ allColumns.add(colName);
+ allIndex.put(keyName, allColumns);
+ }
+ }
+
+ // add primary key to all index
+ if (allIndex.containsKey("PRIMARY")) {
+ List colsInPrimary = allIndex.get("PRIMARY");
+ for (String keyName : allIndex.keySet()) {
+ if (keyName.equals("PRIMARY")) {
+ continue;
+ }
+ allIndex.get(keyName).addAll(colsInPrimary);
+ }
+ }
+ } catch (Exception e) {
+ LOG.error("fail to get all keys from table" + sql, e);
+ } finally {
+ close(rs, stmt, null);
+ }
+
+ LOG.info("all index: " + allIndex.toString());
+ return allIndex;
+ }
+
+ /**
+ *
+ * @param conn
+ * @param table
+ * @param colNamesInCondition
+ * @return
+ */
+ private static List getIndexName(Connection conn, String table,
+ Set colNamesInCondition, String compatibleMode) {
+ List indexNames = new ArrayList();
+ if (colNamesInCondition == null || colNamesInCondition.size() == 0) {
+ LOG.info("there is no qulified conditions in the where clause, skip index selection.");
+ return indexNames;
+ }
+
+ LOG.info("columNamesInConditions: " + String.join(",", colNamesInCondition));
+
+ Map> allIndex = getAllIndex(conn, table, compatibleMode);
+ for (String keyName : allIndex.keySet()) {
+ boolean indexNotMatch = false;
+ // If the index does not have all the column in where conditions, it
+ // can not be chosen
+ // the selected index must start with the columns in where condition
+ if (allIndex.get(keyName).size() < colNamesInCondition.size()) {
+ indexNotMatch = true;
+ } else {
+ // the the first number columns of this index
+ int num = colNamesInCondition.size();
+ for (String colName : allIndex.get(keyName)) {
+ if (!colNamesInCondition.contains(colName)) {
+ indexNotMatch = true;
+ break;
+ }
+ if (--num == 0) {
+ break;
+ }
+ }
+ }
+
+ if (indexNotMatch) {
+ continue;
+ } else {
+ indexNames.add(keyName);
+ }
+ }
+
+ return indexNames;
+ }
+
+ /**
+ * 以 column开头的索引,可能有多个,也可能存在多列的情形
+ *
+ * 所以,需要选择列数最少的
+ *
+ * @param indexNames
+ * @param context
+ */
+ private static void findBestIndex(Connection conn, List indexNames, String table, TaskContext context) {
+ if (indexNames.size() == 0) {
+ LOG.warn("table has no index.");
+ return;
+ }
+
+ Map> allIndexs = new HashMap>();
+ String sql = "show index from " + table + " where key_name in (" + buildPlaceHolder(indexNames.size()) + ")";
+ if (isOracleMode(context.getCompatibleMode())) {
+ Map> allIndexInTab = getAllIndex(conn, table, context.getCompatibleMode());
+ for (String indexName : indexNames) {
+ if (allIndexInTab.containsKey(indexName)) {
+ Map index = new TreeMap();
+ List columnList = allIndexInTab.get(indexName);
+ for (int i = 1; i <= columnList.size(); i++) {
+ index.put(i, columnList.get(i-1));
+ }
+ allIndexs.put(indexName, index);
+ } else {
+ LOG.error("index does not exist: " + indexName);
+ }
+ }
+ } else {
+ PreparedStatement ps = null;
+ ResultSet rs = null;
+ try {
+ ps = conn.prepareStatement(sql);
+ for (int i = 0, n = indexNames.size(); i < n; i++) {
+ ps.setString(i + 1, indexNames.get(i));
+ }
+ rs = ps.executeQuery();
+ while (rs.next()) {
+ String keyName = rs.getString("Key_name");
+ Map index = allIndexs.get(keyName);
+ if (index == null) {
+ index = new TreeMap();
+ allIndexs.put(keyName, index);
+ }
+ int keyInIndex = rs.getInt("Seq_in_index");
+ String column = rs.getString("Column_name");
+ index.put(keyInIndex, column);
+ }
+ } catch (Throwable e) {
+ LOG.error("show index from table fail :" + sql, e);
+ } finally {
+ close(rs, ps, null);
+ }
+ }
+
+ LOG.info("possible index:" + allIndexs + ",where:" + context.getWhere());
+
+ Entry> chooseIndex = null;
+ int columnCount = Integer.MAX_VALUE;
+ for (Entry> entry : allIndexs.entrySet()) {
+ if (entry.getValue().size() < columnCount) {
+ columnCount = entry.getValue().size();
+ chooseIndex = entry;
+ }
+ }
+
+ if (chooseIndex != null) {
+ LOG.info("choose index name:" + chooseIndex.getKey() + ",columns:" + chooseIndex.getValue());
+ context.setIndexName(chooseIndex.getKey());
+ context.setSecondaryIndexColumns(new ArrayList(chooseIndex.getValue().values()));
+ }
+ }
+
+ /**
+ * 由于ObProxy存在bug,事务超时或事务被杀时,conn的close是没有响应的
+ *
+ * @param rs
+ * @param stmt
+ * @param conn
+ */
+ public static void close(final ResultSet rs, final Statement stmt, final Connection conn) {
+ DBUtil.closeDBResources(rs, stmt, conn);
+ }
+
+ /**
+ * 判断是否重复record
+ *
+ * @param savePoint
+ * @param row
+ * @param pkIndexs
+ * @return
+ */
+ public static boolean isPkEquals(Record savePoint, Record row, int[] pkIndexs) {
+ if (savePoint == null || row == null) {
+ return false;
+ }
+ try {
+ for (int index : pkIndexs) {
+ Object left = savePoint.getColumn(index).getRawData();
+ Object right = row.getColumn(index).getRawData();
+ if (!left.equals(right)) {
+ return false;
+ }
+ }
+ } catch (Throwable e) {
+ return false;
+ }
+ return true;
+ }
+
+ public static String buildPlaceHolder(int n) {
+ if (n <= 0) {
+ return "";
+ }
+ StringBuilder str = new StringBuilder(2 * n);
+ str.append('?');
+ for (int i = 1; i < n; i++) {
+ str.append(",?");
+ }
+ return str.toString();
+ }
+
+ public static void binding(PreparedStatement ps, List list) throws SQLException {
+ for (int i = 0, n = list.size(); i < n; i++) {
+ Column c = list.get(i);
+ if(c instanceof BoolColumn){
+ ps.setLong(i + 1, ((BoolColumn)c).asLong());
+ }else if(c instanceof BytesColumn){
+ ps.setBytes(i + 1, ((BytesColumn)c).asBytes());
+ }else if(c instanceof DateColumn){
+ ps.setTimestamp(i + 1, new Timestamp(((DateColumn)c).asDate().getTime()));
+ }else if(c instanceof DoubleColumn){
+ ps.setDouble(i + 1, ((DoubleColumn)c).asDouble());
+ }else if(c instanceof LongColumn){
+ ps.setLong(i + 1, ((LongColumn)c).asLong());
+ }else if(c instanceof StringColumn){
+ ps.setString(i + 1, ((StringColumn)c).asString());
+ }else{
+ ps.setObject(i + 1, c.getRawData());
+ }
+ }
+ }
+
+ public static List buildPoint(Record savePoint, int[] pkIndexs) {
+ List result = new ArrayList(pkIndexs.length);
+ for (int i = 0, n = pkIndexs.length; i < n; i++) {
+ result.add(savePoint.getColumn(pkIndexs[i]));
+ }
+ return result;
+ }
+
+ public static String getCompatibleMode(Connection conn) {
+ String compatibleMode = OB_COMPATIBLE_MODE_MYSQL;
+ String getCompatibleModeSql = "SHOW VARIABLES LIKE 'ob_compatibility_mode'";
+ Statement stmt = null;
+ ResultSet rs = null;
+ try {
+ stmt = conn.createStatement();
+ rs = stmt.executeQuery(getCompatibleModeSql);
+ if (rs.next()) {
+ compatibleMode = rs.getString("VALUE");
+ }
+ } catch (Exception e) {
+ LOG.error("fail to get ob compatible mode, using mysql as default: " + e.getMessage());
+ } finally {
+ DBUtil.closeDBResources(rs, stmt, conn);
+ }
+
+ LOG.info("ob compatible mode is " + compatibleMode);
+ return compatibleMode;
+ }
+
+ public static boolean isOracleMode(String mode) {
+ return (mode != null && OB_COMPATIBLE_MODE_ORACLE.equals(mode));
+ }
+}
diff --git a/oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/util/TaskContext.java b/oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/util/TaskContext.java
new file mode 100644
index 00000000..ba754a37
--- /dev/null
+++ b/oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/util/TaskContext.java
@@ -0,0 +1,176 @@
+package com.alibaba.datax.plugin.reader.oceanbasev10reader.util;
+
+import java.sql.Connection;
+import java.util.Collections;
+import java.util.List;
+
+import com.alibaba.datax.common.element.Record;
+
+public class TaskContext {
+ private Connection conn;
+ private final String table;
+ private String indexName;
+ // 辅助索引的字段列表
+ private List secondaryIndexColumns = Collections.emptyList();
+ private String querySql;
+ private final String where;
+ private final int fetchSize;
+ private long readBatchSize = -1;
+ private boolean weakRead = true;
+ private String userSavePoint;
+ private String compatibleMode = ObReaderUtils.OB_COMPATIBLE_MODE_MYSQL;
+
+ public String getPartitionName() {
+ return partitionName;
+ }
+
+ public void setPartitionName(String partitionName) {
+ this.partitionName = partitionName;
+ }
+
+ private String partitionName;
+
+ // 断点续读的保存点
+ private volatile Record savePoint;
+
+ // pk在column中的index,用于绑定变量时从savePoint中读取值
+ // 如果这个值为null,则表示 不是断点续读的场景
+ private int[] pkIndexs;
+
+ private final List columns;
+
+ private String[] pkColumns;
+
+ private long cost;
+
+ private final int transferColumnNumber;
+
+ public TaskContext(String table, List columns, String where, int fetchSize) {
+ super();
+ this.table = table;
+ this.columns = columns;
+ // 针对只有querySql的场景
+ this.transferColumnNumber = columns == null ? -1 : columns.size();
+ this.where = where;
+ this.fetchSize = fetchSize;
+ }
+
+ public Connection getConn() {
+ return conn;
+ }
+
+ public void setConn(Connection conn) {
+ this.conn = conn;
+ }
+
+ public String getIndexName() {
+ return indexName;
+ }
+
+ public void setIndexName(String indexName) {
+ this.indexName = indexName;
+ }
+
+ public List getSecondaryIndexColumns() {
+ return secondaryIndexColumns;
+ }
+
+ public void setSecondaryIndexColumns(List secondaryIndexColumns) {
+ this.secondaryIndexColumns = secondaryIndexColumns;
+ }
+
+ public String getQuerySql() {
+ if (readBatchSize == -1 || ObReaderUtils.isOracleMode(compatibleMode)) {
+ return querySql;
+ } else {
+ return querySql + " limit " + readBatchSize;
+ }
+ }
+
+ public void setQuerySql(String querySql) {
+ this.querySql = querySql;
+ }
+
+ public String getWhere() {
+ return where;
+ }
+
+ public Record getSavePoint() {
+ return savePoint;
+ }
+
+ public void setSavePoint(Record savePoint) {
+ this.savePoint = savePoint;
+ }
+
+ public int[] getPkIndexs() {
+ return pkIndexs;
+ }
+
+ public void setPkIndexs(int[] pkIndexs) {
+ this.pkIndexs = pkIndexs;
+ }
+
+ public List getColumns() {
+ return columns;
+ }
+
+ public String[] getPkColumns() {
+ return pkColumns;
+ }
+
+ public void setPkColumns(String[] pkColumns) {
+ this.pkColumns = pkColumns;
+ }
+
+ public String getTable() {
+ return table;
+ }
+
+ public int getFetchSize() {
+ return fetchSize;
+ }
+
+ public long getCost() {
+ return cost;
+ }
+
+ public void addCost(long cost) {
+ this.cost += cost;
+ }
+
+ public int getTransferColumnNumber() {
+ return transferColumnNumber;
+ }
+
+ public long getReadBatchSize() {
+ return readBatchSize;
+ }
+
+ public void setReadBatchSize(long readBatchSize) {
+ this.readBatchSize = readBatchSize;
+ }
+
+ public boolean getWeakRead() {
+ return weakRead;
+ }
+
+ public void setWeakRead(boolean weakRead) {
+ this.weakRead = weakRead;
+ }
+
+ public String getUserSavePoint() {
+ return userSavePoint;
+ }
+ public void setUserSavePoint(String userSavePoint) {
+ this.userSavePoint = userSavePoint;
+ }
+
+ public String getCompatibleMode() {
+ return compatibleMode;
+ }
+
+ public void setCompatibleMode(String compatibleMode) {
+ this.compatibleMode = compatibleMode;
+ }
+}
diff --git a/oceanbasev10reader/src/main/resources/plugin.json b/oceanbasev10reader/src/main/resources/plugin.json
new file mode 100644
index 00000000..66acbd62
--- /dev/null
+++ b/oceanbasev10reader/src/main/resources/plugin.json
@@ -0,0 +1,6 @@
+{
+ "name": "oceanbasev10reader",
+ "class": "com.alibaba.datax.plugin.reader.oceanbasev10reader.OceanBaseReader",
+ "description": "read data from oceanbase with SQL interface",
+ "developer": "oceanbase"
+}
\ No newline at end of file
diff --git a/oceanbasev10writer/pom.xml b/oceanbasev10writer/pom.xml
new file mode 100644
index 00000000..cbe19732
--- /dev/null
+++ b/oceanbasev10writer/pom.xml
@@ -0,0 +1,126 @@
+
+
+
+ datax-all
+ com.alibaba.datax
+ 0.0.1-SNAPSHOT
+
+ 4.0.0
+
+ oceanbasev10writer
+
+ com.alibaba.datax
+ 0.0.1-SNAPSHOT
+
+
+
+ com.alibaba.datax
+ datax-common
+ ${datax-project-version}
+
+
+ slf4j-log4j12
+ org.slf4j
+
+
+
+
+ com.alibaba.datax
+ plugin-rdbms-util
+ ${datax-project-version}
+
+
+ org.slf4j
+ slf4j-api
+
+
+ ch.qos.logback
+ logback-classic
+
+
+ org.springframework
+ spring-test
+ 4.0.4.RELEASE
+ test
+
+
+
+
+ com.alipay.oceanbase
+ oceanbase-connector-java
+ 3.2.0
+ system
+ ${basedir}/src/main/libs/oceanbase-connector-java-3.2.0.jar
+
+
+ com.alipay.oceanbase
+ oceanbase-client
+
+
+
+
+
+ log4j
+ log4j
+ 1.2.16
+
+
+ org.json
+ json
+ 20160810
+
+
+ junit
+ junit
+ 4.11
+ test
+
+
+
+
+
+
+ src/main/java
+
+ **/*.properties
+
+
+
+
+
+
+ maven-compiler-plugin
+
+ ${jdk-version}
+ ${jdk-version}
+ ${project-sourceEncoding}
+
+
+
+
+ maven-assembly-plugin
+
+
+ src/main/assembly/package.xml
+
+ datax
+
+
+
+ dwzip
+ package
+
+ single
+
+
+
+
+
+
+
diff --git a/oceanbasev10writer/src/main/assembly/package.xml b/oceanbasev10writer/src/main/assembly/package.xml
new file mode 100644
index 00000000..2529b4d4
--- /dev/null
+++ b/oceanbasev10writer/src/main/assembly/package.xml
@@ -0,0 +1,35 @@
+
+
+
+ dir
+
+ false
+
+
+ src/main/resources
+
+ plugin.json
+ plugin_job_template.json
+
+ plugin/writer/oceanbasev10writer
+
+
+ target/
+
+ oceanbasev10writer-0.0.1-SNAPSHOT.jar
+
+ plugin/writer/oceanbasev10writer
+
+
+
+
+
+ false
+ plugin/writer/oceanbasev10writer/libs
+ runtime
+
+
+
diff --git a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/Config.java b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/Config.java
new file mode 100644
index 00000000..9fa3cd9a
--- /dev/null
+++ b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/Config.java
@@ -0,0 +1,62 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer;
+
+public interface Config {
+
+ String MEMSTORE_THRESHOLD = "memstoreThreshold";
+
+ double DEFAULT_MEMSTORE_THRESHOLD = 0.9d;
+
+ String MEMSTORE_CHECK_INTERVAL_SECOND = "memstoreCheckIntervalSecond";
+
+ long DEFAULT_MEMSTORE_CHECK_INTERVAL_SECOND = 30;
+
+ int DEFAULT_BATCH_SIZE = 100;
+ int MAX_BATCH_SIZE = 4096;
+
+ String FAIL_TRY_COUNT = "failTryCount";
+
+ int DEFAULT_FAIL_TRY_COUNT = 10000;
+
+ String WRITER_THREAD_COUNT = "writerThreadCount";
+
+ int DEFAULT_WRITER_THREAD_COUNT = 1;
+
+ String CONCURRENT_WRITE = "concurrentWrite";
+
+ boolean DEFAULT_CONCURRENT_WRITE = true;
+
+ String OB_VERSION = "obVersion";
+ String TIMEOUT = "timeout";
+
+ String PRINT_COST = "printCost";
+ boolean DEFAULT_PRINT_COST = false;
+
+ String COST_BOUND = "costBound";
+ long DEFAULT_COST_BOUND = 20;
+
+ String MAX_ACTIVE_CONNECTION = "maxActiveConnection";
+ int DEFAULT_MAX_ACTIVE_CONNECTION = 2000;
+
+ String WRITER_SUB_TASK_COUNT = "writerSubTaskCount";
+ int DEFAULT_WRITER_SUB_TASK_COUNT = 1;
+ int MAX_WRITER_SUB_TASK_COUNT = 4096;
+
+ String OB_WRITE_MODE = "obWriteMode";
+ String OB_COMPATIBLE_MODE = "obCompatibilityMode";
+ String OB_COMPATIBLE_MODE_ORACLE = "ORACLE";
+ String OB_COMPATIBLE_MODE_MYSQL = "MYSQL";
+
+ String OCJ_GET_CONNECT_TIMEOUT = "ocjGetConnectTimeout";
+ int DEFAULT_OCJ_GET_CONNECT_TIMEOUT = 5000; // 5s
+
+ String OCJ_PROXY_CONNECT_TIMEOUT = "ocjProxyConnectTimeout";
+ int DEFAULT_OCJ_PROXY_CONNECT_TIMEOUT = 5000; // 5s
+
+ String OCJ_CREATE_RESOURCE_TIMEOUT = "ocjCreateResourceTimeout";
+ int DEFAULT_OCJ_CREATE_RESOURCE_TIMEOUT = 60000; // 60s
+
+ String OB_UPDATE_COLUMNS = "obUpdateColumns";
+
+ String USE_PART_CALCULATOR = "usePartCalculator";
+ boolean DEFAULT_USE_PART_CALCULATOR = false;
+}
diff --git a/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/OceanBaseV10Writer.java b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/OceanBaseV10Writer.java
new file mode 100644
index 00000000..89ef1c52
--- /dev/null
+++ b/oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/OceanBaseV10Writer.java
@@ -0,0 +1,246 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer;
+
+import java.sql.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.DbUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.datax.common.plugin.RecordReceiver;
+import com.alibaba.datax.common.spi.Writer;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter;
+import com.alibaba.datax.plugin.rdbms.writer.Constant;
+import com.alibaba.datax.plugin.rdbms.writer.Key;
+import com.alibaba.datax.plugin.rdbms.writer.util.WriterUtil;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.task.ConcurrentTableWriterTask;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.task.SingleTableWriterTask;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils;
+
+/**
+ * 2016-04-07
+ *
+ * 专门针对OceanBase1.0的Writer
+ *
+ * @author biliang.wbl
+ *
+ */
+public class OceanBaseV10Writer extends Writer {
+ private static DataBaseType DATABASE_TYPE = DataBaseType.OceanBase;
+
+ /**
+ * Job 中的方法仅执行一次，Task 中方法会由框架启动多个 Task 线程并行执行。
+ *
+ * 整个 Writer 执行流程是：
+ *
+ *
+ * Job类init-->prepare-->split
+ *
+ * Task类init-->prepare-->startWrite-->post-->destroy
+ * Task类init-->prepare-->startWrite-->post-->destroy
+ *
+ * Job类post-->destroy
+ *
+ */
+ public static class Job extends Writer.Job {
+ private Configuration originalConfig = null;
+ private CommonRdbmsWriter.Job commonJob;
+ private static final Logger LOG = LoggerFactory.getLogger(Job.class);
+
+ /**
+ * 注意：此方法仅执行一次。 最佳实践：通常在这里对用户的配置进行校验：是否缺失必填项？有无错误值？有没有无关配置项？...
+ * 并给出清晰的报错/警告提示。校验通常建议采用静态工具类进行，以保证本类结构清晰。
+ */
+ @Override
+ public void init() {
+ this.originalConfig = super.getPluginJobConf();
+ checkCompatibleMode(originalConfig);
+ this.commonJob = new CommonRdbmsWriter.Job(DATABASE_TYPE);
+ this.commonJob.init(this.originalConfig);
+ }
+
+ /**
+ * 注意：此方法仅执行一次。 最佳实践：如果 Job 中有需要进行数据同步之前的处理，可以在此处完成，如果没有必要则可以直接去掉。
+ */
+ // 一般来说，是需要推迟到 task 中进行pre 的执行（单表情况例外）
+ @Override
+ public void prepare() {
+ int tableNumber = originalConfig.getInt(Constant.TABLE_NUMBER_MARK);
+ if (tableNumber == 1) {
+ this.commonJob.prepare(this.originalConfig);
+ final String version = fetchServerVersion(originalConfig);
+ originalConfig.set(Config.OB_VERSION, version);
+ }
+
+ String username = originalConfig.getString(Key.USERNAME);
+ String password = originalConfig.getString(Key.PASSWORD);
+
+ // 获取presql配置，并执行
+ List preSqls = originalConfig.getList(Key.PRE_SQL, String.class);
+ if (preSqls == null || preSqls.size() == 0) {
+ return;
+ }
+
+ List