planner: fix having clause item that ref-ed to grouping set expression shouldn't be pushed down through Expand OP (#45651)

close #45647

Author: AilinKid

planner/core/casetest/enforcempp/testdata/enforce_mpp_suite_in.json

@@ -209,8 +209,9 @@
       "explain format = 'brief' select grouping(a,b) from t group by a, b with rollup; -- 5. grouping function contains grouping set column a,c",
       "explain format = 'brief' select a, grouping(b,a) from t group by a,b with rollup; -- 6. resolve normal column a to grouping set column a'",
       "explain format = 'brief' select a+1, grouping(b) from t group by a+1, b with rollup; -- 7. resolve field list a+1 to grouping set column a+1",
-      "explain format = 'brief' SELECT SUM(profit) AS profit FROM sales GROUP BY year+2, year+profit WITH ROLLUP order by year+2",
-      "explain format = 'brief' SELECT year+2, SUM(profit) AS profit FROM sales GROUP BY year+2, year+profit WITH ROLLUP order by year+2"
+      "explain format = 'brief' SELECT SUM(profit) AS profit FROM sales GROUP BY year+2, year+profit WITH ROLLUP order by year+2; -- 8. order by item year+2 resolve to gby grouping expression",
+      "explain format = 'brief' SELECT year+2, SUM(profit) AS profit FROM sales GROUP BY year+2, year+profit WITH ROLLUP order by year+2; -- 9. order by item year+2 resolve to select field",
+      "explain format = 'brief' SELECT year+2 as y, SUM(profit) as profit FROM sales GROUP BY year+2, year+profit WITH ROLLUP having y > 2002 order by year+2, profit; -- 10. having (year+2) shouldn't be pushed down"
     ]
   }
 ]

planner/core/casetest/enforcempp/testdata/enforce_mpp_suite_out.json

@@ -2049,7 +2049,7 @@
         "Warn": null
       },
       {
-        "SQL": "explain format = 'brief' SELECT SUM(profit) AS profit FROM sales GROUP BY year+2, year+profit WITH ROLLUP order by year+2",
+        "SQL": "explain format = 'brief' SELECT SUM(profit) AS profit FROM sales GROUP BY year+2, year+profit WITH ROLLUP order by year+2; -- 8. order by item year+2 resolve to gby grouping expression",
         "Plan": [
           "Projection 8000.00 root  Column#9",
           "└─Sort 8000.00 root  Column#6",
@@ -2067,7 +2067,7 @@
         "Warn": null
       },
       {
-        "SQL": "explain format = 'brief' SELECT year+2, SUM(profit) AS profit FROM sales GROUP BY year+2, year+profit WITH ROLLUP order by year+2",
+        "SQL": "explain format = 'brief' SELECT year+2, SUM(profit) AS profit FROM sales GROUP BY year+2, year+profit WITH ROLLUP order by year+2; -- 9. order by item year+2 resolve to select field",
         "Plan": [
           "Projection 8000.00 root  Column#6->Column#10, Column#9",
           "└─Sort 8000.00 root  Column#6",
@@ -2083,6 +2083,25 @@
           "                    └─TableFullScan 10000.00 mpp[tiflash] table:sales keep order:false, stats:pseudo"
         ],
         "Warn": null
+      },
+      {
+        "SQL": "explain format = 'brief' SELECT year+2 as y, SUM(profit) as profit FROM sales GROUP BY year+2, year+profit WITH ROLLUP having y > 2002 order by year+2, profit; -- 10. having (year+2) shouldn't be pushed down",
+        "Plan": [
+          "Projection 6400.00 root  Column#6, Column#9",
+          "└─Sort 6400.00 root  Column#6, Column#9",
+          "  └─TableReader 6400.00 root  MppVersion: 2, data:ExchangeSender",
+          "    └─ExchangeSender 6400.00 mpp[tiflash]  ExchangeType: PassThrough",
+          "      └─Projection 6400.00 mpp[tiflash]  Column#9, Column#6",
+          "        └─HashAgg 6400.00 mpp[tiflash]  group by:Column#20, Column#21, Column#22, funcs:sum(Column#18)->Column#9, funcs:firstrow(Column#19)->Column#6",
+          "          └─Projection 8000.00 mpp[tiflash]  cast(test.sales.profit, decimal(10,0) BINARY)->Column#18, Column#6->Column#19, Column#6->Column#20, Column#7->Column#21, gid->Column#22",
+          "            └─ExchangeReceiver 8000.00 mpp[tiflash]  ",
+          "              └─ExchangeSender 8000.00 mpp[tiflash]  ExchangeType: HashPartition, Compression: FAST, Hash Cols: [name: Column#6, collate: binary], [name: Column#7, collate: binary], [name: gid, collate: binary]",
+          "                └─Selection 8000.00 mpp[tiflash]  gt(Column#6, 2002)",
+          "                  └─Expand 10000.00 mpp[tiflash]  level-projection:[test.sales.profit, <nil>->Column#6, <nil>->Column#7, 0->gid],[test.sales.profit, Column#6, <nil>->Column#7, 1->gid],[test.sales.profit, Column#6, Column#7, 3->gid]; schema: [test.sales.profit,Column#6,Column#7,gid]",
+          "                    └─Projection 10000.00 mpp[tiflash]  test.sales.profit, plus(test.sales.year, 2)->Column#6, plus(test.sales.year, test.sales.profit)->Column#7",
+          "                      └─TableFullScan 10000.00 mpp[tiflash] table:sales keep order:false, stats:pseudo"
+        ],
+        "Warn": null
       }
     ]
   }

planner/core/logical_plan_builder.go

@@ -1325,6 +1325,13 @@ func (b *PlanBuilder) buildSelection(ctx context.Context, p LogicalPlan, where a
 		if err != nil {
 			return nil, err
 		}
+		// for case: explain SELECT year+2 as y, SUM(profit) AS profit FROM sales GROUP BY year+2, year+profit WITH ROLLUP having y > 2002;
+		// currently, we succeed to resolve y to (year+2), but fail to resolve (year+2) to grouping col, and to base column function: plus(year, 2) instead.
+		// which will cause this selection being pushed down through Expand OP itself.
+		//
+		// In expand, we will additionally project (year+2) out as a new column, let's say grouping_col here, and we wanna it can substitute any upper layer's (year+2)
+		expr = b.replaceGroupingFunc(expr)
+
 		p = np
 		if expr == nil {
 			continue
@@ -2805,11 +2812,11 @@ func (b *PlanBuilder) resolveHavingAndOrderBy(ctx context.Context, sel *ast.Sele
 	}
 	havingAggMapper := extractor.aggMapper
 	extractor.aggMapper = make(map[*ast.AggregateFuncExpr]int)
-	extractor.inExpr = false
 	// Extract agg funcs from order by clause.
 	if sel.OrderBy != nil {
 		extractor.curClause = orderByClause
 		for _, item := range sel.OrderBy.Items {
+			extractor.inExpr = false
 			if ast.HasWindowFlag(item.Expr) {
 				continue
 			}

planner/core/rule_predicate_push_down.go

@@ -453,6 +453,18 @@ func isNullRejected(ctx sessionctx.Context, schema *expression.Schema, expr expr
 	return false
 }
 
+// PredicatePushDown implements LogicalPlan PredicatePushDown interface.
+func (p *LogicalExpand) PredicatePushDown(predicates []expression.Expression, opt *logicalOptimizeOp) (ret []expression.Expression, retPlan LogicalPlan) {
+	// Note that, grouping column related predicates can't be pushed down, since grouping column has nullability change after Expand OP itself.
+	// condition related with grouping column shouldn't be pushed down through it.
+	// currently, since expand is adjacent to aggregate, any filter above aggregate wanted to be push down through expand only have two cases:
+	// 		1. agg function related filters. (these condition is always above aggregate)
+	// 		2. group-by item related filters. (there condition is always related with grouping sets columns, which can't be pushed down)
+	// As a whole, we banned all the predicates pushing-down logic here that remained in Expand OP, and constructing a new selection above it if any.
+	remained, child := p.baseLogicalPlan.PredicatePushDown(nil, opt)
+	return append(remained, predicates...), child
+}
+
 // PredicatePushDown implements LogicalPlan PredicatePushDown interface.
 func (p *LogicalProjection) PredicatePushDown(predicates []expression.Expression, opt *logicalOptimizeOp) (ret []expression.Expression, retPlan LogicalPlan) {
 	canBePushed := make([]expression.Expression, 0, len(predicates))