plan: add cast function for PatternInExpr. (pingcap#1932)

executor/executor_test.go

@@ -469,6 +469,9 @@ func (s *testSuite) TestIn(c *C) {
 	queryStr := `select c2 from t where c1 in ('7', '10', '112', '111', '98', '106', '100', '9', '18', '17') order by c2`
 	r := tk.MustQuery(queryStr)
 	r.Check(testkit.Rows("7", "9", "10", "17", "18", "98", "100", "106", "111", "112"))
+
+	queryStr = `select c2 from t where c1 in ('7a')`
+	tk.MustQuery(queryStr).Check(testkit.Rows("7"))
 }
 
 func (s *testSuite) TestTablePKisHandleScan(c *C) {

plan/typeinferer.go

@@ -16,6 +16,8 @@ package plan
 import (
 	"strings"
 
+	"github.com/juju/errors"
+	"github.com/ngaut/log"
 	"github.com/pingcap/tidb/ast"
 	"github.com/pingcap/tidb/mysql"
 	"github.com/pingcap/tidb/parser/opcode"
@@ -88,6 +90,7 @@ func (v *typeInferrer) Leave(in ast.Node) (out ast.Node, ok bool) {
 		x.SetType(types.NewFieldType(mysql.TypeLonglong))
 		x.Type.Charset = charset.CharsetBin
 		x.Type.Collate = charset.CollationBin
+		v.convertValueToColumnTypeIfNeeded(x)
 	case *ast.PatternLikeExpr:
 		x.SetType(types.NewFieldType(mysql.TypeLonglong))
 		x.Type.Charset = charset.CharsetBin
@@ -373,3 +376,33 @@ func (v *typeInferrer) handleCaseExpr(x *ast.CaseExpr) {
 	// TODO: We need a better way to set charset/collation
 	x.Type.Charset, x.Type.Collate = types.DefaultCharsetForType(x.Type.Tp)
 }
+
+// ConvertValueToColumnTypeIfNeeded checks if the expr in PatternInExpr is column name,
+// and casts function to the items in the list.
+func (v *typeInferrer) convertValueToColumnTypeIfNeeded(x *ast.PatternInExpr) {
+	if cn, ok := x.Expr.(*ast.ColumnNameExpr); ok && cn.Refer != nil {
+		ft := cn.Refer.Column.FieldType
+		for _, expr := range x.List {
+			if valueExpr, ok := expr.(*ast.ValueExpr); ok {
+				newDatum, err := valueExpr.Datum.ConvertTo(&ft)
+				if err != nil {
+					v.err = errors.Trace(err)
+				}
+				cmp, err := newDatum.CompareDatum(valueExpr.Datum)
+				if err != nil {
+					v.err = errors.Trace(err)
+				}
+				if cmp != 0 {
+					// The value will never match the column, do not set newDatum.
+					continue
+				}
+				valueExpr.SetDatum(newDatum)
+			}
+		}
+		if v.err != nil {
+			// TODO: Errors should be handled differently according to query context.
+			log.Errorf("inferor type for pattern in error %v", v.err)
+			v.err = nil
+		}
+	}
+}

util/types/convert_test.go

@@ -486,6 +486,7 @@ func (s *testTypeConvertSuite) TestConvert(c *C) {
 	signedAccept(c, mysql.TypeLong, " 2.35e3  ", "2350")
 	signedAccept(c, mysql.TypeLong, " +2.51 ", "3")
 	signedAccept(c, mysql.TypeLong, " -3.58", "-4")
+	signedDeny(c, mysql.TypeLong, " 1a ", "1")
 
 	// integer from float
 	signedAccept(c, mysql.TypeLong, 234.5456, "235")

util/types/datum.go

@@ -793,59 +793,8 @@ func (d *Datum) convertToString(target *FieldType) (Datum, error) {
 }
 
 func (d *Datum) convertToInt(target *FieldType) (Datum, error) {
-	tp := target.Tp
-	lowerBound := signedLowerBound[tp]
-	upperBound := signedUpperBound[tp]
-	var (
-		val int64
-		err error
-		ret Datum
-	)
-	switch d.k {
-	case KindInt64:
-		val, err = convertIntToInt(d.GetInt64(), lowerBound, upperBound, tp)
-	case KindUint64:
-		val, err = convertUintToInt(d.GetUint64(), upperBound, tp)
-	case KindFloat32, KindFloat64:
-		val, err = convertFloatToInt(d.GetFloat64(), lowerBound, upperBound, tp)
-	case KindString, KindBytes:
-		fval, err1 := StrToFloat(d.GetString())
-		if err1 != nil {
-			return ret, errors.Trace(err1)
-		}
-		val, err = convertFloatToInt(fval, lowerBound, upperBound, tp)
-	case KindMysqlTime:
-		dec := d.GetMysqlTime().ToNumber()
-		dec.Round(dec, 0)
-		val, _ = dec.ToInt()
-		val, err = convertIntToInt(val, lowerBound, upperBound, tp)
-	case KindMysqlDuration:
-		dec := d.GetMysqlDuration().ToNumber()
-		dec.Round(dec, 0)
-		iVal, err1 := dec.ToInt()
-		val, err = convertIntToInt(iVal, lowerBound, upperBound, tp)
-		if err == nil {
-			err = err1
-		}
-	case KindMysqlDecimal:
-		fval, _ := d.GetMysqlDecimal().ToFloat64()
-		val, err = convertFloatToInt(fval, lowerBound, upperBound, tp)
-	case KindMysqlHex:
-		val, err = convertFloatToInt(d.GetMysqlHex().ToNumber(), lowerBound, upperBound, tp)
-	case KindMysqlBit:
-		val, err = convertFloatToInt(d.GetMysqlBit().ToNumber(), lowerBound, upperBound, tp)
-	case KindMysqlEnum:
-		val, err = convertFloatToInt(d.GetMysqlEnum().ToNumber(), lowerBound, upperBound, tp)
-	case KindMysqlSet:
-		val, err = convertFloatToInt(d.GetMysqlSet().ToNumber(), lowerBound, upperBound, tp)
-	default:
-		return invalidConv(d, target.Tp)
-	}
-	ret.SetInt64(val)
-	if err != nil {
-		return ret, errors.Trace(err)
-	}
-	return ret, nil
+	i64, err := d.toSignedInteger(target.Tp)
+	return NewIntDatum(i64), errors.Trace(err)
 }
 
 func (d *Datum) convertToUint(target *FieldType) (Datum, error) {
@@ -1236,7 +1185,10 @@ func (d *Datum) ToDecimal() (*mysql.MyDecimal, error) {
 
 // ToInt64 converts to a int64.
 func (d *Datum) ToInt64() (int64, error) {
-	tp := mysql.TypeLonglong
+	return d.toSignedInteger(mysql.TypeLonglong)
+}
+
+func (d *Datum) toSignedInteger(tp byte) (int64, error) {
 	lowerBound := signedLowerBound[tp]
 	upperBound := signedUpperBound[tp]
 	switch d.Kind() {
@@ -1251,10 +1203,11 @@ func (d *Datum) ToInt64() (int64, error) {
 	case KindString:
 		s := d.GetString()
 		fval, err := StrToFloat(s)
-		if err != nil {
-			return 0, errors.Trace(err)
+		i64, err1 := convertFloatToInt(fval, lowerBound, upperBound, tp)
+		if err == nil {
+			err = err1
 		}
-		return convertFloatToInt(fval, lowerBound, upperBound, tp)
+		return i64, errors.Trace(err)
 	case KindBytes:
 		s := string(d.GetBytes())
 		fval, err := StrToFloat(s)