Merge branch 'hotfix_1.10_4.0.x_29913' into 1.10_release_4.0.x

Author: gituser

.gitignore

@@ -14,4 +14,6 @@ lib/
 .DS_Store
 bin/nohup.out
 .DS_Store
-bin/sideSql.txt
+bin/sideSql.txt
+*.keytab
+krb5.conf

core/src/main/java/com/dtstack/flink/sql/table/AbstractTableParser.java

@@ -26,6 +26,7 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.api.java.tuple.Tuple2;
 
 import java.util.List;
 import java.util.Map;
@@ -49,6 +50,8 @@ public abstract class AbstractTableParser {
     private static Pattern nestJsonFieldKeyPattern = Pattern.compile("(?i)((@*\\S+\\.)*\\S+)\\s+(.+?)\\s+AS\\s+(\\w+)(\\s+NOT\\s+NULL)?$");
     private static Pattern physicalFieldFunPattern = Pattern.compile("\\w+\\((\\w+)\\)$");
     private static Pattern charTypePattern = Pattern.compile("(?i)CHAR\\((\\d*)\\)$");
+    private static Pattern typePattern = Pattern.compile("(\\S+)\\s+(\\w+.*)");
+
 
     private Map<String, Pattern> patternMap = Maps.newHashMap();
 
@@ -95,23 +98,16 @@ public void parseFieldsInfo(String fieldsInfo, AbstractTableInfo tableInfo) {
                 throw new RuntimeException(String.format("table [%s],exists field empty.", tableInfo.getName()));
             }
 
-            String[] fieldInfoArr = fieldRow.split("\\s+");
-
-            String errorMsg = String.format("table [%s] field [%s] format error.", tableInfo.getName(), fieldRow);
-            Preconditions.checkState(fieldInfoArr.length >= 2, errorMsg);
-
             boolean isMatcherKey = dealKeyPattern(fieldRow, tableInfo);
             if (isMatcherKey) {
                 continue;
             }
 
-            //Compatible situation may arise in space in the fieldName
-            String[] filedNameArr = new String[fieldInfoArr.length - 1];
-            System.arraycopy(fieldInfoArr, 0, filedNameArr, 0, fieldInfoArr.length - 1);
-            String fieldName = String.join(" ", filedNameArr);
-            String fieldType = fieldInfoArr[fieldInfoArr.length - 1 ].trim();
+            Tuple2<String, String> t = extractType(fieldRow, tableInfo.getName());
+            String fieldName = t.f0;
+            String fieldType = t.f1;
 
-            Class fieldClass = null;
+            Class fieldClass;
             AbstractTableInfo.FieldExtraInfo fieldExtraInfo = null;
 
             Matcher matcher = charTypePattern.matcher(fieldType);
@@ -123,7 +119,7 @@ public void parseFieldsInfo(String fieldsInfo, AbstractTableInfo tableInfo) {
                 fieldClass = dbTypeConvertToJavaType(fieldType);
             }
 
-            tableInfo.addPhysicalMappings(fieldInfoArr[0], fieldInfoArr[0]);
+            tableInfo.addPhysicalMappings(fieldName, fieldName);
             tableInfo.addField(fieldName);
             tableInfo.addFieldClass(fieldClass);
             tableInfo.addFieldType(fieldType);
@@ -133,11 +129,23 @@ public void parseFieldsInfo(String fieldsInfo, AbstractTableInfo tableInfo) {
         tableInfo.finish();
     }
 
+    private Tuple2<String, String> extractType(String fieldRow, String tableName) {
+        Matcher matcher = typePattern.matcher(fieldRow);
+        if (matcher.matches()) {
+            String fieldName = matcher.group(1);
+            String fieldType = matcher.group(2);
+            return Tuple2.of(fieldName, fieldType);
+        } else {
+            String errorMsg = String.format("table [%s] field [%s] format error.", tableName, fieldRow);
+            throw new RuntimeException(errorMsg);
+        }
+    }
+
     public void dealPrimaryKey(Matcher matcher, AbstractTableInfo tableInfo) {
         String primaryFields = matcher.group(1).trim();
-        String[] splitArry = primaryFields.split(",");
-        List<String> primaryKes = Lists.newArrayList(splitArry);
-        tableInfo.setPrimaryKeys(primaryKes);
+        String[] splitArray = primaryFields.split(",");
+        List<String> primaryKeys = Lists.newArrayList(splitArray);
+        tableInfo.setPrimaryKeys(primaryKeys);
     }
 
     /**