Merge branch '1.8_release_3.10.x' into feat_1.8_redisSink

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

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/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/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();

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/BaseAsyncReqRow.java

@@ -22,6 +22,7 @@
 
 import com.dtstack.flink.sql.enums.ECacheContentType;
 import com.dtstack.flink.sql.enums.ECacheType;
+import com.dtstack.flink.sql.factory.DTThreadFactory;
 import com.dtstack.flink.sql.metric.MetricConstant;
 import com.dtstack.flink.sql.side.cache.AbstractSideCache;
 import com.dtstack.flink.sql.side.cache.CacheObj;
@@ -68,6 +69,7 @@ public abstract class BaseAsyncReqRow extends RichAsyncFunction<CRow, CRow> impl
     private int timeOutNum = 0;
     protected BaseSideInfo sideInfo;
     protected transient Counter parseErrorRecords;
+    private transient ThreadPoolExecutor cancelExecutor;
 
     public BaseAsyncReqRow(BaseSideInfo sideInfo){
         this.sideInfo = sideInfo;
@@ -82,6 +84,8 @@ public void open(Configuration parameters) throws Exception {
         super.open(parameters);
         initCache();
         initMetric();
+        cancelExecutor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(100000),
+                new DTThreadFactory("cancel-timer-executor"));
         LOG.info("async dim table config info: {} ", sideInfo.getSideTableInfo().toString());
     }
 
@@ -248,12 +252,11 @@ public void onProcessingTime(long timestamp) throws Exception {
     }
 
     protected void cancelTimerWhenComplete(ResultFuture<CRow> resultFuture, ScheduledFuture<?> timerFuture){
-        ThreadPoolExecutor executors = new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
         if(resultFuture instanceof StreamRecordQueueEntry){
             StreamRecordQueueEntry streamRecordBufferEntry = (StreamRecordQueueEntry) resultFuture;
             streamRecordBufferEntry.onComplete((Object value) -> {
                 timerFuture.cancel(true);
-            },executors);
+            }, cancelExecutor);
         }
     }
 

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/AuthUtil.java

@@ -0,0 +1,112 @@
+/*
+ * 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.apache.commons.io.FileUtils;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ *  Utility methods for helping with security tasks.
+ * Date: 2019/12/28
+ * Company: www.dtstack.com
+ * @author maqi
+ */
+public class AuthUtil {
+
+    public static String creatJaasFile(String prefix, String suffix, JAASConfig jaasConfig) throws IOException {
+        File krbConf = new File(System.getProperty("user.dir"));
+        File temp = File.createTempFile(prefix, suffix, krbConf);
+        temp.deleteOnExit();
+        FileUtils.writeStringToFile(temp, jaasConfig.toString());
+        return temp.getAbsolutePath();
+    }
+
+
+    public static class JAASConfig {
+        private String entryName;
+        private String loginModule;
+        private String loginModuleFlag;
+        private Map<String, String> loginModuleOptions;
+
+        public JAASConfig(String entryName, String loginModule, String loginModuleFlag, Map<String, String> loginModuleOptions) {
+            this.entryName = entryName;
+            this.loginModule = loginModule;
+            this.loginModuleFlag = loginModuleFlag;
+            this.loginModuleOptions = loginModuleOptions;
+        }
+
+        public static Builder builder() {
+            return new Builder();
+        }
+
+        @Override
+        public String toString() {
+            StringBuilder stringBuilder = new StringBuilder(entryName).append(" {\n\t")
+                    .append(loginModule).append("  ").append(loginModuleFlag).append("\n\t");
+            String[] keys = loginModuleOptions.keySet().toArray(new String[loginModuleOptions.size()]);
+            for (int i = 0; i < keys.length; i++) {
+                stringBuilder.append(keys[i]).append("=").append(loginModuleOptions.get(keys[i]));
+                if (i != keys.length - 1) {
+                    stringBuilder.append("\n\t");
+                } else {
+                    stringBuilder.append(";\n");
+                }
+
+            }
+            stringBuilder.append("\n").append("};");
+            return stringBuilder.toString();
+        }
+
+        public static class Builder {
+            private String entryName;
+            private String loginModule;
+            private String loginModuleFlag;
+            private Map<String, String> loginModuleOptions;
+
+            public Builder setEntryName(String entryName) {
+                this.entryName = entryName;
+                return this;
+            }
+
+            public Builder setLoginModule(String loginModule) {
+                this.loginModule = loginModule;
+                return this;
+            }
+
+            public Builder setLoginModuleFlag(String loginModuleFlag) {
+                this.loginModuleFlag = loginModuleFlag;
+                return this;
+            }
+
+            public Builder setLoginModuleOptions(Map<String, String> loginModuleOptions) {
+                this.loginModuleOptions = loginModuleOptions;
+                return this;
+            }
+
+            public JAASConfig build() {
+                return new JAASConfig(
+                        entryName, loginModule, loginModuleFlag, loginModuleOptions);
+            }
+        }
+    }
+}