planner: do not eliminate group_concat in aggregate elimination (#9967)

cmd/explaintest/r/explain.result

@@ -22,3 +22,18 @@ c1	timestamp	YES		NULL
 desc t id;
 Field	Type	Null	Key	Default	Extra
 id	int(11)	YES		NULL	
+drop table if exists t;
+create table t(id int primary key, a int, b int);
+explain select group_concat(a) from t group by id;
+id	count	task	operator info
+StreamAgg_8	8000.00	root	group by:col_1, funcs:group_concat(col_0, ",")
+└─Projection_18	10000.00	root	cast(test.t.a), test.t.id
+  └─TableReader_15	10000.00	root	data:TableScan_14
+    └─TableScan_14	10000.00	cop	table:t, range:[-inf,+inf], keep order:true, stats:pseudo
+explain select group_concat(a, b) from t group by id;
+id	count	task	operator info
+StreamAgg_8	8000.00	root	group by:col_2, funcs:group_concat(col_0, col_1, ",")
+└─Projection_18	10000.00	root	cast(test.t.a), cast(test.t.b), test.t.id
+  └─TableReader_15	10000.00	root	data:TableScan_14
+    └─TableScan_14	10000.00	cop	table:t, range:[-inf,+inf], keep order:true, stats:pseudo
+drop table t;

cmd/explaintest/t/explain.test

@@ -6,3 +6,9 @@ describe t;
 desc t;
 desc t c1;
 desc t id;
+
+drop table if exists t;
+create table t(id int primary key, a int, b int);
+explain select group_concat(a) from t group by id;
+explain select group_concat(a, b) from t group by id;
+drop table t;

executor/aggregate_test.go

@@ -582,6 +582,7 @@ func (s *testSuite) TestAggEliminator(c *C) {
 	tk.MustQuery("select min(b) from t").Check(testkit.Rows("-2"))
 	tk.MustQuery("select max(b*b) from t").Check(testkit.Rows("4"))
 	tk.MustQuery("select min(b*b) from t").Check(testkit.Rows("1"))
+	tk.MustQuery("select group_concat(b, b) from t group by a").Check(testkit.Rows("-1-1", "-2-2", "11", "<nil>"))
 }
 
 func (s *testSuite) TestMaxMinFloatScalaFunc(c *C) {

planner/core/rule_aggregation_push_down.go

@@ -391,6 +391,20 @@ func (a *aggregationOptimizer) aggPushDown(p LogicalPlan) LogicalPlan {
 // For count(expr), sum(expr), avg(expr), count(distinct expr, [expr...]) we may need to rewrite the expr. Details are shown below.
 // If we can eliminate agg successful, we return a projection. Else we return a nil pointer.
 func (a *aggregationOptimizer) tryToEliminateAggregation(agg *LogicalAggregation) *LogicalProjection {
+	for _, af := range agg.AggFuncs {
+		// TODO(issue #9968): Actually, we can rewrite GROUP_CONCAT when all the
+		// arguments it accepts are promised to be NOT-NULL.
+		// When it accepts only 1 argument, we can extract this argument into a
+		// projection.
+		// When it accepts multiple arguments, we can wrap the arguments with a
+		// function CONCAT_WS and extract this function into a projection.
+		// BUT, GROUP_CONCAT should truncate the final result according to the
+		// system variable `group_concat_max_len`. To ensure the correctness of
+		// the result, we close the elimination of GROUP_CONCAT here.
+		if af.Name == ast.AggFuncGroupConcat {
+			return nil
+		}
+	}
 	schemaByGroupby := expression.NewSchema(agg.groupByCols...)
 	coveredByUniqueKey := false
 	for _, key := range agg.children[0].Schema().Keys {