// Next returns the next row of the sub result.
// If no more row to return, data would be nil.
func (pr *partialResult) Next() (handle int64, data []types.Datum, err error) {
	if !pr.fetched {
		select {
		case err = <-pr.done:
		}
		pr.fetched = true
		if err != nil {
			return 0, nil, err
		}
	}
	if pr.cursor >= len(pr.resp.Rows) {
		return 0, nil, nil
	}
	row := pr.resp.Rows[pr.cursor]
	data, err = tablecodec.DecodeValues(row.Data, pr.fields, pr.index)
	if err != nil {
		return 0, nil, errors.Trace(err)
	}
	if data == nil {
		// When no column is referenced, the data may be nil, like 'select count(*) from t'.
		// In this case, we need to create a zero length datum slice,
		// as caller will check if data is nil to finish iteration.
		data = make([]types.Datum, 0)
	}
	if !pr.aggregate {
		handleBytes := row.GetHandle()
		datums, err := codec.Decode(handleBytes)
		if err != nil {
			return 0, nil, errors.Trace(err)
		}
		handle = datums[0].GetInt64()
	}
	pr.cursor++
	return
}

// Next returns current key and moves iterator to the next step.
func (c *indexIter) Next() (val []interface{}, h int64, err error) {
	if !c.it.Valid() {
		return nil, 0, errors.Trace(io.EOF)
	}
	if !c.it.Key().HasPrefix(c.prefix) {
		return nil, 0, errors.Trace(io.EOF)
	}
	// get indexedValues
	buf := c.it.Key()[len(c.prefix):]
	vv, err := codec.Decode(buf)
	if err != nil {
		return nil, 0, errors.Trace(err)
	}
	// if index is *not* unique, the handle is in keybuf
	if !c.idx.unique {
		h = vv[len(vv)-1].(int64)
		val = vv[0 : len(vv)-1]
	} else {
		// otherwise handle is value
		h, err = decodeHandle(c.it.Value())
		if err != nil {
			return nil, 0, errors.Trace(err)
		}
		val = vv
	}
	// update new iter to next
	err = c.it.Next()
	if err != nil {
		return nil, 0, errors.Trace(err)
	}
	return
}

func (e *Evaluator) decodeValueList(valueListExpr *tipb.Expr) (*decodedValueList, error) {
	if len(valueListExpr.Val) == 0 {
		// Empty value list.
		return &decodedValueList{}, nil
	}
	if e.valueLists == nil {
		e.valueLists = make(map[*tipb.Expr]*decodedValueList)
	}
	decoded := e.valueLists[valueListExpr]
	if decoded != nil {
		return decoded, nil
	}
	list, err := codec.Decode(valueListExpr.Val)
	if err != nil {
		return nil, errors.Trace(err)
	}
	var hasNull bool
	for _, v := range list {
		if v.IsNull() {
			hasNull = true
		}
	}
	decoded = &decodedValueList{values: list, hasNull: hasNull}
	e.valueLists[valueListExpr] = decoded
	return decoded, nil
}

// DecodeValue implements table.Table DecodeValue interface.
func DecodeValue(data []byte, tp *types.FieldType) (types.Datum, error) {
	values, err := codec.Decode(data)
	if err != nil {
		return types.Datum{}, errors.Trace(err)
	}
	return unflatten(values[0], tp)
}

// DecodeRow decodes a byte slice into datums.
// TODO: We should only decode columns in the cols map.
// Row layout: colID1, value1, colID2, value2, .....
func DecodeRow(data []byte, cols map[int64]*types.FieldType) (map[int64]types.Datum, error) {
	if data == nil {
		return nil, nil
	}
	values, err := codec.Decode(data)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if len(values)%2 != 0 {
		return nil, errors.New("Decoded row value length is not even number!")
	}
	row := make(map[int64]types.Datum, len(cols))
	for i := 0; i < len(values); i += 2 {
		cid := values[i]
		id := cid.GetInt64()
		ft, ok := cols[id]
		if ok {
			v := values[i+1]
			v, err = Unflatten(v, ft)
			if err != nil {
				return nil, errors.Trace(err)
			}
			row[id] = v
		}
	}
	return row, nil
}

func decodeValue(data []byte, cols []*model.ColumnInfo) ([]interface{}, error) {
	values, err := codec.Decode(data)
	if err != nil {
		return nil, errors.Trace(err)
	}

	if len(values) != len(cols) {
		return nil, errors.Errorf("Column count does not match, expect %d, actual %d", len(cols), len(values))
	}

	var rvalues []interface{}
	for i, col := range cols {
		// TODO: support more types if we really need.
		switch col.Tp {
		case mysql.TypeString, mysql.TypeVarchar:
			val := string(values[i].([]byte))
			rvalues = append(rvalues, val)
		case mysql.TypeEnum:
			val, err := mysql.ParseEnumValue(col.Elems, values[i].(uint64))
			if err != nil {
				return nil, errors.Trace(err)
			}
			rvalues = append(rvalues, val.String())
		}
	}
	return rvalues, nil
}

// DecodeValue implements table.Table DecodeValue interface.
func (t *Table) DecodeValue(data []byte, col *column.Col) (interface{}, error) {
	values, err := codec.Decode(data)
	if err != nil {
		return nil, errors.Trace(err)
	}

	return t.unflatten(values[0], col)
}

// DecodeValues decodes a byte slice into datums with column types.
func DecodeValues(data []byte, fts []*types.FieldType) ([]types.Datum, error) {
	values, err := codec.Decode(data)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if len(values) > len(fts) {
		return nil, errors.Errorf("invalid column count %d is less than value count %d", len(fts), len(values))
	}
	for i := range values {
		values[i], err = unflatten(values[i], fts[i])
		if err != nil {
			return nil, errors.Trace(err)
		}
	}
	return values, nil
}

func mutationRowsToRows(c *C, mutationRows [][]byte, firstColumn, secondColumn int) [][]types.Datum {
	var rows [][]types.Datum
	for _, mutationRow := range mutationRows {
		datums, err := codec.Decode(mutationRow, 5)
		c.Assert(err, IsNil)
		for i := range datums {
			if i != firstColumn && i != secondColumn {
				// Column ID or handle
				c.Assert(datums[i].GetInt64(), Greater, int64(0))
			}
			if datums[i].Kind() == types.KindBytes {
				datums[i].SetBytesAsString(datums[i].GetBytes())
			}
		}
		row := []types.Datum{datums[firstColumn], datums[secondColumn]}
		rows = append(rows, row)
	}
	return rows
}

// Next returns the next row of the sub result.
// If no more row to return, data would be nil.
func (r *SubResult) Next() (handle int64, data []types.Datum, err error) {
	if r.resp == nil {
		r.resp = new(tipb.SelectResponse)
		var b []byte
		b, err = ioutil.ReadAll(r.reader)
		r.reader.Close()
		if err != nil {
			return 0, nil, errors.Trace(err)
		}
		err = proto.Unmarshal(b, r.resp)
		if err != nil {
			return 0, nil, errors.Trace(err)
		}
		if r.resp.Error != nil {
			return 0, nil, errInvalidResp.Gen("[%d %s]", r.resp.Error.GetCode(), r.resp.Error.GetMsg())
		}
	}
	if r.cursor >= len(r.resp.Rows) {
		return 0, nil, nil
	}
	row := r.resp.Rows[r.cursor]
	data, err = tablecodec.DecodeValues(row.Data, r.fields, r.index)
	if err != nil {
		return 0, nil, errors.Trace(err)
	}
	if data == nil {
		// When no column is referenced, the data may be nil, like 'select count(*) from t'.
		// In this case, we need to create a zero length datum slice,
		// as caller will check if data is nil to finish iteration.
		data = make([]types.Datum, 0)
	}
	if !r.aggregate {
		handleBytes := row.GetHandle()
		datums, err := codec.Decode(handleBytes)
		if err != nil {
			return 0, nil, errors.Trace(err)
		}
		handle = datums[0].GetInt64()
	}
	r.cursor++
	return
}

// DecodeValues decodes a byte slice into datums with column types.
func DecodeValues(data []byte, fts []*types.FieldType, inIndex bool) ([]types.Datum, error) {
	if len(data) == 0 {
		return nil, nil
	}
	values, err := codec.Decode(data, len(fts))
	if err != nil {
		return nil, errors.Trace(err)
	}
	if len(values) > len(fts) {
		return nil, errInvalidColumnCount.Gen("invalid column count %d is less than value count %d", len(fts), len(values))
	}

	for i := range values {
		values[i], err = Unflatten(values[i], fts[i], inIndex)
		if err != nil {
			return nil, errors.Trace(err)
		}
	}
	return values, nil
}

// TableFromPB creates a table statistics from protobuffer.
func TableFromPB(ti *model.TableInfo, tpb *TablePB) (*Table, error) {
	if tpb.GetId() != ti.ID {
		return nil, errors.Errorf("table id not match, expected %d, got %d", ti.ID, tpb.GetId())
	}
	if len(tpb.Columns) != len(ti.Columns) {
		return nil, errors.Errorf("column count not match, expected %d, got %d", len(ti.Columns), len(tpb.Columns))
	}
	for i := range ti.Columns {
		if ti.Columns[i].ID != tpb.Columns[i].GetId() {
			return nil, errors.Errorf("column ID not match, expected %d, got %d", ti.Columns[i].ID, tpb.Columns[i].GetId())
		}
	}
	t := &Table{info: ti}
	t.TS = tpb.GetTs()
	t.Count = tpb.GetCount()
	t.Columns = make([]*Column, len(tpb.GetColumns()))
	for i, cInfo := range t.info.Columns {
		cpb := tpb.Columns[i]
		values, err := codec.Decode(cpb.GetValue())
		if err != nil {
			return nil, errors.Trace(err)
		}
		c := &Column{
			ID:      cpb.GetId(),
			NDV:     cpb.GetNdv(),
			Numbers: cpb.GetNumbers(),
			Values:  make([]types.Datum, len(values)),
			Repeats: cpb.GetRepeats(),
		}
		for i, val := range values {
			c.Values[i], err = tablecodec.Unflatten(val, &cInfo.FieldType)
			if err != nil {
				return nil, errors.Trace(err)
			}
		}
		t.Columns[i] = c
	}
	return t, nil
}

// DecodeValues decodes a byte slice into datums with column types.
func DecodeValues(data []byte, fts []*types.FieldType, inIndex bool) ([]types.Datum, error) {
	if data == nil {
		return nil, nil
	}
	values, err := codec.Decode(data)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if len(values) > len(fts) {
		return nil, errInvalidColumnCount.Gen("invalid column count %d is less than value count %d", len(fts), len(values))
	}
	if inIndex {
		// We don't need to unflatten index columns for now.
		return values, nil
	}

	for i := range values {
		values[i], err = unflatten(values[i], fts[i])
		if err != nil {
			return nil, errors.Trace(err)
		}
	}
	return values, nil
}

// DecodeIndexKey decodes datums from an index key.
func DecodeIndexKey(key kv.Key) ([]types.Datum, error) {
	b := key[prefixLen+idLen:]
	return codec.Decode(b, 1)
}

func (s *testXAPISuite) TestSelect(c *C) {
	defer testleak.AfterTest(c)()
	store := createMemStore(time.Now().Nanosecond())
	count := int64(10)
	err := prepareTableData(store, tbInfo, count, genValues)
	c.Check(err, IsNil)

	// Select Table request.
	txn, err := store.Begin()
	c.Check(err, IsNil)
	client := txn.GetClient()
	req, err := prepareSelectRequest(tbInfo, txn.StartTS())
	c.Check(err, IsNil)
	resp := client.Send(req)
	subResp, err := resp.Next()
	c.Check(err, IsNil)
	data, err := ioutil.ReadAll(subResp)
	c.Check(err, IsNil)
	selResp := new(tipb.SelectResponse)
	proto.Unmarshal(data, selResp)
	c.Check(selResp.Rows, HasLen, int(count))
	for i, row := range selResp.Rows {
		handle := int64(i + 1)
		expectedDatums := []types.Datum{types.NewDatum(handle)}
		expectedDatums = append(expectedDatums, genValues(handle, tbInfo)...)
		var expectedEncoded []byte
		expectedEncoded, err = codec.EncodeValue(nil, expectedDatums...)
		c.Assert(err, IsNil)
		c.Assert(row.Data, BytesEquals, expectedEncoded)
	}
	txn.Commit()

	// Select Index request.
	txn, err = store.Begin()
	c.Check(err, IsNil)
	client = txn.GetClient()
	req, err = prepareIndexRequest(tbInfo, txn.StartTS())
	c.Check(err, IsNil)
	resp = client.Send(req)
	subResp, err = resp.Next()
	c.Check(err, IsNil)
	data, err = ioutil.ReadAll(subResp)
	c.Check(err, IsNil)
	idxResp := new(tipb.SelectResponse)
	proto.Unmarshal(data, idxResp)
	c.Check(idxResp.Rows, HasLen, int(count))
	handles := make([]int, 0, 10)
	for _, row := range idxResp.Rows {
		var err error
		datums, err := codec.Decode(row.Handle)
		c.Check(err, IsNil)
		c.Check(datums, HasLen, 1)
		handles = append(handles, int(datums[0].GetInt64()))
	}
	sort.Ints(handles)
	for i, h := range handles {
		c.Assert(h, Equals, i+1)
	}
	txn.Commit()

	store.Close()
}