Skip to content

Commit

Permalink
sql: allow rename database for sequences without a db name reference
Browse files Browse the repository at this point in the history 
See release note for description. This PR should be included ahead of
the more "general" fix of changing the DefaultExpr with the new database
as it unblocks people using
`experimental_serial_normalization=sql_sequence` from using the database
rename feature.

Release note (sql change): Previously, when we renamed a database, any
table referencing a sequence would be blocked from being able to rename
the table. This is to block cases where if the table's reference to the
sequence contains the database name, and the database name changes, we
have no way of overwriting the table's reference to the sequence in the
new database. However, if no database name is included in the sequence
reference, we should continue to allow the database to rename, as is
implemented with this change.
  • Loading branch information
@otan
otan committed Mar 3, 2020
1 parent 8d2e9d5 commit cee9355
Show file tree
Hide file tree
Showing 2 changed files with 133 additions and 4 deletions.
25 changes: 22 additions & 3 deletions pkg/sql/logictest/testdata/logic_test/rename_database
View file
Original file line number Diff line number Diff line change
Expand Up @@ -136,7 +136,7 @@ ALTER DATABASE u RENAME TO v
statement ok
CREATE VIEW v.v AS SELECT k,v FROM v.kv

statement error cannot rename database because relation "v.public.v" depends on relation "v.public.kv"
statement error cannot rename database because relation "v.public.v" depends on relation "v.public.kv"\s.*you can drop "v.public.v" instead
ALTER DATABASE v RENAME TO u

# Check that the default databases can be renamed like any other.
Expand Down Expand Up @@ -173,7 +173,8 @@ t
v

# Test dependent sequences on different databases upon renames
# return the appropriate error message.
# return the appropriate error message, as well as testing
# renaming databases with sequences in the same DB is successful.
subtest regression_45411

statement ok
Expand All @@ -182,8 +183,26 @@ CREATE DATABASE db1; CREATE SEQUENCE db1.seq
statement ok
CREATE DATABASE db2; CREATE TABLE db2.tbl (a int DEFAULT nextval('db1.seq'))

statement error cannot rename database because relation "db2.public.tbl" depends on relation "db1.public.seq"
statement error cannot rename database because relation "db2.public.tbl" depends on relation "db1.public.seq"\s.*you can drop the column default "a" of "db1.public.seq" referencing "db2.public.tbl"
ALTER DATABASE db1 RENAME TO db3

statement ok
DROP DATABASE db2 CASCADE; DROP DATABASE db1 CASCADE

statement ok
CREATE DATABASE db1; CREATE SEQUENCE db1.a_seq; CREATE SEQUENCE db1.b_seq; USE db1;

statement ok
CREATE TABLE db1.a (a int default nextval('a_seq') + nextval('b_seq') + 1); ALTER DATABASE db1 RENAME TO db2; USE db2;

statement error cannot rename database because relation "db2.public.a" depends on relation "db2.public.a_seq"\s.*you can drop the column default "a" of "db2.public.a_seq" referencing "db2.public.a" or modify the default to not reference the database name "db2"
DROP TABLE db2.a; CREATE TABLE db2.a (a int default nextval('a_seq') + nextval('db2.b_seq') + 1); ALTER DATABASE db2 RENAME TO db1

statement error cannot rename database because relation "db2.public.a" depends on relation "db2.public.a_seq"\s.*you can drop the column default "a" of "db2.public.a_seq" referencing "db2.public.a" or modify the default to not reference the database name "db2"
DROP TABLE db2.a; CREATE TABLE db2.a (a int default nextval('a_seq') + nextval('db2.public.b_seq') + 1); ALTER DATABASE db2 RENAME TO db1

statement ok
DROP TABLE db2.a; CREATE TABLE db2.a (a int default nextval('a_seq') + nextval('public.b_seq') + 1); ALTER DATABASE db2 RENAME TO db1

statement ok
USE defaultdb; DROP DATABASE db1 CASCADE
112 changes: 111 additions & 1 deletion pkg/sql/rename_database.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,14 @@ import (
"context"
"fmt"

"github.com/cockroachdb/cockroach/pkg/sql/parser"
"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
"github.com/cockroachdb/cockroach/pkg/sql/privilege"
"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
"github.com/cockroachdb/cockroach/pkg/util"
"github.com/cockroachdb/cockroach/pkg/util/log"
"github.com/cockroachdb/errors"
)

type renameDatabaseNode struct {
Expand Down Expand Up @@ -78,6 +81,7 @@ func (n *renameDatabaseNode) startExec(params runParams) error {
// strings, they (may) explicitly specify the database name.
// Rather than trying to rewrite them with the changed DB name, we
// simply disallow such renames for now.
// See #34416.
phyAccessor := p.PhysicalSchemaAccessor()
lookupFlags := p.CommonLookupFlags(true /*required*/)
// DDL statements bypass the cache.
Expand Down Expand Up @@ -116,6 +120,20 @@ func (n *renameDatabaseNode) startExec(params runParams) error {
if err != nil {
return err
}

isAllowed, referencedCol, err := isAllowedDependentDescInRenameDatabase(
dependedOn,
tbDesc,
dependentDesc,
dbDesc.Name,
)
if err != nil {
return err
}
if isAllowed {
continue
}

tbTableName := tree.MakeTableNameWithSchema(
tree.Name(dbDesc.Name),
tree.Name(schema),
Expand Down Expand Up @@ -152,7 +170,26 @@ func (n *renameDatabaseNode) startExec(params runParams) error {
dependentDescQualifiedString,
tbTableName.String(),
)
hint := fmt.Sprintf("you can drop %s instead", dependentDescQualifiedString)

// We can have a more specific error message for sequences.
if tbDesc.IsSequence() {
hint := fmt.Sprintf(
"you can drop the column default %q of %q referencing %q",
referencedCol,
tbTableName.String(),
dependentDescQualifiedString,
)
if dependentDesc.GetParentID() == dbDesc.ID {
hint += fmt.Sprintf(
" or modify the default to not reference the database name %q",
dbDesc.Name,
)
}
return sqlbase.NewDependentObjectErrorWithHint(msg, hint)
}

// Otherwise, we default to the view error message.
hint := fmt.Sprintf("you can drop %q instead", dependentDescQualifiedString)
return sqlbase.NewDependentObjectErrorWithHint(msg, hint)
}
}
Expand All @@ -161,6 +198,79 @@ func (n *renameDatabaseNode) startExec(params runParams) error {
return p.renameDatabase(ctx, dbDesc, n.newName)
}

// isAllowedDependentDescInRename determines when rename database is allowed with
// a given {tbDesc, dependentDesc} with the relationship dependedOn on a db named dbName.
// Returns a bool representing whether it's allowed, a string indicating the column name
// found to contain the database (if it exists), and an error if any.
// This is a workaround for #45411 until #34416 is resolved.
func isAllowedDependentDescInRenameDatabase(
dependedOn sqlbase.TableDescriptor_Reference,
tbDesc *sqlbase.TableDescriptor,
dependentDesc *sqlbase.TableDescriptor,
dbName string,
) (bool, string, error) {
// If it is a sequence, and it does not contain the database name, then we have
// no reason to block it's deletion.
if !tbDesc.IsSequence() {
return false, "", nil
}

colIDs := util.MakeFastIntSet()
for _, colID := range dependedOn.ColumnIDs {
colIDs.Add(int(colID))
}

for _, column := range dependentDesc.Columns {
if !colIDs.Contains(int(column.ID)) {
continue
}
colIDs.Remove(int(column.ID))

if column.DefaultExpr == nil {
return false, "", errors.AssertionFailedf(
"rename_database: expected column id %d in table id %d to have a default expr",
dependedOn.ID,
dependentDesc.ID,
)
}
// Try parse the default expression and find the table name direct reference.
parsedExpr, err := parser.ParseExpr(*column.DefaultExpr)
if err != nil {
return false, "", err
}
typedExpr, err := tree.TypeCheck(parsedExpr, nil, &column.Type)
if err != nil {
return false, "", err
}
seqNames, err := getUsedSequenceNames(typedExpr)
if err != nil {
return false, "", err
}
for _, seqName := range seqNames {
parsedSeqName, err := parser.ParseTableName(seqName)
if err != nil {
return false, "", err
}
// There must be at least two parts for this to work.
if parsedSeqName.NumParts >= 2 {
// We only don't allow this if the database name is in there.
// This is always the last argument.
if tree.Name(parsedSeqName.Parts[parsedSeqName.NumParts-1]).Normalize() == tree.Name(dbName).Normalize() {
return false, column.Name, nil
}
}
}
}
if colIDs.Len() > 0 {
return false, "", errors.AssertionFailedf(
"expected to find column ids %s in table id %d",
colIDs.String(),
dependentDesc.ID,
)
}
return true, "", nil
}

func (n *renameDatabaseNode) Next(runParams) (bool, error) { return false, nil }
func (n *renameDatabaseNode) Values() tree.Datums { return tree.Datums{} }
func (n *renameDatabaseNode) Close(context.Context) {}

0 comments on commit cee9355

Please sign in to comment.