Merge pull request #845 from liquidata-inc/ld-master

Drop / create table

Author: erizocosmico

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

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
+}
+

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

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.

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