[fix][dingo-executor] Fix hash join filters out null values issue

Author: githubgxll

dingo-exec/src/main/java/io/dingodb/exec/operator/HashJoinOperator.java

@@ -61,6 +61,17 @@ public boolean push(Context context, @Nullable Object[] tuple, Vertex vertex) {
             if (pin == 0) { // left
                 waitRightFinFlag(param);
                 TupleKey leftKey = HashJoinParam.rtrimTupleKey(new TupleKey(leftMapping.revMap(tuple)));
+
+                if (HashJoinParam.containsNull(leftKey)) {
+                    if ("inner".equalsIgnoreCase(param.getJoinType()) || "right".equalsIgnoreCase(param.getJoinType())) {
+                        return true;
+                    } else if ("left".equalsIgnoreCase(param.getJoinType()) || "full".equalsIgnoreCase(param.getJoinType())) {
+                        Object[] newTuple = Arrays.copyOf(tuple, leftLength + rightLength);
+                        Arrays.fill(newTuple, leftLength, leftLength + rightLength, null);
+                        return pushToNext(param, edge, context, newTuple);
+                    }
+                }
+
                 boolean isEmpty = isEmpty(leftKey, param);
                 if (isEmpty && ("inner".equalsIgnoreCase(param.getJoinType()) || "right".equalsIgnoreCase(param.getJoinType()))) {
                     return true;
@@ -87,12 +98,23 @@ public boolean push(Context context, @Nullable Object[] tuple, Vertex vertex) {
                 }
             } else if (pin == 1) { //right
                 TupleKey rightKey = HashJoinParam.rtrimTupleKey(new TupleKey(rightMapping.revMap(tuple)));
-                if (isEmpty(rightKey, param) && "inner".equalsIgnoreCase(param.getJoinType())) {
-                    return true;
+                if (HashJoinParam.containsNull(rightKey)) {
+                    if ("inner".equalsIgnoreCase(param.getJoinType()) || "left".equalsIgnoreCase(param.getJoinType())) {
+                        return true;
+                    } else if ("right".equalsIgnoreCase(param.getJoinType()) || "full".equalsIgnoreCase(param.getJoinType())) {
+                        List<TupleWithJoinFlag> list = param.getHashMap()
+                            .computeIfAbsent(rightKey, k -> Collections.synchronizedList(new LinkedList<>()));
+                        list.add(new TupleWithJoinFlag(tuple));
+                    }
+                } else {
+                    if (isEmpty(rightKey, param) && "inner".equalsIgnoreCase(param.getJoinType())) {
+                        return true;
+                    }
+                    List<TupleWithJoinFlag> list = param.getHashMap()
+                        .computeIfAbsent(rightKey, k -> Collections.synchronizedList(new LinkedList<>()));
+                    list.add(new TupleWithJoinFlag(tuple));
                 }
-                List<TupleWithJoinFlag> list = param.getHashMap()
-                    .computeIfAbsent(rightKey, k -> Collections.synchronizedList(new LinkedList<>()));
-                list.add(new TupleWithJoinFlag(tuple));
+
             }
             return true;
         } finally {

dingo-exec/src/main/java/io/dingodb/exec/operator/params/HashJoinParam.java

@@ -144,6 +144,15 @@ public static Object[] rtrimTuple(Object[] tuple) {
         return arrayList.toArray();
     }
 
+    public static boolean containsNull(TupleKey key) {
+        for (Object item : key.getTuple()) {
+            if (item == null) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     @Override
     public void init(Vertex vertex) {
         rightFinFlag = false;