diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewAggregateRule.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewAggregateRule.java
index 20efcf839c747a..4e3c2a650e520b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewAggregateRule.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewAggregateRule.java
@@ -17,6 +17,7 @@
 
 package org.apache.doris.nereids.rules.exploration.mv;
 
+import org.apache.doris.common.Pair;
 import org.apache.doris.nereids.jobs.joinorder.hypergraph.Edge;
 import org.apache.doris.nereids.jobs.joinorder.hypergraph.HyperGraph;
 import org.apache.doris.nereids.jobs.joinorder.hypergraph.node.AbstractNode;
@@ -24,54 +25,178 @@
 import org.apache.doris.nereids.rules.exploration.mv.StructInfo.PlanSplitContext;
 import org.apache.doris.nereids.rules.exploration.mv.mapping.ExpressionMapping;
 import org.apache.doris.nereids.rules.exploration.mv.mapping.SlotMapping;
-import org.apache.doris.nereids.trees.expressions.Slot;
-import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
 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.LogicalProject;
+import org.apache.doris.nereids.util.ExpressionUtils;
 
-import com.google.common.collect.ImmutableMultimap;
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Multimap;
 import com.google.common.collect.Sets;
-import static org.apache.doris.nereids.rules.exploration.mv.StructInfo.AGGREGATE_PATTERN_CHECKER;
 
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 /**
  * AbstractMaterializedViewAggregateRule
  * This is responsible for common aggregate rewriting
- * */
+ */
 public abstract class AbstractMaterializedViewAggregateRule extends AbstractMaterializedViewRule {
 
     @Override
     protected Plan rewriteQueryByView(MatchMode matchMode,
             StructInfo queryStructInfo,
             StructInfo viewStructInfo,
-            SlotMapping queryToViewSlotMappings,
+            SlotMapping queryToViewSlotMapping,
             Plan tempRewritedPlan,
             MaterializationContext materializationContext) {
 
-        PlanSplitContext planSplitContext = new PlanSplitContext(Sets.newHashSet(LogicalAggregate.class));
-        viewStructInfo.getTopPlan().accept(StructInfo.PLAN_SPLITTER, planSplitContext);
-        // generate aggregate in mv and mv output expression mapping
-        LogicalAggregate<Plan> bottomAggregate = (LogicalAggregate<Plan>) planSplitContext.getBottomPlan().get(0);
-        Plan topPlan = planSplitContext.getTopPlan();
-        ExpressionMapping aggregateToTopExpressionMapping = generateAggregateToTopMapping(bottomAggregate, topPlan);
+        // get view and query aggregate and top plan correspondingly
+        Pair<Plan, LogicalAggregate<Plan>> viewTopPlanAndAggPair = splitToTopPlanAndAggregate(viewStructInfo);
+        if (viewTopPlanAndAggPair == null) {
+            return null;
+        }
+        Pair<Plan, LogicalAggregate<Plan>> queryTopPlanAndAggPair = splitToTopPlanAndAggregate(queryStructInfo);
+        if (queryTopPlanAndAggPair == null) {
+            return null;
+        }
 
-        return null;
+        // Firstly, handle query group by expression rewrite
+        LogicalAggregate<Plan> queryAggregate = queryTopPlanAndAggPair.value();
+        Plan queryTopPlan = queryTopPlanAndAggPair.key();
+        // query and view have the same dimension, try to rewrite rewrittenQueryGroupExpr
+        LogicalAggregate<Plan> viewAggregate = viewTopPlanAndAggPair.value();
+        boolean needRollUp =
+                queryAggregate.getGroupByExpressions().size() != viewAggregate.getGroupByExpressions().size();
+        if (queryAggregate.getGroupByExpressions().size() == viewAggregate.getGroupByExpressions().size()) {
+            // todo consider alias
+            List<Expression> viewGroupByExpressionQueryBased = ExpressionUtils.replace(
+                    viewAggregate.getGroupByExpressions(),
+                    queryToViewSlotMapping.inverse().toSlotReferenceMap());
+            needRollUp = !queryAggregate.getGroupByExpressions().equals(viewGroupByExpressionQueryBased);
+        }
+        if (!needRollUp) {
+            List<? extends Expression> queryShuttledExpressions = ExpressionUtils.shuttleExpressionWithLineage(
+                    queryTopPlan.getOutput(), queryTopPlan);
+            List<Expression> rewrittenQueryGroupExpr = rewriteExpression(queryShuttledExpressions,
+                    materializationContext.getMvExprToMvScanExprMapping(),
+                    queryToViewSlotMapping,
+                    true);
+            if (rewrittenQueryGroupExpr == null) {
+                // can not rewrite, bail out.
+                return null;
+            }
+            return new LogicalProject<>(
+                    rewrittenQueryGroupExpr.stream().map(NamedExpression.class::cast).collect(Collectors.toList()),
+                    tempRewritedPlan);
+        }
+        // the dimension in query and view are different, try to roll up
+        // Split query aggregate dimension and agg function List<Expression> needPullUpExpression = new ArrayList<>();
+        // Firstly, find the query top output rewriteFunctionExprList which only use query aggregate function,
+        if (viewAggregate.getOutputExpressions().stream().anyMatch(
+                viewExpr -> viewExpr.anyMatch(expr -> expr instanceof AggregateFunction
+                        && ((AggregateFunction) expr).isDistinct())
+        )) {
+            // if mv function contains distinct, can not roll up.
+            return null;
+        }
+        Set<Expression> queryAggGroupSet = new HashSet<>(queryAggregate.getGroupByExpressions());
+        List<NamedExpression> queryAggFunctions = queryAggregate.getOutputExpressions().stream()
+                .filter(expr -> !queryAggGroupSet.contains(expr))
+                .collect(Collectors.toList());
+        Set<Expression> queryAggFunctionSet = new HashSet<>(queryAggFunctions);
+        Pair<List<? extends Expression>, List<? extends Expression>> queryGroupAndFunctionPair
+                = splitToGroupAndFunction(
+                queryTopPlanAndAggPair,
+                queryAggFunctionSet);
+        // filter the expression which use the child agg function in query top plan, only support to reference the
+        // aggregate function directly, will support expression later.
+        List<? extends Expression> queryTopPlanFunctionList = queryGroupAndFunctionPair.value();
+        if (queryTopPlanFunctionList.stream().anyMatch(
+                topAggFunc -> !(topAggFunc instanceof NamedExpression)
+                        && (!queryAggFunctionSet.contains(topAggFunc)
+                        || !queryAggFunctionSet.contains(topAggFunc.child(0))))) {
+            return null;
+        }
+        // Secondly, try to roll up the agg functions and add aggregate
+        Multimap<Expression, Expression> needRollupFunctionExprMap = HashMultimap.create();
+        Map<Expression, Expression> mvExprToMvScanExprQueryBased =
+                materializationContext.getMvExprToMvScanExprMapping().keyPermute(
+                        queryToViewSlotMapping.inverse()).flattenMap().get(0);
+        for (Expression needRollUpExpr : queryTopPlanFunctionList) {
+            Expression needRollupShuttledExpr = ExpressionUtils.shuttleExpressionWithLineage(needRollUpExpr,
+                    queryTopPlan);
+            if (!mvExprToMvScanExprQueryBased.containsKey(needRollupShuttledExpr)) {
+                // function can not rewrite by view
+                return null;
+            }
+            AggregateFunction aggregateFunction = (AggregateFunction) needRollUpExpr.firstMatch(
+                    expr -> expr instanceof AggregateFunction);
+            AggregateFunction rollup = aggregateFunction.getRollup();
+            if (rollup == null) {
+                return null;
+            }
+            // key is query need roll up expr, value is mv scan based roll up expr
+            needRollupFunctionExprMap.put(needRollUpExpr,
+                    rollup.withChildren(mvExprToMvScanExprQueryBased.get(needRollupShuttledExpr)));
+        }
+        // query group rewrite
+        Multimap<Expression, Expression> groupRewrittenExprMap = HashMultimap.create();
+        List<? extends Expression> queryTopPlanGroupExprList = queryGroupAndFunctionPair.key();
+        for (Expression needRewriteGroupExpr : queryTopPlanGroupExprList) {
+            Expression queryGroupShuttledExpr =
+                    ExpressionUtils.shuttleExpressionWithLineage(needRewriteGroupExpr, queryTopPlan);
+            if (!mvExprToMvScanExprQueryBased.containsKey(queryGroupShuttledExpr)) {
+                // group expr can not rewrite by view
+                return null;
+            }
+            groupRewrittenExprMap.put(needRewriteGroupExpr, mvExprToMvScanExprQueryBased.get(queryGroupShuttledExpr));
+        }
+        // rewrite expression for group and function expr
+        List<Expression> rewriteFunctionExprList = rewriteExpression(queryTopPlanFunctionList,
+                new ExpressionMapping(needRollupFunctionExprMap),
+                queryToViewSlotMapping,
+                true);
+        if (rewriteFunctionExprList == null) {
+            return null;
+        }
+        List<Expression> rewriteGroupExprList = rewriteExpression(queryTopPlanGroupExprList,
+                new ExpressionMapping(groupRewrittenExprMap),
+                queryToViewSlotMapping,
+                true);
+        if (rewriteGroupExprList == null) {
+            return null;
+        }
+        // project rewrite
+        return new LogicalAggregate(rewriteGroupExprList, rewriteFunctionExprList, tempRewritedPlan);
     }
 
-    private ExpressionMapping generateAggregateToTopMapping(Plan source, Plan target) {
-        ImmutableMultimap.Builder<Slot, Slot> expressionMappingBuilder = ImmutableMultimap.builder();
-        List<Slot> sourceOutput = source.getOutput();
-        List<Slot> targetOutputOutput = target.getOutput();
-        for (Slot sourceSlot : sourceOutput) {
-            for (Slot targetSlot : targetOutputOutput) {
-                if (sourceSlot.equals(targetSlot)) {
-                    expressionMappingBuilder.put(targetSlot, sourceSlot);
-                }
-            }
+    private Pair<List<? extends Expression>, List<? extends Expression>> splitToGroupAndFunction(
+            Pair<Plan, LogicalAggregate<Plan>> topPlanAndAggPair,
+            Set<? extends Expression> queryAggGroupFunctionSet) {
+        Plan queryTopPlan = topPlanAndAggPair.key();
+        Map<Boolean, ? extends List<? extends Expression>> groupByAndFuncitonMap = queryTopPlan.getExpressions()
+                .stream()
+                .collect(Collectors.partitioningBy(expression -> expression.anyMatch(expr ->
+                        expr instanceof NamedExpression && queryAggGroupFunctionSet.contains((NamedExpression) expr))));
+        return Pair.of(groupByAndFuncitonMap.get(false), groupByAndFuncitonMap.get(true));
+    }
+
+    private Pair<Plan, LogicalAggregate<Plan>> splitToTopPlanAndAggregate(StructInfo structInfo) {
+        Plan topPlan = structInfo.getTopPlan();
+        PlanSplitContext splitContext = new PlanSplitContext(Sets.newHashSet(LogicalAggregate.class));
+        topPlan.accept(StructInfo.PLAN_SPLITTER, splitContext);
+        if (!(splitContext.getBottomPlan() instanceof LogicalAggregate)) {
+            return null;
+        } else {
+            return Pair.of(topPlan, (LogicalAggregate<Plan>) splitContext.getBottomPlan());
         }
-        return new ExpressionMapping(expressionMappingBuilder.build());
     }
 
     // Check Aggregate is simple or not and check join is whether valid or not.
@@ -81,20 +206,19 @@ private ExpressionMapping generateAggregateToTopMapping(Plan source, Plan target
     protected boolean checkPattern(StructInfo structInfo) {
 
         Plan topPlan = structInfo.getTopPlan();
-        Boolean valid = topPlan.accept(AGGREGATE_PATTERN_CHECKER, null);
+        Boolean valid = topPlan.accept(StructInfo.AGGREGATE_PATTERN_CHECKER, null);
         if (!valid) {
             return false;
         }
         HyperGraph hyperGraph = structInfo.getHyperGraph();
-        HashSet<JoinType> requiredJoinType = Sets.newHashSet(JoinType.INNER_JOIN, JoinType.LEFT_OUTER_JOIN);
         for (AbstractNode node : hyperGraph.getNodes()) {
             StructInfoNode structInfoNode = (StructInfoNode) node;
             if (!structInfoNode.getPlan().accept(StructInfo.JOIN_PATTERN_CHECKER,
-                    requiredJoinType)) {
+                    SUPPORTED_JOIN_TYPE_SET)) {
                 return false;
             }
             for (Edge edge : hyperGraph.getEdges()) {
-                if (!edge.getJoin().accept(StructInfo.JOIN_PATTERN_CHECKER, requiredJoinType)) {
+                if (!edge.getJoin().accept(StructInfo.JOIN_PATTERN_CHECKER, SUPPORTED_JOIN_TYPE_SET)) {
                     return false;
                 }
             }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewJoinRule.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewJoinRule.java
index e99ed3e9aa1b4d..1660d11d19b0c7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewJoinRule.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewJoinRule.java
@@ -24,14 +24,10 @@
 import org.apache.doris.nereids.rules.exploration.mv.mapping.SlotMapping;
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.NamedExpression;
-import org.apache.doris.nereids.trees.plans.JoinType;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
 import org.apache.doris.nereids.util.ExpressionUtils;
 
-import com.google.common.collect.Sets;
-
-import java.util.HashSet;
 import java.util.List;
 import java.util.stream.Collectors;
 
@@ -40,14 +36,11 @@
  * This is responsible for common join rewriting
  */
 public abstract class AbstractMaterializedViewJoinRule extends AbstractMaterializedViewRule {
-    private static final HashSet<JoinType> SUPPORTED_JOIN_TYPE_SET =
-            Sets.newHashSet(JoinType.INNER_JOIN, JoinType.LEFT_OUTER_JOIN);
-
     @Override
     protected Plan rewriteQueryByView(MatchMode matchMode,
             StructInfo queryStructInfo,
             StructInfo viewStructInfo,
-            SlotMapping queryToViewSlotMappings,
+            SlotMapping queryToViewSlotMapping,
             Plan tempRewritedPlan,
             MaterializationContext materializationContext) {
 
@@ -57,8 +50,9 @@ protected Plan rewriteQueryByView(MatchMode matchMode,
         // Rewrite top projects, represent the query projects by view
         List<Expression> expressionsRewritten = rewriteExpression(
                 queryShuttleExpression,
-                materializationContext.getViewExpressionIndexMapping(),
-                queryToViewSlotMappings
+                materializationContext.getMvExprToMvScanExprMapping(),
+                queryToViewSlotMapping,
+                true
         );
         // Can not rewrite, bail out
         if (expressionsRewritten.isEmpty()
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewRule.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewRule.java
index 76b9ff094ec4a3..4d118ff6e00e77 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewRule.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewRule.java
@@ -31,6 +31,7 @@
 import org.apache.doris.nereids.trees.expressions.SlotReference;
 import org.apache.doris.nereids.trees.expressions.literal.BooleanLiteral;
 import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.plans.JoinType;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.algebra.CatalogRelation;
 import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
@@ -52,6 +53,9 @@
  */
 public abstract class AbstractMaterializedViewRule {
 
+    public static final HashSet<JoinType> SUPPORTED_JOIN_TYPE_SET =
+            Sets.newHashSet(JoinType.INNER_JOIN, JoinType.LEFT_OUTER_JOIN);
+
     /**
      * The abstract template method for query rewrite, it contains the main logic and different query
      * pattern should override the sub logic.
@@ -123,8 +127,9 @@ protected List<Plan> rewrite(Plan queryPlan, CascadesContext cascadesContext) {
                     // Try to rewrite compensate predicates by using mv scan
                     List<Expression> rewriteCompensatePredicates = rewriteExpression(
                             compensatePredicates.toList(),
-                            materializationContext.getViewExpressionIndexMapping(),
-                            queryToViewSlotMapping);
+                            materializationContext.getMvExprToMvScanExprMapping(),
+                            queryToViewSlotMapping,
+                            true);
                     if (rewriteCompensatePredicates.isEmpty()) {
                         continue;
                     }
@@ -152,19 +157,24 @@ protected List<Plan> rewrite(Plan queryPlan, CascadesContext cascadesContext) {
     protected Plan rewriteQueryByView(MatchMode matchMode,
             StructInfo queryStructInfo,
             StructInfo viewStructInfo,
-            SlotMapping queryToViewSlotMappings,
+            SlotMapping queryToViewSlotMapping,
             Plan tempRewritedPlan,
             MaterializationContext materializationContext) {
         return tempRewritedPlan;
     }
 
     /**
-     * Use target output expression to represent the source expression
+     * Use target expression to represent the source expression.
+     * Visit the source expression, try to replace the source expression with target expression, if found then
+     * replace the source expression by target expression map value.
+     * Note: make the target expression map key to source based according to targetNeedToQueryBased,
+     * if targetNeedToQueryBased is true, we should not make it source based.
      */
     protected List<Expression> rewriteExpression(
             List<? extends Expression> sourceExpressionsToWrite,
-            ExpressionMapping mvExpressionToMvScanExpressionMapping,
-            SlotMapping sourceToTargetMapping) {
+            ExpressionMapping targetExpressionMapping,
+            SlotMapping sourceToTargetMapping,
+            boolean targetNeedToQueryBased) {
         // Firstly, rewrite the target plan output expression using query with inverse mapping
         // then try to use the mv expression to represent the query. if any of source expressions
         // can not be represented by mv, return null
@@ -177,10 +187,10 @@ protected List<Expression> rewriteExpression(
         //     transform source to:
         //        project(slot 2, 1)
         //            target
-        // generate mvSql to mvScan mvExpressionToMvScanExpressionMapping, and change mv sql expression to query based
-        ExpressionMapping expressionMappingKeySourceBased =
-                mvExpressionToMvScanExpressionMapping.keyPermute(sourceToTargetMapping.inverse());
-        List<Map<? extends Expression, ? extends Expression>> flattenExpressionMap =
+        // generate mvSql to mvScan targetExpressionMapping, and change mv sql expression to query based
+        ExpressionMapping expressionMappingKeySourceBased = targetNeedToQueryBased
+                ? targetExpressionMapping : targetExpressionMapping.keyPermute(sourceToTargetMapping.inverse());
+        List<Map<Expression, Expression>> flattenExpressionMap =
                 expressionMappingKeySourceBased.flattenMap();
         // view to view scan expression is 1:1 so get first element
         Map<? extends Expression, ? extends Expression> mvSqlToMvScanMappingQueryBased = flattenExpressionMap.get(0);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/MaterializationContext.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/MaterializationContext.java
index 336c627da66574..9de2e9ed16f320 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/MaterializationContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/MaterializationContext.java
@@ -47,7 +47,7 @@ public class MaterializationContext {
     // Group ids that are rewritten by this mv to reduce rewrite times
     private final Set<GroupId> matchedGroups = new HashSet<>();
     // generate form mv scan plan
-    private ExpressionMapping viewExpressionMapping;
+    private ExpressionMapping mvExprToMvScanExprMapping;
 
     /**
      * MaterializationContext, this contains necessary info for query rewriting by mv
@@ -73,7 +73,7 @@ public MaterializationContext(MTMV mtmv,
                 ExpressionUtils.shuttleExpressionWithLineage(mvOutputExpressions, mvCache.getLogicalPlan()).stream()
                         .map(NamedExpression.class::cast)
                         .collect(Collectors.toList());
-        this.viewExpressionMapping = ExpressionMapping.generate(
+        this.mvExprToMvScanExprMapping = ExpressionMapping.generate(
                 mvOutputExpressions,
                 mvScanPlan.getExpressions());
     }
@@ -106,8 +106,8 @@ public List<Table> getBaseViews() {
         return baseViews;
     }
 
-    public ExpressionMapping getViewExpressionIndexMapping() {
-        return viewExpressionMapping;
+    public ExpressionMapping getMvExprToMvScanExprMapping() {
+        return mvExprToMvScanExprMapping;
     }
 
     /**
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/StructInfo.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/StructInfo.java
index 36d7b8d0b51e6d..6c7ff8b9c034d8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/StructInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/StructInfo.java
@@ -32,9 +32,11 @@
 import org.apache.doris.nereids.trees.plans.algebra.Filter;
 import org.apache.doris.nereids.trees.plans.algebra.Join;
 import org.apache.doris.nereids.trees.plans.algebra.Project;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
 import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
 import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
 import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRepeat;
 import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
 import org.apache.doris.nereids.util.ExpressionUtils;
 
@@ -46,6 +48,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
@@ -55,6 +58,7 @@
  */
 public class StructInfo {
     public static final JoinPatternChecker JOIN_PATTERN_CHECKER = new JoinPatternChecker();
+    public static final AggregatePatternChecker AGGREGATE_PATTERN_CHECKER = new AggregatePatternChecker();
     // struct info splitter
     public static final PlanSplitter PLAN_SPLITTER = new PlanSplitter();
     private static final RelationCollector RELATION_COLLECTOR = new RelationCollector();
@@ -166,6 +170,7 @@ public static List<StructInfo> of(Plan originalPlan) {
         // TODO only consider the inner join currently, Should support outer join
         // Split plan by the boundary which contains multi child
         PlanSplitContext planSplitContext = new PlanSplitContext(Sets.newHashSet(LogicalJoin.class));
+        // if single table without join, the bottom is
         originalPlan.accept(PLAN_SPLITTER, planSplitContext);
 
         List<HyperGraph> structInfos = HyperGraph.toStructInfo(planSplitContext.getBottomPlan());
@@ -274,6 +279,10 @@ public Void visit(Plan plan, PlanSplitContext context) {
             if (context.getTopPlan() == null) {
                 context.setTopPlan(plan);
             }
+            if (plan.children().isEmpty() && context.getBottomPlan() == null) {
+                context.setBottomPlan(plan);
+                return null;
+            }
             if (context.isBoundary(plan)) {
                 context.setBottomPlan(plan);
                 return null;
@@ -348,4 +357,28 @@ public Boolean visit(Plan plan, Set<JoinType> requiredJoinType) {
             return true;
         }
     }
+
+    /**
+     * AggregatePatternChecker
+     */
+    public static class AggregatePatternChecker extends DefaultPlanVisitor<Boolean, Void> {
+        @Override
+        public Boolean visit(Plan plan, Void context) {
+            if (plan instanceof LogicalAggregate) {
+                LogicalAggregate<Plan> aggregate = (LogicalAggregate<Plan>) plan;
+                Optional<LogicalRepeat<?>> sourceRepeat = aggregate.getSourceRepeat();
+                if (sourceRepeat.isPresent()) {
+                    return false;
+                }
+                super.visit(aggregate, context);
+                return true;
+            }
+            if (plan instanceof LogicalProject) {
+                super.visit(plan, context);
+                return true;
+            }
+            super.visit(plan, context);
+            return false;
+        }
+    }
 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/mapping/ExpressionIndexMapping.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/mapping/ExpressionIndexMapping.java
deleted file mode 100644
index f63017633a82aa..00000000000000
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/mapping/ExpressionIndexMapping.java
+++ /dev/null
@@ -1,48 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements.  See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership.  The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License.  You may obtain a copy of the License at
-//
-//   http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied.  See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-package org.apache.doris.nereids.rules.exploration.mv.mapping;
-
-import org.apache.doris.nereids.trees.expressions.Expression;
-
-import com.google.common.collect.ArrayListMultimap;
-import com.google.common.collect.Multimap;
-
-import java.util.List;
-
-/**
- * Expression and it's index mapping
- */
-public class ExpressionIndexMapping extends Mapping {
-    private final Multimap<Expression, Integer> expressionIndexMapping;
-
-    public ExpressionIndexMapping(Multimap<Expression, Integer> expressionIndexMapping) {
-        this.expressionIndexMapping = expressionIndexMapping;
-    }
-
-    public Multimap<Expression, Integer> getExpressionIndexMapping() {
-        return expressionIndexMapping;
-    }
-
-    public static ExpressionIndexMapping generate(List<? extends Expression> expressions) {
-        Multimap<Expression, Integer> expressionIndexMapping = ArrayListMultimap.create();
-        for (int i = 0; i < expressions.size(); i++) {
-            expressionIndexMapping.put(expressions.get(i), i);
-        }
-        return new ExpressionIndexMapping(expressionIndexMapping);
-    }
-}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/mapping/ExpressionMapping.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/mapping/ExpressionMapping.java
index 7c1f06746cdb3c..2f1ed230145f35 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/mapping/ExpressionMapping.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/mapping/ExpressionMapping.java
@@ -22,6 +22,7 @@
 import org.apache.doris.nereids.util.ExpressionUtils;
 
 import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.ImmutableMultimap;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Multimap;
 
@@ -30,25 +31,26 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 
 /**
  * Expression mapping, maybe one expression map to multi expression
  */
 public class ExpressionMapping extends Mapping {
-    private final Multimap<? extends Expression, ? extends Expression> expressionMapping;
+    private final Multimap<Expression, Expression> expressionMapping;
 
-    public ExpressionMapping(Multimap<? extends Expression, ? extends Expression> expressionMapping) {
+    public ExpressionMapping(Multimap<Expression, Expression> expressionMapping) {
         this.expressionMapping = expressionMapping;
     }
 
-    public Multimap<? extends Expression, ? extends Expression> getExpressionMapping() {
+    public Multimap<Expression, Expression> getExpressionMapping() {
         return expressionMapping;
     }
 
     /**
      * ExpressionMapping flatten
      */
-    public List<Map<? extends Expression, ? extends Expression>> flattenMap() {
+    public List<Map<Expression, Expression>> flattenMap() {
         List<List<Pair<Expression, Expression>>> tmpExpressionPairs = new ArrayList<>(this.expressionMapping.size());
         Map<? extends Expression, ? extends Collection<? extends Expression>> expressionMappingMap =
                 expressionMapping.asMap();
@@ -62,7 +64,7 @@ public ExpressionMapping(Multimap<? extends Expression, ? extends Expression> ex
         }
         List<List<Pair<Expression, Expression>>> cartesianExpressionMap = Lists.cartesianProduct(tmpExpressionPairs);
 
-        final List<Map<? extends Expression, ? extends Expression>> flattenedMap = new ArrayList<>();
+        final List<Map<Expression, Expression>> flattenedMap = new ArrayList<>();
         for (List<Pair<Expression, Expression>> listPair : cartesianExpressionMap) {
             final Map<Expression, Expression> expressionMap = new HashMap<>();
             listPair.forEach(pair -> expressionMap.put(pair.key(), pair.value()));
@@ -71,7 +73,8 @@ public ExpressionMapping(Multimap<? extends Expression, ? extends Expression> ex
         return flattenedMap;
     }
 
-    /**Permute the key of expression mapping. this is useful for expression rewrite, if permute key to query based
+    /**
+     * Permute the key of expression mapping. this is useful for expression rewrite, if permute key to query based
      * then when expression rewrite success, we can get the mv scan expression directly.
      */
     public ExpressionMapping keyPermute(SlotMapping slotMapping) {
@@ -86,7 +89,9 @@ public ExpressionMapping keyPermute(SlotMapping slotMapping) {
         return new ExpressionMapping(permutedExpressionMapping);
     }
 
-    /**ExpressionMapping generate*/
+    /**
+     * ExpressionMapping generate
+     */
     public static ExpressionMapping generate(
             List<? extends Expression> sourceExpressions,
             List<? extends Expression> targetExpressions) {
@@ -97,4 +102,25 @@ public static ExpressionMapping generate(
         }
         return new ExpressionMapping(expressionMultiMap);
     }
+
+    @Override
+    public Mapping chainedFold(Mapping target) {
+
+        ImmutableMultimap.Builder<Expression, Expression> foldedMappingBuilder =
+                ImmutableMultimap.builder();
+
+        Multimap<Expression, Expression> targetMapping
+                = ((ExpressionMapping) target).getExpressionMapping();
+        for (Entry<Expression, ? extends Collection<Expression>> exprMapping :
+                this.getExpressionMapping().asMap().entrySet()) {
+            Collection<? extends Expression> valueExpressions = exprMapping.getValue();
+            valueExpressions.forEach(valueExpr -> {
+                if (targetMapping.containsKey(valueExpr)) {
+                    targetMapping.get(valueExpr).forEach(
+                            targetValue -> foldedMappingBuilder.put(exprMapping.getKey(), targetValue));
+                }
+            });
+        }
+        return new ExpressionMapping(foldedMappingBuilder.build());
+    }
 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/mapping/Mapping.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/mapping/Mapping.java
index 17a412dab10d39..3d9f95a5049dc0 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/mapping/Mapping.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/mapping/Mapping.java
@@ -136,4 +136,12 @@ public int hashCode() {
             return Objects.hash(exprId);
         }
     }
+
+    /** Chain fold tow mapping, such as this mapping is {[a -> b]}, the target mapping is
+     *  {[b -> c]} after chain fold, this result will be {[a -> c]}, if the value side in this mapping
+     *  can get the key in the target mapping, will lose the mapping
+     */
+    protected Mapping chainedFold(Mapping target) {
+        return null;
+    }
 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/TreeNode.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/TreeNode.java
index 694f0611567ec8..3d2f3b10e7ef44 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/TreeNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/TreeNode.java
@@ -185,6 +185,23 @@ default boolean anyMatch(Predicate<TreeNode<NODE_TYPE>> predicate) {
         return false;
     }
 
+    /**
+     * iterate top down and test predicate if any matched. Top-down traverse implicitly.
+     * @param predicate predicate
+     * @return true if all predicate return true
+     */
+    default TreeNode<NODE_TYPE> firstMatch(Predicate<TreeNode<NODE_TYPE>> predicate) {
+        if (!predicate.test(this)) {
+            return this;
+        }
+        for (NODE_TYPE child : children()) {
+            if (!child.anyMatch(predicate)) {
+                return this;
+            }
+        }
+        return this;
+    }
+
     /**
      * Collect the nodes that satisfied the predicate.
      */
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/AggregateFunction.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/AggregateFunction.java
index a7e523dfdb549e..7267d457d8caa4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/AggregateFunction.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/AggregateFunction.java
@@ -77,6 +77,10 @@ public boolean isDistinct() {
         return distinct;
     }
 
+    public AggregateFunction getRollup() {
+        return null;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/Sum.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/Sum.java
index b8ee59c81f059a..976c489637631e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/Sum.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/Sum.java
@@ -109,4 +109,9 @@ public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) {
     public List<FunctionSignature> getSignatures() {
         return SIGNATURES;
     }
+
+    @Override
+    public AggregateFunction getRollup() {
+        return this;
+    }
 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/util/ExpressionUtils.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/util/ExpressionUtils.java
index 24e8a055b5b412..2f00313cde92cb 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/util/ExpressionUtils.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/util/ExpressionUtils.java
@@ -207,6 +207,11 @@ public static Expression combine(Class<? extends Expression> type, Collection<Ex
                 .orElse(BooleanLiteral.of(type == And.class));
     }
 
+    public static Expression shuttleExpressionWithLineage(Expression expression, Plan plan) {
+        return shuttleExpressionWithLineage(Lists.newArrayList(expression),
+                plan, ImmutableSet.of(), ImmutableSet.of()).get(0);
+    }
+
     public static List<? extends Expression> shuttleExpressionWithLineage(List<? extends Expression> expressions,
             Plan plan) {
         return shuttleExpressionWithLineage(expressions, plan, ImmutableSet.of(), ImmutableSet.of());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
index fff135baa850b3..e27ac08fbb2c13 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
@@ -122,7 +122,6 @@
 import org.apache.doris.nereids.minidump.MinidumpUtils;
 import org.apache.doris.nereids.parser.NereidsParser;
 import org.apache.doris.nereids.rules.exploration.mv.InitMaterializationContextHook;
-import org.apache.doris.nereids.stats.StatsErrorEstimator;
 import org.apache.doris.nereids.trees.plans.commands.BatchInsertIntoTableCommand;
 import org.apache.doris.nereids.trees.plans.commands.Command;
 import org.apache.doris.nereids.trees.plans.commands.CreateTableCommand;