[Enhancement] support insert local exchange in parttern exchange->join (backport #52021)

be/src/exec/cross_join_node.cpp

@@ -78,6 +78,10 @@ Status CrossJoinNode::init(const TPlanNode& tnode, RuntimeState* state) {
             RETURN_IF_ERROR(
                     Expr::create_expr_trees(_pool, tnode.nestloop_join_node.join_conjuncts, &_join_conjuncts, state));
         }
+
+        if (tnode.nestloop_join_node.__isset.interpolate_passthrough) {
+            _interpolate_passthrough = tnode.nestloop_join_node.interpolate_passthrough;
+        }
         if (tnode.nestloop_join_node.__isset.sql_join_conjuncts) {
             _sql_join_conjuncts = tnode.nestloop_join_node.sql_join_conjuncts;
         }
@@ -697,6 +701,11 @@ std::vector<std::shared_ptr<pipeline::OperatorFactory>> CrossJoinNode::_decompos
         left_ops.emplace_back(std::make_shared<LimitOperatorFactory>(context->next_operator_id(), id(), limit()));
     }
 
+    if (_interpolate_passthrough && !context->is_colocate_group()) {
+        left_ops = context->maybe_interpolate_local_passthrough_exchange(runtime_state(), id(), left_ops,
+                                                                         context->degree_of_parallelism(), true);
+    }
+
     if constexpr (std::is_same_v<BuildFactory, SpillableNLJoinBuildOperatorFactory>) {
         may_add_chunk_accumulate_operator(left_ops, context, id());
     }

be/src/exec/cross_join_node.h

@@ -130,6 +130,7 @@ class CrossJoinNode final : public ExecNode {
     std::vector<uint32_t> _buf_selective;
 
     std::vector<RuntimeFilterBuildDescriptor*> _build_runtime_filters;
+    bool _interpolate_passthrough = false;
 };
 
 } // namespace starrocks

fe/fe-core/src/main/java/com/starrocks/planner/NestLoopJoinNode.java

@@ -123,6 +123,11 @@ protected void toThrift(TPlanNode msg) {
             String sqlJoinPredicate = otherJoinConjuncts.stream().map(Expr::toSql).collect(Collectors.joining(","));
             msg.nestloop_join_node.setSql_join_conjuncts(sqlJoinPredicate);
         }
+        SessionVariable sv = ConnectContext.get().getSessionVariable();
+        if (getCanLocalShuffle()) {
+            msg.nestloop_join_node.setInterpolate_passthrough(sv.isHashJoinInterpolatePassthrough());
+        }
+
 
         if (!buildRuntimeFilters.isEmpty()) {
             msg.nestloop_join_node.setBuild_runtime_filters(

fe/fe-core/src/main/java/com/starrocks/sql/optimizer/operator/physical/PhysicalJoinOperator.java

@@ -33,7 +33,7 @@ public abstract class PhysicalJoinOperator extends PhysicalOperator {
     protected final JoinOperator joinType;
     protected final ScalarOperator onPredicate;
     protected final String joinHint;
-    protected boolean canLocalShuffle;
+    protected boolean outputRequireHashPartition = true;
 
     protected PhysicalJoinOperator(OperatorType operatorType, JoinOperator joinType,
                                    ScalarOperator onPredicate,
@@ -131,11 +131,11 @@ public void fillDisableDictOptimizeColumns(ColumnRefSet columnRefSet) {
         }
     }
 
-    public void setCanLocalShuffle(boolean v) {
-        canLocalShuffle = v;
+    public void setOutputRequireHashPartition(boolean v) {
+        outputRequireHashPartition = v;
     }
 
-    public boolean getCanLocalShuffle() {
-        return canLocalShuffle;
+    public boolean getOutputRequireHashPartition() {
+        return outputRequireHashPartition;
     }
 }

fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/tree/JoinLocalShuffleRule.java

@@ -42,16 +42,22 @@ public Void visit(OptExpression opt, TaskContext context) {
         @Override
         public Void visitPhysicalDistribution(OptExpression opt, TaskContext context) {
             Operator op = opt.getInputs().get(0).getOp();
-            // exchange + local agg + join, then this join can use local shuffle.
+            // 1. exchange + local agg + join, then this join can use local shuffle.
             if ((op instanceof PhysicalHashAggregateOperator) &&
                     ((PhysicalHashAggregateOperator) op).getType().isLocal()) {
                 Operator childOp = opt.getInputs().get(0).getInputs().get(0).getOp();
                 if (childOp instanceof PhysicalJoinOperator) {
                     PhysicalJoinOperator joinOperator = (PhysicalJoinOperator) childOp;
-                    joinOperator.setCanLocalShuffle(true);
+                    joinOperator.setOutputRequireHashPartition(false);
                 }
             }
 
+            // 2. exchange + join
+            if (op instanceof PhysicalJoinOperator) {
+                PhysicalJoinOperator joinOperator = (PhysicalJoinOperator) op;
+                joinOperator.setOutputRequireHashPartition(false);
+            }
+
             for (OptExpression input : opt.getInputs()) {
                 input.getOp().accept(this, input, context);
             }

fe/fe-core/src/main/java/com/starrocks/sql/plan/PlanFragmentBuilder.java

@@ -2463,7 +2463,7 @@ private PlanFragment visitPhysicalJoin(PlanFragment leftFragment, PlanFragment r
                 throw new StarRocksPlannerException("unknown join operator: " + node, INTERNAL_ERROR);
             }
 
-            if (node.getCanLocalShuffle()) {
+            if (!node.getOutputRequireHashPartition()) {
                 joinNode.setCanLocalShuffle(true);
             }
 

fe/fe-core/src/test/java/com/starrocks/sql/plan/JoinLocalShuffleTest.java

@@ -18,6 +18,7 @@
 import com.starrocks.qe.SessionVariable;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
+import org.junit.Test;
 
 public class JoinLocalShuffleTest extends PlanTestBase {
 
@@ -49,4 +50,13 @@ public void joinWithAgg() throws Exception {
         }
         sv.setNewPlanerAggStage(0);
     }
+
+    @Test
+    public void joinUnderExchange() throws Exception {
+        SessionVariable sv = connectContext.getSessionVariable();
+        sv.setInterpolatePassthrough(true);
+        String sql = "select l.* from t0 l join [shuffle] t1 on upper(v1) = v5 join [shuffle] t2 on lower(v1) = v9";
+        String plan = getVerboseExplain(sql);
+        assertContains(plan, "can local shuffle: true");
+    }
 }

gensrc/thrift/PlanNodes.thrift

@@ -682,6 +682,7 @@ struct TNestLoopJoinNode {
     2: optional list<RuntimeFilter.TRuntimeFilterDescription> build_runtime_filters;
     3: optional list<Exprs.TExpr> join_conjuncts
     4: optional string sql_join_conjuncts
+    5: optional bool interpolate_passthrough = false
 }
 
 enum TAggregationOp {