[fix](nereids) fix merge project with window function bug (#19280)

1. don't merge projects if any window function exists
2. bypass SimplifyArithmeticRule for decimalV3 type

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/SimplifyArithmeticRule.java

@@ -81,6 +81,10 @@ public class SimplifyArithmeticRule extends AbstractExpressionRewriteRule {
         List<Operand> variables = Lists.newArrayList();
         List<Operand> constants = Lists.newArrayList();
 
+        // TODO currently we don't process decimalV3 for simplicity.
+        if (flattedExpressions.stream().anyMatch(operand -> operand.expression.getDataType().isDecimalV3Type())) {
+            return arithmetic;
+        }
         // 2. move variables to left side and move constants to right sid.
         flattedExpressions.forEach(operand -> {
             if (operand.expression.isConstant()) {

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/MergeProjects.java

@@ -21,6 +21,7 @@ import org.apache.doris.nereids.rules.Rule;
 import org.apache.doris.nereids.rules.RuleType;
 import org.apache.doris.nereids.rules.rewrite.OneRewriteRuleFactory;
 import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.WindowExpression;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
 
@@ -40,9 +41,12 @@ public class MergeProjects extends OneRewriteRuleFactory {
 
     @Override
     public Rule build() {
+        // TODO modify ExtractAndNormalizeWindowExpression to handle nested window functions
+        // here we just don't merge two projects if there is any window function
         return logicalProject(logicalProject())
-                .then(project -> mergeProjects(project))
-                .toRule(RuleType.MERGE_PROJECTS);
+                .whenNot(project -> containsWindowExpression(project.getProjects())
+                        && containsWindowExpression(project.child().getProjects()))
+                .then(project -> mergeProjects(project)).toRule(RuleType.MERGE_PROJECTS);
     }
 
     public static Plan mergeProjects(LogicalProject project) {
@@ -51,4 +55,8 @@ public class MergeProjects extends OneRewriteRuleFactory {
         LogicalProject newProject = childProject.canEliminate() ? project : childProject;
         return newProject.withProjectsAndChild(projectExpressions, childProject.child(0));
     }
+
+    private boolean containsWindowExpression(List<NamedExpression> expressions) {
+        return expressions.stream().anyMatch(expr -> expr.anyMatch(WindowExpression.class::isInstance));
+    }
 }

regression-test/data/nereids_p0/aggregate/agg_window_project.out

@@ -0,0 +1,7 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select1 --
+1	1	1
+
+-- !select2 --
+10.0000000000
+

regression-test/suites/nereids_p0/aggregate/agg_window_project.groovy

@@ -0,0 +1,68 @@
+/*
+ * 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("agg_window_project") {
+    sql "SET enable_nereids_planner=true"
+    sql "SET enable_fallback_to_original_planner=false"
+    sql "DROP TABLE IF EXISTS test_window_table;"
+    sql """
+        create table test_window_table
+        (
+            a varchar(100) null,
+            b decimalv3(18,10) null
+        ) ENGINE=OLAP
+        DUPLICATE KEY(`a`)
+        DISTRIBUTED BY HASH(`a`) BUCKETS 1
+        PROPERTIES (
+        "replication_allocation" = "tag.location.default: 1"
+        );
+    """
+
+    sql """insert into test_window_table values("1", 1);"""
+
+
+    order_qt_select1 """
+        SELECT `owner`,
+         `melody_num`,
+         first_value(`owner`) over(partition by `owner`,
+        `melody_num`
+        ORDER BY  `super_owner_cnt` DESC) AS `super_owner`
+        FROM 
+            (SELECT `owner`,
+                `melody_num`,
+                count(`owner`)
+                OVER (partition by `owner`) AS `super_owner_cnt`
+            FROM 
+                (SELECT `owner`,
+                `melody_num`
+                FROM 
+                    (SELECT `owner`,
+                sum(`melody_num`) AS `melody_num`
+                    FROM ( 
+                        (SELECT `test_window_table`.`a` AS `owner`,
+                count(*) AS `melody_num`
+                        FROM `test_window_table`
+                        GROUP BY  `test_window_table`.`a` ) ) `owner_melody_num_data`
+                        GROUP BY  `owner` ) `owner_list` ) `super_owner_mapping_raw` ) `super_owner_mapping_raw2`; 
+    """
+
+    order_qt_select2 """select b / 1 * 10 from test_window_table;"""
+
+    sql "DROP TABLE IF EXISTS test_window_table;"
+}