array type support complete

Author: a49a

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

@@ -27,6 +27,8 @@
 import com.google.common.collect.Maps;
 import org.apache.commons.lang3.StringUtils;
 
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.regex.Matcher;
@@ -50,6 +52,9 @@ public abstract class AbstractTableParser {
     private static Pattern physicalFieldFunPattern = Pattern.compile("\\w+\\((\\w+)\\)$");
     private static Pattern charTypePattern = Pattern.compile("(?i)CHAR\\((\\d*)\\)$");
 
+    private static Pattern compositeTypeHeadPattern = Pattern.compile(".+<.+<");
+    private static Pattern compositeTypeTailPattern = Pattern.compile(">\\s*>");
+
     private Map<String, Pattern> patternMap = Maps.newHashMap();
 
     private Map<String, ITableFieldDealHandler> handlerMap = Maps.newHashMap();
@@ -85,16 +90,45 @@ public boolean dealKeyPattern(String fieldRow, AbstractTableInfo tableInfo){
     }
 
     public void parseFieldsInfo(String fieldsInfo, AbstractTableInfo tableInfo){
-        // TODO 替换
+
         List<String> fieldRows = DtStringUtil.splitIgnoreQuota(fieldsInfo, ',');
+
+        ArrayList<String> cache = new ArrayList<>();
+        boolean currentIsCompositeType = false;
         for(String fieldRow : fieldRows){
             fieldRow = fieldRow.trim();
 
             if(StringUtils.isBlank(fieldRow)){
                 throw new RuntimeException(String.format("table [%s],exists field empty.", tableInfo.getName()));
             }
 
-            String[] filedInfoArr = fieldRow.split("\\s+");
+            // 处理复合类型，例如 ARRAY<ROW<foo INT, bar STRING>>
+            String[] filedInfoArr;
+            Matcher headMatcher = compositeTypeHeadPattern.matcher(fieldRow);
+            Matcher tailMatcher = compositeTypeTailPattern.matcher(fieldRow);
+
+            if (tailMatcher.find()) {
+                cache.add(fieldRow);
+                currentIsCompositeType = false;
+                fieldRow = String.join("", cache);
+                cache.clear();
+                String[] tmp = fieldRow.split("\\s+");
+                String[] type = Arrays.copyOfRange(tmp, 1, tmp.length);
+                filedInfoArr = new String[] {
+                    tmp[0],
+                    String.join(" ", type)
+                };
+            } else if (headMatcher.find() || currentIsCompositeType) {
+                currentIsCompositeType = true;
+                StringBuilder builder = new StringBuilder();
+                builder.append(fieldRow);
+                builder.append(",");
+                cache.add(builder.toString());
+                continue;
+            } else {
+                filedInfoArr = fieldRow.split("\\s+");
+            }
+
             if(filedInfoArr.length < 2 ){
                 throw new RuntimeException(String.format("table [%s] field [%s] format error.", tableInfo.getName(), fieldRow));
             }

core/src/main/java/com/dtstack/flink/sql/util/ClassUtil.java

@@ -69,6 +69,7 @@ public static Class<?> stringConvertClass(String str) {
             case "varchar":
             case "char":
             case "text":
+            case "string":
                 return String.class;
 
             case "real":

core/src/main/java/com/dtstack/flink/sql/util/DataTypeUtils.java

@@ -1,5 +1,6 @@
 package com.dtstack.flink.sql.util;
 
+import com.google.common.base.Strings;
 import com.google.common.collect.Iterators;
 import com.google.common.collect.Lists;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
@@ -9,6 +10,7 @@
 import org.apache.flink.table.api.Types;
 
 import java.util.ArrayList;
+import java.util.List;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -25,10 +27,10 @@ public class DataTypeUtils {
     private final static char FIELD_DELIMITER = ',';
     private final static char TYPE_DELIMITER = ' ';
 
-    private final static Pattern COMPLEX_TYPE_PATTERN = Pattern.compile("^\\s*(\\w+)\\s+(.+?<.+>)\\s*,");
     private final static Pattern ATOMIC_TYPE_PATTERN = Pattern.compile("^\\s*(\\w+)\\s+(\\w+)\\s*,");
-    private final static Pattern TAIL_PATTERN = Pattern.compile("^\\s*(\\w+)\\s+(.+?<.+>)\\s*");
-    private final static Pattern ATOMIC_TAIL_PATTERN = Pattern.compile("^\\s*(\\w+)\\s+(\\w+)\\s*");
+    private final static Pattern COMPLEX_TYPE_PATTERN = Pattern.compile("^\\s*(\\w+)\\s+(.+?<.+?>)\\s*,");
+    private final static Pattern ATOMIC_TAIL_PATTERN = Pattern.compile("^\\s*(\\w+)\\s+(\\w+)\\s*$");
+    private final static Pattern COMPLEX_TAIL_PATTERN = Pattern.compile("^\\s*(\\w+)\\s+(.+?<.+>)\\s*$");
 
     private DataTypeUtils() {}
 
@@ -158,8 +160,33 @@ public static TypeInformation convertToAtomicType(String string) {
         }
     }
 
-    public static void splitFields(String fieldStmt) {
-        // TODO 可以把每轮字符流开始的空白字符去掉。
+    public static ArrayList<String> fieldStmtLexer(String fieldStmts) {
+
+        String stmtStream = fieldStmts;
+        ArrayList<String> tokens = new ArrayList<>();
+        while (Strings.isNullOrEmpty(stmtStream)) {
+            Matcher atomicTypeMatcher = ATOMIC_TYPE_PATTERN.matcher(stmtStream);
+            Matcher complexTypeMatcher = COMPLEX_TYPE_PATTERN.matcher(stmtStream);
+            Matcher atomicTypeTailMatcher = ATOMIC_TAIL_PATTERN.matcher(stmtStream);
+            Matcher complexTypeTailMatcher = COMPLEX_TAIL_PATTERN.matcher(stmtStream);
+
+            String fieldStmt;
+
+            if (atomicTypeMatcher.find()) {
+                fieldStmt = atomicTypeMatcher.group(0);
+            } else if (complexTypeMatcher.find()) {
+                fieldStmt = complexTypeMatcher.group(0);
+            } else if (atomicTypeTailMatcher.find()) {
+                fieldStmt = atomicTypeTailMatcher.group(0);
+            } else if (complexTypeTailMatcher.find()) {
+                fieldStmt = complexTypeTailMatcher.group(0);
+            } else {
+                throw new RuntimeException("field declaration statement error" + fieldStmts);
+            }
 
+            tokens.add(fieldStmt);
+            stmtStream = stmtStream.substring(fieldStmt.length() + 1);
+        }
+        return tokens;
     }
 }