Merge branch 'v1.8.0_dev' into feat_1.8_codereview

# Conflicts:
#	core/src/main/java/com/dtstack/flink/sql/util/DateUtil.java

Author: xuchao

.gitlab-ci.yml

@@ -0,0 +1,10 @@
+build:
+  stage: test
+  script:
+    - mvn clean org.jacoco:jacoco-maven-plugin:0.7.8:prepare-agent package -Dmaven.test.failure.ignore=true -q
+    - mvn sonar:sonar -Dsonar.projectKey="dt-insight-engine/flinkStreamSQL" -Dsonar.host.url=http://172.16.100.198:9000 -Dsonar.jdbc.url=jdbc:postgresql://172.16.100.198:5432/sonar -Dsonar.java.binaries=target/sonar
+    - sh ci/sonar_notify.sh
+  only:
+    - 1.8_dev
+  tags:
+    - dt-insight-engine

ci/sonar_notify.sh

@@ -0,0 +1,14 @@
+#!/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" \
+   -H "Content-Type: application/json" \
+   -d "{
+     \"msgtype\": \"markdown\",
+     \"markdown\": {
+         \"title\":\"sonar代码质量\",
+         \"text\": \"## sonar代码质量报告: \n
+> [sonar地址](http://172.16.100.198:9000/dashboard?id=dt-insight-engine/flinkStreamSQL) \n
+> ${sonarreport} \n\"
+     }
+ }"

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

@@ -51,7 +51,11 @@ public void open(Configuration parameters) throws Exception {
                 .put("max_pool_size", rdbSideTableInfo.getAsyncPoolSize())
                 .put("user", rdbSideTableInfo.getUserName())
                 .put("password", rdbSideTableInfo.getPassword())
-                .put("provider_class", DT_PROVIDER_CLASS);
+                .put("provider_class", DT_PROVIDER_CLASS)
+                .put("preferred_test_query", PREFERRED_TEST_QUERY_SQL)
+                .put("idle_connection_test_period", DEFAULT_IDLE_CONNECTION_TEST_PEROID)
+                .put("test_connection_on_checkin", DEFAULT_TEST_CONNECTION_ON_CHECKIN);
+
         System.setProperty("vertx.disableFileCPResolving", "true");
         VertxOptions vo = new VertxOptions();
         vo.setEventLoopPoolSize(rdbSideTableInfo.getAsyncPoolSize());

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

@@ -203,7 +203,6 @@ private static void sqlTranslation(String localSqlPluginPath,
             if (LOG.isInfoEnabled()) {
                 LOG.info("exe-sql:\n" + result.getExecSql());
             }
-
             boolean isSide = false;
             for (String tableName : result.getTargetTableList()) {
                 if (sqlTree.getTmpTableMap().containsKey(tableName)) {
@@ -350,6 +349,7 @@ public static StreamExecutionEnvironment getStreamExeEnv(Properties confProperti
         return env;
     }
 
+
     public static void setLogLevel(ParamsInfo paramsInfo){
         String logLevel = paramsInfo.getConfProp().getProperty(ConfigConstrant.LOG_LEVEL_KEY);
         if(org.apache.commons.lang3.StringUtils.isBlank(logLevel)){

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

@@ -20,8 +20,6 @@
 
 package com.dtstack.flink.sql.side;
 
-import com.dtstack.flink.sql.factory.DTThreadFactory;
-import org.apache.calcite.sql.JoinType;
 import org.apache.flink.api.common.functions.RichFlatMapFunction;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.runtime.types.CRow;
@@ -30,6 +28,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.dtstack.flink.sql.factory.DTThreadFactory;
+import org.apache.calcite.sql.JoinType;
+
 import java.sql.SQLException;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;

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

@@ -23,6 +23,7 @@
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 
 import java.io.Serializable;
+import java.util.Objects;
 
 /**
  * Reason:
@@ -64,4 +65,24 @@ public TypeInformation getTypeInformation() {
     public void setTypeInformation(TypeInformation typeInformation) {
         this.typeInformation = typeInformation;
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        FieldInfo fieldInfo = (FieldInfo) o;
+        return Objects.equals(table, fieldInfo.table) &&
+                Objects.equals(fieldName, fieldInfo.fieldName);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(table, fieldName);
+    }
 }

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

@@ -191,11 +191,7 @@ public JoinInfo dealJoinNode(SqlJoin joinNode, Set<String> sideTableSet,
         tableInfo.setJoinType(joinType);
         tableInfo.setCondition(joinNode.getCondition());
 
-        if(!needBuildTemp){
-            return tableInfo;
-        }
-
-        if(tableInfo.getLeftNode().getKind() != AS){
+        if(tableInfo.getLeftNode().getKind() != AS && needBuildTemp){
             extractTemporaryQuery(tableInfo.getLeftNode(), tableInfo.getLeftTableAlias(), (SqlBasicCall) parentWhere,
                     parentSelectList, queueInfo, joinFieldSet, tableRef);
         }else {

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

@@ -787,9 +787,9 @@ private void joinFun(Object pollObj, Map<String, Table> localTableCache,
             throw new RuntimeException("can't not find side table:" + joinInfo.getRightTableName());
         }
 
-        if(!checkJoinCondition(joinInfo.getCondition(), joinInfo.getRightTableAlias(), sideTableInfo)){
-            throw new RuntimeException("ON condition must contain all equal fields!!!");
-        }
+//        if(!checkJoinCondition(joinInfo.getCondition(), joinInfo.getRightTableAlias(), sideTableInfo)){
+//            throw new RuntimeException("ON condition must contain all equal fields!!!");
+//        }
 
         rightScopeChild.setRowTypeInfo(sideTableInfo.getRowTypeInfo());
 

core/src/main/java/com/dtstack/flink/sql/table/AbstractTableInfoParser.java

@@ -81,7 +81,7 @@ public AbstractTableInfo parseWithTableType(int tableType, CreateTableParser.Sql
                 if(absTableParser == null){
                     String cacheType = MathUtil.getString(props.get(AbstractSideTableInfo.CACHE_KEY));
                     absTableParser = StreamSideFactory.getSqlParser(type, localPluginRoot, cacheType);
-                    sideTableInfoMap.put(type, absTableParser);
+                    sideTableInfoMap.put(type + cacheType, absTableParser);
                 }
             }