From bc898e3cdc305366e4043480280ba3a0eb2811ca Mon Sep 17 00:00:00 2001
From: Gabriel <gabrielleebuaa@gmail.com>
Date: Sun, 3 Mar 2024 19:21:11 +0800
Subject: [PATCH] [pipelineX](bug) Fix incorrect join operator judgement
 (#31690)

* [pipelineX](bug) Fix incorrect join operator judgement

* update
---
 .../translator/PhysicalPlanTranslator.java    |  3 ---
 .../translator/PlanTranslatorContext.java     |  2 --
 .../apache/doris/planner/HashJoinNode.java    | 24 +++++++++++++++----
 .../doris/planner/MultiCastPlanFragment.java  |  1 -
 .../apache/doris/planner/PlanFragment.java    | 10 ++------
 .../org/apache/doris/planner/PlanNode.java    |  4 ++++
 .../org/apache/doris/planner/ScanNode.java    |  2 +-
 7 files changed, 27 insertions(+), 19 deletions(-)

diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
index c64965080f..1be1a6c763 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
@@ -1205,9 +1205,6 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
         hashJoinNode.setDistributeExprLists(distributeExprLists);
         PlanFragment currentFragment = connectJoinNode(hashJoinNode, leftFragment, rightFragment, context, hashJoin);
 
-        if (joinType == JoinType.NULL_AWARE_LEFT_ANTI_JOIN) {
-            currentFragment.setHasNullAwareLeftAntiJoin(true);
-        }
         if (JoinUtils.shouldColocateJoin(physicalHashJoin)) {
             // TODO: add reason
             hashJoinNode.setColocate(true, "");
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java
index df314d7f37..8a723b1fd1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java
@@ -221,8 +221,6 @@ public class PlanTranslatorContext {
         srcFragment.getBuilderRuntimeFilterIds().forEach(targetFragment::setBuilderRuntimeFilterIds);
         targetFragment.setHasColocatePlanNode(targetFragment.hasColocatePlanNode()
                 || srcFragment.hasColocatePlanNode());
-        targetFragment.setHasNullAwareLeftAntiJoin(targetFragment.isHasNullAwareLeftAntiJoin()
-                || srcFragment.isHasNullAwareLeftAntiJoin());
         this.planFragments.remove(srcFragment);
     }
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/HashJoinNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/HashJoinNode.java
index f710a3c0a2..0f099bdca3 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/HashJoinNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/HashJoinNode.java
@@ -784,10 +784,8 @@ public class HashJoinNode extends JoinNodeBase {
             if (eqJoinConjuncts.isEmpty()) {
                 Preconditions.checkState(joinOp == JoinOperator.LEFT_SEMI_JOIN
                         || joinOp == JoinOperator.LEFT_ANTI_JOIN);
-                if (joinOp == JoinOperator.LEFT_SEMI_JOIN) {
-                    msg.hash_join_node.join_op = JoinOperator.NULL_AWARE_LEFT_SEMI_JOIN.toThrift();
-                } else if (joinOp == JoinOperator.LEFT_ANTI_JOIN) {
-                    msg.hash_join_node.join_op = JoinOperator.NULL_AWARE_LEFT_ANTI_JOIN.toThrift();
+                if (joinOp == JoinOperator.LEFT_SEMI_JOIN || joinOp == JoinOperator.LEFT_ANTI_JOIN) {
+                    msg.hash_join_node.join_op = transformJoinOperator().toThrift();
                 }
                 // because eqJoinConjuncts mustn't be empty in thrift
                 // we have to use markJoinConjuncts instead
@@ -970,4 +968,22 @@ public class HashJoinNode extends JoinNodeBase {
             return slotRef;
         }
     }
+
+    private JoinOperator transformJoinOperator() {
+        boolean transformToNullAware = markJoinConjuncts != null && eqJoinConjuncts.isEmpty();
+        if (joinOp == JoinOperator.LEFT_ANTI_JOIN && transformToNullAware) {
+            return JoinOperator.NULL_AWARE_LEFT_ANTI_JOIN;
+        } else if (joinOp == JoinOperator.LEFT_SEMI_JOIN && transformToNullAware) {
+            return JoinOperator.NULL_AWARE_LEFT_SEMI_JOIN;
+        }
+        return joinOp;
+    }
+
+    @Override
+    public boolean isNullAwareLeftAntiJoin() {
+        if (transformJoinOperator() == JoinOperator.NULL_AWARE_LEFT_ANTI_JOIN) {
+            return true;
+        }
+        return children.stream().anyMatch(PlanNode::isNullAwareLeftAntiJoin);
+    }
 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/MultiCastPlanFragment.java b/fe/fe-core/src/main/java/org/apache/doris/planner/MultiCastPlanFragment.java
index 4b2258c8a2..9c54bdb406 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/MultiCastPlanFragment.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/MultiCastPlanFragment.java
@@ -32,7 +32,6 @@ public class MultiCastPlanFragment extends PlanFragment {
         super(planFragment.getFragmentId(), planFragment.getPlanRoot(), planFragment.getDataPartition(),
                 planFragment.getBuilderRuntimeFilterIds(), planFragment.getTargetRuntimeFilterIds());
         this.hasColocatePlanNode = planFragment.hasColocatePlanNode;
-        this.hasNullAwareLeftAntiJoin = planFragment.hasNullAwareLeftAntiJoin;
         this.outputPartition = DataPartition.RANDOM;
         this.children.addAll(planFragment.getChildren());
     }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanFragment.java b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanFragment.java
index f24c16561a..2469d087cd 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanFragment.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanFragment.java
@@ -149,8 +149,6 @@ public class PlanFragment extends TreeNode<PlanFragment> {
     // has colocate plan node
     protected boolean hasColocatePlanNode = false;
 
-    protected boolean hasNullAwareLeftAntiJoin = false;
-
     private TResultSinkType resultSinkType = TResultSinkType.MYSQL_PROTOCAL;
 
     /**
@@ -473,11 +471,7 @@ public class PlanFragment extends TreeNode<PlanFragment> {
         this.bucketNum = bucketNum;
     }
 
-    public boolean isHasNullAwareLeftAntiJoin() {
-        return hasNullAwareLeftAntiJoin;
-    }
-
-    public void setHasNullAwareLeftAntiJoin(boolean hasNullAwareLeftAntiJoin) {
-        this.hasNullAwareLeftAntiJoin = hasNullAwareLeftAntiJoin;
+    public boolean hasNullAwareLeftAntiJoin() {
+        return planRoot.isNullAwareLeftAntiJoin();
     }
 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
index f450758834..a88925f929 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
@@ -261,6 +261,10 @@ public abstract class PlanNode extends TreeNode<PlanNode> implements PlanStats {
         this.fragment = fragment;
     }
 
+    public boolean isNullAwareLeftAntiJoin() {
+        return children.stream().anyMatch(PlanNode::isNullAwareLeftAntiJoin);
+    }
+
     public PlanFragment getFragment() {
         return fragment;
     }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java
index be804e8797..21f6bb07bd 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java
@@ -726,7 +726,7 @@ public abstract class ScanNode extends PlanNode {
         return context != null
                 && context.getSessionVariable().isIgnoreStorageDataDistribution()
                 && context.getSessionVariable().getEnablePipelineXEngine()
-                && !fragment.isHasNullAwareLeftAntiJoin()
+                && !fragment.hasNullAwareLeftAntiJoin()
                 && getScanRangeNum()
                 < ConnectContext.get().getSessionVariable().getParallelExecInstanceNum() * numBackends;
     }