[fix](fe) Fix invalid semi-join transpose when the bottom join is a mark join

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/join/SemiJoinSemiJoinTransposeProject.java

@@ -68,7 +68,15 @@ public Rule build() {
                 .when(this::typeChecker)
                 .when(topSemi -> InnerJoinLAsscomProject.checkReorder(topSemi, topSemi.left().child(), false))
                 .whenNot(join -> join.hasDistributeHint() || join.left().child().hasDistributeHint())
-                .when(join -> join.left().isAllSlots()))
+                .when(join -> join.left().isAllSlots())
+                // the transpose swaps the bottom semi join to the top, so the mark slot
+                // produced by the bottom mark join would be produced by the new top semi
+                // join. if the top semi join references the mark slot in its conjuncts,
+                // those conjuncts would be moved to the new bottom semi join whose children
+                // don't output the mark slot, which makes the mark slot dangling and fails
+                // physical planning with "slot not from children", so the transpose must be
+                // rejected in this case
+                .whenNot(this::isMarkSlotReferencedByTopJoin))
                 .then(topProject -> {
                     LogicalJoin<LogicalProject<LogicalJoin<GroupPlan, GroupPlan>>, GroupPlan> topSemi
                             = topProject.child();
@@ -119,4 +127,24 @@ public Rule build() {
     public boolean typeChecker(LogicalJoin<LogicalProject<LogicalJoin<GroupPlan, GroupPlan>>, GroupPlan> topJoin) {
         return VALID_TYPE_PAIR_SET.contains(Pair.of(topJoin.getJoinType(), topJoin.left().child().getJoinType()));
     }
+
+    /**
+     * check whether the top semi join references the mark slot produced by the bottom mark
+     * join in its conjuncts. in the transposed plan the mark slot is produced by the new
+     * top semi join (built from the bottom semi join), while the top semi join becomes the
+     * new bottom semi join whose children are A and C, which don't output the mark slot.
+     * so if the top semi join's conjuncts reference the mark slot, the transpose would make
+     * the mark slot dangling and must be rejected.
+     */
+    private boolean isMarkSlotReferencedByTopJoin(
+            LogicalJoin<LogicalProject<LogicalJoin<GroupPlan, GroupPlan>>, GroupPlan> topSemi) {
+        LogicalJoin<GroupPlan, GroupPlan> bottomSemi = topSemi.left().child();
+        if (!bottomSemi.isMarkJoin()) {
+            return false;
+        }
+        ExprId markSlotExprId = bottomSemi.getMarkJoinSlotReference().get().getExprId();
+        return topSemi.getExpressions().stream()
+                .flatMap(expr -> expr.getInputSlotExprIds().stream())
+                .anyMatch(markSlotExprId::equals);
+    }
 }

fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/join/SemiJoinSemiJoinTransposeProjectTest.java

@@ -18,6 +18,7 @@
 package org.apache.doris.nereids.rules.exploration.join;
 
 import org.apache.doris.common.Pair;
+import org.apache.doris.nereids.trees.expressions.Slot;
 import org.apache.doris.nereids.trees.plans.JoinType;
 import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
@@ -95,4 +96,58 @@ public void testSemiProjectSemiCommuteMarkJoin() {
                         )
                 );
     }
+
+    @Test
+    public void testSemiProjectSemiCommuteRejectedWhenTopJoinReferencesBottomMarkSlot() {
+        /*
+         * the transpose must be rejected when the bottom semi join is a mark join and the
+         * top semi join references the mark slot in its conjuncts. otherwise the transposed
+         * plan would move the conjuncts that reference the mark slot to a join whose
+         * children don't output the mark slot, which fails physical planning with
+         * "slot not from children".
+         *
+         *        topJoin(references mark)        the transpose is rejected, the plan
+         *        /       \                       keeps the original order:
+         *    abProject    t3                     topJoin
+         *      |                                  /      \
+         * bottomMarkJoin(t1 anti t2)        abProject   t3
+         *    /      \                          |
+         *   t1      t2                   bottomMarkJoin
+         *                                       /      \
+         *                                      t1      t2
+         */
+        // bottom mark join: t1 left anti t2, markJoinConjuncts = (t1#0 = t2#0),
+        // output = [t1#0, t1#1, markSlot]
+        LogicalPlan bottomMarkJoin = new LogicalPlanBuilder(scan1)
+                .markJoinWithMarkConjuncts(scan2, JoinType.LEFT_ANTI_JOIN, Pair.of(0, 0))
+                .build();
+        // project exposes [t1#0, markSlot]
+        LogicalPlan abProject = new LogicalPlanBuilder(bottomMarkJoin)
+                .project(ImmutableList.of(0, 2))
+                .build();
+        // top anti join on t3 whose other conjunct references the mark slot of the bottom
+        // mark join, this is exactly the plan shape that used to trigger the bug
+        Slot markSlot = abProject.getOutput().get(1);
+        LogicalPlan topJoin = new LogicalPlanBuilder(abProject)
+                .join(scan3, JoinType.LEFT_ANTI_JOIN, ImmutableList.of(), ImmutableList.of(markSlot))
+                .projectAll()
+                .build();
+        // the transpose is rejected, so the plan keeps the original order and the
+        // mark join still produces the mark slot below the top anti join
+        PlanChecker.from(MemoTestUtils.createConnectContext(), topJoin)
+                .applyExploration(SemiJoinSemiJoinTransposeProject.INSTANCE.build())
+                .matches(
+                        logicalProject(
+                                logicalJoin(
+                                        logicalProject(
+                                                logicalJoin(
+                                                        logicalOlapScan().when(s -> s.getTable().getName().equals("t1")),
+                                                        logicalOlapScan().when(s -> s.getTable().getName().equals("t2"))
+                                                ).when(join -> join.getJoinType() == JoinType.LEFT_ANTI_JOIN)
+                                        ),
+                                        logicalOlapScan().when(s -> s.getTable().getName().equals("t3"))
+                                ).when(join -> join.getJoinType() == JoinType.LEFT_ANTI_JOIN)
+                        )
+                );
+    }
 }