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.FinishTxn(false)
		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
	if s.IsDDL() || autocommit.ShouldAutocommit(ctx) {
		if err != nil {
			ctx.FinishTxn(true)
		} else {
			err = ctx.FinishTxn(false)
		}
	}
	return rs, errors.Trace(err)
}

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

func runStmt(ctx context.Context, s stmt.Statement, args ...interface{}) (rset.Recordset, error) {
	var err error
	var rs rset.Recordset
	// before every execution, we must clear affectedrows.
	variable.GetSessionVars(ctx).SetAffectedRows(0)
	switch s.(type) {
	case *stmts.PreparedStmt:
		ps := s.(*stmts.PreparedStmt)
		return runPreparedStmt(ctx, ps)
	case *stmts.ExecuteStmt:
		es := s.(*stmts.ExecuteStmt)
		rs, err = runExecute(ctx, es, args...)
		if err != nil {
			return nil, errors.Trace(err)
		}
	default:
		if s.IsDDL() {
			err = ctx.FinishTxn(false)
			if err != nil {
				return nil, errors.Trace(err)
			}
		}
		stmt.BindExecArgs(ctx, args)
		rs, err = s.Exec(ctx)
		stmt.ClearExecArgs(ctx)
	}
	// MySQL DDL should be auto-commit
	if err == nil && (s.IsDDL() || autocommit.ShouldAutocommit(ctx)) {
		err = ctx.FinishTxn(false)
	}
	return rs, errors.Trace(err)
}

func (b *executorBuilder) buildSelectLock(v *plan.SelectLock) Executor {
	src := b.build(v.GetChildByIndex(0))
	if autocommit.ShouldAutocommit(b.ctx) {
		// Locking of rows for update using SELECT FOR UPDATE only applies when autocommit
		// is disabled (either by beginning transaction with START TRANSACTION or by setting
		// autocommit to 0. If autocommit is enabled, the rows matching the specification are not locked.
		// See: https://dev.mysql.com/doc/refman/5.7/en/innodb-locking-reads.html
		return src
	}
	e := &SelectLockExec{
		Src:  src,
		Lock: v.Lock,
		ctx:  b.ctx,
	}
	return e
}

func runStmt(ctx context.Context, s stmt.Statement, args ...interface{}) (rset.Recordset, error) {
	var err error
	var rs rset.Recordset
	// before every execution, we must clear affectedrows.
	variable.GetSessionVars(ctx).SetAffectedRows(0)
	switch ts := s.(type) {
	case *stmts.PreparedStmt:
		rs, err = runPreparedStmt(ctx, ts)
	case *stmts.ExecuteStmt:
		rs, err = runExecute(ctx, ts, args...)
	default:
		if s.IsDDL() {
			err = ctx.FinishTxn(false)
			if err != nil {
				return nil, errors.Trace(err)
			}
		}
		stmt.BindExecArgs(ctx, args)
		rs, err = s.Exec(ctx)
		stmt.ClearExecArgs(ctx)
	}
	// All the history should be added here.
	se := ctx.(*session)
	switch ts := s.(type) {
	case *stmts.PreparedStmt:
		se.history.add(ts.ID, s)
	case *stmts.ExecuteStmt:
		se.history.add(ts.ID, s, args...)
	default:
		se.history.add(0, s)
	}
	// MySQL DDL should be auto-commit
	if s.IsDDL() || autocommit.ShouldAutocommit(ctx) {
		if err != nil {
			ctx.FinishTxn(true)
		} else {
			err = ctx.FinishTxn(false)
		}
	}
	return rs, errors.Trace(err)
}

// Plan implements the plan.Planner interface.
// The whole phase for select is
// `from -> where -> lock -> group by -> having -> select fields -> distinct -> order by -> limit -> final`
func (s *SelectStmt) Plan(ctx context.Context) (plan.Plan, error) {
	var (
		r   plan.Plan
		err error
	)

	if s.From != nil {
		r, err = s.From.Plan(ctx)
		if err != nil {
			return nil, err
		}
	} else if s.Fields != nil {
		// Only evaluate fields values.
		fr := &rsets.SelectFromDualRset{Fields: s.Fields}
		r, err = fr.Plan(ctx)
		if err != nil {
			return nil, err
		}

	}

	// Put RowStackFromPlan here so that we can catch the origin from data after above FROM phase.
	r = &plans.RowStackFromPlan{Src: r}

	if w := s.Where; w != nil {
		r, err = (&rsets.WhereRset{Expr: w.Expr, Src: r}).Plan(ctx)
		if err != nil {
			return nil, err
		}
	}
	lock := s.Lock
	if lock != coldef.SelectLockNone && autocommit.ShouldAutocommit(ctx) {
		// Locking of rows for update using SELECT FOR UPDATE only applies when autocommit
		// is disabled (either by beginning transaction with START TRANSACTION or by setting
		// autocommit to 0. If autocommit is enabled, the rows matching the specification are not locked.
		// See: https://dev.mysql.com/doc/refman/5.7/en/innodb-locking-reads.html
		lock = coldef.SelectLockNone
	}
	r = &plans.SelectLockPlan{Src: r, Lock: lock}

	if err = s.checkOneColumn(ctx); err != nil {
		return nil, errors.Trace(err)
	}

	// Get select list for futher field values evaluation.
	var selectList *plans.SelectList
	if s.selectList == nil {
		selectList, err = plans.ResolveSelectList(s.Fields, r.GetFields())
	} else {
		selectList = s.selectList
	}
	if err != nil {
		return nil, errors.Trace(err)
	}

	var groupBy []expression.Expression
	if s.GroupBy != nil {
		groupBy = s.GroupBy.By
	}

	if s.Having != nil && s.selectList == nil {
		// `having` may contain aggregate functions, and we will add this to hidden fields.
		if err = s.Having.CheckAggregate(selectList); err != nil {
			return nil, errors.Trace(err)
		}
	}

	if s.OrderBy != nil && s.selectList == nil {
		// `order by` may contain aggregate functions, and we will add this to hidden fields.
		if err = s.OrderBy.CheckAggregate(selectList); err != nil {
			return nil, errors.Trace(err)
		}
	}

	if s.selectList == nil {
		s.selectList = selectList
	}

	switch {
	case len(selectList.AggFields) == 0 && s.GroupBy == nil:
		// If no group by and no aggregate functions, we will use SelectFieldsPlan.
		if r, err = (&rsets.SelectFieldsRset{Src: r,
			SelectList: selectList,
		}).Plan(ctx); err != nil {
			return nil, err
		}

	default:
		if r, err = (&rsets.GroupByRset{By: groupBy,
			Src:        r,
			SelectList: selectList,
		}).Plan(ctx); err != nil {
			return nil, err
		}
	}

	if s := s.Having; s != nil {
//.........这里部分代码省略.........