From c8cef12cb41a2d0475293a4b740ed1e79b8ddd00 Mon Sep 17 00:00:00 2001
From: Ewan Chou <coocood@gmail.com>
Date: Mon, 3 Sep 2018 15:15:33 +0800
Subject: [PATCH 1/7] update import package

---
 util/rowcodec/common.go        | 194 ++++++++++++++++++++
 util/rowcodec/decoder.go       | 213 ++++++++++++++++++++++
 util/rowcodec/encoder.go       | 314 +++++++++++++++++++++++++++++++++
 util/rowcodec/rowcodec_test.go | 201 +++++++++++++++++++++
 4 files changed, 922 insertions(+)
 create mode 100644 util/rowcodec/common.go
 create mode 100644 util/rowcodec/decoder.go
 create mode 100644 util/rowcodec/encoder.go
 create mode 100644 util/rowcodec/rowcodec_test.go

diff --git a/util/rowcodec/common.go b/util/rowcodec/common.go
new file mode 100644
index 0000000000000..8a4f06102adcc
--- /dev/null
+++ b/util/rowcodec/common.go
@@ -0,0 +1,194 @@
+// Copyright 2018 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package rowcodec
+
+import (
+	"encoding/binary"
+	"reflect"
+	"unsafe"
+
+	"github.com/pingcap/errors"
+)
+
+// CodecVer is the constant number that represent the new row format.
+const CodecVer = 128
+
+var errInvalidCodecVer = errors.New("invalid codec version")
+
+// row is the struct type used to access the a row.
+type row struct {
+	// small:  colID []byte, offsets []uint16, optimized for most cases.
+	// large:  colID []uint32, offsets []uint32.
+	large          bool
+	numNotNullCols uint16
+	numNullCols    uint16
+	colIDs         []byte
+	offsets        []uint16
+	data           []byte
+
+	// for large row
+	colIDs32  []uint32
+	offsets32 []uint32
+}
+
+func (r *row) getData(i int) []byte {
+	var start, end uint32
+	if r.large {
+		if i > 0 {
+			start = r.offsets32[i-1]
+		}
+		end = r.offsets32[i]
+	} else {
+		if i > 0 {
+			start = uint32(r.offsets[i-1])
+		}
+		end = uint32(r.offsets[i])
+	}
+	return r.data[start:end]
+}
+
+func (r *row) setRowData(rowData []byte) error {
+	if rowData[0] != CodecVer {
+		return errInvalidCodecVer
+	}
+	r.large = rowData[1]&1 > 0
+	r.numNotNullCols = binary.LittleEndian.Uint16(rowData[2:])
+	r.numNullCols = binary.LittleEndian.Uint16(rowData[4:])
+	cursor := 6
+	if r.large {
+		colIDsLen := int(r.numNotNullCols+r.numNullCols) * 4
+		r.colIDs32 = bytesToU32Slice(rowData[cursor : cursor+colIDsLen])
+		cursor += colIDsLen
+		offsetsLen := int(r.numNotNullCols) * 4
+		r.offsets32 = bytesToU32Slice(rowData[cursor : cursor+offsetsLen])
+		cursor += offsetsLen
+	} else {
+		colIDsLen := int(r.numNotNullCols + r.numNullCols)
+		r.colIDs = rowData[cursor : cursor+colIDsLen]
+		cursor += colIDsLen
+		offsetsLen := int(r.numNotNullCols) * 2
+		r.offsets = bytes2U16Slice(rowData[cursor : cursor+offsetsLen])
+		cursor += offsetsLen
+	}
+	r.data = rowData[cursor:]
+	return nil
+}
+
+func bytesToU32Slice(b []byte) []uint32 {
+	if len(b) == 0 {
+		return nil
+	}
+	var u32s []uint32
+	hdr := (*reflect.SliceHeader)(unsafe.Pointer(&u32s))
+	hdr.Len = len(b) / 4
+	hdr.Cap = hdr.Len
+	hdr.Data = uintptr(unsafe.Pointer(&b[0]))
+	return u32s
+}
+
+func bytes2U16Slice(b []byte) []uint16 {
+	if len(b) == 0 {
+		return nil
+	}
+	var u16s []uint16
+	hdr := (*reflect.SliceHeader)(unsafe.Pointer(&u16s))
+	hdr.Len = len(b) / 2
+	hdr.Cap = hdr.Len
+	hdr.Data = uintptr(unsafe.Pointer(&b[0]))
+	return u16s
+}
+
+func u16SliceToBytes(u16s []uint16) []byte {
+	if len(u16s) == 0 {
+		return nil
+	}
+	var b []byte
+	hdr := (*reflect.SliceHeader)(unsafe.Pointer(&b))
+	hdr.Len = len(u16s) * 2
+	hdr.Cap = hdr.Len
+	hdr.Data = uintptr(unsafe.Pointer(&u16s[0]))
+	return b
+}
+
+func u32SliceToBytes(u32s []uint32) []byte {
+	if len(u32s) == 0 {
+		return nil
+	}
+	var b []byte
+	hdr := (*reflect.SliceHeader)(unsafe.Pointer(&b))
+	hdr.Len = len(u32s) * 4
+	hdr.Cap = hdr.Len
+	hdr.Data = uintptr(unsafe.Pointer(&u32s[0]))
+	return b
+}
+
+func encodeInt(buf []byte, iVal int64) []byte {
+	var tmp [8]byte
+	if int64(int8(iVal)) == iVal {
+		buf = append(buf, byte(iVal))
+	} else if int64(int16(iVal)) == iVal {
+		binary.LittleEndian.PutUint16(tmp[:], uint16(iVal))
+		buf = append(buf, tmp[:2]...)
+	} else if int64(int32(iVal)) == iVal {
+		binary.LittleEndian.PutUint32(tmp[:], uint32(iVal))
+		buf = append(buf, tmp[:4]...)
+	} else {
+		binary.LittleEndian.PutUint64(tmp[:], uint64(iVal))
+		buf = append(buf, tmp[:8]...)
+	}
+	return buf
+}
+
+func decodeInt(val []byte) int64 {
+	switch len(val) {
+	case 1:
+		return int64(int8(val[0]))
+	case 2:
+		return int64(int16(binary.LittleEndian.Uint16(val)))
+	case 4:
+		return int64(int32(binary.LittleEndian.Uint32(val)))
+	default:
+		return int64(binary.LittleEndian.Uint64(val))
+	}
+}
+
+func encodeUint(buf []byte, uVal uint64) []byte {
+	var tmp [8]byte
+	if uint64(uint8(uVal)) == uVal {
+		buf = append(buf, byte(uVal))
+	} else if uint64(uint16(uVal)) == uVal {
+		binary.LittleEndian.PutUint16(tmp[:], uint16(uVal))
+		buf = append(buf, tmp[:2]...)
+	} else if uint64(uint32(uVal)) == uVal {
+		binary.LittleEndian.PutUint32(tmp[:], uint32(uVal))
+		buf = append(buf, tmp[:4]...)
+	} else {
+		binary.LittleEndian.PutUint64(tmp[:], uint64(uVal))
+		buf = append(buf, tmp[:8]...)
+	}
+	return buf
+}
+
+func decodeUint(val []byte) uint64 {
+	switch len(val) {
+	case 1:
+		return uint64(val[0])
+	case 2:
+		return uint64(binary.LittleEndian.Uint16(val))
+	case 4:
+		return uint64(binary.LittleEndian.Uint32(val))
+	default:
+		return binary.LittleEndian.Uint64(val)
+	}
+}
diff --git a/util/rowcodec/decoder.go b/util/rowcodec/decoder.go
new file mode 100644
index 0000000000000..01b03366488dd
--- /dev/null
+++ b/util/rowcodec/decoder.go
@@ -0,0 +1,213 @@
+// Copyright 2018 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package rowcodec
+
+import (
+	"math"
+	"time"
+
+	"github.com/pingcap/errors"
+	"github.com/pingcap/parser/mysql"
+	"github.com/pingcap/tidb/sessionctx/stmtctx"
+	"github.com/pingcap/tidb/types"
+	"github.com/pingcap/tidb/types/json"
+	"github.com/pingcap/tidb/util/chunk"
+	"github.com/pingcap/tidb/util/codec"
+)
+
+// Decoder decodes the row to chunk.Chunk.
+type Decoder struct {
+	row
+	requestColIDs []int64
+	handleColID   int64
+	requestTypes  []*types.FieldType
+	origDefaults  [][]byte
+	loc           *time.Location
+}
+
+// NewDecoder creates a NewDecoder.
+// requestColIDs is the columnIDs to decode. tps is the field types for request columns.
+// origDefault is the original default value in old format, if the column ID is not found in the row,
+// the origDefault will be used.
+func NewDecoder(requestColIDs []int64, handleColID int64, tps []*types.FieldType, origDefaults [][]byte,
+	sc *stmtctx.StatementContext) (*Decoder, error) {
+	xOrigDefaultVals := make([][]byte, len(origDefaults))
+	for i := 0; i < len(origDefaults); i++ {
+		if len(origDefaults[i]) == 0 {
+			continue
+		}
+		xDefaultVal, err := convertDefaultValue(origDefaults[i], sc)
+		if err != nil {
+			return nil, err
+		}
+		xOrigDefaultVals[i] = xDefaultVal
+	}
+	return &Decoder{
+		requestColIDs: requestColIDs,
+		handleColID:   handleColID,
+		requestTypes:  tps,
+		origDefaults:  xOrigDefaultVals,
+		loc:           sc.TimeZone,
+	}, nil
+}
+
+func convertDefaultValue(defaultVal []byte, sc *stmtctx.StatementContext) (colVal []byte, err error) {
+	var d types.Datum
+	_, d, err = codec.DecodeOne(defaultVal)
+	if err != nil {
+		return
+	}
+	return encodeDatum(nil, d, sc)
+}
+
+// Decode decodes a row to chunk.
+func (decoder *Decoder) Decode(rowData []byte, handle int64, chk *chunk.Chunk) error {
+	err := decoder.setRowData(rowData)
+	if err != nil {
+		return err
+	}
+	for colIdx, colID := range decoder.requestColIDs {
+		if colID == decoder.handleColID {
+			chk.AppendInt64(colIdx, handle)
+			continue
+		}
+		// Search the column in not-null columns array.
+		i, j := 0, int(decoder.numNotNullCols)
+		var found bool
+		for i < j {
+			h := int(uint(i+j) >> 1) // avoid overflow when computing h
+			// i ≤ h < j
+			var v int64
+			if decoder.large {
+				v = int64(decoder.colIDs32[h])
+			} else {
+				v = int64(decoder.colIDs[h])
+			}
+			if v < colID {
+				i = h + 1
+			} else if v > colID {
+				j = h
+			} else {
+				found = true
+				colData := decoder.getData(h)
+				err := decoder.decodeColData(colIdx, colData, chk)
+				if err != nil {
+					return err
+				}
+				break
+			}
+		}
+		if found {
+			continue
+		}
+		// Search the column in null columns array.
+		i, j = int(decoder.numNotNullCols), int(decoder.numNotNullCols+decoder.numNullCols)
+		for i < j {
+			h := int(uint(i+j) >> 1) // avoid overflow when computing h
+			// i ≤ h < j
+			var v int64
+			if decoder.large {
+				v = int64(decoder.colIDs32[h])
+			} else {
+				v = int64(decoder.colIDs[h])
+			}
+			if v < colID {
+				i = h + 1
+			} else if v > colID {
+				j = h
+			} else {
+				found = true
+				break
+			}
+		}
+		if found || decoder.origDefaults[colIdx] == nil {
+			chk.AppendNull(colIdx)
+		} else {
+			err := decoder.decodeColData(colIdx, decoder.origDefaults[colIdx], chk)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (decoder *Decoder) decodeColData(colIdx int, colData []byte, chk *chunk.Chunk) error {
+	ft := decoder.requestTypes[colIdx]
+	switch ft.Tp {
+	case mysql.TypeLonglong, mysql.TypeLong, mysql.TypeInt24, mysql.TypeShort, mysql.TypeTiny, mysql.TypeYear:
+		if mysql.HasUnsignedFlag(ft.Flag) {
+			chk.AppendUint64(colIdx, decodeUint(colData))
+		} else {
+			chk.AppendInt64(colIdx, decodeInt(colData))
+		}
+	case mysql.TypeFloat:
+		chk.AppendFloat32(colIdx, float32(math.Float64frombits(decodeUint(colData))))
+	case mysql.TypeDouble:
+		chk.AppendFloat64(colIdx, math.Float64frombits(decodeUint(colData)))
+	case mysql.TypeVarString, mysql.TypeVarchar, mysql.TypeString,
+		mysql.TypeBlob, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob:
+		chk.AppendBytes(colIdx, colData)
+	case mysql.TypeNewDecimal:
+		_, dec, _, _, err := codec.DecodeDecimal(colData)
+		if err != nil {
+			return err
+		}
+		chk.AppendMyDecimal(colIdx, dec)
+	case mysql.TypeDate, mysql.TypeDatetime, mysql.TypeTimestamp:
+		var t types.Time
+		t.Type = ft.Tp
+		t.Fsp = ft.Decimal
+		err := t.FromPackedUint(decodeUint(colData))
+		if err != nil {
+			return err
+		}
+		if ft.Tp == mysql.TypeTimestamp && !t.IsZero() {
+			err = t.ConvertTimeZone(time.UTC, decoder.loc)
+			if err != nil {
+				return err
+			}
+		}
+		chk.AppendTime(colIdx, t)
+	case mysql.TypeDuration:
+		var dur types.Duration
+		dur.Duration = time.Duration(decodeInt(colData))
+		dur.Fsp = ft.Decimal
+		chk.AppendDuration(colIdx, dur)
+	case mysql.TypeEnum:
+		// ignore error deliberately, to read empty enum value.
+		enum, err := types.ParseEnumValue(ft.Elems, decodeUint(colData))
+		if err != nil {
+			enum = types.Enum{}
+		}
+		chk.AppendEnum(colIdx, enum)
+	case mysql.TypeSet:
+		set, err := types.ParseSetValue(ft.Elems, decodeUint(colData))
+		if err != nil {
+			return err
+		}
+		chk.AppendSet(colIdx, set)
+	case mysql.TypeBit:
+		byteSize := (ft.Flen + 7) >> 3
+		chk.AppendBytes(colIdx, types.NewBinaryLiteralFromUint(decodeUint(colData), byteSize))
+	case mysql.TypeJSON:
+		var j json.BinaryJSON
+		j.TypeCode = colData[0]
+		j.Value = colData[1:]
+		chk.AppendJSON(colIdx, j)
+	default:
+		return errors.Errorf("unknown type %d", ft.Tp)
+	}
+	return nil
+}
diff --git a/util/rowcodec/encoder.go b/util/rowcodec/encoder.go
new file mode 100644
index 0000000000000..45d565ffb0c44
--- /dev/null
+++ b/util/rowcodec/encoder.go
@@ -0,0 +1,314 @@
+// Copyright 2018 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package rowcodec
+
+import (
+	"math"
+	"sort"
+	"time"
+
+	"github.com/pingcap/errors"
+	"github.com/pingcap/parser/mysql"
+	"github.com/pingcap/parser/terror"
+	"github.com/pingcap/tidb/sessionctx/stmtctx"
+	"github.com/pingcap/tidb/types"
+	"github.com/pingcap/tidb/util/codec"
+)
+
+// Encoder is used to encode a row.
+type Encoder struct {
+	row
+	tempColIDs []int64
+	values     []types.Datum
+	tempData   []byte
+	sc         *stmtctx.StatementContext
+}
+
+func (encoder *Encoder) reset() {
+	encoder.large = false
+	encoder.numNotNullCols = 0
+	encoder.numNullCols = 0
+	encoder.data = encoder.data[:0]
+	encoder.tempColIDs = encoder.tempColIDs[:0]
+	encoder.values = encoder.values[:0]
+}
+
+func (encoder *Encoder) addColumn(colID int64, d types.Datum) {
+	if colID > 255 {
+		encoder.large = true
+	}
+	if d.IsNull() {
+		encoder.numNullCols++
+	} else {
+		encoder.numNotNullCols++
+	}
+	encoder.tempColIDs = append(encoder.tempColIDs, colID)
+	encoder.values = append(encoder.values, d)
+}
+
+// Encode encodes a row from a datums slice.
+func (encoder *Encoder) Encode(colIDs []int64, values []types.Datum, buf []byte) ([]byte, error) {
+	encoder.reset()
+	for i, colID := range colIDs {
+		encoder.addColumn(colID, values[i])
+	}
+	return encoder.build(buf[:0])
+}
+
+func (encoder *Encoder) build(buf []byte) ([]byte, error) {
+	r := &encoder.row
+	// Separate null and not-null column IDs.
+	numCols := len(encoder.tempColIDs)
+	nullIdx := numCols - int(r.numNullCols)
+	notNullIdx := 0
+	if r.large {
+		encoder.initColIDs32()
+		encoder.initOffsets32()
+	} else {
+		encoder.initColIDs()
+		encoder.initOffsets()
+	}
+	for i, colID := range encoder.tempColIDs {
+		if encoder.values[i].IsNull() {
+			if r.large {
+				r.colIDs32[nullIdx] = uint32(colID)
+			} else {
+				r.colIDs[nullIdx] = byte(colID)
+			}
+			nullIdx++
+		} else {
+			if r.large {
+				r.colIDs32[notNullIdx] = uint32(colID)
+			} else {
+				r.colIDs[notNullIdx] = byte(colID)
+			}
+			encoder.values[notNullIdx] = encoder.values[i]
+			notNullIdx++
+		}
+	}
+	if r.large {
+		largeNotNullSorter := (*largeNotNullSorter)(encoder)
+		sort.Sort(largeNotNullSorter)
+		if r.numNullCols > 0 {
+			largeNullSorter := (*largeNullSorter)(encoder)
+			sort.Sort(largeNullSorter)
+		}
+	} else {
+		smallNotNullSorter := (*smallNotNullSorter)(encoder)
+		sort.Sort(smallNotNullSorter)
+		if r.numNullCols > 0 {
+			smallNullSorter := (*smallNullSorter)(encoder)
+			sort.Sort(smallNullSorter)
+		}
+	}
+	for i := 0; i < notNullIdx; i++ {
+		var err error
+		r.data, err = encodeDatum(r.data, encoder.values[i], encoder.sc)
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
+		if len(r.data) > math.MaxUint16 && !r.large {
+			// We need to convert the row to large row.
+			encoder.initColIDs32()
+			for j := 0; j < numCols; j++ {
+				r.colIDs32[j] = uint32(r.colIDs[j])
+			}
+			encoder.initOffsets32()
+			for j := 0; j <= i; j++ {
+				r.offsets32[j] = uint32(r.offsets[j])
+			}
+			r.large = true
+		}
+		if r.large {
+			r.offsets32[i] = uint32(len(r.data))
+		} else {
+			r.offsets[i] = uint16(len(r.data))
+		}
+	}
+	buf = append(buf, CodecVer)
+	flag := byte(0)
+	if r.large {
+		flag = 1
+	}
+	buf = append(buf, flag)
+	buf = append(buf, byte(r.numNotNullCols), byte(r.numNotNullCols>>8))
+	buf = append(buf, byte(r.numNullCols), byte(r.numNullCols>>8))
+	if r.large {
+		buf = append(buf, u32SliceToBytes(r.colIDs32)...)
+		buf = append(buf, u32SliceToBytes(r.offsets32)...)
+	} else {
+		buf = append(buf, r.colIDs...)
+		buf = append(buf, u16SliceToBytes(r.offsets)...)
+	}
+	buf = append(buf, r.data...)
+	return buf, nil
+}
+
+func encodeDatum(buf []byte, d types.Datum, sc *stmtctx.StatementContext) ([]byte, error) {
+	switch d.Kind() {
+	case types.KindInt64:
+		buf = encodeInt(buf, d.GetInt64())
+	case types.KindUint64:
+		buf = encodeUint(buf, d.GetUint64())
+	case types.KindString, types.KindBytes:
+		buf = append(buf, d.GetBytes()...)
+	case types.KindFloat32, types.KindFloat64:
+		buf = encodeUint(buf, uint64(math.Float64bits(d.GetFloat64())))
+	case types.KindMysqlDecimal:
+		var err error
+		buf, err = codec.EncodeDecimal(buf, d.GetMysqlDecimal(), d.Length(), d.Frac())
+		if terror.ErrorEqual(err, types.ErrTruncated) {
+			err = sc.HandleTruncate(err)
+		} else if terror.ErrorEqual(err, types.ErrOverflow) {
+			err = sc.HandleOverflow(err, err)
+		}
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
+	case types.KindMysqlTime:
+		t := d.GetMysqlTime()
+		// Encoding timestamp need to consider timezone.
+		// If it's not in UTC, transform to UTC first.
+		if t.Type == mysql.TypeTimestamp && sc.TimeZone != time.UTC {
+			err := t.ConvertTimeZone(sc.TimeZone, time.UTC)
+			if err != nil {
+				return nil, errors.Trace(err)
+			}
+		}
+		v, err := t.ToPackedUint()
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
+		buf = encodeUint(buf, v)
+	case types.KindMysqlDuration:
+		buf = encodeInt(buf, int64(d.GetMysqlDuration().Duration))
+	case types.KindMysqlEnum:
+		buf = encodeUint(buf, uint64(d.GetMysqlEnum().ToNumber()))
+	case types.KindMysqlSet:
+		buf = encodeUint(buf, uint64(d.GetMysqlSet().ToNumber()))
+	case types.KindMysqlBit, types.KindBinaryLiteral:
+		val, err := types.BinaryLiteral(d.GetBytes()).ToInt(sc)
+		if err != nil {
+			terror.Log(errors.Trace(err))
+		}
+		buf = encodeUint(buf, val)
+	case types.KindMysqlJSON:
+		j := d.GetMysqlJSON()
+		buf = append(buf, j.TypeCode)
+		buf = append(buf, j.Value...)
+	default:
+		return nil, errors.Errorf("unsupport encode type %d", d.Kind())
+	}
+	return buf, nil
+}
+
+func (encoder *Encoder) initColIDs() {
+	numCols := int(encoder.numNotNullCols + encoder.numNullCols)
+	if cap(encoder.colIDs) >= numCols {
+		encoder.colIDs = encoder.colIDs[:numCols]
+	} else {
+		encoder.colIDs = make([]byte, numCols)
+	}
+}
+
+func (encoder *Encoder) initColIDs32() {
+	numCols := int(encoder.numNotNullCols + encoder.numNullCols)
+	if cap(encoder.colIDs32) >= numCols {
+		encoder.colIDs32 = encoder.colIDs32[:numCols]
+	} else {
+		encoder.colIDs32 = make([]uint32, numCols)
+	}
+}
+
+func (encoder *Encoder) initOffsets() {
+	if cap(encoder.offsets) >= int(encoder.numNotNullCols) {
+		encoder.offsets = encoder.offsets[:encoder.numNotNullCols]
+	} else {
+		encoder.offsets = make([]uint16, encoder.numNotNullCols)
+	}
+}
+
+func (encoder *Encoder) initOffsets32() {
+	if cap(encoder.offsets32) >= int(encoder.numNotNullCols) {
+		encoder.offsets32 = encoder.offsets32[:encoder.numNotNullCols]
+	} else {
+		encoder.offsets32 = make([]uint32, encoder.numNotNullCols)
+	}
+}
+
+/*
+	We define several sorters to avoid switch cost in sort functions.
+*/
+
+type largeNotNullSorter Encoder
+
+func (s *largeNotNullSorter) Less(i, j int) bool {
+	return s.colIDs32[i] < s.colIDs32[j]
+}
+
+func (s *largeNotNullSorter) Len() int {
+	return int(s.numNotNullCols)
+}
+
+func (s *largeNotNullSorter) Swap(i, j int) {
+	s.colIDs32[i], s.colIDs32[j] = s.colIDs32[j], s.colIDs32[i]
+	s.values[i], s.values[j] = s.values[j], s.values[i]
+}
+
+type smallNotNullSorter Encoder
+
+func (s *smallNotNullSorter) Less(i, j int) bool {
+	return s.colIDs[i] < s.colIDs[j]
+}
+
+func (s *smallNotNullSorter) Len() int {
+	return int(s.numNotNullCols)
+}
+
+func (s *smallNotNullSorter) Swap(i, j int) {
+	s.colIDs[i], s.colIDs[j] = s.colIDs[j], s.colIDs[i]
+	s.values[i], s.values[j] = s.values[j], s.values[i]
+}
+
+type smallNullSorter Encoder
+
+func (s *smallNullSorter) Less(i, j int) bool {
+	nullCols := s.colIDs[s.numNotNullCols:]
+	return nullCols[i] < nullCols[j]
+}
+
+func (s *smallNullSorter) Len() int {
+	return int(s.numNullCols)
+}
+
+func (s *smallNullSorter) Swap(i, j int) {
+	nullCols := s.colIDs[s.numNotNullCols:]
+	nullCols[i], nullCols[j] = nullCols[j], nullCols[i]
+}
+
+type largeNullSorter Encoder
+
+func (s *largeNullSorter) Less(i, j int) bool {
+	nullCols := s.colIDs32[s.numNotNullCols:]
+	return nullCols[i] < nullCols[j]
+}
+
+func (s *largeNullSorter) Len() int {
+	return int(s.numNullCols)
+}
+
+func (s *largeNullSorter) Swap(i, j int) {
+	nullCols := s.colIDs32[s.numNotNullCols:]
+	nullCols[i], nullCols[j] = nullCols[j], nullCols[i]
+}
diff --git a/util/rowcodec/rowcodec_test.go b/util/rowcodec/rowcodec_test.go
new file mode 100644
index 0000000000000..ca974a4868581
--- /dev/null
+++ b/util/rowcodec/rowcodec_test.go
@@ -0,0 +1,201 @@
+// Copyright 2018 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package rowcodec
+
+import (
+	"math"
+	"testing"
+	"time"
+
+	. "github.com/pingcap/check"
+	"github.com/pingcap/parser/mysql"
+	"github.com/pingcap/tidb/sessionctx/stmtctx"
+	"github.com/pingcap/tidb/types"
+	"github.com/pingcap/tidb/types/json"
+	"github.com/pingcap/tidb/util/chunk"
+	"github.com/pingcap/tidb/util/codec"
+)
+
+func TestT(t *testing.T) {
+	TestingT(t)
+}
+
+var _ = Suite(&testSuite{})
+
+type testSuite struct{}
+
+func (s *testSuite) TestRowCodec(c *C) {
+	colIDs := []int64{1, 2, 3, 4, 5, 10}
+	var rb Encoder
+	rb.sc = new(stmtctx.StatementContext)
+	dt, err := types.ParseDatetime(rb.sc, "2018-01-19 03:14:07.999999")
+	c.Assert(err, IsNil)
+	datums := types.MakeDatums(
+		dt,
+		"abc",
+		nil,
+		1,
+		types.NewDecFromInt(1),
+		"abc",
+	)
+	newRow, err := rb.Encode(colIDs, datums, nil)
+	c.Check(err, IsNil)
+	s.checkDecode(c, rb.sc, newRow)
+
+	// Test large column ID
+	colIDs = []int64{1, 2, 3, 4, 5, 512}
+	newRow, err = rb.Encode(colIDs, datums, nil)
+	c.Check(err, IsNil)
+	s.checkDecode(c, rb.sc, newRow)
+
+	// Test large column value
+	colIDs = []int64{1, 2, 3, 4, 5, 10}
+	datums[5] = types.NewBytesDatum(make([]byte, 65536))
+	newRow, err = rb.Encode(colIDs, datums, nil)
+	c.Check(err, IsNil)
+	s.checkDecode(c, rb.sc, newRow)
+}
+
+func (s *testSuite) TestIntCodec(c *C) {
+	uints := []uint64{255, math.MaxUint16, math.MaxUint32, math.MaxUint32 + 1}
+	sizes := []int{1, 2, 4, 8}
+	for i, v := range uints {
+		data := encodeUint(nil, v)
+		c.Assert(len(data), Equals, sizes[i])
+		c.Assert(decodeUint(data), Equals, v)
+	}
+
+	ints := []int64{127, math.MaxInt16, math.MaxInt32, math.MaxInt32 + 1}
+	for i, v := range ints {
+		data := encodeInt(nil, v)
+		c.Assert(len(data), Equals, sizes[i])
+		c.Assert(decodeInt(data), Equals, v)
+	}
+}
+
+func (s *testSuite) TestMoreTypes(c *C) {
+	colIDs := []int64{1, 2, 3, 4, 5, 6, 7, 8}
+	var rb Encoder
+	rb.sc = new(stmtctx.StatementContext)
+	rb.sc.TimeZone = time.Local
+	ts, err := types.ParseTimestampFromNum(rb.sc, 20181111090909)
+	c.Assert(err, IsNil)
+	datums := types.MakeDatums(
+		float32(1.0),
+		float64(1.0),
+		ts,
+		types.Duration{Duration: time.Minute},
+		types.Enum{Name: "a", Value: 1},
+		types.Set{Name: "a", Value: 1},
+		json.CreateBinary("abc"),
+		types.BitLiteral([]byte{101}),
+	)
+	newRow, err := rb.Encode(colIDs, datums, nil)
+	c.Check(err, IsNil)
+	fieldTypes := []*types.FieldType{
+		types.NewFieldType(mysql.TypeFloat),
+		types.NewFieldType(mysql.TypeDouble),
+		{Tp: mysql.TypeTimestamp, Decimal: 0},
+		types.NewFieldType(mysql.TypeDuration),
+		{Tp: mysql.TypeEnum, Elems: []string{"a"}},
+		{Tp: mysql.TypeSet, Elems: []string{"a"}},
+		types.NewFieldType(mysql.TypeJSON),
+		{Tp: mysql.TypeBit, Flen: 8},
+	}
+	rd, err := NewDecoder(colIDs, 0, fieldTypes, make([][]byte, 8), rb.sc)
+	c.Assert(err, IsNil)
+	chk := chunk.NewChunkWithCapacity(fieldTypes, 1)
+	err = rd.Decode(newRow, 0, chk)
+	c.Assert(err, IsNil)
+	row := chk.GetRow(0)
+	c.Assert(row.GetFloat32(0), Equals, float32(1.0))
+	c.Assert(row.GetFloat64(1), Equals, float64(1.0))
+	c.Assert(row.GetTime(2).String(), Equals, ts.String())
+	c.Assert(row.GetDuration(3, 0), Equals, datums[3].GetMysqlDuration())
+	c.Assert(row.GetEnum(4), Equals, datums[4].GetMysqlEnum())
+	c.Assert(row.GetSet(5), Equals, datums[5].GetMysqlSet())
+	c.Assert(row.GetJSON(6), DeepEquals, datums[6].GetMysqlJSON())
+	c.Assert(row.GetBytes(7), DeepEquals, []byte(datums[7].GetMysqlBit()))
+}
+
+func (s *testSuite) checkDecode(c *C, sc *stmtctx.StatementContext, newRow []byte) {
+	readRowTypes := []*types.FieldType{
+		types.NewFieldType(mysql.TypeVarString),
+		types.NewFieldType(mysql.TypeLonglong),
+		types.NewFieldType(mysql.TypeLonglong),
+		types.NewFieldType(mysql.TypeNewDecimal),
+		types.NewFieldType(mysql.TypeLonglong),
+		types.NewFieldType(mysql.TypeLonglong),
+	}
+	readColIDS := []int64{2, 3, 4, 5, 6, 7}
+	defaultColVal, err := codec.EncodeValue(sc, nil, types.NewIntDatum(5))
+	defaults := [][]byte{nil, defaultColVal, defaultColVal, nil, defaultColVal, nil}
+
+	rd, err := NewDecoder(readColIDS, 7, readRowTypes, defaults, sc)
+	c.Assert(err, IsNil)
+	chk := chunk.NewChunkWithCapacity(readRowTypes, 1)
+	err = rd.Decode(newRow, 1000, chk)
+	c.Assert(err, IsNil)
+	row := chk.GetRow(0)
+	c.Assert(row.GetString(0), Equals, "abc")
+	c.Assert(row.IsNull(1), IsTrue)
+	c.Assert(row.GetInt64(2), Equals, int64(1))
+	c.Assert(row.GetMyDecimal(3).String(), Equals, "1")
+	c.Assert(row.GetInt64(4), Equals, int64(5))
+	c.Assert(row.GetInt64(5), Equals, int64(1000))
+}
+
+func BenchmarkEncode(b *testing.B) {
+	b.ReportAllocs()
+	oldRow := types.MakeDatums(1, "abc", 1.1)
+	var xb Encoder
+	var buf []byte
+	colIDs := []int64{1, 2, 3}
+	var err error
+	for i := 0; i < b.N; i++ {
+		buf, err = xb.Encode(colIDs, oldRow, buf)
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func BenchmarkDecode(b *testing.B) {
+	b.ReportAllocs()
+	oldRow := types.MakeDatums(1, "abc", 1.1)
+	colIDs := []int64{-1, 2, 3}
+	tps := []*types.FieldType{
+		types.NewFieldType(mysql.TypeLonglong),
+		types.NewFieldType(mysql.TypeString),
+		types.NewFieldType(mysql.TypeDouble),
+	}
+	var xb Encoder
+	xb.sc = new(stmtctx.StatementContext)
+	xRowData, err := xb.Encode(colIDs, oldRow, nil)
+	if err != nil {
+		b.Fatal(err)
+	}
+	decoder, err := NewDecoder(colIDs, -1, tps, make([][]byte, 3), xb.sc)
+	if err != nil {
+		b.Fatal(err)
+	}
+	chk := chunk.NewChunkWithCapacity(tps, 1)
+	for i := 0; i < b.N; i++ {
+		chk.Reset()
+		err = decoder.Decode(xRowData, 1, chk)
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+}

From f3ddd548778cf7d5bdd7c1e55c4ef22212409798 Mon Sep 17 00:00:00 2001
From: Ewan Chou <coocood@gmail.com>
Date: Mon, 5 Aug 2019 13:48:37 +0800
Subject: [PATCH 2/7] fix ci

---
 util/rowcodec/rowcodec_test.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/util/rowcodec/rowcodec_test.go b/util/rowcodec/rowcodec_test.go
index ca974a4868581..806650a3836b5 100644
--- a/util/rowcodec/rowcodec_test.go
+++ b/util/rowcodec/rowcodec_test.go
@@ -140,6 +140,7 @@ func (s *testSuite) checkDecode(c *C, sc *stmtctx.StatementContext, newRow []byt
 	}
 	readColIDS := []int64{2, 3, 4, 5, 6, 7}
 	defaultColVal, err := codec.EncodeValue(sc, nil, types.NewIntDatum(5))
+	c.Assert(err, IsNil)
 	defaults := [][]byte{nil, defaultColVal, defaultColVal, nil, defaultColVal, nil}
 
 	rd, err := NewDecoder(readColIDS, 7, readRowTypes, defaults, sc)

From 566503e1a8538bac3f3232e960f0abb9ccf27453 Mon Sep 17 00:00:00 2001
From: Ewan Chou <coocood@gmail.com>
Date: Mon, 5 Aug 2019 17:03:46 +0800
Subject: [PATCH 3/7] address comment

---
 util/rowcodec/common.go        | 13 ++++++++-----
 util/rowcodec/decoder.go       |  2 +-
 util/rowcodec/encoder.go       |  2 +-
 util/rowcodec/rowcodec_test.go |  2 +-
 4 files changed, 11 insertions(+), 8 deletions(-)

diff --git a/util/rowcodec/common.go b/util/rowcodec/common.go
index 8a4f06102adcc..47ca277ac7744 100644
--- a/util/rowcodec/common.go
+++ b/util/rowcodec/common.go
@@ -1,4 +1,4 @@
-// Copyright 2018 PingCAP, Inc.
+// Copyright 2019 PingCAP, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -27,16 +27,19 @@ const CodecVer = 128
 var errInvalidCodecVer = errors.New("invalid codec version")
 
 // row is the struct type used to access the a row.
+// There are two types of row, small and large.
+// small rows use one byte colID and two bytes offset, optimized for most cases.
+// large rows use four bytes colID and four bytes offset.
 type row struct {
-	// small:  colID []byte, offsets []uint16, optimized for most cases.
-	// large:  colID []uint32, offsets []uint32.
 	large          bool
 	numNotNullCols uint16
 	numNullCols    uint16
-	colIDs         []byte
-	offsets        []uint16
 	data           []byte
 
+	// for small rows
+	colIDs  []byte
+	offsets []uint16
+
 	// for large row
 	colIDs32  []uint32
 	offsets32 []uint32
diff --git a/util/rowcodec/decoder.go b/util/rowcodec/decoder.go
index 01b03366488dd..0acd8088d3db6 100644
--- a/util/rowcodec/decoder.go
+++ b/util/rowcodec/decoder.go
@@ -1,4 +1,4 @@
-// Copyright 2018 PingCAP, Inc.
+// Copyright 2019 PingCAP, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
diff --git a/util/rowcodec/encoder.go b/util/rowcodec/encoder.go
index 45d565ffb0c44..ed9a9d1d7c493 100644
--- a/util/rowcodec/encoder.go
+++ b/util/rowcodec/encoder.go
@@ -1,4 +1,4 @@
-// Copyright 2018 PingCAP, Inc.
+// Copyright 2019 PingCAP, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
diff --git a/util/rowcodec/rowcodec_test.go b/util/rowcodec/rowcodec_test.go
index 806650a3836b5..aa14e2fa068f8 100644
--- a/util/rowcodec/rowcodec_test.go
+++ b/util/rowcodec/rowcodec_test.go
@@ -1,4 +1,4 @@
-// Copyright 2018 PingCAP, Inc.
+// Copyright 2019 PingCAP, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

From a7a381893974104ca768d6be5f51c8bf8a7c0037 Mon Sep 17 00:00:00 2001
From: Ewan Chou <coocood@gmail.com>
Date: Fri, 9 Aug 2019 18:52:08 +0800
Subject: [PATCH 4/7] address comment

---
 util/rowcodec/common.go        | 15 ++++++++-------
 util/rowcodec/decoder.go       |  4 ++--
 util/rowcodec/encoder.go       | 24 ++++++++++++------------
 util/rowcodec/rowcodec_test.go |  1 +
 4 files changed, 23 insertions(+), 21 deletions(-)

diff --git a/util/rowcodec/common.go b/util/rowcodec/common.go
index 47ca277ac7744..3ce6f9621755b 100644
--- a/util/rowcodec/common.go
+++ b/util/rowcodec/common.go
@@ -26,12 +26,13 @@ const CodecVer = 128
 
 var errInvalidCodecVer = errors.New("invalid codec version")
 
-// row is the struct type used to access the a row.
+// row is the struct type used to access a row.
 // There are two types of row, small and large.
-// small rows use one byte colID and two bytes offset, optimized for most cases.
-// large rows use four bytes colID and four bytes offset.
+// A small row takes one byte colID and two bytes offset, optimized for most cases.
+// If the max colID is larger than 255 or total value size is larger than 65535, the row type would be large.
+// A large row takes four bytes colID and four bytes offset.
 type row struct {
-	large          bool
+	isLarge        bool
 	numNotNullCols uint16
 	numNullCols    uint16
 	data           []byte
@@ -47,7 +48,7 @@ type row struct {
 
 func (r *row) getData(i int) []byte {
 	var start, end uint32
-	if r.large {
+	if r.isLarge {
 		if i > 0 {
 			start = r.offsets32[i-1]
 		}
@@ -65,11 +66,11 @@ func (r *row) setRowData(rowData []byte) error {
 	if rowData[0] != CodecVer {
 		return errInvalidCodecVer
 	}
-	r.large = rowData[1]&1 > 0
+	r.isLarge = rowData[1]&1 > 0
 	r.numNotNullCols = binary.LittleEndian.Uint16(rowData[2:])
 	r.numNullCols = binary.LittleEndian.Uint16(rowData[4:])
 	cursor := 6
-	if r.large {
+	if r.isLarge {
 		colIDsLen := int(r.numNotNullCols+r.numNullCols) * 4
 		r.colIDs32 = bytesToU32Slice(rowData[cursor : cursor+colIDsLen])
 		cursor += colIDsLen
diff --git a/util/rowcodec/decoder.go b/util/rowcodec/decoder.go
index 0acd8088d3db6..f452b0deb2f63 100644
--- a/util/rowcodec/decoder.go
+++ b/util/rowcodec/decoder.go
@@ -89,7 +89,7 @@ func (decoder *Decoder) Decode(rowData []byte, handle int64, chk *chunk.Chunk) e
 			h := int(uint(i+j) >> 1) // avoid overflow when computing h
 			// i ≤ h < j
 			var v int64
-			if decoder.large {
+			if decoder.isLarge {
 				v = int64(decoder.colIDs32[h])
 			} else {
 				v = int64(decoder.colIDs[h])
@@ -117,7 +117,7 @@ func (decoder *Decoder) Decode(rowData []byte, handle int64, chk *chunk.Chunk) e
 			h := int(uint(i+j) >> 1) // avoid overflow when computing h
 			// i ≤ h < j
 			var v int64
-			if decoder.large {
+			if decoder.isLarge {
 				v = int64(decoder.colIDs32[h])
 			} else {
 				v = int64(decoder.colIDs[h])
diff --git a/util/rowcodec/encoder.go b/util/rowcodec/encoder.go
index ed9a9d1d7c493..c977c0d702722 100644
--- a/util/rowcodec/encoder.go
+++ b/util/rowcodec/encoder.go
@@ -36,7 +36,7 @@ type Encoder struct {
 }
 
 func (encoder *Encoder) reset() {
-	encoder.large = false
+	encoder.isLarge = false
 	encoder.numNotNullCols = 0
 	encoder.numNullCols = 0
 	encoder.data = encoder.data[:0]
@@ -46,7 +46,7 @@ func (encoder *Encoder) reset() {
 
 func (encoder *Encoder) addColumn(colID int64, d types.Datum) {
 	if colID > 255 {
-		encoder.large = true
+		encoder.isLarge = true
 	}
 	if d.IsNull() {
 		encoder.numNullCols++
@@ -63,7 +63,7 @@ func (encoder *Encoder) Encode(colIDs []int64, values []types.Datum, buf []byte)
 	for i, colID := range colIDs {
 		encoder.addColumn(colID, values[i])
 	}
-	return encoder.build(buf[:0])
+	return encoder.build(buf)
 }
 
 func (encoder *Encoder) build(buf []byte) ([]byte, error) {
@@ -72,7 +72,7 @@ func (encoder *Encoder) build(buf []byte) ([]byte, error) {
 	numCols := len(encoder.tempColIDs)
 	nullIdx := numCols - int(r.numNullCols)
 	notNullIdx := 0
-	if r.large {
+	if r.isLarge {
 		encoder.initColIDs32()
 		encoder.initOffsets32()
 	} else {
@@ -81,14 +81,14 @@ func (encoder *Encoder) build(buf []byte) ([]byte, error) {
 	}
 	for i, colID := range encoder.tempColIDs {
 		if encoder.values[i].IsNull() {
-			if r.large {
+			if r.isLarge {
 				r.colIDs32[nullIdx] = uint32(colID)
 			} else {
 				r.colIDs[nullIdx] = byte(colID)
 			}
 			nullIdx++
 		} else {
-			if r.large {
+			if r.isLarge {
 				r.colIDs32[notNullIdx] = uint32(colID)
 			} else {
 				r.colIDs[notNullIdx] = byte(colID)
@@ -97,7 +97,7 @@ func (encoder *Encoder) build(buf []byte) ([]byte, error) {
 			notNullIdx++
 		}
 	}
-	if r.large {
+	if r.isLarge {
 		largeNotNullSorter := (*largeNotNullSorter)(encoder)
 		sort.Sort(largeNotNullSorter)
 		if r.numNullCols > 0 {
@@ -118,7 +118,7 @@ func (encoder *Encoder) build(buf []byte) ([]byte, error) {
 		if err != nil {
 			return nil, errors.Trace(err)
 		}
-		if len(r.data) > math.MaxUint16 && !r.large {
+		if len(r.data) > math.MaxUint16 && !r.isLarge {
 			// We need to convert the row to large row.
 			encoder.initColIDs32()
 			for j := 0; j < numCols; j++ {
@@ -128,9 +128,9 @@ func (encoder *Encoder) build(buf []byte) ([]byte, error) {
 			for j := 0; j <= i; j++ {
 				r.offsets32[j] = uint32(r.offsets[j])
 			}
-			r.large = true
+			r.isLarge = true
 		}
-		if r.large {
+		if r.isLarge {
 			r.offsets32[i] = uint32(len(r.data))
 		} else {
 			r.offsets[i] = uint16(len(r.data))
@@ -138,13 +138,13 @@ func (encoder *Encoder) build(buf []byte) ([]byte, error) {
 	}
 	buf = append(buf, CodecVer)
 	flag := byte(0)
-	if r.large {
+	if r.isLarge {
 		flag = 1
 	}
 	buf = append(buf, flag)
 	buf = append(buf, byte(r.numNotNullCols), byte(r.numNotNullCols>>8))
 	buf = append(buf, byte(r.numNullCols), byte(r.numNullCols>>8))
-	if r.large {
+	if r.isLarge {
 		buf = append(buf, u32SliceToBytes(r.colIDs32)...)
 		buf = append(buf, u32SliceToBytes(r.offsets32)...)
 	} else {
diff --git a/util/rowcodec/rowcodec_test.go b/util/rowcodec/rowcodec_test.go
index aa14e2fa068f8..2c4fde4392e0a 100644
--- a/util/rowcodec/rowcodec_test.go
+++ b/util/rowcodec/rowcodec_test.go
@@ -165,6 +165,7 @@ func BenchmarkEncode(b *testing.B) {
 	colIDs := []int64{1, 2, 3}
 	var err error
 	for i := 0; i < b.N; i++ {
+		buf = buf[:0]
 		buf, err = xb.Encode(colIDs, oldRow, buf)
 		if err != nil {
 			b.Fatal(err)

From 6f2f53ec100bce4e9e048397d39c781f2503b9e3 Mon Sep 17 00:00:00 2001
From: Ewan Chou <coocood@gmail.com>
Date: Mon, 12 Aug 2019 15:44:07 +0800
Subject: [PATCH 5/7] address comment

---
 util/rowcodec/encoder.go | 24 ++++++++++--------------
 1 file changed, 10 insertions(+), 14 deletions(-)

diff --git a/util/rowcodec/encoder.go b/util/rowcodec/encoder.go
index c977c0d702722..e3afa78c83e33 100644
--- a/util/rowcodec/encoder.go
+++ b/util/rowcodec/encoder.go
@@ -44,24 +44,20 @@ func (encoder *Encoder) reset() {
 	encoder.values = encoder.values[:0]
 }
 
-func (encoder *Encoder) addColumn(colID int64, d types.Datum) {
-	if colID > 255 {
-		encoder.isLarge = true
-	}
-	if d.IsNull() {
-		encoder.numNullCols++
-	} else {
-		encoder.numNotNullCols++
-	}
-	encoder.tempColIDs = append(encoder.tempColIDs, colID)
-	encoder.values = append(encoder.values, d)
-}
-
 // Encode encodes a row from a datums slice.
 func (encoder *Encoder) Encode(colIDs []int64, values []types.Datum, buf []byte) ([]byte, error) {
 	encoder.reset()
+	encoder.tempColIDs = append(encoder.tempColIDs, colIDs...)
+	encoder.values = append(encoder.values, values...)
 	for i, colID := range colIDs {
-		encoder.addColumn(colID, values[i])
+		if colID > 255 {
+			encoder.isLarge = true
+		}
+		if values[i].IsNull() {
+			encoder.numNullCols++
+		} else {
+			encoder.numNotNullCols++
+		}
 	}
 	return encoder.build(buf)
 }

From 1caad4ff860c7dc6252a76c22780f061054f92fb Mon Sep 17 00:00:00 2001
From: Ewan Chou <coocood@gmail.com>
Date: Mon, 12 Aug 2019 16:08:38 +0800
Subject: [PATCH 6/7] add NewEncoder function

---
 util/rowcodec/encoder.go       | 14 ++++++++++----
 util/rowcodec/rowcodec_test.go | 27 ++++++++++++---------------
 2 files changed, 22 insertions(+), 19 deletions(-)

diff --git a/util/rowcodec/encoder.go b/util/rowcodec/encoder.go
index e3afa78c83e33..4b5cb2e46144e 100644
--- a/util/rowcodec/encoder.go
+++ b/util/rowcodec/encoder.go
@@ -35,21 +35,27 @@ type Encoder struct {
 	sc         *stmtctx.StatementContext
 }
 
+// NewEncoder creates a new Encoder with column IDs.
+func NewEncoder(colIDs []int64, sc *stmtctx.StatementContext) *Encoder {
+	return &Encoder{
+		tempColIDs: colIDs,
+		sc:         sc,
+	}
+}
+
 func (encoder *Encoder) reset() {
 	encoder.isLarge = false
 	encoder.numNotNullCols = 0
 	encoder.numNullCols = 0
 	encoder.data = encoder.data[:0]
-	encoder.tempColIDs = encoder.tempColIDs[:0]
 	encoder.values = encoder.values[:0]
 }
 
 // Encode encodes a row from a datums slice.
-func (encoder *Encoder) Encode(colIDs []int64, values []types.Datum, buf []byte) ([]byte, error) {
+func (encoder *Encoder) Encode(values []types.Datum, buf []byte) ([]byte, error) {
 	encoder.reset()
-	encoder.tempColIDs = append(encoder.tempColIDs, colIDs...)
 	encoder.values = append(encoder.values, values...)
-	for i, colID := range colIDs {
+	for i, colID := range encoder.tempColIDs {
 		if colID > 255 {
 			encoder.isLarge = true
 		}
diff --git a/util/rowcodec/rowcodec_test.go b/util/rowcodec/rowcodec_test.go
index 2c4fde4392e0a..dfa97fea3acee 100644
--- a/util/rowcodec/rowcodec_test.go
+++ b/util/rowcodec/rowcodec_test.go
@@ -37,8 +37,7 @@ type testSuite struct{}
 
 func (s *testSuite) TestRowCodec(c *C) {
 	colIDs := []int64{1, 2, 3, 4, 5, 10}
-	var rb Encoder
-	rb.sc = new(stmtctx.StatementContext)
+	rb := NewEncoder(colIDs, new(stmtctx.StatementContext))
 	dt, err := types.ParseDatetime(rb.sc, "2018-01-19 03:14:07.999999")
 	c.Assert(err, IsNil)
 	datums := types.MakeDatums(
@@ -49,20 +48,20 @@ func (s *testSuite) TestRowCodec(c *C) {
 		types.NewDecFromInt(1),
 		"abc",
 	)
-	newRow, err := rb.Encode(colIDs, datums, nil)
+	newRow, err := rb.Encode(datums, nil)
 	c.Check(err, IsNil)
 	s.checkDecode(c, rb.sc, newRow)
 
 	// Test large column ID
 	colIDs = []int64{1, 2, 3, 4, 5, 512}
-	newRow, err = rb.Encode(colIDs, datums, nil)
+	newRow, err = rb.Encode(datums, nil)
 	c.Check(err, IsNil)
 	s.checkDecode(c, rb.sc, newRow)
 
 	// Test large column value
 	colIDs = []int64{1, 2, 3, 4, 5, 10}
 	datums[5] = types.NewBytesDatum(make([]byte, 65536))
-	newRow, err = rb.Encode(colIDs, datums, nil)
+	newRow, err = rb.Encode(datums, nil)
 	c.Check(err, IsNil)
 	s.checkDecode(c, rb.sc, newRow)
 }
@@ -86,9 +85,9 @@ func (s *testSuite) TestIntCodec(c *C) {
 
 func (s *testSuite) TestMoreTypes(c *C) {
 	colIDs := []int64{1, 2, 3, 4, 5, 6, 7, 8}
-	var rb Encoder
-	rb.sc = new(stmtctx.StatementContext)
-	rb.sc.TimeZone = time.Local
+	sc := new(stmtctx.StatementContext)
+	sc.TimeZone = time.Local
+	rb := NewEncoder(colIDs, sc)
 	ts, err := types.ParseTimestampFromNum(rb.sc, 20181111090909)
 	c.Assert(err, IsNil)
 	datums := types.MakeDatums(
@@ -101,7 +100,7 @@ func (s *testSuite) TestMoreTypes(c *C) {
 		json.CreateBinary("abc"),
 		types.BitLiteral([]byte{101}),
 	)
-	newRow, err := rb.Encode(colIDs, datums, nil)
+	newRow, err := rb.Encode(datums, nil)
 	c.Check(err, IsNil)
 	fieldTypes := []*types.FieldType{
 		types.NewFieldType(mysql.TypeFloat),
@@ -160,13 +159,12 @@ func (s *testSuite) checkDecode(c *C, sc *stmtctx.StatementContext, newRow []byt
 func BenchmarkEncode(b *testing.B) {
 	b.ReportAllocs()
 	oldRow := types.MakeDatums(1, "abc", 1.1)
-	var xb Encoder
+	xb := NewEncoder([]int64{1, 2, 3}, new(stmtctx.StatementContext))
 	var buf []byte
-	colIDs := []int64{1, 2, 3}
 	var err error
 	for i := 0; i < b.N; i++ {
 		buf = buf[:0]
-		buf, err = xb.Encode(colIDs, oldRow, buf)
+		buf, err = xb.Encode(oldRow, buf)
 		if err != nil {
 			b.Fatal(err)
 		}
@@ -182,9 +180,8 @@ func BenchmarkDecode(b *testing.B) {
 		types.NewFieldType(mysql.TypeString),
 		types.NewFieldType(mysql.TypeDouble),
 	}
-	var xb Encoder
-	xb.sc = new(stmtctx.StatementContext)
-	xRowData, err := xb.Encode(colIDs, oldRow, nil)
+	xb := NewEncoder(colIDs, new(stmtctx.StatementContext))
+	xRowData, err := xb.Encode(oldRow, nil)
 	if err != nil {
 		b.Fatal(err)
 	}

From 583f1b87b6e080853e7736001b63b9481c95f79a Mon Sep 17 00:00:00 2001
From: Ewan Chou <coocood@gmail.com>
Date: Mon, 12 Aug 2019 16:25:35 +0800
Subject: [PATCH 7/7] fix ci

---
 util/rowcodec/rowcodec_test.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/util/rowcodec/rowcodec_test.go b/util/rowcodec/rowcodec_test.go
index dfa97fea3acee..4324404b3225a 100644
--- a/util/rowcodec/rowcodec_test.go
+++ b/util/rowcodec/rowcodec_test.go
@@ -53,13 +53,13 @@ func (s *testSuite) TestRowCodec(c *C) {
 	s.checkDecode(c, rb.sc, newRow)
 
 	// Test large column ID
-	colIDs = []int64{1, 2, 3, 4, 5, 512}
+	rb.tempColIDs = []int64{1, 2, 3, 4, 5, 512}
 	newRow, err = rb.Encode(datums, nil)
 	c.Check(err, IsNil)
 	s.checkDecode(c, rb.sc, newRow)
 
 	// Test large column value
-	colIDs = []int64{1, 2, 3, 4, 5, 10}
+	rb.tempColIDs = []int64{1, 2, 3, 4, 5, 10}
 	datums[5] = types.NewBytesDatum(make([]byte, 65536))
 	newRow, err = rb.Encode(datums, nil)
 	c.Check(err, IsNil)