Merge branch '1.8_release_3.10.x' into feat_1.8_3.10.x_mergedDev

# Conflicts:
#	core/src/main/java/com/dtstack/flink/sql/table/AbstractTableInfoParser.java
#	redis5/redis5-side/redis-all-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAllReqRow.java
#	redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java

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

docs/redisSide.md

@@ -4,6 +4,7 @@
  CREATE TABLE tableName(
      colName cloType,
      ...
+     PRIMARY KEY(colName1,colName2) ,
      PERIOD FOR SYSTEM_TIME
   )WITH(
      type ='redis',
@@ -27,9 +28,10 @@
 | tableName | 注册到flink的表名称(可选填;不填默认和redis对应的表名称相同)|
 | colName | 列名称，维表列名格式 表名:主键名:主键值:列名]|
 | colType | 列类型，当前只支持varchar|
+| PRIMARY KEY |主键，多个字段做为联合主键时以逗号分隔
 | PERIOD FOR SYSTEM_TIME | 关键字表明该定义的表为维表信息|
 
-## 3.参数
+## 4.参数
 
 |参数名称|含义|是否必填|默认值|
 |----|---|---|----|
@@ -51,7 +53,7 @@
     * cacheTTLMs:缓存的过期时间(ms)
   * ALL: 缓存全量表数据
 
-## 4.样例
+## 5.样例
 ```
 create table sideTable(
     channel varchar,
@@ -70,5 +72,10 @@ create table sideTable(
  );
 
 ```
+## 6.缓存redis的存储结构规则
+```
+redis使用散列类型 hash 数据结构，key=tableName_primaryKey1_primaryKey2,value={column1=value1, column2=value2}
+如果以班级class表为例，id和name作为联合主键，那么redis的结构为 <class_1_john ,{id=1, name=john, age=12}>
+```
 
 

impala/impala-side/impala-async-side/src/main/java/com/dtstack/flink/sql/side/impala/ImpalaAsyncReqRow.java

@@ -66,7 +66,10 @@ public void open(Configuration parameters) throws Exception {
                 .put("provider_class", DT_PROVIDER_CLASS)
                 .put("idle_connection_test_period", 300)
                 .put("test_connection_on_checkin", DEFAULT_TEST_CONNECTION_ON_CHECKIN)
-                .put("max_idle_time", 600);
+                .put("max_idle_time", 600)
+                .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");
 

mysql/mysql-side/mysql-all-side/src/main/java/com/dtstack/flink/sql/side/mysql/MysqlAllSideInfo.java

@@ -38,4 +38,9 @@ public class MysqlAllSideInfo extends RdbAllSideInfo {
     public MysqlAllSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, AbstractSideTableInfo sideTableInfo) {
         super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
     }
+
+    @Override
+    public String quoteIdentifier(String identifier) {
+        return "`" + identifier + "`";
+    }
 }

mysql/mysql-side/mysql-async-side/src/main/java/com/dtstack/flink/sql/side/mysql/MysqlAsyncSideInfo.java

@@ -39,4 +39,9 @@ public class MysqlAsyncSideInfo extends RdbAsyncSideInfo {
     public MysqlAsyncSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, AbstractSideTableInfo sideTableInfo) {
         super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
     }
+
+    @Override
+    public String quoteIdentifier(String identifier) {
+        return "`" + identifier + "`";
+    }
 }

polardb/polardb-side/polardb-all-side/src/main/java/com/dtstack/flink/sql/side/polardb/PolardbAllSideInfo.java

@@ -34,5 +34,10 @@ public class PolardbAllSideInfo extends RdbAllSideInfo {
     public PolardbAllSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, AbstractSideTableInfo sideTableInfo) {
         super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
     }
+
+    @Override
+    public String quoteIdentifier(String identifier) {
+        return "`" + identifier + "`";
+    }
 }
 

polardb/polardb-side/polardb-async-side/src/main/java/com/dtstack/flink/sql/side/polardb/PolardbAsyncSideInfo.java

@@ -35,4 +35,9 @@ public class PolardbAsyncSideInfo extends RdbAsyncSideInfo {
     public PolardbAsyncSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, AbstractSideTableInfo sideTableInfo) {
         super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
     }
+
+    @Override
+    public String quoteIdentifier(String identifier) {
+        return "`" + identifier + "`";
+    }
 }

postgresql/postgresql-side/postgresql-async-side/src/main/java/com/dtstack/flink/sql/side/postgresql/PostgresqlAsyncReqRow.java

@@ -62,7 +62,10 @@ public void open(Configuration parameters) throws Exception {
                 .put("driver_class", POSTGRESQL_DRIVER)
                 .put("max_pool_size", rdbSideTableInfo.getAsyncPoolSize())
                 .put("user", rdbSideTableInfo.getUserName())
-                .put("password", rdbSideTableInfo.getPassword());
+                .put("password", rdbSideTableInfo.getPassword())
+                .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);
 
         VertxOptions vo = new VertxOptions();
         vo.setEventLoopPoolSize(DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE);