Vinai/create table select (#449)

This table adds basic functionality for CREATE TABLE SELECT

Author: Vinai Rachakonda

enginetest/enginetests.go

@@ -1343,6 +1343,34 @@ func TestCreateTable(t *testing.T, harness Harness) {
 		require.True(t, indexFound)
 	})
 
+	t.Run("CREATE TABLE (SELECT * )", func(t *testing.T) {
+		TestQuery(t, harness, e, "CREATE TABLE t10 (a INTEGER NOT NULL PRIMARY KEY, "+
+			"b VARCHAR(10))", []sql.Row(nil), nil, nil)
+		TestQuery(t, harness, e, `INSERT INTO t10 VALUES (1, "1"), (2, "2")`, []sql.Row{sql.Row{sql.OkResult{RowsAffected: 0x2, InsertID: 0x0, Info: fmt.Stringer(nil)}}}, nil, nil)
+
+		// Create the table with the data from t10
+		TestQuery(t, harness, e, "CREATE TABLE t10a SELECT * from t10", []sql.Row{sql.Row{sql.OkResult{RowsAffected: 0x2, InsertID: 0x0, Info: fmt.Stringer(nil)}}}, nil, nil)
+
+		db, err := e.Catalog.Database("mydb")
+		require.NoError(t, err)
+
+		t10Table, ok, err := db.GetTableInsensitive(ctx, "t10")
+		require.NoError(t, err)
+		require.True(t, ok)
+		t10aTable, ok, err := db.GetTableInsensitive(ctx, "t10a")
+		require.NoError(t, err)
+		require.True(t, ok)
+
+		require.Equal(t, sql.Schema{
+			{Name: "a", Type: sql.Int32, Nullable: false, PrimaryKey: true, Source: "t10"},
+			{Name: "b", Type: sql.MustCreateStringWithDefaults(sqltypes.VarChar, 10), Nullable: true, Source: "t10"},
+		}, t10Table.Schema())
+		require.Equal(t, sql.Schema{
+			{Name: "a", Type: sql.Int32, Nullable: false, PrimaryKey: true, Source: "t10a"},
+			{Name: "b", Type: sql.MustCreateStringWithDefaults(sqltypes.VarChar, 10), Nullable: true, Source: "t10a"},
+		}, t10aTable.Schema())
+	})
+
 	//TODO: Implement "CREATE TABLE otherDb.tableName"
 }
 

enginetest/script_queries.go

@@ -942,6 +942,44 @@ var ScriptTests = []ScriptTest{
 			},
 		},
 	},
+	{
+		Name: "CREATE TABLE SELECT Queries",
+		SetUpScript: []string{
+			`CREATE TABLE t1 (pk int PRIMARY KEY, v1 varchar(10))`,
+			`INSERT INTO t1 VALUES (1,"1"), (2,"2"), (3,"3")`,
+			`CREATE TABLE t2 AS SELECT * FROM t1`,
+			//`CREATE TABLE t3(v0 int) AS SELECT pk FROM t1`, // parser problems
+			`CREATE TABLE t3 AS SELECT pk FROM t1`,
+			`CREATE TABLE t4 AS SELECT pk, v1 FROM t1`,
+			`CREATE TABLE t5 SELECT * FROM t1 ORDER BY pk LIMIT 1`,
+		},
+		Assertions: []ScriptTestAssertion{
+			{
+				Query:    `SELECT * FROM t2`,
+				Expected: []sql.Row{{1, "1"}, {2, "2"}, {3, "3"}},
+			},
+			{
+				Query:    `SELECT * FROM t3`,
+				Expected: []sql.Row{{1}, {2}, {3}},
+			},
+			{
+				Query:    `SELECT * FROM t4`,
+				Expected: []sql.Row{{1, "1"}, {2, "2"}, {3, "3"}},
+			},
+			{
+				Query:    `SELECT * FROM t5`,
+				Expected: []sql.Row{{1, "1"}},
+			},
+			{
+				Query: `CREATE TABLE test SELECT * FROM t1`,
+				Expected: []sql.Row{sql.Row{sql.OkResult{
+					RowsAffected: 3,
+					InsertID:     0,
+					Info:         nil,
+				}}},
+			},
+		},
+	},
 }
 
 var CreateCheckConstraintsScripts = []ScriptTest{

go.sum

@@ -83,7 +83,6 @@ github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/testify v0.0.0-20161117074351-18a02ba4a312/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
-github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
@@ -149,7 +148,6 @@ gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8
 gopkg.in/src-d/go-errors.v1 v1.0.0 h1:cooGdZnCjYbeS1zb1s6pVAAimTdKceRrpn7aKOnNIfc=
 gopkg.in/src-d/go-errors.v1 v1.0.0/go.mod h1:q1cBlomlw2FnDBDNGlnh6X0jPihy+QxZfMMNxPCbdYg=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
-gopkg.in/yaml.v2 v2.2.4 h1:/eiJrUcujPVeJ3xlSWaiNi3uSVmDGBK1pDHUHAnao1I=
 gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

sql/analyzer/resolve_create_select.go

@@ -0,0 +1,83 @@
+package analyzer
+
+import (
+	"github.com/dolthub/go-mysql-server/sql"
+	"github.com/dolthub/go-mysql-server/sql/plan"
+)
+
+func resolveCreateSelect(ctx *sql.Context, a *Analyzer, n sql.Node, scope *Scope) (sql.Node, error) {
+	planCreate, ok := n.(*plan.CreateTable)
+	if !ok || planCreate.Select() == nil {
+		return n, nil
+	}
+
+	analyzedSelect, err := a.Analyze(ctx, planCreate.Select(), scope)
+	if err != nil {
+		return nil, err
+	}
+
+	// Get the correct schema of the CREATE TABLE based on the select query
+	inputSpec := planCreate.TableSpec()
+	selectSchema := analyzedSelect.Schema()
+	mergedSchema := mergeSchemas(inputSpec.Schema, selectSchema)
+	newSch := make(sql.Schema, len(mergedSchema))
+
+	for i, col := range mergedSchema {
+		tempCol := *col
+		tempCol.Source = planCreate.Name()
+		newSch[i] = &tempCol
+	}
+
+	newSpec := inputSpec.WithSchema(newSch)
+
+	newCreateTable := plan.NewCreateTable(planCreate.Database(), planCreate.Name(), planCreate.IfNotExists(), planCreate.Temporary(), newSpec)
+	analyzedCreate, err := a.Analyze(ctx, newCreateTable, scope)
+	if err != nil {
+		return nil, err
+	}
+
+	return plan.NewTableCopier(planCreate.Database(), stripQueryProcess(analyzedCreate), stripQueryProcess(analyzedSelect), plan.CopierProps{}), nil
+}
+
+// mergeSchemas takes in the table spec of the CREATE TABLE and merges it with the schema used by the
+// select query. The ultimate structure for the new table will be [CREATE TABLE exclusive columns, columns with the same
+// name, SELECT exclusive columns]
+func mergeSchemas(inputSchema sql.Schema, selectSchema sql.Schema) sql.Schema {
+	if inputSchema == nil {
+		return selectSchema
+	}
+
+	// Get the matching columns between the two via name
+	matchingColumns := make([]*sql.Column, 0)
+	leftExclusive := make([]*sql.Column, 0)
+	for _, col := range inputSchema {
+		found := false
+		for _, col2 := range selectSchema {
+			if col.Name == col2.Name {
+				matchingColumns = append(matchingColumns, col)
+				found = true
+			}
+		}
+
+		if !found {
+			leftExclusive = append(leftExclusive, col)
+		}
+	}
+
+	rightExclusive := make([]*sql.Column, 0)
+	for _, col := range selectSchema {
+		found := false
+		for _, col2 := range inputSchema {
+			if col.Name == col2.Name {
+				found = true
+				break
+			}
+		}
+
+		if !found {
+			rightExclusive = append(rightExclusive, col)
+		}
+	}
+
+	return append(append(leftExclusive, matchingColumns...), rightExclusive...)
+}

sql/analyzer/rules.go

@@ -32,6 +32,7 @@ var OnceBeforeDefault = []Rule{
 	{"load_check_constraints", loadChecks},
 	{"resolve_set_variables", resolveSetVariables},
 	{"resolve_create_like", resolveCreateLike},
+	{"resolve_create_select", resolveCreateSelect},
 	{"resolve_subqueries", resolveSubqueries},
 	{"resolve_unions", resolveUnions},
 	{"resolve_describe_query", resolveDescribeQuery},

sql/core.go

@@ -569,6 +569,13 @@ type TemporaryTableDatabase interface {
 	GetAllTemporaryTables(ctx *Context) ([]Table, error)
 }
 
+// TableCopierDatabase is a database that can copy a source table's data (without preserving indexed, fks, etc.) into
+// another destination table.
+type TableCopierDatabase interface {
+	// CopyTableData copies the sourceTable data to the destinationTable and returns the number of rows copied.
+	CopyTableData(ctx *Context, sourceTable string, destinationTable string) (uint64, error)
+}
+
 // GetTableInsensitive implements a case insensitive map lookup for tables keyed off of the table name.
 // Looks for exact matches first.  If no exact matches are found then any table matching the name case insensitively
 // should be returned.  If there is more than one table that matches a case insensitive comparison the resolution

sql/errors.go

@@ -289,6 +289,10 @@ var (
 	// ErrInvalidSyntax is returned for syntax errors that aren't picked up by the parser, e.g. the wrong type of
 	// expression used in part of statement.
 	ErrInvalidSyntax = errors.NewKind("Invalid syntax: %s")
+
+	// ErrTableCopyingNotSupported is returned when a table invokes the TableCopierDatabase interface's
+	// CopyTableData method without supporting the interface
+	ErrTableCopyingNotSupported = errors.NewKind("error: Table copying not supported")
 )
 
 func CastSQLError(err error) (*mysql.SQLError, bool) {

sql/parse/parse.go

@@ -1227,6 +1227,19 @@ func convertCreateTable(ctx *sql.Context, c *sqlparser.DDL) (sql.Node, error) {
 		), nil
 	}
 
+	// In the case that no table spec is given but a SELECT Statement return the CREATE TABLE noder.
+	// if the table spec != nil it will get parsed below.
+	if c.TableSpec == nil && c.OptSelect != nil {
+		tableSpec := &plan.TableSpec{}
+
+		selectNode, err := convertSelectStatement(ctx, c.OptSelect.Select)
+		if err != nil {
+			return nil, err
+		}
+
+		return plan.NewCreateTableSelect(sql.UnresolvedDatabase(c.Table.Qualifier.String()), c.Table.Name.String(), selectNode, tableSpec, plan.IfNotExistsOption(c.IfNotExists), plan.TempTableOption(c.Temporary)), nil
+	}
+
 	schema, err := TableSpecToSchema(nil, c.TableSpec)
 	if err != nil {
 		return nil, err
@@ -1321,6 +1334,15 @@ func convertCreateTable(ctx *sql.Context, c *sqlparser.DDL) (sql.Node, error) {
 		ChDefs:  chDefs,
 	}
 
+	if c.OptSelect != nil {
+		selectNode, err := convertSelectStatement(ctx, c.OptSelect.Select)
+		if err != nil {
+			return nil, err
+		}
+
+		return plan.NewCreateTableSelect(sql.UnresolvedDatabase(qualifier), c.Table.Name.String(), selectNode, tableSpec, plan.IfNotExistsOption(c.IfNotExists), plan.TempTableOption(c.Temporary)), nil
+	}
+
 	return plan.NewCreateTable(
 		sql.UnresolvedDatabase(qualifier), c.Table.Name.String(), plan.IfNotExistsOption(c.IfNotExists), plan.TempTableOption(c.Temporary), tableSpec), nil
 }

sql/parse/parse_test.go

@@ -842,6 +842,13 @@ CREATE TABLE t2
 			}},
 		},
 	),
+	`CREATE TEMPORARY TABLE mytable AS SELECT * from othertable`: plan.NewCreateTableSelect(
+		sql.UnresolvedDatabase(""),
+		"mytable",
+		plan.NewProject([]sql.Expression{expression.NewStar()}, plan.NewUnresolvedTable("othertable", "")),
+		&plan.TableSpec{},
+		plan.IfNotExistsAbsent,
+		plan.IsTempTable),
 	`DROP TABLE foo;`: plan.NewDropTable(
 		sql.UnresolvedDatabase(""), false, "foo",
 	),

sql/plan/ddl.go

@@ -98,28 +98,28 @@ type TableSpec struct {
 	IdxDefs []*IndexDefinition
 }
 
-func (c *TableSpec) WithSchema(schema sql.Schema) (*TableSpec, error) {
+func (c *TableSpec) WithSchema(schema sql.Schema) *TableSpec {
 	nc := *c
 	nc.Schema = schema
-	return &nc, nil
+	return &nc
 }
 
-func (c *TableSpec) WithForeignKeys(fkDefs []*sql.ForeignKeyConstraint) (*TableSpec, error) {
+func (c *TableSpec) WithForeignKeys(fkDefs []*sql.ForeignKeyConstraint) *TableSpec {
 	nc := *c
 	nc.FkDefs = fkDefs
-	return &nc, nil
+	return &nc
 }
 
-func (c *TableSpec) WithCheckConstraints(chDefs []*sql.CheckConstraint) (*TableSpec, error) {
+func (c *TableSpec) WithCheckConstraints(chDefs []*sql.CheckConstraint) *TableSpec {
 	nc := *c
 	nc.ChDefs = chDefs
-	return &nc, nil
+	return &nc
 }
 
-func (c *TableSpec) WithIndices(idxDefs []*IndexDefinition) (*TableSpec, error) {
+func (c *TableSpec) WithIndices(idxDefs []*IndexDefinition) *TableSpec {
 	nc := *c
 	nc.IdxDefs = idxDefs
-	return &nc, nil
+	return &nc
 }
 
 // CreateTable is a node describing the creation of some table.
@@ -133,6 +133,7 @@ type CreateTable struct {
 	idxDefs     []*IndexDefinition
 	like        sql.Node
 	temporary   TempTableOption
+	selectNode  sql.Node
 }
 
 var _ sql.Databaser = (*CreateTable)(nil)
@@ -168,6 +169,25 @@ func NewCreateTableLike(db sql.Database, name string, likeTable sql.Node, ifn If
 	}
 }
 
+// NewCreateTableSelect create a new CreateTable node for CREATE TABLE [AS] SELECT
+func NewCreateTableSelect(db sql.Database, name string, selectNode sql.Node, tableSpec *TableSpec, ifn IfNotExistsOption, temp TempTableOption) *CreateTable {
+	for _, s := range tableSpec.Schema {
+		s.Source = name
+	}
+
+	return &CreateTable{
+		ddlNode:     ddlNode{db: db},
+		schema:      tableSpec.Schema,
+		fkDefs:      tableSpec.FkDefs,
+		chDefs:      tableSpec.ChDefs,
+		idxDefs:     tableSpec.IdxDefs,
+		name:        name,
+		selectNode:  selectNode,
+		ifNotExists: ifn,
+		temporary:   temp,
+	}
+}
+
 // WithDatabase implements the sql.Databaser interface.
 func (c *CreateTable) WithDatabase(db sql.Database) (sql.Node, error) {
 	nc := *c
@@ -317,6 +337,8 @@ func (c *CreateTable) createChecks(ctx *sql.Context, tableNode sql.Table) error
 func (c *CreateTable) Children() []sql.Node {
 	if c.like != nil {
 		return []sql.Node{c.like}
+	} else if c.selectNode != nil {
+		return []sql.Node{c.selectNode}
 	}
 	return nil
 }
@@ -326,8 +348,16 @@ func (c *CreateTable) WithChildren(children ...sql.Node) (sql.Node, error) {
 	if len(children) == 0 {
 		return c, nil
 	} else if len(children) == 1 {
+		child := children[0]
 		nc := *c
-		nc.like = children[0]
+
+		switch child.(type) {
+		case *Project, *Limit:
+			nc.selectNode = child
+		default:
+			nc.like = child
+		}
+
 		return &nc, nil
 	} else {
 		return nil, sql.ErrInvalidChildrenNumber.New(c, len(children), 1)
@@ -429,6 +459,21 @@ func (c *CreateTable) Like() sql.Node {
 	return c.like
 }
 
+func (c *CreateTable) Select() sql.Node {
+	return c.selectNode
+}
+
+func (c *CreateTable) TableSpec() *TableSpec {
+	tableSpec := TableSpec{}
+
+	ret := tableSpec.WithSchema(c.schema)
+	ret = tableSpec.WithForeignKeys(c.fkDefs)
+	ret = tableSpec.WithIndices(c.idxDefs)
+	ret = tableSpec.WithCheckConstraints(c.chDefs)
+
+	return ret
+}
+
 func (c *CreateTable) Name() string {
 	return c.name
 }