[fix](mv) Fix use sync mv wrongly when use rbo materialized view rewrite rule

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/mv/SelectMaterializedIndexWithAggregate.java

@@ -42,6 +42,7 @@
 import org.apache.doris.nereids.trees.expressions.VirtualSlotReference;
 import org.apache.doris.nereids.trees.expressions.WhenClause;
 import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Avg;
 import org.apache.doris.nereids.trees.expressions.functions.agg.BitmapUnion;
 import org.apache.doris.nereids.trees.expressions.functions.agg.BitmapUnionCount;
 import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
@@ -859,7 +860,8 @@ private PreAggStatus checkAggFunc(
                         matchingAggType, normalizeName(aggFunc.child(0).toSql())));
 
             boolean contains = containsAllColumn(aggFunc.child(0), ctx.keyNameToColumn.keySet());
-            if (contains || ctx.keyNameToColumn.containsKey(childNameWithFuncName)) {
+            if ((contains || ctx.keyNameToColumn.containsKey(childNameWithFuncName))
+                    && checkWhenUseKey(aggFunc, matchingAggType)) {
                 if (canUseKeyColumn || ctx.isDupKeysOrMergeOnWrite || (!ctx.isBaseIndex() && contains)) {
                     return PreAggStatus.on();
                 } else {
@@ -943,8 +945,7 @@ private PreAggStatus checkSubExpressions(AggregateFunction aggFunc, AggregateTyp
                             returnExp.toSql(), matchingAggType));
                 }
                 if (ctx.keyNameToColumn.containsKey(exprName)) {
-                    if (matchingAggType != AggregateType.MAX && matchingAggType != AggregateType.MIN
-                            && (aggFunc instanceof Count && !aggFunc.isDistinct())) {
+                    if (!checkWhenUseKey(aggFunc, matchingAggType)) {
                         return PreAggStatus.off("agg on key column should be MAX, MIN or COUNT DISTINCT.");
                     }
                 }
@@ -981,6 +982,15 @@ private PreAggStatus checkSubExpressions(AggregateFunction aggFunc, AggregateTyp
         }
     }
 
+    // agg on key column should be MAX, MIN, COUNT DISTINCT, SUM DISTINCT, AVG DISTINCT. return true if valid
+    private static boolean checkWhenUseKey(AggregateFunction aggFunc, AggregateType matchingAggType) {
+        return matchingAggType == AggregateType.MAX
+                || matchingAggType == AggregateType.MIN
+                || (aggFunc instanceof Sum && aggFunc.isDistinct())
+                || (aggFunc instanceof Count && aggFunc.isDistinct())
+                || (aggFunc instanceof Avg && aggFunc.isDistinct());
+    }
+
     private static class CheckContext {
 
         public final LogicalOlapScan scan;

regression-test/data/mv_p0/agg_use_key_direct/agg_use_key_direct.out

@@ -0,0 +1,16 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select_count --
+100
+
+-- !select_sum --
+4950
+
+-- !select_max --
+-4
+
+-- !select_min --
+-4
+
+-- !select_avg --
+-4
+

regression-test/suites/mv_p0/agg_use_key_direct/agg_use_key_direct.groovy

@@ -0,0 +1,63 @@
+// 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_use_key_direct") {
+
+    String db = context.config.getDbNameByFile(context.file)
+
+    def tblName = "agg_use_key_direct"
+
+    sql "drop table if exists ${tblName} force;"
+    sql """
+        create table ${tblName} (
+            k1 int null,
+            k2 int not null,
+            k3 bigint null,
+            k4 bigint null,
+            k5 varchar(100) null
+        )
+        duplicate key (k1, k2, k3)
+        distributed by hash(k1) buckets 3
+        properties("replication_num" = "1");
+    """
+    sql "insert into ${tblName} select e1, -4, -4, -4, 'd' from (select 1 k1) as t lateral view explode_numbers(100) tmp1 as e1;"
+    create_sync_mv(db, tblName, "common_mv", """select k1, k3, sum(k2), count(k4) from ${tblName} group by k1, k3;""")
+
+    if (enable_sync_mv_cost_based_rewrite()) {
+        sql """set enable_sync_mv_cost_based_rewrite = false;"""
+    }
+
+    mv_rewrite_fail("""select count(k1) from agg_use_key_direct""", "common_mv")
+    mv_rewrite_fail("""select sum(k1) from agg_use_key_direct""", "common_mv")
+    mv_rewrite_fail("""select avg(k3) from agg_use_key_direct""", "common_mv")
+
+
+    mv_rewrite_success("""select count(distinct k1) from agg_use_key_direct""", "common_mv")
+    order_qt_select_count """select count(distinct k1) from agg_use_key_direct"""
+
+    mv_rewrite_success("""select sum(distinct k1) from agg_use_key_direct""", "common_mv")
+    order_qt_select_sum """select sum(distinct k1) from agg_use_key_direct"""
+
+    mv_rewrite_success("""select max(distinct k3) from agg_use_key_direct""", "common_mv")
+    order_qt_select_max """select max(distinct k3) from agg_use_key_direct"""
+
+    mv_rewrite_success("""select min(distinct k3) from agg_use_key_direct""", "common_mv")
+    order_qt_select_min """select min(distinct k3) from agg_use_key_direct"""
+
+    mv_rewrite_success("""select avg(distinct k3) from agg_use_key_direct""", "common_mv")
+    order_qt_select_avg """select min(distinct k3) from agg_use_key_direct"""
+}