Merge branch 'v1.5.0_dev_feature_redis' into 'v1.5.0_dev'

修改redis维表内连接问题

修改redis维表内连接问题

See merge request !123

Author: simenliuxing

redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java

@@ -31,6 +31,7 @@
 import io.lettuce.core.api.async.RedisStringAsyncCommands;
 import io.lettuce.core.cluster.RedisClusterClient;
 import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
+import org.apache.calcite.sql.JoinType;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
 import org.apache.flink.calcite.shaded.com.google.common.collect.Maps;
@@ -124,15 +125,11 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
             Integer conValIndex = sideInfo.getEqualValIndex().get(i);
             Object equalObj = input.getField(conValIndex);
 
-            String value = "";
-
             if(equalObj == null){
-                resultFuture.complete(null);
-                value = "null";
-            } else {
-                value = equalObj.toString();
+                dealMissKey(input, resultFuture);
+                return;
             }
-
+            String value = equalObj.toString();
             keyData.add(sideInfo.getEqualFieldList().get(i));
             keyData.add(value);
         }
@@ -165,8 +162,7 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
         List<String> value = async.keys(key + ":*").get();
         String[] values = value.toArray(new String[value.size()]);
         if (values.length == 0){
-            Row row = fillData(input, null);
-            resultFuture.complete(Collections.singleton(row));
+            dealMissKey(input, resultFuture);
         } else {
             RedisFuture<List<KeyValue<String, String>>> future = ((RedisStringAsyncCommands) async).mget(values);
             future.thenAccept(new Consumer<List<KeyValue<String, String>>>() {