[improvement](mtmv) Support union rewrite when the materialized view is not enough to provide all the data for the query

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

@@ -17,27 +17,25 @@
 
 package org.apache.doris.mtmv;
 
-import org.apache.doris.catalog.Column;
 import org.apache.doris.catalog.MTMV;
-import org.apache.doris.catalog.Partition;
+import org.apache.doris.nereids.CascadesContext;
 import org.apache.doris.nereids.NereidsPlanner;
 import org.apache.doris.nereids.StatementContext;
-import org.apache.doris.nereids.analyzer.UnboundResultSink;
-import org.apache.doris.nereids.analyzer.UnboundTableSink;
+import org.apache.doris.nereids.jobs.executor.Rewriter;
 import org.apache.doris.nereids.parser.NereidsParser;
 import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.rules.exploration.mv.MaterializedViewUtils;
+import org.apache.doris.nereids.rules.rewrite.EliminateSort;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.commands.ExplainCommand.ExplainLevel;
 import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
-import org.apache.doris.nereids.trees.plans.logical.LogicalTableSink;
+import org.apache.doris.nereids.trees.plans.logical.LogicalResultSink;
 import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
-import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.OriginStatement;
 
-import com.google.common.collect.Lists;
-
-import java.util.stream.Collectors;
+import com.google.common.collect.ImmutableList;
 
 /**
  * The cache for materialized view cache
@@ -70,27 +68,25 @@ public static MTMVCache from(MTMV mtmv, ConnectContext connectContext) {
         if (mvSqlStatementContext.getConnectContext().getStatementContext() == null) {
             mvSqlStatementContext.getConnectContext().setStatementContext(mvSqlStatementContext);
         }
-        unboundMvPlan = unboundMvPlan.accept(new DefaultPlanVisitor<LogicalPlan, Void>() {
-            // convert to table sink to eliminate sort under table sink, because sort under result sink can not be
-            // eliminated
-            @Override
-            public LogicalPlan visitUnboundResultSink(UnboundResultSink<? extends Plan> unboundResultSink,
-                    Void context) {
-                return new UnboundTableSink<>(mtmv.getFullQualifiers(),
-                        mtmv.getBaseSchema().stream().map(Column::getName).collect(Collectors.toList()),
-                        Lists.newArrayList(),
-                        mtmv.getPartitions().stream().map(Partition::getName).collect(Collectors.toList()),
-                        unboundResultSink.child());
-            }
-        }, null);
+        // Can not convert to table sink, because use the same column from different table when self join
+        // the out slot is wrong
         Plan originPlan = planner.plan(unboundMvPlan, PhysicalProperties.ANY, ExplainLevel.REWRITTEN_PLAN);
-        // eliminate logicalTableSink because sink operator is useless in query rewrite by materialized view
-        Plan mvPlan = planner.getCascadesContext().getRewritePlan().accept(new DefaultPlanRewriter<Object>() {
+        // Eliminate result sink because sink operator is useless in query rewrite by materialized view
+        // and the top sort can also be removed
+        Plan mvPlan = originPlan.accept(new DefaultPlanRewriter<Object>() {
             @Override
-            public Plan visitLogicalTableSink(LogicalTableSink<? extends Plan> logicalTableSink, Object context) {
-                return logicalTableSink.child().accept(this, context);
+            public Plan visitLogicalResultSink(LogicalResultSink<? extends Plan> logicalResultSink, Object context) {
+                return logicalResultSink.child().accept(this, context);
             }
         }, null);
+        // Optimize by rules to remove top sort
+        CascadesContext parentCascadesContext = CascadesContext.initContext(mvSqlStatementContext, mvPlan,
+                PhysicalProperties.ANY);
+        mvPlan = MaterializedViewUtils.rewriteByRules(parentCascadesContext, childContext -> {
+            Rewriter.getCteChildrenRewriter(childContext,
+                    ImmutableList.of(Rewriter.custom(RuleType.ELIMINATE_SORT, EliminateSort::new))).execute();
+            return childContext.getRewritePlan();
+        }, mvPlan, originPlan);
         return new MTMVCache(mvPlan, originPlan);
     }
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/CommonSubExpressionOpt.java

@@ -83,8 +83,9 @@ private List<List<NamedExpression>> computeMultiLayerProjections(
                         // 'case slot whenClause2 END'
                         // This is illegal.
                         Expression rewritten = expr.accept(ExpressionReplacer.INSTANCE, aliasMap);
-                        Alias alias = new Alias(rewritten);
-                        aliasMap.put(expr, alias);
+                        // if rewritten is already alias, use it directly, because in materialized view rewriting
+                        // Should keep out slot immutably after rewritten successfully
+                        aliasMap.put(expr, rewritten instanceof Alias ? (Alias) rewritten : new Alias(rewritten));
                     }
                 });
                 layer.addAll(aliasMap.values());