branch-3.0: [fix](nereids) fix distinct window compute wrong result (#48987)

fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Analyzer.java

@@ -30,7 +30,6 @@
 import org.apache.doris.nereids.rules.analysis.CheckPolicy;
 import org.apache.doris.nereids.rules.analysis.CollectJoinConstraint;
 import org.apache.doris.nereids.rules.analysis.CollectSubQueryAlias;
-import org.apache.doris.nereids.rules.analysis.EliminateDistinctConstant;
 import org.apache.doris.nereids.rules.analysis.EliminateGroupByConstant;
 import org.apache.doris.nereids.rules.analysis.EliminateLogicalSelectHint;
 import org.apache.doris.nereids.rules.analysis.FillUpMissingSlots;
@@ -41,7 +40,6 @@
 import org.apache.doris.nereids.rules.analysis.NormalizeRepeat;
 import org.apache.doris.nereids.rules.analysis.OneRowRelationExtractAggregate;
 import org.apache.doris.nereids.rules.analysis.ProjectToGlobalAggregate;
-import org.apache.doris.nereids.rules.analysis.ProjectWithDistinctToAggregate;
 import org.apache.doris.nereids.rules.analysis.ReplaceExpressionByChildOutput;
 import org.apache.doris.nereids.rules.analysis.SubqueryToApply;
 import org.apache.doris.nereids.rules.analysis.VariableToLiteral;
@@ -106,13 +104,6 @@ private static List<RewriteJob> buildAnalyzerJobs() {
             bottomUp(new AddInitMaterializationHook()),
             bottomUp(
                     new ProjectToGlobalAggregate(),
-                    // this rule check's the logicalProject node's isDistinct property
-                    // and replace the logicalProject node with a LogicalAggregate node
-                    // so any rule before this, if create a new logicalProject node
-                    // should make sure isDistinct property is correctly passed around.
-                    // please see rule BindSlotReference or BindFunction for example
-                    new EliminateDistinctConstant(),
-                    new ProjectWithDistinctToAggregate(),
                     new ReplaceExpressionByChildOutput(),
                     new OneRowRelationExtractAggregate()
             ),

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/ProjectToGlobalAggregate.java

@@ -17,13 +17,24 @@
 
 package org.apache.doris.nereids.rules.analysis;
 
+import org.apache.doris.nereids.exceptions.AnalysisException;
 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.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
 import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitors;
+import org.apache.doris.nereids.trees.plans.LimitPhase;
+import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
 
 import com.google.common.collect.ImmutableList;
 
+import java.util.List;
+
 /**
  * ProjectToGlobalAggregate.
  * <p>
@@ -43,17 +54,110 @@ public class ProjectToGlobalAggregate extends OneAnalysisRuleFactory {
     @Override
     public Rule build() {
         return RuleType.PROJECT_TO_GLOBAL_AGGREGATE.build(
-           logicalProject().then(project -> {
-               boolean needGlobalAggregate = project.getProjects()
-                       .stream()
-                       .anyMatch(p -> p.accept(ExpressionVisitors.CONTAINS_AGGREGATE_CHECKER, null));
-
-               if (needGlobalAggregate) {
-                   return new LogicalAggregate<>(ImmutableList.of(), project.getProjects(), project.child());
-               } else {
-                   return project;
-               }
-           })
+            logicalProject().then(project -> {
+                project = distinctConstantsToLimit1(project);
+                Plan result = projectToAggregate(project);
+                return distinctToAggregate(result, project);
+            })
         );
     }
+
+    // select distinct 1,2,3 from tbl
+    //               ↓
+    // select 1,2,3 from (select 1, 2, 3 from tbl limit 1) as tmp
+    private static LogicalProject<Plan> distinctConstantsToLimit1(LogicalProject<Plan> project) {
+        if (!project.isDistinct()) {
+            return project;
+        }
+
+        boolean allSelectItemAreConstants = true;
+        for (NamedExpression selectItem : project.getProjects()) {
+            if (!selectItem.isConstant()) {
+                allSelectItemAreConstants = false;
+                break;
+            }
+        }
+
+        if (allSelectItemAreConstants) {
+            return new LogicalProject<>(
+                    project.getProjects(),
+                    new LogicalLimit<>(1, 0, LimitPhase.ORIGIN, project.child())
+            );
+        }
+        return project;
+    }
+
+    // select avg(xxx) from tbl
+    //         ↓
+    // LogicalAggregate(groupBy=[], output=[avg(xxx)])
+    private static Plan projectToAggregate(LogicalProject<Plan> project) {
+        // contains aggregate functions, like sum, avg ?
+        for (NamedExpression selectItem : project.getProjects()) {
+            if (selectItem.accept(ExpressionVisitors.CONTAINS_AGGREGATE_CHECKER, null)) {
+                return new LogicalAggregate<>(ImmutableList.of(), project.getProjects(), project.child());
+            }
+        }
+        return project;
+    }
+
+    private static Plan distinctToAggregate(Plan result, LogicalProject<Plan> originProject) {
+        if (!originProject.isDistinct()) {
+            return result;
+        }
+        if (result instanceof LogicalProject) {
+            // remove distinct: select distinct fun(xxx) as c1 from tbl
+            //
+            // LogicalProject(distinct=true, output=[fun(xxx) as c1])
+            //                  ↓
+            // LogicalAggregate(groupBy=[c1], output=[c1])
+            //                  |
+            //   LogicalProject(output=[fun(xxx) as c1])
+            LogicalProject<?> project = (LogicalProject<?>) result;
+
+            ImmutableList.Builder<NamedExpression> bottomProjectOutput
+                    = ImmutableList.builderWithExpectedSize(project.getProjects().size());
+            ImmutableList.Builder<NamedExpression> topAggOutput
+                    = ImmutableList.builderWithExpectedSize(project.getProjects().size());
+
+            boolean hasComplexExpr = false;
+            for (NamedExpression selectItem : project.getProjects()) {
+                if (selectItem.isSlot()) {
+                    topAggOutput.add(selectItem);
+                    bottomProjectOutput.add(selectItem);
+                } else if (isAliasLiteral(selectItem)) {
+                    // stay in agg, and eliminate by `ELIMINATE_GROUP_BY_CONSTANT`
+                    topAggOutput.add(selectItem);
+                } else {
+                    // `FillUpMissingSlots` not support find complex expr in aggregate,
+                    // so we should push down into the bottom project
+                    hasComplexExpr = true;
+                    topAggOutput.add(selectItem.toSlot());
+                    bottomProjectOutput.add(selectItem);
+                }
+            }
+
+            if (!hasComplexExpr) {
+                List<Slot> projects = (List) project.getProjects();
+                return new LogicalAggregate(projects, projects, project.child());
+            }
+
+            LogicalProject<?> removeDistinct = new LogicalProject<>(bottomProjectOutput.build(), project.child());
+            ImmutableList<NamedExpression> aggOutput = topAggOutput.build();
+            return new LogicalAggregate(aggOutput, aggOutput, removeDistinct);
+        } else if (result instanceof LogicalAggregate) {
+            // remove distinct: select distinct avg(xxx) as c1 from tbl
+            //
+            // LogicalProject(distinct=true, output=[avg(xxx) as c1])
+            //                  ↓
+            //  LogicalAggregate(output=[avg(xxx) as c1])
+            return result;
+        } else {
+            // never reach
+            throw new AnalysisException("Unsupported");
+        }
+    }
+
+    private static boolean isAliasLiteral(NamedExpression selectItem) {
+        return selectItem instanceof Alias && selectItem.child(0) instanceof Literal;
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/ReplaceExpressionByChildOutput.java

@@ -53,21 +53,27 @@ public List<Rule> buildRules() {
                 ))
                 .add(RuleType.REPLACE_SORT_EXPRESSION_BY_CHILD_OUTPUT.build(
                         logicalSort(logicalAggregate()).then(sort -> {
-                            LogicalAggregate<Plan> aggregate = sort.child();
-                            Map<Expression, Slot> sMap = buildOutputAliasMap(aggregate.getOutputExpressions());
+                            LogicalAggregate<Plan> agg = sort.child();
+                            Map<Expression, Slot> sMap = buildOutputAliasMap(agg.getOutputExpressions());
+                            if (sMap.isEmpty() && isSelectDistinct(agg)) {
+                                sMap = getSelectDistinctExpressions(agg);
+                            }
                             return replaceSortExpression(sort, sMap);
                         })
                 )).add(RuleType.REPLACE_SORT_EXPRESSION_BY_CHILD_OUTPUT.build(
                         logicalSort(logicalHaving(logicalAggregate())).then(sort -> {
-                            LogicalAggregate<Plan> aggregate = sort.child().child();
-                            Map<Expression, Slot> sMap = buildOutputAliasMap(aggregate.getOutputExpressions());
+                            LogicalAggregate<Plan> agg = sort.child().child();
+                            Map<Expression, Slot> sMap = buildOutputAliasMap(agg.getOutputExpressions());
+                            if (sMap.isEmpty() && isSelectDistinct(agg)) {
+                                sMap = getSelectDistinctExpressions(agg);
+                            }
                             return replaceSortExpression(sort, sMap);
                         })
                 ))
                 .build();
     }
 
-    private Map<Expression, Slot> buildOutputAliasMap(List<NamedExpression> output) {
+    private static Map<Expression, Slot> buildOutputAliasMap(List<NamedExpression> output) {
         Map<Expression, Slot> sMap = Maps.newHashMapWithExpectedSize(output.size());
         for (NamedExpression expr : output) {
             if (expr instanceof Alias) {
@@ -93,4 +99,22 @@ private LogicalPlan replaceSortExpression(LogicalSort<? extends LogicalPlan> sor
 
         return changed ? new LogicalSort<>(newKeys.build(), sort.child()) : sort;
     }
+
+    private static boolean isSelectDistinct(LogicalAggregate<? extends Plan> agg) {
+        return agg.getGroupByExpressions().equals(agg.getOutputExpressions())
+                && agg.getGroupByExpressions().equals(agg.child().getOutput());
+    }
+
+    private static Map<Expression, Slot> getSelectDistinctExpressions(LogicalAggregate<? extends Plan> agg) {
+        Plan child = agg.child();
+        List<NamedExpression> selectItems;
+        if (child instanceof LogicalProject) {
+            selectItems = ((LogicalProject<?>) child).getProjects();
+        } else if (child instanceof LogicalAggregate) {
+            selectItems = ((LogicalAggregate<?>) child).getOutputExpressions();
+        } else {
+            selectItems = ImmutableList.of();
+        }
+        return buildOutputAliasMap(selectItems);
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/ExpressionColumnFilterConverter.java

@@ -21,6 +21,7 @@
 import org.apache.doris.analysis.LiteralExpr;
 import org.apache.doris.analysis.NullLiteral;
 import org.apache.doris.analysis.SlotRef;
+import org.apache.doris.common.Config;
 import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
 import org.apache.doris.nereids.trees.expressions.EqualTo;
 import org.apache.doris.nereids.trees.expressions.Expression;
@@ -99,6 +100,10 @@ public Expression visitComparisonPredicate(ComparisonPredicate predicate, Void u
 
     @Override
     public Expression visitInPredicate(InPredicate predicate, Void unused) {
+        if (predicate.getOptions().size() > Config.max_distribution_pruner_recursion_depth) {
+            return null;
+        }
+
         List<Expr> literals = predicate.getOptions().stream()
                 .map(expr -> ((Expr) ((Literal) expr).toLegacyLiteral()))
                 .collect(Collectors.toList());

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

@@ -18,6 +18,7 @@
 package org.apache.doris.nereids.util;
 
 import org.apache.doris.catalog.TableIf.TableType;
+import org.apache.doris.common.Config;
 import org.apache.doris.common.MaterializedViewException;
 import org.apache.doris.common.NereidsException;
 import org.apache.doris.common.Pair;
@@ -808,7 +809,8 @@ public static Optional<Expression> checkAndMaybeCommute(Expression expression) {
         }
         if (expression instanceof InPredicate) {
             InPredicate predicate = ((InPredicate) expression);
-            if (!predicate.getCompareExpr().isSlot()) {
+            if (!predicate.getCompareExpr().isSlot()
+                    || predicate.getOptions().size() > Config.max_distribution_pruner_recursion_depth) {
                 return Optional.empty();
             }
             return Optional.ofNullable(predicate.getOptions().stream()