Merge branch '1.8_release_3.10.x' into hotfix_1.8_3.10.x_26972

Author: dapeng

ci/sonar_notify.sh

@@ -1,7 +1,7 @@
 #!/bin/bash
 #参考钉钉文档 https://open-doc.dingtalk.com/microapp/serverapi2/qf2nxq
  sonarreport=$(curl -s http://172.16.100.198:8082/?projectname=dt-insight-engine/flinkStreamSQL)
- curl -s "https://oapi.dingtalk.com/robot/send?access_token=71555061297a53d3ac922a6f4d94285d8e23bccdca0c00b4dc6df0a2d49da724" \
+ curl -s "https://oapi.dingtalk.com/robot/send?access_token=58fd731d8bed3b17708d3aa27e49a7e2c41c7e6545f6c4be3170963a7bba7e2a" \
    -H "Content-Type: application/json" \
    -d "{
      \"msgtype\": \"markdown\",

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

@@ -292,9 +292,7 @@ public static Set<URL> registerTable(SqlTree sqlTree, StreamExecutionEnvironment
 
                 RowTypeInfo typeInfo = new RowTypeInfo(adaptTable.getSchema().getFieldTypes(), adaptTable.getSchema().getFieldNames());
                 DataStream adaptStream = tableEnv.toRetractStream(adaptTable, typeInfo)
-                        .map((Tuple2<Boolean, Row> f0) -> {
-                            return f0.f1;
-                        })
+                        .map((Tuple2<Boolean, Row> f0) -> f0.f1)
                         .returns(typeInfo);
 
                 String fields = String.join(",", typeInfo.getFieldNames());

core/src/main/java/com/dtstack/flink/sql/option/OptionParser.java

@@ -26,6 +26,7 @@
 import org.apache.commons.lang.StringUtils;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Field;
+import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Map;
 import java.io.File;
@@ -102,8 +103,8 @@ public List<String> getProgramExeArgList() throws Exception {
                 continue;
             } else if (OPTION_SQL.equalsIgnoreCase(key)) {
                 File file = new File(value.toString());
-                String content = FileUtils.readFile(file, "UTF-8");
-                value = URLEncoder.encode(content, Charsets.UTF_8.name());
+                String content = FileUtils.readFile(file, StandardCharsets.UTF_8.name());
+                value = URLEncoder.encode(content, StandardCharsets.UTF_8.name());
             }
             args.add("-" + key);
             args.add(value.toString());

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

@@ -41,6 +41,8 @@ public class CreateTableParser implements IParser {
 
     private static final Pattern PATTERN = Pattern.compile(PATTERN_STR);
 
+    private static final Pattern PROP_PATTERN = Pattern.compile("^'\\s*(.+)\\s*'$");
+
     public static CreateTableParser newInstance(){
         return new CreateTableParser();
     }
@@ -69,18 +71,27 @@ public void parseSql(String sql, SqlTree sqlTree) {
     }
 
     private Map parseProp(String propsStr){
-        String[] strs = propsStr.trim().split("'\\s*,");
+        propsStr = propsStr.replaceAll("'\\s*,", "'|");
+        String[] strs = propsStr.trim().split("\\|");
         Map<String, Object> propMap = Maps.newHashMap();
         for(int i=0; i<strs.length; i++){
             List<String> ss = DtStringUtil.splitIgnoreQuota(strs[i], '=');
             String key = ss.get(0).trim();
-            String value = ss.get(1).trim().replaceAll("'", "").trim();
+            String value = extractValue(ss.get(1).trim());
             propMap.put(key, value);
         }
 
         return propMap;
     }
 
+    private String extractValue(String value) {
+        Matcher matcher = PROP_PATTERN.matcher(value);
+        if (matcher.find()) {
+            return matcher.group(1);
+        }
+        throw new RuntimeException("[" + value + "] format is invalid");
+    }
+
     public static class SqlParserResult{
 
         private String tableName;

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

@@ -20,17 +20,18 @@
 
 package com.dtstack.flink.sql.parser;
 
-import org.apache.calcite.config.Lex;
-import org.apache.calcite.sql.SqlBasicCall;
 import org.apache.calcite.sql.SqlInsert;
 import org.apache.calcite.sql.SqlJoin;
 import org.apache.calcite.sql.SqlKind;
-import org.apache.calcite.sql.SqlMatchRecognize;
 import org.apache.calcite.sql.SqlNode;
-import org.apache.calcite.sql.SqlOrderBy;
 import org.apache.calcite.sql.SqlSelect;
-import org.apache.calcite.sql.parser.SqlParseException;
-import org.apache.calcite.sql.parser.SqlParser;
+import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.SqlBasicCall;
+import org.apache.calcite.sql.SqlMatchRecognize;
+import org.apache.calcite.sql.SqlOrderBy;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlAsOperator;
+import org.apache.calcite.sql.parser.SqlParserPos;
 import org.apache.commons.lang3.StringUtils;
 import com.google.common.collect.Lists;
 import org.apache.flink.table.calcite.FlinkPlannerImpl;
@@ -49,6 +50,9 @@
 
 public class InsertSqlParser implements IParser {
 
+    // 用来标识当前解析节点的上一层节点是否为 insert 节点
+    private static Boolean parentIsInsert = false;
+
     @Override
     public boolean verify(String sql) {
         return StringUtils.isNotBlank(sql) && sql.trim().toLowerCase().startsWith("insert");
@@ -78,13 +82,19 @@ private static void parseNode(SqlNode sqlNode, SqlParseResult sqlParseResult){
                 SqlNode sqlTarget = ((SqlInsert)sqlNode).getTargetTable();
                 SqlNode sqlSource = ((SqlInsert)sqlNode).getSource();
                 sqlParseResult.addTargetTable(sqlTarget.toString());
+                parentIsInsert = true;
                 parseNode(sqlSource, sqlParseResult);
                 break;
             case SELECT:
-                SqlNode sqlFrom = ((SqlSelect)sqlNode).getFrom();
-                if(sqlFrom.getKind() == IDENTIFIER){
+                SqlSelect sqlSelect = (SqlSelect) sqlNode;
+                if (parentIsInsert) {
+                    rebuildSelectNode(sqlSelect.getSelectList(), sqlSelect);
+                }
+                SqlNode sqlFrom = ((SqlSelect) sqlNode).getFrom();
+                if (sqlFrom.getKind() == IDENTIFIER) {
                     sqlParseResult.addSourceTable(sqlFrom.toString());
-                }else{
+                } else {
+                    parentIsInsert = false;
                     parseNode(sqlFrom, sqlParseResult);
                 }
                 break;
@@ -141,6 +151,42 @@ private static void parseNode(SqlNode sqlNode, SqlParseResult sqlParseResult){
         }
     }
 
+    /**
+     * 将第一层 select 中的 sqlNode 转化为 AsNode，解决字段名冲突问题
+     * @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)) {
+                sqlNodes.add(selectList.get(index));
+                continue;
+            }
+            sqlNodes.add(transformToAsNode(selectList.get(index)));
+        }
+        sqlSelect.setSelectList(sqlNodes);
+    }
+
+    /**
+     * 将 sqlNode 转化为 AsNode
+     * @param sqlNode 需要转化的 sqlNode
+     * @return 重新构造的 AsNode
+     */
+    public static SqlBasicCall transformToAsNode(SqlNode sqlNode) {
+        String asName = "";
+        SqlParserPos pos = new SqlParserPos(sqlNode.getParserPosition().getLineNum(),
+                                            sqlNode.getParserPosition().getEndColumnNum());
+        if (sqlNode.getKind().equals(SqlKind.IDENTIFIER)) {
+            asName = ((SqlIdentifier) sqlNode).names.get(1);
+        }
+        SqlNode[] operands = new SqlNode[2];
+        operands[0] = sqlNode;
+        operands[1] = new SqlIdentifier(asName, null, pos);
+        return new SqlBasicCall(new SqlAsOperator(), operands, pos);
+    }
+
     public static class SqlParseResult {
 
         private List<String> sourceTableList = Lists.newArrayList();

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

@@ -68,7 +68,7 @@ public static SqlTree parseSql(String sql) throws Exception {
             throw new RuntimeException("need to set local sql plugin root");
         }
 
-        sql = sql.replaceAll("--.*", "")
+        sql = DtStringUtil.dealSqlComment(sql)
                 .replaceAll("\r\n", " ")
                 .replaceAll("\n", " ")
                 .replace("\t", " ").trim();

core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java

@@ -115,7 +115,11 @@ public void exec(String sql,
         SideSQLParser sideSQLParser = new SideSQLParser();
         sideSQLParser.setLocalTableCache(localTableCache);
         Queue<Object> exeQueue = sideSQLParser.getExeQueue(sql, sideTableMap.keySet(), scope);
-        Object pollObj = null;
+        Object pollObj;
+
+        //need clean
+        boolean preIsSideJoin = false;
+        List<FieldReplaceInfo> replaceInfoList = Lists.newArrayList();
 
         while((pollObj = exeQueue.poll()) != null){
 

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

@@ -106,6 +106,57 @@ public static String replaceIgnoreQuota(String str, String oriStr, String replac
         return str.replaceAll(splitPatternStr, replaceStr);
     }
 
+    /**
+     * 处理 sql 中 "--" 注释，而不删除引号内的内容
+     *
+     * @param sql 解析出来的 sql
+     * @return 返回无注释内容的 sql
+     */
+    public static String dealSqlComment(String sql) {
+        boolean inQuotes = false;
+        boolean inSingleQuotes = false;
+        int bracketLeftNum = 0;
+        StringBuilder b = new StringBuilder(sql.length());
+        char[] chars = sql.toCharArray();
+        for (int index = 0; index < chars.length; index ++) {
+            if (index == chars.length) {
+                return b.toString();
+            }
+            StringBuilder tempSb = new StringBuilder(2);
+            if (index > 1) {
+                tempSb.append(chars[index - 1]);
+                tempSb.append(chars[index]);
+            }
+
+            if (tempSb.toString().equals("--")) {
+                if (inQuotes) {
+                    b.append(chars[index]);
+                } else if (inSingleQuotes) {
+                    b.append(chars[index]);
+                } else if (bracketLeftNum > 0) {
+                    b.append(chars[index]);
+                } else {
+                    b.deleteCharAt(b.length() - 1);
+                    while (chars[index] != '\n') {
+                        // 判断注释内容是不是行尾或者 sql 的最后一行
+                        if (index == chars.length - 1) {
+                            break;
+                        }
+                        index++;
+                    }
+                }
+            } else if (chars[index] == '\"' && '\\' != chars[index] && !inSingleQuotes) {
+                inQuotes = !inQuotes;
+                b.append(chars[index]);
+            } else if (chars[index] == '\'' && '\\' != chars[index] && !inQuotes) {
+                inSingleQuotes = !inSingleQuotes;
+                b.append(chars[index]);
+            } else {
+                b.append(chars[index]);
+            }
+        }
+        return b.toString();
+    }
 
     public static String col2string(Object column, String type) {
         String rowData = column.toString();

core/src/main/java/com/dtstack/flink/sql/watermarker/WaterMarkerAssigner.java

@@ -39,11 +39,7 @@
 public class WaterMarkerAssigner {
 
     public boolean checkNeedAssignWaterMarker(AbstractSourceTableInfo tableInfo){
-        if(Strings.isNullOrEmpty(tableInfo.getEventTimeField())){
-            return false;
-        }
-
-        return true;
+        return !Strings.isNullOrEmpty(tableInfo.getEventTimeField());
     }
 
     public DataStream assignWaterMarker(DataStream<Row> dataStream, RowTypeInfo typeInfo, AbstractSourceTableInfo sourceTableInfo){

core/src/test/java/com/dtstack/flink/sql/util/TestDtStringUtil.java

@@ -0,0 +1,38 @@
+/*
+ * 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 org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * @author tiezhu
+ * Date 2020/6/17 星期三
+ */
+public class TestDtStringUtil {
+    @Test
+    public void dealSqlCommentTest() {
+        String testSQLWithComment = "CREATE TABLE MyTable --this is a comment\n";
+        Assert.assertEquals("CREATE TABLE MyTable ", DtStringUtil.dealSqlComment(testSQLWithComment));
+        testSQLWithComment = "CREATE TABLE 'MyTable--' --this is  a comment";
+        Assert.assertEquals("CREATE TABLE 'MyTable--' ", DtStringUtil.dealSqlComment(testSQLWithComment));
+        testSQLWithComment = "CREATE TABLE MyTable -- this is a '--comment'\n";
+        Assert.assertEquals("CREATE TABLE MyTable ", DtStringUtil.dealSqlComment(testSQLWithComment));
+    }
+}