Merge #74636

74636: schemachanger: address error message inconsistencies with legacy schema changer r=fqazi a=fqazi These changes will address the following issues, which will also allow us to enable logictest support in a separate PR: 1. Cases where the declarative schema changer currently either does not generate an error message or generates a message that is inconsistent with the legacy schema changer. 2. A case inside DROP SCHEMA inside the legacy schema changer, where the message generated for privilege errors is not consistent with other DROP operations. 3. Fixing an issue inside the declarative schema changer where we should have been avoiding leased descriptors for event logging. Co-authored-by: Faizan Qazi <[email protected]>
cockroachdb · Jan 13, 2022 · 1db01e8 · 1db01e8
2 parents a763b09 + 3a38ef9
commit 1db01e8
Show file tree

Hide file tree

Showing 9 changed files with 86 additions and 26 deletions.
diff --git a/pkg/sql/drop_schema.go b/pkg/sql/drop_schema.go
@@ -102,7 +102,7 @@ func (p *planner) DropSchema(ctx context.Context, n *tree.DropSchema) (planNode,
 			}
 			if !(isAdmin || hasOwnership) {
 				return nil, pgerror.Newf(pgcode.InsufficientPrivilege,
-					"permission denied to drop schema %q", sc.GetName())
+					"must be owner of schema %q", sc.GetName())
 			}
 			namesBefore := len(d.objectNamesToDelete)
 			if err := d.collectObjectsInSchema(ctx, p, db, sc); err != nil {

diff --git a/pkg/sql/logictest/testdata/logic_test/new_schema_changer b/pkg/sql/logictest/testdata/logic_test/new_schema_changer
@@ -391,7 +391,7 @@ CREATE VIEW v4Dep AS (SELECT n2, n1 FROM v2Dep);
 statement ok
 explain (DDL, DEPS) DROP VIEW v1Dep CASCADE;
 
-statement error pq: cannot drop view "v1dep" because view "v2dep" depends on it
+statement error pq: cannot drop relation "v1dep" because view "v2dep" depends on it
 DROP VIEW v1Dep RESTRICT;
 
 statement error pq: "v1dep" is not a materialized view
@@ -453,6 +453,15 @@ statement ok
 CREATE SEQUENCE defaultdb.sq1 OWNED BY defaultdb.shipments.carrier;
 
 statement ok
+CREATE TABLE defaultdb.sq1dep (
+	rand_col INT8 DEFAULT nextval('defaultdb.sq1')
+);
+
+statement error cannot drop table a because other objects depend on it
+DROP TABLE defaultdb.shipments;
+
+statement ok
+DROP TABLE defaultdb.sq1dep;
 DROP TABLE defaultdb.shipments;
 
 statement ok
@@ -466,11 +475,10 @@ CREATE TABLE defaultdb.shipments (
     CONSTRAINT fk_orders FOREIGN KEY (customer_id) REFERENCES defaultdb.orders(customer)
  );
 
-
 statement ok
 CREATE VIEW defaultdb.v1 as (select customer_id, carrier from defaultdb.shipments);
 
-statement error pq: cannot drop table "defaultdb.public.shipments" because view "defaultdb.public.v1" depends on it
+statement error pq: cannot drop relation "shipments" because view "v1" depends on it
 DROP TABLE defaultdb.shipments;
 
 statement ok
@@ -664,7 +672,7 @@ BLAH5
 statement error schema "sc1" is not empty and CASCADE was not specified
 DROP SCHEMA db1.sc1
 
-statement error database "db1" has a non-empty schema "public" and CASCADE was not specified
+statement error database "db1" is not empty and RESTRICT was specified
 DROP DATABASE db1 RESTRICT
 
 statement ok
@@ -1268,3 +1276,31 @@ DROP TABLE ttc3;
 
 statement ok
 DROP TYPE d_tc;
+
+subtest empty-database
+
+statement error empty database name
+DROP DATABASE ""
+
+subtest schema-permission-error
+
+user root
+
+statement ok
+CREATE SCHEMA sc1;
+
+statement ok
+CREATE DATABASE db1;
+
+user testuser
+
+statement ok
+SET experimental_use_new_schema_changer = 'on'
+
+statement error must be owner of schema \"sc1\"
+DROP SCHEMA sc1;
+
+statement error user testuser does not have DROP privilege on database db1
+DROP DATABASE db1;
+
+user root
diff --git a/pkg/sql/logictest/testdata/logic_test/schema b/pkg/sql/logictest/testdata/logic_test/schema
@@ -423,7 +423,7 @@ CREATE TYPE privs.denied AS ENUM ('denied')
 statement error pq: must be owner of schema "privs"
 ALTER SCHEMA privs RENAME TO denied
 
-statement error pq: permission denied to drop schema "privs"
+statement error must be owner of schema \"privs\"
 DROP SCHEMA privs
 
 # Test the usage privilege.

diff --git a/pkg/sql/schemachanger/scbuild/internal/scbuildstmt/drop_database.go b/pkg/sql/schemachanger/scbuild/internal/scbuildstmt/drop_database.go
@@ -48,7 +48,7 @@ func DropDatabase(b BuildCtx, n *tree.DropDatabase) {
 			// Block drops if cascade is not set.
 			if n.DropBehavior == tree.DropRestrict && (nodeAdded || !schemaDroppedIDs.Empty()) {
 				panic(pgerror.Newf(pgcode.DependentObjectsStillExist,
-					"database %q has a non-empty schema %q and CASCADE was not specified", db.GetName(), schema.GetName()))
+					"database %q is not empty and RESTRICT was specified", db.GetName()))
 			}
 			// If no schema exists to depend on, then depend on dropped IDs
 			if !nodeAdded {

diff --git a/pkg/sql/schemachanger/scbuild/internal/scbuildstmt/drop_schema.go b/pkg/sql/schemachanger/scbuild/internal/scbuildstmt/drop_schema.go
@@ -42,11 +42,6 @@ func dropSchema(
 	behavior tree.DropBehavior,
 ) (nodeAdded bool, dropIDs catalog.DescriptorIDSet) {
 	descsThatNeedElements := catalog.DescriptorIDSet{}
-	// For non-user defined schemas, another check will be
-	// done each object as we go to drop them.
-	if sc.SchemaKind() == catalog.SchemaUserDefined {
-		b.MustOwn(sc)
-	}
 	_, objectIDs := b.CatalogReader().ReadObjectNamesAndIDs(b, db, sc)
 	for _, id := range objectIDs {
 		// For dependency tracking we will still track that these elements were

diff --git a/pkg/sql/schemachanger/scbuild/internal/scbuildstmt/drop_table.go b/pkg/sql/schemachanger/scbuild/internal/scbuildstmt/drop_table.go
@@ -86,8 +86,8 @@ func dropTableDependents(b BuildCtx, tbl catalog.TableDescriptor, behavior tree.
 				onErrPanic(err)
 
 				return pgerror.Newf(
-					pgcode.DependentObjectsStillExist, "cannot drop table %q because view %q depends on it",
-					name, depViewName)
+					pgcode.DependentObjectsStillExist, "cannot drop relation %q because view %q depends on it",
+					name.Object(), depViewName.Object())
 			}
 			dropView(c, dependentDesc, behavior)
 			return nil
@@ -124,14 +124,36 @@ func dropTableDependents(b BuildCtx, tbl catalog.TableDescriptor, behavior tree.
 				ReferenceColumns: fk.OriginColumns,
 			})
 		})
-		// Detect any sequence ownerships and clean them up no cascade
-		// is required.
+		// Detect any sequence ownerships and clean them up if only the
+		// cascade option is specified, specifically if other objects rely
+		// on the sequence.
+		cleanSequenceOwnedBy := func(seq catalog.TableDescriptor) {
+			dropSequence(b, seq, tree.DropCascade)
+			if behavior == tree.DropCascade {
+				return
+			}
+			scpb.ForEachRelationDependedOnBy(c, func(
+				status scpb.Status,
+				targetStatus scpb.Status,
+				depBy *scpb.RelationDependedOnBy,
+			) {
+				if depBy.TableID != seq.GetID() {
+					return
+				}
+				if depBy.DependedOnBy == tbl.GetID() {
+					return
+				}
+				panic(pgerror.Newf(
+					pgcode.DependentObjectsStillExist,
+					"cannot drop table a because other objects depend on it"))
+			})
+		}
 		scpb.ForEachSequenceOwnedBy(c, func(_, _ scpb.Status, sequenceOwnedBy *scpb.SequenceOwnedBy) {
 			if sequenceOwnedBy.OwnerTableID != tbl.GetID() {
 				return
 			}
-			sequence := c.MustReadTable(sequenceOwnedBy.SequenceID)
-			dropSequence(b, sequence, tree.DropCascade)
+			seq := c.MustReadTable(sequenceOwnedBy.SequenceID)
+			cleanSequenceOwnedBy(seq)
 		})
 	}
 }
diff --git a/pkg/sql/schemachanger/scbuild/internal/scbuildstmt/drop_view.go b/pkg/sql/schemachanger/scbuild/internal/scbuildstmt/drop_view.go
@@ -97,13 +97,12 @@ func dropViewDependents(b BuildCtx, view catalog.TableDescriptor, behavior tree.
 			depViewName, err := b.CatalogReader().GetQualifiedTableNameByID(b, int64(dep.DependedOnBy), tree.ResolveRequireViewDesc)
 			onErrPanic(err)
 			if dependentDesc.GetParentID() != view.GetParentID() {
-				panic(sqlerrors.NewDependentObjectErrorf("cannot drop view %q because view %q depends on it",
-					name.FQString(), depViewName.FQString()))
-			} else {
-				panic(errors.WithHintf(
-					sqlerrors.NewDependentObjectErrorf("cannot drop view %q because view %q depends on it",
-						name.Object(), depViewName.Object()),
+				panic(errors.WithHintf(sqlerrors.NewDependentObjectErrorf("cannot drop relation %q because view %q depends on it",
+					name.Object(), depViewName.FQString()),
 					"you can drop %s instead.", depViewName.Object()))
+			} else {
+				panic(sqlerrors.NewDependentObjectErrorf("cannot drop relation %q because view %q depends on it",
+					name.Object(), depViewName.Object()))
 			}
 		}
 		// Decompose and recursively attempt to drop.

diff --git a/pkg/sql/schemachanger/scbuild/name_resolver.go b/pkg/sql/schemachanger/scbuild/name_resolver.go
@@ -33,6 +33,9 @@ func (b buildCtx) ResolveDatabase(
 		if p.IsExistenceOptional {
 			return nil
 		}
+		if string(name) == "" {
+			panic(pgerror.New(pgcode.Syntax, "empty database name"))
+		}
 		panic(sqlerrors.NewUndefinedDatabaseError(name.String()))
 	}
 	if err := b.AuthorizationAccessor().CheckPrivilege(b, db, p.RequiredPrivilege); err != nil {
@@ -50,7 +53,7 @@ func (b buildCtx) ResolveSchema(
 		if p.IsExistenceOptional {
 			return db, nil
 		}
-		panic(sqlerrors.NewUndefinedSchemaError(name.String()))
+		panic(sqlerrors.NewUndefinedSchemaError(name.Schema()))
 	}
 	switch sc.SchemaKind() {
 	case catalog.SchemaPublic, catalog.SchemaVirtual, catalog.SchemaTemporary:

diff --git a/pkg/sql/schemachanger/scdeps/exec_deps.go b/pkg/sql/schemachanger/scdeps/exec_deps.go
@@ -135,6 +135,7 @@ func (d *txnDeps) GetFullyQualifiedName(ctx context.Context, id descpb.ID) (stri
 		tree.CommonLookupFlags{
 			Required:       true,
 			IncludeDropped: true,
+			AvoidLeased:    true,
 		})
 	if err != nil {
 		return "", err
@@ -149,14 +150,17 @@ func (d *txnDeps) GetFullyQualifiedName(ctx context.Context, id descpb.ID) (stri
 			tree.CommonLookupFlags{
 				IncludeDropped: true,
 				Required:       true,
+				AvoidLeased:    true,
 			})
 		if err != nil {
 			return "", err
 		}
 		schemaDesc, err := d.descsCollection.GetImmutableSchemaByID(ctx, d.txn, objectDesc.GetParentSchemaID(),
 			tree.SchemaLookupFlags{
 				Required:       true,
-				IncludeDropped: true})
+				IncludeDropped: true,
+				AvoidLeased:    true,
+			})
 		if err != nil {
 			return "", err
 		}
@@ -173,6 +177,7 @@ func (d *txnDeps) GetFullyQualifiedName(ctx context.Context, id descpb.ID) (stri
 			tree.CommonLookupFlags{
 				IncludeDropped: true,
 				Required:       true,
+				AvoidLeased:    true,
 			})
 		if err != nil {
 			return "", err