Merge branch 'hotfix_1.8_3.9.x_23318' into 1.8_release_3.9.x

Author: gituser

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}>
+```
 
 

redis5/redis5-side/redis-all-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAllReqRow.java

@@ -19,10 +19,14 @@
 package com.dtstack.flink.sql.side.redis;
 
 import com.dtstack.flink.sql.side.*;
+import com.dtstack.flink.sql.side.redis.enums.RedisType;
 import com.dtstack.flink.sql.side.redis.table.RedisSideReqRow;
 import com.dtstack.flink.sql.side.redis.table.RedisSideTableInfo;
 import com.esotericsoftware.minlog.Log;
 import org.apache.calcite.sql.JoinType;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang.StringUtils;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import com.google.common.collect.Maps;
@@ -85,13 +89,14 @@ protected void reloadCache() {
         }
 
         cacheRef.set(newCache);
-        LOG.info("----- Redis all cacheRef reload end:{}", Calendar.getInstance());
+        LOG.info("----- Redis all cacheRef reload end:{}", newCache.size());
     }
 
     @Override
     public void flatMap(Row row, Collector<Row> out) throws Exception {
         Map<String, String> inputParams = Maps.newHashMap();
-        for(Integer conValIndex : sideInfo.getEqualValIndex()){
+        for (int i = 0; i < sideInfo.getEqualValIndex().size(); i++) {
+            Integer conValIndex = sideInfo.getEqualValIndex().get(i);
             Object equalObj = row.getField(conValIndex);
             if(equalObj == null){
                 if(sideInfo.getJoinType() == JoinType.LEFT){
@@ -100,107 +105,55 @@ public void flatMap(Row row, Collector<Row> out) throws Exception {
                 }
                 return;
             }
-            String columnName = sideInfo.getEqualFieldList().get(conValIndex);
-            inputParams.put(columnName, equalObj.toString());
+            inputParams.put(sideInfo.getEqualFieldList().get(i), equalObj.toString());
+        }
+        String key = buildCacheKey(inputParams);
+        if(StringUtils.isBlank(key)){
+            return;
         }
-        String key = buildKey(inputParams);
 
         Map<String, String> cacheMap = cacheRef.get().get(key);
-
-        if (cacheMap == null){
-            if(sideInfo.getJoinType() == JoinType.LEFT){
-                Row data = fillData(row, null);
-                out.collect(data);
-            }else{
+        if(MapUtils.isEmpty(cacheMap)){
+            if(sideInfo.getJoinType() != JoinType.LEFT){
                 return;
             }
-
+            Row data = fillData(row, null);
+            out.collect(data);
             return;
-        }
 
+        }
         Row newRow = fillData(row, cacheMap);
         out.collect(newRow);
     }
 
-    private String buildKey(Map<String, String> inputParams) {
-        String tableName = tableInfo.getTableName();
-        StringBuilder key = new StringBuilder();
-        for (int i=0; i<inputParams.size(); i++){
-            key.append(tableName).append(":").append(inputParams.keySet().toArray()[i]).append(":")
-                    .append(inputParams.get(inputParams.keySet().toArray()[i]));
+    private String buildCacheKey(Map<String, String> refData) {
+        StringBuilder keyBuilder = new StringBuilder(tableInfo.getTableName());
+        List<String> primaryKeys = tableInfo.getPrimaryKeys();
+        for(String primaryKey : primaryKeys){
+            if(!refData.containsKey(primaryKey)){
+                return null;
+            }
+            keyBuilder.append("_").append(refData.get(primaryKey));
         }
-        return key.toString();
+        return keyBuilder.toString();
     }
 
+
     private void loadData(Map<String, Map<String, String>> tmpCache) throws SQLException {
         JedisCommands jedis = null;
-
         try {
-            for(int i=0; i<CONN_RETRY_NUM; i++){
-
-                try{
-                    jedis = getJedis(tableInfo);
-                    break;
-                }catch (Exception e){
-                    if(i == CONN_RETRY_NUM - 1){
-                        throw new RuntimeException("", e);
-                    }
-
-                    try {
-                        String jedisInfo = "url:" + tableInfo.getUrl() + ",pwd:" + tableInfo.getPassword() + ",database:" + tableInfo.getDatabase();
-                        LOG.warn("get conn fail, wait for 5 sec and try again, connInfo:" + jedisInfo);
-                        Thread.sleep(5 * 1000);
-                    } catch (InterruptedException e1) {
-                        LOG.error("", e1);
-                    }
-                }
+            StringBuilder keyPattern = new StringBuilder(tableInfo.getTableName());
+            for(String key : tableInfo.getPrimaryKeys()){
+                keyPattern.append("_").append("*");
+            };
+            jedis = getJedisWithRetry(CONN_RETRY_NUM);
+            Set<String> keys = getRedisKeys(RedisType.parse(tableInfo.getRedisType()), jedis, keyPattern.toString());
+            if(CollectionUtils.isEmpty(keys)){
+                return;
             }
-
-            if (tableInfo.getRedisType() != 3){
-                String perKey = tableInfo.getTableName() + "*";
-                Set<String> keys = ((Jedis) jedis).keys(perKey);
-                List<String> newPerKeys = new LinkedList<>();
-                for (String key : keys){
-                    String[] splitKey = key.split(":");
-                    String newKey = splitKey[0] + ":" + splitKey[1] + ":" + splitKey[2];
-                    newPerKeys.add(newKey);
-                }
-                List<String> list = newPerKeys.stream().distinct().collect(Collectors.toList());
-                for(String key : list){
-                    Map<String, String> kv = Maps.newHashMap();
-                    String[] primaryKv = key.split(":");
-                    kv.put(primaryKv[1], primaryKv[2]);
-                    String pattern = key + "*";
-                    Set<String> realKeys = ((Jedis) jedis).keys(pattern);
-                    for (String realKey : realKeys){
-                        kv.put(realKey.split(":")[3], jedis.get(realKey));
-                    }
-                    tmpCache.put(key, kv);
-                }
-            } else {
-                String perKey = tableInfo.getTableName() + "*";
-                Set<String> keys = keys((JedisCluster) jedis, perKey);
-                List<String> newPerKeys = new LinkedList<>();
-                for (String key : keys){
-                    String[] splitKey = key.split(":");
-                    String newKey = splitKey[0] + ":" + splitKey[1] + ":" + splitKey[2];
-                    newPerKeys.add(newKey);
-                }
-                List<String> list = newPerKeys.stream().distinct().collect(Collectors.toList());
-                for(String key : list){
-                    Map<String, String> kv = Maps.newHashMap();
-                    String[] primaryKv = key.split(":");
-                    kv.put(primaryKv[1], primaryKv[2]);
-                    String pattern = key + "*";
-                    Set<String> realKeys = keys((JedisCluster) jedis, pattern);
-                    for (String realKey : realKeys){
-                        kv.put(realKey.split(":")[3], jedis.get(realKey));
-                    }
-                    tmpCache.put(key, kv);
-                }
+            for(String key : keys){
+                tmpCache.put(key, jedis.hgetAll(key));
             }
-
-
         } catch (Exception e){
             LOG.error("", e);
         } finally {
@@ -223,7 +176,7 @@ private void loadData(Map<String, Map<String, String>> tmpCache) throws SQLExcep
     private JedisCommands getJedis(RedisSideTableInfo tableInfo) {
         String url = tableInfo.getUrl();
         String password = tableInfo.getPassword();
-        String database = tableInfo.getDatabase();
+        String database = tableInfo.getDatabase() == null ? "0" : tableInfo.getDatabase();
         int timeout = tableInfo.getTimeout();
         if (timeout == 0){
             timeout = 1000;
@@ -245,33 +198,58 @@ private JedisCommands getJedis(RedisSideTableInfo tableInfo) {
         }
         JedisCommands jedis = null;
         GenericObjectPoolConfig poolConfig = setPoolConfig(tableInfo.getMaxTotal(), tableInfo.getMaxIdle(), tableInfo.getMinIdle());
-        switch (tableInfo.getRedisType()){
+        switch (RedisType.parse(tableInfo.getRedisType())){
             //单机
-            case 1:
+            case STANDALONE:
                 pool = new JedisPool(poolConfig, firstIp, Integer.parseInt(firstPort), timeout, password, Integer.parseInt(database));
                 jedis = pool.getResource();
                 break;
             //哨兵
-            case 2:
+            case SENTINEL:
                 jedisSentinelPool = new JedisSentinelPool(tableInfo.getMasterName(), ipPorts, poolConfig, timeout, password, Integer.parseInt(database));
                 jedis = jedisSentinelPool.getResource();
                 break;
             //集群
-            case 3:
+            case CLUSTER:
                 jedis = new JedisCluster(addresses, timeout, timeout,1, poolConfig);
+            default:
+                break;
         }
 
         return jedis;
     }
 
-    private Set<String> keys(JedisCluster jedisCluster, String pattern){
+    private JedisCommands getJedisWithRetry(int retryNum) {
+        while (retryNum-- > 0){
+            try {
+                return getJedis(tableInfo);
+            } catch (Exception e) {
+                if(retryNum <= 0){
+                    throw new RuntimeException("getJedisWithRetry error", e);
+                }
+                try {
+                    String jedisInfo = "url:" + tableInfo.getUrl() + ",pwd:" + tableInfo.getPassword() + ",database:" + tableInfo.getDatabase();
+                    LOG.warn("get conn fail, wait for 5 sec and try again, connInfo:" + jedisInfo);
+                    Thread.sleep(5 * 1000);
+                } catch (InterruptedException e1) {
+                    LOG.error("", e1);
+                }
+            }
+        }
+        return null;
+    }
+
+    private Set<String> getRedisKeys(RedisType redisType, JedisCommands jedis, String keyPattern){
+        if(!redisType.equals(RedisType.CLUSTER)){
+            return ((Jedis) jedis).keys(keyPattern);
+        }
         Set<String> keys = new TreeSet<>();
-        Map<String, JedisPool> clusterNodes = jedisCluster.getClusterNodes();
+        Map<String, JedisPool> clusterNodes = ((JedisCluster)jedis).getClusterNodes();
         for(String k : clusterNodes.keySet()){
             JedisPool jp = clusterNodes.get(k);
             Jedis connection = jp.getResource();
             try {
-                keys.addAll(connection.keys(pattern));
+                keys.addAll(connection.keys(keyPattern));
             } catch (Exception e){
                 LOG.error("Getting keys error: {}", e);
             } finally {