[BugFix] Fix push down decimal window function error (backport #46199) (#46313)

Co-authored-by: Seaven <[email protected]>

Author: mergify[bot]

fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/tree/pdagg/PushDownDistinctAggregateRewriter.java

@@ -442,6 +442,10 @@ public RewriteInfo visitLogicalWindow(OptExpression optExpression, RewriteInfo r
                 Type[] argTypes = newArgs.stream().map(ScalarOperator::getType).toArray(Type[]::new);
                 Function newFunc = Expr.getBuiltinFunction(funcName, argTypes,
                         Function.CompareMode.IS_NONSTRICT_SUPERTYPE_OF);
+                if (FunctionSet.SUM.equals(funcName) && argTypes[0].isDecimalOfAnyVersion()) {
+                    newFunc = DecimalV3FunctionAnalyzer.rectifyAggregationFunction((AggregateFunction) newFunc,
+                            argTypes[0], rewriteCallOp.getType());
+                }
                 CallOperator newWindowCall = new CallOperator(funcName, newFunc.getReturnType(), newArgs, newFunc);
                 newWindowCalls.put(entry.getKey(), newWindowCall);
             }

fe/fe-core/src/test/java/com/starrocks/sql/plan/AggregatePushDownTest.java

@@ -218,4 +218,18 @@ public void testPruneColsAfterPushdownAgg_2() throws Exception {
                         "  |  \n" +
                         "  2:EXCHANGE");
     }
+
+    @Test
+    public void testPruneDistinctWindow() throws Exception {
+        String sql = "select distinct t1c, t1d, t1g, amount " +
+                " from (" +
+                " select  t1b, t1c, t1d, t1g, id_date, \n" +
+                "     sum(id_decimal)over(partition by t1c) as amount\n" +
+                "from test_all_type_not_null) tt";
+        String plan = getVerboseExplain(sql);
+        assertContains(plan, "  5:ANALYTIC\n" +
+                "  |  functions: [, sum[([12: sum, DECIMAL128(38,2), true]);" +
+                " args: DECIMAL128; result: DECIMAL128(38,2); args nullable: true; result nullable: true], ]");
+        assertContains(plan, "2:AGGREGATE (update finalize)");
+    }
 }