Repository: src-d/go-mysql-server
Pull request #845 (2019-10-17 18:40:51)

Drop / create table
Implemented DROP TABLE. Also:

* 24-bit integers
* Primary key attribute for columns
* Deprecated Alterable in favor of TableCreator. Primary difference is the new interface takes a context argument.


Patch:
diff --git a/memory/database.go b/memory/database.go
index b5b87a7..133c4f5 100644
--- a/memory/database.go
+++ b/memory/database.go
@@ -33,8 +33,8 @@ func (d *Database) AddTable(name string, t sql.Table) {
 	d.tables[name] = t
 }
 
-// Create creates a table with the given name and schema
-func (d *Database) Create(name string, schema sql.Schema) error {
+// CreateTable creates a table with the given name and schema
+func (d *Database) CreateTable(ctx *sql.Context, name string, schema sql.Schema) error {
 	_, ok := d.tables[name]
 	if ok {
 		return sql.ErrTableAlreadyExists.New(name)
@@ -43,3 +43,15 @@ func (d *Database) Create(name string, schema sql.Schema) error {
 	d.tables[name] = NewTable(name, schema)
 	return nil
 }
+
+// DropTable drops the table with the given name
+func (d *Database) DropTable(ctx *sql.Context, name string) error {
+	_, ok := d.tables[name]
+	if !ok {
+		return sql.ErrTableNotFound.New(name)
+	}
+
+	delete(d.tables, name)
+	return nil
+}
+
diff --git a/sql/core.go b/sql/core.go
index 7a88750..2d49ece 100644
--- a/sql/core.go
+++ b/sql/core.go
@@ -230,9 +230,14 @@ type Database interface {
 	Tables() map[string]Table
 }
 
-// Alterable should be implemented by databases that can handle DDL statements
-type Alterable interface {
-	Create(name string, schema Schema) error
+// TableCreator should be implemented by databases that can create new tables.
+type TableCreator interface {
+	CreateTable(ctx *Context, name string, schema Schema) error
+}
+
+// TableDropper should be implemented by databases that can drop tables.
+type TableDropper interface {
+	DropTable(ctx *Context, name string) error
 }
 
 // Lockable should be implemented by tables that can be locked and unlocked.
diff --git a/sql/parse/parse.go b/sql/parse/parse.go
index 5b6b9ca..f4dfe6f 100644
--- a/sql/parse/parse.go
+++ b/sql/parse/parse.go
@@ -47,6 +47,17 @@ var (
 	unlockTablesRegex    = regexp.MustCompile(`^unlock\s+tables$`)
 	lockTablesRegex      = regexp.MustCompile(`^lock\s+tables\s`)
 	setRegex             = regexp.MustCompile(`^set\s+`)
+	createViewRegex      = regexp.MustCompile(`^create\s+view\s+`)
+)
+
+// These constants aren't exported from vitess for some reason. This could be removed if we changed this.
+const (
+	colKeyNone sqlparser.ColumnKeyOption = iota
+	colKeyPrimary
+	colKeySpatialKey
+	colKeyUnique
+	colKeyUniqueKey
+	colKey
 )
 
 // Parse parses the given SQL sentence and returns the corresponding node.
@@ -93,6 +104,9 @@ func Parse(ctx *sql.Context, query string) (sql.Node, error) {
 		return parseLockTables(ctx, s)
 	case setRegex.MatchString(lowerQuery):
 		s = fixSetQuery(s)
+	case createViewRegex.MatchString(lowerQuery):
+		// CREATE VIEW parses as a CREATE DDL statement with an empty table spec
+		return nil, ErrUnsupportedFeature.New("CREATE VIEW")
 	}
 
 	stmt, err := sqlparser.Parse(s)
@@ -144,7 +158,12 @@ func convert(ctx *sql.Context, stmt sqlparser.Statement, query string) (sql.Node
 	case *sqlparser.Insert:
 		return convertInsert(ctx, n)
 	case *sqlparser.DDL:
-		return convertDDL(n)
+		// unlike other statements, DDL statements have loose parsing by default
+		ddl, err := sqlparser.ParseStrictDDL(query)
+		if err != nil {
+			return nil, err
+		}
+		return convertDDL(ddl.(*sqlparser.DDL))
 	case *sqlparser.Set:
 		return convertSet(ctx, n)
 	case *sqlparser.Use:
@@ -354,13 +373,23 @@ func convertDDL(c *sqlparser.DDL) (sql.Node, error) {
 	switch c.Action {
 	case sqlparser.CreateStr:
 		return convertCreateTable(c)
+	case sqlparser.DropStr:
+		return convertDropTable(c)
 	default:
 		return nil, ErrUnsupportedSyntax.New(c)
 	}
 }
 
+func convertDropTable(c *sqlparser.DDL) (sql.Node, error) {
+	tableNames := make([]string, len(c.FromTables))
+	for i, t := range c.FromTables {
+		tableNames[i] = t.Name.String()
+	}
+	return plan.NewDropTable(sql.UnresolvedDatabase(""), c.IfExists, tableNames...), nil
+}
+
 func convertCreateTable(c *sqlparser.DDL) (sql.Node, error) {
-	schema, err := columnDefinitionToSchema(c.TableSpec.Columns)
+	schema, err := tableSpecToSchema(c.TableSpec)
 	if err != nil {
 		return nil, err
 	}
@@ -462,6 +491,7 @@ func convertUpdate(ctx *sql.Context, d *sqlparser.Update) (sql.Node, error) {
 		if err != nil {
 			return nil, err
 		}
+
 	}
 
 	if d.Limit != nil {
@@ -474,27 +504,56 @@ func convertUpdate(ctx *sql.Context, d *sqlparser.Update) (sql.Node, error) {
 	return plan.NewUpdate(node, updateExprs), nil
 }
 
-func columnDefinitionToSchema(colDef []*sqlparser.ColumnDefinition) (sql.Schema, error) {
+func tableSpecToSchema(tableSpec *sqlparser.TableSpec) (sql.Schema, error) {
 	var schema sql.Schema
-	for _, cd := range colDef {
-		typ := cd.Type
-		internalTyp, err := sql.MysqlTypeToType(typ.SQLType())
+	for _, cd := range tableSpec.Columns {
+		column, err := getColumn(cd, tableSpec.Indexes)
 		if err != nil {
 			return nil, err
 		}
 
-		schema = append(schema, &sql.Column{
-			Nullable: !bool(typ.NotNull),
-			Type:     internalTyp,
-			Name:     cd.Name.String(),
-			// TODO
-			Default: nil,
-		})
+		schema = append(schema, column)
 	}
 
 	return schema, nil
 }
 
+// getColumn returns the sql.Column for the column definition given, as part of a create table statement.
+func getColumn(cd *sqlparser.ColumnDefinition, indexes []*sqlparser.IndexDefinition) (*sql.Column, error) {
+	typ := cd.Type
+	internalTyp, err := sql.MysqlTypeToType(typ.SQLType())
+	if err != nil {
+		return nil, err
+	}
+
+	// Primary key info can either be specified in the column's type info (for in-line declarations), or in a slice of
+	// indexes attached to the table def. We have to check both places to find if a column is part of the primary key
+	isPkey := cd.Type.KeyOpt == colKeyPrimary
+
+	if !isPkey {
+	OuterLoop:
+		for _, index := range indexes {
+			if index.Info.Primary {
+				for _, indexCol := range index.Columns {
+					if indexCol.Column.Equal(cd.Name) {
+						isPkey = true
+						break OuterLoop
+					}
+				}
+			}
+		}
+	}
+
+	return &sql.Column{
+		Nullable:   !bool(typ.NotNull),
+		Type:       internalTyp,
+		Name:       cd.Name.String(),
+		PrimaryKey: isPkey,
+		// TODO
+		Default: nil,
+	}, nil
+}
+
 func columnsToStrings(cols sqlparser.Columns) []string {
 	res := make([]string, len(cols))
 	for i, c := range cols {
diff --git a/sql/plan/ddl.go b/sql/plan/ddl.go
index 3eb9733..d9acf8a 100644
--- a/sql/plan/ddl.go
+++ b/sql/plan/ddl.go
@@ -1,12 +1,14 @@
 package plan
 
 import (
+	"fmt"
 	"github.com/src-d/go-mysql-server/sql"
 	"gopkg.in/src-d/go-errors.v1"
 )
 
 // ErrCreateTable is thrown when the database doesn't support table creation
-var ErrCreateTable = errors.NewKind("tables cannot be created on database %s")
+var ErrCreateTableNotSupported = errors.NewKind("tables cannot be created on database %s")
+var ErrDropTableNotSupported = errors.NewKind("tables cannot be dropped on database %s")
 
 // CreateTable is a node describing the creation of some table.
 type CreateTable struct {
@@ -50,12 +52,12 @@ func (c *CreateTable) Resolved() bool {
 
 // RowIter implements the Node interface.
 func (c *CreateTable) RowIter(s *sql.Context) (sql.RowIter, error) {
-	d, ok := c.db.(sql.Alterable)
-	if !ok {
-		return nil, ErrCreateTable.New(c.db.Name())
+	creatable, ok := c.db.(sql.TableCreator)
+	if ok {
+		return sql.RowsToRowIter(), creatable.CreateTable(s, c.name, c.schema)
 	}
 
-	return sql.RowsToRowIter(), d.Create(c.name, c.schema)
+	return nil, ErrCreateTableNotSupported.New(c.db.Name())
 }
 
 // Schema implements the Node interface.
@@ -75,3 +77,86 @@ func (c *CreateTable) WithChildren(children ...sql.Node) (sql.Node, error) {
 func (c *CreateTable) String() string {
 	return "CreateTable"
 }
+
+// DropTable is a node describing dropping one or more tables
+type DropTable struct {
+	db       sql.Database
+	names    []string
+	ifExists bool
+}
+
+// NewDropTable creates a new DropTable node
+func NewDropTable(db sql.Database, ifExists bool, tableNames ...string) *DropTable {
+	return &DropTable{
+		db:       db,
+		names:    tableNames,
+		ifExists: ifExists,
+	}
+}
+
+var _ sql.Databaser = (*DropTable)(nil)
+
+// Database implements the sql.Databaser interface.
+func (d *DropTable) Database() sql.Database {
+	return d.db
+}
+
+// WithDatabase implements the sql.Databaser interface.
+func (d *DropTable) WithDatabase(db sql.Database) (sql.Node, error) {
+	nc := *d
+	nc.db = db
+	return &nc, nil
+}
+
+// Resolved implements the Resolvable interface.
+func (d *DropTable) Resolved() bool {
+	_, ok := d.db.(sql.UnresolvedDatabase)
+	return !ok
+}
+
+// RowIter implements the Node interface.
+func (d *DropTable) RowIter(s *sql.Context) (sql.RowIter, error) {
+	droppable, ok := d.db.(sql.TableDropper)
+	if !ok {
+		return nil, ErrDropTableNotSupported.New(d.db.Name())
+	}
+
+	var err error
+	for _, tableName := range d.names {
+		_, ok := d.db.Tables()[tableName]
+		if !ok {
+			if d.ifExists {
+				continue
+			}
+			return nil, sql.ErrTableNotFound.New(tableName)
+		}
+		err = droppable.DropTable(s, tableName)
+		if err != nil {
+			break
+		}
+	}
+
+	return sql.RowsToRowIter(), err
+}
+
+// Schema implements the Node interface.
+func (d *DropTable) Schema() sql.Schema { return nil }
+
+// Children implements the Node interface.
+func (d *DropTable) Children() []sql.Node { return nil }
+
+// WithChildren implements the Node interface.
+func (d *DropTable) WithChildren(children ...sql.Node) (sql.Node, error) {
+	if len(children) != 0 {
+		return nil, sql.ErrInvalidChildrenNumber.New(d, len(children), 0)
+	}
+	return d, nil
+}
+
+func (d *DropTable) String() string {
+	ifExists := ""
+	if d.ifExists {
+		ifExists = "if exists "
+	}
+	return fmt.Sprintf("Drop table %s%s", ifExists, d.names)
+}
diff --git a/sql/type.go b/sql/type.go
index 94e8253..6327244 100644
--- a/sql/type.go
+++ b/sql/type.go
@@ -124,6 +124,8 @@ type Column struct {
 	Nullable bool
 	// Source is the name of the table this column came from.
 	Source string
+	// PrimaryKey is true if the column is part of the primary key for its table.
+	PrimaryKey bool
 }
 
 // Check ensures the value is correct for this column.
@@ -184,6 +186,10 @@ var (
 	Int16 = numberT{t: sqltypes.Int16}
 	// Uint16 is an unsigned integer of 16 bits
 	Uint16 = numberT{t: sqltypes.Uint16}
+	// Int24 is an integer of 24 bits.
+	Int24 = numberT{t: sqltypes.Int24}
+	// Uint24 is an unsigned integer of 24 bits.
+	Uint24 = numberT{t: sqltypes.Uint24}
 	// Int32 is an integer of 32 bits.
 	Int32 = numberT{t: sqltypes.Int32}
 	// Uint32 is an unsigned integer of 32 bits.
@@ -246,12 +252,16 @@ func MysqlTypeToType(sql query.Type) (Type, error) {
 		return Int16, nil
 	case sqltypes.Uint16:
 		return Uint16, nil
+	case sqltypes.Int24:
+		return Int24, nil
+	case sqltypes.Uint24:
+		return Uint24, nil
 	case sqltypes.Int32:
 		return Int32, nil
-	case sqltypes.Int64:
-		return Int64, nil
 	case sqltypes.Uint32:
 		return Uint32, nil
+	case sqltypes.Int64:
+		return Int64, nil
 	case sqltypes.Uint64:
 		return Uint64, nil
 	case sqltypes.Float32:


Test patch:
diff --git a/engine_test.go b/engine_test.go
index e1274e2..5070589 100644
--- a/engine_test.go
+++ b/engine_test.go
@@ -2443,7 +2443,7 @@ func TestAmbiguousColumnResolution(t *testing.T) {
 	require.Equal(expected, rs)
 }
 
-func TestDDL(t *testing.T) {
+func TestCreateTable(t *testing.T) {
 	require := require.New(t)
 
 	e := newEngine(t)
@@ -2474,6 +2474,83 @@ func TestDDL(t *testing.T) {
 	}
 
 	require.Equal(s, testTable.Schema())
+
+	testQuery(t, e,
+		"CREATE TABLE t2 (a INTEGER NOT NULL PRIMARY KEY, "+
+				"b VARCHAR(10) NOT NULL)",
+		[]sql.Row(nil),
+	)
+
+	db, err = e.Catalog.Database("mydb")
+	require.NoError(err)
+
+	testTable, ok = db.Tables()["t2"]
+	require.True(ok)
+
+	s = sql.Schema{
+		{Name: "a", Type: sql.Int32, Nullable: false, PrimaryKey: true, Source: "t2"},
+		{Name: "b", Type: sql.Text, Nullable: false, Source: "t2"},
+	}
+
+	require.Equal(s, testTable.Schema())
+
+	testQuery(t, e,
+		"CREATE TABLE t3(a INTEGER NOT NULL,"+
+				"b TEXT NOT NULL,"+
+				"c bool, primary key (a,b))",
+		[]sql.Row(nil),
+	)
+
+	db, err = e.Catalog.Database("mydb")
+	require.NoError(err)
+
+	testTable, ok = db.Tables()["t3"]
+	require.True(ok)
+
+	s = sql.Schema{
+		{Name: "a", Type: sql.Int32, Nullable: false, PrimaryKey: true, Source: "t3"},
+		{Name: "b", Type: sql.Text, Nullable: false, PrimaryKey: true, Source: "t3"},
+		{Name: "c", Type: sql.Uint8, Nullable: true, Source: "t3"},
+	}
+
+	require.Equal(s, testTable.Schema())
+}
+
+func TestDropTable(t *testing.T) {
+	require := require.New(t)
+
+	e := newEngine(t)
+	db, err := e.Catalog.Database("mydb")
+	require.NoError(err)
+
+	_, ok := db.Tables()["mytable"]
+	require.True(ok)
+
+	testQuery(t, e,
+		"DROP TABLE IF EXISTS mytable, not_exist",
+		[]sql.Row(nil),
+	)
+
+	_, ok = db.Tables()["mytable"]
+	require.False(ok)
+
+	_, ok = db.Tables()["othertable"]
+	require.True(ok)
+	_, ok = db.Tables()["tabletest"]
+	require.True(ok)
+
+	testQuery(t, e,
+		"DROP TABLE IF EXISTS othertable, tabletest",
+		[]sql.Row(nil),
+	)
+
+	_, ok = db.Tables()["othertable"]
+	require.False(ok)
+	_, ok = db.Tables()["tabletest"]
+	require.False(ok)
+
+	_, _, err = e.Query(newCtx(), "DROP TABLE not_exist")
+	require.Error(err)
 }
 
 func TestNaturalJoin(t *testing.T) {
diff --git a/memory/database_test.go b/memory/database_test.go
index e41c9e2..c6a2f91 100644
--- a/memory/database_test.go
+++ b/memory/database_test.go
@@ -19,8 +19,7 @@ func TestDatabase_AddTable(t *testing.T) {
 	tables := db.Tables()
 	require.Equal(0, len(tables))
 
-	var altDb sql.Alterable = db
-	err := altDb.Create("test_table", nil)
+	err := db.CreateTable(sql.NewEmptyContext(), "test_table", nil)
 	require.NoError(err)
 
 	tables = db.Tables()
@@ -29,6 +28,6 @@ func TestDatabase_AddTable(t *testing.T) {
 	require.True(ok)
 	require.NotNil(tt)
 
-	err = altDb.Create("test_table", nil)
+	err = db.CreateTable(sql.NewEmptyContext(), "test_table", nil)
 	require.Error(err)
 }
diff --git a/sql/parse/parse_test.go b/sql/parse/parse_test.go
index ad8bfc6..b66dca9 100644
--- a/sql/parse/parse_test.go
+++ b/sql/parse/parse_test.go
@@ -51,6 +51,60 @@ var fixtures = map[string]sql.Node{
 			Nullable: true,
 		}},
 	),
+	`CREATE TABLE t1(a INTEGER NOT NULL PRIMARY KEY, b TEXT)`: plan.NewCreateTable(
+		sql.UnresolvedDatabase(""),
+		"t1",
+		sql.Schema{{
+			Name:     "a",
+			Type:     sql.Int32,
+			Nullable: false,
+			PrimaryKey: true,
+		}, {
+			Name:     "b",
+			Type:     sql.Text,
+			Nullable: true,
+			PrimaryKey: false,
+		}},
+	),
+	`CREATE TABLE t1(a INTEGER, b TEXT, PRIMARY KEY (a))`: plan.NewCreateTable(
+		sql.UnresolvedDatabase(""),
+		"t1",
+		sql.Schema{{
+			Name:     "a",
+			Type:     sql.Int32,
+			Nullable: true,
+			PrimaryKey: true,
+		}, {
+			Name:     "b",
+			Type:     sql.Text,
+			Nullable: true,
+			PrimaryKey: false,
+		}},
+	),
+	`CREATE TABLE t1(a INTEGER, b TEXT, PRIMARY KEY (a, b))`: plan.NewCreateTable(
+		sql.UnresolvedDatabase(""),
+		"t1",
+		sql.Schema{{
+			Name:     "a",
+			Type:     sql.Int32,
+			Nullable: true,
+			PrimaryKey: true,
+		}, {
+			Name:     "b",
+			Type:     sql.Text,
+			Nullable: true,
+			PrimaryKey: true,
+		}},
+	),
+	`DROP TABLE foo;`: plan.NewDropTable(
+		sql.UnresolvedDatabase(""), false, "foo",
+	),
+	`DROP TABLE IF EXISTS foo;`: plan.NewDropTable(
+		sql.UnresolvedDatabase(""), true, "foo",
+	),
+	`DROP TABLE IF EXISTS foo, bar, baz;`: plan.NewDropTable(
+		sql.UnresolvedDatabase(""), true, "foo", "bar", "baz",
+	),
 	`DESCRIBE TABLE foo;`: plan.NewDescribe(
 		plan.NewUnresolvedTable("foo", ""),
 	),
@@ -1242,6 +1296,7 @@ var fixturesErrors = map[string]*errors.Kind{
 	`SELECT INTERVAL 1 DAY + INTERVAL 1 DAY`:                  ErrUnsupportedSyntax,
 	`SELECT '2018-05-01' + (INTERVAL 1 DAY + INTERVAL 1 DAY)`: ErrUnsupportedSyntax,
 	`SELECT AVG(DISTINCT foo) FROM b`:                         ErrUnsupportedSyntax,
+	`CREATE VIEW view1 AS SELECT x FROM t1 WHERE x>0`:         ErrUnsupportedFeature,
 }
 
 func TestParseErrors(t *testing.T) {
diff --git a/sql/plan/ddl_test.go b/sql/plan/ddl_test.go
index a644857..a226a63 100644
--- a/sql/plan/ddl_test.go
+++ b/sql/plan/ddl_test.go
@@ -22,15 +22,7 @@ func TestCreateTable(t *testing.T) {
 		{Name: "c2", Type: sql.Int32},
 	}
 
-	c := NewCreateTable(db, "testTable", s)
-
-	rows, err := c.RowIter(sql.NewEmptyContext())
-
-	require.NoError(err)
-
-	r, err := rows.Next()
-	require.Equal(err, io.EOF)
-	require.Nil(r)
+	createTable(t, db, "testTable", s)
 
 	tables = db.Tables()
 
@@ -43,3 +35,59 @@ func TestCreateTable(t *testing.T) {
 		require.Equal("testTable", s.Source)
 	}
 }
+
+func TestDropTable(t *testing.T) {
+	require := require.New(t)
+
+	db := memory.NewDatabase("test")
+
+	s := sql.Schema{
+		{Name: "c1", Type: sql.Text},
+		{Name: "c2", Type: sql.Int32},
+	}
+
+	createTable(t, db, "testTable1", s)
+	createTable(t, db, "testTable2", s)
+	createTable(t, db, "testTable3", s)
+
+	d := NewDropTable(db, false, "testTable1", "testTable2")
+	rows, err := d.RowIter(sql.NewEmptyContext())
+	require.NoError(err)
+
+	r, err := rows.Next()
+	require.Equal(err, io.EOF)
+	require.Nil(r)
+
+	_, ok := db.Tables()["testTable1"]
+	require.False(ok)
+	_, ok = db.Tables()["testTable2"]
+	require.False(ok)
+	_, ok = db.Tables()["testTable3"]
+	require.True(ok)
+
+	d = NewDropTable(db, false, "testTable1")
+	_, err = d.RowIter(sql.NewEmptyContext())
+	require.Error(err)
+
+	d = NewDropTable(db, true, "testTable1")
+	_, err = d.RowIter(sql.NewEmptyContext())
+	require.NoError(err)
+
+	d = NewDropTable(db, true, "testTable1", "testTable2", "testTable3")
+	_, err = d.RowIter(sql.NewEmptyContext())
+	require.NoError(err)
+
+	_, ok = db.Tables()["testTable3"]
+	require.False(ok)
+}
+
+func createTable(t *testing.T, db sql.Database, name string, schema sql.Schema) {
+	c := NewCreateTable(db, name, schema)
+
+	rows, err := c.RowIter(sql.NewEmptyContext())
+	require.NoError(t, err)
+
+	r, err := rows.Next()
+	require.Equal(t, err, io.EOF)
+	require.Nil(t, r)
+}
\ No newline at end of file
diff --git a/sql/type_test.go b/sql/type_test.go
index 5ca9ac9..0dab602 100644
--- a/sql/type_test.go
+++ b/sql/type_test.go
@@ -192,6 +192,8 @@ func TestNumberComparison(t *testing.T) {
 		{Uint8, uint8(42)},
 		{Int16, int16(42)},
 		{Uint16, uint16(42)},
+		{Int24, int32(42)},
+		{Uint24, uint32(42)},
 		{Int32, int32(42)},
 		{Uint32, uint32(42)},
 		{Int64, int64(42)},