planner: stop pushing Agg down through Projection if substitution fail (#50932)

expression/util.go

@@ -409,6 +409,7 @@ func SetExprColumnInOperand(expr Expression) Expression {
 
 // ColumnSubstitute substitutes the columns in filter to expressions in select fields.
 // e.g. select * from (select b as a from t) k where a < 10 => select * from (select b as a from t where b < 10) k.
+// TODO: remove this function and only use ColumnSubstituteImpl since this function swallows the error, which seems unsafe.
 func ColumnSubstitute(expr Expression, schema *Schema, newExprs []Expression) Expression {
 	_, _, resExpr := ColumnSubstituteImpl(expr, schema, newExprs, false)
 	return resExpr

planner/core/casetest/integration_test.go

@@ -3280,6 +3280,15 @@ func TestTiFlashPartitionTableScan(t *testing.T) {
 	tk.MustExec("drop table hp_t;")
 }
 
+func TestIssue50926(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t (a int)")
+	tk.MustExec("create or replace definer='root'@'localhost' view v (a,b) AS select 1 as a, json_object('k', '0') as b from t")
+	tk.MustQuery("select sum(json_extract(b, '$.path')) from v group by a").Check(testkit.Rows()) // no error
+}
+
 func TestTiFlashFineGrainedShuffle(t *testing.T) {
 	store, dom := testkit.CreateMockStoreAndDomain(t)
 	tk := testkit.NewTestKit(t, store)

planner/core/rule_aggregation_push_down.go

@@ -552,7 +552,12 @@
 				noSideEffects := true
 				newGbyItems := make([]expression.Expression, 0, len(agg.GroupByItems))
 				for _, gbyItem := range agg.GroupByItems {
-					newGbyItems = append(newGbyItems, expression.ColumnSubstitute(gbyItem, proj.schema, proj.Exprs))
+					_, failed, groupBy := expression.ColumnSubstituteImpl(gbyItem, proj.schema, proj.Exprs, true)
+					if failed {
+						noSideEffects = false
+						break
+					}
+					newGbyItems = append(newGbyItems, groupBy)
 					if ExprsHasSideEffects(newGbyItems) {
 						noSideEffects = false
 						break
@@ -567,7 +572,12 @@
 						oldAggFuncsArgs = append(oldAggFuncsArgs, aggFunc.Args)
 						newArgs := make([]expression.Expression, 0, len(aggFunc.Args))
 						for _, arg := range aggFunc.Args {
-							newArgs = append(newArgs, expression.ColumnSubstitute(arg, proj.schema, proj.Exprs))
+							_, failed, newArg := expression.ColumnSubstituteImpl(arg, proj.schema, proj.Exprs, true)
+							if failed {
+								noSideEffects = false
+								break
+							}
+							newArgs = append(newArgs, newArg)
 						}
 						if ExprsHasSideEffects(newArgs) {
 							noSideEffects = false
@@ -579,7 +589,12 @@
 							oldAggOrderItems = append(oldAggOrderItems, aggFunc.OrderByItems)
 							newOrderByItems := make([]expression.Expression, 0, len(aggFunc.OrderByItems))
 							for _, oby := range aggFunc.OrderByItems {
-								newOrderByItems = append(newOrderByItems, expression.ColumnSubstitute(oby.Expr, proj.schema, proj.Exprs))
+								_, failed, byItem := expression.ColumnSubstituteImpl(oby.Expr, proj.schema, proj.Exprs, true)
+								if failed {
+									noSideEffects = false
+									break
+								}
+								newOrderByItems = append(newOrderByItems, byItem)
 							}
 							if ExprsHasSideEffects(newOrderByItems) {
 								noSideEffects = false
@@ -598,6 +613,9 @@
 					}
 				}
 				for i, funcsArgs := range oldAggFuncsArgs {
+					if !noSideEffects {
+						break
+					}
 					for j := range funcsArgs {
 						if oldAggFuncsArgs[i][j].GetType().EvalType() != newAggFuncsArgs[i][j].GetType().EvalType() {
 							noSideEffects = false
@@ -614,9 +632,6 @@
 							break
 						}
 					}
-					if !noSideEffects {
-						break
-					}
 				}
 				if noSideEffects {
 					agg.GroupByItems = newGbyItems