[fix][dingo-exec] Resolve the issue of null values in join connection columns

Author: guojn1

dingo-calcite/src/main/java/io/dingodb/calcite/visitor/function/DingoHashJoinVisitFun.java

@@ -112,6 +112,7 @@ public static List<Vertex> visit(
                 rel.getJoinType() == JoinRelType.LEFT || rel.getJoinType() == JoinRelType.FULL,
                 rel.getJoinType() == JoinRelType.RIGHT || rel.getJoinType() == JoinRelType.FULL
             );
+            param.setJoinType(rel.getJoinType().lowerName);
             param.setOtherExpr(otherCondition);
             Vertex vertex = new Vertex(HASH_JOIN, param);
             vertex.setId(idGenerator.getOperatorId(taskId));

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

@@ -60,6 +60,15 @@ public boolean push(Context context, @Nullable Object[] tuple, Vertex vertex) {
             if (pin == 0) { // left
                 waitRightFinFlag(param);
                 TupleKey leftKey = new TupleKey(leftMapping.revMap(tuple));
+                boolean isEmpty = isEmpty(leftKey, param);
+                if (isEmpty && ("inner".equalsIgnoreCase(param.getJoinType()) || "right".equalsIgnoreCase(param.getJoinType()))) {
+                    return true;
+                }
+                if (isEmpty && "left".equalsIgnoreCase(param.getJoinType())) {
+                    Object[] newTuple = Arrays.copyOf(tuple, leftLength + rightLength);
+                    Arrays.fill(newTuple, leftLength, leftLength + rightLength, null);
+                    return pushToNext(param, edge, context, newTuple);
+                }
                 List<TupleWithJoinFlag> rightList = param.getHashMap().get(leftKey);
                 if (rightList != null) {
                     for (TupleWithJoinFlag t : rightList) {
@@ -77,6 +86,9 @@ public boolean push(Context context, @Nullable Object[] tuple, Vertex vertex) {
                 }
             } else if (pin == 1) { //right
                 TupleKey rightKey = new TupleKey(rightMapping.revMap(tuple));
+                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));
@@ -163,4 +175,19 @@ private static boolean pushToNext(HashJoinParam param, Edge edge, Context contex
             return edge.transformToNext(context, tuple);
         }
     }
+
+    public static boolean isEmpty(TupleKey tupleKey, HashJoinParam param) {
+        if (param.rightMappingEmpty && param.leftMappingEmpty) {
+            return false;
+        }
+        Object[] tuple = tupleKey.getTuple();
+        if (tuple != null) {
+            for (Object item : tuple) {
+                if (item != null) {
+                    return false;
+                }
+            }
+        }
+        return true;
+    }
 }

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

@@ -57,17 +57,20 @@ public class HashJoinParam extends AbstractParams {
     @Setter
     private transient CompletableFuture<Void> future;
 
-    @Getter
     @Setter
     public Profile profileLeft;
-    @Getter
     @Setter
     public Profile profileRight;
 
-    @Getter
     @Setter
     public SqlExpr otherExpr;
 
+    @Setter
+    public String joinType;
+
+    public boolean leftMappingEmpty;
+    public boolean rightMappingEmpty;
+
 
     public HashJoinParam(
         TupleMapping leftMapping,
@@ -83,6 +86,8 @@ public HashJoinParam(
         this.rightLength = rightLength;
         this.leftRequired = leftRequired;
         this.rightRequired = rightRequired;
+        this.leftMappingEmpty = this.leftMapping.size() == 0;
+        this.rightMappingEmpty = this.rightMapping.size() == 0;
     }
 
     @Override

dingo-executor/src/main/java/io/dingodb/server/executor/prepare/PrepareMeta.java

@@ -629,14 +629,19 @@ public static void initTableByTemplate(String schema,
     }
 
     public static void synchronizeTenant() {
-        List<Object> tenantObjList = io.dingodb.meta.InfoSchemaService.root().listTenant();
-        tenantObjList.forEach(object -> {
-            Tenant tenant = (Tenant) object;
-            if (!MetaService.ROOT.existsTenant(tenant.getId())) {
-                MetaService.ROOT.createTenant(tenant);
-                LogUtils.info(log, "synchronize tenant id to coordinator:{}", tenant.getId());
-            }
-        });
+        try {
+            List<Object> tenantObjList = io.dingodb.meta.InfoSchemaService.root().listTenant();
+            tenantObjList.forEach(object -> {
+                Tenant tenant = (Tenant) object;
+                if (!MetaService.ROOT.existsTenant(tenant.getId())) {
+                    MetaService.ROOT.createTenant(tenant);
+                    LogUtils.info(log, "synchronize tenant id to coordinator:{}", tenant.getId());
+                }
+            });
+            LogUtils.info(log, "synchronizeTenant done");
+        } catch (Exception e) {
+            LogUtils.error(log, e.getMessage(), e);
+        }
     }
 
     private static boolean continueRetry() {