Merge remote-tracking branch 'origin/1.8_test_3.10.x' into feat_1.8_hbaseRefactor

# Conflicts:
#	hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseOutputFormat.java

Author: dapeng

.gitignore

@@ -8,6 +8,7 @@ target/
 *.eclipse.*
 *.iml
 plugins/
+sqlplugins/
 lib/
 .vertx/
 .DS_Store

clickhouse/clickhouse-sink/src/main/java/com/dtstack/flink/sql/sink/clickhouse/ClickhouseDialect.java

@@ -0,0 +1,46 @@
+/*
+ * 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.sink.clickhouse;
+
+import com.dtstack.flink.sql.sink.rdb.dialect.JDBCDialect;
+
+import java.util.Optional;
+
+/**
+ * Date: 2020/1/15
+ * Company: www.dtstack.com
+ * @author maqi
+ */
+public class ClickhouseDialect implements JDBCDialect {
+
+    @Override
+    public boolean canHandle(String url) {
+        return url.startsWith("jdbc:clickhouse:");
+    }
+
+    @Override
+    public Optional<String> defaultDriverName() {
+        return Optional.of("ru.yandex.clickhouse.ClickHouseDriver");
+    }
+
+    @Override
+    public String getUpdateStatement(String tableName, String[] fieldNames, String[] conditionFields) {
+        throw new RuntimeException("Clickhouse does not support update sql, please remove primary key or use append mode");
+    }
+}

core/pom.xml

@@ -122,6 +122,7 @@
             <artifactId>junit</artifactId>
             <version>4.12</version>
         </dependency>
+
     </dependencies>
 
     <build>

core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java

@@ -38,7 +38,6 @@
 import org.apache.flink.types.Row;
 
 import com.dtstack.flink.sql.classloader.ClassLoaderManager;
-import com.dtstack.flink.sql.constrant.ConfigConstrant;
 import com.dtstack.flink.sql.enums.ClusterMode;
 import com.dtstack.flink.sql.enums.ECacheType;
 import com.dtstack.flink.sql.enums.EPluginLoadMode;

core/src/main/java/com/dtstack/flink/sql/format/dtnest/DtNestRowDeserializationSchema.java

@@ -24,6 +24,7 @@
 import org.apache.flink.api.common.serialization.AbstractDeserializationSchema;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
@@ -34,6 +35,7 @@
 import org.apache.flink.types.Row;
 
 import java.io.IOException;
+import java.lang.reflect.Array;
 import java.sql.Date;
 import java.sql.Time;
 import java.sql.Timestamp;
@@ -43,7 +45,7 @@
 
 /**
  * source data parse to json format
- *
+ * <p>
  * Date: 2019/12/12
  * Company: www.dtstack.com
  *
@@ -53,51 +55,33 @@ public class DtNestRowDeserializationSchema extends AbstractDeserializationSchem
 
     private final ObjectMapper objectMapper = new ObjectMapper();
 
-    private Map<String, String> rowAndFieldMapping;
-    private Map<String, JsonNode> nodeAndJsonNodeMapping = Maps.newHashMap();
+    private final Map<String, String> rowAndFieldMapping;
+    private final Map<String, JsonNode> nodeAndJsonNodeMapping = Maps.newHashMap();
 
     private final String[] fieldNames;
     private final TypeInformation<?>[] fieldTypes;
-    private List<AbstractTableInfo.FieldExtraInfo> fieldExtraInfos;
+    private final List<AbstractTableInfo.FieldExtraInfo> fieldExtraInfos;
+    private final String charsetName;
 
-    public DtNestRowDeserializationSchema(TypeInformation<Row> typeInfo, Map<String, String> rowAndFieldMapping, List<AbstractTableInfo.FieldExtraInfo> fieldExtraInfos) {
+    public DtNestRowDeserializationSchema(TypeInformation<Row> typeInfo, Map<String, String> rowAndFieldMapping,
+                                          List<AbstractTableInfo.FieldExtraInfo> fieldExtraInfos,
+                                          String charsetName) {
         this.fieldNames = ((RowTypeInfo) typeInfo).getFieldNames();
         this.fieldTypes = ((RowTypeInfo) typeInfo).getFieldTypes();
         this.rowAndFieldMapping = rowAndFieldMapping;
         this.fieldExtraInfos = fieldExtraInfos;
+        this.charsetName = charsetName;
     }
 
     @Override
     public Row deserialize(byte[] message) throws IOException {
-        JsonNode root = objectMapper.readTree(message);
+        String decoderStr = new String(message, charsetName);
+        JsonNode root = objectMapper.readTree(decoderStr);
         this.parseTree(root, null);
-        Row row = new Row(fieldNames.length);
-
-        try {
-            for (int i = 0; i < fieldNames.length; i++) {
-                JsonNode node = getIgnoreCase(fieldNames[i]);
-                AbstractTableInfo.FieldExtraInfo fieldExtraInfo = fieldExtraInfos.get(i);
-
-                if (node == null) {
-                    if (fieldExtraInfo != null && fieldExtraInfo.getNotNull()) {
-                        throw new IllegalStateException("Failed to find field with name '"
-                                + fieldNames[i] + "'.");
-                    } else {
-                        row.setField(i, null);
-                    }
-                } else {
-                    // Read the value as specified type
-                    Object value = convert(node, fieldTypes[i]);
-                    row.setField(i, value);
-                }
-            }
-            return row;
-        } finally {
-            nodeAndJsonNodeMapping.clear();
-        }
+        return convertTopRow();
     }
 
-    private void parseTree(JsonNode jsonNode, String prefix){
+    private void parseTree(JsonNode jsonNode, String prefix) {
         if (jsonNode.isArray()) {
             ArrayNode array = (ArrayNode) jsonNode;
             for (int i = 0; i < array.size(); i++) {
@@ -116,15 +100,15 @@ private void parseTree(JsonNode jsonNode, String prefix){
             return;
         }
         Iterator<String> iterator = jsonNode.fieldNames();
-        while (iterator.hasNext()){
+        while (iterator.hasNext()) {
             String next = iterator.next();
             JsonNode child = jsonNode.get(next);
             String nodeKey = getNodeKey(prefix, next);
 
             nodeAndJsonNodeMapping.put(nodeKey, child);
-            if(child.isArray()){
+            if (child.isArray()) {
                 parseTree(child, nodeKey);
-            }else {
+            } else {
                 parseTree(child, nodeKey);
             }
         }
@@ -135,8 +119,8 @@ private JsonNode getIgnoreCase(String key) {
         return nodeAndJsonNodeMapping.get(nodeMappingKey);
     }
 
-    private String getNodeKey(String prefix, String nodeName){
-        if(Strings.isNullOrEmpty(prefix)){
+    private String getNodeKey(String prefix, String nodeName) {
+        if (Strings.isNullOrEmpty(prefix)) {
             return nodeName;
         }
         return prefix + "." + nodeName;
@@ -160,15 +144,19 @@ private Object convert(JsonNode node, TypeInformation<?> info) {
             } else {
                 return node.asText();
             }
-        }  else if (info.getTypeClass().equals(Types.SQL_DATE.getTypeClass())) {
+        } else if (info.getTypeClass().equals(Types.SQL_DATE.getTypeClass())) {
             return Date.valueOf(node.asText());
         } else if (info.getTypeClass().equals(Types.SQL_TIME.getTypeClass())) {
             // local zone
             return Time.valueOf(node.asText());
         } else if (info.getTypeClass().equals(Types.SQL_TIMESTAMP.getTypeClass())) {
             // local zone
             return Timestamp.valueOf(node.asText());
-        }  else {
+        }  else if (info instanceof RowTypeInfo) {
+            return convertRow(node, (RowTypeInfo) info);
+        } else if (info instanceof ObjectArrayTypeInfo) {
+            return convertObjectArray(node, ((ObjectArrayTypeInfo) info).getComponentInfo());
+        } else {
             // for types that were specified without JSON schema
             // e.g. POJOs
             try {
@@ -179,5 +167,55 @@ private Object convert(JsonNode node, TypeInformation<?> info) {
         }
     }
 
+    private Row convertTopRow() {
+        Row row = new Row(fieldNames.length);
+        try {
+            for (int i = 0; i < fieldNames.length; i++) {
+                JsonNode node = getIgnoreCase(fieldNames[i]);
+                AbstractTableInfo.FieldExtraInfo fieldExtraInfo = fieldExtraInfos.get(i);
+
+                if (node == null) {
+                    if (fieldExtraInfo != null && fieldExtraInfo.getNotNull()) {
+                        throw new IllegalStateException("Failed to find field with name '"
+                            + fieldNames[i] + "'.");
+                    } else {
+                        row.setField(i, null);
+                    }
+                } else {
+                    // Read the value as specified type
+                    Object value = convert(node, fieldTypes[i]);
+                    row.setField(i, value);
+                }
+            }
+            return row;
+        } finally {
+            nodeAndJsonNodeMapping.clear();
+        }
+    }
+
+    private Row convertRow(JsonNode node, RowTypeInfo info) {
+        final String[] names = info.getFieldNames();
+        final TypeInformation<?>[] types = info.getFieldTypes();
+
+        final Row row = new Row(names.length);
+        for (int i = 0; i < names.length; i++) {
+            final String name = names[i];
+            final JsonNode subNode = node.get(name);
+            if (subNode == null) {
+                row.setField(i, null);
+            } else {
+                row.setField(i, convert(subNode, types[i]));
+            }
+        }
 
+        return row;
+    }
+
+    private Object convertObjectArray(JsonNode node, TypeInformation<?> elementType) {
+        final Object[] array = (Object[]) Array.newInstance(elementType.getTypeClass(), node.size());
+        for (int i = 0; i < node.size(); i++) {
+            array[i] = convert(node.get(i), elementType);
+        }
+        return array;
+    }
 }

core/src/main/java/com/dtstack/flink/sql/parser/CreateTmpTableParser.java

@@ -22,11 +22,7 @@
 
 import com.dtstack.flink.sql.util.DtStringUtil;
 import org.apache.calcite.config.Lex;
-import org.apache.calcite.sql.SqlBasicCall;
-import org.apache.calcite.sql.SqlJoin;
-import org.apache.calcite.sql.SqlKind;
-import org.apache.calcite.sql.SqlNode;
-import org.apache.calcite.sql.SqlSelect;
+import org.apache.calcite.sql.*;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
 import com.google.common.collect.Lists;
@@ -164,6 +160,10 @@ private static void parseNode(SqlNode sqlNode, CreateTmpTableParser.SqlParserRes
                     parseNode(unionRight, sqlParseResult);
                 }
                 break;
+            case MATCH_RECOGNIZE:
+                SqlMatchRecognize node = (SqlMatchRecognize) sqlNode;
+                sqlParseResult.addSourceTable(node.getTableRef().toString());
+                break;
             default:
                 //do nothing
                 break;

core/src/main/java/com/dtstack/flink/sql/parser/InsertSqlParser.java

@@ -16,7 +16,6 @@
  * limitations under the License.
  */
 
- 
 
 package com.dtstack.flink.sql.parser;
 
@@ -153,14 +152,16 @@ private static void parseNode(SqlNode sqlNode, SqlParseResult sqlParseResult){
 
     /**
      * 将第一层 select 中的 sqlNode 转化为 AsNode，解决字段名冲突问题
+     * 仅对 table.xx 这种类型的字段进行替换
      * @param selectList select Node 的 select 字段
      * @param sqlSelect 第一层解析出来的 selectNode
      */
     private static void rebuildSelectNode(SqlNodeList selectList, SqlSelect sqlSelect) {
         SqlNodeList sqlNodes = new SqlNodeList(selectList.getParserPosition());
 
         for (int index = 0; index < selectList.size(); index++) {
-            if (selectList.get(index).getKind().equals(SqlKind.AS)) {
+            if (selectList.get(index).getKind().equals(SqlKind.AS)
+                    || ((SqlIdentifier) selectList.get(index)).names.size() == 1) {
                 sqlNodes.add(selectList.get(index));
                 continue;
             }