diff --git a/.gitattributes b/.gitattributes
new file mode 100644
index 0000000000000..1ad886cf2491c
--- /dev/null
+++ b/.gitattributes
@@ -0,0 +1,5 @@
+# Set the default behavior, in case people don't have core.autocrlf set.
+* text=auto
+
+# Declare files that will always have LF line endings on checkout.
+*.y text eol=lf
diff --git a/ddl/ddl.go b/ddl/ddl.go
index 280629d9baeab..21af80eca9649 100644
--- a/ddl/ddl.go
+++ b/ddl/ddl.go
@@ -188,6 +188,7 @@ var (
 	ErrPartitionFuncNotAllowed = terror.ClassDDL.New(codeErrPartitionFuncNotAllowed, mysql.MySQLErrName[mysql.ErrPartitionFuncNotAllowed])
 	// ErrUniqueKeyNeedAllFieldsInPf returns must include all columns in the table's partitioning function.
 	ErrUniqueKeyNeedAllFieldsInPf = terror.ClassDDL.New(codeUniqueKeyNeedAllFieldsInPf, mysql.MySQLErrName[mysql.ErrUniqueKeyNeedAllFieldsInPf])
+	errWrongExprInPartitionFunc   = terror.ClassDDL.New(codeWrongExprInPartitionFunc, mysql.MySQLErrName[mysql.ErrWrongExprInPartitionFunc])
 )
 
 // DDL is responsible for updating schema in data store and maintaining in-memory InfoSchema cache.
@@ -606,6 +607,7 @@ const (
 	codeErrFieldTypeNotAllowedAsPartitionField = terror.ErrCode(mysql.ErrFieldTypeNotAllowedAsPartitionField)
 	codeUniqueKeyNeedAllFieldsInPf             = terror.ErrCode(mysql.ErrUniqueKeyNeedAllFieldsInPf)
 	codePrimaryCantHaveNull                    = terror.ErrCode(mysql.ErrPrimaryCantHaveNull)
+	codeWrongExprInPartitionFunc               = terror.ErrCode(mysql.ErrWrongExprInPartitionFunc)
 )
 
 func init() {
@@ -652,6 +654,7 @@ func init() {
 		codeErrFieldTypeNotAllowedAsPartitionField: mysql.ErrFieldTypeNotAllowedAsPartitionField,
 		codeUniqueKeyNeedAllFieldsInPf:             mysql.ErrUniqueKeyNeedAllFieldsInPf,
 		codePrimaryCantHaveNull:                    mysql.ErrPrimaryCantHaveNull,
+		codeWrongExprInPartitionFunc:               mysql.ErrWrongExprInPartitionFunc,
 	}
 	terror.ErrClassToMySQLCodes[terror.ClassDDL] = ddlMySQLErrCodes
 }
diff --git a/ddl/ddl_api.go b/ddl/ddl_api.go
index af41a36617b11..2ab73110f6a9f 100644
--- a/ddl/ddl_api.go
+++ b/ddl/ddl_api.go
@@ -912,7 +912,7 @@ func (d *ddl) CreateTable(ctx sessionctx.Context, s *ast.CreateTableStmt) (err e
 			return errors.Trace(err)
 		}
 
-		if err = checkPartitionFuncValid(s.Partition.Expr); err != nil {
+		if err = checkPartitionFuncValid(ctx, tbInfo, s.Partition.Expr); err != nil {
 			return errors.Trace(err)
 		}
 
diff --git a/ddl/partition.go b/ddl/partition.go
index 7d3f4ac8865f7..82d88439f479c 100644
--- a/ddl/partition.go
+++ b/ddl/partition.go
@@ -24,6 +24,7 @@ import (
 	"github.com/pingcap/tidb/expression"
 	"github.com/pingcap/tidb/meta"
 	"github.com/pingcap/tidb/model"
+	"github.com/pingcap/tidb/mysql"
 	"github.com/pingcap/tidb/parser/opcode"
 	"github.com/pingcap/tidb/sessionctx"
 	"github.com/pingcap/tidb/table"
@@ -103,28 +104,43 @@ func checkPartitionNameUnique(tbInfo *model.TableInfo, pi *model.PartitionInfo)
 }
 
 // checkPartitionFuncValid checks partition function validly.
-func checkPartitionFuncValid(expr ast.ExprNode) error {
+func checkPartitionFuncValid(ctx sessionctx.Context, tblInfo *model.TableInfo, expr ast.ExprNode) error {
 	switch v := expr.(type) {
-	case *ast.CaseExpr:
-		return ErrPartitionFunctionIsNotAllowed
+	case *ast.FuncCastExpr, *ast.CaseExpr:
+		return errors.Trace(ErrPartitionFunctionIsNotAllowed)
 	case *ast.FuncCallExpr:
 		// check function which allowed in partitioning expressions
 		// see https://dev.mysql.com/doc/mysql-partitioning-excerpt/5.7/en/partitioning-limitations-functions.html
 		switch v.FnName.L {
 		case ast.Abs, ast.Ceiling, ast.DateDiff, ast.Day, ast.DayOfMonth, ast.DayOfWeek, ast.DayOfYear, ast.Extract, ast.Floor,
 			ast.Hour, ast.MicroSecond, ast.Minute, ast.Mod, ast.Month, ast.Quarter, ast.Second, ast.TimeToSec, ast.ToDays,
-			ast.ToSeconds, ast.UnixTimestamp, ast.Weekday, ast.Year, ast.YearWeek:
+			ast.ToSeconds, ast.Weekday, ast.Year, ast.YearWeek:
 			return nil
-		default:
-			return ErrPartitionFunctionIsNotAllowed
+		case ast.UnixTimestamp:
+			if len(v.Args) == 1 {
+				col, err := expression.RewriteSimpleExprWithTableInfo(ctx, tblInfo, v.Args[0])
+				if err != nil {
+					return errors.Trace(err)
+				}
+				if col.GetType().Tp != mysql.TypeTimestamp {
+					return errors.Trace(errWrongExprInPartitionFunc)
+				}
+				return nil
+			}
 		}
+		return errors.Trace(ErrPartitionFunctionIsNotAllowed)
 	case *ast.BinaryOperationExpr:
 		// The DIV operator (opcode.IntDiv) is also supported; the / operator ( opcode.Div ) is not permitted.
 		// see https://dev.mysql.com/doc/refman/5.7/en/partitioning-limitations.html
-		if v.Op == opcode.Div {
-			return ErrPartitionFunctionIsNotAllowed
+		switch v.Op {
+		case opcode.Or, opcode.And, opcode.Xor, opcode.LeftShift, opcode.RightShift, opcode.BitNeg, opcode.Div:
+			return errors.Trace(ErrPartitionFunctionIsNotAllowed)
 		}
 		return nil
+	case *ast.UnaryOperationExpr:
+		if v.Op == opcode.BitNeg {
+			return errors.Trace(ErrPartitionFunctionIsNotAllowed)
+		}
 	}
 	return nil
 }
diff --git a/docs/design/2018-07-01-refactor-aggregate-framework.md b/docs/design/2018-07-01-refactor-aggregate-framework.md
index 9d49668f8f2eb..b3dbdef80ae19 100644
--- a/docs/design/2018-07-01-refactor-aggregate-framework.md
+++ b/docs/design/2018-07-01-refactor-aggregate-framework.md
@@ -31,6 +31,7 @@ The `AggFunc` interface contains the following functions:
 - `ResetPartialResult` resets the partial result to the original state for a specific aggregate function. It converts the input `PartialResult` to the specific data structure which stores the partial result and then resets every field to the proper original state.
 - `UpdatePartialResult` updates the specific partial result for an aggregate function using the input rows which all belong to the same data group. It converts the `PartialResult` to the specific data structure which stores the partial result and then iterates on the input rows and updates that partial result according to the functionality and the state of the aggregate function.
 - `AppendFinalResult2Chunk` finalizes the partial result and appends the final result directly to the input `Chunk`. Like other operations, it converts the input `PartialResult` to the specific data structure firstly, calculates the final result and appends that final result to the `Chunk` provided.
+- `MergePartialResult` evaluates the final result using the input `PartialResults`. Suppose the input `PartialResults` names are `dst` and `src` respectively. It converts `dst` and `src` to the same data structure firstly, merges the partial results and store the result in `dst`. 
 
 The new framework uses the `Build()` function to build an executable aggregate function. Its input parameters are:
 - `aggFuncDesc`: the aggregate function representation used by the planner layer.
diff --git a/docs/design/README.md b/docs/design/README.md
index 0b67696f6a8fb..36639dd95e2a3 100644
--- a/docs/design/README.md
+++ b/docs/design/README.md
@@ -4,8 +4,9 @@ A proposal template: [TEMPLATE.md](./TEMPLATE.md)
 
 ## Proposed
 
-## In Progress
-
 - [Proposal: A new aggregate function execution framework](./2018-07-01-refactor-aggregate-framework.md)
+- [Proposal: A new storage row format for efficient decoding](./2018-07-19-row-format.md)
+
+## In Progress
 
 ## Completed
diff --git a/expression/builtin_cast.go b/expression/builtin_cast.go
index fe0694293e46b..50b88e0bff3dd 100644
--- a/expression/builtin_cast.go
+++ b/expression/builtin_cast.go
@@ -1633,11 +1633,13 @@ func (i inCastContext) String() string {
 	return "__cast_ctx"
 }
 
-// inUnionCastContext is session key value that indicates whether executing in union cast context.
+// inUnionCastContext is session key value that indicates whether executing in
+// union cast context.
 // @see BuildCastFunction4Union
 const inUnionCastContext inCastContext = 0
 
-// BuildCastFunction4Union build a implicitly CAST ScalarFunction from the Union Expression.
+// BuildCastFunction4Union build a implicitly CAST ScalarFunction from the Union
+// Expression.
 func BuildCastFunction4Union(ctx sessionctx.Context, expr Expression, tp *types.FieldType) (res Expression) {
 	ctx.SetValue(inUnionCastContext, struct{}{})
 	defer func() {
@@ -1673,17 +1675,16 @@ func BuildCastFunction(ctx sessionctx.Context, expr Expression, tp *types.FieldT
 		Function: f,
 	}
 	// We do not fold CAST if the eval type of this scalar function is ETJson
-	// since we may reset the flag of the field type of CastAsJson later which would
-	// affect the evaluation of it.
+	// since we may reset the flag of the field type of CastAsJson later which
+	// would affect the evaluation of it.
 	if tp.EvalType() != types.ETJson {
 		res = FoldConstant(res)
 	}
 	return res
 }
 
-// WrapWithCastAsInt wraps `expr` with `cast` if the return type
-// of expr is not type int,
-// otherwise, returns `expr` directly.
+// WrapWithCastAsInt wraps `expr` with `cast` if the return type of expr is not
+// type int, otherwise, returns `expr` directly.
 func WrapWithCastAsInt(ctx sessionctx.Context, expr Expression) Expression {
 	if expr.GetType().EvalType() == types.ETInt {
 		return expr
@@ -1695,9 +1696,8 @@ func WrapWithCastAsInt(ctx sessionctx.Context, expr Expression) Expression {
 	return BuildCastFunction(ctx, expr, tp)
 }
 
-// WrapWithCastAsReal wraps `expr` with `cast` if the return type
-// of expr is not type real,
-// otherwise, returns `expr` directly.
+// WrapWithCastAsReal wraps `expr` with `cast` if the return type of expr is not
+// type real, otherwise, returns `expr` directly.
 func WrapWithCastAsReal(ctx sessionctx.Context, expr Expression) Expression {
 	if expr.GetType().EvalType() == types.ETReal {
 		return expr
@@ -1709,9 +1709,8 @@ func WrapWithCastAsReal(ctx sessionctx.Context, expr Expression) Expression {
 	return BuildCastFunction(ctx, expr, tp)
 }
 
-// WrapWithCastAsDecimal wraps `expr` with `cast` if the return type
-// of expr is not type decimal,
-// otherwise, returns `expr` directly.
+// WrapWithCastAsDecimal wraps `expr` with `cast` if the return type of expr is
+// not type decimal, otherwise, returns `expr` directly.
 func WrapWithCastAsDecimal(ctx sessionctx.Context, expr Expression) Expression {
 	if expr.GetType().EvalType() == types.ETDecimal {
 		return expr
@@ -1723,15 +1722,22 @@ func WrapWithCastAsDecimal(ctx sessionctx.Context, expr Expression) Expression {
 	return BuildCastFunction(ctx, expr, tp)
 }
 
-// WrapWithCastAsString wraps `expr` with `cast` if the return type
-// of expr is not type string,
-// otherwise, returns `expr` directly.
+// WrapWithCastAsString wraps `expr` with `cast` if the return type of expr is
+// not type string, otherwise, returns `expr` directly.
 func WrapWithCastAsString(ctx sessionctx.Context, expr Expression) Expression {
-	if expr.GetType().EvalType() == types.ETString {
+	exprTp := expr.GetType()
+	if exprTp.EvalType() == types.ETString {
 		return expr
 	}
-	argLen := expr.GetType().Flen
-	if expr.GetType().EvalType() == types.ETInt {
+	argLen := exprTp.Flen
+	// If expr is decimal, we should take the decimal point and negative sign
+	// into consideration, so we set `expr.GetType().Flen + 2` as the `argLen`.
+	// Since the length of float and double is not accurate, we do not handle
+	// them.
+	if exprTp.Tp == mysql.TypeNewDecimal && argLen != types.UnspecifiedFsp {
+		argLen += 2
+	}
+	if exprTp.EvalType() == types.ETInt {
 		argLen = mysql.MaxIntWidth
 	}
 	tp := types.NewFieldType(mysql.TypeVarString)
@@ -1740,9 +1746,8 @@ func WrapWithCastAsString(ctx sessionctx.Context, expr Expression) Expression {
 	return BuildCastFunction(ctx, expr, tp)
 }
 
-// WrapWithCastAsTime wraps `expr` with `cast` if the return type
-// of expr is not same as type of the specified `tp` ,
-// otherwise, returns `expr` directly.
+// WrapWithCastAsTime wraps `expr` with `cast` if the return type of expr is not
+// same as type of the specified `tp` , otherwise, returns `expr` directly.
 func WrapWithCastAsTime(ctx sessionctx.Context, expr Expression, tp *types.FieldType) Expression {
 	exprTp := expr.GetType().Tp
 	if tp.Tp == exprTp {
@@ -1769,9 +1774,8 @@ func WrapWithCastAsTime(ctx sessionctx.Context, expr Expression, tp *types.Field
 	return BuildCastFunction(ctx, expr, tp)
 }
 
-// WrapWithCastAsDuration wraps `expr` with `cast` if the return type
-// of expr is not type duration,
-// otherwise, returns `expr` directly.
+// WrapWithCastAsDuration wraps `expr` with `cast` if the return type of expr is
+// not type duration, otherwise, returns `expr` directly.
 func WrapWithCastAsDuration(ctx sessionctx.Context, expr Expression) Expression {
 	if expr.GetType().Tp == mysql.TypeDuration {
 		return expr
@@ -1790,9 +1794,8 @@ func WrapWithCastAsDuration(ctx sessionctx.Context, expr Expression) Expression
 	return BuildCastFunction(ctx, expr, tp)
 }
 
-// WrapWithCastAsJSON wraps `expr` with `cast` if the return type
-// of expr is not type json,
-// otherwise, returns `expr` directly.
+// WrapWithCastAsJSON wraps `expr` with `cast` if the return type of expr is not
+// type json, otherwise, returns `expr` directly.
 func WrapWithCastAsJSON(ctx sessionctx.Context, expr Expression) Expression {
 	if expr.GetType().Tp == mysql.TypeJSON && !mysql.HasParseToJSONFlag(expr.GetType().Flag) {
 		return expr
diff --git a/expression/builtin_time.go b/expression/builtin_time.go
index 7d44e77746a80..5b0d76d7cf069 100644
--- a/expression/builtin_time.go
+++ b/expression/builtin_time.go
@@ -5410,7 +5410,7 @@ func (b *builtinLastDaySig) evalTime(row chunk.Row) (types.Time, bool, error) {
 // getExpressionFsp calculates the fsp from given expression.
 func getExpressionFsp(ctx sessionctx.Context, expression Expression) (int, error) {
 	constExp, isConstant := expression.(*Constant)
-	if isConstant && types.IsString(expression.GetType()) && !isTemporalColumn(expression) {
+	if isConstant && types.IsString(expression.GetType().Tp) && !isTemporalColumn(expression) {
 		str, isNil, err := constExp.EvalString(ctx, chunk.Row{})
 		if isNil || err != nil {
 			return 0, errors.Trace(err)
diff --git a/expression/integration_test.go b/expression/integration_test.go
index e57b5ef431b3f..20369d6038ab3 100644
--- a/expression/integration_test.go
+++ b/expression/integration_test.go
@@ -925,6 +925,13 @@ func (s *testIntegrationSuite) TestStringBuiltin(c *C) {
 	result.Check(testkit.Rows("2 0 3 0"))
 	result = tk.MustQuery(`select field("abc", "a", 1), field(1.3, "1.3", 1.5);`)
 	result.Check(testkit.Rows("1 1"))
+
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t(a decimal(11, 8), b decimal(11,8))")
+	tk.MustExec("insert into t values('114.57011441','38.04620115'), ('-38.04620119', '38.04620115');")
+	result = tk.MustQuery("select a,b,concat_ws(',',a,b) from t")
+	result.Check(testkit.Rows("114.57011441 38.04620115 114.57011441,38.04620115",
+		"-38.04620119 38.04620115 -38.04620119,38.04620115"))
 }
 
 func (s *testIntegrationSuite) TestEncryptionBuiltin(c *C) {
diff --git a/expression/typeinfer_test.go b/expression/typeinfer_test.go
index 20832ba88d373..2aac3d967a5da 100644
--- a/expression/typeinfer_test.go
+++ b/expression/typeinfer_test.go
@@ -403,7 +403,7 @@ func (s *testInferTypeSuite) createTestCase4StrFuncs() []typeInferTestCase {
 		{"reverse(c_bigint_d   )", mysql.TypeVarString, charset.CharsetUTF8, 0, 20, types.UnspecifiedLength},
 		{"reverse(c_float_d    )", mysql.TypeVarString, charset.CharsetUTF8, 0, 12, types.UnspecifiedLength},
 		{"reverse(c_double_d   )", mysql.TypeVarString, charset.CharsetUTF8, 0, 22, types.UnspecifiedLength},
-		{"reverse(c_decimal    )", mysql.TypeVarString, charset.CharsetUTF8, 0, 6, types.UnspecifiedLength},
+		{"reverse(c_decimal    )", mysql.TypeVarString, charset.CharsetUTF8, 0, 8, types.UnspecifiedLength},
 		{"reverse(c_char       )", mysql.TypeVarString, charset.CharsetUTF8, 0, 20, types.UnspecifiedLength},
 		{"reverse(c_varchar    )", mysql.TypeVarString, charset.CharsetUTF8, 0, 20, types.UnspecifiedLength},
 		{"reverse(c_text_d     )", mysql.TypeVarString, charset.CharsetUTF8, 0, 65535, types.UnspecifiedLength},
diff --git a/infoschema/tables.go b/infoschema/tables.go
index 3e5507a156c01..f0a88941c76a1 100644
--- a/infoschema/tables.go
+++ b/infoschema/tables.go
@@ -792,8 +792,12 @@ func dataForColumns(ctx sessionctx.Context, schemas []*model.DBInfo) [][]types.D
 func dataForColumnsInTable(schema *model.DBInfo, tbl *model.TableInfo) [][]types.Datum {
 	var rows [][]types.Datum
 	for i, col := range tbl.Columns {
+		var charMaxLen, charOctLen, numericPrecision, numericScale, datetimePrecision interface{}
 		colLen, decimal := col.Flen, col.Decimal
 		defaultFlen, defaultDecimal := mysql.GetDefaultFieldLengthAndDecimal(col.Tp)
+		if decimal == types.UnspecifiedLength {
+			decimal = defaultDecimal
+		}
 		if colLen == types.UnspecifiedLength {
 			colLen = defaultFlen
 		}
@@ -808,6 +812,8 @@ func dataForColumnsInTable(schema *model.DBInfo, tbl *model.TableInfo) [][]types
 			if len(col.Elems) != 0 {
 				colLen += (len(col.Elems) - 1)
 			}
+			charMaxLen = colLen
+			charOctLen = colLen
 		} else if col.Tp == mysql.TypeEnum {
 			// Example: In MySQL enum('a', 'ab', 'cdef') has length 4, because
 			// the longest string in the enum is 'cdef'
@@ -818,9 +824,16 @@ func dataForColumnsInTable(schema *model.DBInfo, tbl *model.TableInfo) [][]types
 					colLen = len(ele)
 				}
 			}
-		}
-		if decimal == types.UnspecifiedLength {
-			decimal = defaultDecimal
+			charMaxLen = colLen
+			charOctLen = colLen
+		} else if types.IsString(col.Tp) {
+			charMaxLen = colLen
+			charOctLen = colLen
+		} else if types.IsTypeFractionable(col.Tp) {
+			datetimePrecision = decimal
+		} else if types.IsTypeNumeric(col.Tp) {
+			numericPrecision = colLen
+			numericScale = decimal
 		}
 		columnType := col.FieldType.InfoSchemaStr()
 		columnDesc := table.NewColDesc(table.ToColumn(col))
@@ -837,19 +850,19 @@ func dataForColumnsInTable(schema *model.DBInfo, tbl *model.TableInfo) [][]types
 			columnDefault,                        // COLUMN_DEFAULT
 			columnDesc.Null,                      // IS_NULLABLE
 			types.TypeToStr(col.Tp, col.Charset), // DATA_TYPE
-			colLen,                            // CHARACTER_MAXIMUM_LENGTH
-			colLen,                            // CHARACTER_OCTET_LENGTH
-			decimal,                           // NUMERIC_PRECISION
-			0,                                 // NUMERIC_SCALE
-			0,                                 // DATETIME_PRECISION
-			col.Charset,                       // CHARACTER_SET_NAME
-			col.Collate,                       // COLLATION_NAME
-			columnType,                        // COLUMN_TYPE
-			columnDesc.Key,                    // COLUMN_KEY
-			columnDesc.Extra,                  // EXTRA
-			"select,insert,update,references", // PRIVILEGES
-			columnDesc.Comment,                // COLUMN_COMMENT
-			col.GeneratedExprString,           // GENERATION_EXPRESSION
+			charMaxLen,                           // CHARACTER_MAXIMUM_LENGTH
+			charOctLen,                           // CHARACTER_OCTET_LENGTH
+			numericPrecision,                     // NUMERIC_PRECISION
+			numericScale,                         // NUMERIC_SCALE
+			datetimePrecision,                    // DATETIME_PRECISION
+			col.Charset,                          // CHARACTER_SET_NAME
+			col.Collate,                          // COLLATION_NAME
+			columnType,                           // COLUMN_TYPE
+			columnDesc.Key,                       // COLUMN_KEY
+			columnDesc.Extra,                     // EXTRA
+			"select,insert,update,references",    // PRIVILEGES
+			columnDesc.Comment,                   // COLUMN_COMMENT
+			col.GeneratedExprString,              // GENERATION_EXPRESSION
 		)
 		// In mysql, 'character_set_name' and 'collation_name' are setted to null when column type is non-varchar or non-blob in information_schema.
 		if col.Tp != mysql.TypeVarchar && col.Tp != mysql.TypeBlob {
@@ -1180,7 +1193,7 @@ func (s schemasSorter) Less(i, j int) bool {
 }
 
 func (it *infoschemaTable) getRows(ctx sessionctx.Context, cols []*table.Column) (fullRows [][]types.Datum, err error) {
-	is := it.handle.Get()
+	is := ctx.GetSessionVars().TxnCtx.InfoSchema.(InfoSchema)
 	dbs := is.AllSchemas()
 	sort.Sort(schemasSorter(dbs))
 	switch it.meta.Name.O {
diff --git a/infoschema/tables_test.go b/infoschema/tables_test.go
index 6db86406e8a53..82f08876f5982 100644
--- a/infoschema/tables_test.go
+++ b/infoschema/tables_test.go
@@ -23,6 +23,32 @@ import (
 	"github.com/pingcap/tidb/util/testleak"
 )
 
+func (s *testSuite) TestInfoschemaFielValue(c *C) {
+	testleak.BeforeTest()
+	defer testleak.AfterTest(c)()
+	store, err := mockstore.NewMockTikvStore()
+	c.Assert(err, IsNil)
+	defer store.Close()
+	session.SetStatsLease(0)
+	do, err := session.BootstrapSession(store)
+	c.Assert(err, IsNil)
+	defer do.Close()
+
+	tk := testkit.NewTestKit(c, store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists numschema, timeschema")
+	tk.MustExec("create table numschema(i int(2), f float(4,2), d decimal(4,3))")
+	tk.MustExec("create table timeschema(d date, dt datetime(3), ts timestamp(3), t time(4), y year(4))")
+	tk.MustExec("create table strschema(c char(3), c2 varchar(3), b blob(3), t text(3))")
+
+	tk.MustQuery("select CHARACTER_MAXIMUM_LENGTH,CHARACTER_OCTET_LENGTH,NUMERIC_PRECISION,NUMERIC_SCALE,DATETIME_PRECISION from information_schema.COLUMNS where table_name='numschema'").
+		Check(testkit.Rows("<nil> <nil> 2 0 <nil>", "<nil> <nil> 4 2 <nil>", "<nil> <nil> 4 3 <nil>")) // FIXME: for mysql first one will be "<nil> <nil> 10 0 <nil>"
+	tk.MustQuery("select CHARACTER_MAXIMUM_LENGTH,CHARACTER_OCTET_LENGTH,NUMERIC_PRECISION,NUMERIC_SCALE,DATETIME_PRECISION from information_schema.COLUMNS where table_name='timeschema'").
+		Check(testkit.Rows("<nil> <nil> <nil> <nil> <nil>", "<nil> <nil> <nil> <nil> 3", "<nil> <nil> <nil> <nil> 3", "<nil> <nil> <nil> <nil> 4", "<nil> <nil> <nil> <nil> <nil>"))
+	tk.MustQuery("select CHARACTER_MAXIMUM_LENGTH,CHARACTER_OCTET_LENGTH,NUMERIC_PRECISION,NUMERIC_SCALE,DATETIME_PRECISION from information_schema.COLUMNS where table_name='strschema'").
+		Check(testkit.Rows("3 3 <nil> <nil> <nil>", "3 3 <nil> <nil> <nil>", "3 3 <nil> <nil> <nil>", "3 3 <nil> <nil> <nil>")) // FIXME: for mysql last two will be "255 255 <nil> <nil> <nil>", "255 255 <nil> <nil> <nil>"
+}
+
 func (s *testSuite) TestDataForTableRowsCountField(c *C) {
 	testleak.BeforeTest()
 	defer testleak.AfterTest(c)()
diff --git a/parser/misc.go b/parser/misc.go
index 8374c5ae838e0..fafd507d72791 100644
--- a/parser/misc.go
+++ b/parser/misc.go
@@ -444,6 +444,8 @@ var tokenMap = map[string]int{
 	"STORED":                   stored,
 	"STRAIGHT_JOIN":            straightJoin,
 	"SUBDATE":                  subDate,
+	"SUBPARTITION":             subpartition,
+	"SUBPARTITIONS":            subpartitions,
 	"SUBSTR":                   substring,
 	"SUBSTRING":                substring,
 	"SUM":                      sum,
diff --git a/parser/opcode/opcode.go b/parser/opcode/opcode.go
index 26fddbe891316..670fb2b883bd1 100644
--- a/parser/opcode/opcode.go
+++ b/parser/opcode/opcode.go
@@ -134,5 +134,5 @@ var opsLiteral = map[Op]string{
 
 // Format the ExprNode into a Writer.
 func (o Op) Format(w io.Writer) {
-	fmt.Fprintf(w, opsLiteral[o])
+	fmt.Fprintf(w, "%s", opsLiteral[o])
 }
diff --git a/parser/opcode/opcode_test.go b/parser/opcode/opcode_test.go
index 98e4f9d85cf99..a4c2821b249f1 100644
--- a/parser/opcode/opcode_test.go
+++ b/parser/opcode/opcode_test.go
@@ -13,7 +13,10 @@
 
 package opcode
 
-import "testing"
+import (
+	"bytes"
+	"testing"
+)
 
 func TestT(t *testing.T) {
 	op := Plus
@@ -21,6 +24,18 @@ func TestT(t *testing.T) {
 		t.Fatalf("invalid op code")
 	}
 
+	if len(Ops) != len(opsLiteral) {
+		t.Error("inconsistent count ops and opsliteral")
+	}
+	var buf bytes.Buffer
+	for op := range Ops {
+		op.Format(&buf)
+		if buf.String() != opsLiteral[op] {
+			t.Error("format op fail", op)
+		}
+		buf.Reset()
+	}
+
 	// Test invalid opcode
 	defer func() {
 		recover()
diff --git a/parser/parser.y b/parser/parser.y
index 5a4ded0b4dec9..85895fe731f5e 100644
--- a/parser/parser.y
+++ b/parser/parser.y
@@ -376,6 +376,8 @@ import (
 	start		"START"
 	statsPersistent	"STATS_PERSISTENT"
 	status		"STATUS"
+	subpartition	"SUBPARTITION"
+	subpartitions	"SUBPARTITIONS"
 	super		"SUPER"
 	some 		"SOME"
 	global		"GLOBAL"
@@ -721,6 +723,8 @@ import (
 	StatsPersistentVal		"stats_persistent value"
 	StringName			"string literal or identifier"
 	StringList 			"string list"
+	SubPartitionOpt			"SubPartition option"
+	SubPartitionNumOpt		"SubPartition NUM option"
 	Symbol				"Constraint Symbol"
 	TableAsName			"table alias name"
 	TableAsNameOpt 			"table alias name optional"
@@ -1796,11 +1800,11 @@ PartitionOpt:
 	{
 		$$ = nil
 	}
-|	"PARTITION" "BY" "RANGE" '(' Expression ')' PartitionNumOpt  PartitionDefinitionListOpt
+|	"PARTITION" "BY" "RANGE" '(' Expression ')' PartitionNumOpt SubPartitionOpt PartitionDefinitionListOpt
 	{
 		var defs []*ast.PartitionDefinition
-		if $8 != nil {
-			defs = $8.([]*ast.PartitionDefinition)
+		if $9 != nil {
+			defs = $9.([]*ast.PartitionDefinition)
 		}
 		$$ = &ast.PartitionOptions{
 			Tp:		model.PartitionTypeRange,
@@ -1821,6 +1825,18 @@ PartitionOpt:
 		}
 	}
 
+SubPartitionOpt:
+	{}
+|	"SUBPARTITION" "BY" "HASH" '(' Expression ')' SubPartitionNumOpt
+	{}
+|	"SUBPARTITION" "BY" "KEY" '(' ColumnNameList ')' SubPartitionNumOpt
+	{}
+
+SubPartitionNumOpt:
+	{}
+|	"SUBPARTITIONS" NUM
+	{}
+
 PartitionNumOpt:
 	{}
 |	"PARTITIONS" NUM
@@ -2780,8 +2796,8 @@ UnReservedKeyword:
 | "COLUMNS" | "COMMIT" | "COMPACT" | "COMPRESSED" | "CONSISTENT" | "DATA" | "DATE" %prec lowerThanStringLitToken| "DATETIME" | "DAY" | "DEALLOCATE" | "DO" | "DUPLICATE"
 | "DYNAMIC"| "END" | "ENGINE" | "ENGINES" | "ENUM" | "ERRORS" | "ESCAPE" | "EXECUTE" | "FIELDS" | "FIRST" | "FIXED" | "FLUSH" | "FORMAT" | "FULL" |"GLOBAL"
 | "HASH" | "HOUR" | "LESS" | "LOCAL" | "NAMES" | "OFFSET" | "PASSWORD" %prec lowerThanEq | "PREPARE" | "QUICK" | "REDUNDANT"
-| "ROLLBACK" | "SESSION" | "SIGNED" | "SNAPSHOT" | "START" | "STATUS" | "TABLES" | "TABLESPACE" | "TEXT" | "THAN" | "TIME" %prec lowerThanStringLitToken | "TIMESTAMP" %prec lowerThanStringLitToken
-| "TRACE" | "TRANSACTION" | "TRUNCATE" | "UNKNOWN" | "VALUE" | "WARNINGS" | "YEAR" | "MODE"  | "WEEK"  | "ANY" | "SOME" | "USER" | "IDENTIFIED"
+| "ROLLBACK" | "SESSION" | "SIGNED" | "SNAPSHOT" | "START" | "STATUS" | "SUBPARTITIONS" | "SUBPARTITION" | "TABLES" | "TABLESPACE" | "TEXT" | "THAN" | "TIME" %prec lowerThanStringLitToken 
+| "TIMESTAMP" %prec lowerThanStringLitToken | "TRACE" | "TRANSACTION" | "TRUNCATE" | "UNKNOWN" | "VALUE" | "WARNINGS" | "YEAR" | "MODE"  | "WEEK"  | "ANY" | "SOME" | "USER" | "IDENTIFIED"
 | "COLLATION" | "COMMENT" | "AVG_ROW_LENGTH" | "CONNECTION" | "CHECKSUM" | "COMPRESSION" | "KEY_BLOCK_SIZE" | "MASTER" | "MAX_ROWS"
 | "MIN_ROWS" | "NATIONAL" | "ROW" | "ROW_FORMAT" | "QUARTER" | "GRANTS" | "TRIGGERS" | "DELAY_KEY_WRITE" | "ISOLATION" | "JSON"
 | "REPEATABLE" | "COMMITTED" | "UNCOMMITTED" | "ONLY" | "SERIALIZABLE" | "LEVEL" | "VARIABLES" | "SQL_CACHE" | "INDEXES" | "PROCESSLIST"
diff --git a/parser/parser_test.go b/parser/parser_test.go
index b723e8ea6bb61..0097a57c5f6c9 100644
--- a/parser/parser_test.go
+++ b/parser/parser_test.go
@@ -87,7 +87,7 @@ func (s *testParserSuite) TestSimple(c *C) {
 		"date", "datediff", "datetime", "deallocate", "do", "from_days", "end", "engine", "engines", "execute", "first", "full",
 		"local", "names", "offset", "password", "prepare", "quick", "rollback", "session", "signed",
 		"start", "global", "tables", "tablespace", "text", "time", "timestamp", "tidb", "transaction", "truncate", "unknown",
-		"value", "warnings", "year", "now", "substr", "substring", "mode", "any", "some", "user", "identified",
+		"value", "warnings", "year", "now", "substr", "subpartition", "subpartitions", "substring", "mode", "any", "some", "user", "identified",
 		"collation", "comment", "avg_row_length", "checksum", "compression", "connection", "key_block_size",
 		"max_rows", "min_rows", "national", "row", "quarter", "escape", "grants", "status", "fields", "triggers",
 		"delay_key_write", "isolation", "partitions", "repeatable", "committed", "uncommitted", "only", "serializable", "level",
@@ -2390,6 +2390,14 @@ func (s *testParserSuite) TestTablePartition(c *C) {
 		{`create table t1 (a int) partition by range (a)
 		  (PARTITION p0 VALUES LESS THAN (63340531200) COMMENT 'xxx' ENGINE = MyISAM ,
 		   PARTITION p1 VALUES LESS THAN (63342604800) ENGINE = MyISAM)`, true},
+		{`create table t (id int)
+		    partition by range (id)
+		    subpartition by key (id) subpartitions 2
+		    (partition p0 values less than (42))`, true},
+		{`create table t (id int)
+		    partition by range (id)
+		    subpartition by hash (id)
+		    (partition p0 values less than (42))`, true},
 	}
 	s.RunTest(c, table)
 
diff --git a/plan/logical_plan_builder.go b/plan/logical_plan_builder.go
index 5f18b30bff543..db807b5315e11 100644
--- a/plan/logical_plan_builder.go
+++ b/plan/logical_plan_builder.go
@@ -854,6 +854,13 @@ func matchField(f *ast.SelectField, col *ast.ColumnNameExpr, ignoreAsName bool)
 		if f.AsName.L == "" || ignoreAsName {
 			if curCol, isCol := f.Expr.(*ast.ColumnNameExpr); isCol {
 				return curCol.Name.Name.L == col.Name.Name.L
+			} else if _, isFunc := f.Expr.(*ast.FuncCallExpr); isFunc {
+				// Fix issue 7331
+				// If there are some function calls in SelectField, we check if
+				// ColumnNameExpr in GroupByClause matches one of these function calls.
+				// Example: select concat(k1,k2) from t group by `concat(k1,k2)`,
+				// `concat(k1,k2)` matches with function call concat(k1, k2).
+				return strings.ToLower(f.Text()) == col.Name.Name.L
 			}
 			// a expression without as name can't be matched.
 			return false
diff --git a/plan/logical_plan_test.go b/plan/logical_plan_test.go
index c0615ffe6ed2e..eb80d298edde2 100644
--- a/plan/logical_plan_test.go
+++ b/plan/logical_plan_test.go
@@ -538,6 +538,17 @@ func (s *testPlanSuite) TestTablePartition(c *C) {
 			best: "Dual->Projection",
 			is:   is1,
 		},
+		{
+			// NULL will be located in the first partition.
+			sql:  "select * from t where t.h is null",
+			best: "Partition(41)->Projection",
+			is:   is,
+		},
+		{
+			sql:  "select * from t where t.h is null or t.h > 70",
+			best: "UnionAll{Partition(41)->Partition(44)}->Projection",
+			is:   is1,
+		},
 	}
 	for _, ca := range tests {
 		comment := Commentf("for %s", ca.sql)
@@ -1171,6 +1182,14 @@ func (s *testPlanSuite) TestValidate(c *C) {
 			sql: "select a from t having sum(avg(a))",
 			err: ErrInvalidGroupFuncUse,
 		},
+		{
+			sql: "select concat(c_str, d_str) from t group by `concat(c_str, d_str)`",
+			err: nil,
+		},
+		{
+			sql: "select concat(c_str, d_str) from t group by `concat(c_str,d_str)`",
+			err: ErrUnknownColumn,
+		},
 	}
 	for _, tt := range tests {
 		sql := tt.sql
diff --git a/plan/rule_partition_processor.go b/plan/rule_partition_processor.go
index 3015ee0785400..ab37b28cc0e5e 100644
--- a/plan/rule_partition_processor.go
+++ b/plan/rule_partition_processor.go
@@ -14,12 +14,10 @@ package plan
 
 import (
 	"github.com/juju/errors"
-	"github.com/pingcap/tidb/ast"
 	"github.com/pingcap/tidb/expression"
 	"github.com/pingcap/tidb/sessionctx"
 	"github.com/pingcap/tidb/table/tables"
 	"github.com/pingcap/tidb/util/ranger"
-	log "github.com/sirupsen/logrus"
 )
 
 // partitionProcessor rewrites the ast for table partition.
@@ -76,8 +74,10 @@ func (s *partitionProcessor) prune(ds *DataSource) (LogicalPlan, error) {
 	}
 
 	var partitionExprs []expression.Expression
+	var col *expression.Column
 	if table, ok := ds.table.(partitionTable); ok {
 		partitionExprs = table.PartitionExpr().Ranges
+		col = table.PartitionExpr().Column
 	}
 	if len(partitionExprs) == 0 {
 		return nil, errors.New("partition expression missing")
@@ -86,8 +86,6 @@ func (s *partitionProcessor) prune(ds *DataSource) (LogicalPlan, error) {
 	// Rewrite data source to union all partitions, during which we may prune some
 	// partitions according to the filter conditions pushed to the DataSource.
 	children := make([]LogicalPlan, 0, len(pi.Definitions))
-
-	col := partitionExprAccessColumn(partitionExprs[0])
 	for i, expr := range partitionExprs {
 		if col != nil {
 			// If the selection condition would never be satisified, prune that partition.
@@ -143,21 +141,3 @@ func (s *partitionProcessor) canBePrune(ctx sessionctx.Context, col *expression.
 	}
 	return len(r) == 0, nil
 }
-
-// partitionExprAccessColumn extracts the column which is used by the partition expression.
-// If the partition expression is not a simple operation on one column, return nil.
-func partitionExprAccessColumn(expr expression.Expression) *expression.Column {
-	lt, ok := expr.(*expression.ScalarFunction)
-	if !ok || lt.FuncName.L != ast.LT {
-		// The partition expression is constructed by us, its format should always be
-		// expr < value, such as "id < 42", "timestamp(id) < maxvalue"
-		log.Warnf("illegal partition expr:%s", expr.String())
-		return nil
-	}
-	tmp := lt.GetArgs()[0]
-	col, ok := tmp.(*expression.Column)
-	if !ok {
-		return nil
-	}
-	return col
-}
diff --git a/store/tikv/coprocessor.go b/store/tikv/coprocessor.go
index 9e9392fa9ca87..e5f15c148a283 100644
--- a/store/tikv/coprocessor.go
+++ b/store/tikv/coprocessor.go
@@ -23,6 +23,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/cznic/mathutil"
 	"github.com/juju/errors"
 	"github.com/opentracing/opentracing-go"
 	"github.com/pingcap/kvproto/pkg/coprocessor"
@@ -235,6 +236,9 @@ func (r *copRanges) split(key []byte) (*copRanges, *copRanges) {
 	return r.slice(0, n), r.slice(n, r.len())
 }
 
+// rangesPerTask limits the length of the ranges slice sent in one copTask.
+const rangesPerTask = 25000
+
 func buildCopTasks(bo *Backoffer, cache *RegionCache, ranges *copRanges, desc bool, streaming bool) ([]*copTask, error) {
 	start := time.Now()
 	rangesLen := ranges.len()
@@ -245,12 +249,19 @@ func buildCopTasks(bo *Backoffer, cache *RegionCache, ranges *copRanges, desc bo
 
 	var tasks []*copTask
 	appendTask := func(region RegionVerID, ranges *copRanges) {
-		tasks = append(tasks, &copTask{
-			region:   region,
-			ranges:   ranges,
-			respChan: make(chan *copResponse, 1),
-			cmdType:  cmdType,
-		})
+		// TiKV will return gRPC error if the message is too large. So we need to limit the length of the ranges slice
+		// to make sure the message can be sent successfully.
+		rLen := ranges.len()
+		for i := 0; i < rLen; {
+			nextI := mathutil.Min(i+rangesPerTask, rLen)
+			tasks = append(tasks, &copTask{
+				region:   region,
+				ranges:   ranges.slice(i, nextI),
+				respChan: make(chan *copResponse, 1),
+				cmdType:  cmdType,
+			})
+			i = nextI
+		}
 	}
 
 	err := splitRanges(bo, cache, ranges, appendTask)
diff --git a/store/tikv/gcworker/gc_worker.go b/store/tikv/gcworker/gc_worker.go
index d29c4b91b6022..9f4885321540f 100644
--- a/store/tikv/gcworker/gc_worker.go
+++ b/store/tikv/gcworker/gc_worker.go
@@ -216,6 +216,7 @@ func (w *GCWorker) storeIsBootstrapped() bool {
 // Leader of GC worker checks if it should start a GC job every tick.
 func (w *GCWorker) leaderTick(ctx context.Context) error {
 	if w.gcIsRunning {
+		log.Infof("[gc worker] leaderTick on %s: there's already a gc job running. skipped.", w.uuid)
 		return nil
 	}
 
@@ -231,6 +232,7 @@ func (w *GCWorker) leaderTick(ctx context.Context) error {
 	// wait a while before starting a new job.
 	if time.Since(w.lastFinish) < gcWaitTime {
 		w.gcIsRunning = false
+		log.Infof("[gc worker] leaderTick on %s: another gc job has just finished. skipped.", w.uuid)
 		return nil
 	}
 
@@ -288,6 +290,7 @@ func (w *GCWorker) checkGCInterval(now time.Time) (bool, error) {
 	}
 
 	if lastRun != nil && lastRun.Add(*runInterval).After(now) {
+		log.Infof("[gc worker] leaderTick on %s: gc interval (%v) haven't past since last run (%v). no need to gc", w.uuid, runInterval, lastRun)
 		return false, nil
 	}
 
@@ -307,6 +310,8 @@ func (w *GCWorker) calculateNewSafePoint(now time.Time) (*time.Time, error) {
 	safePoint := now.Add(-*lifeTime)
 	// We should never decrease safePoint.
 	if lastSafePoint != nil && safePoint.Before(*lastSafePoint) {
+		log.Info("[gc worker] leaderTick on %s: last safe point (%v) is later than current one (%v). no need to gc. "+
+			"this might be caused by manually enlarging gc lifetime.", w.uuid, lastSafePoint, safePoint)
 		return nil, nil
 	}
 	return &safePoint, nil
diff --git a/table/tables/partition.go b/table/tables/partition.go
index 792ce7ad46494..5f72a99e40a33 100644
--- a/table/tables/partition.go
+++ b/table/tables/partition.go
@@ -98,14 +98,17 @@ func newPartitionedTable(tbl *Table, tblInfo *model.TableInfo) (table.Table, err
 //      p1 values less than (y1)
 //      p2 values less than (y2)
 //      p3 values less than (y3))
-// Ranges: (x < y1); (y1 <= x < y2); (y2 <= x < y3)
+// Ranges: (x < y1 or x is null); (y1 <= x < y2); (y2 <= x < y3)
 // UpperBounds: (x < y1); (x < y2); (x < y3)
 type PartitionExpr struct {
+	// Column is the column appeared in the by range expression, partition pruning need this to work.
+	Column      *expression.Column
 	Ranges      []expression.Expression
 	UpperBounds []expression.Expression
 }
 
 func generatePartitionExpr(tblInfo *model.TableInfo) (*PartitionExpr, error) {
+	var column *expression.Column
 	// The caller should assure partition info is not nil.
 	pi := tblInfo.GetPartitionInfo()
 	ctx := mock.NewContext()
@@ -129,6 +132,19 @@ func generatePartitionExpr(tblInfo *model.TableInfo) (*PartitionExpr, error) {
 
 		if i > 0 {
 			fmt.Fprintf(&buf, " and ((%s) >= (%s))", pi.Expr, pi.Definitions[i-1].LessThan[0])
+		} else {
+			// NULL will locate in the first partition, so its expression is (expr < value or expr is null).
+			fmt.Fprintf(&buf, " or ((%s) is null)", pi.Expr)
+
+			// Extracts the column of the partition expression, it will be used by partition prunning.
+			if tmp, err1 := expression.ParseSimpleExprWithTableInfo(ctx, pi.Expr, tblInfo); err1 == nil {
+				if col, ok := tmp.(*expression.Column); ok {
+					column = col
+				}
+			}
+			if column == nil {
+				log.Warnf("partition pruning won't work on this expr:%s", pi.Expr)
+			}
 		}
 
 		expr, err = expression.ParseSimpleExprWithTableInfo(ctx, buf.String(), tblInfo)
@@ -141,6 +157,7 @@ func generatePartitionExpr(tblInfo *model.TableInfo) (*PartitionExpr, error) {
 		buf.Reset()
 	}
 	return &PartitionExpr{
+		Column:      column,
 		Ranges:      partitionPruneExprs,
 		UpperBounds: locateExprs,
 	}, nil
diff --git a/table/tables/tables_test.go b/table/tables/tables_test.go
index 553d97cbe5b9c..0eccc2d025d0b 100644
--- a/table/tables/tables_test.go
+++ b/table/tables/tables_test.go
@@ -341,6 +341,8 @@ PARTITION BY RANGE ( id ) (
 
 	_, err := ts.se.Execute(context.Background(), "set @@session.tidb_enable_table_partition=1")
 	c.Assert(err, IsNil)
+	_, err = ts.se.Execute(context.Background(), "drop table if exists t1;")
+	c.Assert(err, IsNil)
 	_, err = ts.se.Execute(context.Background(), createTable1)
 	c.Assert(err, IsNil)
 	tb, err := ts.dom.InfoSchema().TableByName(model.NewCIStr("test"), model.NewCIStr("t1"))
@@ -422,3 +424,44 @@ PARTITION BY RANGE ( id ) (
 		c.Assert(pd.ID, Equals, p.GetPhysicalID())
 	}
 }
+
+func (ts *testSuite) TestGeneratePartitionExpr(c *C) {
+	_, err := ts.se.Execute(context.Background(), "use test")
+	c.Assert(err, IsNil)
+	_, err = ts.se.Execute(context.Background(), "set @@session.tidb_enable_table_partition=1")
+	c.Assert(err, IsNil)
+	_, err = ts.se.Execute(context.Background(), "drop table if exists t1;")
+	c.Assert(err, IsNil)
+	_, err = ts.se.Execute(context.Background(), `create table t1 (id int)
+							partition by range (id) (
+							partition p0 values less than (4),
+							partition p1 values less than (7),
+							partition p3 values less than maxvalue)`)
+	c.Assert(err, IsNil)
+
+	tbl, err := ts.dom.InfoSchema().TableByName(model.NewCIStr("test"), model.NewCIStr("t1"))
+	c.Assert(err, IsNil)
+	type partitionExpr interface {
+		PartitionExpr() *tables.PartitionExpr
+	}
+	pe := tbl.(partitionExpr).PartitionExpr()
+	c.Assert(pe.Column.TblName.L, Equals, "t1")
+	c.Assert(pe.Column.ColName.L, Equals, "id")
+
+	ranges := []string{
+		"or(lt(t1.id, 4), isnull(t1.id))",
+		"and(lt(t1.id, 7), ge(t1.id, 4))",
+		"and(1, ge(t1.id, 7))",
+	}
+	upperBounds := []string{
+		"lt(t1.id, 4)",
+		"lt(t1.id, 7)",
+		"1",
+	}
+	for i, expr := range pe.Ranges {
+		c.Assert(expr.String(), Equals, ranges[i])
+	}
+	for i, expr := range pe.UpperBounds {
+		c.Assert(expr.String(), Equals, upperBounds[i])
+	}
+}
diff --git a/types/etc.go b/types/etc.go
index efe0cc18d8262..56e551f9735bd 100644
--- a/types/etc.go
+++ b/types/etc.go
@@ -73,9 +73,14 @@ func IsTypeTime(tp byte) bool {
 	return tp == mysql.TypeDatetime || tp == mysql.TypeDate || tp == mysql.TypeTimestamp
 }
 
-// IsTypeFloat returns a boolean indicating whether the tp is floating-point type.
-func IsTypeFloat(tp byte) bool {
-	return tp == mysql.TypeFloat || tp == mysql.TypeDouble
+// IsTypeNumeric returns a boolean indicating whether the tp is numeric type.
+func IsTypeNumeric(tp byte) bool {
+	switch tp {
+	case mysql.TypeBit, mysql.TypeTiny, mysql.TypeInt24, mysql.TypeLong, mysql.TypeLonglong, mysql.TypeNewDecimal,
+		mysql.TypeDecimal, mysql.TypeFloat, mysql.TypeDouble, mysql.TypeShort:
+		return true
+	}
+	return false
 }
 
 // IsTemporalWithDate returns a boolean indicating
@@ -87,7 +92,7 @@ func IsTemporalWithDate(tp byte) bool {
 // IsBinaryStr returns a boolean indicating
 // whether the field type is a binary string type.
 func IsBinaryStr(ft *FieldType) bool {
-	if ft.Collate == charset.CollationBin && IsString(ft) {
+	if ft.Collate == charset.CollationBin && IsString(ft.Tp) {
 		return true
 	}
 	return false
@@ -96,7 +101,7 @@ func IsBinaryStr(ft *FieldType) bool {
 // IsNonBinaryStr returns a boolean indicating
 // whether the field type is a non-binary string type.
 func IsNonBinaryStr(ft *FieldType) bool {
-	if ft.Collate != charset.CollationBin && IsString(ft) {
+	if ft.Collate != charset.CollationBin && IsString(ft.Tp) {
 		return true
 	}
 	return false
@@ -104,8 +109,8 @@ func IsNonBinaryStr(ft *FieldType) bool {
 
 // IsString returns a boolean indicating
 // whether the field type is a string type.
-func IsString(ft *FieldType) bool {
-	return IsTypeChar(ft.Tp) || IsTypeBlob(ft.Tp) || IsTypeVarchar(ft.Tp) || IsTypeUnspecified(ft.Tp)
+func IsString(tp byte) bool {
+	return IsTypeChar(tp) || IsTypeBlob(tp) || IsTypeVarchar(tp) || IsTypeUnspecified(tp)
 }
 
 var type2Str = map[byte]string{