planner, executor: set new child after injecting Project operator

cmd/explaintest/r/topn_pushdown.result

@@ -0,0 +1,9 @@
+explain select * from ((select 4 as a) union all (select 33 as a)) tmp order by a desc limit 1;
+id	count	task	operator info
+TopN_17	1.00	root	tmp.a:desc, offset:0, count:1
+└─Union_21	2.00	root	
+  ├─Projection_22	1.00	root	cast(a)
+  │ └─Projection_23	1.00	root	4
+  │   └─TableDual_24	1.00	root	rows:1
+  └─Projection_26	1.00	root	33
+    └─TableDual_27	1.00	root	rows:1

cmd/explaintest/t/topn_pushdown.test

@@ -0,0 +1 @@
+explain select * from ((select 4 as a) union all (select 33 as a)) tmp order by a desc limit 1;

executor/join_test.go

@@ -1368,3 +1368,16 @@ func (s *testSuite2) TestScalarFuncNullSemiJoin(c *C) {
 	tk.MustExec("insert into s values(null, 1)")
 	tk.MustQuery("select a in (select a+b from s) from t").Check(testkit.Rows("<nil>", "<nil>"))
 }
+
+func (s *testSuite2) TestInjectProjOnTopN(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t1")
+	tk.MustExec("drop table if exists t2")
+	tk.MustExec("create table t1(a bigint, b bigint)")
+	tk.MustExec("create table t2(a bigint, b bigint)")
+	tk.MustExec("insert into t1 values(1, 1)")
+	tk.MustQuery("select t1.a+t1.b as result from t1 left join t2 on 1 = 0 order by result limit 20;").Check(testkit.Rows(
+		"2",
+	))
+}

planner/core/physical_plan_test.go

@@ -115,7 +115,7 @@ func (s *testPlanSuite) TestDAGPlanBuilderSimpleCase(c *C) {
 		// Test TopN push down in table single read.
 		{
 			sql:  "select c from t order by t.a + t.b limit 1",
-			best: "TableReader(Table(t)->TopN([plus(test.t.a, test.t.b)],0,1))->Projection->TopN([col_3],0,1)->Projection",
+			best: "TableReader(Table(t)->TopN([plus(test.t.a, test.t.b)],0,1))->Projection->TopN([col_3],0,1)->Projection->Projection",
 		},
 		// Test Limit push down in table single read.
 		{
@@ -1219,7 +1219,7 @@ func (s *testPlanSuite) TestAggEliminater(c *C) {
 		// If max/min contains scalar function, we can still do transformation.
 		{
 			sql:  "select max(a+1) from t;",
-			best: "TableReader(Table(t)->Sel([not(isnull(plus(test.t.a, 1)))])->TopN([plus(test.t.a, 1) true],0,1))->Projection->TopN([col_1 true],0,1)->Projection->StreamAgg",
+			best: "TableReader(Table(t)->Sel([not(isnull(plus(test.t.a, 1)))])->TopN([plus(test.t.a, 1) true],0,1))->Projection->TopN([col_1 true],0,1)->Projection->Projection->StreamAgg",
 		},
 		// Do nothing to max+min.
 		{

planner/core/rule_inject_extra_projection.go

@@ -39,8 +39,8 @@ func NewProjInjector() *projInjector {
 }
 
 func (pe *projInjector) inject(plan PhysicalPlan) PhysicalPlan {
-	for _, child := range plan.Children() {
-		pe.inject(child)
+	for i, child := range plan.Children() {
+		plan.Children()[i] = pe.inject(child)
 	}
 
 	switch p := plan.(type) {
@@ -205,5 +205,9 @@ func injectProjBelowSort(p PhysicalPlan, orderByItems []*ByItems) PhysicalPlan {
 	if origChildProj, isChildProj := childPlan.(*PhysicalProjection); isChildProj {
 		refine4NeighbourProj(bottomProj, origChildProj)
 	}
+
+	// if topProj.Schema().Len() == 0 {
+	// 	return p
+	// }
 	return topProj
 }

planner/core/rule_topn_push_down.go

@@ -14,6 +14,7 @@
 package core
 
 import (
+	"github.com/cznic/mathutil"
 	"github.com/pingcap/tidb/expression"
 )
 
@@ -38,6 +39,18 @@ func (s *baseLogicalPlan) pushDownTopN(topN *LogicalTopN) LogicalPlan {
 
 // setChild set p as topn's child.
 func (lt *LogicalTopN) setChild(p LogicalPlan) LogicalPlan {
+	// Remove this TopN if its child is a TableDual.
+	dual, isDual := p.(*LogicalTableDual)
+	if isDual {
+		numDualRows := uint64(dual.RowCount)
+		if numDualRows < lt.Offset {
+			dual.RowCount = 0
+			return dual
+		}
+		dual.RowCount = int(mathutil.MinUint64(numDualRows-lt.Offset, lt.Count))
+		return dual
+	}
+
 	if lt.isLimit() {
 		limit := LogicalLimit{
 			Count:  lt.Count,