[fix](mtmv) check groupby in agg-bottom-plan when rewrite agg query by mv (#34274)

check groupby in agg-bottom-plan when rewrite and rollup agg query by mv

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewAggregateRule.java

@@ -230,6 +230,7 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
         // split the query top plan expressions to group expressions and functions, if can not, bail out.
         Pair<Set<? extends Expression>, Set<? extends Expression>> queryGroupAndFunctionPair
                 = topPlanSplitToGroupAndFunction(queryTopPlanAndAggPair, queryStructInfo);
+        Set<? extends Expression> queryTopPlanGroupBySet = queryGroupAndFunctionPair.key();
         Set<? extends Expression> queryTopPlanFunctionSet = queryGroupAndFunctionPair.value();
         // try to rewrite, contains both roll up aggregate functions and aggregate group expression
         List<NamedExpression> finalOutputExpressions = new ArrayList<>();
@@ -284,6 +285,31 @@ public abstract class AbstractMaterializedViewAggregateRule extends AbstractMate
         }
         // add project to guarantee group by column ref is slot reference,
         // this is necessary because physical createHash will need slotReference later
+        if (queryGroupByExpressions.size() != queryTopPlanGroupBySet.size()) {
+            for (Expression expression : queryGroupByExpressions) {
+                if (queryTopPlanGroupBySet.contains(expression)) {
+                    continue;
+                }
+                Expression queryGroupShuttledExpr = ExpressionUtils.shuttleExpressionWithLineage(
+                        expression, queryTopPlan, queryStructInfo.getTableBitSet());
+                AggregateExpressionRewriteContext context = new AggregateExpressionRewriteContext(true,
+                        mvExprToMvScanExprQueryBased, queryTopPlan, queryStructInfo.getTableBitSet());
+                // group by expression maybe group by a + b, so we need expression rewriter
+                Expression rewrittenGroupByExpression = queryGroupShuttledExpr.accept(AGGREGATE_EXPRESSION_REWRITER,
+                        context);
+                if (!context.isValid()) {
+                    // group expr can not rewrite by view
+                    materializationContext.recordFailReason(queryStructInfo,
+                            "View dimensions doesn't not cover the query dimensions in bottom agg ",
+                            () -> String.format("mvExprToMvScanExprQueryBased is %s,\n queryGroupShuttledExpr is %s",
+                                    mvExprToMvScanExprQueryBased, queryGroupShuttledExpr));
+                    return null;
+                }
+                NamedExpression groupByExpression = rewrittenGroupByExpression instanceof NamedExpression
+                        ? (NamedExpression) rewrittenGroupByExpression : new Alias(rewrittenGroupByExpression);
+                finalGroupExpressions.add(groupByExpression);
+            }
+        }
         List<Expression> copiedFinalGroupExpressions = new ArrayList<>(finalGroupExpressions);
         List<NamedExpression> projectsUnderAggregate = copiedFinalGroupExpressions.stream()
                 .map(NamedExpression.class::cast)

regression-test/data/nereids_rules_p0/mv/agg_with_roll_up/aggregate_with_roll_up.out

@@ -341,3 +341,17 @@
 2023-12-11	2	mm	4	\N	1
 2023-12-12	2	mi	5	\N	2
 
+-- !query32_0_before --
+2023-12-08	2
+2023-12-09	1
+2023-12-10	2
+2023-12-11	1
+2023-12-12	2
+
+-- !query32_0_after --
+2023-12-08	2
+2023-12-09	1
+2023-12-10	2
+2023-12-11	1
+2023-12-12	2
+

regression-test/suites/nereids_rules_p0/mv/agg_with_roll_up/aggregate_with_roll_up.groovy

@@ -1360,4 +1360,29 @@ suite("aggregate_with_roll_up") {
     order_qt_query31_0_after "${query31_0}"
     sql """ DROP MATERIALIZED VIEW IF EXISTS mv31_0"""
 
+    // should rewrite fail, because the part of query is join but mv is aggregate
+    def mv32_0 = """
+            select 
+              o_orderdate, 
+              count(*) 
+            from 
+              orders 
+            group by 
+              o_orderdate;
+    """
+    def query32_0 = """
+            select 
+              o_orderdate, 
+              count(*) 
+            from 
+              orders 
+            group by 
+              o_orderdate, 
+              o_shippriority;
+    """
+    order_qt_query32_0_before "${query32_0}"
+    check_mv_rewrite_fail(db, mv32_0, query32_0, "mv32_0")
+    order_qt_query32_0_after "${query32_0}"
+    sql """ DROP MATERIALIZED VIEW IF EXISTS mv32_0"""
+
 }