Merge pull request #18 from liquidata-inc/zachmu/drop-table

Drop and create table support, as well as support for 24-bit integers.

Author: zachmu

memory/database.go

@@ -43,3 +43,25 @@ func (d *Database) Create(name string, schema sql.Schema) error {
 	d.tables[name] = NewTable(name, schema)
 	return nil
 }
+
+// Create 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)
+	}
+
+	d.tables[name] = NewTable(name, schema)
+	return nil
+}
+
+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
+}
+

sql/core.go

@@ -230,11 +230,22 @@ type Database interface {
 	Tables() map[string]Table
 }
 
+// DEPRECATED. Use TableCreator and TableDropper.
 // 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.
 type Lockable interface {
 	Nameable

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 {

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) {

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 ErrDropTableNotSupported = errors.NewKind("tables cannot be dropped on database %s")
 
 // CreateTable is a node describing the creation of some table.
 type CreateTable struct {
@@ -15,6 +17,13 @@ type CreateTable struct {
 	schema sql.Schema
 }
 
+// DropTable is a node describing dropping a table
+type DropTable struct {
+	db       sql.Database
+	names    []string
+	ifExists bool
+}
+
 // NewCreateTable creates a new CreateTable node
 func NewCreateTable(db sql.Database, name string, schema sql.Schema) *CreateTable {
 	for _, s := range schema {
@@ -50,6 +59,12 @@ func (c *CreateTable) Resolved() bool {
 
 // RowIter implements the Node interface.
 func (c *CreateTable) RowIter(s *sql.Context) (sql.RowIter, error) {
+	creatable, ok := c.db.(sql.TableCreator)
+	if ok {
+		return sql.RowsToRowIter(), creatable.CreateTable(s, c.name, c.schema)
+	}
+
+	// TODO: phase out this interface
 	d, ok := c.db.(sql.Alterable)
 	if !ok {
 		return nil, ErrCreateTable.New(c.db.Name())
@@ -75,3 +90,79 @@ func (c *CreateTable) WithChildren(children ...sql.Node) (sql.Node, error) {
 func (c *CreateTable) String() string {
 	return "CreateTable"
 }
+
+// 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)
+}