fix

Author: feiniaofeiafei

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

@@ -183,6 +183,7 @@ public boolean isFuncDeps(Set<Slot> dominate, Set<Slot> dependency) {
         return items.contains(new FuncDepsItem(dominate, dependency));
     }
 
+    // 这个也是判断是否为双射的
     public boolean isCircleDeps(Set<Slot> dominate, Set<Slot> dependency) {
         return items.contains(new FuncDepsItem(dominate, dependency))
                 && items.contains(new FuncDepsItem(dependency, dominate));
@@ -201,16 +202,30 @@ public Map<Set<Slot>, Set<Set<Slot>>> getREdges() {
     }
 
     /**
-     * find the determinants of dependencies
+     * Finds all slot sets that have a bijective relationship with the given slot set.
+     * Given edges containing:
+     *   {A} -> {{B}, {C}}
+     *   {B} -> {{A}, {D}}
+     *   {C} -> {{A}}
+     * When slot = {A}, returns {{B}} because {A} and {B} mutually determine each other.
+     * {C} is not returned because {C} does not determine {A} (one-way dependency only).
      */
-    public Set<Set<Slot>> findDeterminats(Set<Slot> dependency) {
-        Set<Set<Slot>> determinants = new HashSet<>();
-        for (FuncDepsItem item : items) {
-            if (item.dependencies.equals(dependency)) {
-                determinants.add(item.determinants);
+    public Set<Set<Slot>> findBijectionSlots(Set<Slot> slot) {
+        Set<Set<Slot>> bijectionSlots = new HashSet<>();
+        if (!edges.containsKey(slot)) {
+            return bijectionSlots;
+        }
+        for (Set<Slot> dep : edges.get(slot)) {
+            if (!edges.containsKey(dep)) {
+                continue;
+            }
+            for (Set<Slot> det : edges.get(dep)) {
+                if (det.equals(slot)) {
+                    bijectionSlots.add(dep);
+                }
             }
         }
-        return determinants;
+        return bijectionSlots;
     }
 
     @Override

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownAggThroughJoinOnPkFk.java

@@ -47,6 +47,7 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 /**
  * Push down agg through join with foreign key:
@@ -131,13 +132,27 @@ private LogicalAggregate<?> eliminatePrimaryOutput(LogicalAggregate<?> agg, Plan
         if (primary.getOutputSet().stream().noneMatch(aggInputs::contains)) {
             return agg;
         }
-        Set<Slot> primaryOutputSet = primary.getOutputSet();
-        Set<Slot> primarySlots = Sets.intersection(aggInputs, primaryOutputSet);
+        // Firstly, using fd to eliminate group by key.
+        // group by primary_table_pk, primary_table_other
+        // -> group by primary_table_pk
+        Set<Set<Slot>> groupBySlots = new HashSet<>();
+        for (Expression expression : agg.getGroupByExpressions()) {
+            groupBySlots.add(expression.getInputSlots());
+        }
         DataTrait dataTrait = child.getLogicalProperties().getTrait();
         FuncDeps funcDeps = dataTrait.getAllValidFuncDeps(Sets.union(foreign.getOutputSet(), primary.getOutputSet()));
+        Set<Slot> foreignOutput = Sets.intersection(agg.getOutputSet(), foreign.getOutputSet());
+        Set<Set<Slot>> minGroupBySlots = funcDeps.eliminateDeps(groupBySlots, foreignOutput);
+        List<Slot> minGroupBySlotList = minGroupBySlots.stream().flatMap(Set::stream).collect(Collectors.toList());
+
+        // Secondly, put bijective slot into map: {primary_table_pk : foreign_table_fk}
+        // Bijective slots are mutually interchangeable within GROUP BY keys.
+        // group by primary_table_pk equals group by foreign_table_fk
+        Set<Slot> primaryOutputSet = primary.getOutputSet();
+        Set<Slot> primarySlots = Sets.intersection(aggInputs, primaryOutputSet);
         HashMap<Slot, Slot> primaryToForeignDeps = new HashMap<>();
         for (Slot slot : primarySlots) {
-            Set<Set<Slot>> replacedSlotSets = funcDeps.findDeterminats(ImmutableSet.of(slot));
+            Set<Set<Slot>> replacedSlotSets = funcDeps.findBijectionSlots(ImmutableSet.of(slot));
             for (Set<Slot> replacedSlots : replacedSlotSets) {
                 if (primaryOutputSet.stream().noneMatch(replacedSlots::contains)
                         && replacedSlots.size() == 1) {
@@ -147,7 +162,9 @@ private LogicalAggregate<?> eliminatePrimaryOutput(LogicalAggregate<?> agg, Plan
             }
         }
 
-        Set<Expression> newGroupBySlots = constructNewGroupBy(agg, primaryOutputSet, primaryToForeignDeps);
+        // Thirdly, construct new Agg below join.
+        Set<Expression> newGroupBySlots = constructNewGroupBy(minGroupBySlotList, primaryOutputSet,
+                primaryToForeignDeps);
         List<NamedExpression> newOutput = constructNewOutput(
                 agg, primaryOutputSet, primaryToForeignDeps, funcDeps, primary);
         if (newGroupBySlots == null || newOutput == null) {
@@ -156,10 +173,10 @@ private LogicalAggregate<?> eliminatePrimaryOutput(LogicalAggregate<?> agg, Plan
         return agg.withGroupByAndOutput(ImmutableList.copyOf(newGroupBySlots), ImmutableList.copyOf(newOutput));
     }
 
-    private @Nullable Set<Expression> constructNewGroupBy(LogicalAggregate<?> agg, Set<Slot> primaryOutputs,
-            Map<Slot, Slot> primaryToForeignBiDeps) {
+    private @Nullable Set<Expression> constructNewGroupBy(List<? extends Expression> gbyExpression,
+            Set<Slot> primaryOutputs, Map<Slot, Slot> primaryToForeignBiDeps) {
         Set<Expression> newGroupBySlots = new HashSet<>();
-        for (Expression expression : agg.getGroupByExpressions()) {
+        for (Expression expression : gbyExpression) {
             if (!(expression instanceof Slot)) {
                 return null;
             }

regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy

@@ -3537,4 +3537,9 @@ class Suite implements GroovyInterceptable {
             GlobalLock.unlock(lockName)
         }
     }
+
+    def explain_and_result = { tag, sql ->
+        "qt_${tag}_shape"          "explain shape plan ${sql}"
+        "order_qt_${tag}_result"   "${sql}"
+    }
 }