Merge branch 'v1.8.0_dev_bugfix_leftjoinnull' into 'v1.8.0_dev'

left join null nosql



See merge request !135

Author: simenliuxing

cassandra/cassandra-side/cassandra-all-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAllReqRow.java

@@ -134,7 +134,11 @@ public void flatMap(Row value, Collector<Row> out) throws Exception {
         for (Integer conValIndex : sideInfo.getEqualValIndex()) {
             Object equalObj = value.getField(conValIndex);
             if (equalObj == null) {
-                out.collect(null);
+                if(sideInfo.getJoinType() == JoinType.LEFT){
+                    Row data = fillData(value, null);
+                    out.collect(data);
+                }
+                return;
             }
 
             inputParams.add(equalObj);

cassandra/cassandra-side/cassandra-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncReqRow.java

@@ -170,7 +170,8 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
             Integer conValIndex = sideInfo.getEqualValIndex().get(i);
             Object equalObj = input.getField(conValIndex);
             if (equalObj == null) {
-                resultFuture.complete(null);
+                dealMissKey(input, resultFuture);
+                return;
             }
             inputParams.add(equalObj);
             stringBuffer.append(sideInfo.getEqualFieldList().get(i))

hbase/hbase-side/hbase-all-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAllReqRow.java

@@ -22,6 +22,7 @@
 
 import com.dtstack.flink.sql.side.*;
 import com.dtstack.flink.sql.side.hbase.table.HbaseSideTableInfo;
+import org.apache.calcite.sql.JoinType;
 import org.apache.commons.collections.map.HashedMap;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import com.google.common.collect.Maps;
@@ -119,7 +120,11 @@ public void flatMap(Row value, Collector<Row> out) throws Exception {
             Integer conValIndex = sideInfo.getEqualValIndex().get(i);
             Object equalObj = value.getField(conValIndex);
             if(equalObj == null){
-                out.collect(null);
+                if(sideInfo.getJoinType() == JoinType.LEFT){
+                    Row data = fillData(value, null);
+                    out.collect(data);
+                }
+                return;
             }
             refData.put(sideInfo.getEqualFieldList().get(i), equalObj);
         }

hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAsyncReqRow.java

@@ -128,7 +128,7 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
             Integer conValIndex = sideInfo.getEqualValIndex().get(i);
             Object equalObj = input.getField(conValIndex);
             if(equalObj == null){
-                resultFuture.complete(null);
+                dealMissKey(input, resultFuture);
                 return;
             }
 

mongo/mongo-side/mongo-all-side/src/main/java/com/dtstack/flink/sql/side/mongo/MongoAllReqRow.java

@@ -132,7 +132,11 @@ public void flatMap(Row value, Collector<Row> out) throws Exception {
         for (Integer conValIndex : sideInfo.getEqualValIndex()) {
             Object equalObj = value.getField(conValIndex);
             if (equalObj == null) {
-                out.collect(null);
+                if(sideInfo.getJoinType() == JoinType.LEFT){
+                    Row data = fillData(value, null);
+                    out.collect(data);
+                }
+                return;
             }
 
             inputParams.add(equalObj);

mongo/mongo-side/mongo-async-side/src/main/java/com/dtstack/flink/sql/side/mongo/MongoAsyncReqRow.java

@@ -129,7 +129,7 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
             Integer conValIndex = sideInfo.getEqualValIndex().get(i);
             Object equalObj = input.getField(conValIndex);
             if (equalObj == null) {
-                resultFuture.complete(null);
+                dealMissKey(input, resultFuture);
                 return;
             }
             basicDBObject.put(sideInfo.getEqualFieldList().get(i), equalObj);

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

@@ -93,7 +93,11 @@ public void flatMap(Row row, Collector<Row> out) throws Exception {
         for(Integer conValIndex : sideInfo.getEqualValIndex()){
             Object equalObj = row.getField(conValIndex);
             if(equalObj == null){
-                out.collect(null);
+                if(sideInfo.getJoinType() == JoinType.LEFT){
+                    Row data = fillData(row, null);
+                    out.collect(data);
+                }
+                return;
             }
             String columnName = sideInfo.getEqualFieldList().get(conValIndex);
             inputParams.put(columnName, equalObj.toString());