planner: don't decorrelate the APPLY when the inner's projection refe…

…rence no column (#32370)

close #32089

expression/util.go

@@ -190,7 +190,7 @@ func ExtractColumnsAndCorColumnsFromExpressions(result []*Column, list []Express
 }
 
 // ExtractColumnSet extracts the different values of `UniqueId` for columns in expressions.
-func ExtractColumnSet(exprs []Expression) *intsets.Sparse {
+func ExtractColumnSet(exprs ...Expression) *intsets.Sparse {
 	set := &intsets.Sparse{}
 	for _, expr := range exprs {
 		extractColumnSet(expr, set)

planner/core/rule_decorrelate.go

@@ -154,6 +154,21 @@ func (s *decorrelateSolver) optimize(ctx context.Context, p LogicalPlan, opt *lo
 				return s.optimize(ctx, p, opt)
 			}
 		} else if proj, ok := innerPlan.(*LogicalProjection); ok {
+			allConst := true
+			for _, expr := range proj.Exprs {
+				if len(expression.ExtractCorColumns(expr)) > 0 || !expression.ExtractColumnSet(expr).IsEmpty() {
+					allConst = false
+					break
+				}
+			}
+			if allConst && apply.JoinType == LeftOuterJoin {
+				// If the projection just references some constant. We cannot directly pull it up when the APPLY is an outer join.
+				//  e.g. select (select 1 from t1 where t1.a=t2.a) from t2; When the t1.a=t2.a is false the join's output is NULL.
+				//       But if we pull the projection upon the APPLY. It will return 1 since the projection is evaluated after the join.
+				// We disable the decorrelation directly for now.
+				// TODO: Actually, it can be optimized. We need to first push the projection down to the selection. And then the APPLY can be decorrelated.
+				goto NoOptimize
+			}
 			for i, expr := range proj.Exprs {
 				proj.Exprs[i] = expr.Decorrelate(outerPlan.Schema())
 			}
@@ -315,6 +330,7 @@ func (s *decorrelateSolver) optimize(ctx context.Context, p LogicalPlan, opt *lo
 			return s.optimize(ctx, p, opt)
 		}
 	}
+NoOptimize:
 	newChildren := make([]LogicalPlan, 0, len(p.Children()))
 	for _, child := range p.Children() {
 		np, err := s.optimize(ctx, child, opt)

planner/core/rule_partition_processor.go

@@ -930,7 +930,7 @@ func makePartitionByFnCol(sctx sessionctx.Context, columns []*expression.Column,
 		monotonous = getMonotoneMode(raw.FuncName.L)
 		// Check the partitionExpr is in the form: fn(col, ...)
 		// There should be only one column argument, and it should be the first parameter.
-		if expression.ExtractColumnSet(args).Len() == 1 {
+		if expression.ExtractColumnSet(args...).Len() == 1 {
 			if col1, ok := args[0].(*expression.Column); ok {
 				col = col1
 			}

planner/core/testdata/plan_suite_unexported_in.json

@@ -131,7 +131,8 @@
       "select t1.b from t t1 where t1.b in (select t2.b from t t2 where t2.a = t1.a order by t2.a)",
       "select t1.b from t t1 where exists(select t2.b from t t2 where t2.a = t1.a order by t2.a)",
       // `Sort` will not be eliminated, if it is not the top level operator.
-      "select t1.b from t t1 where t1.b = (select t2.b from t t2 where t2.a = t1.a order by t2.a limit 1)"
+      "select t1.b from t t1 where t1.b = (select t2.b from t t2 where t2.a = t1.a order by t2.a limit 1)",
+      "select (select 1 from t t1 where t1.a = t2.a) from t t2"
     ]
   },
   {

planner/core/testdata/plan_suite_unexported_out.json

@@ -117,7 +117,8 @@
       "Join{DataScan(t1)->DataScan(t2)->Aggr(max(test.t.a),firstrow(test.t.b))}(test.t.b,test.t.b)->Projection->Sel([eq(test.t.b, Column#25)])->Projection",
       "Join{DataScan(t1)->DataScan(t2)}(test.t.a,test.t.a)(test.t.b,test.t.b)->Projection",
       "Join{DataScan(t1)->DataScan(t2)}(test.t.a,test.t.a)->Projection",
-      "Apply{DataScan(t1)->DataScan(t2)->Sel([eq(test.t.a, test.t.a)])->Projection->Sort->Limit}->Projection->Sel([eq(test.t.b, test.t.b)])->Projection"
+      "Apply{DataScan(t1)->DataScan(t2)->Sel([eq(test.t.a, test.t.a)])->Projection->Sort->Limit}->Projection->Sel([eq(test.t.b, test.t.b)])->Projection",
+      "Apply{DataScan(t2)->DataScan(t1)->Sel([eq(test.t.a, test.t.a)])->Projection}->Projection"
     ]
   },
   {

util/ranger/detacher.go

@@ -198,7 +198,7 @@ func extractIndexPointRangesForCNF(sctx sessionctx.Context, conds []expression.E
 	offset := int(-1)
 	for i, cond := range conds {
 		tmpConds := []expression.Expression{cond}
-		colSets := expression.ExtractColumnSet(tmpConds)
+		colSets := expression.ExtractColumnSet(cond)
 		if colSets.Len() == 0 {
 			continue
 		}