修复redis all cache中取值bug

dapeng · dapeng · commit c15fcacde7f1 · 2020-03-12T12:53:57.000+08:00
diff --git a/redis5/redis5-side/redis-all-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAllReqRow.java b/redis5/redis5-side/redis-all-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAllReqRow.java
@@ -95,7 +95,8 @@ protected void reloadCache() {
     @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){
@@ -104,8 +105,7 @@ 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)){

Original file line number	Diff line number	Diff line change
`@@ -95,7 +95,8 @@ protected void reloadCache() {`
`95`	`95`	`@Override`
`96`	`96`	`public void flatMap(Row row, Collector<Row> out) throws Exception {`
`97`	`97`	`Map<String, String> inputParams = Maps.newHashMap();`
`98`		`- for(Integer conValIndex : sideInfo.getEqualValIndex()){`
	`98`	`+ for (int i = 0; i < sideInfo.getEqualValIndex().size(); i++) {`
	`99`	`+ Integer conValIndex = sideInfo.getEqualValIndex().get(i);`
`99`	`100`	`Object equalObj = row.getField(conValIndex);`
`100`	`101`	`if(equalObj == null){`
`101`	`102`	`if(sideInfo.getJoinType() == JoinType.LEFT){`
`@@ -104,8 +105,7 @@ public void flatMap(Row row, Collector<Row> out) throws Exception {`
`104`	`105`	`}`
`105`	`106`	`return;`
`106`	`107`	`}`
`107`		`- String columnName = sideInfo.getEqualFieldList().get(conValIndex);`
`108`		`- inputParams.put(columnName, equalObj.toString());`
	`108`	`+ inputParams.put(sideInfo.getEqualFieldList().get(i), equalObj.toString());`
`109`	`109`	`}`
`110`	`110`	`String key = buildCacheKey(inputParams);`
`111`	`111`	`if(StringUtils.isBlank(key)){`