Skip to content

oceanbase数据库兼容多表并行数据迁移 #2309

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
licc1021 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

oceanbase数据库兼容多表并行数据迁移 #2309

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 14 additions & 13 deletions ...rc/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/ColumnMetaCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,8 @@
import java.sql.Connection;
import java.sql.SQLException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Triple;
Expand All @@ -15,27 +17,26 @@ public class ColumnMetaCache {
private static final Logger LOG = LoggerFactory.getLogger(ColumnMetaCache.class);

private static String tableName;
private static Triple<List<String>, List<Integer>, List<String>> columnMeta = null;
//oceanbase数据库可以兼容多表并行进行数据迁移
private static Map<String ,Triple<List<String>, List<Integer>, List<String>>> columnMeta = new ConcurrentHashMap<>();

public ColumnMetaCache() {

}

public static void init(Connection connection, final String tableName, final List<String> columns) throws SQLException {
if (columnMeta == null) {
synchronized(ColumnMetaCache.class) {
ColumnMetaCache.tableName = tableName;
if (columnMeta == null) {
columnMeta = DBUtil.getColumnMetaData(connection,
tableName, StringUtils.join(columns, ","));
LOG.info("fetch columnMeta of table {} success", tableName);
}
}
Triple<List<String>, List<Integer>, List<String>> meta = columnMeta.get(tableName);
if (meta == null) {
ColumnMetaCache.tableName = tableName;
meta = DBUtil.getColumnMetaData(connection,
tableName, StringUtils.join(columns, ","));
columnMeta.put(tableName,meta);
LOG.info("fetch columnMeta of table {} success", tableName);
}
}
public static Triple<List<String>, List<Integer>, List<String>> getColumnMeta() {
return columnMeta;

public static Triple<List<String>, List<Integer>, List<String>> getColumnMeta(String tableName) {
return columnMeta.get(tableName);
}

}
2 changes: 1 addition & 1 deletion ...va/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/ConcurrentTableWriterTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -190,7 +190,7 @@ public void startWriteWithConnection(RecordReceiver recordReceiver, TaskPluginCo
LOG.warn("getColumnMetaData of table {} failed, retry the {} times ...", this.table, retryTimes);
}
ColumnMetaCache.init(connection, this.table, this.columns);
this.resultSetMetaData = ColumnMetaCache.getColumnMeta();
this.resultSetMetaData = ColumnMetaCache.getColumnMeta(this.table);
needRetry = false;
} catch (SQLException e) {
needRetry = true;
Expand Down