func (s *testAggFuncSuite) TestCount(c *C) {
	// Compose aggregate exec for "select c1, count(c2) from t";
	// c1  c2
	// 1	1
	// 2	1
	// 3    nil
	c1 := ast.NewValueExpr(0)
	rf1 := &ast.ResultField{Expr: c1}
	col1 := &ast.ColumnNameExpr{Refer: rf1}
	fc1 := &ast.AggregateFuncExpr{
		F:    ast.AggFuncFirstRow,
		Args: []ast.ExprNode{col1},
	}
	c2 := ast.NewValueExpr(0)
	rf2 := &ast.ResultField{Expr: c2}
	col2 := &ast.ColumnNameExpr{Refer: rf2}
	fc2 := &ast.AggregateFuncExpr{
		F:    ast.AggFuncCount,
		Args: []ast.ExprNode{col2},
	}
	row1 := []interface{}{1, 1}
	row2 := []interface{}{2, 1}
	row3 := []interface{}{3, nil}
	data := []([]interface{}){row1, row2, row3}

	rows := make([]*Row, 0, 3)
	for _, d := range data {
		rows = append(rows, &Row{Data: d})
	}
	src := &mockExec{
		rows:   rows,
		fields: []*ast.ResultField{rf1, rf2},
	}
	agg := &AggregateExec{
		AggFuncs: []*ast.AggregateFuncExpr{fc1, fc2},
		Src:      src,
	}
	var (
		row *Row
		cnt int
	)
	for {
		r, err := agg.Next()
		c.Assert(err, IsNil)
		if r == nil {
			break
		}
		row = r
		cnt++
	}
	c.Assert(cnt, Equals, 1)
	c.Assert(row, NotNil)
	ctx := mock.NewContext()
	val, err := evaluator.Eval(ctx, fc1)
	c.Assert(err, IsNil)
	c.Assert(val, Equals, 1)
	val, err = evaluator.Eval(ctx, fc2)
	c.Assert(err, IsNil)
	c.Assert(val, Equals, int64(2))
}

// Next implements Executor Next interface.
func (e *SortExec) Next() (*Row, error) {
	if !e.fetched {
		for {
			srcRow, err := e.Src.Next()
			if err != nil {
				return nil, errors.Trace(err)
			}
			if srcRow == nil {
				break
			}
			orderRow := &orderByRow{
				row: srcRow,
				key: make([]interface{}, len(e.ByItems)),
			}
			for i, byItem := range e.ByItems {
				orderRow.key[i], err = evaluator.Eval(e.ctx, byItem.Expr)
				if err != nil {
					return nil, errors.Trace(err)
				}
			}
			e.Rows = append(e.Rows, orderRow)
		}
		sort.Sort(e)
		e.fetched = true
	}
	if e.err != nil {
		return nil, errors.Trace(e.err)
	}
	if e.Idx >= len(e.Rows) {
		return nil, nil
	}
	row := e.Rows[e.Idx].row
	e.Idx++
	return row, nil
}

func (e *UpdateExec) fetchRows() error {
	for {
		row, err := e.SelectExec.Next()
		if err != nil {
			return errors.Trace(err)
		}
		if row == nil {
			return nil
		}
		data := make([]interface{}, len(e.SelectExec.Fields()))
		newData := make([]interface{}, len(e.SelectExec.Fields()))
		for i, f := range e.SelectExec.Fields() {
			data[i] = f.Expr.GetValue()
			newData[i] = data[i]
			if e.OrderedList[i] != nil {
				val, err := evaluator.Eval(e.ctx, e.OrderedList[i].Expr)
				if err != nil {
					return errors.Trace(err)
				}
				newData[i] = val
			}
		}
		row.Data = data
		e.rows = append(e.rows, row)
		e.newRowsData = append(e.newRowsData, newData)
	}
}

// Fetch a single row from src and update each aggregate function.
// If the first return value is false, it means there is no more data from src.
func (e *AggregateExec) innerNext() (bool, error) {
	if e.Src != nil {
		srcRow, err := e.Src.Next()
		if err != nil {
			return false, errors.Trace(err)
		}
		if srcRow == nil {
			return false, nil
		}
	} else {
		// If Src is nil, only one row should be returned.
		if e.executed {
			return false, nil
		}
	}
	e.executed = true
	// TODO: evaluate groupby key and set currentgroup to each aggfunc.
	groupKey := singleGroup
	e.groupMap[groupKey] = true
	for _, af := range e.AggFuncs {
		for _, arg := range af.Args {
			_, err := evaluator.Eval(e.ctx, arg)
			if err != nil {
				return false, errors.Trace(err)
			}
		}
		af.CurrentGroup = groupKey
		af.Update()
	}
	return true, nil
}

// Build builds a prepared statement into an executor.
func (e *ExecuteExec) Build() error {
	vars := variable.GetSessionVars(e.Ctx)
	if e.Name != "" {
		e.ID = vars.PreparedStmtNameToID[e.Name]
	}
	v := vars.PreparedStmts[e.ID]
	if v == nil {
		return ErrStmtNotFound
	}
	prepared := v.(*Prepared)

	if len(prepared.Params) != len(e.UsingVars) {
		return ErrWrongParamCount
	}

	for i, usingVar := range e.UsingVars {
		val, err := evaluator.Eval(e.Ctx, usingVar)
		if err != nil {
			return errors.Trace(err)
		}
		prepared.Params[i].SetValue(val)
	}

	if optimizer.IsSupported(prepared.Stmt) {
		if prepared.SchemaVersion != e.IS.SchemaMetaVersion() {
			// If the schema version has changed we need to prepare it again,
			// if this time it failed, the real reason for the error is schema changed.
			err := optimizer.Prepare(e.IS, e.Ctx, prepared.Stmt)
			if err != nil {
				return ErrSchemaChanged.Gen("Schema change casued error: %s", err.Error())
			}
			prepared.SchemaVersion = e.IS.SchemaMetaVersion()
		}
		sb := &subqueryBuilder{is: e.IS}
		plan, err := optimizer.Optimize(e.Ctx, prepared.Stmt, sb)
		if err != nil {
			return errors.Trace(err)
		}
		b := newExecutorBuilder(e.Ctx, e.IS)
		stmtExec := b.build(plan)
		if b.err != nil {
			return errors.Trace(b.err)
		}
		e.StmtExec = stmtExec
	} else {
		conv := converter.Converter{}
		oStmt, err := conv.Convert(prepared.Stmt)
		if err != nil {
			return errors.Trace(err)
		}
		e.OldStmt = oStmt
	}
	return nil
}

func (e *AggregateExec) getGroupKey() (string, error) {
	if len(e.GroupByItems) == 0 {
		return singleGroup, nil
	}
	vals := make([]interface{}, 0, len(e.GroupByItems))
	for _, item := range e.GroupByItems {
		v, err := evaluator.Eval(e.ctx, item.Expr)
		if err != nil {
			return "", errors.Trace(err)
		}
		vals = append(vals, v)
	}
	bs, err := codec.EncodeValue([]byte{}, vals...)
	if err != nil {
		return "", errors.Trace(err)
	}
	return string(bs), nil
}

func (r *preEvaluator) Leave(in ast.Node) (ast.Node, bool) {
	if expr, ok := in.(ast.ExprNode); ok {
		if _, ok = expr.(*ast.ValueExpr); ok {
			return in, true
		} else if ast.IsPreEvaluable(expr) {
			val, err := evaluator.Eval(r.ctx, expr)
			if err != nil {
				r.err = err
				return in, false
			}
			if ast.IsConstant(expr) {
				// The expression is constant, rewrite the expression to value expression.
				valExpr := &ast.ValueExpr{}
				valExpr.SetText(expr.Text())
				valExpr.SetType(expr.GetType())
				valExpr.SetValue(val)
				return valExpr, true
			}
			expr.SetValue(val)
		}
	}
	return in, true
}

// Fetch a single row from src and update each aggregate function.
// If the first return value is false, it means there is no more data from src.
func (e *AggregateExec) innerNext() (bool, error) {
	if e.Src != nil {
		srcRow, err := e.Src.Next()
		if err != nil {
			return false, errors.Trace(err)
		}
		if srcRow == nil {
			return false, nil
		}
	} else {
		// If Src is nil, only one row should be returned.
		if e.executed {
			return false, nil
		}
	}
	e.executed = true
	groupKey, err := e.getGroupKey()
	if err != nil {
		return false, errors.Trace(err)
	}
	if _, ok := e.groupMap[groupKey]; !ok {
		e.groupMap[groupKey] = true
		e.groups = append(e.groups, groupKey)
	}
	for _, af := range e.AggFuncs {
		for _, arg := range af.Args {
			_, err := evaluator.Eval(e.ctx, arg)
			if err != nil {
				return false, errors.Trace(err)
			}
		}
		af.CurrentGroup = groupKey
		af.Update()
	}
	return true, nil
}