[feat](Nereids) Optimize query by pushing down aggregation through jo…

…in on foreign key (apache#36035)

## Proposed changes

This PR optimizes query performance by pushing down aggregations through
joins when grouped by a foreign key. This adjustment reduces data
processing overhead above the join, improving both speed and resource
efficiency.

Transformation Example:

Before Optimization:
```
Aggregation(group by fk)
     |
   Join(pk = fk)
   /  \
  pk  fk
```
After Optimization:
```
 Join(pk = fk)
 /     \
pk  Aggregation(group by fk)
       |
      fk
```

fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java

@@ -110,6 +110,7 @@
 import org.apache.doris.nereids.rules.rewrite.PushConjunctsIntoJdbcScan;
 import org.apache.doris.nereids.rules.rewrite.PushConjunctsIntoOdbcScan;
 import org.apache.doris.nereids.rules.rewrite.PushDownAggThroughJoin;
+import org.apache.doris.nereids.rules.rewrite.PushDownAggThroughJoinOnPkFk;
 import org.apache.doris.nereids.rules.rewrite.PushDownAggThroughJoinOneSide;
 import org.apache.doris.nereids.rules.rewrite.PushDownDistinctThroughJoin;
 import org.apache.doris.nereids.rules.rewrite.PushDownFilterThroughProject;
@@ -348,8 +349,9 @@ public class Rewriter extends AbstractBatchJobExecutor {
             ),
 
             // this rule should be invoked after topic "Join pull up"
-            topic("eliminate group by keys according to fd items",
-                    topDown(new EliminateGroupByKey())
+            topic("eliminate Aggregate according to fd items",
+                    topDown(new EliminateGroupByKey()),
+                    topDown(new PushDownAggThroughJoinOnPkFk())
             ),
 
             topic("Limit optimization",

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

@@ -62,6 +62,7 @@ public int hashCode() {
     }
 
     private final Set<FuncDepsItem> items;
+    // determinants -> dependencies
     private final Map<Set<Slot>, Set<Set<Slot>>> edges;
 
     public FuncDeps() {
@@ -159,6 +160,24 @@ 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));
+    }
+
+    /**
+     * find the determinants of dependencies
+     */
+    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);
+            }
+        }
+        return determinants;
+    }
+
     @Override
     public String toString() {
         return items.toString();

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

@@ -186,7 +186,7 @@ public enum RuleType {
 
     PUSH_DOWN_AGG_THROUGH_JOIN_ONE_SIDE(RuleTypeClass.REWRITE),
     PUSH_DOWN_AGG_THROUGH_JOIN(RuleTypeClass.REWRITE),
-
+    PUSH_DOWN_AGG_THROUGH_JOIN_ON_PKFK(RuleTypeClass.REWRITE),
     TRANSPOSE_LOGICAL_SEMI_JOIN_LOGICAL_JOIN(RuleTypeClass.REWRITE),
     TRANSPOSE_LOGICAL_SEMI_JOIN_LOGICAL_JOIN_PROJECT(RuleTypeClass.REWRITE),
     LOGICAL_SEMI_JOIN_COMMUTE(RuleTypeClass.REWRITE),