Repository: src-d/go-mysql-server
Pull request #816 (2019-09-11 21:51:30)

Fixed INSERT bugs, added UINT64 support, and added tests
I found some bugs for INSERT statements, and implemented a few tests as well (some specific to the bug fixes, others for general INSERT testing).

One bug dealt with integer values greater than an INT64. By default, all numbers are parsed to INT64, meaning that a column with type `sql.Uint64` cannot actually hold the entire range. I added another check that only attempts to parse as a UINT64 in the event that the INT64 parsing fails.

Another bug dealt with the absence of columns in an INSERT statement. For example: `INSERT INTO tablename VALUES ...` would parse as an empty column list, rather than the entire column list. This behavior has been fixed.

Another bug dealt with a mismatch between the number of columns supplied and the number of values supplied for an INSERT statement. This has also been fixed.

The last bug dealt with the smallest integer types. The `Convert` function was missing entries for those types, and causing inserts on those columns to fail that should otherwise be valid. This has also been fixed.


Patch:
diff --git a/sql/parse/parse.go b/sql/parse/parse.go
index 53da95a..4f05916 100644
--- a/sql/parse/parse.go
+++ b/sql/parse/parse.go
@@ -9,7 +9,7 @@ import (
 	"strconv"
 	"strings"
 
-	opentracing "github.com/opentracing/opentracing-go"
+	"github.com/opentracing/opentracing-go"
 	"github.com/src-d/go-mysql-server/sql"
 	"github.com/src-d/go-mysql-server/sql/expression"
 	"github.com/src-d/go-mysql-server/sql/expression/function"
@@ -903,7 +903,12 @@ func convertVal(v *sqlparser.SQLVal) (sql.Expression, error) {
 		//TODO: Use smallest integer representation and widen later.
 		val, err := strconv.ParseInt(string(v.Val), 10, 64)
 		if err != nil {
-			return nil, err
+			// Might be a uint64 value that is greater than int64 max
+			val, checkErr := strconv.ParseUint(string(v.Val), 10, 64)
+			if checkErr != nil {
+				return nil, err
+			}
+			return expression.NewLiteral(val, sql.Uint64), nil
 		}
 		return expression.NewLiteral(val, sql.Int64), nil
 	case sqlparser.FloatVal:
diff --git a/sql/plan/insert.go b/sql/plan/insert.go
index 538fc16..d795cf2 100644
--- a/sql/plan/insert.go
+++ b/sql/plan/insert.go
@@ -11,6 +11,11 @@ import (
 
 // ErrInsertIntoNotSupported is thrown when a table doesn't support inserts
 var ErrInsertIntoNotSupported = errors.NewKind("table doesn't support INSERT INTO")
+var ErrInsertIntoMismatchValueCount =
+	errors.NewKind("number of values does not match number of columns provided")
+var ErrInsertIntoUnsupportedValues = errors.NewKind("%T is unsupported for inserts")
+var ErrInsertIntoDuplicateColumn = errors.NewKind("duplicate column name %v")
+var ErrInsertIntoNonexistentColumn = errors.NewKind("invalid column name %v")
 
 // InsertInto is a node describing the insertion into some table.
 type InsertInto struct {
@@ -67,6 +72,25 @@ func (p *InsertInto) Execute(ctx *sql.Context) (int, error) {
 
 	dstSchema := p.Left.Schema()
 	projExprs := make([]sql.Expression, len(dstSchema))
+
+	// If no columns are given, we assume the full schema in order
+	if len(p.Columns) == 0 {
+		p.Columns = make([]string, len(dstSchema))
+		for i, f := range dstSchema {
+			p.Columns[i] = f.Name
+		}
+	} else {
+		err = p.validateColumns(ctx, dstSchema)
+		if err != nil {
+			return 0, err
+		}
+	}
+
+	err = p.validateValueCount(ctx)
+	if err != nil {
+		return 0, err
+	}
+
 	for i, f := range dstSchema {
 		found := false
 		for j, col := range p.Columns {
@@ -78,8 +102,7 @@ func (p *InsertInto) Execute(ctx *sql.Context) (int, error) {
 		}
 
 		if !found {
-			def, _ := f.Type.Convert(nil)
-			projExprs[i] = expression.NewLiteral(def, f.Type)
+			projExprs[i] = expression.NewLiteral(nil, f.Type)
 		}
 	}
 
@@ -138,3 +161,36 @@ func (p InsertInto) String() string {
 	_ = pr.WriteChildren(p.Left.String(), p.Right.String())
 	return pr.String()
 }
+
+func (p *InsertInto) validateValueCount(ctx *sql.Context) error {
+	switch node := p.Right.(type) {
+	case *Values:
+		for _, exprTuple := range node.ExpressionTuples {
+			if len(exprTuple) != len(p.Columns) {
+				return ErrInsertIntoMismatchValueCount.New()
+			}
+		}
+	default:
+		return ErrInsertIntoUnsupportedValues.New(node)
+	}
+	return nil
+}
+
+func (p *InsertInto) validateColumns(ctx *sql.Context, dstSchema sql.Schema) error {
+	dstColNames := make(map[string]struct{})
+	for _, dstCol := range dstSchema {
+		dstColNames[dstCol.Name] = struct{}{}
+	}
+	columnNames := make(map[string]struct{})
+	for _, columnName := range p.Columns {
+		if _, exists := dstColNames[columnName]; !exists {
+			return ErrInsertIntoNonexistentColumn.New(columnName)
+		}
+		if _, exists := columnNames[columnName]; !exists {
+			columnNames[columnName] = struct{}{}
+		} else {
+			return ErrInsertIntoDuplicateColumn.New(columnName)
+		}
+	}
+	return nil
+}
diff --git a/sql/type.go b/sql/type.go
index bb68e4e..5de48b8 100644
--- a/sql/type.go
+++ b/sql/type.go
@@ -343,10 +343,18 @@ func (t numberT) Convert(v interface{}) (interface{}, error) {
 	}
 
 	switch t.t {
+	case sqltypes.Int8:
+		return cast.ToInt8E(v)
+	case sqltypes.Int16:
+		return cast.ToInt16E(v)
 	case sqltypes.Int32:
 		return cast.ToInt32E(v)
 	case sqltypes.Int64:
 		return cast.ToInt64E(v)
+	case sqltypes.Uint8:
+		return cast.ToUint8E(v)
+	case sqltypes.Uint16:
+		return cast.ToUint16E(v)
 	case sqltypes.Uint32:
 		return cast.ToUint32E(v)
 	case sqltypes.Uint64:


Test patch:
diff --git a/engine_test.go b/engine_test.go
index 7aabd1a..a0d69bf 100644
--- a/engine_test.go
+++ b/engine_test.go
@@ -700,6 +700,7 @@ var queries = []struct {
 			{"bigtable", "InnoDB", "10", "Fixed", int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), nil, nil, nil, "utf8_bin", nil, nil},
 			{"floattable", "InnoDB", "10", "Fixed", int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), nil, nil, nil, "utf8_bin", nil, nil},
 			{"niltable", "InnoDB", "10", "Fixed", int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), nil, nil, nil, "utf8_bin", nil, nil},
+			{"typestable", "InnoDB", "10", "Fixed", int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), nil, nil, nil, "utf8_bin", nil, nil},
 		},
 	},
 	{
@@ -710,6 +711,7 @@ var queries = []struct {
 			{"bigtable", "InnoDB", "10", "Fixed", int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), nil, nil, nil, "utf8_bin", nil, nil},
 			{"floattable", "InnoDB", "10", "Fixed", int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), nil, nil, nil, "utf8_bin", nil, nil},
 			{"niltable", "InnoDB", "10", "Fixed", int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), nil, nil, nil, "utf8_bin", nil, nil},
+			{"typestable", "InnoDB", "10", "Fixed", int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), nil, nil, nil, "utf8_bin", nil, nil},
 		},
 	},
 	{
@@ -852,6 +854,7 @@ var queries = []struct {
 			{"bigtable"},
 			{"floattable"},
 			{"niltable"},
+			{"typestable"},
 		},
 	},
 	{
@@ -881,6 +884,21 @@ var queries = []struct {
 			{"f64"},
 			{"b"},
 			{"f"},
+			{"id"},
+			{"i8"},
+			{"i16"},
+			{"i32"},
+			{"i64"},
+			{"u8"},
+			{"u16"},
+			{"u32"},
+			{"u64"},
+			{"ti"},
+			{"da"},
+			{"te"},
+			{"bo"},
+			{"js"},
+			{"bl"},
 		},
 	},
 	{
@@ -900,6 +918,21 @@ var queries = []struct {
 			{"f64"},
 			{"b"},
 			{"f"},
+			{"id"},
+			{"i8"},
+			{"i16"},
+			{"i32"},
+			{"i64"},
+			{"u8"},
+			{"u16"},
+			{"u32"},
+			{"u64"},
+			{"ti"},
+			{"da"},
+			{"te"},
+			{"bo"},
+			{"js"},
+			{"bl"},
 		},
 	},
 	{
@@ -919,6 +952,21 @@ var queries = []struct {
 			{"f64"},
 			{"b"},
 			{"f"},
+			{"id"},
+			{"i8"},
+			{"i16"},
+			{"i32"},
+			{"i64"},
+			{"u8"},
+			{"u16"},
+			{"u32"},
+			{"u64"},
+			{"ti"},
+			{"da"},
+			{"te"},
+			{"bo"},
+			{"js"},
+			{"bl"},
 		},
 	},
 	{
@@ -955,6 +1003,7 @@ var queries = []struct {
 			{"bigtable", "InnoDB", "10", "Fixed", int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), nil, nil, nil, "utf8_bin", nil, nil},
 			{"floattable", "InnoDB", "10", "Fixed", int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), nil, nil, nil, "utf8_bin", nil, nil},
 			{"niltable", "InnoDB", "10", "Fixed", int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), nil, nil, nil, "utf8_bin", nil, nil},
+			{"typestable", "InnoDB", "10", "Fixed", int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), nil, nil, nil, "utf8_bin", nil, nil},
 		},
 	},
 	{
@@ -1068,6 +1117,7 @@ var queries = []struct {
 			{"bigtable"},
 			{"floattable"},
 			{"niltable"},
+			{"typestable"},
 		},
 	},
 	{
@@ -1079,6 +1129,7 @@ var queries = []struct {
 			{"bigtable", "BASE TABLE"},
 			{"floattable", "BASE TABLE"},
 			{"niltable", "BASE TABLE"},
+			{"typestable", "BASE TABLE"},
 		},
 	},
 	{
@@ -1095,6 +1146,7 @@ var queries = []struct {
 			{"bigtable"},
 			{"floattable"},
 			{"niltable"},
+			{"typestable"},
 		},
 	},
 	{
@@ -1755,16 +1807,123 @@ func TestOrderByColumns(t *testing.T) {
 }
 
 func TestInsertInto(t *testing.T) {
-	e := newEngine(t)
-	testQuery(t, e,
-		"INSERT INTO mytable (s, i) VALUES ('x', 999);",
-		[]sql.Row{{int64(1)}},
-	)
+	timeParse := func(layout string, value string) time.Time {
+		t, _ := time.Parse(layout, value)
+		return t
+	}
 
-	testQuery(t, e,
-		"SELECT i FROM mytable WHERE s = 'x';",
-		[]sql.Row{{int64(999)}},
-	)
+	var insertions = []struct {
+		insertQuery    string
+		expectedInsert []sql.Row
+		selectQuery    string
+		expectedSelect []sql.Row
+	}{
+		{
+			"INSERT INTO mytable (s, i) VALUES ('x', 999);",
+			[]sql.Row{{int64(1)}},
+			"SELECT i FROM mytable WHERE s = 'x';",
+			[]sql.Row{{int64(999)}},
+		},
+		{
+			"INSERT INTO mytable VALUES (999, 'x');",
+			[]sql.Row{{int64(1)}},
+			"SELECT i FROM mytable WHERE s = 'x';",
+			[]sql.Row{{int64(999)}},
+		},
+		{
+			`INSERT INTO typestable VALUES (
+			999, 127, 32767, 2147483647, 9223372036854775807,
+			255, 65535, 4294967295, 18446744073709551615,
+			3.40282346638528859811704183484516925440e+38, 1.797693134862315708145274237317043567981e+308,
+			'2132-04-05 12:51:36', '2231-11-07',
+			'random text', true, '{"key":"value"}', 'blobdata'
+			);`,
+			[]sql.Row{{int64(1)}},
+			"SELECT * FROM typestable WHERE id = 999;",
+			[]sql.Row{{
+				int64(999), int64(math.MaxInt8), int64(math.MaxInt16), int64(math.MaxInt32), int64(math.MaxInt64),
+				int64(math.MaxUint8), int64(math.MaxUint16), int64(math.MaxUint32), uint64(math.MaxUint64),
+				float64(math.MaxFloat32), float64(math.MaxFloat64),
+				timeParse(sql.TimestampLayout, "2132-04-05 12:51:36"), timeParse(sql.DateLayout, "2231-11-07"),
+				"random text", true, `{"key":"value"}`, "blobdata",
+			}},
+		},
+		{
+			`INSERT INTO typestable VALUES (
+			999, -128, -32768, -2147483648, -9223372036854775808,
+			0, 0, 0, 0,
+			1.401298464324817070923729583289916131280e-45, 4.940656458412465441765687928682213723651e-324,
+			'0010-04-05 12:51:36', '0101-11-07',
+			'', false, '', ''
+			);`,
+			[]sql.Row{{int64(1)}},
+			"SELECT * FROM typestable WHERE id = 999;",
+			[]sql.Row{{
+				int64(999), int64(-math.MaxInt8-1), int64(-math.MaxInt16-1), int64(-math.MaxInt32-1), int64(-math.MaxInt64-1),
+				int64(0), int64(0), int64(0), int64(0),
+				float64(math.SmallestNonzeroFloat32), float64(math.SmallestNonzeroFloat64),
+				timeParse(sql.TimestampLayout, "0010-04-05 12:51:36"), timeParse(sql.DateLayout, "0101-11-07"),
+				"", false, ``, "",
+			}},
+		},
+		{
+			`INSERT INTO typestable VALUES (999, null, null, null, null, null, null, null, null,
+			null, null, null, null, null, null, null, null);`,
+			[]sql.Row{{int64(1)}},
+			"SELECT * FROM typestable WHERE id = 999;",
+			[]sql.Row{{int64(999), nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil}},
+		},
+	}
+
+	for _, insertion := range insertions {
+		e := newEngine(t)
+		ctx := newCtx()
+		testQueryWithContext(ctx, t, e, insertion.insertQuery, insertion.expectedInsert)
+		testQueryWithContext(ctx, t, e, insertion.selectQuery, insertion.expectedSelect)
+	}
+}
+
+func TestInsertIntoErrors(t *testing.T) {
+	var expectedFailures = []struct {
+		name  string
+		query string
+	}{
+		{
+			"too few values",
+			"INSERT INTO mytable (s, i) VALUES ('x');",
+		},
+		{
+			"too many values one column",
+			"INSERT INTO mytable (s) VALUES ('x', 999);",
+		},
+		{
+			"too many values two columns",
+			"INSERT INTO mytable (i, s) VALUES (999, 'x', 'y');",
+		},
+		{
+			"too few values no columns specified",
+			"INSERT INTO mytable VALUES (999);",
+		},
+		{
+			"too many values no columns specified",
+			"INSERT INTO mytable VALUES (999, 'x', 'y');",
+		},
+		{
+			"non-existent column",
+			"INSERT INTO mytable (i, s, z) VALUES (999, 'x', 999);",
+		},
+		{
+			"duplicate column",
+			"INSERT INTO mytable (i, s, s) VALUES (999, 'x', 'x');",
+		},
+	}
+
+	for _, expectedFailure := range expectedFailures {
+		t.Run(expectedFailure.name, func(t *testing.T) {
+			_, _, err := newEngine(t).Query(newCtx(), expectedFailure.query)
+			require.Error(t, err)
+		})
+	}
 }
 
 const testNumPartitions = 5
@@ -2213,6 +2372,26 @@ func newEngineWithParallelism(t *testing.T, parallelism int) *sqle.Engine {
 		sql.NewRow(nil, nil, nil),
 	)
 
+	typestable := memory.NewPartitionedTable("typestable", sql.Schema{
+		{Name: "id", Type: sql.Int64, Source: "typestable"},
+		{Name: "i8", Type: sql.Int8, Source: "typestable", Nullable: true},
+		{Name: "i16", Type: sql.Int16, Source: "typestable", Nullable: true},
+		{Name: "i32", Type: sql.Int32, Source: "typestable", Nullable: true},
+		{Name: "i64", Type: sql.Int64, Source: "typestable", Nullable: true},
+		{Name: "u8", Type: sql.Uint8, Source: "typestable", Nullable: true},
+		{Name: "u16", Type: sql.Uint16, Source: "typestable", Nullable: true},
+		{Name: "u32", Type: sql.Uint32, Source: "typestable", Nullable: true},
+		{Name: "u64", Type: sql.Uint64, Source: "typestable", Nullable: true},
+		{Name: "f32", Type: sql.Float32, Source: "typestable", Nullable: true},
+		{Name: "f64", Type: sql.Float64, Source: "typestable", Nullable: true},
+		{Name: "ti", Type: sql.Timestamp, Source: "typestable", Nullable: true},
+		{Name: "da", Type: sql.Date, Source: "typestable", Nullable: true},
+		{Name: "te", Type: sql.Text, Source: "typestable", Nullable: true},
+		{Name: "bo", Type: sql.Boolean, Source: "typestable", Nullable: true},
+		{Name: "js", Type: sql.JSON, Source: "typestable", Nullable: true},
+		{Name: "bl", Type: sql.Blob, Source: "typestable", Nullable: true},
+	}, testNumPartitions)
+
 	db := memory.NewDatabase("mydb")
 	db.AddTable("mytable", table)
 	db.AddTable("othertable", table2)
@@ -2220,6 +2399,7 @@ func newEngineWithParallelism(t *testing.T, parallelism int) *sqle.Engine {
 	db.AddTable("bigtable", bigtable)
 	db.AddTable("floattable", floatTable)
 	db.AddTable("niltable", nilTable)
+	db.AddTable("typestable", typestable)
 
 	db2 := memory.NewDatabase("foo")
 	db2.AddTable("other_table", table4)