// AddRecord implements table.Table AddRecord interface.
func (t *Table) AddRecord(ctx context.Context, r []interface{}) (recordID int64, err error) {
	id := variable.GetSessionVars(ctx).LastInsertID
	// Already have auto increment ID
	if id != 0 {
		recordID = int64(id)
	} else {
		recordID, err = t.alloc.Alloc(t.ID)
		if err != nil {
			return 0, err
		}
	}
	txn, err := ctx.GetTxn(false)
	if err != nil {
		return 0, err
	}
	for _, v := range t.indices {
		if v == nil {
			continue
		}
		colVals, _ := v.FetchValues(r)
		if err = v.X.Create(txn, colVals, recordID); err != nil {
			if errors2.ErrorEqual(err, kv.ErrKeyExists) {
				// Get the duplicate row handle
				iter, _, terr := v.X.Seek(txn, colVals)
				if terr != nil {
					return 0, errors.Trace(terr)
				}
				_, h, terr := iter.Next()
				if terr != nil {
					return 0, errors.Trace(terr)
				}
				return h, errors.Trace(err)
			}
			return 0, errors.Trace(err)
		}
	}

	// split a record into multiple kv pair
	// first key -> LOCK
	k := t.RecordKey(recordID, nil)
	// A new row with current txn-id as lockKey
	err = txn.Set([]byte(k), []byte(txn.String()))
	if err != nil {
		return 0, err
	}
	// column key -> column value
	for _, c := range t.Cols() {
		colKey := t.RecordKey(recordID, c)
		data, err := t.EncodeValue(r[c.Offset])
		if err != nil {
			return 0, err
		}
		err = txn.Set([]byte(colKey), data)
		if err != nil {
			return 0, err
		}
	}
	variable.GetSessionVars(ctx).AddAffectedRows(1)
	return recordID, nil
}

// IsAutocommit checks if it is in the auto-commit mode.
func (s *session) isAutocommit(ctx context.Context) bool {
	sessionVar := variable.GetSessionVars(ctx)
	autocommit := sessionVar.GetSystemVar("autocommit")
	if autocommit.IsNull() {
		if s.initing {
			return false
		}
		autocommitStr, err := s.GetGlobalSysVar(ctx, "autocommit")
		if err != nil {
			log.Errorf("Get global sys var error: %v", err)
			return false
		}
		autocommit.SetString(autocommitStr)
		err = sessionVar.SetSystemVar("autocommit", autocommit)
		if err != nil {
			log.Errorf("Set session sys var error: %v", err)
		}
	}
	autocommitStr := autocommit.GetString()
	if autocommitStr == "ON" || autocommitStr == "on" || autocommitStr == "1" {
		variable.GetSessionVars(ctx).SetStatusFlag(mysql.ServerStatusAutocommit, true)
		return true
	}
	variable.GetSessionVars(ctx).SetStatusFlag(mysql.ServerStatusAutocommit, false)
	return false
}

// IsAutocommit checks if it is in the auto-commit mode.
func (s *session) isAutocommit(ctx context.Context) (bool, error) {
	sessionVar := variable.GetSessionVars(ctx)
	autocommit := sessionVar.GetSystemVar("autocommit")
	if autocommit.IsNull() {
		if s.initing {
			return false, nil
		}
		autocommitStr, err := s.GetGlobalSysVar(ctx, "autocommit")
		if err != nil {
			return false, errors.Trace(err)
		}
		autocommit.SetString(autocommitStr)
		err = sessionVar.SetSystemVar("autocommit", autocommit)
		if err != nil {
			return false, errors.Trace(err)
		}
	}
	autocommitStr := autocommit.GetString()
	if autocommitStr == "ON" || autocommitStr == "on" || autocommitStr == "1" {
		variable.GetSessionVars(ctx).SetStatusFlag(mysql.ServerStatusAutocommit, true)
		return true, nil
	}
	variable.GetSessionVars(ctx).SetStatusFlag(mysql.ServerStatusAutocommit, false)
	return false, nil
}

// If forceNew is true, GetTxn() must return a new transaction.
// In this situation, if current transaction is still in progress,
// there will be an implicit commit and create a new transaction.
func (s *session) GetTxn(forceNew bool) (kv.Transaction, error) {
	var err error
	if s.txn == nil {
		s.resetHistory()
		s.txn, err = s.store.Begin()
		if err != nil {
			return nil, errors.Trace(err)
		}
		if !s.isAutocommit(s) {
			variable.GetSessionVars(s).SetStatusFlag(mysql.ServerStatusInTrans, true)
		}
		log.Infof("New txn:%s in session:%d", s.txn, s.sid)
	} else if forceNew {
		err = s.CommitTxn()
		if err != nil {
			return nil, errors.Trace(err)
		}
		s.txn, err = s.store.Begin()
		if err != nil {
			return nil, errors.Trace(err)
		}
		if !s.isAutocommit(s) {
			variable.GetSessionVars(s).SetStatusFlag(mysql.ServerStatusInTrans, true)
		}
		log.Warnf("Force new txn:%s in session:%d", s.txn, s.sid)
	}
	retryInfo := variable.GetSessionVars(s).RetryInfo
	if retryInfo.Retrying {
		s.txn.SetOption(kv.RetryAttempts, retryInfo.Attempts)
	}
	return s.txn, nil
}

// If forceNew is true, GetTxn() must return a new transaction.
// In this situation, if current transaction is still in progress,
// there will be an implicit commit and create a new transaction.
func (s *session) GetTxn(forceNew bool) (kv.Transaction, error) {
	var err error
	if s.txn == nil {
		s.resetHistory()
		s.txn, err = s.store.Begin()
		if err != nil {
			return nil, errors.Trace(err)
		}
		if !s.isAutocommit(s) {
			variable.GetSessionVars(s).SetStatusFlag(mysql.ServerStatusInTrans, true)
		}
		log.Infof("New txn:%s in session:%d", s.txn, s.sid)
		return s.txn, nil
	}
	if forceNew {
		err = s.FinishTxn(false)
		if err != nil {
			return nil, errors.Trace(err)
		}
		s.txn, err = s.store.Begin()
		if err != nil {
			return nil, errors.Trace(err)
		}
		if !s.isAutocommit(s) {
			variable.GetSessionVars(s).SetStatusFlag(mysql.ServerStatusInTrans, true)
		}
		log.Warnf("Force new txn:%s in session:%d", s.txn, s.sid)
	}
	return s.txn, nil
}

// IsAutocommit checks if it is in the auto-commit mode.
func (s *session) isAutocommit(ctx context.Context) bool {
	if ctx.Value(&sqlexec.RestrictedSQLExecutorKeyType{}) != nil {
		return false
	}
	autocommit, ok := variable.GetSessionVars(ctx).Systems["autocommit"]
	if !ok {
		if s.initing {
			return false
		}
		var err error
		autocommit, err = s.GetGlobalSysVar(ctx, "autocommit")
		if err != nil {
			log.Errorf("Get global sys var error: %v", err)
			return false
		}
		variable.GetSessionVars(ctx).Systems["autocommit"] = autocommit
		ok = true
	}
	if ok && (autocommit == "ON" || autocommit == "on" || autocommit == "1") {
		variable.GetSessionVars(ctx).SetStatusFlag(mysql.ServerStatusAutocommit, true)
		return true
	}
	variable.GetSessionVars(ctx).SetStatusFlag(mysql.ServerStatusAutocommit, false)
	return false
}

// execExecSelect implements `insert table select ... from ...`.
func (s *InsertValues) execSelect(t table.Table, cols []*column.Col, ctx context.Context) (rset.Recordset, error) {
	r, err := s.Sel.Plan(ctx)
	if err != nil {
		return nil, errors.Trace(err)
	}
	defer r.Close()
	if len(r.GetFields()) != len(cols) {
		return nil, errors.Errorf("Column count %d doesn't match value count %d", len(cols), len(r.GetFields()))
	}

	var bufRecords [][]interface{}
	var lastInsertIds []uint64
	for {
		var row *plan.Row
		row, err = r.Next(ctx)
		if err != nil {
			return nil, errors.Trace(err)
		}
		if row == nil {
			break
		}
		data0 := make([]interface{}, len(t.Cols()))
		marked := make(map[int]struct{}, len(cols))
		for i, d := range row.Data {
			data0[cols[i].Offset] = d
			marked[cols[i].Offset] = struct{}{}
		}

		if err = s.initDefaultValues(ctx, t, data0, marked); err != nil {
			return nil, errors.Trace(err)
		}

		if err = column.CastValues(ctx, data0, cols); err != nil {
			return nil, errors.Trace(err)
		}

		if err = column.CheckNotNull(t.Cols(), data0); err != nil {
			return nil, errors.Trace(err)
		}
		var v interface{}
		v, err = types.Clone(data0)
		if err != nil {
			return nil, errors.Trace(err)
		}

		bufRecords = append(bufRecords, v.([]interface{}))
		lastInsertIds = append(lastInsertIds, variable.GetSessionVars(ctx).LastInsertID)
	}

	for i, r := range bufRecords {
		variable.GetSessionVars(ctx).SetLastInsertID(lastInsertIds[i])
		if _, err = t.AddRecord(ctx, r); err != nil {
			return nil, errors.Trace(err)
		}
	}
	return nil, nil
}

func (s *session) Retry() error {
	variable.GetSessionVars(s).RetryInfo.Retrying = true
	nh := s.history.clone()
	// Debug infos.
	if len(nh.history) == 0 {
		s.debugInfos[retryEmptyHistoryList] = true
	} else {
		s.debugInfos[retryEmptyHistoryList] = false
	}
	defer func() {
		s.history.history = nh.history
		variable.GetSessionVars(s).RetryInfo.Retrying = false
	}()

	if forUpdate := s.Value(forupdate.ForUpdateKey); forUpdate != nil {
		return errors.Errorf("can not retry select for update statement")
	}
	var err error
	retryCnt := 0
	for {
		variable.GetSessionVars(s).RetryInfo.Attempts = retryCnt + 1
		s.resetHistory()
		log.Info("RollbackTxn for retry txn.")
		err = s.RollbackTxn()
		if err != nil {
			// TODO: handle this error.
			log.Errorf("rollback txn failed, err:%v", errors.ErrorStack(err))
		}
		success := true
		variable.GetSessionVars(s).RetryInfo.ResetOffset()
		for _, sr := range nh.history {
			st := sr.st
			log.Warnf("Retry %s", st.OriginText())
			_, err = runStmt(s, st)
			if err != nil {
				if kv.IsRetryableError(err) {
					success = false
					break
				}
				log.Warnf("session:%v, err:%v", s, err)
				return errors.Trace(err)
			}
		}
		if success {
			err = s.CommitTxn()
			if !kv.IsRetryableError(err) {
				break
			}
		}
		retryCnt++
		if (s.maxRetryCnt != unlimitedRetryCnt) && (retryCnt >= s.maxRetryCnt) {
			return errors.Trace(err)
		}
		kv.BackOff(retryCnt)
	}
	return err
}

// See http://dev.mysql.com/doc/refman/5.7/en/information-functions.html#function_last-insert-id
func builtinLastInsertID(args []types.Datum, ctx context.Context) (d types.Datum, err error) {
	if len(args) == 1 {
		id, err := args[0].ToInt64()
		if err != nil {
			return d, errors.Trace(err)
		}
		variable.GetSessionVars(ctx).SetLastInsertID(uint64(id))
	}

	d.SetUint64(variable.GetSessionVars(ctx).LastInsertID)
	return
}

// execExecSelect implements `insert table select ... from ...`.
func (s *InsertIntoStmt) execSelect(t table.Table, cols []*column.Col, ctx context.Context) (_ rset.Recordset, err error) {
	r, err := s.Sel.Plan(ctx)
	if err != nil {
		return nil, errors.Trace(err)
	} else if len(r.GetFields()) != len(cols) {
		return nil, errors.Errorf("Column count %d doesn't match value count %d", len(cols), len(r.GetFields()))
	}

	var bufRecords [][]interface{}
	var lastInsertIds []uint64
	err = r.Do(ctx, func(_ interface{}, data []interface{}) (more bool, err error) {
		data0 := make([]interface{}, len(t.Cols()))
		marked := make(map[int]struct{}, len(cols))
		for i, d := range data {
			data0[cols[i].Offset] = d
			marked[cols[i].Offset] = struct{}{}
		}

		if err = s.initDefaultValues(ctx, t, t.Cols(), data0, marked); err != nil {
			return false, errors.Trace(err)
		}

		if err = column.CastValues(ctx, data0, cols); err != nil {
			return false, errors.Trace(err)
		}

		if err = column.CheckNotNull(t.Cols(), data0); err != nil {
			return false, errors.Trace(err)
		}

		v, err := types.Clone(data0)
		if err != nil {
			return false, errors.Trace(err)
		}

		bufRecords = append(bufRecords, v.([]interface{}))
		lastInsertIds = append(lastInsertIds, variable.GetSessionVars(ctx).LastInsertID)
		return true, nil
	})
	if err != nil {
		return nil, errors.Trace(err)
	}

	for i, r := range bufRecords {
		variable.GetSessionVars(ctx).SetLastInsertID(lastInsertIds[i])

		if _, err = t.AddRecord(ctx, r); err != nil {
			return nil, errors.Trace(err)
		}
	}

	return nil, nil
}

// AddRecord implements table.Table AddRecord interface.
func (t *Table) AddRecord(ctx context.Context, r []interface{}) (recordID int64, err error) {
	id := variable.GetSessionVars(ctx).LastInsertID
	// Already have auto increment ID
	if id != 0 {
		recordID = int64(id)
	} else {
		recordID, err = t.alloc.Alloc(t.ID)
		if err != nil {
			return 0, errors.Trace(err)
		}
	}
	txn, err := ctx.GetTxn(false)
	if err != nil {
		return 0, errors.Trace(err)
	}
	for _, v := range t.indices {
		if v == nil {
			continue
		}
		colVals, _ := v.FetchValues(r)
		if err = v.X.Create(txn, colVals, recordID); err != nil {
			if terror.ErrorEqual(err, kv.ErrKeyExists) {
				// Get the duplicate row handle
				// For insert on duplicate syntax, we should update the row
				iter, _, err1 := v.X.Seek(txn, colVals)
				if err1 != nil {
					return 0, errors.Trace(err1)
				}
				_, h, err1 := iter.Next()
				if err1 != nil {
					return 0, errors.Trace(err1)
				}
				return h, errors.Trace(err)
			}
			return 0, errors.Trace(err)
		}
	}

	if err := t.LockRow(ctx, recordID); err != nil {
		return 0, errors.Trace(err)
	}

	// column key -> column value
	for _, c := range t.Cols() {
		k := t.RecordKey(recordID, c)
		if err := t.SetColValue(txn, k, r[c.Offset]); err != nil {
			return 0, errors.Trace(err)
		}
	}
	variable.GetSessionVars(ctx).AddAffectedRows(1)
	return recordID, nil
}

func (s *testStmtSuite) TestSetCharsetStmt(c *C) {
	testSQL := `SET NAMES utf8;`

	stmtList, err := tidb.Compile(testSQL)
	c.Assert(err, IsNil)
	c.Assert(stmtList, HasLen, 1)

	testStmt, ok := stmtList[0].(*stmts.SetCharsetStmt)
	c.Assert(ok, IsTrue)

	c.Assert(testStmt.IsDDL(), IsFalse)
	c.Assert(len(testStmt.OriginText()), Greater, 0)

	ctx := mock.NewContext()
	variable.BindSessionVars(ctx)
	sessionVars := variable.GetSessionVars(ctx)
	for _, v := range variable.SetNamesVariables {
		c.Assert(sessionVars.Systems[v] != "utf8", IsTrue)
	}
	_, err = testStmt.Exec(ctx)
	c.Assert(err, IsNil)
	for _, v := range variable.SetNamesVariables {
		c.Assert(sessionVars.Systems[v], Equals, "utf8")
	}
	c.Assert(sessionVars.Systems[variable.CollationConnection], Equals, "utf8_general_ci")

	mf := newMockFormatter()
	testStmt.Explain(nil, mf)
	c.Assert(mf.Len(), Greater, 0)
}

func replaceRow(ctx context.Context, t table.Table, handle int64, replaceRow []interface{}) error {
	row, err := t.Row(ctx, handle)
	if err != nil {
		return errors.Trace(err)
	}

	result := 0
	isReplace := false
	touched := make(map[int]bool, len(row))
	for i, val := range row {
		result, err = types.Compare(val, replaceRow[i])
		if err != nil {
			return errors.Trace(err)
		}
		if result != 0 {
			touched[i] = true
			isReplace = true
		}
	}

	if isReplace {
		variable.GetSessionVars(ctx).AddAffectedRows(1)
		if err = t.UpdateRecord(ctx, handle, row, replaceRow, touched); err != nil {
			return errors.Trace(err)
		}
	}

	return nil
}

// 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),
	}
	domain, err := domap.Get(store)
	if err != nil {
		return nil, err
	}
	sessionctx.BindDomain(s, domain)

	variable.BindSessionVars(s)
	variable.GetSessionVars(s).SetStatusFlag(mysql.ServerStatusAutocommit, true)
	sessionMu.Lock()
	defer sessionMu.Unlock()

	_, ok := storeBootstrapped[store.UUID()]
	if !ok {
		bootstrap(s)
		storeBootstrapped[store.UUID()] = true
	}
	// Add auth here
	return s, nil
}

// Exec implements the stmt.Statement Exec interface.
func (s *ExecuteStmt) Exec(ctx context.Context) (_ rset.Recordset, err error) {
	// Get prepared statement.
	vars := variable.GetSessionVars(ctx)
	if len(s.Name) == 0 {
		s.Name = getPreparedStmtIDKey(s.ID)
	}
	vs, ok := vars.PreparedStmts[s.Name]
	if !ok {
		return nil, errors.Errorf("Can not find prepared statement with name %s", s.Name)
	}
	ps, ok := vs.(*PreparedStmt)
	if !ok {
		return nil, errors.Errorf("Statement %s is not PreparedStmt, but %T", s.Name, vs)
	}

	// Fill param markers.
	if len(s.UsingVars) != len(ps.Params) {
		return nil, errors.Errorf("Parameter number does not match between prepared statement(%d) and execute statement(%d)", len(ps.Params), len(s.UsingVars))
	}
	for i, v := range s.UsingVars {
		ps.Params[i].Expr = v
	}
	ps.InPrepare = false

	// Run statement.
	return ps.SQLStmt.Exec(ctx)
}

func (s *InsertValues) getRow(ctx context.Context, t table.Table, cols []*column.Col, list []expression.Expression, m map[interface{}]interface{}) ([]interface{}, error) {
	r := make([]interface{}, len(t.Cols()))
	marked := make(map[int]struct{}, len(list))
	for i, expr := range list {
		// For "insert into t values (default)" Default Eval.
		m[expression.ExprEvalDefaultName] = cols[i].Name.O

		val, err := expr.Eval(ctx, m)
		if err != nil {
			return nil, errors.Trace(err)
		}
		r[cols[i].Offset] = val
		marked[cols[i].Offset] = struct{}{}
	}

	// Clear last insert id.
	variable.GetSessionVars(ctx).SetLastInsertID(0)

	err := s.initDefaultValues(ctx, t, r, marked)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if err = column.CastValues(ctx, r, cols); err != nil {
		return nil, errors.Trace(err)
	}
	if err = column.CheckNotNull(t.Cols(), r); err != nil {
		return nil, errors.Trace(err)
	}

	return r, nil
}

// Eval implements the Expression Eval interface.
func (v *Variable) Eval(ctx context.Context, args map[interface{}]interface{}) (interface{}, error) {
	name := strings.ToLower(v.Name)
	sessionVars := variable.GetSessionVars(ctx)
	if !v.IsSystem {
		// user vars
		if value, ok := sessionVars.Users[name]; ok {
			return value, nil
		}
		// select null user vars is permitted.
		return nil, nil
	}

	sysVar, ok := variable.SysVars[name]
	if !ok {
		// select null sys vars is not permitted
		return nil, errors.Errorf("Unknown system variable '%s'", name)
	}

	if !v.IsGlobal {
		if value, ok := sessionVars.Systems[name]; ok {
			return value, nil
		}
	}

	// TODO: select global sys var from kv store, now we only get it from memory
	return sysVar.Value, nil
}

func (t *testPrivilegeSuite) TestCheckDBPrivilege(c *C) {
	se := newSession(c, t.store, t.dbName)
	mustExec(c, se, `CREATE USER 'test'@'localhost' identified by '123';`)
	pc := &privileges.UserPrivileges{}
	db := &model.DBInfo{
		Name: model.NewCIStr("test"),
	}
	ctx, _ := se.(context.Context)
	variable.GetSessionVars(ctx).User = "[email protected]"
	r, err := pc.Check(ctx, db, nil, mysql.SelectPriv)
	c.Assert(err, IsNil)
	c.Assert(r, IsFalse)

	mustExec(c, se, `GRANT SELECT ON *.* TO  'test'@'localhost';`)
	pc = &privileges.UserPrivileges{}
	r, err = pc.Check(ctx, db, nil, mysql.SelectPriv)
	c.Assert(err, IsNil)
	c.Assert(r, IsTrue)
	r, err = pc.Check(ctx, db, nil, mysql.UpdatePriv)
	c.Assert(err, IsNil)
	c.Assert(r, IsFalse)

	mustExec(c, se, `GRANT Update ON test.* TO  'test'@'localhost';`)
	pc = &privileges.UserPrivileges{}
	r, err = pc.Check(ctx, db, nil, mysql.UpdatePriv)
	c.Assert(err, IsNil)
	c.Assert(r, IsTrue)
}

// 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) (ast.Statement, error) {
	ast.SetFlag(node)
	if _, ok := node.(*ast.UpdateStmt); ok {
		sVars := variable.GetSessionVars(ctx)
		sVars.InUpdateStmt = true
		defer func() {
			sVars.InUpdateStmt = false
		}()
	}

	is := sessionctx.GetDomain(ctx).InfoSchema()
	if err := plan.Preprocess(node, is, ctx); err != nil {
		return nil, errors.Trace(err)
	}
	// Validate should be after NameResolve.
	if err := plan.Validate(node, false); err != nil {
		return nil, errors.Trace(err)
	}
	sb := NewSubQueryBuilder(is)

	p, err := plan.Optimize(ctx, node, sb, is)
	if err != nil {
		return nil, errors.Trace(err)
	}
	_, isDDL := node.(ast.DDLNode)
	sa := &statement{
		is:    is,
		plan:  p,
		text:  node.Text(),
		isDDL: isDDL,
	}
	return sa, nil
}

func runStmt(ctx context.Context, s ast.Statement, args ...interface{}) (ast.RecordSet, error) {
	var err error
	var rs ast.RecordSet
	// before every execution, we must clear affectedrows.
	variable.GetSessionVars(ctx).SetAffectedRows(0)
	if s.IsDDL() {
		err = ctx.CommitTxn()
		if err != nil {
			return nil, errors.Trace(err)
		}
	}
	rs, err = s.Exec(ctx)
	// All the history should be added here.
	se := ctx.(*session)
	se.history.add(0, s)
	// MySQL DDL should be auto-commit.
	ac, err1 := autocommit.ShouldAutocommit(ctx)
	if err1 != nil {
		return nil, errors.Trace(err1)
	}
	if s.IsDDL() || ac {
		if err != nil {
			log.Info("RollbackTxn for ddl/autocommit error.")
			ctx.RollbackTxn()
		} else {
			err = ctx.CommitTxn()
		}
	}
	return rs, errors.Trace(err)
}