branch-4.1: [fix](fd)drop Function dependencies from join outer side #65982 (#66775)

Cherry-picked from #65982

Co-authored-by: minghong <zhouminghong@selectdb.com>

Author: github-actions[bot]

fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DataTrait.java

@@ -237,6 +237,16 @@ public void addFuncDepsDG(DataTrait fd) {
             fdDgBuilder.addDeps(fd.fdDg);
         }
 
+        /**
+         * Add FDs from the nullable side of an outer join, filtering out edges whose
+         * determinant may be NULL in the immediate child's current output — those would
+         * be invalidated by null-extension of unmatched rows. Determinants are canonicalized
+         * against childOutput (by ExprId) before the nullability check.
+         */
+        public void addFuncDepsDGForOuterJoinNullableSide(DataTrait fd, List<Slot> childOutput) {
+            fdDgBuilder.addDepsForOuterJoinNullableSide(fd.fdDg, childOutput);
+        }
+
         /**add Dependency relation for dominate and dependency*/
         public void addDeps(Set<Slot> dominate, Set<Slot> dependency) {
             if (dominate.isEmpty() || dependency.isEmpty()) {

fe/fe-core/src/main/java/org/apache/doris/nereids/properties/FuncDepsDG.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.nereids.properties;
 
+import org.apache.doris.nereids.trees.expressions.ExprId;
 import org.apache.doris.nereids.trees.expressions.Slot;
 
 import com.google.common.collect.ImmutableList;
@@ -219,6 +220,43 @@ public void addDeps(FuncDepsDG funcDepsDG) {
             }
         }
 
+        /**
+         * Add FD edges from the nullable side of an outer join. Only keep edges whose
+         * determinant slots are all NOT NULL in the immediate child's current output:
+         * matched rows always carry a non-null determinant, while unmatched rows contribute
+         * (NULL, NULL), so they cannot collide. Edges with a nullable determinant are dropped
+         * because a matched row with determinant=NULL could conflict with an unmatched (NULL, NULL).
+         * The determinant slots stored in the graph may carry a stale nullable flag (slot
+         * equality/hash use only ExprId and getOrCreateNode never replaces the stored object),
+         * so each determinant is canonicalized against the child's current output before the
+         * nullability check.
+         */
+        public void addDepsForOuterJoinNullableSide(FuncDepsDG funcDepsDG, List<Slot> childOutput) {
+            Map<ExprId, Slot> outputSlotMap = new HashMap<>();
+            for (Slot slot : childOutput) {
+                outputSlotMap.put(slot.getExprId(), slot);
+            }
+            for (DGItem dgItem : funcDepsDG.dgItems) {
+                Set<Slot> canonicalSlots = new HashSet<>();
+                boolean allNotNull = true;
+                for (Slot slot : dgItem.slots) {
+                    Slot outputSlot = outputSlotMap.get(slot.getExprId());
+                    // a determinant not in the child's output cannot be trusted; drop the edge
+                    if (outputSlot == null || outputSlot.nullable()) {
+                        allNotNull = false;
+                        break;
+                    }
+                    canonicalSlots.add(outputSlot);
+                }
+                if (!allNotNull) {
+                    continue;
+                }
+                for (int childIdx : dgItem.children) {
+                    addDeps(canonicalSlots, funcDepsDG.dgItems.get(childIdx).slots);
+                }
+            }
+        }
+
         public void replace(Map<Slot, Slot> replaceSlotMap) {
             for (DGItem item : dgItems) {
                 item.replace(replaceSlotMap);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalJoin.java

@@ -711,11 +711,50 @@ public void computeEqualSet(Builder builder) {
 
     @Override
     public void computeFd(Builder builder) {
-        if (!joinType.isLeftSemiOrAntiJoin()) {
-            builder.addFuncDepsDG(right().getLogicalProperties().getTrait());
-        }
-        if (!joinType.isRightSemiOrAntiJoin()) {
-            builder.addFuncDepsDG(left().getLogicalProperties().getTrait());
+        switch (joinType) {
+            case INNER_JOIN:
+            case ASOF_LEFT_INNER_JOIN:
+            case ASOF_RIGHT_INNER_JOIN:
+            case CROSS_JOIN:
+                builder.addFuncDepsDG(left().getLogicalProperties().getTrait());
+                builder.addFuncDepsDG(right().getLogicalProperties().getTrait());
+                break;
+            case LEFT_SEMI_JOIN:
+            case LEFT_ANTI_JOIN:
+            case NULL_AWARE_LEFT_ANTI_JOIN:
+                // Semi/anti joins only output the left side; right-side FDs are irrelevant.
+                builder.addFuncDepsDG(left().getLogicalProperties().getTrait());
+                break;
+            case LEFT_OUTER_JOIN:
+            case ASOF_LEFT_OUTER_JOIN:
+                // Left side preserved; right side nullable — keep only FDs whose
+                // determinant is NOT NULL in the right child's current output.
+                builder.addFuncDepsDG(left().getLogicalProperties().getTrait());
+                builder.addFuncDepsDGForOuterJoinNullableSide(
+                        right().getLogicalProperties().getTrait(), right().getOutput());
+                break;
+            case RIGHT_SEMI_JOIN:
+            case RIGHT_ANTI_JOIN:
+                // Semi/anti joins only output the right side; left-side FDs are irrelevant.
+                builder.addFuncDepsDG(right().getLogicalProperties().getTrait());
+                break;
+            case RIGHT_OUTER_JOIN:
+            case ASOF_RIGHT_OUTER_JOIN:
+                // Right side preserved; left side nullable — keep only FDs whose
+                // determinant is NOT NULL in the left child's current output.
+                builder.addFuncDepsDG(right().getLogicalProperties().getTrait());
+                builder.addFuncDepsDGForOuterJoinNullableSide(
+                        left().getLogicalProperties().getTrait(), left().getOutput());
+                break;
+            case FULL_OUTER_JOIN:
+                // Both sides are nullable; keep only FDs whose determinant is NOT NULL.
+                builder.addFuncDepsDGForOuterJoinNullableSide(
+                        left().getLogicalProperties().getTrait(), left().getOutput());
+                builder.addFuncDepsDGForOuterJoinNullableSide(
+                        right().getLogicalProperties().getTrait(), right().getOutput());
+                break;
+            default:
+                break;
         }
     }
 

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalHashJoin.java

@@ -357,11 +357,50 @@ public void computeEqualSet(DataTrait.Builder builder) {
 
     @Override
     public void computeFd(DataTrait.Builder builder) {
-        if (!joinType.isLeftSemiOrAntiJoin()) {
-            builder.addFuncDepsDG(right().getLogicalProperties().getTrait());
-        }
-        if (!joinType.isRightSemiOrAntiJoin()) {
-            builder.addFuncDepsDG(left().getLogicalProperties().getTrait());
+        switch (joinType) {
+            case INNER_JOIN:
+            case ASOF_LEFT_INNER_JOIN:
+            case ASOF_RIGHT_INNER_JOIN:
+            case CROSS_JOIN:
+                builder.addFuncDepsDG(left().getLogicalProperties().getTrait());
+                builder.addFuncDepsDG(right().getLogicalProperties().getTrait());
+                break;
+            case LEFT_SEMI_JOIN:
+            case LEFT_ANTI_JOIN:
+            case NULL_AWARE_LEFT_ANTI_JOIN:
+                // Semi/anti joins only output the left side; right-side FDs are irrelevant.
+                builder.addFuncDepsDG(left().getLogicalProperties().getTrait());
+                break;
+            case LEFT_OUTER_JOIN:
+            case ASOF_LEFT_OUTER_JOIN:
+                // Left side preserved; right side nullable — keep only FDs whose
+                // determinant is NOT NULL in the right child's current output.
+                builder.addFuncDepsDG(left().getLogicalProperties().getTrait());
+                builder.addFuncDepsDGForOuterJoinNullableSide(
+                        right().getLogicalProperties().getTrait(), right().getOutput());
+                break;
+            case RIGHT_SEMI_JOIN:
+            case RIGHT_ANTI_JOIN:
+                // Semi/anti joins only output the right side; left-side FDs are irrelevant.
+                builder.addFuncDepsDG(right().getLogicalProperties().getTrait());
+                break;
+            case RIGHT_OUTER_JOIN:
+            case ASOF_RIGHT_OUTER_JOIN:
+                // Right side preserved; left side nullable — keep only FDs whose
+                // determinant is NOT NULL in the left child's current output.
+                builder.addFuncDepsDG(right().getLogicalProperties().getTrait());
+                builder.addFuncDepsDGForOuterJoinNullableSide(
+                        left().getLogicalProperties().getTrait(), left().getOutput());
+                break;
+            case FULL_OUTER_JOIN:
+                // Both sides are nullable; keep only FDs whose determinant is NOT NULL.
+                builder.addFuncDepsDGForOuterJoinNullableSide(
+                        left().getLogicalProperties().getTrait(), left().getOutput());
+                builder.addFuncDepsDGForOuterJoinNullableSide(
+                        right().getLogicalProperties().getTrait(), right().getOutput());
+                break;
+            default:
+                break;
         }
     }
 }

fe/fe-core/src/test/java/org/apache/doris/nereids/properties/FdTest.java

@@ -19,6 +19,10 @@
 
 import org.apache.doris.nereids.trees.expressions.Slot;
 import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalHashJoin;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan;
 import org.apache.doris.nereids.util.PlanChecker;
 import org.apache.doris.utframe.TestWithFeService;
 
@@ -27,6 +31,7 @@
 import org.junit.jupiter.api.Test;
 
 import java.util.Set;
+import java.util.function.Predicate;
 
 class FdTest extends TestWithFeService {
     @Override
@@ -46,6 +51,13 @@ protected void runBeforeAll() throws Exception {
                 + "UNIQUE KEY(id)\n"
                 + "distributed by hash(id) buckets 10\n"
                 + "properties('replication_num' = '1');");
+        createTable("create table test.nullable_uni (\n"
+                + "id int,\n"
+                + "id2 int not null,\n"
+                + "name varchar(128) not null)\n"
+                + "UNIQUE KEY(id)\n"
+                + "distributed by hash(id) buckets 10\n"
+                + "properties('replication_num' = '1');");
         connectContext.setDatabase("test");
         connectContext.getSessionVariable().setDisableNereidsRules("PRUNE_EMPTY_PARTITION");
     }
@@ -147,38 +159,139 @@ void testJoin() {
         Assertions.assertTrue(plan.getLogicalProperties().getTrait()
                 .isDependent(ImmutableSet.of(plan.getOutput().get(1)), ImmutableSet.of(plan.getOutput().get(2))));
 
-        // foj
+        // foj: both sides nullable — keep FDs with NOT NULL determinants
         plan = PlanChecker.from(connectContext)
                 .analyze("select t1.id, t1.id2, t2.id, t2.id2 "
                         + "from uni as t1 full outer join uni as t2 on t1.id2 = t2.id2")
                 .rewrite()
                 .getPlan();
+        // t1.id is NOT NULL, so {t1.id} -> {t1.id2} survives null extension
         Assertions.assertTrue(plan.getLogicalProperties().getTrait()
                 .isDependent(ImmutableSet.of(plan.getOutput().get(0)), ImmutableSet.of(plan.getOutput().get(1))));
+        // t2.id is NOT NULL, so {t2.id} -> {t2.id2} survives null extension
         Assertions.assertTrue(plan.getLogicalProperties().getTrait()
                 .isDependent(ImmutableSet.of(plan.getOutput().get(2)), ImmutableSet.of(plan.getOutput().get(3))));
 
-        // loj
+        // loj: left side preserved, right side nullable — only NOT NULL-determinant FDs from right propagate
         plan = PlanChecker.from(connectContext)
                 .analyze("select t1.id, t1.id2, t2.id, t2.id2 "
                         + "from uni as t1 left outer join uni as t2 on t1.id2 = t2.id2")
                 .rewrite()
                 .getPlan();
+        // t1.id is NOT NULL, left side always preserved
         Assertions.assertTrue(plan.getLogicalProperties().getTrait()
                 .isDependent(ImmutableSet.of(plan.getOutput().get(0)), ImmutableSet.of(plan.getOutput().get(1))));
+        // t2.id is NOT NULL, so {t2.id} -> {t2.id2} survives null extension
         Assertions.assertTrue(plan.getLogicalProperties().getTrait()
                 .isDependent(ImmutableSet.of(plan.getOutput().get(2)), ImmutableSet.of(plan.getOutput().get(3))));
 
-        // roj
+        // roj: right side preserved, left side nullable — only NOT NULL-determinant FDs from left propagate
         plan = PlanChecker.from(connectContext)
                 .analyze("select t1.id, t1.id2, t2.id, t2.id2 "
                         + "from uni as t1 right outer join uni as t2 on t1.id2 = t2.id2")
                 .rewrite()
                 .getPlan();
+        // t1.id is NOT NULL, so {t1.id} -> {t1.id2} survives null extension
         Assertions.assertTrue(plan.getLogicalProperties().getTrait()
                 .isDependent(ImmutableSet.of(plan.getOutput().get(0)), ImmutableSet.of(plan.getOutput().get(1))));
+        // t2.id is NOT NULL, right side always preserved
         Assertions.assertTrue(plan.getLogicalProperties().getTrait()
                 .isDependent(ImmutableSet.of(plan.getOutput().get(2)), ImmutableSet.of(plan.getOutput().get(3))));
+
+        // loj with nullable determinant: FD should be dropped
+        plan = PlanChecker.from(connectContext)
+                .analyze("select t1.id, t1.id2, t2.id, t2.id2 "
+                        + "from uni as t1 left outer join nullable_uni as t2 on t1.id2 = t2.id2")
+                .rewrite()
+                .getPlan();
+        // t1 side preserved
+        Assertions.assertTrue(plan.getLogicalProperties().getTrait()
+                .isDependent(ImmutableSet.of(plan.getOutput().get(0)), ImmutableSet.of(plan.getOutput().get(1))));
+        // t2.id is nullable, so {t2.id} -> {t2.id2} should be dropped
+        Assertions.assertFalse(plan.getLogicalProperties().getTrait()
+                .isDependent(ImmutableSet.of(plan.getOutput().get(2)), ImmutableSet.of(plan.getOutput().get(3))));
+
+        // foj with nullable determinant on one side
+        plan = PlanChecker.from(connectContext)
+                .analyze("select t1.id, t1.id2, t2.id, t2.id2 "
+                        + "from uni as t1 full outer join nullable_uni as t2 on t1.id2 = t2.id2")
+                .rewrite()
+                .getPlan();
+        // t1.id is NOT NULL, so {t1.id} -> {t1.id2} survives
+        Assertions.assertTrue(plan.getLogicalProperties().getTrait()
+                .isDependent(ImmutableSet.of(plan.getOutput().get(0)), ImmutableSet.of(plan.getOutput().get(1))));
+        // t2.id is nullable, so {t2.id} -> {t2.id2} should be dropped
+        Assertions.assertFalse(plan.getLogicalProperties().getTrait()
+                .isDependent(ImmutableSet.of(plan.getOutput().get(2)), ImmutableSet.of(plan.getOutput().get(3))));
+    }
+
+    @Test
+    void testNestedOuterJoinNullableDeterminant() {
+        // Reduced failing tree from review "Check determinant nullability against the current child output":
+        //   Aggregate(group by r_id, c)
+        //     RightOuterJoin
+        //       Project(l_id, r_id, coalesce(r_id, 1) AS c)
+        //         LeftOuterJoin
+        //           Scan L
+        //           Scan R(r_id NOT NULL UNIQUE)
+        //       Scan V
+        // r_id is NOT NULL in R but becomes nullable at the inner LOJ output; the Project derives
+        // r_id -> c from the expression. At the outer join output this FD must be dropped:
+        // unmatched V rows inject (r_id=NULL, c=NULL), which collides with the Project's own
+        // (r_id=NULL, c=1). After rewrite the sub-query alias is inlined into a plain project
+        // (LogicalSubQueryAliasToLogicalProject) whose trait keeps the stale non-nullable r_id,
+        // so the outer join must still be checked against the immediate child's current output.
+        // c is kept in the select list so that it is not pruned away before the trait check.
+        // Disable join reorder to keep the join tree stable (v LEFT OUTER JOIN p as written).
+        connectContext.getSessionVariable().setDisableJoinReorder(true);
+        String sql = "select p.id, p.c, count(*) "
+                + "from uni as v "
+                + "left outer join ("
+                + "select l.id2, r.id, coalesce(r.id, 1) as c "
+                + "from agg as l left outer join uni as r on l.id2 = r.id2) p "
+                + "on v.id2 = p.id2 "
+                + "group by p.id, p.c";
+
+        LogicalAggregate<?> aggregate = (LogicalAggregate<?>) findNode(
+                PlanChecker.from(connectContext).analyze(sql).getPlan(), n -> n instanceof LogicalAggregate);
+        Assertions.assertNotNull(aggregate);
+        // group by (r_id, c); both are plain slots after subquery inlining
+        Slot rId = (Slot) aggregate.getGroupByExpressions().get(0);
+        Slot c = (Slot) aggregate.getGroupByExpressions().get(1);
+
+        // logical path: the outer join's trait must not contain r_id -> c
+        Plan rewritten = PlanChecker.from(connectContext).analyze(sql).rewrite().getPlan();
+        LogicalJoin<?, ?> outerJoin = (LogicalJoin<?, ?>) findNode(rewritten, n -> n instanceof LogicalJoin);
+        Assertions.assertNotNull(outerJoin, "rewritten plan: " + rewritten.treeString());
+        Assertions.assertFalse(outerJoin.getLogicalProperties().getTrait()
+                        .isDependent(ImmutableSet.of(rId), ImmutableSet.of(c)),
+                "r_id -> c must be dropped at the outer join since r_id is nullable on the outer side");
+
+        // physical path: PhysicalHashJoin must drop r_id -> c as well; pick the outer join
+        // (its subtree contains the inner join). implement() applies the implementation rules
+        // directly (no CBO), so no table statistics are required.
+        PhysicalPlan physicalPlan = PlanChecker.from(connectContext)
+                .analyze(sql).rewrite().implement().getPhysicalPlan();
+        PhysicalHashJoin<?, ?> physicalOuterJoin = (PhysicalHashJoin<?, ?>) findNode(physicalPlan,
+                n -> n instanceof PhysicalHashJoin
+                        && n.anyMatch(p -> p instanceof PhysicalHashJoin && p != n));
+        Assertions.assertNotNull(physicalOuterJoin, "physical plan: " + physicalPlan.treeString());
+        Assertions.assertFalse(physicalOuterJoin.getLogicalProperties().getTrait()
+                        .isDependent(ImmutableSet.of(rId), ImmutableSet.of(c)),
+                "physical join must also drop r_id -> c since r_id is nullable on the outer side");
+    }
+
+    private Plan findNode(Plan plan, Predicate<Plan> predicate) {
+        if (predicate.test(plan)) {
+            return plan;
+        }
+        for (Plan child : plan.children()) {
+            Plan found = findNode(child, predicate);
+            if (found != null) {
+                return found;
+            }
+        }
+        return null;
     }
 
     @Test