Merge pull request #1716 from dolthub/fulghum/revert-dac7262c9b7904b931f9792ac8ce3eb52f6f9ccf

Revert "allow renaming views with `RENAME TABLE` statement (#1712)"

Author: fulghum

enginetest/enginetests.go

@@ -1897,6 +1897,10 @@ func TestRecursiveViewDefinition(t *testing.T, harness Harness) {
 	db, err := e.Analyzer.Catalog.Database(ctx, "mydb")
 	require.NoError(t, err)
 
+	if pdb, ok := db.(mysql_db.PrivilegedDatabase); ok {
+		db = pdb.Unwrap()
+	}
+
 	vdb, ok := db.(sql.ViewDatabase)
 	require.True(t, ok, "expected sql.ViewDatabase")
 

enginetest/queries/script_queries.go

@@ -2902,35 +2902,6 @@ var ScriptTests = []ScriptTest{
 			},
 		},
 	},
-	{
-		Name: "rename views with RENAME TABLE ... TO .. statement",
-		SetUpScript: []string{
-			"create table t1 (id int primary key, v1 int);",
-			"create view v1 as select * from t1;",
-		},
-		Assertions: []ScriptTestAssertion{
-			{
-				Query:    "show tables;",
-				Expected: []sql.Row{{"myview"}, {"t1"}, {"v1"}},
-			},
-			{
-				Query:    "rename table v1 to view1",
-				Expected: []sql.Row{{types.OkResult{RowsAffected: 0}}},
-			},
-			{
-				Query:    "show tables;",
-				Expected: []sql.Row{{"myview"}, {"t1"}, {"view1"}},
-			},
-			{
-				Query:    "rename table view1 to newViewName, t1 to newTableName",
-				Expected: []sql.Row{{types.OkResult{RowsAffected: 0}}},
-			},
-			{
-				Query:    "show tables;",
-				Expected: []sql.Row{{"myview"}, {"newTableName"}, {"newViewName"}},
-			},
-		},
-	},
 }
 
 var SpatialScriptTests = []ScriptTest{

sql/analyzer/resolve_views.go

@@ -18,6 +18,7 @@ import (
 	"fmt"
 
 	"github.com/dolthub/go-mysql-server/sql"
+	"github.com/dolthub/go-mysql-server/sql/mysql_db"
 	"github.com/dolthub/go-mysql-server/sql/parse"
 	"github.com/dolthub/go-mysql-server/sql/plan"
 	"github.com/dolthub/go-mysql-server/sql/transform"
@@ -50,7 +51,11 @@ func resolveViews(ctx *sql.Context, a *Analyzer, n sql.Node, scope *Scope, sel R
 				return nil, transform.SameTree, err
 			}
 
-			if vdb, vok := db.(sql.ViewDatabase); vok {
+			maybeVdb := db
+			if privilegedDatabase, ok := maybeVdb.(mysql_db.PrivilegedDatabase); ok {
+				maybeVdb = privilegedDatabase.Unwrap()
+			}
+			if vdb, vok := maybeVdb.(sql.ViewDatabase); vok {
 				viewDef, vdok, verr := vdb.GetViewDefinition(ctx, viewName)
 				if verr != nil {
 					return nil, transform.SameTree, verr

sql/errors.go

@@ -555,9 +555,6 @@ var (
 	// are automatically rolled back. Clients receiving this error must retry the transaction.
 	ErrLockDeadlock = errors.NewKind("serialization failure: %s, try restarting transaction.")
 
-	// ErrViewsNotSupported is returned when attempting to access a view on a database that doesn't support them.
-	ErrViewsNotSupported = errors.NewKind("database '%s' doesn't support views")
-
 	// ErrExistingView is returned when a CREATE VIEW statement uses a name that already exists
 	ErrExistingView = errors.NewKind("the view %s.%s already exists")
 

sql/information_schema/information_schema.go

@@ -2793,6 +2793,9 @@ func viewsInDatabase(ctx *Context, db Database) ([]ViewDefinition, error) {
 	var views []ViewDefinition
 	dbName := db.Name()
 
+	if privilegedDatabase, ok := db.(mysql_db.PrivilegedDatabase); ok {
+		db = privilegedDatabase.Unwrap()
+	}
 	if vdb, ok := db.(ViewDatabase); ok {
 		dbViews, err := vdb.AllViews(ctx)
 		if err != nil {

sql/mysql_db/privileged_database_provider.go

@@ -116,7 +116,6 @@ var _ sql.TableCopierDatabase = PrivilegedDatabase{}
 var _ sql.ReadOnlyDatabase = PrivilegedDatabase{}
 var _ sql.TemporaryTableDatabase = PrivilegedDatabase{}
 var _ sql.CollatedDatabase = PrivilegedDatabase{}
-var _ sql.ViewDatabase = PrivilegedDatabase{}
 
 // NewPrivilegedDatabase returns a new PrivilegedDatabase.
 func NewPrivilegedDatabase(grantTables *MySQLDb, db sql.Database) sql.Database {
@@ -385,38 +384,6 @@ func (pdb PrivilegedDatabase) UpdateEvent(ctx *sql.Context, ed sql.EventDefiniti
 	return sql.ErrEventsNotSupported.New(pdb.db.Name())
 }
 
-// CreateView implements sql.ViewDatabase
-func (pdb PrivilegedDatabase) CreateView(ctx *sql.Context, name string, selectStatement, createViewStmt string) error {
-	if db, ok := pdb.db.(sql.ViewDatabase); ok {
-		return db.CreateView(ctx, name, selectStatement, createViewStmt)
-	}
-	return sql.ErrViewsNotSupported.New(pdb.db.Name())
-}
-
-// DropView implements sql.ViewDatabase
-func (pdb PrivilegedDatabase) DropView(ctx *sql.Context, name string) error {
-	if db, ok := pdb.db.(sql.ViewDatabase); ok {
-		return db.DropView(ctx, name)
-	}
-	return sql.ErrViewsNotSupported.New(pdb.db.Name())
-}
-
-// GetViewDefinition implements sql.ViewDatabase
-func (pdb PrivilegedDatabase) GetViewDefinition(ctx *sql.Context, viewName string) (sql.ViewDefinition, bool, error) {
-	if db, ok := pdb.db.(sql.ViewDatabase); ok {
-		return db.GetViewDefinition(ctx, viewName)
-	}
-	return sql.ViewDefinition{}, false, sql.ErrViewsNotSupported.New(pdb.db.Name())
-}
-
-// AllViews implements sql.ViewDatabase
-func (pdb PrivilegedDatabase) AllViews(ctx *sql.Context) ([]sql.ViewDefinition, error) {
-	if db, ok := pdb.db.(sql.ViewDatabase); ok {
-		return db.AllViews(ctx)
-	}
-	return nil, sql.ErrViewsNotSupported.New(pdb.db.Name())
-}
-
 // CopyTableData implements the interface sql.TableCopierDatabase.
 func (pdb PrivilegedDatabase) CopyTableData(ctx *sql.Context, sourceTable string, destinationTable string) (uint64, error) {
 	if db, ok := pdb.db.(sql.TableCopierDatabase); ok {

sql/plan/alter_table.go

@@ -53,31 +53,6 @@ func (r *RenameTable) String() string {
 	return fmt.Sprintf("Rename table %s to %s", r.OldNames, r.NewNames)
 }
 
-func (r *RenameTable) RowIter(ctx *sql.Context, row sql.Row) (sql.RowIter, error) {
-	// TODO: 'ALTER TABLE newViewName RENAME view1' should fail on renaming views - should be fixed in vitess
-	renamer, _ := r.Db.(sql.TableRenamer)
-	viewDb, _ := r.Db.(sql.ViewDatabase)
-	viewRegistry := ctx.GetViewRegistry()
-
-	for i, oldName := range r.OldNames {
-		if tbl, exists := r.tableExists(ctx, oldName); exists {
-			err := r.renameTable(ctx, renamer, tbl, oldName, r.NewNames[i])
-			if err != nil {
-				return nil, err
-			}
-		} else {
-			success, err := r.renameView(ctx, viewDb, viewRegistry, oldName, r.NewNames[i])
-			if err != nil {
-				return nil, err
-			} else if !success {
-				return nil, sql.ErrTableNotFound.New(oldName)
-			}
-		}
-	}
-
-	return sql.RowsToRowIter(sql.NewRow(types.NewOkResult(0))), nil
-}
-
 func (r *RenameTable) WithChildren(children ...sql.Node) (sql.Node, error) {
 	return NillaryWithChildren(r, children...)
 }
@@ -99,106 +74,6 @@ func (*RenameTable) CollationCoercibility(ctx *sql.Context) (collation sql.Colla
 	return sql.Collation_binary, 7
 }
 
-func (r *RenameTable) tableExists(ctx *sql.Context, name string) (sql.Table, bool) {
-	tbl, ok, err := r.Db.GetTableInsensitive(ctx, name)
-	if err != nil || !ok {
-		return nil, false
-	}
-	return tbl, true
-}
-
-func (r *RenameTable) renameTable(ctx *sql.Context, renamer sql.TableRenamer, tbl sql.Table, oldName, newName string) error {
-	if renamer == nil {
-		return sql.ErrRenameTableNotSupported.New(r.Db.Name())
-	}
-
-	if fkTable, ok := tbl.(sql.ForeignKeyTable); ok {
-		parentFks, err := fkTable.GetReferencedForeignKeys(ctx)
-		if err != nil {
-			return err
-		}
-		for _, parentFk := range parentFks {
-			//TODO: support renaming tables across databases for foreign keys
-			if strings.ToLower(parentFk.Database) != strings.ToLower(parentFk.ParentDatabase) {
-				return fmt.Errorf("updating foreign key table names across databases is not yet supported")
-			}
-			parentFk.ParentTable = newName
-			childTbl, ok, err := r.Db.GetTableInsensitive(ctx, parentFk.Table)
-			if err != nil {
-				return err
-			}
-			if !ok {
-				return sql.ErrTableNotFound.New(parentFk.Table)
-			}
-			childFkTbl, ok := childTbl.(sql.ForeignKeyTable)
-			if !ok {
-				return fmt.Errorf("referenced table `%s` supports foreign keys but declaring table `%s` does not", parentFk.ParentTable, parentFk.Table)
-			}
-			err = childFkTbl.UpdateForeignKey(ctx, parentFk.Name, parentFk)
-			if err != nil {
-				return err
-			}
-		}
-
-		fks, err := fkTable.GetDeclaredForeignKeys(ctx)
-		if err != nil {
-			return err
-		}
-		for _, fk := range fks {
-			fk.Table = newName
-			err = fkTable.UpdateForeignKey(ctx, fk.Name, fk)
-			if err != nil {
-				return err
-			}
-		}
-	}
-
-	err := renamer.RenameTable(ctx, oldName, newName)
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (r *RenameTable) renameView(ctx *sql.Context, viewDb sql.ViewDatabase, vr *sql.ViewRegistry, oldName, newName string) (bool, error) {
-	if viewDb != nil {
-		oldView, exists, err := viewDb.GetViewDefinition(ctx, oldName)
-		if err != nil {
-			return false, err
-		} else if !exists {
-			return false, nil
-		}
-
-		err = viewDb.DropView(ctx, oldName)
-		if err != nil {
-			return false, err
-		}
-
-		err = viewDb.CreateView(ctx, newName, oldView.TextDefinition, oldView.CreateViewStatement)
-		if err != nil {
-			return false, err
-		}
-
-		return true, nil
-	} else {
-		view, exists := vr.View(r.Db.Name(), oldName)
-		if !exists {
-			return false, nil
-		}
-
-		err := vr.Delete(r.Db.Name(), oldName)
-		if err != nil {
-			return false, nil
-		}
-		err = vr.Register(r.Db.Name(), sql.NewView(newName, view.Definition(), view.TextDefinition(), view.CreateStatement()))
-		if err != nil {
-			return false, nil
-		}
-		return true, nil
-	}
-}
-
 type AddColumn struct {
 	ddlNode
 	Table     sql.Node

sql/rowexec/ddl.go

@@ -233,7 +233,72 @@ func (b *BaseBuilder) buildDropCheck(ctx *sql.Context, n *plan.DropCheck, row sq
 }
 
 func (b *BaseBuilder) buildRenameTable(ctx *sql.Context, n *plan.RenameTable, row sql.Row) (sql.RowIter, error) {
-	return n.RowIter(ctx, row)
+	renamer, ok := n.Db.(sql.TableRenamer)
+	if !ok {
+		return nil, sql.ErrRenameTableNotSupported.New(n.Db.Name())
+	}
+
+	var err error
+	for i, oldName := range n.OldNames {
+		var tbl sql.Table
+		var ok bool
+		tbl, ok, err = n.Db.GetTableInsensitive(ctx, oldName)
+		if err != nil {
+			return nil, err
+		}
+
+		if !ok {
+			return nil, sql.ErrTableNotFound.New(oldName)
+		}
+
+		if fkTable, ok := tbl.(sql.ForeignKeyTable); ok {
+			parentFks, err := fkTable.GetReferencedForeignKeys(ctx)
+			if err != nil {
+				return nil, err
+			}
+			for _, parentFk := range parentFks {
+				//TODO: support renaming tables across databases for foreign keys
+				if strings.ToLower(parentFk.Database) != strings.ToLower(parentFk.ParentDatabase) {
+					return nil, fmt.Errorf("updating foreign key table names across databases is not yet supported")
+				}
+				parentFk.ParentTable = n.NewNames[i]
+				childTbl, ok, err := n.Db.GetTableInsensitive(ctx, parentFk.Table)
+				if err != nil {
+					return nil, err
+				}
+				if !ok {
+					return nil, sql.ErrTableNotFound.New(parentFk.Table)
+				}
+				childFkTbl, ok := childTbl.(sql.ForeignKeyTable)
+				if !ok {
+					return nil, fmt.Errorf("referenced table `%s` supports foreign keys but declaring table `%s` does not", parentFk.ParentTable, parentFk.Table)
+				}
+				err = childFkTbl.UpdateForeignKey(ctx, parentFk.Name, parentFk)
+				if err != nil {
+					return nil, err
+				}
+			}
+
+			fks, err := fkTable.GetDeclaredForeignKeys(ctx)
+			if err != nil {
+				return nil, err
+			}
+			for _, fk := range fks {
+				fk.Table = n.NewNames[i]
+				err = fkTable.UpdateForeignKey(ctx, fk.Name, fk)
+				if err != nil {
+					return nil, err
+				}
+			}
+		}
+
+		err = renamer.RenameTable(ctx, tbl.Name(), n.NewNames[i])
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return sql.RowsToRowIter(sql.NewRow(types.NewOkResult(0))), nil
 }
 
 func (b *BaseBuilder) buildModifyColumn(ctx *sql.Context, n *plan.ModifyColumn, row sql.Row) (sql.RowIter, error) {
@@ -791,6 +856,7 @@ func (b *BaseBuilder) buildAlterDB(ctx *sql.Context, n *plan.AlterDB, row sql.Ro
 
 func (b *BaseBuilder) buildCreateTable(ctx *sql.Context, n *plan.CreateTable, row sql.Row) (sql.RowIter, error) {
 	var err error
+	var vd sql.ViewDatabase
 
 	// If it's set to Invalid, then no collation has been explicitly defined
 	if n.Collation == sql.Collation_Unspecified {
@@ -862,11 +928,13 @@ func (b *BaseBuilder) buildCreateTable(ctx *sql.Context, n *plan.CreateTable, ro
 		return sql.RowsToRowIter(), err
 	}
 
-	if vdb, vok := n.Db.(sql.ViewDatabase); vok {
-		_, ok, err := vdb.GetViewDefinition(ctx, n.Name())
+	vd, _ = maybePrivDb.(sql.ViewDatabase)
+	if vd != nil {
+		_, ok, err := vd.GetViewDefinition(ctx, n.Name())
 		if err != nil {
 			return nil, err
 		}
+
 		if ok {
 			return nil, sql.ErrTableAlreadyExists.New(n.Name())
 		}

sql/rowexec/show.go

@@ -193,8 +193,11 @@ func (b *BaseBuilder) buildShowTables(ctx *sql.Context, n *plan.ShowTables, row
 	}
 
 	// TODO: currently there is no way to see views AS OF a particular time
-	db := n.Database()
-	if vdb, ok := db.(sql.ViewDatabase); ok {
+	maybeVdb := n.Database()
+	if privilegedDatabase, ok := maybeVdb.(mysql_db.PrivilegedDatabase); ok {
+		maybeVdb = privilegedDatabase.Unwrap()
+	}
+	if vdb, ok := maybeVdb.(sql.ViewDatabase); ok {
 		views, err := vdb.AllViews(ctx)
 		if err != nil {
 			return nil, err
@@ -208,7 +211,7 @@ func (b *BaseBuilder) buildShowTables(ctx *sql.Context, n *plan.ShowTables, row
 		}
 	}
 
-	for _, view := range ctx.GetViewRegistry().ViewsInDatabase(db.Name()) {
+	for _, view := range ctx.GetViewRegistry().ViewsInDatabase(maybeVdb.Name()) {
 		row := sql.Row{view.Name()}
 		if n.Full {
 			row = append(row, "VIEW")