add error msg

Author: a49a

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.Preconditions;
 import com.google.common.base.Strings;
 import com.google.common.collect.Iterators;
 import com.google.common.collect.Lists;
@@ -21,26 +22,21 @@
  **/
 public class DataTypeUtils {
 
-    private final static Pattern COMPOUND_TYPE_PATTERN = Pattern.compile("(.+?)<(.+)>");
+    private final static Pattern COMPOSITE_TYPE_PATTERN = Pattern.compile("(.+?)<(.+)>");
     private final static String ARRAY = "ARRAY";
     private final static String ROW = "ROW";
     private final static char FIELD_DELIMITER = ',';
     private final static char TYPE_DELIMITER = ' ';
 
-    private final static Pattern ATOMIC_TYPE_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() {}
 
-    public static TypeInformation convertToCollectionType(String string) {
-        Matcher matcher = COMPOUND_TYPE_PATTERN.matcher(string);
+    public static TypeInformation convertToCompositeType(String string) {
+        Matcher matcher = COMPOSITE_TYPE_PATTERN.matcher(string);
         // TODO 现在只支持ARRAY类型后续可以加入 MAP等类型
         if (matcher.find() && ARRAY.equals(matcher.group(1))) {
             return convertToArray(string);
         } else {
-            throw new RuntimeException("");
+            throw new RuntimeException("type " + string + "is not support!");
         }
     }
 
@@ -50,7 +46,7 @@ public static TypeInformation convertToCollectionType(String string) {
      * @return
      */
     public static TypeInformation convertToArray(String arrayTypeString) {
-        Matcher matcher = COMPOUND_TYPE_PATTERN.matcher(arrayTypeString);
+        Matcher matcher = COMPOSITE_TYPE_PATTERN.matcher(arrayTypeString);
         if (matcher.find() && ARRAY.equals(matcher.group(1))) {
             String elementTypeString = matcher.group(2);
             TypeInformation elementType;
@@ -61,7 +57,7 @@ public static TypeInformation convertToArray(String arrayTypeString) {
             }
             return Types.OBJECT_ARRAY(elementType);
         } else {
-            throw new RuntimeException("convert to array type error!");
+            throw new RuntimeException(arrayTypeString + "convert to array type error!");
         }
 
     }
@@ -71,19 +67,17 @@ public static TypeInformation convertToArray(String arrayTypeString) {
      * @param string
      */
     public static RowTypeInfo convertToRow(String string) {
-        Matcher matcher = COMPOUND_TYPE_PATTERN.matcher(string);
+        Matcher matcher = COMPOSITE_TYPE_PATTERN.matcher(string);
 
         if (matcher.find() &&
-            ROW.equals(
-                matcher.group(1).toUpperCase()
-            )
+            ROW.equals(matcher.group(1).toUpperCase())
         ) {
             String elementTypeStr = matcher.group(2);
             Iterable<String> typeInfo = splitCompositeTypeFields(elementTypeStr);
             Tuple2<TypeInformation[], String[]> tuple = genFieldInfo(typeInfo);
             return new RowTypeInfo(tuple.f0, tuple.f1);
         } else {
-            throw new RuntimeException("");
+            throw new RuntimeException(string + "convert to row type error!");
         }
     }
 
@@ -95,28 +89,26 @@ private static Iterable<String> splitCompositeTypeFields(String string) {
     }
 
     private static Tuple2<TypeInformation[], String[]> genFieldInfo(Iterable<String> typeInfo) {
-        int fieldsSize = Iterators.size(typeInfo.iterator());
         ArrayList<TypeInformation> types = Lists.newArrayList();
         ArrayList<String> fieldNames = Lists.newArrayList();
+
         for (String type : typeInfo) {
             Iterable<String> fieldInfo = Splitter
                 .on(TYPE_DELIMITER)
                 .trimResults()
                 .omitEmptyStrings()
                 .split(type);
+
             ArrayList<String> array = Lists.newArrayList(fieldInfo.iterator());
-            if (array.size() == 2) {
-                TypeInformation fieldType = convertToAtomicType(array.get(1));
-                types.add(fieldType);
-                fieldNames.add(array.get(0));
-            } else {
-                throw new RuntimeException();
-            }
+            Preconditions.checkState(array.size() == 2, "field info must be name with type");
+            TypeInformation fieldType = convertToAtomicType(array.get(1));
+            types.add(fieldType);
+            fieldNames.add(array.get(0));
         }
 
-        TypeInformation[] typesArray = types.toArray(new TypeInformation[types.size()]);
-        String[] fieldNamesArray = fieldNames.toArray(new String[fieldNames.size()]);
-        return Tuple2.of(typesArray, fieldNamesArray);
+        TypeInformation[] typeArray = types.toArray(new TypeInformation[types.size()]);
+        String[] fieldNameArray = fieldNames.toArray(new String[fieldNames.size()]);
+        return Tuple2.of(typeArray, fieldNameArray);
     }
 
     /**
@@ -156,42 +148,8 @@ public static TypeInformation convertToAtomicType(String string) {
             case "TIMESTAMP":
                 return Types.SQL_TIMESTAMP();
             default:
-                throw new RuntimeException("type " + string + "not supported");
+                throw new RuntimeException("type " + string + "not supported, please refer to the flink doc!");
         }
     }
 
-    /**
-     * 目前这个方法未使用，设置当初是想字段声明走统一的词法分析器(分词器)。
-     * @param fieldStmts
-     * @return
-     */
-    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;
-    }
 }