func (s *testPlanSuite) TestNullRejectFinder(c *C) {
	cases := []struct {
		expr    string
		notNull bool
	}{
		{"a = 1", true},
		{"a != 100 and a > 0", true},
		{"a is null", false},
		{"a is not null", true},
		{"a is true", true},
		{"a is not true", false},
		{"a is false", true},
		{"a is not false", false},
		{"a != 0 and a is not false", true},
		{"a > 0 or true", false},
	}
	for _, ca := range cases {
		sql := "select * from t where " + ca.expr
		comment := Commentf("for expr %s", ca.expr)
		s, err := parser.ParseOneStmt(sql, "", "")
		c.Assert(err, IsNil, comment)
		finder := &nullRejectFinder{nullRejectTables: map[*ast.TableName]bool{}}
		stmt := s.(*ast.SelectStmt)
		mockResolve(stmt)
		stmt.Where.Accept(finder)
		if ca.notNull {
			c.Assert(finder.nullRejectTables, HasLen, 1, comment)
		} else {
			c.Assert(finder.nullRejectTables, HasLen, 0, comment)
		}
	}
}

func (s *testPlanSuite) TestMultiColumnIndex(c *C) {
	cases := []struct {
		sql              string
		accessEqualCount int
		usedColumnCount  int
	}{
		{"select * from t where c = 0 and d = 0 and e = 0", 3, 3},
		{"select * from t where c = 0 and d = 0 and e > 0", 2, 3},
		{"select * from t where d > 0 and e = 0 and c = 0", 1, 2},
	}
	for _, ca := range cases {
		comment := Commentf("for %s", ca.sql)
		s, err := parser.ParseOneStmt(ca.sql, "", "")
		c.Assert(err, IsNil, comment)
		stmt := s.(*ast.SelectStmt)
		ast.SetFlag(stmt)
		mockResolve(stmt)
		b := &planBuilder{}
		p := b.buildFrom(stmt)
		err = Refine(p)
		c.Assert(err, IsNil)
		idxScan, ok := p.(*IndexScan)
		c.Assert(ok, IsTrue)
		c.Assert(idxScan.AccessEqualCount, Equals, ca.accessEqualCount)
		c.Assert(idxScan.Ranges[0].LowVal, HasLen, ca.usedColumnCount)
	}
}

func statement(ctx context.Context, sql string) stmt.Statement {
	log.Debug("[ddl] Compile", sql)
	s, _ := parser.ParseOneStmt(sql, "", "")
	compiler := &executor.Compiler{}
	stm, _ := compiler.Compile(ctx, s)
	return stm
}

func (s *testPlanSuite) TestVisitCount(c *C) {
	sqls := []string{
		"select t1.c1, t2.c2 from t1, t2",
		"select * from t1 left join t2 on t1.c1 = t2.c1",
		"select * from t1 group by t1.c1 having sum(t1.c2) = 1",
		"select * from t1 where t1.c1 > 2 order by t1.c2 limit 100",
		"insert t1 values (1), (2)",
		"delete from t1 where false",
		"truncate table t1",
		"do 1",
		"show databases",
	}
	for _, sql := range sqls {
		stmt, err := parser.ParseOneStmt(sql, "", "")
		c.Assert(err, IsNil, Commentf(sql))
		ast.SetFlag(stmt)
		mockJoinResolve(c, stmt)
		b := &planBuilder{}
		p := b.build(stmt)
		c.Assert(b.err, IsNil)
		visitor := &countVisitor{}
		for i := 0; i < 5; i++ {
			visitor.skipAt = i
			visitor.enterCount = 0
			visitor.leaveCount = 0
			p.Accept(visitor)
			c.Assert(visitor.enterCount, Equals, visitor.leaveCount, Commentf(sql))
		}
	}
}

func (ts *testNameResolverSuite) TestNameResolver(c *C) {
	store, err := tidb.NewStore(tidb.EngineGoLevelDBMemory)
	c.Assert(err, IsNil)
	defer store.Close()
	testKit := testkit.NewTestKit(c, store)
	testKit.MustExec("use test")
	testKit.MustExec("create table t1 (c1 int, c2 int)")
	testKit.MustExec("create table t2 (c1 int, c2 int)")
	testKit.MustExec("create table t3 (c1 int, c2 int)")
	ctx := testKit.Se.(context.Context)
	domain := sessionctx.GetDomain(ctx)
	db.BindCurrentSchema(ctx, "test")
	for _, tc := range resolverTestCases {
		node, err := parser.ParseOneStmt(tc.src, "", "")
		c.Assert(err, IsNil)
		resolveErr := plan.ResolveName(node, domain.InfoSchema(), ctx)
		if tc.valid {
			c.Assert(resolveErr, IsNil)
			verifier := &resolverVerifier{c: c, src: tc.src}
			node.Accept(verifier)
		} else {
			c.Assert(resolveErr, NotNil, Commentf("%s", tc.src))
		}
	}
}

func (s *testPlanSuite) TestFilterRate(c *C) {
	cases := []struct {
		expr string
		rate float64
	}{
		{expr: "a = 1", rate: rateEqual},
		{expr: "a > 1", rate: rateGreaterOrLess},
		{expr: "a between 1 and 100", rate: rateBetween},
		{expr: "a is null", rate: rateIsNull},
		{expr: "a is not null", rate: rateFull - rateIsNull},
		{expr: "a is true", rate: rateFull - rateIsNull - rateIsFalse},
		{expr: "a is not true", rate: rateIsNull + rateIsFalse},
		{expr: "a is false", rate: rateIsFalse},
		{expr: "a is not false", rate: rateFull - rateIsFalse},
		{expr: "a like 'a'", rate: rateLike},
		{expr: "a not like 'a'", rate: rateFull - rateLike},
		{expr: "a in (1, 2, 3)", rate: rateEqual * 3},
		{expr: "a not in (1, 2, 3)", rate: rateFull - rateEqual*3},
		{expr: "a > 1 and a < 9", rate: float64(rateGreaterOrLess) * float64(rateGreaterOrLess)},
		{expr: "a = 1 or a = 2", rate: rateEqual + rateEqual - rateEqual*rateEqual},
		{expr: "a != 1", rate: rateNotEqual},
	}
	for _, ca := range cases {
		sql := "select 1 from dual where " + ca.expr
		s, err := parser.ParseOneStmt(sql, "", "")
		c.Assert(err, IsNil, Commentf("for expr %s", ca.expr))
		stmt := s.(*ast.SelectStmt)
		rate := guesstimateFilterRate(stmt.Where)
		c.Assert(rate, Equals, ca.rate, Commentf("for expr %s", ca.expr))
	}
}

func (s *testPlanSuite) TestBuilder(c *C) {
	cases := []struct {
		sqlStr  string
		planStr string
	}{
		{
			sqlStr:  "select 1",
			planStr: "Fields",
		},
		{
			sqlStr:  "select a from t",
			planStr: "Table(t)->Fields",
		},
		{
			sqlStr:  "select a from t where a = 1",
			planStr: "Table(t)->Filter->Fields",
		},
		{
			sqlStr:  "select a from t where a = 1 order by a",
			planStr: "Table(t)->Filter->Fields->Sort",
		},
		{
			sqlStr:  "select a from t where a = 1 order by a limit 1",
			planStr: "Table(t)->Filter->Fields->Sort->Limit",
		},
		{
			sqlStr:  "select a from t where a = 1 limit 1",
			planStr: "Table(t)->Filter->Fields->Limit",
		},
		{
			sqlStr:  "select a from t where a = 1 limit 1 for update",
			planStr: "Table(t)->Filter->Lock->Fields->Limit",
		},
		{
			sqlStr:  "admin show ddl",
			planStr: "ShowDDL",
		},
		{
			sqlStr:  "admin check table t",
			planStr: "CheckTable",
		},
	}
	var stmt ast.StmtNode
	for _, ca := range cases {
		s, err := parser.ParseOneStmt(ca.sqlStr, "", "")
		c.Assert(err, IsNil, Commentf("for expr %s", ca.sqlStr))
		if strings.HasPrefix(ca.sqlStr, "select") {
			stmt = s.(*ast.SelectStmt)
		} else if strings.HasPrefix(ca.sqlStr, "admin") {
			stmt = s.(*ast.AdminStmt)
		}
		mockResolve(stmt)
		p, err := BuildPlan(stmt)
		c.Assert(err, IsNil)
		explainStr, err := Explain(p)
		c.Assert(err, IsNil)
		c.Assert(explainStr, Equals, ca.planStr, Commentf("for expr %s", ca.sqlStr))
	}
}

func (s *testPlanSuite) TestIndexHint(c *C) {
	defer testleak.AfterTest(c)()
	cases := []struct {
		sql     string
		explain string
	}{
		{
			"select * from t1 force index (i1) where t1.i1 > 0 and t1.i2 = 0",
			"Index(t1.i1)->Fields",
		},
		{
			"select * from t1 use index (i1) where t1.i1 > 0 and t1.i2 = 0",
			"Index(t1.i1)->Fields",
		},
		{
			"select * from t1 ignore index (i2) where t1.i1 > 0 and t1.i2 = 0",
			"Index(t1.i1)->Fields",
		},
		{
			"select * from t1 use index (i1, i2) where t1.i1 > 0 and t1.i2 between 0 and 2 and t1.i3 = 0",
			"Index(t1.i2)->Fields",
		},
		{
			"select * from t1 ignore index (i1, i2, i3) where t1.i1 = 0 and t1.i2 = 0 and t1.i3 = 0",
			"Table(t1)->Fields",
		},
		{
			"select * from t1 use index () where t1.i1 = 0 and t1.i2 = 0 and t1.i3 = 0",
			"Table(t1)->Fields",
		},
		{
			"select * from t1 use index (i1) ignore index (i1) where t1.i1 = 0",
			"Table(t1)->Fields",
		},
	}
	for _, ca := range cases {
		comment := Commentf("for %s", ca.sql)
		s, err := parser.ParseOneStmt(ca.sql, "", "")
		c.Assert(err, IsNil, comment)
		stmt := s.(*ast.SelectStmt)
		mockJoinResolve(c, stmt)
		ast.SetFlag(stmt)
		p, err := BuildPlan(stmt, nil)
		c.Assert(err, IsNil)
		c.Assert(ToString(p), Equals, ca.explain, comment)
	}
}

func (s *testPlanSuite) TestBuilder(c *C) {
	cases := []struct {
		sqlStr  string
		planStr string
	}{
		{
			sqlStr:  "select 1",
			planStr: "Fields",
		},
		{
			sqlStr:  "select a from t",
			planStr: "Table(t)->Fields",
		},
		{
			sqlStr:  "select a from t where a = 1",
			planStr: "Table(t)->Filter->Fields",
		},
		{
			sqlStr:  "select a from t where a = 1 order by a",
			planStr: "Table(t)->Filter->Fields->Sort",
		},
		{
			sqlStr:  "select a from t where a = 1 order by a limit 1",
			planStr: "Table(t)->Filter->Fields->Sort->Limit",
		},
		{
			sqlStr:  "select a from t where a = 1 limit 1",
			planStr: "Table(t)->Filter->Fields->Limit",
		},
		{
			sqlStr:  "select a from t where a = 1 limit 1 for update",
			planStr: "Table(t)->Filter->Lock->Fields->Limit",
		},
	}
	for _, ca := range cases {
		s, err := parser.ParseOneStmt(ca.sqlStr, "", "")
		c.Assert(err, IsNil, Commentf("for expr %s", ca.sqlStr))
		stmt := s.(*ast.SelectStmt)
		mockResolve(stmt)
		p, err := BuildPlan(stmt)
		c.Assert(err, IsNil)
		explainStr, err := Explain(p)
		c.Assert(err, IsNil)
		c.Assert(explainStr, Equals, ca.planStr, Commentf("for expr %s", ca.sqlStr))
	}
}

func (s *testPlanSuite) TestSplitWhere(c *C) {
	cases := []struct {
		expr  string
		count int
	}{
		{"a = 1 and b = 2 and c = 3", 3},
		{"(a = 1 and b = 2) and c = 3", 3},
		{"a = 1 and (b = 2 and c = 3 or d = 4)", 2},
		{"a = 1 and (b = 2 or c = 3) and d = 4", 3},
		{"(a = 1 and b = 2) and (c = 3 and d = 4)", 4},
	}
	for _, ca := range cases {
		sql := "select 1 from dual where " + ca.expr
		comment := Commentf("for expr %s", ca.expr)
		s, err := parser.ParseOneStmt(sql, "", "")
		c.Assert(err, IsNil, comment)
		stmt := s.(*ast.SelectStmt)
		conditions := splitWhere(stmt.Where)
		c.Assert(conditions, HasLen, ca.count, comment)
	}
}

func parseExpr(c *C, expr string) ast.ExprNode {
	s, err := parser.ParseOneStmt("select "+expr, "", "")
	c.Assert(err, IsNil)
	stmt := s.(*ast.SelectStmt)
	return stmt.Fields.Fields[0].Expr
}

// ParseDDL only use to parse ddl sql, no longer use.
// sqlparser will native support DDL.
func ParseDDL(sql string) (ast.StmtNode, error) {
	return parser.ParseOneStmt(sql, "", "")
}

func (sei *session) handleDDL(sql string) error {
	stmt, err := parser.ParseOneStmt(sql, "", "")
	if err != nil {
		glog.Infof("parse ddl sql(%s) error:%v", sql, err)
		return sei.writeError(mysql.NewDefaultError(mysql.ER_SYNTAX_ERROR))
	}
	switch v := stmt.(type) {
	case *ast.CreateDatabaseStmt:
		dbname := v.Name
		id, rows, err := sei.ddlManage().CreateDatabase(sei.user, dbname, sei.dbnum)
		glog.Infof("DDL plan id(%v)", id)
		if err != nil {
			glog.Infof("CREATE DATABASE has an error(%v)", err)
			err = sei.writeError(err)
		} else {
			// one time only creata a db
			r := &mysql.Result{
				AffectedRows: rows,
			}
			err = sei.writeOK(r)
		}
		return err

	case *ast.CreateTableStmt:
		if sei.db == "" {
			return sei.writeError(mysql.NewDefaultError(mysql.ER_NO_DB_ERROR))
		}
		table := &meta.Table{
			Scheme: "hash",
			Name:   v.Table.Name.String(),
			PartitionKey: &meta.Key{
				Name: sei.partitionKey,
			},
			ColsLen: len(v.Cols),
		}

		existMap := make(map[string]bool)
		// get constraints
		for _, constraint := range v.Constraints {
			if constraint.Tp == ast.ConstraintPrimaryKey ||
				constraint.Tp == ast.ConstraintUniq {
				if len(constraint.Keys) > 1 {
					err := mysql.NewError(mysql.ER_SYNTAX_ERROR,
						"not support constraint keys' length > 1")
					return sei.writeError(err)
				}
				// get type
				name := constraint.Keys[0].Column.Name.String()
				index, typ := sei.getFieldType(v.Cols, name)
				if typ == meta.TypeKeyUnknow {
					err := mysql.NewError(mysql.ER_SYNTAX_ERROR,
						"unsupport key's type ")
					return sei.writeError(err)
				}

				if constraint.Tp == ast.ConstraintPrimaryKey && sei.partitionKey == "" {
					// set primary key for partition key
					table.PartitionKey.Name = name
					table.PartitionKey.Type = typ
					table.PartitionKey.Index = index
				}

				table.AutoKeys = append(table.AutoKeys, &meta.Key{
					Name:  name,
					Type:  typ,
					Index: index,
				})
				existMap[name] = true
			}
		}
		// check auto increment
		for _, col := range v.Cols {
			for n, option := range col.Options {
				t := sei.getOneFeildType(col)
				if t == meta.TypeKeyUnknow {
					err := mysql.NewError(mysql.ER_SYNTAX_ERROR,
						"unsupport key's type ")
					return sei.writeError(err)
				}
				switch option.Tp {
				case ast.ColumnOptionAutoIncrement, ast.ColumnOptionPrimaryKey, ast.ColumnOptionUniq:
					if ast.ColumnOptionPrimaryKey == option.Tp && table.PartitionKey.Name == "" {
						table.PartitionKey.Name = col.Name.Name.String()
						table.PartitionKey.Type = t
						table.PartitionKey.Index = n
					}
					// check if exist not append
					if existMap[col.Name.Name.String()] {
						continue
					}
					table.AutoKeys = append(table.AutoKeys, &meta.Key{
						Name:  col.Name.Name.String(),
						Type:  t,
						Index: n,
					})
					existMap[col.Name.Name.String()] = true
				}
				if option.Tp == ast.ColumnOptionAutoIncrement {
					glog.Infof("record auto increment option index(%v)", n)
					// record
//.........这里部分代码省略.........

func (s *testPlanSuite) TestJoinPath(c *C) {
	cases := []struct {
		sql     string
		explain string
	}{
		{
			"select * from t1, t2 where t1.c1 > 0",
			"InnerJoin{Table(t1)->Table(t2)}->Fields",
		},
		{
			"select * from t1 left join t2 on 1 where t2.c1 != 0",
			"InnerJoin{Table(t2)->Table(t1)}->Fields",
		},
		{
			"select * from t1 left join t2 on 1 where t2.c1 != 0 or t1.c1 != 0",
			"OuterJoin{Table(t1)->Table(t2)}->Fields",
		},
		{
			"select * from t1 left join t2 on t1.i1 = t2.i1 where t1.i1 = 1",
			"OuterJoin{Index(t1.i1)->Index(t2.i1)}->Fields",
		},
		{
			"select * from t1 join t2 on t2.c1 = t1.i1",
			"InnerJoin{Table(t2)->Index(t1.i1)}->Fields",
		},
		{
			"select * from t1, t2 where t2.c1 = t1.i1",
			"InnerJoin{Table(t2)->Index(t1.i1)}->Fields",
		},
		{
			`select * from
				t1 left join
					(t2 join
						t3
					on t2.i1 = t3.c1)
				on t1.c1 = t2.i2`,
			"OuterJoin{Table(t1)->InnerJoin{Index(t2.i2)->Table(t3)}}->Fields",
		},
		{
			`select * from
				(t1, t2) left join
					t3
				on t2.c1 = t3.i1
			where t2.i2 between 1 and 4 and t1.i1 = 3`,
			"OuterJoin{InnerJoin{Index(t1.i1)->Index(t2.i2)}->Index(t3.i1)}->Fields",
		},
		{
			`select * from
				t1 join (
					(t2 join t3
						on t2.i3 = t3.i3 and t2.c2 > t3.c3
					) left join t4
					on t3.c3 = t4.i4
				)
				on t1.i1 = 1 and t1.c1 = t2.i2`,
			"InnerJoin{Index(t1.i1)->OuterJoin{InnerJoin{Index(t2.i2)->Index(t3.i3)}->Index(t4.i4)}}->Fields",
		},
		{
			`select * from
				t1 join (
					t2 left join (
						t3 join t4
						on t3.i3 = t4.c4
					)
					on t2.i2 = t3.c3
				)
				on t1.i1 = t2.c2`,
			"InnerJoin{OuterJoin{Table(t2)->InnerJoin{Table(t4)->Index(t3.i3)}}->Index(t1.i1)}->Fields",
		},
		{
			`select * from
				t1 join (
					(t2 join t3
						on t2.i2 = t3.i3
					)
				) on t1.i1 = t2.i2
				where t1.i1 = 1`,
			"InnerJoin{Index(t1.i1)->Index(t2.i2)->Index(t3.i3)}->Fields",
		},
	}
	for _, ca := range cases {
		comment := Commentf("for %s", ca.sql)
		s, err := parser.ParseOneStmt(ca.sql, "", "")
		c.Assert(err, IsNil, comment)
		stmt := s.(*ast.SelectStmt)
		mockJoinResolve(c, stmt)
		ast.SetFlag(stmt)
		p, err := BuildPlan(stmt, nil)
		c.Assert(err, IsNil)
		c.Assert(ToString(p), Equals, ca.explain, comment)
	}
}

func (ts *testFlagSuite) TestFlag(c *C) {
	cases := []struct {
		expr string
		flag uint64
	}{
		{
			"1 between 0 and 2",
			ast.FlagConstant,
		},
		{
			"case 1 when 1 then 1 else 0 end",
			ast.FlagConstant,
		},
		{
			"case 1 when 1 then 1 else 0 end",
			ast.FlagConstant,
		},
		{
			"1 = ANY (select 1) OR exists (select 1)",
			ast.FlagHasSubquery,
		},
		{
			"1 in (1) or 1 is true or null is null or 'abc' like 'abc' or 'abc' rlike 'abc'",
			ast.FlagConstant,
		},
		{
			"row (1, 1) = row (1, 1)",
			ast.FlagConstant,
		},
		{
			"(1 + a) > ?",
			ast.FlagHasReference | ast.FlagHasParamMarker,
		},
		{
			"trim('abc ')",
			ast.FlagHasFunc,
		},
		{
			"now() + EXTRACT(YEAR FROM '2009-07-02') + CAST(1 AS UNSIGNED)",
			ast.FlagHasFunc,
		},
		{
			"substring('abc', 1)",
			ast.FlagHasFunc,
		},
		{
			"sum(a)",
			ast.FlagHasAggregateFunc | ast.FlagHasReference,
		},
		{
			"(select 1) as a",
			ast.FlagHasSubquery,
		},
		{
			"@auto_commit",
			ast.FlagHasVariable,
		},
		{
			"default(a)",
			ast.FlagHasDefault,
		},
	}
	for _, ca := range cases {
		stmt, err := parser.ParseOneStmt("select "+ca.expr, "", "")
		c.Assert(err, IsNil)
		selectStmt := stmt.(*ast.SelectStmt)
		ast.SetFlag(selectStmt)
		expr := selectStmt.Fields.Fields[0].Expr
		c.Assert(expr.GetFlag(), Equals, ca.flag, Commentf("For %s", ca.expr))
	}
}

//.........这里部分代码省略.........
		{
			sql: "select count(*) from t a join t b on a.a = b.d order by sum(a.d)",
			ans: map[string][]string{
				"TableScan_1": {"a", "d"},
				"TableScan_2": {"d"},
			},
		},
		{
			sql: "select count(b.a) from t a join t b on a.a = b.d group by b.b order by sum(a.d)",
			ans: map[string][]string{
				"TableScan_1": {"a", "d"},
				"TableScan_2": {"a", "b", "d"},
			},
		},
		{
			sql: "select * from (select count(b.a) from t a join t b on a.a = b.d group by b.b having sum(a.d) < 0) tt",
			ans: map[string][]string{
				"TableScan_1": {"a", "d"},
				"TableScan_2": {"a", "b", "d"},
			},
		},
		{
			sql: "select (select count(a) from t where b = k.a) from t k",
			ans: map[string][]string{
				"TableScan_1": {"a"},
				"TableScan_2": {"a", "b"},
			},
		},
		{
			sql: "select exists (select count(*) from t where b = k.a) from t k",
			ans: map[string][]string{
				"TableScan_1": {"a"},
				"TableScan_2": {"b"},
			},
		},
		{
			sql: "select b = (select count(*) from t where b = k.a) from t k",
			ans: map[string][]string{
				"TableScan_1": {"a", "b"},
				"TableScan_2": {"b"},
			},
		},
		{
			sql: "select exists (select count(a) from t where b = k.a) from t k",
			ans: map[string][]string{
				"TableScan_1": {"a"},
				"TableScan_2": {"b"},
			},
		},
		{
			sql: "select a as c1, b as c2 from t order by 1, c1 + c2 + c",
			ans: map[string][]string{
				"TableScan_1": {"a", "b", "c"},
			},
		},
		{
			sql: "select a from t where b < any (select c from t)",
			ans: map[string][]string{
				"TableScan_1": {"a", "b"},
				"TableScan_2": {"c"},
			},
		},
		{
			sql: "select a from t where (b,a) = all (select c,d from t)",
			ans: map[string][]string{
				"TableScan_1": {"a", "b"},
				"TableScan_2": {"c", "d"},
			},
		},
		{
			sql: "select a from t where (b,a) in (select c,d from t)",
			ans: map[string][]string{
				"TableScan_1": {"a", "b"},
				"TableScan_2": {"c", "d"},
			},
		},
	}
	for _, ca := range cases {
		comment := Commentf("for %s", ca.sql)
		stmt, err := parser.ParseOneStmt(ca.sql, "", "")
		c.Assert(err, IsNil, comment)
		ast.SetFlag(stmt)

		err = newMockResolve(stmt)
		c.Assert(err, IsNil)

		builder := &planBuilder{
			colMapper: make(map[*ast.ColumnNameExpr]expression.Expression),
			allocator: new(idAllocator)}
		p := builder.build(stmt).(LogicalPlan)
		c.Assert(builder.err, IsNil, comment)

		_, err = p.PredicatePushDown(nil)
		c.Assert(err, IsNil)
		_, err = p.PruneColumnsAndResolveIndices(p.GetSchema())
		c.Assert(err, IsNil)
		check(p, c, ca.ans, comment)
	}
	UseNewPlanner = false
}

func (s *testPlanSuite) TestColumnPruning(c *C) {
	UseNewPlanner = true
	defer testleak.AfterTest(c)()
	cases := []struct {
		sql string
		ans map[string][]string
	}{
		{
			sql: "select count(*) from t group by a",
			ans: map[string][]string{
				"*plan.NewTableScan_1": {"a"},
			},
		},
		{
			sql: "select count(*) from t",
			ans: map[string][]string{
				"*plan.NewTableScan_1": {},
			},
		},
		{
			sql: "select count(*) from t a join t b where a.a < 1",
			ans: map[string][]string{
				"*plan.NewTableScan_1": {"a"},
				"*plan.NewTableScan_2": {},
			},
		},
		{
			sql: "select count(*) from t a join t b on a.a = b.d",
			ans: map[string][]string{
				"*plan.NewTableScan_1": {"a"},
				"*plan.NewTableScan_2": {"d"},
			},
		},
		{
			sql: "select count(*) from t a join t b on a.a = b.d order by sum(a.d)",
			ans: map[string][]string{
				"*plan.NewTableScan_1": {"a", "d"},
				"*plan.NewTableScan_2": {"d"},
			},
		},
		{
			sql: "select count(b.a) from t a join t b on a.a = b.d group by b.b order by sum(a.d)",
			ans: map[string][]string{
				"*plan.NewTableScan_1": {"a", "d"},
				"*plan.NewTableScan_2": {"a", "b", "d"},
			},
		},
		{
			sql: "select * from (select count(b.a) from t a join t b on a.a = b.d group by b.b having sum(a.d) < 0) tt",
			ans: map[string][]string{
				"*plan.NewTableScan_1": {"a", "d"},
				"*plan.NewTableScan_2": {"a", "b", "d"},
			},
		},
		{
			sql: "select (select count(a) from t where b = k.a) from t k",
			ans: map[string][]string{
				"*plan.NewTableScan_1": {"a"},
				"*plan.NewTableScan_2": {"a", "b"},
			},
		},
		{
			sql: "select exists (select count(*) from t where b = k.a) from t k",
			ans: map[string][]string{
				"*plan.NewTableScan_1": {"a"},
				"*plan.NewTableScan_2": {"b"},
			},
		},
		{
			sql: "select b = (select count(*) from t where b = k.a) from t k",
			ans: map[string][]string{
				"*plan.NewTableScan_1": {"a", "b"},
				"*plan.NewTableScan_2": {"b"},
			},
		},
		{
			sql: "select exists (select count(a) from t where b = k.a) from t k",
			ans: map[string][]string{
				"*plan.NewTableScan_1": {"a"},
				"*plan.NewTableScan_2": {"b"},
			},
		},
	}
	for _, ca := range cases {
		comment := Commentf("for %s", ca.sql)
		stmt, err := parser.ParseOneStmt(ca.sql, "", "")
		c.Assert(err, IsNil, comment)
		ast.SetFlag(stmt)

		err = newMockResolve(stmt)
		c.Assert(err, IsNil)

		builder := &planBuilder{}
		p := builder.build(stmt)
		c.Assert(builder.err, IsNil)

		_, err = builder.predicatePushDown(p, []expression.Expression{})
		c.Assert(err, IsNil)
		_, _, err = pruneColumnsAndResolveIndices(p, p.GetSchema())
		c.Assert(err, IsNil)
//.........这里部分代码省略.........

func (s *testPlanSuite) TestPredicatePushDown(c *C) {
	UseNewPlanner = true
	defer testleak.AfterTest(c)()
	cases := []struct {
		sql   string
		first string
		best  string
	}{
		{
			sql:   "select a from (select a from t where d = 0) k where k.a = 5",
			first: "DataScan(t)->Selection->Projection->Selection->Projection",
			best:  "DataScan(t)->Selection->Projection->Projection",
		},
		{
			sql:   "select a from (select 1+2 as a from t where d = 0) k where k.a = 5",
			first: "DataScan(t)->Selection->Projection->Selection->Projection",
			best:  "DataScan(t)->Selection->Projection->Selection->Projection",
		},
		{
			sql:   "select a from (select d as a from t where d = 0) k where k.a = 5",
			first: "DataScan(t)->Selection->Projection->Selection->Projection",
			best:  "DataScan(t)->Selection->Projection->Projection",
		},
		{
			sql:   "select * from t ta join t tb on ta.d = tb.d and ta.d > 1 where tb.a = 0",
			first: "Join{DataScan(t)->DataScan(t)}->Selection->Projection",
			best:  "Join{DataScan(t)->Selection->DataScan(t)->Selection}->Projection",
		},
		{
			sql:   "select * from t ta join t tb on ta.d = tb.d where ta.d > 1 and tb.a = 0",
			first: "Join{DataScan(t)->DataScan(t)}->Selection->Projection",
			best:  "Join{DataScan(t)->Selection->DataScan(t)->Selection}->Projection",
		},
		{
			sql:   "select * from t ta left outer join t tb on ta.d = tb.d and ta.d > 1 where tb.a = 0",
			first: "Join{DataScan(t)->DataScan(t)}->Selection->Projection",
			best:  "Join{DataScan(t)->DataScan(t)}->Selection->Projection",
		},
		{
			sql:   "select * from t ta right outer join t tb on ta.d = tb.d and ta.a > 1 where tb.a = 0",
			first: "Join{DataScan(t)->DataScan(t)}->Selection->Projection",
			best:  "Join{DataScan(t)->Selection->DataScan(t)->Selection}->Projection",
		},
		{
			sql:   "select a, d from (select * from t union all select * from t union all select * from t) z where a < 10",
			first: "UnionAll{DataScan(t)->Projection->DataScan(t)->Projection->DataScan(t)->Projection}->Selection->Projection",
			best:  "UnionAll{DataScan(t)->Selection->Projection->DataScan(t)->Selection->Projection->DataScan(t)->Selection->Projection}->Projection",
		},
		{
			sql:   "select (select count(*) from t where t.a = k.a) from t k",
			first: "DataScan(t)->Apply(DataScan(t)->Selection->Aggr->Projection->MaxOneRow)->Projection",
			best:  "DataScan(t)->Apply(DataScan(t)->Selection->Aggr->Projection->MaxOneRow)->Projection",
		},
		{
			sql:   "select a from t where exists(select 1 from t as x where x.a < t.a)",
			first: "DataScan(t)->Apply(DataScan(t)->Selection->Projection->Exists)->Selection->Projection",
			best:  "DataScan(t)->Apply(DataScan(t)->Selection->Projection->Exists)->Selection->Projection",
		},
	}
	for _, ca := range cases {
		comment := Commentf("for %s", ca.sql)
		stmt, err := parser.ParseOneStmt(ca.sql, "", "")
		c.Assert(err, IsNil, comment)
		ast.SetFlag(stmt)

		err = newMockResolve(stmt)
		c.Assert(err, IsNil)

		builder := &planBuilder{}
		p := builder.build(stmt)
		c.Assert(builder.err, IsNil)
		c.Assert(ToString(p), Equals, ca.first, Commentf("for %s", ca.sql))

		_, err = builder.predicatePushDown(p, []expression.Expression{})
		c.Assert(err, IsNil)
		_, _, err = pruneColumnsAndResolveIndices(p, p.GetSchema())
		c.Assert(err, IsNil)
		c.Assert(ToString(p), Equals, ca.best, Commentf("for %s", ca.sql))
	}
	UseNewPlanner = false
}

func (s *testPlanSuite) TestBestPlan(c *C) {
	cases := []struct {
		sql  string
		best string
	}{
		{
			sql:  "select * from t",
			best: "Table(t)->Fields",
		},
		{
			sql:  "select * from t order by a",
			best: "Table(t)->Fields",
		},
		{
			sql:  "select * from t where b = 1 order by a",
			best: "Index(t.b)->Filter->Fields->Sort",
		},
		{
			sql:  "select * from t where (a between 1 and 2) and (b = 3)",
			best: "Index(t.b)->Filter->Fields",
		},
		{
			sql:  "select * from t where a > 0 order by b limit 100",
			best: "Index(t.b)->Filter->Fields->Limit",
		},
		{
			sql:  "select * from t where d = 0",
			best: "Table(t)->Filter->Fields",
		},
		{
			sql:  "select * from t where c = 0 and d = 0",
			best: "Index(t.c_d)->Filter->Fields",
		},
		{
			sql:  "select * from t where b like 'abc%'",
			best: "Index(t.b)->Filter->Fields",
		},
		{
			sql:  "select * from t where d",
			best: "Table(t)->Filter->Fields",
		},
		{
			sql:  "select * from t where a is null",
			best: "Range(t)->Filter->Fields",
		},
	}
	for _, ca := range cases {
		comment := Commentf("for %s", ca.sql)
		s, err := parser.ParseOneStmt(ca.sql, "", "")
		c.Assert(err, IsNil, comment)
		stmt := s.(*ast.SelectStmt)
		ast.SetFlag(stmt)
		mockResolve(stmt)

		p, err := BuildPlan(stmt)
		c.Assert(err, IsNil)
		alts, err := Alternatives(p)
		c.Assert(err, IsNil)

		err = Refine(p)
		c.Assert(err, IsNil)
		bestCost := EstimateCost(p)
		bestPlan := p

		for _, alt := range alts {
			c.Assert(Refine(alt), IsNil)
			cost := EstimateCost(alt)
			if cost < bestCost {
				bestCost = cost
				bestPlan = alt
			}
		}
		explainStr, err := Explain(bestPlan)
		c.Assert(err, IsNil)
		c.Assert(explainStr, Equals, ca.best, Commentf("for %s cost %v", ca.sql, bestCost))
	}
}

func (s *testPlanSuite) TestBestPlan(c *C) {
	cases := []struct {
		sql  string
		best string
	}{
		{
			sql:  "select * from t",
			best: "Table(t)->Fields",
		},
		{
			sql:  "select * from t order by a",
			best: "Table(t)->Fields",
		},
		{
			sql:  "select * from t where b = 1 order by a",
			best: "Index(t.b)->Fields->Sort",
		},
		{
			sql:  "select * from t where (a between 1 and 2) and (b = 3)",
			best: "Index(t.b)->Fields",
		},
		{
			sql:  "select * from t where a > 0 order by b limit 100",
			best: "Index(t.b)->Fields->Limit",
		},
		{
			sql:  "select * from t where d = 0",
			best: "Table(t)->Fields",
		},
		{
			sql:  "select * from t where c = 0 and d = 0",
			best: "Index(t.c_d_e)->Fields",
		},
		{
			sql:  "select * from t where c = 0 and d = 0 and e = 0",
			best: "Index(t.c_d_e)->Fields",
		},
		{
			sql:  "select * from t where (d = 0 and e = 0) and c = 0",
			best: "Index(t.c_d_e)->Fields",
		},
		{
			sql:  "select * from t where e = 0 and (d = 0 and c = 0)",
			best: "Index(t.c_d_e)->Fields",
		},
		{
			sql:  "select * from t where b like 'abc%'",
			best: "Index(t.b)->Fields",
		},
		{
			sql:  "select * from t where d",
			best: "Table(t)->Fields",
		},
		{
			sql:  "select * from t where a is null",
			best: "Range(t)->Fields",
		},
		{
			sql:  "select a from t where a = 1 limit 1 for update",
			best: "Range(t)->Lock->Fields->Limit",
		},
		{
			sql:  "admin show ddl",
			best: "ShowDDL",
		},
		{
			sql:  "admin check table t",
			best: "CheckTable",
		},
	}
	for _, ca := range cases {
		comment := Commentf("for %s", ca.sql)
		stmt, err := parser.ParseOneStmt(ca.sql, "", "")
		c.Assert(err, IsNil, comment)
		ast.SetFlag(stmt)
		mockResolve(stmt)

		p, err := BuildPlan(stmt, nil)
		c.Assert(err, IsNil)

		err = Refine(p)
		c.Assert(ToString(p), Equals, ca.best, Commentf("for %s cost %v", ca.sql, EstimateCost(p)))
	}
}