[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());