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

# Conflicts:
#	rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/all/AbstractRdbAllReqRow.java

Author: dapeng

.gitignore

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

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\",

clickhouse/clickhouse-side/clickhouse-async-side/src/main/java/com/dtstack/flink/sql/side/clickhouse/ClickhouseAsyncReqRow.java

@@ -67,8 +67,6 @@ public void open(Configuration parameters) throws Exception {
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSqlClient(JDBCClient.createNonShared(vertx, clickhouseClientConfig));
-        setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,
-                new LinkedBlockingQueue<>(10000), new DTThreadFactory("clickhouseAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));
     }
 
 }

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

@@ -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/option/Options.java

@@ -72,6 +72,10 @@ public class Options {
     @OptionRequired(description = "log level")
     private String logLevel = "info";
 
+    @OptionRequired(description = "file add to ship file")
+    private  String addShipfile;
+
+
     public String getMode() {
         return mode;
     }
@@ -183,4 +187,13 @@ public String getLogLevel() {
 	public void setLogLevel(String logLevel) {
 		this.logLevel = logLevel;
 	}
+
+    public String getAddShipfile() {
+        return addShipfile;
+    }
+
+    public void setAddShipfile(String addShipfile) {
+        this.addShipfile = addShipfile;
+    }
+
 }

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/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,21 +16,21 @@
  * limitations under the License.
  */
 
- 
 
 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 +49,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 +81,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 +150,44 @@ 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)
+                    || ((SqlIdentifier) selectList.get(index)).names.size() == 1) {
+                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();