[enhance](aggregate) Add rewrite rule DecomposeRepeatWithPreAggregation

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

@@ -60,6 +60,7 @@
 import org.apache.doris.nereids.rules.rewrite.CountDistinctRewrite;
 import org.apache.doris.nereids.rules.rewrite.CountLiteralRewrite;
 import org.apache.doris.nereids.rules.rewrite.CreatePartitionTopNFromWindow;
+import org.apache.doris.nereids.rules.rewrite.DecomposeRepeatWithPreAggregation;
 import org.apache.doris.nereids.rules.rewrite.DecoupleEncodeDecode;
 import org.apache.doris.nereids.rules.rewrite.DeferMaterializeTopNResult;
 import org.apache.doris.nereids.rules.rewrite.DistinctAggStrategySelector;
@@ -902,7 +903,8 @@ private static List<RewriteJob> getWholeTreeRewriteJobs(
                     rewriteJobs.addAll(jobs(topic("or expansion",
                             custom(RuleType.OR_EXPANSION, () -> OrExpansion.INSTANCE))));
                 }
-
+                rewriteJobs.add(topic("repeat rewrite",
+                        custom(RuleType.DECOMPOSE_REPEAT, () -> DecomposeRepeatWithPreAggregation.INSTANCE)));
                 rewriteJobs.addAll(jobs(topic("split multi distinct",
                         custom(RuleType.DISTINCT_AGG_STRATEGY_SELECTOR, () -> DistinctAggStrategySelector.INSTANCE))));
 

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

@@ -414,7 +414,7 @@ public enum RuleType {
     SHOR_CIRCUIT_POINT_QUERY(RuleTypeClass.REWRITE),
     // skew rewrtie
     SALT_JOIN(RuleTypeClass.REWRITE),
-
+    DECOMPOSE_REPEAT(RuleTypeClass.REWRITE),
     DISTINCT_AGGREGATE_SPLIT(RuleTypeClass.REWRITE),
     PROCESS_SCALAR_AGG_MUST_USE_MULTI_DISTINCT(RuleTypeClass.REWRITE),
 

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/Alias.java

@@ -171,6 +171,10 @@ public Alias withChildren(List<Expression> children) {
         return new Alias(exprId, children, name, qualifier, nameFromChild);
     }
 
+    public Alias withExprId(ExprId exprId) {
+        return new Alias(exprId, children, name, qualifier, nameFromChild);
+    }
+
     public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) {
         return visitor.visitAlias(this, context);
     }

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

@@ -24,6 +24,7 @@
 import org.apache.doris.nereids.trees.expressions.NamedExpression;
 import org.apache.doris.nereids.trees.expressions.Slot;
 import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.GroupingScalarFunction;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.PlanType;
 import org.apache.doris.nereids.trees.plans.algebra.Repeat;
@@ -34,6 +35,7 @@
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 
+import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
@@ -125,6 +127,19 @@ public List<NamedExpression> getOutputs() {
         return outputExpressions;
     }
 
+    /**
+     * get groupingScalarFunction with Alias
+     */
+    public List<NamedExpression> getGroupingScalarFunctionAlias() {
+        List<NamedExpression> functionList = new ArrayList<>();
+        for (NamedExpression outputExpression : outputExpressions) {
+            if (outputExpression.containsType(GroupingScalarFunction.class)) {
+                functionList.add(outputExpression);
+            }
+        }
+        return functionList;
+    }
+
     @Override
     public String toString() {
         return Utils.toSqlString("LogicalRepeat",

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

@@ -39,6 +39,7 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.stream.Stream;
 
 /**
  * PhysicalRepeat.
@@ -117,7 +118,7 @@ public String toString() {
 
     @Override
     public List<Slot> computeOutput() {
-        return outputExpressions.stream()
+        return Stream.concat(outputExpressions.stream(), Stream.of(groupingId))
                 .map(NamedExpression::toSlot)
                 .collect(ImmutableList.toImmutableList());
     }

fe/fe-core/src/test/java/org/apache/doris/nereids/mv/PreMaterializedViewRewriterTest.java

@@ -2977,6 +2977,7 @@ private void checkIfEquals(String originalSql, List<String> equivalentSqlList) {
 
     private CascadesContext initOriginal(String sql) {
         CascadesContext cascadesContext = createCascadesContext(sql, connectContext);
+        connectContext.setThreadLocalInfo();
         PlanChecker.from(cascadesContext).analyze().rewrite();
         return cascadesContext;
     }