Add CHAR and DATETIME types support (#823)

Add CHAR and DATETIME types support

Author: ajnavarro

_integration/go/mysql_test.go

@@ -82,7 +82,7 @@ func TestGrafana(t *testing.T) {
 				{"name", "TEXT"},
 				{"email", "TEXT"},
 				{"phone_numbers", "JSON"},
-				{"created_at", "DATETIME"},
+				{"created_at", "TIMESTAMP"},
 			},
 		},
 		{

engine_test.go

@@ -2277,7 +2277,7 @@ func TestDDL(t *testing.T) {
 	testQuery(t, e,
 		"CREATE TABLE t1(a INTEGER, b TEXT, c DATE, "+
 			"d TIMESTAMP, e VARCHAR(20), f BLOB NOT NULL, "+
-			"b1 BOOL, b2 BOOLEAN NOT NULL)",
+			"b1 BOOL, b2 BOOLEAN NOT NULL, g DATETIME, h CHAR(40))",
 		[]sql.Row(nil),
 	)
 
@@ -2296,6 +2296,8 @@ func TestDDL(t *testing.T) {
 		{Name: "f", Type: sql.Blob, Source: "t1"},
 		{Name: "b1", Type: sql.Uint8, Nullable: true, Source: "t1"},
 		{Name: "b2", Type: sql.Uint8, Source: "t1"},
+		{Name: "g", Type: sql.Datetime, Nullable: true, Source: "t1"},
+		{Name: "h", Type: sql.Text, Nullable: true, Source: "t1"},
 	}
 
 	require.Equal(s, testTable.Schema())

sql/parse/parse_test.go

@@ -13,7 +13,7 @@ import (
 )
 
 var fixtures = map[string]sql.Node{
-	`CREATE TABLE t1(a INTEGER, b TEXT, c DATE, d TIMESTAMP, e VARCHAR(20), f BLOB NOT NULL)`: plan.NewCreateTable(
+	`CREATE TABLE t1(a INTEGER, b TEXT, c DATE, d TIMESTAMP, e VARCHAR(20), f BLOB NOT NULL, g DATETIME, h CHAR(40))`: plan.NewCreateTable(
 		sql.UnresolvedDatabase(""),
 		"t1",
 		sql.Schema{{
@@ -40,6 +40,14 @@ var fixtures = map[string]sql.Node{
 			Name:     "f",
 			Type:     sql.Blob,
 			Nullable: false,
+		}, {
+			Name:     "g",
+			Type:     sql.Datetime,
+			Nullable: true,
+		}, {
+			Name:     "h",
+			Type:     sql.Text,
+			Nullable: true,
 		}},
 	),
 	`DESCRIBE TABLE foo;`: plan.NewDescribe(

sql/type.go

@@ -27,7 +27,10 @@ var (
 	// ErrConvertingToTime is thrown when a value cannot be converted to a Time
 	ErrConvertingToTime = errors.NewKind("value %q can't be converted to time.Time")
 
-	// ErrVarCharTruncation is thrown when a value is textually longer than the destination capacity
+	// ErrCharTruncation is thrown when a Char value is textually longer than the destination capacity
+	ErrCharTruncation = errors.NewKind("string value of %q is longer than destination capacity %d")
+
+	// ErrVarCharTruncation is thrown when a VarChar value is textually longer than the destination capacity
 	ErrVarCharTruncation = errors.NewKind("string value of %q is longer than destination capacity %d")
 
 	// ErrValueNotNil is thrown when a value that was expected to be nil, is not
@@ -198,6 +201,8 @@ var (
 	Timestamp timestampT
 	// Date is a date with day, month and year.
 	Date dateT
+	// Datetime is a date and a time
+	Datetime datetimeT
 	// Text is a string type.
 	Text textT
 	// Boolean is a boolean type.
@@ -218,6 +223,11 @@ func Array(underlying Type) Type {
 	return arrayT{underlying}
 }
 
+// Char returns a new Char type of the given length.
+func Char(length int) Type {
+	return charT{length: length}
+}
+
 // VarChar returns a new VarChar type of the given length.
 func VarChar(length int) Type {
 	return varCharT{length: length}
@@ -254,10 +264,16 @@ func MysqlTypeToType(sql query.Type) (Type, error) {
 		return Date, nil
 	case sqltypes.Text:
 		return Text, nil
+	case sqltypes.Char:
+		// Since we can't get the size of the sqltypes.Char to instantiate a
+		// specific Char(length) type we return a Text here
+		return Text, nil
 	case sqltypes.VarChar:
 		// Since we can't get the size of the sqltypes.VarChar to instantiate a
 		// specific VarChar(length) type we return a Text here
 		return Text, nil
+	case sqltypes.Datetime:
+		return Datetime, nil
 	case sqltypes.Bit:
 		return Boolean, nil
 	case sqltypes.TypeJSON:
@@ -597,6 +613,109 @@ func (t dateT) Compare(a, b interface{}) (int, error) {
 	return 0, nil
 }
 
+type datetimeT struct{}
+
+// DatetimeLayout is the layout of the MySQL date format in the representation
+// Go understands.
+const DatetimeLayout = "2006-01-02 15:04:05"
+
+func (t datetimeT) String() string { return "DATETIME" }
+
+func (t datetimeT) Type() query.Type {
+	return sqltypes.Datetime
+}
+
+func (t datetimeT) SQL(v interface{}) (sqltypes.Value, error) {
+	if v == nil {
+		return sqltypes.NULL, nil
+	}
+
+	v, err := t.Convert(v)
+	if err != nil {
+		return sqltypes.Value{}, err
+	}
+
+	return sqltypes.MakeTrusted(
+		sqltypes.Datetime,
+		[]byte(v.(time.Time).Format(DatetimeLayout)),
+	), nil
+}
+
+func (t datetimeT) Convert(v interface{}) (interface{}, error) {
+	switch value := v.(type) {
+	case time.Time:
+		return value.UTC(), nil
+	case string:
+		t, err := time.Parse(DatetimeLayout, value)
+		if err != nil {
+			return nil, ErrConvertingToTime.Wrap(err, v)
+		}
+		return t.UTC(), nil
+	default:
+		ts, err := Int64.Convert(v)
+		if err != nil {
+			return nil, ErrInvalidType.New(reflect.TypeOf(v))
+		}
+
+		return time.Unix(ts.(int64), 0).UTC(), nil
+	}
+}
+
+func (t datetimeT) Compare(a, b interface{}) (int, error) {
+	av := a.(time.Time)
+	bv := b.(time.Time)
+	if av.Before(bv) {
+		return -1, nil
+	} else if av.After(bv) {
+		return 1, nil
+	}
+	return 0, nil
+}
+
+type charT struct {
+	length int
+}
+
+func (t charT) Capacity() int { return t.length }
+
+func (t charT) String() string { return fmt.Sprintf("CHAR(%d)", t.length) }
+
+func (t charT) Type() query.Type {
+	return sqltypes.Char
+}
+
+func (t charT) SQL(v interface{}) (sqltypes.Value, error) {
+	if v == nil {
+		return sqltypes.MakeTrusted(sqltypes.Char, nil), nil
+	}
+
+	v, err := t.Convert(v)
+	if err != nil {
+		return sqltypes.Value{}, err
+	}
+
+	return sqltypes.MakeTrusted(sqltypes.Char, []byte(v.(string))), nil
+}
+
+// Converts any value that can be casted to a string
+func (t charT) Convert(v interface{}) (interface{}, error) {
+	val, err := cast.ToStringE(v)
+	if err != nil {
+		return nil, ErrConvertToSQL.New(t)
+	}
+
+	if len(val) > t.length {
+		return nil, ErrCharTruncation.New(val, t.length)
+	}
+	return val, nil
+}
+
+// Compares two strings lexicographically
+func (t charT) Compare(a interface{}, b interface{}) (int, error) {
+	return strings.Compare(a.(string), b.(string)), nil
+}
+
+
 type varCharT struct {
 	length int
 }
@@ -1013,9 +1132,9 @@ func IsInteger(t Type) bool {
 	return IsSigned(t) || IsUnsigned(t)
 }
 
-// IsTime checks if t is a timestamp or date.
+// IsTime checks if t is a timestamp, date or datetime
 func IsTime(t Type) bool {
-	return t == Timestamp || t == Date
+	return t == Timestamp || t == Date || t == Datetime
 }
 
 // IsDecimal checks if t is decimal type.
@@ -1025,7 +1144,13 @@ func IsDecimal(t Type) bool {
 
 // IsText checks if t is a text type.
 func IsText(t Type) bool {
-	return t == Text || t == Blob || t == JSON || IsVarChar(t)
+	return t == Text || t == Blob || t == JSON || IsVarChar(t) || IsChar(t)
+}
+
+// IsChar checks if t is a Char type.
+func IsChar(t Type) bool {
+	_, ok := t.(charT)
+	return ok
 }
 
 // IsVarChar checks if t is a varchar type.
@@ -1082,9 +1207,13 @@ func MySQLTypeName(t Type) string {
 	case sqltypes.Float64:
 		return "DOUBLE"
 	case sqltypes.Timestamp:
+		return "TIMESTAMP"
+	case sqltypes.Datetime:
 		return "DATETIME"
 	case sqltypes.Date:
 		return "DATE"
+	case sqltypes.Char:
+		return "CHAR"
 	case sqltypes.VarChar:
 		return "VARCHAR"
 	case sqltypes.Text:

sql/type_test.go

@@ -236,40 +236,57 @@ func TestExtraTimestamps(t *testing.T) {
 	}
 }
 
-func TestDate(t *testing.T) {
+// Generic tests for Date and Datetime.
+// typ should be Date or Datetime
+func commonTestsDatesTypes(typ Type, layout string, t *testing.T) {
 	require := require.New(t)
 	now := time.Now().UTC()
-	v, err := Date.Convert(now)
+	v, err := typ.Convert(now)
 	require.NoError(err)
-	require.Equal(now.Format(DateLayout), v.(time.Time).Format(DateLayout))
+	require.Equal(now.Format(layout), v.(time.Time).Format(layout))
 
-	v, err = Date.Convert(now.Format(DateLayout))
+	v, err = typ.Convert(now.Format(layout))
 	require.NoError(err)
 	require.Equal(
-		now.Format(DateLayout),
-		v.(time.Time).Format(DateLayout),
+		now.Format(layout),
+		v.(time.Time).Format(layout),
 	)
 
-	v, err = Date.Convert(now.Unix())
+	v, err = typ.Convert(now.Unix())
 	require.NoError(err)
 	require.Equal(
-		now.Format(DateLayout),
-		v.(time.Time).Format(DateLayout),
+		now.Format(layout),
+		v.(time.Time).Format(layout),
 	)
 
-	sql, err := Date.SQL(now)
+	sql, err := typ.SQL(now)
 	require.NoError(err)
-	require.Equal([]byte(now.Format(DateLayout)), sql.Raw())
+	require.Equal([]byte(now.Format(layout)), sql.Raw())
+
+	after := now.Add(26 * time.Hour)
+	lt(t, typ, now, after)
+	eq(t, typ, now, now)
+	gt(t, typ, after, now)
+}
+
+func TestDate(t *testing.T) {
+	commonTestsDatesTypes(Date, DateLayout, t)
 
+	now := time.Now().UTC()
 	after := now.Add(time.Second)
 	eq(t, Date, now, after)
 	eq(t, Date, now, now)
 	eq(t, Date, after, now)
+}
 
-	after = now.Add(26 * time.Hour)
-	lt(t, Date, now, after)
-	eq(t, Date, now, now)
-	gt(t, Date, after, now)
+func TestDatetime(t *testing.T) {
+	commonTestsDatesTypes(Datetime, DatetimeLayout, t)
+
+	now := time.Now().UTC()
+	after := now.Add(time.Millisecond)
+	lt(t, Datetime, now, after)
+	eq(t, Datetime, now, now)
+	gt(t, Datetime, after, now)
 }
 
 func TestBlob(t *testing.T) {
@@ -325,20 +342,22 @@ func TestTuple(t *testing.T) {
 	gt(t, typ, []interface{}{1, 2, 4}, []interface{}{1, 2, 3})
 }
 
-func TestVarChar(t *testing.T) {
-	typ := VarChar(3)
-	require.True(t, IsVarChar(typ))
+// Generic test for Char and VarChar types.
+// genType should be sql.Char or sql.VarChar
+func testCharTypes(genType func(int) Type, checkType func(Type) bool, t *testing.T) {
+	typ := genType(3)
+	require.True(t, checkType(typ))
 	require.True(t, IsText(typ))
 	convert(t, typ, "foo", "foo")
 	fooByte := []byte{'f', 'o', 'o'}
 	convert(t, typ, fooByte, "foo")
 
-	typ = VarChar(1)
+	typ = genType(1)
 	convertErr(t, typ, "foo")
 	convertErr(t, typ, fooByte)
 	convertErr(t, typ, 123)
 
-	typ = VarChar(10)
+	typ = genType(10)
 	convert(t, typ, 123, "123")
 	convertErr(t, typ, 1234567890123)
 
@@ -353,10 +372,18 @@ func TestVarChar(t *testing.T) {
 	require.NoError(t, err)
 
 	convert(t, typ, text, "abc")
-	typ1 := VarChar(1)
+	typ1 := genType(1)
 	convertErr(t, typ1, text)
 }
 
+func TestChar(t *testing.T) {
+	testCharTypes(Char, IsChar, t)
+}
+
+func TestVarChar(t *testing.T) {
+	testCharTypes(VarChar, IsVarChar, t)
+}
+
 func TestArray(t *testing.T) {
 	require := require.New(t)