add license and doc, update split part

a49a · a49a · commit faafbd0db0a9 · 2020-07-21T14:20:48.000+08:00
diff --git a/core/src/main/java/com/dtstack/flink/sql/table/AbstractTableParser.java b/core/src/main/java/com/dtstack/flink/sql/table/AbstractTableParser.java
@@ -16,8 +16,6 @@
  * limitations under the License.
  */
 
-
-
 package com.dtstack.flink.sql.table;
 
 import com.dtstack.flink.sql.util.ClassUtil;
@@ -27,7 +25,6 @@
 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;
@@ -52,9 +49,6 @@ 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();
@@ -89,76 +83,56 @@ public boolean dealKeyPattern(String fieldRow, AbstractTableInfo tableInfo){
         return false;
     }
 
-    public void parseFieldsInfo(String fieldsInfo, AbstractTableInfo tableInfo){
+    public void parseFieldsInfo(String fieldsInfo, AbstractTableInfo tableInfo) {
 
         List<String> fieldRows = DtStringUtil.splitIgnoreQuota(fieldsInfo, ',');
 
-        ArrayList<String> buffer = new ArrayList<>();
-
         for (String fieldRow : fieldRows) {
             fieldRow = fieldRow.trim();
 
             if (StringUtils.isBlank(fieldRow)) {
                 throw new RuntimeException(String.format("table [%s],exists field empty.", tableInfo.getName()));
             }
 
-            // 处理复合类型，例如 ARRAY<ROW<foo INT, bar STRING>>
-            // 把ARRAY类型的长串字符压入Buffer
-            Matcher headMatcher = compositeTypeHeadPattern.matcher(fieldRow);
-            Matcher tailMatcher = compositeTypeTailPattern.matcher(fieldRow);
-            boolean isNotTail = !tailMatcher.matches();
-            boolean isToNeedPush = headMatcher.matches() || !buffer.isEmpty();
+            String[] fieldInfoArr = fieldRow.split("\\s+");
 
-            if (isNotTail && isToNeedPush) {
-                writeBuffer(buffer, fieldRow);
-            } else {
-                String[] fieldInfoArr;
-                if (tailMatcher.matches()) {
-                    buffer.add(fieldRow);
-                    fieldRow = String.join("", buffer);
-                    fieldInfoArr = readBuffer(buffer);
-                }  else {
-                    fieldInfoArr = fieldRow.split("\\s+");
-                }
+            String errorMsg = String.format("table [%s] field [%s] format error.", tableInfo.getName(), fieldRow);
+            Preconditions.checkState(fieldInfoArr.length >= 2, errorMsg);
 
-                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;
+            }
 
-                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();
 
-                //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();
-
-                Class fieldClass = null;
-                AbstractTableInfo.FieldExtraInfo fieldExtraInfo = null;
-
-                Matcher matcher = charTypePattern.matcher(fieldType);
-                if (matcher.find()) {
-                    fieldClass = dbTypeConvertToJavaType(CHAR_TYPE_NO_LENGTH);
-                    fieldExtraInfo = new AbstractTableInfo.FieldExtraInfo();
-                    fieldExtraInfo.setLength(Integer.valueOf(matcher.group(1)));
-                } else {
-                    fieldClass = dbTypeConvertToJavaType(fieldType);
-                }
+            Class fieldClass = null;
+            AbstractTableInfo.FieldExtraInfo fieldExtraInfo = null;
 
-                tableInfo.addPhysicalMappings(fieldInfoArr[0], fieldInfoArr[0]);
-                tableInfo.addField(fieldName);
-                tableInfo.addFieldClass(fieldClass);
-                tableInfo.addFieldType(fieldType);
-                tableInfo.addFieldExtraInfo(fieldExtraInfo);
+            Matcher matcher = charTypePattern.matcher(fieldType);
+            if (matcher.find()) {
+                fieldClass = dbTypeConvertToJavaType(CHAR_TYPE_NO_LENGTH);
+                fieldExtraInfo = new AbstractTableInfo.FieldExtraInfo();
+                fieldExtraInfo.setLength(Integer.valueOf(matcher.group(1)));
+            } else {
+                fieldClass = dbTypeConvertToJavaType(fieldType);
             }
+
+            tableInfo.addPhysicalMappings(fieldInfoArr[0], fieldInfoArr[0]);
+            tableInfo.addField(fieldName);
+            tableInfo.addFieldClass(fieldClass);
+            tableInfo.addFieldType(fieldType);
+            tableInfo.addFieldExtraInfo(fieldExtraInfo);
         }
 
         tableInfo.finish();
     }
 
-    public void dealPrimaryKey(Matcher matcher, AbstractTableInfo tableInfo){
+    public void dealPrimaryKey(Matcher matcher, AbstractTableInfo tableInfo) {
         String primaryFields = matcher.group(1).trim();
         String[] splitArry = primaryFields.split(",");
         List<String> primaryKes = Lists.newArrayList(splitArry);
diff --git a/core/src/main/java/com/dtstack/flink/sql/util/DataTypeUtils.java b/core/src/main/java/com/dtstack/flink/sql/util/DataTypeUtils.java
@@ -1,8 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 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;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -11,7 +27,6 @@
 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;
 
diff --git a/core/src/main/java/com/dtstack/flink/sql/util/DtStringUtil.java b/core/src/main/java/com/dtstack/flink/sql/util/DtStringUtil.java
@@ -89,6 +89,12 @@ public static List<String> splitIgnoreQuota(String str, char delimiter){
             } else if (c == ')' && !inSingleQuotes && !inQuotes) {
                 bracketLeftNum--;
                 b.append(c);
+            } else if (c == '<' && !inSingleQuotes && !inQuotes) {
+                bracketLeftNum++;
+                b.append(c);
+            } else if (c == '>' && !inSingleQuotes && !inQuotes) {
+                bracketLeftNum--;
+                b.append(c);
             } else {
                 b.append(c);
             }
diff --git a/docs/plugin/kafkaSource.md b/docs/plugin/kafkaSource.md
@@ -130,7 +130,85 @@ CREATE TABLE MyTable(
  );
 ```
 
-数组类型字段解析示例
+#### 数组类型字段解析示例
+
+声明数据类型为Array可以使用CROSS JOIN UNNEST语句对Array类型进行展开。（新模式，推荐）
+
+json样例
+```
+{
+    "id": 2,
+    "some_users": [
+        "foo",
+        "bar"
+    ]
+}
+```
+
+SQL样例
+```
+CREATE TABLE ods (
+    id INT,
+    some_users ARRAY<STRING>>
+) WITH (
+    ...
+);
+
+CREATE TABLE dwd (
+    id INT,
+    user_no INT,
+    user_info VARCHAR
+) WITH (
+    type ='console',
+);
+
+INSERT INTO dwd
+    SELECT id, user_info
+    FROM ods_foo
+    CROSS JOIN UNNEST(ods_foo.some_users) AS A (user_info);
+```
+
+json样例
+```
+{
+    "id": 4,
+    "some_users": [
+        {
+            "user_no": 12,
+            "user_info": "foo"
+
+        },
+        {
+            "user_no": 14,
+            "user_info": "bar"
+        }
+    ]
+
+```
+
+SQL样例
+```
+CREATE TABLE ods (
+    id INT,
+    some_users ARRAY<ROW<user_no INT, user_info STRING>>
+) WITH (
+    ...
+);
+
+CREATE TABLE dwd (
+    id INT,
+    user_no INT,
+    user_info VARCHAR
+) WITH (
+    type ='console',
+);
+
+INSERT INTO dwd
+    SELECT id, user_no, user_info
+    FROM ods_foo
+    CROSS JOIN UNNEST(ods_foo.some_users) AS A (user_no, user_info);
+```
+##### 旧模式
 
 json: {"name":"tom", "obj":{"channel": "root"}, "user": [{"pv": 4}, {"pv": 10}], "xctime":1572932485}
 ```
