func (ts *testSuite) TestConstraintNames(c *C) {
	se, _ := tidb.CreateSession(ts.store)
	ctx := se.(context.Context)
	schemaName := model.NewCIStr("test_constraint")
	tblName := model.NewCIStr("t")
	tbIdent := table.Ident{
		Schema: schemaName,
		Name:   tblName,
	}

	err := sessionctx.GetDomain(ctx).DDL().CreateSchema(ctx, tbIdent.Schema)
	c.Assert(err, IsNil)

	tbStmt := statement("create table t (a int, b int, index a (a, b), index a (a))").(*stmts.CreateTableStmt)
	err = sessionctx.GetDomain(ctx).DDL().CreateTable(ctx, tbIdent, tbStmt.Cols, tbStmt.Constraints)
	c.Assert(err, NotNil)

	tbStmt = statement("create table t (a int, b int, index A (a, b), index (a))").(*stmts.CreateTableStmt)
	err = sessionctx.GetDomain(ctx).DDL().CreateTable(ctx, tbIdent, tbStmt.Cols, tbStmt.Constraints)
	c.Assert(err, IsNil)
	tbl, err := sessionctx.GetDomain(ctx).InfoSchema().TableByName(schemaName, tblName)
	indices := tbl.Indices()
	c.Assert(len(indices), Equals, 2)
	c.Assert(indices[0].Name.O, Equals, "A")
	c.Assert(indices[1].Name.O, Equals, "a_2")

	err = sessionctx.GetDomain(ctx).DDL().DropSchema(ctx, tbIdent.Schema)
	c.Assert(err, IsNil)
}

// CreateSession creates a new session environment.
func CreateSession(store kv.Storage) (Session, error) {
	s := &session{
		values:      make(map[fmt.Stringer]interface{}),
		store:       store,
		sid:         atomic.AddInt64(&sessionID, 1),
		debugInfos:  make(map[string]interface{}),
		maxRetryCnt: 10,
		parser:      parser.New(),
	}
	domain, err := domap.Get(store)
	if err != nil {
		return nil, errors.Trace(err)
	}
	sessionctx.BindDomain(s, domain)

	variable.BindSessionVars(s)
	variable.GetSessionVars(s).SetStatusFlag(mysql.ServerStatusAutocommit, true)

	// session implements variable.GlobalVarAccessor. Bind it to ctx.
	variable.BindGlobalVarAccessor(s, s)

	// session implements autocommit.Checker. Bind it to ctx
	autocommit.BindAutocommitChecker(s, s)
	sessionMu.Lock()
	defer sessionMu.Unlock()

	ver := getStoreBootstrapVersion(store)
	if ver == notBootstrapped {
		// if no bootstrap and storage is remote, we must use a little lease time to
		// bootstrap quickly, after bootstrapped, we will reset the lease time.
		// TODO: Using a bootstap tool for doing this may be better later.
		if !localstore.IsLocalStore(store) {
			sessionctx.GetDomain(s).SetLease(chooseMinLease(100*time.Millisecond, schemaLease))
		}

		s.SetValue(context.Initing, true)
		bootstrap(s)
		s.ClearValue(context.Initing)

		if !localstore.IsLocalStore(store) {
			sessionctx.GetDomain(s).SetLease(schemaLease)
		}

		finishBootstrap(store)
	} else if ver < currentBootstrapVersion {
		s.SetValue(context.Initing, true)
		upgrade(s)
		s.ClearValue(context.Initing)
	}

	// TODO: Add auth here
	privChecker := &privileges.UserPrivileges{}
	privilege.BindPrivilegeChecker(s, privChecker)
	return s, nil
}

func (s *testDBSuite) TestTruncateTable(c *C) {
	defer testleak.AfterTest(c)
	store, err := tidb.NewStore("memory://truncate_table")
	c.Assert(err, IsNil)
	tk := testkit.NewTestKit(c, store)
	tk.MustExec("use test")
	tk.MustExec("create table t (c1 int, c2 int)")
	tk.MustExec("insert t values (1, 1), (2, 2)")
	ctx := tk.Se.(context.Context)
	is := sessionctx.GetDomain(ctx).InfoSchema()
	oldTblInfo, err := is.TableByName(model.NewCIStr("test"), model.NewCIStr("t"))
	c.Assert(err, IsNil)
	oldTblID := oldTblInfo.Meta().ID

	tk.MustExec("truncate table t")

	tk.MustExec("insert t values (3, 3), (4, 4)")
	tk.MustQuery("select * from t").Check(testkit.Rows("3 3", "4 4"))

	is = sessionctx.GetDomain(ctx).InfoSchema()
	newTblInfo, err := is.TableByName(model.NewCIStr("test"), model.NewCIStr("t"))
	c.Assert(err, IsNil)
	c.Assert(newTblInfo.Meta().ID, Greater, oldTblID)

	// verify that the old table data has been deleted by background worker.
	tablePrefix := tablecodec.EncodeTablePrefix(oldTblID)
	hasOldTableData := true
	for i := 0; i < 30; i++ {
		err = kv.RunInNewTxn(store, false, func(txn kv.Transaction) error {
			it, err1 := txn.Seek(tablePrefix)
			if err1 != nil {
				return err1
			}
			if !it.Valid() {
				hasOldTableData = false
			} else {
				hasOldTableData = it.Key().HasPrefix(tablePrefix)
			}
			it.Close()
			return nil
		})
		c.Assert(err, IsNil)
		if !hasOldTableData {
			break
		}
		time.Sleep(time.Millisecond * 100)
	}
	c.Assert(hasOldTableData, IsFalse)
}

func (s *session) checkSchemaValid() error {
	var ts uint64
	if s.txn != nil {
		ts = s.txn.StartTS()
	} else {
		s.schemaVerInCurrTxn = 0
	}

	var err error
	var currSchemaVer int64
	for i := 0; i < schemaExpiredRetryTimes; i++ {
		currSchemaVer, err = sessionctx.GetDomain(s).SchemaValidity.Check(ts, s.schemaVerInCurrTxn)
		if err == nil {
			if s.txn == nil {
				s.schemaVerInCurrTxn = currSchemaVer
			}
			return nil
		}
		log.Infof("schema version original %d, current %d, sleep time %v",
			s.schemaVerInCurrTxn, currSchemaVer, checkSchemaValiditySleepTime)
		if terror.ErrorEqual(err, domain.ErrInfoSchemaChanged) {
			break
		}
		time.Sleep(checkSchemaValiditySleepTime)
	}
	return errors.Trace(err)
}

// Compile compiles an ast.StmtNode to a stmt.Statement.
// If it is supported to use new plan and executer, it optimizes the node to
// a plan, and we wrap the plan in an adapter as stmt.Statement.
// If it is not supported, the node will be converted to old statement.
func (c *Compiler) Compile(ctx context.Context, node ast.StmtNode) (stmt.Statement, error) {
	if optimizer.IsSupported(node) {
		ast.SetFlag(node)
		if err := optimizer.Validate(node, false); err != nil {
			return nil, errors.Trace(err)
		}
		is := sessionctx.GetDomain(ctx).InfoSchema()
		if err := optimizer.ResolveName(node, is, ctx); err != nil {
			return nil, errors.Trace(err)
		}
		p, err := optimizer.Optimize(ctx, node)
		if err != nil {
			return nil, errors.Trace(err)
		}
		sa := &statementAdapter{
			is:   is,
			plan: p,
		}
		return sa, nil
	}
	c.converter = &converter.Converter{}
	s, err := c.converter.Convert(node)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return s, nil
}

func (s *testDBSuite) SetUpSuite(c *C) {
	trySkipDBTest(c)

	var err error

	s.schemaName = "test_db"

	uri := "memory://test"
	s.store, err = tidb.NewStore(uri)
	c.Assert(err, IsNil)

	s.db, err = sql.Open("tidb", fmt.Sprintf("%s/%s", uri, s.schemaName))
	c.Assert(err, IsNil)

	s.s, err = tidb.CreateSession(s.store)
	c.Assert(err, IsNil)

	s.mustExec(c, "create table t1 (c1 int, c2 int, c3 int, primary key(c1))")
	s.mustExec(c, "create table t2 (c1 int, c2 int, c3 int)")

	// set proper schema lease
	s.lease = 500 * time.Millisecond
	ctx := s.s.(context.Context)
	sessionctx.GetDomain(ctx).SetLease(s.lease)
}

func (s *ShowPlan) fetchShowDatabases(ctx context.Context) error {
	dbs := sessionctx.GetDomain(ctx).InfoSchema().AllSchemaNames()

	// TODO: let information_schema be the first database
	sort.Strings(dbs)
	m := map[interface{}]interface{}{}
	for _, d := range dbs {
		if s.Pattern != nil {
			s.Pattern.Expr = expression.Value{Val: d}
		} else if s.Where != nil {
			m[expression.ExprEvalIdentFunc] = func(name string) (interface{}, error) {
				if strings.EqualFold(name, "Database") {
					return d, nil
				}

				return nil, errors.Errorf("unknown field %s", name)
			}
		}
		match, err := s.evalCondition(ctx, m)
		if err != nil {
			return errors.Trace(err)
		}
		if !match {
			continue
		}
		s.rows = append(s.rows, &plan.Row{Data: []interface{}{d}})
	}
	return nil
}

func (s *ShowPlan) fetchShowTables(ctx context.Context) error {
	is := sessionctx.GetDomain(ctx).InfoSchema()
	dbName := model.NewCIStr(s.DBName)
	if !is.SchemaExists(dbName) {
		return errors.Errorf("Can not find DB: %s", dbName)
	}

	// sort for tables
	var tableNames []string
	for _, v := range is.SchemaTables(dbName) {
		tableNames = append(tableNames, v.TableName().L)
	}

	sort.Strings(tableNames)

	for _, v := range tableNames {
		data := []interface{}{v}
		if s.Full {
			// TODO: support "VIEW" later if we have supported view feature.
			// now, just use "BASE TABLE".
			data = append(data, "BASE TABLE")
		}
		s.rows = append(s.rows, &plan.Row{Data: data})
	}
	return nil
}

// Compile compiles an ast.StmtNode to a stmt.Statement.
// If it is supported to use new plan and executer, it optimizes the node to
// a plan, and we wrap the plan in an adapter as stmt.Statement.
// If it is not supported, the node will be converted to old statement.
func (c *Compiler) Compile(ctx context.Context, node ast.StmtNode) (stmt.Statement, error) {
	if optimizer.IsSupported(node) {
		ast.SetFlag(node)

		is := sessionctx.GetDomain(ctx).InfoSchema()
		if err := optimizer.Preprocess(node, is, ctx); err != nil {
			return nil, errors.Trace(err)
		}
		// Validate should be after NameResolve.
		if err := optimizer.Validate(node, false); err != nil {
			return nil, errors.Trace(err)
		}
		sb := NewSubQueryBuilder(is)
		p, err := optimizer.Optimize(ctx, node, sb)
		if err != nil {
			return nil, errors.Trace(err)
		}
		sa := &statementAdapter{
			is:   is,
			plan: p,
		}
		return sa, nil
	}
	c.converter = &converter.Converter{}
	s, err := c.converter.Convert(node)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return s, nil
}

// Exec implements the stmt.Statement Exec interface.
func (s *CreateIndexStmt) Exec(ctx context.Context) (rset.Recordset, error) {
	err := sessionctx.GetDomain(ctx).DDL().CreateIndex(ctx, s.TableIdent.Full(ctx), s.Unique, model.NewCIStr(s.IndexName), s.IndexColNames)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return nil, nil
}

func (s *testBinlogSuite) SetUpSuite(c *C) {
	logLevel := os.Getenv("log_level")
	log.SetLevelByString(logLevel)
	store, err := tikv.NewMockTikvStore()
	c.Assert(err, IsNil)
	s.store = store
	tidb.SetSchemaLease(0)
	s.unixFile = "/tmp/mock-binlog-pump"
	os.Remove(s.unixFile)
	l, err := net.Listen("unix", s.unixFile)
	c.Assert(err, IsNil)
	s.serv = grpc.NewServer()
	s.pump = new(mockBinlogPump)
	binlog.RegisterPumpServer(s.serv, s.pump)
	go s.serv.Serve(l)
	opt := grpc.WithDialer(func(addr string, timeout time.Duration) (net.Conn, error) {
		return net.DialTimeout("unix", addr, timeout)
	})
	clientCon, err := grpc.Dial(s.unixFile, opt, grpc.WithInsecure())
	c.Assert(err, IsNil)
	c.Assert(clientCon, NotNil)
	binloginfo.PumpClient = binlog.NewPumpClient(clientCon)
	s.tk = testkit.NewTestKit(c, s.store)
	s.tk.MustExec("use test")
	domain := sessionctx.GetDomain(s.tk.Se.(context.Context))
	s.ddl = domain.DDL()
}

func (s *session) Execute(sql string) ([]ast.RecordSet, error) {
	if err := s.checkSchemaValidOrRollback(); err != nil {
		return nil, errors.Trace(err)
	}
	charset, collation := getCtxCharsetInfo(s)
	rawStmts, err := s.ParseSQL(sql, charset, collation)
	if err != nil {
		log.Warnf("compiling %s, error: %v", sql, err)
		return nil, errors.Trace(err)
	}

	var rs []ast.RecordSet
	ph := sessionctx.GetDomain(s).PerfSchema()
	for i, rst := range rawStmts {
		st, err1 := Compile(s, rst)
		if err1 != nil {
			log.Errorf("Syntax error: %s", sql)
			log.Errorf("Error occurs at %s.", err1)
			return nil, errors.Trace(err1)
		}
		id := variable.GetSessionVars(s).ConnectionID
		s.stmtState = ph.StartStatement(sql, id, perfschema.CallerNameSessionExecute, rawStmts[i])
		r, err := runStmt(s, st)
		ph.EndStatement(s.stmtState)
		if err != nil {
			log.Warnf("session:%v, err:%v", s, err)
			return nil, errors.Trace(err)
		}
		if r != nil {
			rs = append(rs, r)
		}
	}
	return rs, nil
}

func (ts *testSuite) TestUniqueIndexMultipleNullEntries(c *C) {
	_, err := ts.se.Execute("CREATE TABLE test.t (a int primary key auto_increment, b varchar(255) unique)")
	c.Assert(err, IsNil)
	ctx := ts.se.(context.Context)
	dom := sessionctx.GetDomain(ctx)
	tb, err := dom.InfoSchema().TableByName(model.NewCIStr("test"), model.NewCIStr("t"))
	c.Assert(err, IsNil)
	c.Assert(tb.Meta().ID, Greater, int64(0))
	c.Assert(tb.Meta().Name.L, Equals, "t")
	c.Assert(tb.Meta(), NotNil)
	c.Assert(tb.Indices(), NotNil)
	c.Assert(string(tb.FirstKey()), Not(Equals), "")
	c.Assert(string(tb.IndexPrefix()), Not(Equals), "")
	c.Assert(string(tb.RecordPrefix()), Not(Equals), "")
	c.Assert(tables.FindIndexByColName(tb, "b"), NotNil)

	autoid, err := tb.AllocAutoID()
	c.Assert(err, IsNil)
	c.Assert(autoid, Greater, int64(0))

	_, err = tb.AddRecord(ctx, types.MakeDatums(1, nil))
	c.Assert(err, IsNil)
	_, err = tb.AddRecord(ctx, types.MakeDatums(2, nil))
	c.Assert(err, IsNil)
	_, err = ts.se.Execute("drop table test.t")
	c.Assert(err, IsNil)
}

func (ts *testInfoBinderSuite) TestInfoBinder(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 t (c1 int, c2 int)")
	domain := sessionctx.GetDomain(testKit.Se.(context.Context))

	src := "SELECT c1 from t"
	l := parser.NewLexer(src)
	c.Assert(parser.YYParse(l), Equals, 0)
	stmts := l.Stmts()
	c.Assert(len(stmts), Equals, 1)
	v := &optimizer.InfoBinder{
		Info:          domain.InfoSchema(),
		DefaultSchema: model.NewCIStr("test"),
	}
	selectStmt := stmts[0].(*ast.SelectStmt)
	selectStmt.Accept(v)

	verifier := &binderVerifier{
		c: c,
	}
	selectStmt.Accept(verifier)
}

// Open returns a new connection to the database.  The name is a string in a
// driver-specific format.
//
// Open may return a cached connection (one previously closed), but doing so is
// unnecessary; the sql package maintains a pool of idle connections for
// efficient re-use.
//
// The returned connection is only used by one goroutine at a time.
func (d *sqlDriver) Open(name string) (driver.Conn, error) {
	store, err := NewStore(name)
	if err != nil {
		return nil, errors.Trace(err)
	}

	driver := &sqlDriver{}
	switch {
	case strings.HasPrefix(name, "file://"):
		name = name[len("file://"):]
	case strings.HasPrefix(name, "memory://"):
		name = name[len("memory://"):]
	}
	name = filepath.Clean(name)
	if name == "" || name == "." || name == string(os.PathSeparator) {
		return nil, errors.Errorf("invalid DB name %q", name)
	}

	sess, err := CreateSession(store)
	if err != nil {
		return nil, errors.Trace(err)
	}
	s := sess.(*session)
	defer d.lock()()
	DBName := model.NewCIStr(name[strings.LastIndex(name, "/")+1:])
	domain := sessionctx.GetDomain(s)
	if !domain.InfoSchema().SchemaExists(DBName) {
		err = domain.DDL().CreateSchema(s, DBName)
		if err != nil {
			return nil, errors.Trace(err)
		}
	}

	return newDriverConn(s, driver, DBName.O)
}

// Exec implements the stmt.Statement Exec interface.
func (s *DropDatabaseStmt) Exec(ctx context.Context) (rset.Recordset, error) {
	err := sessionctx.GetDomain(ctx).DDL().DropSchema(ctx, model.NewCIStr(s.Name))
	if terror.ErrorEqual(err, ddl.ErrNotExists) && s.IfExists {
		err = nil
	}
	return nil, errors.Trace(err)
}

// Next implements Executor Next interface.
func (e *CheckTableExec) Next() (*Row, error) {
	if e.done {
		return nil, nil
	}

	dbName := model.NewCIStr(db.GetCurrentSchema(e.ctx))
	is := sessionctx.GetDomain(e.ctx).InfoSchema()

	for _, t := range e.tables {
		tb, err := is.TableByName(dbName, t.Name)
		if err != nil {
			return nil, errors.Trace(err)
		}
		for _, idx := range tb.Indices() {
			txn, err := e.ctx.GetTxn(false)
			if err != nil {
				return nil, errors.Trace(err)
			}
			err = inspectkv.CompareIndexData(txn, tb, idx)
			if err != nil {
				return nil, errors.Errorf("%v err:%v", t.Name, err)
			}
		}
	}
	e.done = true

	return nil, nil
}

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))
		}
	}
}

// Open returns a new connection to the database.
//
// The dsn must be a URL format 'engine://path/dbname?params'.
// Engine is the storage name registered with RegisterStore.
// Path is the storage specific format.
// Params is key-value pairs split by '&', optional params are storage specific.
// Examples:
//    goleveldb://relative/path/test
//    boltdb:///absolute/path/test
//    hbase://zk1,zk2,zk3/hbasetbl/test?tso=zk
//
// Open may return a cached connection (one previously closed), but doing so is
// unnecessary; the sql package maintains a pool of idle connections for
// efficient re-use.
//
// The returned connection is only used by one goroutine at a time.
func (d *sqlDriver) Open(dsn string) (driver.Conn, error) {
	storePath, dbName, err := parseDriverDSN(dsn)
	if err != nil {
		return nil, errors.Trace(err)
	}
	store, err := NewStore(storePath)
	if err != nil {
		return nil, errors.Trace(err)
	}

	sess, err := CreateSession(store)
	if err != nil {
		return nil, errors.Trace(err)
	}
	s := sess.(*session)

	d.lock()
	defer d.unlock()

	DBName := model.NewCIStr(dbName)
	domain := sessionctx.GetDomain(s)
	cs := &coldef.CharsetOpt{
		Chs: "utf8",
		Col: "utf8_bin",
	}
	if !domain.InfoSchema().SchemaExists(DBName) {
		err = domain.DDL().CreateSchema(s, DBName, cs)
		if err != nil {
			return nil, errors.Trace(err)
		}
	}
	driver := &sqlDriver{}
	return newDriverConn(s, driver, DBName.O)
}

func (ts *testSuite) TestTypes(c *C) {
	_, err := ts.se.Execute("CREATE TABLE test.t (c1 tinyint, c2 smallint, c3 int, c4 bigint, c5 text, c6 blob, c7 varchar(64), c8 time, c9 timestamp not null default CURRENT_TIMESTAMP, c10 decimal)")
	c.Assert(err, IsNil)
	ctx := ts.se.(context.Context)
	dom := sessionctx.GetDomain(ctx)
	_, err = dom.InfoSchema().TableByName(model.NewCIStr("test"), model.NewCIStr("t"))
	c.Assert(err, IsNil)
	_, err = ts.se.Execute("insert test.t values (1, 2, 3, 4, '5', '6', '7', '10:10:10', null, 1.4)")
	c.Assert(err, IsNil)
	rs, err := ts.se.Execute("select * from test.t where c1 = 1")
	row, err := rs[0].FirstRow()
	c.Assert(err, IsNil)
	c.Assert(row, NotNil)
	_, err = ts.se.Execute("drop table test.t")
	c.Assert(err, IsNil)

	_, err = ts.se.Execute("CREATE TABLE test.t (c1 tinyint unsigned, c2 smallint unsigned, c3 int unsigned, c4 bigint unsigned, c5 double)")
	c.Assert(err, IsNil)
	_, err = ts.se.Execute("insert test.t values (1, 2, 3, 4, 5)")
	c.Assert(err, IsNil)
	rs, err = ts.se.Execute("select * from test.t where c1 = 1")
	row, err = rs[0].FirstRow()
	c.Assert(err, IsNil)
	c.Assert(row, NotNil)
	_, err = ts.se.Execute("drop table test.t")
	c.Assert(err, IsNil)
}