diff --git a/util/rowcodec/common.go b/util/rowcodec/common.go
new file mode 100644
index 0000000000000..3ce6f9621755b
--- /dev/null
+++ b/util/rowcodec/common.go
@@ -0,0 +1,198 @@
+// 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.
+// 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 a row.
+// There are two types of row, small and large.
+// 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 {
+	isLarge        bool
+	numNotNullCols uint16
+	numNullCols    uint16
+	data           []byte
+
+	// for small rows
+	colIDs  []byte
+	offsets []uint16
+
+	// for large row
+	colIDs32  []uint32
+	offsets32 []uint32
+}
+
+func (r *row) getData(i int) []byte {
+	var start, end uint32
+	if r.isLarge {
+		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.isLarge = rowData[1]&1 > 0
+	r.numNotNullCols = binary.LittleEndian.Uint16(rowData[2:])
+	r.numNullCols = binary.LittleEndian.Uint16(rowData[4:])
+	cursor := 6
+	if r.isLarge {
+		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..f452b0deb2f63
--- /dev/null
+++ b/util/rowcodec/decoder.go
@@ -0,0 +1,213 @@
+// 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.
+// 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.isLarge {
+				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.isLarge {
+				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..4b5cb2e46144e
--- /dev/null
+++ b/util/rowcodec/encoder.go
@@ -0,0 +1,316 @@
+// 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.
+// 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
+}
+
+// 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.values = encoder.values[:0]
+}
+
+// Encode encodes a row from a datums slice.
+func (encoder *Encoder) Encode(values []types.Datum, buf []byte) ([]byte, error) {
+	encoder.reset()
+	encoder.values = append(encoder.values, values...)
+	for i, colID := range encoder.tempColIDs {
+		if colID > 255 {
+			encoder.isLarge = true
+		}
+		if values[i].IsNull() {
+			encoder.numNullCols++
+		} else {
+			encoder.numNotNullCols++
+		}
+	}
+	return encoder.build(buf)
+}
+
+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.isLarge {
+		encoder.initColIDs32()
+		encoder.initOffsets32()
+	} else {
+		encoder.initColIDs()
+		encoder.initOffsets()
+	}
+	for i, colID := range encoder.tempColIDs {
+		if encoder.values[i].IsNull() {
+			if r.isLarge {
+				r.colIDs32[nullIdx] = uint32(colID)
+			} else {
+				r.colIDs[nullIdx] = byte(colID)
+			}
+			nullIdx++
+		} else {
+			if r.isLarge {
+				r.colIDs32[notNullIdx] = uint32(colID)
+			} else {
+				r.colIDs[notNullIdx] = byte(colID)
+			}
+			encoder.values[notNullIdx] = encoder.values[i]
+			notNullIdx++
+		}
+	}
+	if r.isLarge {
+		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.isLarge {
+			// 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.isLarge = true
+		}
+		if r.isLarge {
+			r.offsets32[i] = uint32(len(r.data))
+		} else {
+			r.offsets[i] = uint16(len(r.data))
+		}
+	}
+	buf = append(buf, CodecVer)
+	flag := byte(0)
+	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.isLarge {
+		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..4324404b3225a
--- /dev/null
+++ b/util/rowcodec/rowcodec_test.go
@@ -0,0 +1,200 @@
+// 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.
+// 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}
+	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(
+		dt,
+		"abc",
+		nil,
+		1,
+		types.NewDecFromInt(1),
+		"abc",
+	)
+	newRow, err := rb.Encode(datums, nil)
+	c.Check(err, IsNil)
+	s.checkDecode(c, rb.sc, newRow)
+
+	// Test large column ID
+	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
+	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)
+	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}
+	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(
+		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(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))
+	c.Assert(err, IsNil)
+	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)
+	xb := NewEncoder([]int64{1, 2, 3}, new(stmtctx.StatementContext))
+	var buf []byte
+	var err error
+	for i := 0; i < b.N; i++ {
+		buf = buf[:0]
+		buf, err = xb.Encode(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),
+	}
+	xb := NewEncoder(colIDs, new(stmtctx.StatementContext))
+	xRowData, err := xb.Encode(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)
+		}
+	}
+}