[fix](nereids) project distinct to agg checker exclude window function

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

@@ -62,7 +62,6 @@
 import org.apache.doris.nereids.trees.expressions.functions.table.TableValuedFunction;
 import org.apache.doris.nereids.trees.expressions.literal.IntegerLikeLiteral;
 import org.apache.doris.nereids.trees.expressions.literal.StringLiteral;
-import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitors;
 import org.apache.doris.nereids.trees.plans.JoinType;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.algebra.Aggregate;
@@ -739,9 +738,7 @@ private Plan bindProject(MatchingContext<LogicalProject<Plan>> ctx) {
     }
 
     private List<NamedExpression> adjustProjectionAggNullable(List<NamedExpression> expressions) {
-        boolean hasAggregation = expressions.stream()
-                .anyMatch(expr -> expr.accept(ExpressionVisitors.CONTAINS_AGGREGATE_CHECKER, null));
-        if (!hasAggregation) {
+        if (!ExpressionUtils.hasNonWindowAggregateFunction(expressions)) {
             return expressions;
         }
         boolean hasOnlyFullGroupBy = SqlModeHelper.hasOnlyFullGroupBy();

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

@@ -37,6 +37,7 @@
 import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
 import org.apache.doris.nereids.trees.plans.logical.LogicalSort;
 import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.util.ExpressionUtils;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
@@ -136,7 +137,7 @@ private void checkExpressionInputTypes(Plan plan) {
 
     private void checkAggregate(LogicalAggregate<? extends Plan> aggregate) {
         for (Expression expr : aggregate.getGroupByExpressions()) {
-            if (expr.anyMatch(AggregateFunction.class::isInstance)) {
+            if (ExpressionUtils.hasNonWindowAggregateFunction(expr)) {
                 throw new AnalysisException(
                         "GROUP BY expression must not contain aggregate functions: " + expr.toSql());
             }

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

@@ -19,16 +19,13 @@
 
 import org.apache.doris.nereids.rules.Rule;
 import org.apache.doris.nereids.rules.RuleType;
-import org.apache.doris.nereids.trees.expressions.NamedExpression;
-import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitors;
 import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
 import org.apache.doris.nereids.trees.plans.logical.LogicalOneRowRelation;
 import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.util.ExpressionUtils;
 
 import com.google.common.collect.ImmutableList;
 
-import java.util.List;
-
 /**
  * OneRowRelationExtractAggregate.
  * <p>
@@ -50,19 +47,13 @@ public class OneRowRelationExtractAggregate extends OneAnalysisRuleFactory {
     @Override
     public Rule build() {
         return RuleType.ONE_ROW_RELATION_EXTRACT_AGGREGATE.build(
-                logicalOneRowRelation().then(relation -> {
-                    List<NamedExpression> outputs = relation.getOutputs();
-                    boolean needGlobalAggregate = outputs
-                            .stream()
-                            .anyMatch(p -> p.accept(ExpressionVisitors.CONTAINS_AGGREGATE_CHECKER, null));
-                    if (needGlobalAggregate) {
-                        LogicalRelation newRelation = new LogicalOneRowRelation(relation.getRelationId(),
-                                ImmutableList.of());
-                        return new LogicalAggregate<>(ImmutableList.of(), relation.getOutputs(), newRelation);
-                    } else {
-                        return relation;
-                    }
-                })
+                logicalOneRowRelation()
+                       .when(relation -> ExpressionUtils.hasNonWindowAggregateFunction(relation.getOutputs()))
+                       .then(relation -> {
+                           LogicalRelation newRelation = new LogicalOneRowRelation(relation.getRelationId(),
+                                   ImmutableList.of());
+                           return new LogicalAggregate<>(ImmutableList.of(), relation.getOutputs(), newRelation);
+                       })
         );
     }
 }

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

@@ -19,9 +19,8 @@
 
 import org.apache.doris.nereids.rules.Rule;
 import org.apache.doris.nereids.rules.RuleType;
-import org.apache.doris.nereids.trees.expressions.NamedExpression;
-import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitors;
 import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.util.ExpressionUtils;
 
 import com.google.common.collect.ImmutableList;
 
@@ -44,21 +43,9 @@ public class ProjectToGlobalAggregate extends OneAnalysisRuleFactory {
     @Override
     public Rule build() {
         return RuleType.PROJECT_TO_GLOBAL_AGGREGATE.build(
-           logicalProject().then(project -> {
-               boolean needGlobalAggregate = false;
-               for (NamedExpression output : project.getProjects()) {
-                   if (output.accept(ExpressionVisitors.CONTAINS_AGGREGATE_CHECKER, null)) {
-                       needGlobalAggregate = true;
-                       break;
-                   }
-               }
-
-               if (needGlobalAggregate) {
-                   return new LogicalAggregate<>(ImmutableList.of(), project.getProjects(), project.child());
-               } else {
-                   return project;
-               }
-           })
+           logicalProject()
+                   .when(project -> ExpressionUtils.hasNonWindowAggregateFunction(project.getProjects()))
+                   .then(project -> new LogicalAggregate<>(ImmutableList.of(), project.getProjects(), project.child()))
         );
     }
 }

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

@@ -19,10 +19,9 @@
 
 import org.apache.doris.nereids.rules.Rule;
 import org.apache.doris.nereids.rules.RuleType;
-import org.apache.doris.nereids.trees.expressions.Expression;
-import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
 import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
 import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.util.ExpressionUtils;
 
 /**
  * ProjectWithDistinctToAggregate.
@@ -46,12 +45,8 @@ public Rule build() {
         return RuleType.PROJECT_WITH_DISTINCT_TO_AGGREGATE.build(
             logicalProject()
                 .when(LogicalProject::isDistinct)
-                .whenNot(project -> project.getProjects().stream().anyMatch(this::hasAggregateFunction))
+                .whenNot(project -> ExpressionUtils.hasNonWindowAggregateFunction(project.getProjects()))
                 .then(project -> new LogicalAggregate<>(project.getProjects(), false, project.child()))
         );
     }
-
-    private boolean hasAggregateFunction(Expression expression) {
-        return expression.anyMatch(AggregateFunction.class::isInstance);
-    }
 }

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

@@ -64,6 +64,7 @@
 import org.apache.doris.nereids.trees.expressions.literal.StringLiteral;
 import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
 import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
+import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitors;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalEmptyRelation;
 import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
@@ -1186,4 +1187,23 @@ public static String slotListShapeInfo(List<Slot> materializedSlots) {
         shapeBuilder.append(")");
         return shapeBuilder.toString();
     }
+
+    /**
+     * has aggregate function, exclude the window function
+     */
+    public static boolean hasNonWindowAggregateFunction(Collection<? extends Expression> expressions) {
+        for (Expression expression : expressions) {
+            if (hasNonWindowAggregateFunction(expression)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    /**
+     * has aggregate function, exclude the window function
+     */
+    public static boolean hasNonWindowAggregateFunction(Expression expression) {
+        return expression.accept(ExpressionVisitors.CONTAINS_AGGREGATE_CHECKER, null);
+    }
 }

regression-test/data/nereids_rules_p0/project_distinct_to_agg/project_distinct_to_agg.out

@@ -0,0 +1,53 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !with_windows_1_shape --
+PhysicalResultSink
+--hashAgg[GLOBAL]
+----hashAgg[LOCAL]
+------PhysicalProject
+--------PhysicalWindow
+----------PhysicalProject
+------------PhysicalOlapScan[tbl_project_distinct_to_agg]
+
+-- !with_windows_1_result --
+1	12
+2	12
+
+-- !with_windows_2_shape --
+PhysicalResultSink
+--hashAgg[GLOBAL]
+----hashAgg[LOCAL]
+------PhysicalProject
+--------PhysicalWindow
+----------PhysicalQuickSort[LOCAL_SORT]
+------------PhysicalUnion
+
+-- !with_windows_2_result --
+100
+
+-- !order_by_shape --
+PhysicalResultSink
+--PhysicalLimit[GLOBAL]
+----PhysicalUnion
+
+-- !order_by_result --
+101
+
+-- !constant_shape --
+PhysicalResultSink
+--PhysicalLimit[GLOBAL]
+----PhysicalLimit[LOCAL]
+------PhysicalProject
+--------PhysicalStorageLayerAggregate[tbl_project_distinct_to_agg]
+
+-- !constant_result --
+1	2	3
+
+-- !agg_shape --
+PhysicalResultSink
+--hashAgg[GLOBAL]
+----hashAgg[LOCAL]
+------PhysicalOlapScan[tbl_project_distinct_to_agg]
+
+-- !agg_result --
+7
+

regression-test/plugins/plugin_planner.groovy

@@ -0,0 +1,28 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.apache.doris.regression.suite.Suite
+
+Suite.metaClass.explainAndResult = { String tag, String sql ->
+    "qt_${tag}_shape"          "explain shape plan ${sql}"
+    "qt_${tag}_result"         "${sql}"
+}
+
+Suite.metaClass.explainAndOrderResult = { String tag, String sql ->
+    "qt_${tag}_shape"          "explain shape plan ${sql}"
+    "order_qt_${tag}_result"   "${sql}"
+}

regression-test/suites/nereids_rules_p0/project_distinct_to_agg/project_distinct_to_agg.groovy

@@ -0,0 +1,32 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS
+// OF ANY KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite('project_distinct_to_agg') {
+    def tbl = 'tbl_project_distinct_to_agg'
+    sql "SET ignore_shape_nodes='PhysicalDistribute'"
+    sql "drop table if exists ${tbl} force"
+    sql "create table ${tbl} (a int) distributed by hash(a) buckets 10 properties ('replication_num' = '1')"
+    sql "insert into ${tbl} values (1), (1), (1), (2), (2)"
+
+    explainAndOrderResult 'with_windows_1',  "select distinct a, max(a + 10) over() from ${tbl}"
+    explainAndOrderResult 'with_windows_2',  'select distinct sum(value) over(partition by id) from (select 100 value, 1 id union all select 100, 2)a'
+    explainAndResult 'order_by',  'select distinct value+1 from (select 100 value, 1 id union all select 100, 2)a order by value+1'
+    explainAndOrderResult 'constant',  "select distinct 1, 2, 3 from ${tbl}"
+    explainAndOrderResult 'agg',  "select distinct sum(a) from ${tbl}"
+
+    sql "drop table if exists ${tbl} force"
+}