本次提交更新了转义的字符不需要进行大小写转换

oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/util/ObReaderUtils.java

@@ -1,37 +1,22 @@
 package com.alibaba.datax.plugin.reader.oceanbasev10reader.util;
 
-import java.sql.Connection;
+import com.alibaba.datax.common.element.*;
-import java.sql.PreparedStatement;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Statement;
-import java.sql.Timestamp;
-import java.util.*;
-import java.util.Map.Entry;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
 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;
 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 java.sql.*;
-import com.alibaba.datax.common.element.BytesColumn;
+import java.util.*;
-import com.alibaba.datax.common.element.Column;
+import java.util.Map.Entry;
-import com.alibaba.datax.common.element.DateColumn;
+import java.util.regex.Matcher;
-import com.alibaba.datax.common.element.DoubleColumn;
+import java.util.regex.Pattern;
-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.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;
-
-import javax.xml.crypto.Data;
 
 public class ObReaderUtils {
 
@@ -53,21 +38,32 @@ public class ObReaderUtils {
         return new HashSet(Arrays.asList(keywords.split(",")));
     }
 
-    public static void escapeDatabaseKeywords(List<String> keywords) {
+    public static String escapeDatabaseKeywords(String keyword) {
         if (databaseKeywords == null) {
-            if (isOracleMode(compatibleMode.toString())) {
+            if (isOracleMode(compatibleMode)) {
                 databaseKeywords = keywordsFromString2HashSet(ORACLE_KEYWORDS);
             } else {
                 databaseKeywords = keywordsFromString2HashSet(MYSQL_KEYWORDS);
             }
         }
-        char escapeChar = isOracleMode(compatibleMode.toString()) ? '"' : '`';
+        char escapeChar = isOracleMode(compatibleMode) ? '"' : '`';
-        for (int i = 0; i < keywords.size(); i++) {
-            String keyword = keywords.get(i);
         if (databaseKeywords.contains(keyword.toUpperCase())) {
             keyword = escapeChar + keyword + escapeChar;
         }
-            keywords.set(i, keyword);
+        return keyword;
+    }
+
+    public static void escapeDatabaseKeywords(List<String> keywords) {
+        for (int i = 0; i < keywords.size(); i++) {
+            keywords.set(i, escapeDatabaseKeywords(keywords.get(i)));
+        }
+    }
+
+    public static Boolean isEscapeMode(String keyword) {
+        if (isOracleMode(compatibleMode)) {
+            return keyword.startsWith("\"") && keyword.endsWith("\"");
+        } else {
+            return keyword.startsWith("`") && keyword.endsWith("`");
         }
     }
 
@@ -151,12 +147,15 @@ public class ObReaderUtils {
             ps = conn.createStatement();
             rs = ps.executeQuery(sql);
             while (rs.next()) {
-                String columnName = StringUtils.lowerCase(rs.getString("Column_name"));
+                String columnName = rs.getString("Column_name");
+                columnName = escapeDatabaseKeywords(columnName);
+                if (!isEscapeMode(columnName)) {
+                    columnName.toLowerCase();
+                }
                 if (!realIndex.contains(columnName)) {
                     realIndex.add(columnName);
                 }
             }
-            escapeDatabaseKeywords(realIndex);
 
             String[] pks = new String[realIndex.size()];
             realIndex.toArray(pks);

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

@@ -62,6 +62,7 @@ public class ConcurrentTableWriterTask extends CommonRdbmsWriter.Task {
 	private ObPartitionIdCalculator partCalculator = null;
 
 	private HashMap<Long, List<Record>> groupInsertValues;
+	List<Record> unknownPartRecords = new ArrayList<Record>();
 //	private List<Record> unknownPartRecords;
 	private List<Integer> partitionKeyIndexes;
 	
@@ -307,6 +308,18 @@ public class ConcurrentTableWriterTask extends CommonRdbmsWriter.Task {
 				}
 			}
 		}
+		if(unknownPartRecords.size()>0){
+			int retry = 0;
+			while (true) {
+				try {
+					concurrentWriter.addBatchRecords(unknownPartRecords);
+					break;
+				} catch (InterruptedException e) {
+					retry++;
+					LOG.info("Concurrent table writer is interrupted, retry {}", retry);
+				}
+			}
+		}
 	}
 	
 	private void addRecordToCache(final Record record) {
@@ -347,10 +360,11 @@ public class ConcurrentTableWriterTask extends CommonRdbmsWriter.Task {
 				groupInsertValues.put(partId, groupValues);
 			}
 		} else {
-			LOG.warn("add unknown part record {}", record);
+			LOG.debug("add unknown part record {}", record);
-			List<Record> unknownPartRecords = new ArrayList<Record>();
+
 			unknownPartRecords.add(record);
 			int i = 0;
+			if(unknownPartRecords.size()>batchSize){
 				while (true) {
 					if (i > 0) {
 						LOG.info("retry add batch record the {} times", i);
@@ -363,6 +377,8 @@ public class ConcurrentTableWriterTask extends CommonRdbmsWriter.Task {
 					}
 				}
 			}
+
+		}
 	}
 
 	private void checkMemStore() {

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/util/ObWriterUtils.java

@@ -11,7 +11,6 @@ import org.slf4j.LoggerFactory;
 
 import java.sql.*;
 import java.util.*;
-import java.util.stream.Collectors;
 
 public class ObWriterUtils {
 
@@ -26,8 +25,8 @@ public class ObWriterUtils {
 		return new HashSet(Arrays.asList(keywords.split(",")));
 	}
 
-	public static void escapeDatabaseKeywords(List<String> keywords) {
+	//java中的String的坑
-		//判断是否需要更改关键字集合
+	public static String escapeDatabaseKeywords(String keyword) {
 		if (databaseKeywords == null) {
 			if (isOracleMode()) {
 				databaseKeywords = keywordsFromString2HashSet(ORACLE_KEYWORDS);
@@ -36,12 +35,22 @@ public class ObWriterUtils {
 			}
 		}
 		char escapeChar = isOracleMode() ? '"' : '`';
-		for (int i = 0; i < keywords.size(); i++) {
-			String keyword = keywords.get(i);
 		if (databaseKeywords.contains(keyword.toUpperCase())) {
 			keyword = escapeChar + keyword + escapeChar;
 		}
-			keywords.set(i, keyword);
+		return keyword;
+	}
+
+	public static void escapeDatabaseKeywords(List<String> keywords) {
+		for (int i = 0; i < keywords.size(); i++) {
+			keywords.set(i, escapeDatabaseKeywords(keywords.get(i)));
+		}
+	}
+	public static Boolean isEscapeMode(String keyword){
+		if(isOracleMode()){
+			return keyword.startsWith("\"") && keyword.endsWith("\"");
+		}else{
+			return keyword.startsWith("`") && keyword.endsWith("`");
 		}
 	}
 	public static boolean isMemstoreFull(Connection conn, double memstoreThreshold) {
@@ -94,7 +103,16 @@ public class ObWriterUtils {
 	}
 
 	private static int[] getColumnIndex(List<String> columnsInIndex, List<String> allColumns) {
-		allColumns = allColumns.stream().map(String::toUpperCase).collect(Collectors.toList());
+		/**
+		 * JDK8的stream模型：将一种数据结构转化成通用的数据模型，并可在该模型上进行操作
+		 * map:接受一个函数引用，用于操作元素
+		 * collect：接受一个Collectors方法，用于将中间数据模型转化成目标数据结构
+		 */
+		for (int i = 0; i < allColumns.size(); i++) {
+			if (!ObWriterUtils.isEscapeMode(allColumns.get(i))) {
+				allColumns.set(i, allColumns.get(i).toUpperCase());
+			}
+		}
 		int[] colIdx = new int[columnsInIndex.size()];
 		for (int i = 0; i < columnsInIndex.size(); i++) {
 			int index = allColumns.indexOf(columnsInIndex.get(i));
@@ -146,7 +164,11 @@ public class ObWriterUtils {
 			rs = stmt.executeQuery(sql);
 			while (rs.next()) {
 				String keyName = rs.getString("Key_name");
-				String columnName = StringUtils.upperCase(rs.getString("Column_name"));
+				String columnName = rs.getString("Column_name");
+				columnName=escapeDatabaseKeywords(columnName);
+				if(!ObWriterUtils.isEscapeMode(columnName)){
+					columnName=columnName.toUpperCase();
+				}
 				List<String> s = uniqueKeys.get(keyName);
 				if (s == null) {
 					s = new ArrayList();
@@ -159,6 +181,7 @@ public class ObWriterUtils {
 		} finally {
 			asyncClose(rs, stmt, null);
 		}
+		//ObWriterUtils.escapeDatabaseKeywords(uniqueKeys);
 		return uniqueKeys;
 	}
 
@@ -315,6 +338,7 @@ public class ObWriterUtils {
 	 * @param e
 	 * @return
 	 */
+
 	public static boolean isFatalError(SQLException e) {
 		String sqlState = e.getSQLState();
 		if (StringUtils.startsWith(sqlState, "08")) {