[Improvement](nereids) Make sure to catch and record exception for every materialization context (#29953)

1. Make sure instance when change params of StructInfo,Predicates.
2. Catch and record exception for every materialization context, this make sure that if throw exception when one materialization context rewrite, it will not influence others.
3. Support to mv rewrite when hava count function when aggregate without group by

fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVCache.java

@@ -63,7 +63,9 @@ public class MTMVCache {
         StatementContext mvSqlStatementContext = new StatementContext(connectContext,
                 new OriginStatement(mtmv.getQuerySql(), 0));
         NereidsPlanner planner = new NereidsPlanner(mvSqlStatementContext);
-
+        if (mvSqlStatementContext.getConnectContext().getStatementContext() == null) {
+            mvSqlStatementContext.getConnectContext().setStatementContext(mvSqlStatementContext);
+        }
         Plan mvRewrittenPlan =
                 planner.plan(unboundMvPlan, PhysicalProperties.ANY, ExplainLevel.REWRITTEN_PLAN);
         Plan mvPlan = mvRewrittenPlan instanceof LogicalResultSink

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewAggregateRule.java

@@ -23,8 +23,8 @@ import org.apache.doris.nereids.jobs.joinorder.hypergraph.edge.JoinEdge;
 import org.apache.doris.nereids.jobs.joinorder.hypergraph.node.AbstractNode;
 import org.apache.doris.nereids.jobs.joinorder.hypergraph.node.StructInfoNode;
 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.Alias;
 import org.apache.doris.nereids.trees.expressions.Any;
 import org.apache.doris.nereids.trees.expressions.Cast;
 import org.apache.doris.nereids.trees.expressions.ExprId;
@@ -45,7 +45,6 @@ import org.apache.doris.nereids.types.BigIntType;
 import org.apache.doris.nereids.util.ExpressionUtils;
 
 import com.google.common.collect.ArrayListMultimap;
-import com.google.common.collect.HashMultimap;
 import com.google.common.collect.Multimap;
 import com.google.common.collect.Sets;
 
@@ -64,7 +63,7 @@ import java.util.stream.Collectors;
  */
 public abstract class AbstractMaterializedViewAggregateRule extends AbstractMaterializedViewRule {
 
-    protected static final Multimap<Expression, Expression>
+    protected static final Multimap<Function, Expression>
             AGGREGATE_ROLL_UP_EQUIVALENT_FUNCTION_MAP = ArrayListMultimap.create();
 
     static {
@@ -95,7 +94,7 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
     protected Plan rewriteQueryByView(MatchMode matchMode,
             StructInfo queryStructInfo,
             StructInfo viewStructInfo,
-            SlotMapping queryToViewSlotMapping,
+            SlotMapping viewToQuerySlotMapping,
             Plan tempRewritedPlan,
             MaterializationContext materializationContext) {
         // get view and query aggregate and top plan correspondingly
@@ -115,12 +114,11 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
         }
         // Firstly,if group by expression between query and view is equals, try to rewrite expression directly
         Plan queryTopPlan = queryTopPlanAndAggPair.key();
-        SlotMapping viewToQurySlotMapping = queryToViewSlotMapping.inverse();
-        if (isGroupByEquals(queryTopPlanAndAggPair, viewTopPlanAndAggPair, viewToQurySlotMapping)) {
+        if (isGroupByEquals(queryTopPlanAndAggPair, viewTopPlanAndAggPair, viewToQuerySlotMapping)) {
             List<Expression> rewrittenQueryExpressions = rewriteExpression(queryTopPlan.getExpressions(),
                     queryTopPlan,
                     materializationContext.getMvExprToMvScanExprMapping(),
-                    queryToViewSlotMapping,
+                    viewToQuerySlotMapping,
                     true);
             if (!rewrittenQueryExpressions.isEmpty()) {
                 return new LogicalProject<>(
@@ -133,15 +131,17 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
             materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
                     Pair.of("Can not rewrite expression when no roll up",
                             String.format("expressionToWrite = %s,\n mvExprToMvScanExprMapping = %s,\n"
-                                            + "queryToViewSlotMapping = %s",
+                                            + "viewToQuerySlotMapping = %s",
                                     queryTopPlan.getExpressions(),
                                     materializationContext.getMvExprToMvScanExprMapping(),
-                                    queryToViewSlotMapping)));
+                                    viewToQuerySlotMapping)));
         }
         // if view is scalar aggregate but query is not. Or if query is scalar aggregate but view is not
         // Should not rewrite
-        if (queryTopPlanAndAggPair.value().getGroupByExpressions().isEmpty()
-                || viewTopPlanAndAggPair.value().getGroupByExpressions().isEmpty()) {
+        List<Expression> queryGroupByExpressions = queryTopPlanAndAggPair.value().getGroupByExpressions();
+        List<Expression> viewGroupByExpressions = viewTopPlanAndAggPair.value().getGroupByExpressions();
+        if ((queryGroupByExpressions.isEmpty() && !viewGroupByExpressions.isEmpty())
+                || (!queryGroupByExpressions.isEmpty() && viewGroupByExpressions.isEmpty())) {
             materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
                     Pair.of("only one the of query or view is scalar aggregate and "
                                     + "can not rewrite expression meanwhile",
@@ -154,53 +154,42 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
         // split the query top plan expressions to group expressions and functions, if can not, bail out.
         Pair<Set<? extends Expression>, Set<? extends Expression>> queryGroupAndFunctionPair
                 = topPlanSplitToGroupAndFunction(queryTopPlanAndAggPair);
-        // this map will be used to rewrite expression
-        Multimap<Expression, Expression> needRollupExprMap = HashMultimap.create();
-        Multimap<Expression, Expression> groupRewrittenExprMap = HashMultimap.create();
-        // permute the mv expr mapping to query based
-        Map<Expression, Expression> mvExprToMvScanExprQueryBased =
-                materializationContext.getMvExprToMvScanExprMapping().keyPermute(viewToQurySlotMapping)
-                        .flattenMap().get(0);
         Set<? extends Expression> queryTopPlanFunctionSet = queryGroupAndFunctionPair.value();
         // try to rewrite, contains both roll up aggregate functions and aggregate group expression
         List<NamedExpression> finalAggregateExpressions = new ArrayList<>();
         List<Expression> finalGroupExpressions = new ArrayList<>();
-        for (Expression topExpression : queryTopPlan.getExpressions()) {
+        List<? extends Expression> queryExpressions = queryTopPlan.getExpressions();
+        // permute the mv expr mapping to query based
+        Map<Expression, Expression> mvExprToMvScanExprQueryBased =
+                materializationContext.getMvExprToMvScanExprMapping().keyPermute(viewToQuerySlotMapping)
+                        .flattenMap().get(0);
+        for (Expression topExpression : queryExpressions) {
             // if agg function, try to roll up and rewrite
             if (queryTopPlanFunctionSet.contains(topExpression)) {
                 Expression queryFunctionShuttled = ExpressionUtils.shuttleExpressionWithLineage(
                         topExpression,
                         queryTopPlan);
                 // try to roll up
-                AggregateFunction queryFunction = (AggregateFunction) queryFunctionShuttled.firstMatch(
-                        expr -> expr instanceof AggregateFunction);
-                Function rollupAggregateFunction = rollup(queryFunction, queryFunctionShuttled,
-                        mvExprToMvScanExprQueryBased);
+                List<Object> queryFunctions =
+                        queryFunctionShuttled.collectFirst(expr -> expr instanceof AggregateFunction);
+                if (queryFunctions.isEmpty()) {
+                    materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
+                            Pair.of("Can not found query function",
+                                    String.format("queryFunctionShuttled = %s", queryFunctionShuttled)));
+                    return null;
+                }
+                Function rollupAggregateFunction = rollup((AggregateFunction) queryFunctions.get(0),
+                        queryFunctionShuttled, mvExprToMvScanExprQueryBased);
                 if (rollupAggregateFunction == null) {
                     materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
                             Pair.of("Query function roll up fail",
                                     String.format("queryFunction = %s,\n queryFunctionShuttled = %s,\n"
                                                     + "mvExprToMvScanExprQueryBased = %s",
-                                            queryFunction, queryFunctionShuttled, mvExprToMvScanExprQueryBased)));
+                                            queryFunctions.get(0), queryFunctionShuttled,
+                                            mvExprToMvScanExprQueryBased)));
                     return null;
                 }
-                // key is query need roll up expr, value is mv scan based roll up expr
-                needRollupExprMap.put(queryFunctionShuttled, rollupAggregateFunction);
-                // rewrite query function expression by mv expression
-                ExpressionMapping needRollupExprMapping = new ExpressionMapping(needRollupExprMap);
-                Expression rewrittenFunctionExpression = rewriteExpression(topExpression,
-                        queryTopPlan,
-                        needRollupExprMapping,
-                        queryToViewSlotMapping,
-                        false);
-                if (rewrittenFunctionExpression == null) {
-                    materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
-                            Pair.of("Roll up expression can not rewrite by view", String.format(
-                                    "topExpression = %s,\n needRollupExprMapping = %s,\n queryToViewSlotMapping = %s",
-                                    topExpression, needRollupExprMapping, queryToViewSlotMapping)));
-                    return null;
-                }
-                finalAggregateExpressions.add((NamedExpression) rewrittenFunctionExpression);
+                finalAggregateExpressions.add(new Alias(rollupAggregateFunction));
             } else {
                 // if group by expression, try to rewrite group by expression
                 Expression queryGroupShuttledExpr =
@@ -213,26 +202,9 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
                                             mvExprToMvScanExprQueryBased, queryGroupShuttledExpr)));
                     return null;
                 }
-                groupRewrittenExprMap.put(queryGroupShuttledExpr,
-                        mvExprToMvScanExprQueryBased.get(queryGroupShuttledExpr));
-                // rewrite query group expression by mv expression
-                ExpressionMapping groupRewrittenExprMapping = new ExpressionMapping(groupRewrittenExprMap);
-                Expression rewrittenGroupExpression = rewriteExpression(
-                        topExpression,
-                        queryTopPlan,
-                        groupRewrittenExprMapping,
-                        queryToViewSlotMapping,
-                        true);
-                if (rewrittenGroupExpression == null) {
-                    materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
-                            Pair.of("Query dimensions can not be rewritten by view",
-                                    String.format("topExpression is %s,\n groupRewrittenExprMapping is %s,\n"
-                                                    + "queryToViewSlotMapping = %s",
-                                            topExpression, groupRewrittenExprMapping, queryToViewSlotMapping)));
-                    return null;
-                }
-                finalAggregateExpressions.add((NamedExpression) rewrittenGroupExpression);
-                finalGroupExpressions.add(rewrittenGroupExpression);
+                Expression expression = mvExprToMvScanExprQueryBased.get(queryGroupShuttledExpr);
+                finalAggregateExpressions.add((NamedExpression) expression);
+                finalGroupExpressions.add(expression);
             }
         }
         // add project to guarantee group by column ref is slot reference,
@@ -271,7 +243,7 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
 
     private boolean isGroupByEquals(Pair<Plan, LogicalAggregate<Plan>> queryTopPlanAndAggPair,
             Pair<Plan, LogicalAggregate<Plan>> viewTopPlanAndAggPair,
-            SlotMapping viewToQurySlotMapping) {
+            SlotMapping viewToQuerySlotMapping) {
         Plan queryTopPlan = queryTopPlanAndAggPair.key();
         Plan viewTopPlan = viewTopPlanAndAggPair.key();
         LogicalAggregate<Plan> queryAggregate = queryTopPlanAndAggPair.value();
@@ -282,7 +254,7 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
         Set<? extends Expression> viewGroupShuttledExpressionQueryBased = ExpressionUtils.shuttleExpressionWithLineage(
                         viewAggregate.getGroupByExpressions(), viewTopPlan)
                 .stream()
-                .map(expr -> ExpressionUtils.replace(expr, viewToQurySlotMapping.toSlotReferenceMap()))
+                .map(expr -> ExpressionUtils.replace(expr, viewToQuerySlotMapping.toSlotReferenceMap()))
                 .collect(Collectors.toSet());
         return queryGroupShuttledExpression.equals(viewGroupShuttledExpressionQueryBased);
     }
@@ -309,19 +281,20 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
         }
         Expression rollupParam = null;
         Expression viewRollupFunction = null;
-        if (mvExprToMvScanExprQueryBased.containsKey(queryAggregateFunctionShuttled)) {
-            // function can rewrite by view
+        // handle simple aggregate function roll up which is not in the AGGREGATE_ROLL_UP_EQUIVALENT_FUNCTION_MAP
+        if (mvExprToMvScanExprQueryBased.containsKey(queryAggregateFunctionShuttled)
+                && AGGREGATE_ROLL_UP_EQUIVALENT_FUNCTION_MAP.keySet().stream()
+                .noneMatch(aggFunction -> aggFunction.equals(queryAggregateFunction))) {
             rollupParam = mvExprToMvScanExprQueryBased.get(queryAggregateFunctionShuttled);
             viewRollupFunction = queryAggregateFunctionShuttled;
         } else {
-            // function can not rewrite by view, try to use complex roll up param
+            // handle complex functions roll up
             // eg: query is count(distinct param), mv sql is bitmap_union(to_bitmap(param))
             for (Expression mvExprShuttled : mvExprToMvScanExprQueryBased.keySet()) {
                 if (!(mvExprShuttled instanceof Function)) {
                     continue;
                 }
-                if (isAggregateFunctionEquivalent(queryAggregateFunction, queryAggregateFunctionShuttled,
-                        (Function) mvExprShuttled)) {
+                if (isAggregateFunctionEquivalent(queryAggregateFunction, (Function) mvExprShuttled)) {
                     rollupParam = mvExprToMvScanExprQueryBased.get(mvExprShuttled);
                     viewRollupFunction = mvExprShuttled;
                 }
@@ -429,13 +402,12 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
      * This will check the count(distinct a) in query is equivalent to  bitmap_union(to_bitmap(a)) in mv,
      * and then check their arguments is equivalent.
      */
-    private boolean isAggregateFunctionEquivalent(Function queryFunction, Expression queryFunctionShuttled,
-            Function viewFunction) {
+    private boolean isAggregateFunctionEquivalent(Function queryFunction, Function viewFunction) {
         if (queryFunction.equals(viewFunction)) {
             return true;
         }
         // check the argument of rollup function is equivalent to view function or not
-        for (Map.Entry<Expression, Collection<Expression>> equivalentFunctionEntry :
+        for (Map.Entry<Function, Collection<Expression>> equivalentFunctionEntry :
                 AGGREGATE_ROLL_UP_EQUIVALENT_FUNCTION_MAP.asMap().entrySet()) {
             if (equivalentFunctionEntry.getKey().equals(queryFunction)) {
                 // check is have equivalent function or not

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewJoinRule.java

@@ -23,6 +23,7 @@ import org.apache.doris.nereids.jobs.joinorder.hypergraph.edge.JoinEdge;
 import org.apache.doris.nereids.jobs.joinorder.hypergraph.node.AbstractNode;
 import org.apache.doris.nereids.jobs.joinorder.hypergraph.node.StructInfoNode;
 import org.apache.doris.nereids.rules.exploration.mv.mapping.SlotMapping;
+import org.apache.doris.nereids.trees.expressions.Alias;
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.NamedExpression;
 import org.apache.doris.nereids.trees.plans.Plan;
@@ -47,7 +48,7 @@ public abstract class AbstractMaterializedViewJoinRule extends AbstractMateriali
     protected Plan rewriteQueryByView(MatchMode matchMode,
             StructInfo queryStructInfo,
             StructInfo viewStructInfo,
-            SlotMapping queryToViewSlotMapping,
+            SlotMapping targetToSourceMapping,
             Plan tempRewritedPlan,
             MaterializationContext materializationContext) {
         // Rewrite top projects, represent the query projects by view
@@ -55,19 +56,18 @@ public abstract class AbstractMaterializedViewJoinRule extends AbstractMateriali
                 queryStructInfo.getExpressions(),
                 queryStructInfo.getOriginalPlan(),
                 materializationContext.getMvExprToMvScanExprMapping(),
-                queryToViewSlotMapping,
+                targetToSourceMapping,
                 true
         );
         // Can not rewrite, bail out
-        if (expressionsRewritten.isEmpty()
-                || expressionsRewritten.stream().anyMatch(expr -> !(expr instanceof NamedExpression))) {
+        if (expressionsRewritten.isEmpty()) {
             materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
                     Pair.of("Rewrite expressions by view in join fail",
                             String.format("expressionToRewritten is %s,\n mvExprToMvScanExprMapping is %s,\n"
-                                            + "queryToViewSlotMapping = %s",
+                                            + "targetToSourceMapping = %s",
                                     queryStructInfo.getExpressions(),
                                     materializationContext.getMvExprToMvScanExprMapping(),
-                                    queryToViewSlotMapping)));
+                                    targetToSourceMapping)));
             return null;
         }
         // record the group id in materializationContext, and when rewrite again in
@@ -77,7 +77,10 @@ public abstract class AbstractMaterializedViewJoinRule extends AbstractMateriali
                     queryStructInfo.getOriginalPlan().getGroupExpression().get().getOwnerGroup().getGroupId());
         }
         return new LogicalProject<>(
-                expressionsRewritten.stream().map(NamedExpression.class::cast).collect(Collectors.toList()),
+                expressionsRewritten.stream()
+                        .map(expression -> expression instanceof NamedExpression ? expression : new Alias(expression))
+                        .map(NamedExpression.class::cast)
+                        .collect(Collectors.toList()),
                 tempRewritedPlan);
     }
 

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewRule.java

@@ -48,6 +48,7 @@ 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;
 import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
 import org.apache.doris.nereids.util.ExpressionUtils;
 import org.apache.doris.nereids.util.TypeUtils;
 
@@ -72,163 +73,222 @@ public abstract class AbstractMaterializedViewRule implements ExplorationRuleFac
             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.
+     * The abstract template method for query rewrite, it contains the main logic, try to rewrite query by
+     * multi materialization every time. if exception it will catch the exception and record it to
+     * materialization context.
      */
-    protected List<Plan> rewrite(Plan queryPlan, CascadesContext cascadesContext) {
+    public List<Plan> rewrite(Plan queryPlan, CascadesContext cascadesContext) {
+        List<Plan> rewrittenPlans = new ArrayList<>();
+        // already rewrite or query is invalid, bail out
+        List<StructInfo> queryStructInfos = checkQuery(queryPlan, cascadesContext);
+        if (queryStructInfos.isEmpty()) {
+            return rewrittenPlans;
+        }
+        for (MaterializationContext context : cascadesContext.getMaterializationContexts()) {
+            if (checkIfRewritten(queryPlan, context)) {
+                continue;
+            }
+            // TODO Just support only one query struct info, support multi later.
+            StructInfo queryStructInfo = queryStructInfos.get(0);
+            try {
+                rewrittenPlans.addAll(doRewrite(queryStructInfo, cascadesContext, context));
+            } catch (Exception exception) {
+                context.recordFailReason(queryStructInfo.getOriginalPlanId(),
+                        Pair.of("Materialized view rule exec fail", exception.toString()));
+            }
+        }
+        return rewrittenPlans;
+    }
+
+    /**
+     * Check query is valid or not, if valid return the query struct infos, if invalid return empty list.
+     */
+    protected List<StructInfo> checkQuery(Plan queryPlan, CascadesContext cascadesContext) {
+        List<StructInfo> validQueryStructInfos = new ArrayList<>();
         List<MaterializationContext> materializationContexts = cascadesContext.getMaterializationContexts();
-        List<Plan> rewriteResults = new ArrayList<>();
         if (materializationContexts.isEmpty()) {
-            return rewriteResults;
+            return validQueryStructInfos;
         }
         List<StructInfo> queryStructInfos = MaterializedViewUtils.extractStructInfo(queryPlan, cascadesContext);
         // TODO Just Check query queryPlan firstly, support multi later.
         StructInfo queryStructInfo = queryStructInfos.get(0);
         if (!checkPattern(queryStructInfo)) {
-            materializationContexts.forEach(ctx -> ctx.recordFailReason(queryStructInfo.getOriginalPlanId(),
-                    Pair.of("Query struct info is invalid",
-                            String.format("queryPlan is %s", queryPlan.treeString()))));
+            cascadesContext.getMaterializationContexts().forEach(ctx ->
+                    ctx.recordFailReason(queryStructInfo.getOriginalPlanId(),
+                            Pair.of("Query struct info is invalid",
+                                    String.format("queryPlan is %s", queryPlan.treeString())))
+            );
+            return validQueryStructInfos;
+        }
+        validQueryStructInfos.add(queryStructInfo);
+        return validQueryStructInfos;
+    }
+
+    /**
+     * The abstract template method for query rewrite, it contains the main logic, try to rewrite query by
+     * only one materialization every time. Different query pattern should override the sub logic.
+     */
+    protected List<Plan> doRewrite(StructInfo queryStructInfo, CascadesContext cascadesContext,
+            MaterializationContext materializationContext) {
+        List<Plan> rewriteResults = new ArrayList<>();
+        List<StructInfo> viewStructInfos = MaterializedViewUtils.extractStructInfo(
+                materializationContext.getMvPlan(), cascadesContext);
+        if (viewStructInfos.size() > 1) {
+            // view struct info should only have one
+            materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
+                    Pair.of("The num of view struct info is more then one",
+                            String.format("mv plan is %s", materializationContext.getMvPlan().treeString())));
             return rewriteResults;
         }
-        for (MaterializationContext materializationContext : materializationContexts) {
-            // already rewrite, bail out
-            if (checkIfRewritten(queryPlan, materializationContext)) {
+        StructInfo viewStructInfo = viewStructInfos.get(0);
+        if (!checkPattern(viewStructInfo)) {
+            materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
+                    Pair.of("View struct info is invalid",
+                            String.format(", view plan is %s", viewStructInfo.getOriginalPlan().treeString())));
+            return rewriteResults;
+        }
+        MatchMode matchMode = decideMatchMode(queryStructInfo.getRelations(), viewStructInfo.getRelations());
+        if (MatchMode.COMPLETE != matchMode) {
+            materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
+                    Pair.of("Match mode is invalid", String.format("matchMode is %s", matchMode)));
+            return rewriteResults;
+        }
+        List<RelationMapping> queryToViewTableMappings = RelationMapping.generate(queryStructInfo.getRelations(),
+                viewStructInfo.getRelations());
+        // if any relation in query and view can not map, bail out.
+        if (queryToViewTableMappings == null) {
+            materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
+                    Pair.of("Query to view table mapping is null", ""));
+            return rewriteResults;
+        }
+        for (RelationMapping queryToViewTableMapping : queryToViewTableMappings) {
+            SlotMapping queryToViewSlotMapping = SlotMapping.generate(queryToViewTableMapping);
+            if (queryToViewSlotMapping == null) {
+                materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
+                        Pair.of("Query to view slot mapping is null", ""));
                 continue;
             }
-            List<StructInfo> viewStructInfos = MaterializedViewUtils.extractStructInfo(
-                    materializationContext.getMvPlan(), cascadesContext);
-            if (viewStructInfos.size() > 1) {
-                // view struct info should only have one
+            SlotMapping viewToQuerySlotMapping = queryToViewSlotMapping.inverse();
+            LogicalCompatibilityContext compatibilityContext = LogicalCompatibilityContext.from(
+                    queryToViewTableMapping, queryToViewSlotMapping, queryStructInfo, viewStructInfo);
+            ComparisonResult comparisonResult = StructInfo.isGraphLogicalEquals(queryStructInfo, viewStructInfo,
+                    compatibilityContext);
+            if (comparisonResult.isInvalid()) {
                 materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
-                        Pair.of("The num of view struct info is more then one",
-                                String.format("mv plan is %s", materializationContext.getMvPlan().treeString())));
-                return rewriteResults;
-            }
-            StructInfo viewStructInfo = viewStructInfos.get(0);
-            if (!checkPattern(viewStructInfo)) {
-                materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
-                        Pair.of("View struct info is invalid",
-                                String.format(", view plan is %s", viewStructInfo.getOriginalPlan().treeString())));
+                        Pair.of("The graph logic between query and view is not consistent",
+                                comparisonResult.getErrorMessage()));
                 continue;
             }
-            MatchMode matchMode = decideMatchMode(queryStructInfo.getRelations(), viewStructInfo.getRelations());
-            if (MatchMode.COMPLETE != matchMode) {
+            SplitPredicate compensatePredicates = predicatesCompensate(queryStructInfo, viewStructInfo,
+                    viewToQuerySlotMapping, comparisonResult, cascadesContext);
+            // Can not compensate, bail out
+            if (compensatePredicates.isInvalid()) {
                 materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
-                        Pair.of("Match mode is invalid", String.format("matchMode is %s", matchMode)));
+                        Pair.of("Predicate compensate fail",
+                                String.format("query predicates = %s,\n query equivalenceClass = %s, \n"
+                                                + "view predicates = %s,\n query equivalenceClass = %s\n",
+                                        queryStructInfo.getPredicates(),
+                                        queryStructInfo.getEquivalenceClass(),
+                                        viewStructInfo.getPredicates(),
+                                        viewStructInfo.getEquivalenceClass())));
                 continue;
             }
-            List<RelationMapping> queryToViewTableMappings = RelationMapping.generate(queryStructInfo.getRelations(),
-                    viewStructInfo.getRelations());
-            // if any relation in query and view can not map, bail out.
-            if (queryToViewTableMappings == null) {
+            Plan rewrittenPlan;
+            Plan mvScan = materializationContext.getMvScanPlan();
+            Plan originalPlan = queryStructInfo.getOriginalPlan();
+            if (compensatePredicates.isAlwaysTrue()) {
+                rewrittenPlan = mvScan;
+            } else {
+                // Try to rewrite compensate predicates by using mv scan
+                List<Expression> rewriteCompensatePredicates = rewriteExpression(compensatePredicates.toList(),
+                        originalPlan, materializationContext.getMvExprToMvScanExprMapping(),
+                        viewToQuerySlotMapping, true);
+                if (rewriteCompensatePredicates.isEmpty()) {
+                    materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
+                            Pair.of("Rewrite compensate predicate by view fail", String.format(
+                                    "compensatePredicates = %s,\n mvExprToMvScanExprMapping = %s,\n"
+                                            + "viewToQuerySlotMapping = %s",
+                                    compensatePredicates,
+                                    materializationContext.getMvExprToMvScanExprMapping(),
+                                    viewToQuerySlotMapping)));
+                    continue;
+                }
+                rewrittenPlan = new LogicalFilter<>(Sets.newHashSet(rewriteCompensatePredicates), mvScan);
+            }
+            // Rewrite query by view
+            rewrittenPlan = rewriteQueryByView(matchMode, queryStructInfo, viewStructInfo, viewToQuerySlotMapping,
+                    rewrittenPlan, materializationContext);
+            if (rewrittenPlan == null) {
+                continue;
+            }
+            rewrittenPlan = rewriteByRules(cascadesContext, rewrittenPlan, originalPlan);
+            if (!isOutputValid(originalPlan, rewrittenPlan)) {
+                ObjectId planObjId = originalPlan.getGroupExpression().map(GroupExpression::getId)
+                        .orElseGet(() -> new ObjectId(-1));
+                materializationContext.recordFailReason(planObjId, Pair.of(
+                        "RewrittenPlan output logical properties is different with target group",
+                        String.format("planOutput logical properties = %s,\n"
+                                        + "groupOutput logical properties = %s", rewrittenPlan.getLogicalProperties(),
+                                originalPlan.getLogicalProperties())));
+                continue;
+            }
+            // check the partitions used by rewritten plan is valid or not
+            Set<Long> invalidPartitionsQueryUsed =
+                    calcInvalidPartitions(rewrittenPlan, materializationContext, cascadesContext);
+            if (!invalidPartitionsQueryUsed.isEmpty()) {
                 materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
-                        Pair.of("Query to view table mapping is null", ""));
-                return rewriteResults;
-            }
-            for (RelationMapping queryToViewTableMapping : queryToViewTableMappings) {
-                SlotMapping queryToViewSlotMapping = SlotMapping.generate(queryToViewTableMapping);
-                if (queryToViewSlotMapping == null) {
-                    materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
-                            Pair.of("Query to view slot mapping is null", ""));
-                    continue;
-                }
-                LogicalCompatibilityContext compatibilityContext = LogicalCompatibilityContext.from(
-                        queryToViewTableMapping, queryToViewSlotMapping, queryStructInfo, viewStructInfo);
-                ComparisonResult comparisonResult = StructInfo.isGraphLogicalEquals(queryStructInfo, viewStructInfo,
-                        compatibilityContext);
-                if (comparisonResult.isInvalid()) {
-                    materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
-                            Pair.of("The graph logic between query and view is not consistent",
-                                    comparisonResult.getErrorMessage()));
-                    continue;
-                }
-                SplitPredicate compensatePredicates = predicatesCompensate(queryStructInfo, viewStructInfo,
-                        queryToViewSlotMapping, comparisonResult, cascadesContext);
-                // Can not compensate, bail out
-                if (compensatePredicates.isInvalid()) {
-                    materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
-                            Pair.of("Predicate compensate fail",
-                                    String.format("query predicates = %s,\n query equivalenceClass = %s, \n"
-                                                    + "view predicates = %s,\n query equivalenceClass = %s\n",
-                                            queryStructInfo.getPredicates(),
-                                            queryStructInfo.getEquivalenceClass(),
-                                            viewStructInfo.getPredicates(),
-                                            viewStructInfo.getEquivalenceClass())));
-                    continue;
-                }
-                Plan rewrittenPlan;
-                Plan mvScan = materializationContext.getMvScanPlan();
-                if (compensatePredicates.isAlwaysTrue()) {
-                    rewrittenPlan = mvScan;
-                } else {
-                    // Try to rewrite compensate predicates by using mv scan
-                    List<Expression> rewriteCompensatePredicates = rewriteExpression(compensatePredicates.toList(),
-                            queryPlan, materializationContext.getMvExprToMvScanExprMapping(), queryToViewSlotMapping,
-                            true);
-                    if (rewriteCompensatePredicates.isEmpty()) {
-                        materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
-                                Pair.of("Rewrite compensate predicate by view fail", String.format(
-                                        "compensatePredicates = %s,\n mvExprToMvScanExprMapping = %s,\n"
-                                                + "queryToViewSlotMapping = %s",
-                                        compensatePredicates,
-                                        materializationContext.getMvExprToMvScanExprMapping(),
-                                        queryToViewSlotMapping)));
-                        continue;
-                    }
-                    rewrittenPlan = new LogicalFilter<>(Sets.newHashSet(rewriteCompensatePredicates), mvScan);
-                }
-                // Rewrite query by view
-                rewrittenPlan = rewriteQueryByView(matchMode, queryStructInfo, viewStructInfo, queryToViewSlotMapping,
-                        rewrittenPlan, materializationContext);
-                if (rewrittenPlan == null) {
-                    continue;
-                }
-                // run rbo job on mv rewritten plan
-                CascadesContext rewrittenPlanContext = CascadesContext.initContext(
-                        cascadesContext.getStatementContext(), rewrittenPlan,
-                        cascadesContext.getCurrentJobContext().getRequiredProperties());
-                Rewriter.getWholeTreeRewriter(rewrittenPlanContext).execute();
-                rewrittenPlan = rewrittenPlanContext.getRewritePlan();
-                if (!checkOutput(queryPlan, rewrittenPlan, materializationContext)) {
-                    continue;
-                }
-                // check the partitions used by rewritten plan is valid or not
-                Set<Long> invalidPartitionsQueryUsed =
-                        calcInvalidPartitions(rewrittenPlan, materializationContext, cascadesContext);
-                if (!invalidPartitionsQueryUsed.isEmpty()) {
-                    materializationContext.recordFailReason(queryStructInfo.getOriginalPlanId(),
-                            Pair.of("Check partition query used validation fail",
-                                    String.format("the partition used by query is invalid by materialized view,"
-                                                    + "invalid partition info query used is %s",
-                                            materializationContext.getMTMV().getPartitions().stream()
-                                                    .filter(partition ->
-                                                            invalidPartitionsQueryUsed.contains(partition.getId()))
-                                                    .collect(Collectors.toSet()))));
-                    continue;
-                }
-                recordIfRewritten(queryPlan, materializationContext);
-                rewriteResults.add(rewrittenPlan);
+                        Pair.of("Check partition query used validation fail",
+                                String.format("the partition used by query is invalid by materialized view,"
+                                                + "invalid partition info query used is %s",
+                                        materializationContext.getMTMV().getPartitions().stream()
+                                                .filter(partition ->
+                                                        invalidPartitionsQueryUsed.contains(partition.getId()))
+                                                .collect(Collectors.toSet()))));
+                continue;
             }
+            recordIfRewritten(originalPlan, materializationContext);
+            rewriteResults.add(rewrittenPlan);
         }
         return rewriteResults;
     }
 
     /**
-     * Check the logical properties of rewritten plan by mv is the same with source plan
+     * Rewrite by rules and try to make output is the same after optimize by rules
      */
-    protected boolean checkOutput(Plan sourcePlan, Plan rewrittenPlan, MaterializationContext materializationContext) {
+    protected Plan rewriteByRules(CascadesContext cascadesContext, Plan rewrittenPlan, Plan originPlan) {
+        // run rbo job on mv rewritten plan
+        CascadesContext rewrittenPlanContext = CascadesContext.initContext(
+                cascadesContext.getStatementContext(), rewrittenPlan,
+                cascadesContext.getCurrentJobContext().getRequiredProperties());
+        Rewriter.getWholeTreeRewriter(rewrittenPlanContext).execute();
+        rewrittenPlan = rewrittenPlanContext.getRewritePlan();
+        List<Slot> originPlanOutput = originPlan.getOutput();
+        List<Slot> rewrittenPlanOutput = rewrittenPlan.getOutput();
+        if (originPlanOutput.size() != rewrittenPlanOutput.size()) {
+            return null;
+        }
+        List<NamedExpression> expressions = new ArrayList<>();
+        // should add project above rewritten plan if top plan is not project, if aggregate above will nu
+        if (!isOutputValid(originPlan, rewrittenPlan)) {
+            for (int i = 0; i < originPlanOutput.size(); i++) {
+                expressions.add(((NamedExpression) normalizeExpression(originPlanOutput.get(i),
+                        rewrittenPlanOutput.get(i))));
+            }
+            return new LogicalProject<>(expressions, rewrittenPlan, false);
+        }
+        return rewrittenPlan;
+    }
+
+    /**
+     * Check the logical properties of rewritten plan by mv is the same with source plan
+     * if same return true, if different return false
+     */
+    protected boolean isOutputValid(Plan sourcePlan, Plan rewrittenPlan) {
         if (sourcePlan.getGroupExpression().isPresent() && !rewrittenPlan.getLogicalProperties()
                 .equals(sourcePlan.getGroupExpression().get().getOwnerGroup().getLogicalProperties())) {
-            ObjectId planObjId = sourcePlan.getGroupExpression().map(GroupExpression::getId)
-                    .orElseGet(() -> new ObjectId(-1));
-            materializationContext.recordFailReason(planObjId, Pair.of(
-                    "RewrittenPlan output logical properties is different with target group",
-                    String.format("planOutput logical properties = %s,\n"
-                                    + "groupOutput logical properties = %s", rewrittenPlan.getLogicalProperties(),
-                            sourcePlan.getGroupExpression().get().getOwnerGroup().getLogicalProperties())));
             return false;
         }
-        return true;
+        return sourcePlan.getLogicalProperties().equals(rewrittenPlan.getLogicalProperties());
     }
 
     /**
@@ -271,7 +331,7 @@ public abstract class AbstractMaterializedViewRule implements ExplorationRuleFac
      * Rewrite query by view, for aggregate or join rewriting should be different inherit class implementation
      */
     protected Plan rewriteQueryByView(MatchMode matchMode, StructInfo queryStructInfo, StructInfo viewStructInfo,
-            SlotMapping queryToViewSlotMapping, Plan tempRewritedPlan, MaterializationContext materializationContext) {
+            SlotMapping viewToQuerySlotMapping, Plan tempRewritedPlan, MaterializationContext materializationContext) {
         return tempRewritedPlan;
     }
 
@@ -298,7 +358,7 @@ public abstract class AbstractMaterializedViewRule implements ExplorationRuleFac
      *         target
      */
     protected List<Expression> rewriteExpression(List<? extends Expression> sourceExpressionsToWrite, Plan sourcePlan,
-            ExpressionMapping targetExpressionMapping, SlotMapping sourceToTargetMapping,
+            ExpressionMapping targetExpressionMapping, SlotMapping targetToSourceMapping,
             boolean targetExpressionNeedSourceBased) {
         // Firstly, rewrite the target expression using source with inverse mapping
         // then try to use the target expression to represent the query. if any of source expressions
@@ -307,14 +367,13 @@ public abstract class AbstractMaterializedViewRule implements ExplorationRuleFac
         List<? extends Expression> sourceShuttledExpressions = ExpressionUtils.shuttleExpressionWithLineage(
                 sourceExpressionsToWrite, sourcePlan);
         ExpressionMapping expressionMappingKeySourceBased = targetExpressionNeedSourceBased
-                ? targetExpressionMapping.keyPermute(sourceToTargetMapping.inverse()) : targetExpressionMapping;
+                ? targetExpressionMapping.keyPermute(targetToSourceMapping) : targetExpressionMapping;
         // target to target replacement expression mapping, because mv is 1:1 so get first element
         List<Map<Expression, Expression>> flattenExpressionMap = expressionMappingKeySourceBased.flattenMap();
         Map<? extends Expression, ? extends Expression> targetToTargetReplacementMapping = flattenExpressionMap.get(0);
 
         List<Expression> rewrittenExpressions = new ArrayList<>();
-        for (int index = 0; index < sourceShuttledExpressions.size(); index++) {
-            Expression expressionShuttledToRewrite = sourceShuttledExpressions.get(index);
+        for (Expression expressionShuttledToRewrite : sourceShuttledExpressions) {
             if (expressionShuttledToRewrite instanceof Literal) {
                 rewrittenExpressions.add(expressionShuttledToRewrite);
                 continue;
@@ -327,39 +386,31 @@ public abstract class AbstractMaterializedViewRule implements ExplorationRuleFac
                 // if contains any slot to rewrite, which means can not be rewritten by target, bail out
                 return ImmutableList.of();
             }
-            Expression sourceExpression = sourceExpressionsToWrite.get(index);
-            if (sourceExpression instanceof NamedExpression
-                    && replacedExpression.nullable() != sourceExpression.nullable()) {
-                // if enable join eliminate, query maybe inner join and mv maybe outer join.
-                // If the slot is at null generate side, the nullable maybe different between query and view
-                // So need to force to consistent.
-                replacedExpression = sourceExpression.nullable()
-                        ? new Nullable(replacedExpression) : new NonNullable(replacedExpression);
-            }
-            if (sourceExpression instanceof NamedExpression) {
-                NamedExpression sourceNamedExpression = (NamedExpression) sourceExpression;
-                replacedExpression = new Alias(sourceNamedExpression.getExprId(), replacedExpression,
-                        sourceNamedExpression.getName());
-            }
             rewrittenExpressions.add(replacedExpression);
         }
         return rewrittenExpressions;
     }
 
     /**
-     * Rewrite single expression, the logic is the same with above
+     * Normalize expression with query, keep the consistency of exprId and nullable props with
+     * query
      */
-    protected Expression rewriteExpression(Expression sourceExpressionsToWrite, Plan sourcePlan,
-            ExpressionMapping targetExpressionMapping, SlotMapping sourceToTargetMapping,
-            boolean targetExpressionNeedSourceBased) {
-        List<Expression> expressionToRewrite = new ArrayList<>();
-        expressionToRewrite.add(sourceExpressionsToWrite);
-        List<Expression> rewrittenExpressions = rewriteExpression(expressionToRewrite, sourcePlan,
-                targetExpressionMapping, sourceToTargetMapping, targetExpressionNeedSourceBased);
-        if (rewrittenExpressions.isEmpty()) {
-            return null;
+    protected Expression normalizeExpression(Expression sourceExpression, Expression replacedExpression) {
+        if (sourceExpression instanceof NamedExpression
+                && replacedExpression.nullable() != sourceExpression.nullable()) {
+            // if enable join eliminate, query maybe inner join and mv maybe outer join.
+            // If the slot is at null generate side, the nullable maybe different between query and view
+            // So need to force to consistent.
+            replacedExpression = sourceExpression.nullable()
+                    ? new Nullable(replacedExpression) : new NonNullable(replacedExpression);
         }
-        return rewrittenExpressions.get(0);
+        if (sourceExpression instanceof NamedExpression
+                && !sourceExpression.equals(replacedExpression)) {
+            NamedExpression sourceNamedExpression = (NamedExpression) sourceExpression;
+            replacedExpression = new Alias(sourceNamedExpression.getExprId(), replacedExpression,
+                    sourceNamedExpression.getName());
+        }
+        return replacedExpression;
     }
 
     /**
@@ -371,7 +422,7 @@ public abstract class AbstractMaterializedViewRule implements ExplorationRuleFac
     protected SplitPredicate predicatesCompensate(
             StructInfo queryStructInfo,
             StructInfo viewStructInfo,
-            SlotMapping queryToViewSlotMapping,
+            SlotMapping viewToQuerySlotMapping,
             ComparisonResult comparisonResult,
             CascadesContext cascadesContext
     ) {
@@ -379,15 +430,16 @@ public abstract class AbstractMaterializedViewRule implements ExplorationRuleFac
         List<Expression> queryPulledUpExpressions = ImmutableList.copyOf(comparisonResult.getQueryExpressions());
         // set pulled up expression to queryStructInfo predicates and update related predicates
         if (!queryPulledUpExpressions.isEmpty()) {
-            queryStructInfo.addPredicates(queryPulledUpExpressions);
+            queryStructInfo = queryStructInfo.withPredicates(
+                    queryStructInfo.getPredicates().merge(queryPulledUpExpressions));
         }
         List<Expression> viewPulledUpExpressions = ImmutableList.copyOf(comparisonResult.getViewExpressions());
         // set pulled up expression to viewStructInfo predicates and update related predicates
         if (!viewPulledUpExpressions.isEmpty()) {
-            viewStructInfo.addPredicates(viewPulledUpExpressions);
+            viewStructInfo = viewStructInfo.withPredicates(
+                    viewStructInfo.getPredicates().merge(viewPulledUpExpressions));
         }
         // viewEquivalenceClass to query based
-        SlotMapping viewToQuerySlotMapping = queryToViewSlotMapping.inverse();
         // equal predicate compensate
         final Set<Expression> equalCompensateConjunctions = Predicates.compensateEquivalence(
                 queryStructInfo,
@@ -422,11 +474,11 @@ public abstract class AbstractMaterializedViewRule implements ExplorationRuleFac
                 // query has not null reject predicates, so return
                 return SplitPredicate.INVALID_INSTANCE;
             }
+            SlotMapping queryToViewMapping = viewToQuerySlotMapping.inverse();
             Set<Expression> queryUsedNeedRejectNullSlotsViewBased = nullRejectPredicates.stream()
                     .map(expression -> TypeUtils.isNotNull(expression).orElse(null))
                     .filter(Objects::nonNull)
-                    .map(expr -> ExpressionUtils.replace((Expression) expr,
-                            queryToViewSlotMapping.toSlotReferenceMap()))
+                    .map(expr -> ExpressionUtils.replace((Expression) expr, queryToViewMapping.toSlotReferenceMap()))
                     .collect(Collectors.toSet());
             if (requireNoNullableViewSlot.stream().anyMatch(
                     set -> Sets.intersection(set, queryUsedNeedRejectNullSlotsViewBased).isEmpty())) {

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/MaterializedViewAggregateRule.java

@@ -28,7 +28,7 @@ import java.util.List;
 
 /**
  * This is responsible for aggregate rewriting according to different pattern
- * */
+ */
 public class MaterializedViewAggregateRule extends AbstractMaterializedViewAggregateRule {
 
     public static final MaterializedViewAggregateRule INSTANCE = new MaterializedViewAggregateRule();

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/Predicates.java

@@ -29,6 +29,7 @@ import org.apache.doris.nereids.util.Utils;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Sets;
 
+import java.util.Collection;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
@@ -39,32 +40,30 @@ import java.util.Set;
 import java.util.stream.Collectors;
 
 /**
- * This record the predicates which can be pulled up or some other type predicates
+ * This record the predicates which can be pulled up or some other type predicates.
+ * Also contains the necessary method for predicates process
  */
 public class Predicates {
 
     // Predicates that can be pulled up
-    private final Set<Expression> pulledUpPredicates = new HashSet<>();
+    private final Set<Expression> pulledUpPredicates;
 
-    private Predicates() {
+    public Predicates(Set<Expression> pulledUpPredicates) {
+        this.pulledUpPredicates = pulledUpPredicates;
     }
 
-    public static Predicates of() {
-        return new Predicates();
-    }
-
-    public static Predicates of(List<? extends Expression> pulledUpPredicates) {
-        Predicates predicates = new Predicates();
-        pulledUpPredicates.forEach(predicates::addPredicate);
-        return predicates;
+    public static Predicates of(Set<Expression> pulledUpPredicates) {
+        return new Predicates(pulledUpPredicates);
     }
 
     public Set<Expression> getPulledUpPredicates() {
         return pulledUpPredicates;
     }
 
-    public void addPredicate(Expression expression) {
-        this.pulledUpPredicates.add(expression);
+    public Predicates merge(Collection<Expression> predicates) {
+        Set<Expression> mergedPredicates = new HashSet<>(predicates);
+        mergedPredicates.addAll(this.pulledUpPredicates);
+        return new Predicates(mergedPredicates);
     }
 
     public Expression composedExpression() {
@@ -98,8 +97,7 @@ public class Predicates {
         if (queryEquivalenceClass.isEmpty() && viewEquivalenceClass.isEmpty()) {
             equalCompensateConjunctions.add(BooleanLiteral.TRUE);
         }
-        if (queryEquivalenceClass.isEmpty()
-                && !viewEquivalenceClass.isEmpty()) {
+        if (queryEquivalenceClass.isEmpty() && !viewEquivalenceClass.isEmpty()) {
             return null;
         }
         EquivalenceClassSetMapping queryToViewEquivalenceMapping =

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/StructInfo.java

@@ -17,7 +17,9 @@
 
 package org.apache.doris.nereids.rules.exploration.mv;
 
+import org.apache.doris.common.Pair;
 import org.apache.doris.nereids.jobs.joinorder.hypergraph.HyperGraph;
+import org.apache.doris.nereids.jobs.joinorder.hypergraph.edge.JoinEdge;
 import org.apache.doris.nereids.jobs.joinorder.hypergraph.node.StructInfoNode;
 import org.apache.doris.nereids.memo.Group;
 import org.apache.doris.nereids.memo.GroupExpression;
@@ -44,6 +46,7 @@ import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
 import org.apache.doris.nereids.trees.plans.visitor.ExpressionLineageReplacer;
 import org.apache.doris.nereids.util.ExpressionUtils;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
@@ -60,6 +63,8 @@ import javax.annotation.Nullable;
 
 /**
  * StructInfo for plan, this contains necessary info for query rewrite by materialized view
+ * the struct info is used by all materialization, so it's struct info should only get, should not
+ * modify, if wanting to modify, should copy and then modify
  */
 public class StructInfo {
     public static final JoinPatternChecker JOIN_PATTERN_CHECKER = new JoinPatternChecker();
@@ -70,58 +75,76 @@ public class StructInfo {
     private static final PredicateCollector PREDICATE_COLLECTOR = new PredicateCollector();
     // source data
     private final Plan originalPlan;
-    private ObjectId originalPlanId;
+    private final ObjectId originalPlanId;
     private final HyperGraph hyperGraph;
-    private boolean valid = true;
+    private final boolean valid;
     // derived data following
     // top plan which may include project or filter, except for join and scan
-    private Plan topPlan;
+    private final Plan topPlan;
     // bottom plan which top plan only contain join or scan. this is needed by hyper graph
-    private Plan bottomPlan;
-    private final List<CatalogRelation> relations = new ArrayList<>();
+    private final Plan bottomPlan;
+    private final List<CatalogRelation> relations;
     // this is for LogicalCompatibilityContext later
-    private final Map<RelationId, StructInfoNode> relationIdStructInfoNodeMap = new HashMap<>();
+    private final Map<RelationId, StructInfoNode> relationIdStructInfoNodeMap;
     // this recorde the predicates which can pull up, not shuttled
     private Predicates predicates;
     // split predicates is shuttled
-    private SplitPredicate splitPredicate;
-    private EquivalenceClass equivalenceClass;
+    private final SplitPredicate splitPredicate;
+    private final EquivalenceClass equivalenceClass;
     // Key is the expression shuttled and the value is the origin expression
     // this is for building LogicalCompatibilityContext later.
-    private final Map<Expression, Expression> shuttledHashConjunctsToConjunctsMap = new HashMap<>();
+    private final Map<Expression, Expression> shuttledHashConjunctsToConjunctsMap;
     // Record the exprId and the corresponding expr map, this is used by expression shuttled
-    private final Map<ExprId, Expression> namedExprIdAndExprMapping = new HashMap<>();
+    private final Map<ExprId, Expression> namedExprIdAndExprMapping;
 
-    private StructInfo(Plan originalPlan, @Nullable Plan topPlan, @Nullable Plan bottomPlan, HyperGraph hyperGraph) {
+    /**
+     * The construct method for StructInfo
+     */
+    public StructInfo(Plan originalPlan, ObjectId originalPlanId, HyperGraph hyperGraph, boolean valid, Plan topPlan,
+            Plan bottomPlan, List<CatalogRelation> relations,
+            Map<RelationId, StructInfoNode> relationIdStructInfoNodeMap,
+            @Nullable Predicates predicates,
+            Map<Expression, Expression> shuttledHashConjunctsToConjunctsMap,
+            Map<ExprId, Expression> namedExprIdAndExprMapping) {
         this.originalPlan = originalPlan;
-        this.originalPlanId = originalPlan.getGroupExpression()
-                .map(GroupExpression::getId).orElseGet(() -> new ObjectId(-1));
+        this.originalPlanId = originalPlanId;
         this.hyperGraph = hyperGraph;
+        this.valid = valid;
         this.topPlan = topPlan;
         this.bottomPlan = bottomPlan;
-        init();
-    }
-
-    private void init() {
-        // split the top plan to two parts by join node
-        if (topPlan == null || bottomPlan == null) {
-            PlanSplitContext planSplitContext = new PlanSplitContext(Sets.newHashSet(LogicalJoin.class));
-            originalPlan.accept(PLAN_SPLITTER, planSplitContext);
-            this.bottomPlan = planSplitContext.getBottomPlan();
-            this.topPlan = planSplitContext.getTopPlan();
+        this.relations = relations;
+        this.relationIdStructInfoNodeMap = relationIdStructInfoNodeMap;
+        this.predicates = predicates;
+        if (predicates == null) {
+            // collect predicate from top plan which not in hyper graph
+            Set<Expression> topPlanPredicates = new HashSet<>();
+            topPlan.accept(PREDICATE_COLLECTOR, topPlanPredicates);
+            this.predicates = Predicates.of(topPlanPredicates);
         }
-        collectStructInfoFromGraph();
-        initPredicates();
+        Pair<SplitPredicate, EquivalenceClass> derivedPredicates = predicatesDerive(this.predicates, originalPlan);
+        this.splitPredicate = derivedPredicates.key();
+        this.equivalenceClass = derivedPredicates.value();
+        this.shuttledHashConjunctsToConjunctsMap = shuttledHashConjunctsToConjunctsMap;
+        this.namedExprIdAndExprMapping = namedExprIdAndExprMapping;
     }
 
-    public void addPredicates(List<Expression> canPulledUpExpressions) {
-        canPulledUpExpressions.forEach(this.predicates::addPredicate);
-        predicatesDerive();
+    /**
+     * Construct StructInfo with new predicates
+     */
+    public StructInfo withPredicates(Predicates predicates) {
+        return new StructInfo(this.originalPlan, this.originalPlanId, this.hyperGraph, this.valid, this.topPlan,
+                this.bottomPlan, this.relations, this.relationIdStructInfoNodeMap, predicates,
+                this.shuttledHashConjunctsToConjunctsMap, this.namedExprIdAndExprMapping);
     }
 
-    private void collectStructInfoFromGraph() {
+    private static boolean collectStructInfoFromGraph(HyperGraph hyperGraph,
+            Plan topPlan,
+            Map<Expression, Expression> shuttledHashConjunctsToConjunctsMap,
+            Map<ExprId, Expression> namedExprIdAndExprMapping,
+            ImmutableList.Builder<CatalogRelation> relationBuilder,
+            Map<RelationId, StructInfoNode> relationIdStructInfoNodeMap) {
         // Collect expression from join condition in hyper graph
-        this.hyperGraph.getJoinEdges().forEach(edge -> {
+        for (JoinEdge edge : hyperGraph.getJoinEdges()) {
             List<Expression> hashJoinConjuncts = edge.getHashJoinConjuncts();
             // shuttle expression in edge for the build of LogicalCompatibilityContext later.
             // Record the exprId to expr map in the processing to strut info
@@ -132,34 +155,31 @@ public class StructInfo {
                                 Lists.newArrayList(conjunctExpr),
                                 ImmutableSet.of(),
                                 ImmutableSet.of());
-                this.topPlan.accept(ExpressionLineageReplacer.INSTANCE, replaceContext);
+                topPlan.accept(ExpressionLineageReplacer.INSTANCE, replaceContext);
                 // Replace expressions by expression map
                 List<Expression> replacedExpressions = replaceContext.getReplacedExpressions();
                 shuttledHashConjunctsToConjunctsMap.put(replacedExpressions.get(0), conjunctExpr);
                 // Record this, will be used in top level expression shuttle later, see the method
                 // ExpressionLineageReplacer#visitGroupPlan
-                this.namedExprIdAndExprMapping.putAll(replaceContext.getExprIdExpressionMap());
+                namedExprIdAndExprMapping.putAll(replaceContext.getExprIdExpressionMap());
             });
             List<Expression> otherJoinConjuncts = edge.getOtherJoinConjuncts();
             if (!otherJoinConjuncts.isEmpty()) {
-                this.valid = false;
+                return false;
             }
-        });
-        if (!this.isValid()) {
-            return;
         }
         // Collect relations from hyper graph which in the bottom plan
-        this.hyperGraph.getNodes().forEach(node -> {
+        hyperGraph.getNodes().forEach(node -> {
             // plan relation collector and set to map
             Plan nodePlan = node.getPlan();
             List<CatalogRelation> nodeRelations = new ArrayList<>();
             nodePlan.accept(RELATION_COLLECTOR, nodeRelations);
-            this.relations.addAll(nodeRelations);
+            relationBuilder.addAll(nodeRelations);
             // every node should only have one relation, this is for LogicalCompatibilityContext
             relationIdStructInfoNodeMap.put(nodeRelations.get(0).getRelationId(), (StructInfoNode) node);
         });
         // Collect expression from where in hyper graph
-        this.hyperGraph.getFilterEdges().forEach(filterEdge -> {
+        hyperGraph.getFilterEdges().forEach(filterEdge -> {
             List<? extends Expression> filterExpressions = filterEdge.getExpressions();
             filterExpressions.forEach(predicate -> {
                 // this is used for LogicalCompatibilityContext
@@ -168,28 +188,18 @@ public class StructInfo {
                                 ExpressionUtils.shuttleExpressionWithLineage(predicate, topPlan), predicate));
             });
         });
-    }
-
-    private void initPredicates() {
-        // Collect predicate from top plan which not in hyper graph
-        this.predicates = Predicates.of();
-        Set<Expression> topPlanPredicates = new HashSet<>();
-        topPlan.accept(PREDICATE_COLLECTOR, topPlanPredicates);
-        topPlanPredicates.forEach(this.predicates::addPredicate);
-        predicatesDerive();
+        return true;
     }
 
     // derive some useful predicate by predicates
-    private void predicatesDerive() {
+    private Pair<SplitPredicate, EquivalenceClass> predicatesDerive(Predicates predicates, Plan originalPlan) {
         // construct equivalenceClass according to equals predicates
         List<Expression> shuttledExpression = ExpressionUtils.shuttleExpressionWithLineage(
-                        new ArrayList<>(this.predicates.getPulledUpPredicates()), originalPlan).stream()
+                        new ArrayList<>(predicates.getPulledUpPredicates()), originalPlan).stream()
                 .map(Expression.class::cast)
                 .collect(Collectors.toList());
         SplitPredicate splitPredicate = Predicates.splitPredicates(ExpressionUtils.and(shuttledExpression));
-        this.splitPredicate = splitPredicate;
-
-        this.equivalenceClass = new EquivalenceClass();
+        EquivalenceClass equivalenceClass = new EquivalenceClass();
         for (Expression expression : ExpressionUtils.extractConjunction(splitPredicate.getEqualPredicate())) {
             if (expression instanceof Literal) {
                 continue;
@@ -201,6 +211,7 @@ public class StructInfo {
                         (SlotReference) equalTo.getArguments().get(1));
             }
         }
+        return Pair.of(splitPredicate, equivalenceClass);
     }
 
     /**
@@ -216,11 +227,39 @@ public class StructInfo {
 
         List<HyperGraph> structInfos = HyperGraph.toStructInfo(planSplitContext.getBottomPlan());
         return structInfos.stream()
-                .map(hyperGraph -> new StructInfo(originalPlan, planSplitContext.getTopPlan(),
+                .map(hyperGraph -> StructInfo.of(originalPlan, planSplitContext.getTopPlan(),
                         planSplitContext.getBottomPlan(), hyperGraph))
                 .collect(Collectors.toList());
     }
 
+    /**
+     * The construct method for init StructInfo
+     */
+    public static StructInfo of(Plan originalPlan, @Nullable Plan topPlan, @Nullable Plan bottomPlan,
+            HyperGraph hyperGraph) {
+        ObjectId originalPlanId = originalPlan.getGroupExpression()
+                .map(GroupExpression::getId).orElseGet(() -> new ObjectId(-1));
+        // if any of topPlan or bottomPlan is null, split the top plan to two parts by join node
+        if (topPlan == null || bottomPlan == null) {
+            PlanSplitContext planSplitContext = new PlanSplitContext(Sets.newHashSet(LogicalJoin.class));
+            originalPlan.accept(PLAN_SPLITTER, planSplitContext);
+            bottomPlan = planSplitContext.getBottomPlan();
+            topPlan = planSplitContext.getTopPlan();
+        }
+        // collect struct info fromGraph
+        ImmutableList.Builder<CatalogRelation> relationBuilder = ImmutableList.builder();
+        Map<RelationId, StructInfoNode> relationIdStructInfoNodeMap = new HashMap<>();
+        Map<Expression, Expression> shuttledHashConjunctsToConjunctsMap = new HashMap<>();
+        Map<ExprId, Expression> namedExprIdAndExprMapping = new HashMap<>();
+        boolean valid = collectStructInfoFromGraph(hyperGraph, topPlan, shuttledHashConjunctsToConjunctsMap,
+                namedExprIdAndExprMapping,
+                relationBuilder,
+                relationIdStructInfoNodeMap);
+        return new StructInfo(originalPlan, originalPlanId, hyperGraph, valid, topPlan, bottomPlan,
+                relationBuilder.build(), relationIdStructInfoNodeMap, null, shuttledHashConjunctsToConjunctsMap,
+                namedExprIdAndExprMapping);
+    }
+
     /**
      * Build Struct info from group.
      * Maybe return multi structInfo when original plan already be rewritten by mv

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

@@ -38,10 +38,14 @@ import java.util.function.BiFunction;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 
-/** NormalizeToSlot */
+/**
+ * NormalizeToSlot
+ */
 public interface NormalizeToSlot {
 
-    /** NormalizeSlotContext */
+    /**
+     * NormalizeSlotContext
+     */
     class NormalizeToSlotContext {
         private final Map<Expression, NormalizeToSlotTriplet> normalizeToSlotMap;
 
@@ -51,11 +55,11 @@ public interface NormalizeToSlot {
 
         /**
          * build normalization context by follow step.
-         *   1. collect all exists alias by input parameters existsAliases build a reverted map: expr -> alias
-         *   2. for all input source expressions, use existsAliasMap to construct triple:
-         *     origin expr, pushed expr and alias to replace origin expr,
-         *     see more detail in {@link NormalizeToSlotTriplet}
-         *   3. construct a map: original expr -> triple constructed by step 2
+         * 1. collect all exists alias by input parameters existsAliases build a reverted map: expr -> alias
+         * 2. for all input source expressions, use existsAliasMap to construct triple:
+         * origin expr, pushed expr and alias to replace origin expr,
+         * see more detail in {@link NormalizeToSlotTriplet}
+         * 3. construct a map: original expr -> triple constructed by step 2
          */
         public static NormalizeToSlotContext buildContext(
                 Set<Alias> existsAliases, Collection<? extends Expression> sourceExpressions) {
@@ -65,7 +69,6 @@ public interface NormalizeToSlot {
             for (Alias existsAlias : existsAliases) {
                 existsAliasMap.put(existsAlias.child(), existsAlias);
             }
-
             for (Expression expression : sourceExpressions) {
                 if (normalizeToSlotMap.containsKey(expression)) {
                     continue;
@@ -186,7 +189,9 @@ public interface NormalizeToSlot {
         }
     }
 
-    /** NormalizeToSlotTriplet */
+    /**
+     * NormalizeToSlotTriplet
+     */
     class NormalizeToSlotTriplet {
         // which expression need to normalized to slot?
         // e.g. `a + 1`

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

@@ -199,23 +199,6 @@ public interface TreeNode<NODE_TYPE extends TreeNode<NODE_TYPE>> {
         return false;
     }
 
-    /**
-     * iterate top down and test predicate if any matched. Top-down traverse implicitly.
-     * @param predicate predicate
-     * @return the first node which match the predicate
-     */
-    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 child;
-            }
-        }
-        return this;
-    }
-
     /**
      * Collect the nodes that satisfied the predicate.
      */

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

@@ -61,6 +61,10 @@ public class LogicalProject<CHILD_TYPE extends Plan> extends LogicalUnary<CHILD_
         this(projects, ImmutableList.of(), false, true, ImmutableList.of(child));
     }
 
+    public LogicalProject(List<NamedExpression> projects, CHILD_TYPE child, boolean canEliminate) {
+        this(projects, ImmutableList.of(), false, canEliminate, ImmutableList.of(child));
+    }
+
     public LogicalProject(List<NamedExpression> projects, List<NamedExpression> excepts,
             boolean isDistinct, List<Plan> child) {
         this(projects, excepts, isDistinct, true, Optional.empty(), Optional.empty(), child);

fe/fe-core/src/main/java/org/apache/doris/nereids/util/ExpressionUtils.java

@@ -231,7 +231,9 @@ public class ExpressionUtils {
             Plan plan,
             Set<TableType> targetTypes,
             Set<String> tableIdentifiers) {
-
+        if (expressions.isEmpty()) {
+            return ImmutableList.of();
+        }
         ExpressionLineageReplacer.ExpressionReplaceContext replaceContext =
                 new ExpressionLineageReplacer.ExpressionReplaceContext(
                         expressions.stream().map(Expression.class::cast).collect(Collectors.toList()),