Merge branch 'feat_1.8_3.10.x_mergedDev' into 'v1.8.0_dev'

3.10.x merged dev

See merge request dt-insight-engine/flinkStreamSQL!4

Author: todd5167

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

@@ -206,7 +206,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)) {
@@ -233,7 +232,6 @@ private static void sqlTranslation(String localSqlPluginPath,
                         System.out.println(result.getExecSql());
                         FlinkSQLExec.sqlUpdate(tableEnv, result.getExecSql(), queryConfig);
                         if (LOG.isInfoEnabled()) {
-                            System.out.println();
                             LOG.info("exec sql: " + result.getExecSql());
                         }
                     }
@@ -355,6 +353,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,14 +20,15 @@
 
 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;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.Collector;
 
+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

@@ -190,11 +190,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

@@ -793,9 +793,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);
                 }
             }