diff --git a/executor/executor_test.go b/executor/executor_test.go
index eae41e98913fb..791f1eaa94190 100644
--- a/executor/executor_test.go
+++ b/executor/executor_test.go
@@ -1122,6 +1122,100 @@ func (s *testSuite) TestJoin(c *C) {
 
 }
 
+func (s *testSuite) TestMultiJoin(c *C) {
+	defer testleak.AfterTest(c)()
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t35(a35 int primary key, b35 int, x35 int)")
+	tk.MustExec("create table t40(a40 int primary key, b40 int, x40 int)")
+	tk.MustExec("create table t14(a14 int primary key, b14 int, x14 int)")
+	tk.MustExec("create table t42(a42 int primary key, b42 int, x42 int)")
+	tk.MustExec("create table t15(a15 int primary key, b15 int, x15 int)")
+	tk.MustExec("create table t7(a7 int primary key, b7 int, x7 int)")
+	tk.MustExec("create table t64(a64 int primary key, b64 int, x64 int)")
+	tk.MustExec("create table t19(a19 int primary key, b19 int, x19 int)")
+	tk.MustExec("create table t9(a9 int primary key, b9 int, x9 int)")
+	tk.MustExec("create table t8(a8 int primary key, b8 int, x8 int)")
+	tk.MustExec("create table t57(a57 int primary key, b57 int, x57 int)")
+	tk.MustExec("create table t37(a37 int primary key, b37 int, x37 int)")
+	tk.MustExec("create table t44(a44 int primary key, b44 int, x44 int)")
+	tk.MustExec("create table t38(a38 int primary key, b38 int, x38 int)")
+	tk.MustExec("create table t18(a18 int primary key, b18 int, x18 int)")
+	tk.MustExec("create table t62(a62 int primary key, b62 int, x62 int)")
+	tk.MustExec("create table t4(a4 int primary key, b4 int, x4 int)")
+	tk.MustExec("create table t48(a48 int primary key, b48 int, x48 int)")
+	tk.MustExec("create table t31(a31 int primary key, b31 int, x31 int)")
+	tk.MustExec("create table t16(a16 int primary key, b16 int, x16 int)")
+	tk.MustExec("create table t12(a12 int primary key, b12 int, x12 int)")
+	tk.MustExec("insert into t35 values(1,1,1)")
+	tk.MustExec("insert into t40 values(1,1,1)")
+	tk.MustExec("insert into t14 values(1,1,1)")
+	tk.MustExec("insert into t42 values(1,1,1)")
+	tk.MustExec("insert into t15 values(1,1,1)")
+	tk.MustExec("insert into t7 values(1,1,1)")
+	tk.MustExec("insert into t64 values(1,1,1)")
+	tk.MustExec("insert into t19 values(1,1,1)")
+	tk.MustExec("insert into t9 values(1,1,1)")
+	tk.MustExec("insert into t8 values(1,1,1)")
+	tk.MustExec("insert into t57 values(1,1,1)")
+	tk.MustExec("insert into t37 values(1,1,1)")
+	tk.MustExec("insert into t44 values(1,1,1)")
+	tk.MustExec("insert into t38 values(1,1,1)")
+	tk.MustExec("insert into t18 values(1,1,1)")
+	tk.MustExec("insert into t62 values(1,1,1)")
+	tk.MustExec("insert into t4 values(1,1,1)")
+	tk.MustExec("insert into t48 values(1,1,1)")
+	tk.MustExec("insert into t31 values(1,1,1)")
+	tk.MustExec("insert into t16 values(1,1,1)")
+	tk.MustExec("insert into t12 values(1,1,1)")
+	tk.MustExec("insert into t35 values(7,7,7)")
+	tk.MustExec("insert into t40 values(7,7,7)")
+	tk.MustExec("insert into t14 values(7,7,7)")
+	tk.MustExec("insert into t42 values(7,7,7)")
+	tk.MustExec("insert into t15 values(7,7,7)")
+	tk.MustExec("insert into t7 values(7,7,7)")
+	tk.MustExec("insert into t64 values(7,7,7)")
+	tk.MustExec("insert into t19 values(7,7,7)")
+	tk.MustExec("insert into t9 values(7,7,7)")
+	tk.MustExec("insert into t8 values(7,7,7)")
+	tk.MustExec("insert into t57 values(7,7,7)")
+	tk.MustExec("insert into t37 values(7,7,7)")
+	tk.MustExec("insert into t44 values(7,7,7)")
+	tk.MustExec("insert into t38 values(7,7,7)")
+	tk.MustExec("insert into t18 values(7,7,7)")
+	tk.MustExec("insert into t62 values(7,7,7)")
+	tk.MustExec("insert into t4 values(7,7,7)")
+	tk.MustExec("insert into t48 values(7,7,7)")
+	tk.MustExec("insert into t31 values(7,7,7)")
+	tk.MustExec("insert into t16 values(7,7,7)")
+	tk.MustExec("insert into t12 values(7,7,7)")
+	result := tk.MustQuery(`SELECT x4,x8,x38,x44,x31,x9,x57,x48,x19,x40,x14,x12,x7,x64,x37,x18,x62,x35,x42,x15,x16 FROM
+t35,t40,t14,t42,t15,t7,t64,t19,t9,t8,t57,t37,t44,t38,t18,t62,t4,t48,t31,t16,t12
+WHERE b48=a57
+AND a4=b19
+AND a14=b16
+AND b37=a48
+AND a40=b42
+AND a31=7
+AND a15=b40
+AND a38=b8
+AND b15=a31
+AND b64=a18
+AND b12=a44
+AND b7=a8
+AND b35=a16
+AND a12=b14
+AND a64=b57
+AND b62=a7
+AND a35=b38
+AND b9=a19
+AND a62=b18
+AND b4=a37
+AND b44=a42`)
+	result.Check(testkit.Rows("7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7"))
+}
+
 func (s *testSuite) TestIndexScan(c *C) {
 	defer testleak.AfterTest(c)()
 	tk := testkit.NewTestKit(c, s.store)
diff --git a/expression/expression.go b/expression/expression.go
index 83e761bd44540..3264b81bac232 100644
--- a/expression/expression.go
+++ b/expression/expression.go
@@ -258,6 +258,7 @@ func ScalarFuncs2Exprs(funcs []*ScalarFunction) []Expression {
 // DeepCopy implements Expression interface.
 func (sf *ScalarFunction) DeepCopy() Expression {
 	newFunc := &ScalarFunction{FuncName: sf.FuncName, Function: sf.Function, RetType: sf.RetType}
+	newFunc.Args = make([]Expression, 0, len(sf.Args))
 	for _, arg := range sf.Args {
 		newFunc.Args = append(newFunc.Args, arg.DeepCopy())
 	}
diff --git a/plan/join_reorder.go b/plan/join_reorder.go
new file mode 100644
index 0000000000000..ba138ddbe3e5a
--- /dev/null
+++ b/plan/join_reorder.go
@@ -0,0 +1,206 @@
+// Copyright 2016 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package plan
+
+import (
+	"sort"
+
+	"github.com/ngaut/log"
+	"github.com/pingcap/tidb/ast"
+	"github.com/pingcap/tidb/expression"
+)
+
+// tryToGetJoinGroup tries to fetch a whole join group, which all joins is cartesian join.
+func tryToGetJoinGroup(j *Join) ([]LogicalPlan, bool) {
+	if j.reordered || !j.cartesianJoin {
+		return nil, false
+	}
+	lChild := j.GetChildByIndex(0).(LogicalPlan)
+	rChild := j.GetChildByIndex(1).(LogicalPlan)
+	if nj, ok := lChild.(*Join); ok {
+		plans, valid := tryToGetJoinGroup(nj)
+		return append(plans, rChild), valid
+	}
+	return []LogicalPlan{lChild, rChild}, true
+}
+
+func findColumnIndexByGroup(groups []LogicalPlan, col *expression.Column) int {
+	for i, plan := range groups {
+		idx := plan.GetSchema().GetIndex(col)
+		if idx != -1 {
+			return i
+		}
+	}
+	log.Errorf("Unknown columns %s, from id %s, position %d", col.ToString(), col.FromID, col.Position)
+	return -1
+}
+
+type joinReOrderSolver struct {
+	graph      []edgeList
+	group      []LogicalPlan
+	visited    []bool
+	resultJoin LogicalPlan
+	groupRank  []*rankInfo
+	allocator  *idAllocator
+}
+
+type edgeList []*rankInfo
+
+func (l edgeList) Len() int {
+	return len(l)
+}
+
+func (l edgeList) Less(i, j int) bool {
+	return l[i].rate < l[j].rate
+}
+
+func (l edgeList) Swap(i, j int) {
+	l[i], l[j] = l[j], l[i]
+}
+
+type rankInfo struct {
+	nodeID int
+	rate   float64
+}
+
+func (e *joinReOrderSolver) Less(i, j int) bool {
+	return e.groupRank[i].rate < e.groupRank[j].rate
+}
+
+func (e *joinReOrderSolver) Swap(i, j int) {
+	e.groupRank[i], e.groupRank[j] = e.groupRank[j], e.groupRank[i]
+}
+
+func (e *joinReOrderSolver) Len() int {
+	return len(e.groupRank)
+}
+
+// reorderJoin implements a simple join reorder algorithm. It will extract all the equal conditions and compose them to a graph.
+// Then walk through the graph and pick the nodes connected by some edges to compose a join tree.
+// We will pick the node with least result set as early as possible.
+func (e *joinReOrderSolver) reorderJoin(group []LogicalPlan, conds []expression.Expression) {
+	e.graph = make([]edgeList, len(group))
+	e.group = group
+	e.visited = make([]bool, len(group))
+	e.resultJoin = nil
+	e.groupRank = make([]*rankInfo, len(group))
+	for i := 0; i < len(e.groupRank); i++ {
+		e.groupRank[i] = &rankInfo{
+			nodeID: i,
+			rate:   1.0,
+		}
+	}
+	for _, cond := range conds {
+		if f, ok := cond.(*expression.ScalarFunction); ok {
+			if f.FuncName.L == ast.EQ {
+				lCol, lok := f.Args[0].(*expression.Column)
+				rCol, rok := f.Args[1].(*expression.Column)
+				if lok && rok && !lCol.Correlated && !rCol.Correlated {
+					lID := findColumnIndexByGroup(group, lCol)
+					rID := findColumnIndexByGroup(group, rCol)
+					if lID != rID {
+						e.graph[lID] = append(e.graph[lID], &rankInfo{nodeID: rID})
+						e.graph[rID] = append(e.graph[rID], &rankInfo{nodeID: lID})
+						continue
+					}
+				}
+			}
+			id := -1
+			rate := 1.0
+			cols, _ := extractColumn(f, nil, nil)
+			for _, col := range cols {
+				idx := findColumnIndexByGroup(group, col)
+				if id == -1 {
+					switch f.FuncName.L {
+					case ast.EQ:
+						rate *= 0.1
+					case ast.LT, ast.LE, ast.GE, ast.GT:
+						rate *= 0.3
+					// TODO: Estimate it more precisely in future.
+					default:
+						rate *= 0.9
+					}
+					id = idx
+				} else {
+					id = -1
+					break
+				}
+			}
+			if id != -1 {
+				e.groupRank[id].rate *= rate
+			}
+		}
+	}
+	for _, node := range e.graph {
+		for _, edge := range node {
+			edge.rate = e.groupRank[edge.nodeID].rate
+		}
+	}
+	sort.Sort(e)
+	for _, edge := range e.graph {
+		sort.Sort(edge)
+	}
+	var cartesianJoinGroup []LogicalPlan
+	for j := 0; j < len(e.groupRank); j++ {
+		i := e.groupRank[j].nodeID
+		if !e.visited[i] {
+			e.resultJoin = e.group[i]
+			e.walkGraphAndComposeJoin(i)
+			cartesianJoinGroup = append(cartesianJoinGroup, e.resultJoin)
+		}
+	}
+	e.makeBushyJoin(cartesianJoinGroup)
+}
+
+// Make cartesian join as bushy tree.
+func (e *joinReOrderSolver) makeBushyJoin(cartesianJoinGroup []LogicalPlan) {
+	for len(cartesianJoinGroup) > 1 {
+		resultJoinGroup := make([]LogicalPlan, 0, len(cartesianJoinGroup))
+		for i := 0; i < len(cartesianJoinGroup); i += 2 {
+			if i+1 == len(cartesianJoinGroup) {
+				resultJoinGroup = append(resultJoinGroup, cartesianJoinGroup[i])
+				break
+			}
+			resultJoinGroup = append(resultJoinGroup, e.newJoin(cartesianJoinGroup[i], cartesianJoinGroup[i+1]))
+		}
+		cartesianJoinGroup = resultJoinGroup
+	}
+	e.resultJoin = cartesianJoinGroup[0]
+}
+
+func (e *joinReOrderSolver) newJoin(lChild, rChild LogicalPlan) *Join {
+	join := &Join{
+		JoinType:        InnerJoin,
+		reordered:       true,
+		baseLogicalPlan: newBaseLogicalPlan(Jn, e.allocator),
+	}
+	join.initID()
+	join.SetChildren(lChild, rChild)
+	join.SetSchema(append(lChild.GetSchema().DeepCopy(), rChild.GetSchema().DeepCopy()...))
+	lChild.SetParents(join)
+	rChild.SetParents(join)
+	return join
+}
+
+// walkGraph implements a dfs algorithm. Each time it picks a edge with lowest rate, which has been sorted before.
+func (e *joinReOrderSolver) walkGraphAndComposeJoin(u int) {
+	e.visited[u] = true
+	for _, edge := range e.graph[u] {
+		v := edge.nodeID
+		if !e.visited[v] {
+			e.resultJoin = e.newJoin(e.resultJoin, e.group[v])
+			e.walkGraphAndComposeJoin(v)
+		}
+	}
+}
diff --git a/plan/logical_plan_builder.go b/plan/logical_plan_builder.go
index 59ea07d1af5bb..6959c1d2ea67f 100644
--- a/plan/logical_plan_builder.go
+++ b/plan/logical_plan_builder.go
@@ -209,6 +209,8 @@ func (b *planBuilder) buildNewJoin(join *ast.Join) LogicalPlan {
 		joinPlan.LeftConditions = leftCond
 		joinPlan.RightConditions = rightCond
 		joinPlan.OtherConditions = otherCond
+	} else if joinPlan.JoinType == InnerJoin {
+		joinPlan.cartesianJoin = true
 	}
 	if join.Tp == ast.LeftJoin {
 		joinPlan.JoinType = LeftOuterJoin
diff --git a/plan/logical_plans.go b/plan/logical_plans.go
index 7f202411fdec8..31170de5639ed 100644
--- a/plan/logical_plans.go
+++ b/plan/logical_plans.go
@@ -25,10 +25,8 @@ import (
 type JoinType int
 
 const (
-	// CrossJoin means Cartesian Product, but not used now.
-	CrossJoin JoinType = iota
 	// InnerJoin means inner join.
-	InnerJoin
+	InnerJoin JoinType = iota
 	// LeftOuterJoin means left join.
 	LeftOuterJoin
 	// RightOuterJoin means right join.
@@ -43,8 +41,10 @@ const (
 type Join struct {
 	baseLogicalPlan
 
-	JoinType JoinType
-	anti     bool
+	JoinType      JoinType
+	anti          bool
+	reordered     bool
+	cartesianJoin bool
 
 	EqualConditions []*expression.ScalarFunction
 	LeftConditions  []expression.Expression
diff --git a/plan/new_plan_test.go b/plan/new_plan_test.go
index 73eff36701d2e..7c7dac7557fa4 100644
--- a/plan/new_plan_test.go
+++ b/plan/new_plan_test.go
@@ -191,6 +191,64 @@ func (s *testPlanSuite) TestPredicatePushDown(c *C) {
 	UseNewPlanner = false
 }
 
+func (s *testPlanSuite) TestJoinReOrder(c *C) {
+	UseNewPlanner = true
+	defer testleak.AfterTest(c)()
+	cases := []struct {
+		sql  string
+		best string
+	}{
+		{
+			sql:  "select * from t t1, t t2, t t3, t t4, t t5, t t6 where t1.a = t2.b and t2.a = t3.b and t3.c = t4.a and t4.d = t2.c and t5.d = t6.d",
+			best: "LeftHashJoin{LeftHashJoin{LeftHashJoin{LeftHashJoin{Table(t)->Table(t)}(t1.a,t2.b)->Table(t)}(t2.a,t3.b)->Table(t)}(t3.c,t4.a)(t2.c,t4.d)->LeftHashJoin{Table(t)->Table(t)}(t5.d,t6.d)}->Projection",
+		},
+		{
+			sql:  "select * from t t1, t t2, t t3, t t4, t t5, t t6, t t7, t t8 where t1.a = t8.a",
+			best: "LeftHashJoin{LeftHashJoin{LeftHashJoin{LeftHashJoin{Table(t)->Table(t)}(t1.a,t8.a)->Table(t)}->LeftHashJoin{Table(t)->Table(t)}}->LeftHashJoin{LeftHashJoin{Table(t)->Table(t)}->Table(t)}}->Projection",
+		},
+		{
+			sql:  "select * from t t1, t t2, t t3, t t4, t t5 where t1.a = t5.a and t5.a = t4.a and t4.a = t3.a and t3.a = t2.a and t2.a = t1.a and t1.a = t3.a and t2.a = t4.a and t5.b < 8",
+			best: "LeftHashJoin{LeftHashJoin{LeftHashJoin{RightHashJoin{Table(t)->Selection->Table(t)}(t5.a,t1.a)->Table(t)}(t1.a,t2.a)->Table(t)}(t2.a,t3.a)(t1.a,t3.a)->Table(t)}(t5.a,t4.a)(t3.a,t4.a)(t2.a,t4.a)->Projection",
+		},
+		{
+			sql:  "select * from t t1, t t2, t t3, t t4, t t5 where t1.a = t5.a and t5.a = t4.a and t4.a = t3.a and t3.a = t2.a and t2.a = t1.a and t1.a = t3.a and t2.a = t4.a and t3.b = 1 and t4.a = 1",
+			best: "LeftHashJoin{LeftHashJoin{LeftHashJoin{LeftHashJoin{Table(t)->Selection->Table(t)}(t3.a,t4.a)->Table(t)}(t4.a,t5.a)->Table(t)}(t5.a,t1.a)(t3.a,t1.a)->Table(t)}(t3.a,t2.a)(t1.a,t2.a)(t4.a,t2.a)->Projection",
+		},
+		{
+			sql:  "select * from t o where o.b in (select t3.c from t t1, t t2, t t3 where t1.a = t3.a and t2.a = t3.a and t2.a = o.a)",
+			best: "Table(t)->Apply(LeftHashJoin{RightHashJoin{Table(t)->Selection->Table(t)}(t2.a,t3.a)->Table(t)}(t3.a,t1.a)->Projection)->Selection->Projection",
+		},
+	}
+	for _, ca := range cases {
+		comment := Commentf("for %s", ca.sql)
+		stmt, err := s.ParseOneStmt(ca.sql, "", "")
+		c.Assert(err, IsNil, comment)
+		ast.SetFlag(stmt)
+
+		err = newMockResolve(stmt)
+		c.Assert(err, IsNil)
+
+		builder := &planBuilder{
+			allocator: new(idAllocator),
+			ctx:       mock.NewContext(),
+			colMapper: make(map[*ast.ColumnNameExpr]int),
+		}
+		p := builder.build(stmt)
+		c.Assert(builder.err, IsNil)
+		lp := p.(LogicalPlan)
+
+		_, lp, err = lp.PredicatePushDown(nil)
+		c.Assert(err, IsNil)
+		_, err = lp.PruneColumnsAndResolveIndices(lp.GetSchema())
+		c.Assert(err, IsNil)
+		_, res, _, err := lp.convert2PhysicalPlan(nil)
+		c.Assert(err, IsNil)
+		p = res.p.PushLimit(nil)
+		c.Assert(ToString(p), Equals, ca.best, Commentf("for %s", ca.sql))
+	}
+	UseNewPlanner = false
+}
+
 func (s *testPlanSuite) TestCBO(c *C) {
 	UseNewPlanner = true
 	defer testleak.AfterTest(c)()
diff --git a/plan/physical_plan_builder.go b/plan/physical_plan_builder.go
index c5dc6b6b965fa..205c6b5dbe6fa 100644
--- a/plan/physical_plan_builder.go
+++ b/plan/physical_plan_builder.go
@@ -173,9 +173,12 @@ func (p *DataSource) handleIndexScan(prop requiredProperty, index *model.IndexIn
 
 // convert2PhysicalPlan implements LogicalPlan convert2PhysicalPlan interface.
 func (p *DataSource) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo, *physicalPlanInfo, uint64, error) {
+	sortedRes, unsortedRes, cnt := p.getPlanInfo(prop)
+	if sortedRes != nil {
+		return sortedRes, unsortedRes, cnt, nil
+	}
 	statsTbl := p.statisticTable
 	indices, includeTableScan := availableIndices(p.table)
-	var sortedRes, unsortedRes *physicalPlanInfo
 	var err error
 	if includeTableScan {
 		sortedRes, unsortedRes, err = p.handleTableScan(prop)
@@ -195,25 +198,34 @@ func (p *DataSource) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanI
 			unsortedRes = unsortedIsRes
 		}
 	}
+	p.storePlanInfo(prop, sortedRes, unsortedRes, uint64(statsTbl.Count))
 	return sortedRes, unsortedRes, uint64(statsTbl.Count), nil
 }
 
-func addPlanToResponse(p PhysicalPlan, res *physicalPlanInfo) *physicalPlanInfo {
+func addPlanToResponse(p PhysicalPlan, planInfo *physicalPlanInfo) *physicalPlanInfo {
 	np := p.Copy()
-	np.SetChildren(res.p)
-	return &physicalPlanInfo{p: np, cost: res.cost}
+	np.SetChildren(planInfo.p)
+	return &physicalPlanInfo{p: np, cost: planInfo.cost}
 }
 
 // convert2PhysicalPlan implements LogicalPlan convert2PhysicalPlan interface.
 func (p *Limit) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo, *physicalPlanInfo, uint64, error) {
-	res0, res1, count, err := p.GetChildByIndex(0).(LogicalPlan).convert2PhysicalPlan(prop)
+	sortedPlanInfo, unSortedPlanInfo, count := p.getPlanInfo(prop)
+	var err error
+	if sortedPlanInfo != nil {
+		return sortedPlanInfo, unSortedPlanInfo, count, nil
+	}
+	sortedPlanInfo, unSortedPlanInfo, count, err = p.GetChildByIndex(0).(LogicalPlan).convert2PhysicalPlan(prop)
 	if err != nil {
 		return nil, nil, 0, errors.Trace(err)
 	}
 	if p.Offset+p.Count < count {
 		count = p.Offset + p.Count
 	}
-	return addPlanToResponse(p, res0), addPlanToResponse(p, res1), count, nil
+	sortedPlanInfo = addPlanToResponse(p, sortedPlanInfo)
+	unSortedPlanInfo = addPlanToResponse(p, unSortedPlanInfo)
+	p.storePlanInfo(prop, sortedPlanInfo, unSortedPlanInfo, count)
+	return sortedPlanInfo, unSortedPlanInfo, count, nil
 }
 
 func estimateJoinCount(lc uint64, rc uint64) uint64 {
@@ -245,20 +257,20 @@ func (p *Join) handleLeftJoin(prop requiredProperty, innerJoin bool) (*physicalP
 	if !allLeft {
 		prop = nil
 	}
-	lRes0, lRes1, lCount, err := lChild.convert2PhysicalPlan(prop)
+	lSortedPlanInfo, lUnSortedPlanInfo, lCount, err := lChild.convert2PhysicalPlan(prop)
 	if err != nil {
 		return nil, nil, 0, errors.Trace(err)
 	}
 	if !allLeft {
-		lRes0.cost = math.MaxFloat64
+		lSortedPlanInfo.cost = math.MaxFloat64
 	}
-	rRes0, rRes1, rCount, err := rChild.convert2PhysicalPlan(nil)
+	rSortedPlanInfo, rUnSortedPlanInfo, rCount, err := rChild.convert2PhysicalPlan(nil)
 	if err != nil {
 		return nil, nil, 0, errors.Trace(err)
 	}
-	res0 := join.matchProperty(prop, []uint64{lCount, rCount}, lRes0, rRes0)
-	res1 := join.matchProperty(prop, []uint64{lCount, rCount}, lRes1, rRes1)
-	return res0, res1, estimateJoinCount(lCount, rCount), nil
+	sortedPlanInfo := join.matchProperty(prop, []uint64{lCount, rCount}, lSortedPlanInfo, rSortedPlanInfo)
+	unSortedPlanInfo := join.matchProperty(prop, []uint64{lCount, rCount}, lUnSortedPlanInfo, rUnSortedPlanInfo)
+	return sortedPlanInfo, unSortedPlanInfo, estimateJoinCount(lCount, rCount), nil
 }
 
 func (p *Join) handleRightJoin(prop requiredProperty, innerJoin bool) (*physicalPlanInfo, *physicalPlanInfo, uint64, error) {
@@ -282,27 +294,31 @@ func (p *Join) handleRightJoin(prop requiredProperty, innerJoin bool) (*physical
 	} else {
 		join.JoinType = RightOuterJoin
 	}
-	lRes0, lRes1, lCount, err := lChild.convert2PhysicalPlan(nil)
+	lSortedPlanInfo, lUnSortedPlanInfo, lCount, err := lChild.convert2PhysicalPlan(nil)
 	if err != nil {
 		return nil, nil, 0, errors.Trace(err)
 	}
 	if !allRight {
 		prop = nil
 	}
-	rRes0, rRes1, rCount, err := rChild.convert2PhysicalPlan(prop)
+	rSortedPlanInfo, rUnSortedPlanInfo, rCount, err := rChild.convert2PhysicalPlan(prop)
 	if err != nil {
 		return nil, nil, 0, errors.Trace(err)
 	}
 	if !allRight {
-		rRes0.cost = math.MaxFloat64
+		rSortedPlanInfo.cost = math.MaxFloat64
 	}
-	res0 := join.matchProperty(prop, []uint64{lCount, rCount}, lRes0, rRes0)
-	res1 := join.matchProperty(prop, []uint64{lCount, rCount}, lRes1, rRes1)
-	return res0, res1, estimateJoinCount(lCount, rCount), nil
+	sortedPlanInfo := join.matchProperty(prop, []uint64{lCount, rCount}, lSortedPlanInfo, rSortedPlanInfo)
+	unSortedPlanInfo := join.matchProperty(prop, []uint64{lCount, rCount}, lUnSortedPlanInfo, rUnSortedPlanInfo)
+	return sortedPlanInfo, unSortedPlanInfo, estimateJoinCount(lCount, rCount), nil
 }
 
 // convert2PhysicalPlan implements LogicalPlan convert2PhysicalPlan interface.
 func (p *Join) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo, *physicalPlanInfo, uint64, error) {
+	sortedPlanInfo, unsortedPlanInfo, cnt := p.getPlanInfo(prop)
+	if sortedPlanInfo != nil {
+		return sortedPlanInfo, unsortedPlanInfo, cnt, nil
+	}
 	switch p.JoinType {
 	case SemiJoin, SemiJoinWithAux:
 		lChild := p.GetChildByIndex(0).(LogicalPlan)
@@ -325,92 +341,133 @@ func (p *Join) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo, *
 		if !allLeft {
 			prop = nil
 		}
-		lRes0, lRes1, lCount, err := lChild.convert2PhysicalPlan(prop)
+		lSortedPlanInfo, lUnSortedPlanInfo, lCount, err := lChild.convert2PhysicalPlan(prop)
 		if err != nil {
 			return nil, nil, 0, errors.Trace(err)
 		}
-		rRes0, rRes1, rCount, err := rChild.convert2PhysicalPlan(nil)
+		rSortedPlanInfo, rUnSortedPlanInfo, rCount, err := rChild.convert2PhysicalPlan(nil)
 		if err != nil {
 			return nil, nil, 0, errors.Trace(err)
 		}
-		res0 := join.matchProperty(prop, []uint64{lCount, rCount}, lRes0, rRes0)
-		res1 := join.matchProperty(prop, []uint64{lCount, rCount}, lRes1, rRes1)
+		sortedPlanInfo := join.matchProperty(prop, []uint64{lCount, rCount}, lSortedPlanInfo, rSortedPlanInfo)
+		unSortedPlanInfo := join.matchProperty(prop, []uint64{lCount, rCount}, lUnSortedPlanInfo, rUnSortedPlanInfo)
 		if p.JoinType == SemiJoin {
 			lCount = uint64(float64(lCount) * selectionFactor)
 		}
 		if !allLeft {
-			res0.cost = math.MaxFloat64
+			sortedPlanInfo.cost = math.MaxFloat64
 		}
-		return res0, res1, lCount, err
+		p.storePlanInfo(prop, sortedPlanInfo, unSortedPlanInfo, lCount)
+		return sortedPlanInfo, unSortedPlanInfo, lCount, nil
 	case LeftOuterJoin:
-		return p.handleLeftJoin(prop, false)
+		sortedPlanInfo, unSortedPlanInfo, count, err := p.handleLeftJoin(prop, false)
+		if err != nil {
+			return nil, nil, 0, errors.Trace(err)
+		}
+		p.storePlanInfo(prop, sortedPlanInfo, unSortedPlanInfo, count)
+		return sortedPlanInfo, unSortedPlanInfo, count, nil
 	case RightOuterJoin:
-		return p.handleRightJoin(prop, false)
+		sortedPlanInfo, unSortedPlanInfo, count, err := p.handleRightJoin(prop, false)
+		if err != nil {
+			return nil, nil, 0, errors.Trace(err)
+		}
+		p.storePlanInfo(prop, sortedPlanInfo, unSortedPlanInfo, count)
+		return sortedPlanInfo, unSortedPlanInfo, count, nil
 	default:
-		lres0, lres1, count, err := p.handleLeftJoin(prop, true)
+		lSortedPlanInfo, lunSortedPlanInfo, count, err := p.handleLeftJoin(prop, true)
 		if err != nil {
 			return nil, nil, 0, errors.Trace(err)
 		}
-		rres0, rres1, _, err := p.handleRightJoin(prop, true)
+		rsortedPlanInfo, runSortedPlanInfo, _, err := p.handleRightJoin(prop, true)
 		if err != nil {
 			return nil, nil, 0, errors.Trace(err)
 		}
-		if rres0.cost < lres0.cost {
-			lres0 = rres0
+		if rsortedPlanInfo.cost < lSortedPlanInfo.cost {
+			lSortedPlanInfo = rsortedPlanInfo
 		}
-		if rres1.cost < lres1.cost {
-			lres1 = rres1
+		if runSortedPlanInfo.cost < lunSortedPlanInfo.cost {
+			lunSortedPlanInfo = runSortedPlanInfo
 		}
-		return lres0, lres1, count, errors.Trace(err)
+		p.storePlanInfo(prop, lSortedPlanInfo, lunSortedPlanInfo, count)
+		return lSortedPlanInfo, lunSortedPlanInfo, count, nil
 	}
 }
 
 // convert2PhysicalPlan implements LogicalPlan convert2PhysicalPlan interface.
 func (p *Aggregation) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo, *physicalPlanInfo, uint64, error) {
-	_, res, cnt, err := p.GetChildByIndex(0).(LogicalPlan).convert2PhysicalPlan(nil)
+	var err error
+	_, planInfo, cnt := p.getPlanInfo(prop)
+	if planInfo != nil {
+		return planInfo, planInfo, cnt, nil
+	}
+	_, planInfo, cnt, err = p.GetChildByIndex(0).(LogicalPlan).convert2PhysicalPlan(nil)
 	if len(prop) != 0 {
-		return &physicalPlanInfo{cost: math.MaxFloat64}, addPlanToResponse(p, res), cnt / 3, errors.Trace(err)
+		return &physicalPlanInfo{cost: math.MaxFloat64}, addPlanToResponse(p, planInfo), cnt / 3, errors.Trace(err)
 	}
-	res = addPlanToResponse(p, res)
-	return res, res, cnt / 3, errors.Trace(err)
+	planInfo = addPlanToResponse(p, planInfo)
+	p.storePlanInfo(prop, planInfo, planInfo, cnt/3)
+	return planInfo, planInfo, cnt / 3, errors.Trace(err)
 }
 
 // convert2PhysicalPlan implements LogicalPlan convert2PhysicalPlan interface.
 func (p *NewUnion) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo, *physicalPlanInfo, uint64, error) {
-	count := uint64(0)
-	var res0Collection, res1Collection []*physicalPlanInfo
+	var err error
+	sortedPlanInfo, unSortedPlanInfo, count := p.getPlanInfo(prop)
+	if sortedPlanInfo != nil {
+		return sortedPlanInfo, unSortedPlanInfo, count, nil
+	}
+	count = uint64(0)
+	var sortedPlanInfoCollection, unSortedPlanInfoCollection []*physicalPlanInfo
 	for _, child := range p.GetChildren() {
 		newProp := make(requiredProperty, 0, len(prop))
 		for _, c := range prop {
 			idx := p.GetSchema().GetIndex(c.col)
 			newProp = append(newProp, &columnProp{col: child.GetSchema()[idx], desc: c.desc})
 		}
-		res0, res1, cnt, err := child.(LogicalPlan).convert2PhysicalPlan(newProp)
+		var cnt uint64
+		sortedPlanInfo, unSortedPlanInfo, cnt, err = child.(LogicalPlan).convert2PhysicalPlan(newProp)
 		count += cnt
 		if err != nil {
 			return nil, nil, 0, errors.Trace(err)
 		}
-		res0Collection = append(res0Collection, res0)
-		res1Collection = append(res1Collection, res1)
+		sortedPlanInfoCollection = append(sortedPlanInfoCollection, sortedPlanInfo)
+		unSortedPlanInfoCollection = append(unSortedPlanInfoCollection, unSortedPlanInfo)
 	}
-	return p.matchProperty(prop, nil, res0Collection...), p.matchProperty(prop, nil, res1Collection...), count, nil
+	sortedPlanInfo = p.matchProperty(prop, nil, sortedPlanInfoCollection...)
+	unSortedPlanInfo = p.matchProperty(prop, nil, unSortedPlanInfoCollection...)
+	p.storePlanInfo(prop, sortedPlanInfo, unSortedPlanInfo, count)
+	return sortedPlanInfo, unSortedPlanInfo, count, nil
 }
 
 // convert2PhysicalPlan implements LogicalPlan convert2PhysicalPlan interface.
 func (p *Selection) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo, *physicalPlanInfo, uint64, error) {
-	res0, res1, count, err := p.GetChildByIndex(0).(LogicalPlan).convert2PhysicalPlan(prop)
+	var err error
+	sortedPlanInfo, unSortedPlanInfo, count := p.getPlanInfo(prop)
+	if sortedPlanInfo != nil {
+		return sortedPlanInfo, unSortedPlanInfo, count, nil
+	}
+	sortedPlanInfo, unSortedPlanInfo, count, err = p.GetChildByIndex(0).(LogicalPlan).convert2PhysicalPlan(prop)
 	if err != nil {
 		return nil, nil, 0, errors.Trace(err)
 	}
 	if _, ok := p.GetChildByIndex(0).(*DataSource); ok {
 		count = uint64(float64(count) * selectionFactor)
-		return res0, res1, count, nil
+		return sortedPlanInfo, unSortedPlanInfo, count, nil
 	}
-	return p.matchProperty(prop, nil, res0), p.matchProperty(prop, nil, res1), count / 3, nil
+	count /= 3
+	sortedPlanInfo = p.matchProperty(prop, nil, sortedPlanInfo)
+	unSortedPlanInfo = p.matchProperty(prop, nil, unSortedPlanInfo)
+	p.storePlanInfo(prop, sortedPlanInfo, unSortedPlanInfo, count)
+	return sortedPlanInfo, unSortedPlanInfo, count, nil
 }
 
 // convert2PhysicalPlan implements LogicalPlan convert2PhysicalPlan interface.
 func (p *Projection) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo, *physicalPlanInfo, uint64, error) {
+	var err error
+	sortedPlanInfo, unSortedPlanInfo, count := p.getPlanInfo(prop)
+	if sortedPlanInfo != nil {
+		return sortedPlanInfo, unSortedPlanInfo, count, nil
+	}
 	newProp := make(requiredProperty, 0, len(prop))
 	childSchema := p.GetChildByIndex(0).GetSchema()
 	usedCols := make([]bool, len(childSchema))
@@ -431,16 +488,18 @@ loop:
 			break loop
 		}
 	}
-	res0, res1, count, err := p.GetChildByIndex(0).(LogicalPlan).convert2PhysicalPlan(newProp)
+	sortedPlanInfo, unSortedPlanInfo, count, err = p.GetChildByIndex(0).(LogicalPlan).convert2PhysicalPlan(newProp)
 	if err != nil {
 		return nil, nil, count, errors.Trace(err)
 	}
-	res1 = addPlanToResponse(p, res1)
+	unSortedPlanInfo = addPlanToResponse(p, unSortedPlanInfo)
 	if !canPassSort {
-		return &physicalPlanInfo{cost: math.MaxFloat64}, res1, count, nil
+		return &physicalPlanInfo{cost: math.MaxFloat64}, unSortedPlanInfo, count, nil
 	}
 
-	return addPlanToResponse(p, res0), res1, count, nil
+	sortedPlanInfo = addPlanToResponse(p, sortedPlanInfo)
+	p.storePlanInfo(prop, sortedPlanInfo, unSortedPlanInfo, count)
+	return sortedPlanInfo, unSortedPlanInfo, count, nil
 }
 
 func matchProp(target, new requiredProperty) bool {
@@ -459,6 +518,11 @@ func matchProp(target, new requiredProperty) bool {
 
 // convert2PhysicalPlan implements LogicalPlan convert2PhysicalPlan interface.
 func (p *NewSort) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo, *physicalPlanInfo, uint64, error) {
+	var err error
+	sortedPlanInfo, unSortedPlanInfo, count := p.getPlanInfo(prop)
+	if sortedPlanInfo != nil {
+		return sortedPlanInfo, unSortedPlanInfo, count, nil
+	}
 	selfProp := make(requiredProperty, 0, len(p.ByItems))
 	for _, by := range p.ByItems {
 		if col, ok := by.Expr.(*expression.Column); ok {
@@ -468,26 +532,34 @@ func (p *NewSort) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo
 			break
 		}
 	}
-	res0, res1, count, err := p.GetChildByIndex(0).(LogicalPlan).convert2PhysicalPlan(selfProp)
+	sortedPlanInfo, unSortedPlanInfo, count, err = p.GetChildByIndex(0).(LogicalPlan).convert2PhysicalPlan(selfProp)
 	if err != nil {
 		return nil, nil, 0, errors.Trace(err)
 	}
 	cnt := float64(count)
 	sortCost := cnt*math.Log2(cnt)*cpuFactor + memoryFactor*cnt
 	if len(selfProp) == 0 {
-		res0 = addPlanToResponse(p, res1)
-	} else if sortCost+res1.cost < res0.cost {
-		res0.cost = sortCost + res1.cost
-		res0 = addPlanToResponse(p, res1)
+		sortedPlanInfo = addPlanToResponse(p, unSortedPlanInfo)
+	} else if sortCost+unSortedPlanInfo.cost < sortedPlanInfo.cost {
+		sortedPlanInfo.cost = sortCost + unSortedPlanInfo.cost
+		sortedPlanInfo = addPlanToResponse(p, unSortedPlanInfo)
 	}
 	if matchProp(prop, selfProp) {
-		return res0, res0, count, nil
+		return sortedPlanInfo, sortedPlanInfo, count, nil
 	}
-	return &physicalPlanInfo{cost: math.MaxFloat64}, res0, count, nil
+	unSortedPlanInfo = sortedPlanInfo
+	sortedPlanInfo = &physicalPlanInfo{cost: math.MaxFloat64}
+	p.storePlanInfo(prop, sortedPlanInfo, unSortedPlanInfo, count)
+	return sortedPlanInfo, unSortedPlanInfo, count, nil
 }
 
 // convert2PhysicalPlan implements LogicalPlan convert2PhysicalPlan interface.
 func (p *Apply) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo, *physicalPlanInfo, uint64, error) {
+	var err error
+	sortedPlanInfo, unSortedPlanInfo, count := p.getPlanInfo(prop)
+	if sortedPlanInfo != nil {
+		return sortedPlanInfo, unSortedPlanInfo, count, nil
+	}
 	child := p.GetChildByIndex(0).(LogicalPlan)
 	_, innerRes, _, err := p.InnerPlan.convert2PhysicalPlan(nil)
 	if err != nil {
@@ -499,79 +571,119 @@ func (p *Apply) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo,
 		InnerPlan:   innerRes.p,
 	}
 	np.SetSchema(p.GetSchema())
-	res0, res1, count, err := child.convert2PhysicalPlan(prop)
+	sortedPlanInfo, unSortedPlanInfo, count, err = child.convert2PhysicalPlan(prop)
 	if err != nil {
 		return nil, nil, 0, errors.Trace(err)
 	}
-	return addPlanToResponse(np, res0), addPlanToResponse(np, res1), count, nil
+	sortedPlanInfo = addPlanToResponse(np, sortedPlanInfo)
+	unSortedPlanInfo = addPlanToResponse(np, unSortedPlanInfo)
+	p.storePlanInfo(prop, sortedPlanInfo, unSortedPlanInfo, count)
+	return sortedPlanInfo, unSortedPlanInfo, count, nil
 }
 
 // convert2PhysicalPlan implements LogicalPlan convert2PhysicalPlan interface.
 func (p *Distinct) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo, *physicalPlanInfo, uint64, error) {
+	var err error
+	sortedPlanInfo, unSortedPlanInfo, count := p.getPlanInfo(prop)
+	if sortedPlanInfo != nil {
+		return sortedPlanInfo, unSortedPlanInfo, count, nil
+	}
 	child := p.GetChildByIndex(0).(LogicalPlan)
-	res0, res1, count, err := child.convert2PhysicalPlan(prop)
+	sortedPlanInfo, unSortedPlanInfo, count, err = child.convert2PhysicalPlan(prop)
 	if err != nil {
 		return nil, nil, 0, errors.Trace(err)
 	}
-	return addPlanToResponse(p, res0), addPlanToResponse(p, res1), uint64(float64(count) * distinctFactor), nil
+	sortedPlanInfo = addPlanToResponse(p, sortedPlanInfo)
+	unSortedPlanInfo = addPlanToResponse(p, unSortedPlanInfo)
+	count = uint64(float64(count) * distinctFactor)
+	p.storePlanInfo(prop, sortedPlanInfo, unSortedPlanInfo, count)
+	return sortedPlanInfo, unSortedPlanInfo, count, nil
 }
 
 // convert2PhysicalPlan implements LogicalPlan convert2PhysicalPlan interface.
 func (p *NewTableDual) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo, *physicalPlanInfo, uint64, error) {
-	res := &physicalPlanInfo{p: p, cost: 1.0}
-	return res, res, 1, nil
+	planInfo := &physicalPlanInfo{p: p, cost: 1.0}
+	return planInfo, planInfo, 1, nil
 }
 
 // convert2PhysicalPlan implements LogicalPlan convert2PhysicalPlan interface.
 func (p *MaxOneRow) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo, *physicalPlanInfo, uint64, error) {
+	var err error
+	sortedPlanInfo, unSortedPlanInfo, count := p.getPlanInfo(prop)
+	if sortedPlanInfo != nil {
+		return sortedPlanInfo, unSortedPlanInfo, count, nil
+	}
 	child := p.GetChildByIndex(0).(LogicalPlan)
-	res0, res1, count, err := child.convert2PhysicalPlan(prop)
+	sortedPlanInfo, unSortedPlanInfo, count, err = child.convert2PhysicalPlan(prop)
 	if err != nil {
 		return nil, nil, 0, errors.Trace(err)
 	}
-	return addPlanToResponse(p, res0), addPlanToResponse(p, res1), count, nil
+	sortedPlanInfo = addPlanToResponse(p, sortedPlanInfo)
+	unSortedPlanInfo = addPlanToResponse(p, unSortedPlanInfo)
+	return sortedPlanInfo, unSortedPlanInfo, count, nil
 }
 
 // convert2PhysicalPlan implements LogicalPlan convert2PhysicalPlan interface.
 func (p *Exists) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo, *physicalPlanInfo, uint64, error) {
+	var err error
+	sortedPlanInfo, unSortedPlanInfo, count := p.getPlanInfo(prop)
+	if sortedPlanInfo != nil {
+		return sortedPlanInfo, unSortedPlanInfo, count, nil
+	}
 	child := p.GetChildByIndex(0).(LogicalPlan)
-	res0, res1, count, err := child.convert2PhysicalPlan(prop)
+	sortedPlanInfo, unSortedPlanInfo, count, err = child.convert2PhysicalPlan(prop)
 	if err != nil {
 		return nil, nil, 0, errors.Trace(err)
 	}
-	return addPlanToResponse(p, res0), addPlanToResponse(p, res1), count, nil
+	sortedPlanInfo = addPlanToResponse(p, sortedPlanInfo)
+	unSortedPlanInfo = addPlanToResponse(p, unSortedPlanInfo)
+	return sortedPlanInfo, unSortedPlanInfo, count, nil
 }
 
 // convert2PhysicalPlan implements LogicalPlan convert2PhysicalPlan interface.
 func (p *Trim) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo, *physicalPlanInfo, uint64, error) {
+	var err error
+	sortedPlanInfo, unSortedPlanInfo, count := p.getPlanInfo(prop)
+	if sortedPlanInfo != nil {
+		return sortedPlanInfo, unSortedPlanInfo, count, nil
+	}
 	child := p.GetChildByIndex(0).(LogicalPlan)
-	res0, res1, count, err := child.convert2PhysicalPlan(prop)
+	sortedPlanInfo, unSortedPlanInfo, count, err = child.convert2PhysicalPlan(prop)
 	if err != nil {
 		return nil, nil, 0, errors.Trace(err)
 	}
-	return addPlanToResponse(p, res0), addPlanToResponse(p, res1), count, nil
+	sortedPlanInfo = addPlanToResponse(p, sortedPlanInfo)
+	unSortedPlanInfo = addPlanToResponse(p, unSortedPlanInfo)
+	return sortedPlanInfo, unSortedPlanInfo, count, nil
 }
 
 // convert2PhysicalPlan implements LogicalPlan convert2PhysicalPlan interface.
 func (p *SelectLock) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo, *physicalPlanInfo, uint64, error) {
+	var err error
+	sortedPlanInfo, unSortedPlanInfo, count := p.getPlanInfo(prop)
+	if sortedPlanInfo != nil {
+		return sortedPlanInfo, unSortedPlanInfo, count, nil
+	}
 	child := p.GetChildByIndex(0).(LogicalPlan)
-	res0, res1, count, err := child.convert2PhysicalPlan(prop)
+	sortedPlanInfo, unSortedPlanInfo, count, err = child.convert2PhysicalPlan(prop)
 	if err != nil {
 		return nil, nil, 0, errors.Trace(err)
 	}
-	return addPlanToResponse(p, res0), addPlanToResponse(p, res1), count, nil
+	sortedPlanInfo = addPlanToResponse(p, sortedPlanInfo)
+	unSortedPlanInfo = addPlanToResponse(p, unSortedPlanInfo)
+	return sortedPlanInfo, unSortedPlanInfo, count, nil
 }
 
 // convert2PhysicalPlan implements LogicalPlan convert2PhysicalPlan interface.
 func (p *Insert) convert2PhysicalPlan(prop requiredProperty) (*physicalPlanInfo, *physicalPlanInfo, uint64, error) {
 	if len(p.GetChildren()) == 0 {
-		res := &physicalPlanInfo{p: p}
-		return res, res, 0, nil
+		planInfo := &physicalPlanInfo{p: p}
+		return planInfo, planInfo, 0, nil
 	}
 	child := p.GetChildByIndex(0).(LogicalPlan)
-	res0, res1, count, err := child.convert2PhysicalPlan(prop)
+	sortedPlanInfo, unSortedPlanInfo, count, err := child.convert2PhysicalPlan(prop)
 	if err != nil {
 		return nil, nil, 0, errors.Trace(err)
 	}
-	return addPlanToResponse(p, res0), addPlanToResponse(p, res1), count, nil
+	return addPlanToResponse(p, sortedPlanInfo), addPlanToResponse(p, unSortedPlanInfo), count, nil
 }
diff --git a/plan/plan.go b/plan/plan.go
index 4a6e1c0472665..98909ae0ef45a 100644
--- a/plan/plan.go
+++ b/plan/plan.go
@@ -152,9 +152,33 @@ type PhysicalPlan interface {
 }
 
 type baseLogicalPlan struct {
+	propLen          int
+	sortedPlanInfo   *physicalPlanInfo
+	unSortedPlanInfo *physicalPlanInfo
+	count            uint64
 	basePlan
 }
 
+func (p *baseLogicalPlan) getPlanInfo(prop requiredProperty) (*physicalPlanInfo, *physicalPlanInfo, uint64) {
+	if p.sortedPlanInfo == nil {
+		return nil, nil, 0
+	}
+	if len(prop) > p.propLen {
+		return nil, nil, 0
+	}
+	if len(prop) == p.propLen {
+		return p.sortedPlanInfo, p.unSortedPlanInfo, p.count
+	}
+	return p.unSortedPlanInfo, p.unSortedPlanInfo, p.count
+}
+
+func (p *baseLogicalPlan) storePlanInfo(prop requiredProperty, sortedPlanInfo, unSortedPlanInfo *physicalPlanInfo, cnt uint64) {
+	p.propLen = len(prop)
+	p.sortedPlanInfo = sortedPlanInfo
+	p.unSortedPlanInfo = unSortedPlanInfo
+	p.count = cnt
+}
+
 func newBaseLogicalPlan(tp string, a *idAllocator) baseLogicalPlan {
 	return baseLogicalPlan{
 		basePlan: basePlan{
@@ -190,7 +214,7 @@ func (p *baseLogicalPlan) PruneColumnsAndResolveIndices(parentUsedCols []*expres
 	return outer, errors.Trace(err)
 }
 
-func (p *baseLogicalPlan) initID() {
+func (p *basePlan) initID() {
 	p.id = p.tp + p.allocator.allocID()
 }
 
diff --git a/plan/predicate_push_down.go b/plan/predicate_push_down.go
index 9e48e7a7c0729..e55668240be25 100644
--- a/plan/predicate_push_down.go
+++ b/plan/predicate_push_down.go
@@ -78,6 +78,16 @@ func (p *NewTableDual) PredicatePushDown(predicates []expression.Expression) ([]
 // PredicatePushDown implements LogicalPlan PredicatePushDown interface.
 func (p *Join) PredicatePushDown(predicates []expression.Expression) (ret []expression.Expression, retPlan LogicalPlan, err error) {
 	//TODO: add null rejecter.
+	groups, valid := tryToGetJoinGroup(p)
+	if valid {
+		e := joinReOrderSolver{allocator: p.allocator}
+		e.reorderJoin(groups, predicates)
+		newJoin := e.resultJoin
+		parent := p.parents[0]
+		newJoin.SetParents(parent)
+		parent.ReplaceChild(p, newJoin)
+		return newJoin.PredicatePushDown(predicates)
+	}
 	var leftCond, rightCond []expression.Expression
 	retPlan = p
 	leftPlan := p.GetChildByIndex(0).(LogicalPlan)
diff --git a/plan/stringer.go b/plan/stringer.go
index d4409a01b1535..d9b47f7932305 100644
--- a/plan/stringer.go
+++ b/plan/stringer.go
@@ -59,6 +59,11 @@ func toString(in Plan, strs []string, idxs []int) ([]string, []int) {
 		} else {
 			str = "LeftHashJoin{" + strings.Join(children, "->") + "}"
 		}
+		for _, eq := range x.EqualConditions {
+			l := eq.Args[0].ToString()
+			r := eq.Args[1].ToString()
+			str += fmt.Sprintf("(%s,%s)", l, r)
+		}
 	case *PhysicalHashSemiJoin:
 		last := len(idxs) - 1
 		idx := idxs[last]
diff --git a/server/conn.go b/server/conn.go
index 4d24c07f1013f..80c1dfe8c9dc8 100644
--- a/server/conn.go
+++ b/server/conn.go
@@ -382,7 +382,11 @@ func (cc *clientConn) handleQuery(sql string) (err error) {
 		err = cc.writeOK()
 	}
 	costTime := time.Now().Sub(startTs)
-	log.Debugf("[TIME_QUERY] %v %s", costTime, sql)
+	if costTime < time.Second {
+		log.Debugf("[TIME_QUERY] %v %s", costTime, sql)
+	} else {
+		log.Warnf("[TIME_QUERY] %v %s", costTime, sql)
+	}
 	metrics.Query(costTime)
 	return errors.Trace(err)
 }