apache · englefly · Jun 25, 2026 · Jun 30, 2026 · Jul 5, 2026 · Jul 5, 2026
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
@@ -267,7 +267,8 @@ public void run() throws JobException {
                 try {
                     executeWithRetry(execPartitionNames, tableWithPartKey);
                 } catch (Exception e) {
-                    LOG.error("Execution failed after retries: {}", e.getMessage());
+                    LOG.error("Execution failed after retries, mvName: {}, taskId: {}",
+                            mtmv.getName(), getTaskId(), e);
                     throw new JobException(e.getMessage(), e);
                 }
                 completedPartitions.addAll(execPartitionNames);
@@ -277,7 +278,8 @@ public void run() throws JobException {
                     mtmv.getDatabase().getFullName(), mtmv.getName(), getTaskId());
         } catch (Throwable e) {
             if (getStatus() == TaskStatus.RUNNING) {
-                LOG.warn("run task failed: {}", e.getMessage());
+                LOG.warn("run task failed, mvName: {}, taskId: {}",
+                        mtmv.getName(), getTaskId(), e);
                 throw new JobException(e.getMessage(), e);
             } else {
                 // if status is not `RUNNING`,maybe the task was canceled, therefore, it is a normal situation

@@ -111,6 +111,7 @@ public class MTMVPlanUtil {
             RuleType.ELIMINATE_JOIN_BY_FK,
             RuleType.ELIMINATE_JOIN_BY_UK,
             RuleType.ELIMINATE_GROUP_BY_KEY_BY_UNIFORM,
+            RuleType.ELIMINATE_GROUP_BY_KEY,
             RuleType.ELIMINATE_GROUP_BY,
             RuleType.SALT_JOIN
     );

@@ -678,7 +678,6 @@ public class Rewriter extends AbstractBatchJobExecutor {
                         cascadesContext -> cascadesContext.rewritePlanContainsTypes(LogicalAggregate.class)
                                 || cascadesContext.rewritePlanContainsTypes(LogicalJoin.class)
                                 || cascadesContext.rewritePlanContainsTypes(LogicalUnion.class),
-                        topDown(new EliminateGroupByKey()),
                         topDown(new PushDownAggThroughJoinOnPkFk()),
                         topDown(new PullUpJoinFromUnionAll())
                 ),
@@ -916,6 +915,11 @@ private static List<RewriteJob> getWholeTreeRewriteJobs(
                             )));
                     rewriteJobs.addAll(jobs(topic("convert outer join to anti",
                             custom(RuleType.CONVERT_OUTER_JOIN_TO_ANTI, ConvertOuterJoinToAntiJoin::new))));
+                    rewriteJobs.addAll(jobs(topic("eliminate Aggregate according to fd items",
+                            cascadesContext -> cascadesContext.rewritePlanContainsTypes(LogicalAggregate.class)
+                                    || cascadesContext.rewritePlanContainsTypes(LogicalJoin.class)
+                                    || cascadesContext.rewritePlanContainsTypes(LogicalUnion.class),
+                            custom(RuleType.ELIMINATE_GROUP_BY_KEY, EliminateGroupByKey::new))));
                     rewriteJobs.addAll(jobs(topic("eliminate group by key by uniform",
                             custom(RuleType.ELIMINATE_GROUP_BY_KEY_BY_UNIFORM, EliminateGroupByKeyByUniform::new))));
                     if (needOrExpansion) {

@@ -68,6 +68,7 @@ public class PreMaterializedViewRewriter {
         NEED_PRE_REWRITE_RULE_TYPES.set(RuleType.DISTINCT_AGGREGATE_SPLIT.ordinal());
         NEED_PRE_REWRITE_RULE_TYPES.set(RuleType.PROCESS_SCALAR_AGG_MUST_USE_MULTI_DISTINCT.ordinal());
         NEED_PRE_REWRITE_RULE_TYPES.set(RuleType.ELIMINATE_GROUP_BY_KEY_BY_UNIFORM.ordinal());
+        NEED_PRE_REWRITE_RULE_TYPES.set(RuleType.ELIMINATE_GROUP_BY_KEY.ordinal());
         NEED_PRE_REWRITE_RULE_TYPES.set(RuleType.SALT_JOIN.ordinal());
         NEED_PRE_REWRITE_RULE_TYPES.set(RuleType.PULL_UP_PROJECT_EXPR_UNDER_TOPN.ordinal());
     }

@@ -231,10 +231,10 @@ public Rule build() {
                 List<Expression> groupByExprs = agg.getGroupByExpressions();
                 ExpressionRewriteContext context = new ExpressionRewriteContext(agg, ctx.cascadesContext);
                 List<Expression> newGroupByExprs = rewriter.rewrite(groupByExprs, context);
-
+                boolean groupByChanged = !newGroupByExprs.equals(groupByExprs);
                 List<NamedExpression> outputExpressions = agg.getOutputExpressions();
                 RewriteResult<NamedExpression> result = rewriteAll(outputExpressions, rewriter, context);
-                if (!result.changed) {
+                if (!result.changed && !groupByChanged) {
                     return agg;
                 }
                 return new LogicalAggregate<>(newGroupByExprs, result.result,

@@ -17,90 +17,230 @@
 
 package org.apache.doris.nereids.rules.rewrite;
 
-import org.apache.doris.nereids.annotation.DependsRules;
+import org.apache.doris.nereids.jobs.JobContext;
 import org.apache.doris.nereids.properties.DataTrait;
 import org.apache.doris.nereids.properties.FuncDeps;
-import org.apache.doris.nereids.rules.Rule;
-import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ExprId;
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.NamedExpression;
 import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AnyValue;
 import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.algebra.Aggregate;
 import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
 
-import com.google.common.collect.ImmutableList;
+import com.google.common.collect.LinkedHashMultimap;
+import com.google.common.collect.Multimap;
 
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
 
-
 /**
  * Eliminate group by key based on fd item information.
  * such as:
  *  for a -> b, we can get:
  *          group by a, b, c  => group by a, c
+ *
+ * When a group-by key is FD-redundant but still needed in the output,
+ * it is wrapped with any_value() and assigned a fresh ExprId.
+ * Upper plan references are rewritten via ExprIdRewriter so that
+ * all ancestor nodes see the new ExprIds.
  */
-@DependsRules({EliminateGroupBy.class, ColumnPruning.class})
-public class EliminateGroupByKey implements RewriteRuleFactory {
+public class EliminateGroupByKey extends DefaultPlanRewriter<Map<ExprId, ExprId>> implements CustomRewriter {
+    private ExprIdRewriter exprIdReplacer;
+
+    @Override
+    public Plan rewriteRoot(Plan plan, JobContext jobContext) {
+        if (!plan.containsType(Aggregate.class)) {
+            return plan;
+        }
+        Map<ExprId, ExprId> replaceMap = new HashMap<>();
+        ExprIdRewriter.ReplaceRule replaceRule = new ExprIdRewriter.ReplaceRule(replaceMap, false);
+        exprIdReplacer = new ExprIdRewriter(replaceRule, jobContext);
+        return plan.accept(this, replaceMap);
+    }
+
+    @Override
+    public Plan visit(Plan plan, Map<ExprId, ExprId> replaceMap) {
+        plan = visitChildren(this, plan, replaceMap);
+        plan = exprIdReplacer.rewriteExpr(plan, replaceMap);
+        return plan;
+    }
+
+    @Override
+    public Plan visitLogicalProject(LogicalProject<? extends Plan> proj, Map<ExprId, ExprId> replaceMap) {
+        proj = visitChildren(this, proj, replaceMap);
+
+        // Find the Aggregate child, possibly through a Filter
+        Plan child = proj.child(0);
+        LogicalAggregate<? extends Plan> agg;
+        boolean hasFilter = child instanceof LogicalFilter;
+        if (hasFilter && child.child(0) instanceof LogicalAggregate) {
+            agg = (LogicalAggregate<? extends Plan>) child.child(0);
+        } else if (child instanceof LogicalAggregate) {
+            agg = (LogicalAggregate<? extends Plan>) child;
+        } else {
+            return exprIdReplacer.rewriteExpr(proj, replaceMap);
+        }
+
+        // Don't transform if source repeat is present
+        if (agg.getSourceRepeat().isPresent()) {
+            return exprIdReplacer.rewriteExpr(proj, replaceMap);
+        }
+
+        // Rewrite proj and the filter (if present) through the replaceMap accumulated
+        // by visitChildren, so that ExprId replacements from nested rewrites
+        // (e.g. inner aggregates) are reflected in the required-output slot set.
+        proj = (LogicalProject<? extends Plan>) exprIdReplacer.rewriteExpr(proj, replaceMap);
+        if (hasFilter) {
+            child = exprIdReplacer.rewriteExpr(child, replaceMap);
+        }
+
+        // Compute requireOutput: slots needed by the Project (and Filter, if present)
+        Set<Slot> requireOutput = new HashSet<>(proj.getInputSlots());
+        if (hasFilter) {
+            requireOutput.addAll(child.getInputSlots());
+        }
+
+        // Transform the aggregate
+        EliminateResult result = eliminateGroupByKeyWithMap(agg, requireOutput);
+        if (!result.changed) {
+            return proj;
+        }
+
+        // Merge into the global replaceMap so that all ancestor nodes get rewritten
+        replaceMap.putAll(result.replaceMap);
+
+        // Rebuild the child chain with the new aggregate,
+        // and rewrite the Filter (if present) and Project expressions
+        Plan newChild;
+        if (hasFilter) {
+            Plan updatedFilter = child.withChildren(result.newAgg);
+            newChild = exprIdReplacer.rewriteExpr(updatedFilter, replaceMap);
+        } else {
+            newChild = result.newAgg;
+        }
+        Plan newProj = exprIdReplacer.rewriteExpr(proj.withChildren(newChild), replaceMap);
+        return newProj;
+    }
 
     @Override
-    public List<Rule> buildRules() {
-        return ImmutableList.of(
-                RuleType.ELIMINATE_GROUP_BY_KEY.build(
-                        logicalProject(logicalAggregate().when(agg -> !agg.getSourceRepeat().isPresent()))
-                                .then(proj -> {
-                                    LogicalAggregate<? extends Plan> agg = proj.child();
-                                    LogicalAggregate<Plan> newAgg = eliminateGroupByKey(agg, proj.getInputSlots());
-                                    if (newAgg == null) {
-                                        return null;
-                                    }
-                                    return proj.withChildren(newAgg);
-                                })),
-                RuleType.ELIMINATE_FILTER_GROUP_BY_KEY.build(
-                        logicalProject(logicalFilter(logicalAggregate()
-                                .when(agg -> !agg.getSourceRepeat().isPresent())))
-                                .then(proj -> {
-                                    LogicalAggregate<? extends Plan> agg = proj.child().child();
-                                    Set<Slot> requireSlots = new HashSet<>(proj.getInputSlots());
-                                    requireSlots.addAll(proj.child(0).getInputSlots());
-                                    LogicalAggregate<Plan> newAgg = eliminateGroupByKey(agg, requireSlots);
-                                    if (newAgg == null) {
-                                        return null;
-                                    }
-                                    return proj.withChildren(proj.child().withChildren(newAgg));
-                                })
-                )
-        );
+    public Plan visitLogicalCTEConsumer(LogicalCTEConsumer cteConsumer, Map<ExprId, ExprId> replaceMap) {
+        // When a producer aggregate's output slot is wrapped with any_value(),
+        // a fresh ExprId is recorded in replaceMap. The CTE consumer's producerToConsumerSlotMap
+        // still references the old ExprId, so we must rebuild both maps with the new ExprIds.
+        Map<Slot, Slot> newConsumerToProducer = new LinkedHashMap<>();
+        Multimap<Slot, Slot> newProducerToConsumer = LinkedHashMultimap.create();
+        for (Slot producerSlot : cteConsumer.getConsumerToProducerOutputMap().values()) {
+            ExprId newExprId = resolveExprIdChain(producerSlot.getExprId(), replaceMap);
+            Slot effectiveProducerSlot = newExprId != null
+                    ? (Slot) producerSlot.withExprId(newExprId)
+                    : producerSlot;
+            for (Slot consumerSlot : cteConsumer.getProducerToConsumerOutputMap().get(producerSlot)) {
+                newProducerToConsumer.put(effectiveProducerSlot, consumerSlot);
+                newConsumerToProducer.put(consumerSlot, effectiveProducerSlot);
+            }
+        }
+        return cteConsumer.withTwoMaps(newConsumerToProducer, newProducerToConsumer);
+    }
+
+    /** Follow transitive ExprId chain to find the final replacement, or null if none. */
+    private static ExprId resolveExprIdChain(ExprId exprId, Map<ExprId, ExprId> replaceMap) {
+        ExprId newId = replaceMap.get(exprId);
+        if (newId == null) {
+            return null;
+        }
+        ExprId lastId = newId;
+        while (true) {
+            ExprId next = replaceMap.get(lastId);
+            if (next == null) {
+                return lastId;
+            }
+            lastId = next;
+        }
+    }
+
+    /** Result of eliminateGroupByKey: the new aggregate and a map of old->new ExprIds. */
+    private static class EliminateResult {
+        final LogicalAggregate<Plan> newAgg;
+        final Map<ExprId, ExprId> replaceMap;
+        final boolean changed;
+
+        EliminateResult(LogicalAggregate<Plan> newAgg, Map<ExprId, ExprId> replaceMap, boolean changed) {
+            this.newAgg = newAgg;
+            this.replaceMap = replaceMap;
+            this.changed = changed;
+        }
     }
 
-    LogicalAggregate<Plan> eliminateGroupByKey(LogicalAggregate<? extends Plan> agg, Set<Slot> requireOutput) {
-        Set<Expression> removeExpression = findCanBeRemovedExpressions(agg, requireOutput,
+    EliminateResult eliminateGroupByKeyWithMap(LogicalAggregate<? extends Plan> agg, Set<Slot> requireOutput) {
+        FindResult result = findCanBeRemovedExpressionsInternal(agg, requireOutput,
                 agg.child().getLogicalProperties().getTrait());
+        Set<Expression> removeExpression = result.removeExpression;
+        Set<Expression> wrapWithAnyValue = result.wrapWithAnyValue;
+
         List<Expression> newGroupExpression = new ArrayList<>();
         for (Expression expression : agg.getGroupByExpressions()) {
-            if (!removeExpression.contains(expression)) {
+            if (!removeExpression.contains(expression)
+                    && !wrapWithAnyValue.contains(expression)) {
                 newGroupExpression.add(expression);
             }
         }
         List<NamedExpression> newOutput = new ArrayList<>();
+        Map<ExprId, ExprId> replaceMap = new HashMap<>();
+        boolean changed = !removeExpression.isEmpty() || !wrapWithAnyValue.isEmpty();
         for (NamedExpression expression : agg.getOutputExpressions()) {
-            if (!removeExpression.contains(expression)) {
-                newOutput.add(expression);
+            if (removeExpression.contains(expression)) {
+                continue;
             }
+            if (wrapWithAnyValue.contains(expression)) {
+                // expression is FD-redundant but needed in output: wrap with any_value
+                // Use fresh ExprId (auto-generated by Alias) to avoid ExprId collision,
+                // and record the mapping for rewriting upper plan references.
+                Alias newAlias = new Alias(new AnyValue(expression.toSlot()), expression.getName());
+                replaceMap.put(expression.getExprId(), newAlias.getExprId());
+                expression = newAlias;
+            }
+            newOutput.add(expression);
         }
-        return agg.withGroupByAndOutput(newGroupExpression, newOutput);
+        return new EliminateResult(agg.withGroupByAndOutput(newGroupExpression, newOutput), replaceMap, changed);
     }
 
     /**
-     * return removeExpression
+     * Return expressions that can be completely removed from both group-by and output.
+     * Kept for backward compatibility with external callers (e.g. PushDownAggThroughJoinOnPkFk).
      */
     public static Set<Expression> findCanBeRemovedExpressions(LogicalAggregate<? extends Plan> agg,
             Set<Slot> requireOutput, DataTrait dataTrait) {
+        FindResult result = findCanBeRemovedExpressionsInternal(agg, requireOutput, dataTrait);
+        return new HashSet<>(result.removeExpression);
+    }
+
+    /** Result of findCanBeRemovedExpressionsInternal: two sets of expressions. */
+    private static class FindResult {
+        final Set<Expression> removeExpression;   // remove from group-by and output
+        final Set<Expression> wrapWithAnyValue;   // remove from group-by, wrap with ANY_VALUE in output
+
+        FindResult(Set<Expression> removeExpression, Set<Expression> wrapWithAnyValue) {
+            this.removeExpression = removeExpression;
+            this.wrapWithAnyValue = wrapWithAnyValue;
+        }
+    }
+
+    private static FindResult findCanBeRemovedExpressionsInternal(LogicalAggregate<? extends Plan> agg,
+            Set<Slot> requireOutput, DataTrait dataTrait) {
         Map<Expression, Set<Slot>> groupBySlots = new HashMap<>();
         Set<Slot> validSlots = new HashSet<>();
         for (Expression expression : agg.getGroupByExpressions()) {
@@ -110,17 +250,24 @@ public static Set<Expression> findCanBeRemovedExpressions(LogicalAggregate<? ext
 
         FuncDeps funcDeps = dataTrait.getAllValidFuncDeps(validSlots);
         if (funcDeps.isEmpty()) {
-            return new HashSet<>();
+            return new FindResult(new HashSet<>(), new HashSet<>());
         }
 
         Set<Set<Slot>> minGroupBySlots = funcDeps.eliminateDeps(new HashSet<>(groupBySlots.values()), requireOutput);
         Set<Expression> removeExpression = new HashSet<>();
+        Set<Expression> wrapWithAnyValue = new HashSet<>();
         for (Entry<Expression, Set<Slot>> entry : groupBySlots.entrySet()) {
-            if (!minGroupBySlots.contains(entry.getValue())
-                    && !requireOutput.containsAll(entry.getValue())) {
-                removeExpression.add(entry.getKey());
+            if (!minGroupBySlots.contains(entry.getValue())) {
+                // FD redundant: can remove from group-by
+                if (!requireOutput.containsAll(entry.getValue())) {
+                    // Not needed in output either: remove completely
+                    removeExpression.add(entry.getKey());
+                } else {
+                    // Still needed in output: remove from group-by, wrap with ANY_VALUE in output
+                    wrapWithAnyValue.add(entry.getKey());
+                }
             }
         }
-        return removeExpression;
+        return new FindResult(removeExpression, wrapWithAnyValue);
     }
 }
@@ -76,7 +76,7 @@ public static Plan rewrite(LogicalAggregate<? extends Plan> agg, DistinctSelecto
         // construct cte consumer and aggregate
         List<LogicalAggregate<Plan>> newAggs = new ArrayList<>();
         // All otherAggFuncs are placed in the first one
-        Map<Alias, Alias> newToOriginDistinctFuncAlias = new HashMap<>();
+        Map<Alias, Alias> newToOriginDistinctFuncAlias = new LinkedHashMap<>();
         List<Expression> outputJoinGroupBys = new ArrayList<>();
         for (int i = 0; i < distinctFuncWithAliasReplaced.size(); ++i) {
             List<Alias> aliases = distinctFuncWithAliasReplaced.get(i);