// CompareRows returns:
// -1 if left is smaller than right
// 0 if left and right are equal
// +1 if left is bigger than right
func CompareRows(fields []mproto.Field, compareCount int, left, right []sqltypes.Value) (int, error) {
	for i := 0; i < compareCount; i++ {
		fieldType := fields[i].Type
		lv, err := mproto.Convert(fieldType, left[i])
		if err != nil {
			return 0, err
		}
		rv, err := mproto.Convert(fieldType, right[i])
		if err != nil {
			return 0, err
		}
		switch l := lv.(type) {
		case int64:
			r := rv.(int64)
			if l < r {
				return -1, nil
			} else if l > r {
				return 1, nil
			}
		case float64:
			r := rv.(float64)
			if l < r {
				return -1, nil
			} else if l > r {
				return 1, nil
			}
		case []byte:
			r := rv.([]byte)
			return bytes.Compare(l, r), nil
		default:
			return 0, fmt.Errorf("Unsuported type %T returned by mysql.proto.Convert", l)
		}
	}
	return 0, nil
}

func (sr *StreamResult) Next() (row []interface{}) {
	if sr.qr == nil {
		// we need to read the next record that may contain
		// multiple rows
		qr, ok := <-sr.sr
		if !ok {
			if sr.errFunc() != nil {
				log.Warningf("vt: error reading the next value %v", sr.errFunc())
				sr.err = sr.errFunc()
			}
			return nil
		}
		sr.qr = qr
		sr.index = 0
	}

	row = make([]interface{}, len(sr.qr.Rows[sr.index]))
	for i, v := range sr.qr.Rows[sr.index] {
		var err error
		row[i], err = mproto.Convert(sr.columns.Fields[i].Type, v)
		if err != nil {
			panic(err) // unexpected
		}
	}

	sr.index++
	if sr.index == len(sr.qr.Rows) {
		// we reached the end of our rows, nil it so next run
		// will fetch the next one
		sr.qr = nil
	}

	return row
}

// Map2 is for a non-unique vindex.
func (lkp *lookup) Map2(vcursor planbuilder.VCursor, ids []interface{}) ([][]key.KeyspaceId, error) {
	out := make([][]key.KeyspaceId, 0, len(ids))
	bq := &tproto.BoundQuery{
		Sql: lkp.sel,
	}
	for _, id := range ids {
		bq.BindVariables = map[string]interface{}{
			lkp.From: id,
		}
		result, err := vcursor.Execute(bq)
		if err != nil {
			return nil, fmt.Errorf("lookup.Map: %v", err)
		}
		var ksids []key.KeyspaceId
		for _, row := range result.Rows {
			inum, err := mproto.Convert(result.Fields[0], row[0])
			if err != nil {
				return nil, fmt.Errorf("lookup.Map: %v", err)
			}
			num, err := getNumber(inum)
			if err != nil {
				return nil, fmt.Errorf("lookup.Map: %v", err)
			}
			ksids = append(ksids, vhash(num))
		}
		out = append(out, ksids)
	}
	return out, nil
}

// Map1 is for a unique vindex.
func (lkp *lookup) Map1(vcursor planbuilder.VCursor, ids []interface{}) ([]key.KeyspaceId, error) {
	out := make([]key.KeyspaceId, 0, len(ids))
	bq := &tproto.BoundQuery{
		Sql: lkp.sel,
	}
	for _, id := range ids {
		bq.BindVariables = map[string]interface{}{
			lkp.From: id,
		}
		result, err := vcursor.Execute(bq)
		if err != nil {
			return nil, fmt.Errorf("lookup.Map: %v", err)
		}
		if len(result.Rows) == 0 {
			out = append(out, "")
			continue
		}
		if len(result.Rows) != 1 {
			return nil, fmt.Errorf("lookup.Map: unexpected multiple results from vindex %s: %v", lkp.Table, id)
		}
		inum, err := mproto.Convert(result.Fields[0], result.Rows[0][0])
		if err != nil {
			return nil, fmt.Errorf("lookup.Map: %v", err)
		}
		num, err := getNumber(inum)
		if err != nil {
			return nil, fmt.Errorf("lookup.Map: %v", err)
		}
		out = append(out, vhash(num))
	}
	return out, nil
}

func (rtr *Router) deleteVindexEntries(vcursor *requestContext, plan *planbuilder.Plan, ks, shard string, ksid []byte) error {
	result, err := rtr.scatterConn.Execute(
		vcursor.ctx,
		plan.Subquery,
		vcursor.bindVariables,
		ks,
		[]string{shard},
		vcursor.tabletType,
		NewSafeSession(vcursor.session),
		vcursor.notInTransaction)
	if err != nil {
		return err
	}
	if len(result.Rows) == 0 {
		return nil
	}
	for i, colVindex := range plan.Table.Owned {
		keys := make(map[interface{}]bool)
		for _, row := range result.Rows {
			k, err := mproto.Convert(result.Fields[i], row[i])
			if err != nil {
				return err
			}
			switch k := k.(type) {
			case []byte:
				keys[string(k)] = true
			default:
				keys[k] = true
			}
		}
		var ids []interface{}
		for k := range keys {
			ids = append(ids, k)
		}
		switch vindex := colVindex.Vindex.(type) {
		case planbuilder.Functional:
			if err = vindex.Delete(vcursor, ids, ksid); err != nil {
				return err
			}
		case planbuilder.Lookup:
			if err = vindex.Delete(vcursor, ids, ksid); err != nil {
				return err
			}
		default:
			panic("unexpceted")
		}
	}
	return nil
}

// populateRow populates a row of data using the table's field descriptions.
// The returned types for "dest" include the list from the interface
// specification at https://golang.org/pkg/database/sql/driver/#Value
// and in addition the type "uint64" for unsigned BIGINT MySQL records.
func populateRow(dest []driver.Value, fields []*query.Field, row []sqltypes.Value) error {
	if len(dest) != len(fields) {
		return fmt.Errorf("length mismatch: dest is %d, fields are %d", len(dest), len(fields))
	}
	if len(fields) != len(row) {
		return fmt.Errorf("internal error: length mismatch: dest is %d, fields are %d", len(fields), len(row))
	}
	var err error
	for i := range dest {
		dest[i], err = mproto.Convert(fields[i], row[i])
		if err != nil {
			return fmt.Errorf("conversion error: field: %v, val: %v: %v", fields[i], row[i], err)
		}
	}
	return nil
}