func (r *DistinctDefaultPlan) fetchAll(ctx context.Context) error {
	t, err := memkv.CreateTemp(true)
	if err != nil {
		return errors.Trace(err)
	}
	defer func() {
		if derr := t.Drop(); derr != nil && err == nil {
			err = derr
		}
	}()
	for {
		row, err := r.Src.Next(ctx)
		if row == nil || err != nil {
			return errors.Trace(err)
		}
		var v []interface{}
		// get distinct key
		key := row.Data[0:r.HiddenFieldOffset]
		v, err = t.Get(key)
		if err != nil {
			return errors.Trace(err)
		}

		if len(v) == 0 {
			// no group for key, save data for this group
			r.rows = append(r.rows, row)
			if err := t.Set(key, []interface{}{true}); err != nil {
				return errors.Trace(err)
			}
		}
	}
}

func (a *AggregateDistinct) clear() {
	if a.Distinct == nil {
		return
	}

	// drop does nothing, no need to check error
	a.Distinct.Drop()
	// CreateTemp returns no error, no need to check error
	// later we may use another better way instead of memkv
	a.Distinct, _ = memkv.CreateTemp(true)
}

// CreateAggregateDistinct creates a distinct for function f.
func CreateAggregateDistinct(f string, distinct bool) *AggregateDistinct {
	a := &AggregateDistinct{}

	switch strings.ToLower(f) {
	case "count", "sum", "avg", "group_concat":
		// only these aggregate functions support distinct
		if distinct {
			a.Distinct, _ = memkv.CreateTemp(true)
		}
	}
	return a
}

func (c *Call) createDistinct() *aggregateDistinct {
	a := new(aggregateDistinct)

	switch strings.ToLower(c.F) {
	case "count", "sum", "avg", "group_concat":
		// only these aggregate functions support distinct
		if c.Distinct {
			a.Distinct, _ = memkv.CreateTemp(true)
		}
	}

	return a
}

func (p *UnionPlan) fetchAll(ctx context.Context) error {
	if len(p.Srcs) == 0 {
		return nil
	}
	t, err := memkv.CreateTemp(true)
	if err != nil {
		return err
	}

	defer func() {
		if derr := t.Drop(); derr != nil && err == nil {
			err = derr
		}
	}()

	// Fetch results of the first select statement.
	src := p.Srcs[0]
	for {
		var row *plan.Row
		row, err = src.Next(ctx)
		if err != nil {
			return errors.Trace(err)
		}
		if row == nil {
			break
		}
		p.rows = append(p.rows, row)
		err = t.Set(row.Data, []interface{}{true})
		if err != nil {
			return errors.Trace(err)
		}
	}

	// Update return result types
	// e,g, for select 'abc' union select 'a', we can only get result type after
	// executing first select.
	for i, f := range p.RFields {
		if f.Col.FieldType.Tp == 0 {
			f.Col.FieldType = src.GetFields()[i].Col.FieldType
		}
	}

	// Fetch results of the following select statements.
	for i := range p.Distincts {
		err = p.fetchSrc(ctx, i, t)
		if err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

// Do : Distinct plan use an in-memory temp table for storing items that has same
// key, the value in temp table is an array of record handles.
func (r *DistinctDefaultPlan) Do(ctx context.Context, f plan.RowIterFunc) (err error) {
	t, err := memkv.CreateTemp(true)
	if err != nil {
		return
	}

	defer func() {
		if derr := t.Drop(); derr != nil && err == nil {
			err = derr
		}
	}()

	var rows [][]interface{}
	if err = r.Src.Do(ctx, func(id interface{}, in []interface{}) (bool, error) {
		var v []interface{}
		// get distinct key
		key := in[0:r.HiddenFieldOffset]
		v, err = t.Get(key)
		if err != nil {
			return false, err
		}

		if len(v) == 0 {
			// no group for key, save data for this group
			rows = append(rows, in)
			if err := t.Set(key, []interface{}{true}); err != nil {
				return false, err
			}
		}

		return true, nil
	}); err != nil {
		return
	}

	var more bool
	for _, row := range rows {
		if more, err = f(nil, row); !more || err != nil {
			break
		}
	}
	return types.EOFAsNil(err)
}

func (p *UnionPlan) fetchAll(ctx context.Context) error {
	if len(p.Srcs) == 0 {
		return nil
	}
	t, err := memkv.CreateTemp(true)
	if err != nil {
		return err
	}

	defer func() {
		if derr := t.Drop(); derr != nil && err == nil {
			err = derr
		}
	}()

	// Fetch results of the first select statement.
	src := p.Srcs[0]
	for {
		var row *plan.Row
		row, err = src.Next(ctx)
		if err != nil {
			return errors.Trace(err)
		}
		if row == nil {
			break
		}
		p.rows = append(p.rows, row)
		err = t.Set(row.Data, []interface{}{true})
		if err != nil {
			return errors.Trace(err)
		}
	}

	// Fetch results of the following select statements.
	for i := range p.Distincts {
		err = p.fetchSrc(ctx, i, t)
		if err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

// Do implements plan.Plan Do interface.
// Table: Subject_Selection
// Subject   Semester   Attendee
// ---------------------------------
// ITB001    1          John
// ITB001    1          Bob
// ITB001    1          Mickey
// ITB001    2          Jenny
// ITB001    2          James
// MKB114    1          John
// MKB114    1          Erica
// refs: http://stackoverflow.com/questions/2421388/using-group-by-on-multiple-columns
func (r *GroupByDefaultPlan) Do(ctx context.Context, f plan.RowIterFunc) (err error) {
	// TODO: now we have to use this to save group key -> row index
	// later we will serialize group by items into a string key and then use a map instead.
	t, err := memkv.CreateTemp(true)
	if err != nil {
		return err
	}

	defer func() {
		if derr := t.Drop(); derr != nil && err == nil {
			err = derr
		}
	}()

	k := make([]interface{}, len(r.By))

	// save output group by result
	var outRows []*groupRow

	err = r.Src.Do(ctx, func(rid interface{}, in []interface{}) (more bool, err error) {
		out := make([]interface{}, len(r.Fields))

		// TODO: later order by will use the same mechanism, so we may use another plan to do this
		m := map[interface{}]interface{}{}
		// must first eval none aggregate fields, because alias group by will use this.
		if err := r.evalNoneAggFields(ctx, out, m, in); err != nil {
			return false, err
		}

		if err := r.evalGroupKey(ctx, k, out, in); err != nil {
			return false, err
		}

		// get row index with k.
		v, err := t.Get(k)
		if err != nil {
			return false, err
		}

		index := 0
		if len(v) == 0 {
			// no group for key, save data for this group
			index = len(outRows)
			outRows = append(outRows, &groupRow{Row: out, Args: m})

			if err := t.Set(k, []interface{}{index}); err != nil {
				return false, err
			}
		} else {
			// we have already saved data in the group by key, use this
			index = v[0].(int)

			// we will use this context args to evaluate aggregate fields.
			m = outRows[index].Args
		}

		// eval aggregate fields
		if err := r.evalAggFields(ctx, out, m, in); err != nil {
			return false, err
		}

		return true, nil
	})

	if err != nil {
		return err
	}

	if len(outRows) == 0 {
		// empty table
		var out []interface{}
		out, err = r.evalEmptyTable(ctx)
		if err != nil || out == nil {
			return err
		}

		_, err = f(nil, out)

		return err
	}

	// we don't consider implicit GROUP BY sorting now.
	// Relying on implicit GROUP BY sorting in MySQL 5.7 is deprecated.
	// To achieve a specific sort order of grouped results,
	// it is preferable to use an explicit ORDER BY clause.
	// GROUP BY sorting is a MySQL extension that may change in a future release
	var more bool
	for _, row := range outRows {
//.........这里部分代码省略.........

func (r *GroupByDefaultPlan) fetchAll(ctx context.Context) error {
	// TODO: now we have to use this to save group key -> row index
	// later we will serialize group by items into a string key and then use a map instead.
	t, err := memkv.CreateTemp(true)
	if err != nil {
		return errors.Trace(err)
	}

	defer func() {
		if derr := t.Drop(); derr != nil && err == nil {
			err = derr
		}
	}()
	k := make([]interface{}, len(r.By))
	for {
		srcRow, err1 := r.Src.Next(ctx)
		if err1 != nil {
			return errors.Trace(err1)
		}
		if srcRow == nil {
			break
		}
		row := &plan.Row{
			Data: make([]interface{}, len(r.Fields)),
			// must save FromData info for inner sub query use.
			FromData: srcRow.Data,
		}
		// TODO: later order by will use the same mechanism, so we may use another plan to do this
		evalArgs := map[interface{}]interface{}{}
		// must first eval none aggregate fields, because alias group by will use this.
		if err := r.evalNoneAggFields(ctx, row.Data, evalArgs, srcRow.Data); err != nil {
			return errors.Trace(err)
		}

		// update outer query becuase group by may use it if it has a subquery.
		updateRowStack(ctx, row.Data, row.FromData)

		if err := r.evalGroupKey(ctx, k, row.Data, srcRow.Data); err != nil {
			return errors.Trace(err)
		}

		// get row index with k.
		v, err1 := t.Get(k)
		if err1 != nil {
			return errors.Trace(err1)
		}

		index := 0
		if len(v) == 0 {
			// no group for key, save data for this group
			index = len(r.rows)
			r.rows = append(r.rows, &groupRow{Row: row, Args: evalArgs})

			if err := t.Set(k, []interface{}{index}); err != nil {
				return errors.Trace(err)
			}
		} else {
			// we have already saved data in the group by key, use this
			index = v[0].(int)

			// we will use this context args to evaluate aggregate fields.
			evalArgs = r.rows[index].Args
		}

		// eval aggregate fields
		if err := r.evalAggFields(ctx, row.Data, evalArgs, srcRow.Data); err != nil {
			return errors.Trace(err)
		}
	}
	if len(r.rows) == 0 {
		// empty table
		var out []interface{}
		out, err = r.evalEmptyTable(ctx)
		if err != nil || out == nil {
			return errors.Trace(err)
		}
		row := &plan.Row{Data: out}
		r.rows = append(r.rows, &groupRow{Row: row, Args: map[interface{}]interface{}{}})
	} else {
		for _, row := range r.rows {
			// eval aggregate done
			if err := r.evalAggDone(ctx, row.Row.Data, row.Args); err != nil {
				return errors.Trace(err)
			}
		}
	}
	return nil
}

// Do implements plan.Plan Do interface.
// In union plan, we should evaluate each selects from left to right, and union them according to distinct option
// Use the first select ResultFields as the final ResultFields but we should adjust its Flen for the rest Select ResultFields.
func (p *UnionPlan) Do(ctx context.Context, f plan.RowIterFunc) error {
	if len(p.Srcs) == 0 {
		return nil
	}
	t, err := memkv.CreateTemp(true)
	if err != nil {
		return err
	}

	defer func() {
		if derr := t.Drop(); derr != nil && err == nil {
			err = derr
		}
	}()
	// The final result
	var rows [][]interface{}

	// Eval the first select statement
	src := p.Srcs[0]
	src.Do(ctx, func(rid interface{}, in []interface{}) (bool, error) {
		rows = append(rows, in)
		if err := t.Set(in, []interface{}{true}); err != nil {
			return false, err
		}
		return true, nil
	})
	// Use the ResultFields of the first select statement as the final ResultFields
	rfs := src.GetFields()

	// Eval the following select statements
	for i, distinct := range p.Distincts {
		src = p.Srcs[i+1]
		// Eval src
		if len(src.GetFields()) != len(rfs) {
			return errors.New("The used SELECT statements have a different number of columns")
		}
		src.Do(ctx, func(rid interface{}, in []interface{}) (bool, error) {
			srcRfs := src.GetFields()
			for i := range in {
				// The column value should be casted as the same type of the first select statement in corresponding position
				srcRf := srcRfs[i]
				rf := rfs[i]
				/*
				 * The lengths of the columns in the UNION result take into account the values retrieved by all of the SELECT statements
				 * SELECT REPEAT('a',1) UNION SELECT REPEAT('b',10);
				 * +---------------+
				 * | REPEAT('a',1) |
				 * +---------------+
				 * | a             |
				 * | bbbbbbbbbb    |
				 * +---------------+
				 */
				if srcRf.Flen > rf.Col.Flen {
					rf.Col.Flen = srcRf.Col.Flen
				}
				in[i], err = rf.Col.CastValue(ctx, in[i])
				if err != nil {
					return false, err
				}
			}
			if distinct {
				// distinct union, check duplicate
				v, getErr := t.Get(in)
				if getErr != nil {
					return false, getErr
				}
				if len(v) > 0 {
					// Find duplicate, ignore it
					return true, nil
				}
			}
			rows = append(rows, in)
			if err := t.Set(in, []interface{}{true}); err != nil {
				return false, err
			}
			return true, nil
		})
	}
	var more bool
	for _, row := range rows {
		if more, err = f(nil, row); !more || err != nil {
			break
		}
	}
	return nil
}