plan: fix a but about limit pushdown (#4449) (#4702)

plan/dag_plan_test.go

@@ -440,6 +440,22 @@ func (s *testPlanSuite) TestDAGPlanTopN(c *C) {
 			sql:  "select * from t t1 left join t t2 on t1.b = t2.b left join t t3 on t2.b = t3.b limit 1",
 			best: "LeftHashJoin{LeftHashJoin{TableReader(Table(t)->Limit)->TableReader(Table(t))}(t1.b,t2.b)->TableReader(Table(t))}(t2.b,t3.b)->Limit",
 		},
+		{
+			sql:  "select * from t where b = 1 and c = 1 order by c limit 1",
+			best: "IndexLookUp(Index(t.c_d_e)[[1,1]], Table(t)->Sel([eq(test.t.b, 1)]))->Limit",
+		},
+		{
+			sql:  "select * from t where c = 1 order by c limit 1",
+			best: "IndexLookUp(Index(t.c_d_e)[[1,1]]->Limit, Table(t))->Limit",
+		},
+		{
+			sql:  "select * from t order by a limit 1",
+			best: "TableReader(Table(t)->Limit)->Limit",
+		},
+		{
+			sql:  "select c from t order by c limit 1",
+			best: "IndexReader(Index(t.c_d_e)[[<nil>,+inf]]->Limit)->Limit",
+		},
 	}
 	for _, tt := range tests {
 		comment := Commentf("for %s", tt.sql)

plan/new_physical_plan_builder.go

@@ -802,6 +802,7 @@ func (p *DataSource) convertToIndexScan(prop *requiredProp, idx *model.IndexInfo
 			is.Desc = true
 			cop.cst = rowCount * descScanFactor
 		}
+		cop.keepOrder = true
 		is.addPushedDownSelection(cop, p, prop.expectedCnt)
 		if p.unionScanSchema != nil {
 			task = addUnionScan(cop, p)
@@ -960,6 +961,7 @@ func (p *DataSource) convertToTableScan(prop *requiredProp) (task task, err erro
 			copTask.cst = rowCount * descScanFactor
 		}
 		ts.KeepOrder = true
+		copTask.keepOrder = true
 		ts.addPushedDownSelection(copTask, p.profile, prop.expectedCnt)
 		if p.unionScanSchema != nil {
 			task = addUnionScan(copTask, p)

plan/task.go

@@ -44,6 +44,8 @@ type copTask struct {
 	cst       float64
 	// indexPlanFinished means we have finished index plan.
 	indexPlanFinished bool
+	// keepOrder indicates if the plan scans data by order.
+	keepOrder bool
 }
 
 func (t *copTask) invalid() bool {
@@ -276,16 +278,18 @@ func (p *Limit) attach2Task(tasks ...task) task {
 	}
 	t := tasks[0].copy()
 	if cop, ok := t.(*copTask); ok {
-		// If the task is copTask, the Limit can always be pushed down.
-		// When limit be pushed down, it should remove its offset.
-		pushedDownLimit := Limit{Count: p.Offset + p.Count}.init(p.allocator, p.ctx)
-		pushedDownLimit.profile = p.profile
-		if cop.tablePlan != nil {
-			pushedDownLimit.SetSchema(cop.tablePlan.Schema())
-		} else {
-			pushedDownLimit.SetSchema(cop.indexPlan.Schema())
+		// If the table/index scans data by order and applies a double read, the limit cannot be pushed to the table side.
+		if !cop.keepOrder || !cop.indexPlanFinished || cop.indexPlan == nil {
+			// When limit be pushed down, it should remove its offset.
+			pushedDownLimit := Limit{Count: p.Offset + p.Count}.init(p.allocator, p.ctx)
+			pushedDownLimit.profile = p.profile
+			if cop.tablePlan != nil {
+				pushedDownLimit.SetSchema(cop.tablePlan.Schema())
+			} else {
+				pushedDownLimit.SetSchema(cop.indexPlan.Schema())
+			}
+			cop = attachPlan2Task(pushedDownLimit, cop).(*copTask)
 		}
-		cop = attachPlan2Task(pushedDownLimit, cop).(*copTask)
 		t = finishCopTask(cop, p.ctx, p.allocator)
 	}
 	if !p.partial {