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

#50932) (#50991)

close #50926

pkg/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

pkg/planner/core/casetest/BUILD.bazel

@@ -12,7 +12,7 @@ go_test(
     ],
     data = glob(["testdata/**"]),
     flaky = True,
-    shard_count = 20,
+    shard_count = 21,
     deps = [
         "//pkg/domain",
         "//pkg/parser",

pkg/planner/core/casetest/integration_test.go

@@ -404,6 +404,15 @@ func TestTiFlashFineGrainedShuffle(t *testing.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 TestFixControl45132(t *testing.T) {
 	store := testkit.CreateMockStore(t)
 	tk := testkit.NewTestKit(t, store)

pkg/planner/core/rule_aggregation_push_down.go

@@ -554,7 +554,12 @@ func (a *aggregationPushDownSolver) aggPushDown(p LogicalPlan, opt *logicalOptim
 				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
@@ -569,7 +574,12 @@ func (a *aggregationPushDownSolver) aggPushDown(p LogicalPlan, opt *logicalOptim
 						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
@@ -581,7 +591,12 @@ func (a *aggregationPushDownSolver) aggPushDown(p LogicalPlan, opt *logicalOptim
 							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
@@ -600,6 +615,9 @@ func (a *aggregationPushDownSolver) aggPushDown(p LogicalPlan, opt *logicalOptim
 					}
 				}
 				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
@@ -616,9 +634,6 @@ func (a *aggregationPushDownSolver) aggPushDown(p LogicalPlan, opt *logicalOptim
 							break
 						}
 					}
-					if !noSideEffects {
-						break
-					}
 				}
 				if noSideEffects {
 					agg.GroupByItems = newGbyItems