sql: owning a schema gives privilege to drop tables in schema

pkg/sql/authorization.go

@@ -565,3 +565,41 @@ func (p *planner) checkCanAlterToNewOwner(
 
 	return nil
 }
+
+// HasOwnershipOnSchema checks if the current user has ownership on the schema.
+// For schemas, we cannot always use HasOwnership as not every schema has a
+// descriptor.
+func (p *planner) HasOwnershipOnSchema(ctx context.Context, schemaID descpb.ID) (bool, error) {
+	resolvedSchema, err := p.Descriptors().ResolveSchemaByID(
+		ctx, p.Txn(), schemaID,
+	)
+	if err != nil {
+		return false, err
+	}
+
+	hasOwnership := false
+	switch resolvedSchema.Kind {
+	case catalog.SchemaPublic:
+		// admin is the owner of the public schema.
+		hasOwnership, err = p.UserHasAdminRole(ctx, p.User())
+		if err != nil {
+			return false, err
+		}
+	case catalog.SchemaVirtual:
+		// Cannot drop on virtual schemas.
+	case catalog.SchemaTemporary:
+		// The user only owns the temporary schema that corresponds to the
+		// TemporarySchemaID in the sessionData.
+		hasOwnership = p.SessionData() != nil &&
+			p.SessionData().TemporarySchemaID == uint32(resolvedSchema.ID)
+	case catalog.SchemaUserDefined:
+		hasOwnership, err = p.HasOwnership(ctx, resolvedSchema.Desc)
+		if err != nil {
+			return false, err
+		}
+	default:
+		panic(errors.AssertionFailedf("unknown schema kind %d", resolvedSchema.Kind))
+	}
+
+	return hasOwnership, nil
+}

pkg/sql/catalog/descs/collection.go

@@ -890,6 +890,7 @@ func (tc *Collection) ResolveSchemaByID(
 
 	// If this collection is attached to a session and the session has created
 	// a temporary schema, then check if the schema ID matches.
+	// This check doesn't work after switching databases. Issue #53163.
 	if tc.sessionData != nil && tc.sessionData.TemporarySchemaID == uint32(schemaID) {
 		return catalog.ResolvedSchema{
 			Kind: catalog.SchemaTemporary,

pkg/sql/drop_cascade.go

@@ -100,7 +100,7 @@ func (d *dropCascadeState) resolveCollectedObjects(
 						" dropped or made public before dropping database %s",
 					objName.FQString(), tree.AsString((*tree.Name)(&dbName)))
 			}
-			if err := p.prepareDropWithTableDesc(ctx, tbDesc); err != nil {
+			if err := p.canDropTable(ctx, tbDesc); err != nil {
 				return err
 			}
 			// Recursively check permissions on all dependent views, since some may

pkg/sql/drop_table.go

@@ -173,20 +173,31 @@ func (p *planner) prepareDrop(
 	if tableDesc == nil {
 		return nil, err
 	}
-	if err := p.prepareDropWithTableDesc(ctx, tableDesc); err != nil {
+	if err := p.canDropTable(ctx, tableDesc); err != nil {
 		return nil, err
 	}
 	return tableDesc, nil
 }
 
-// prepareDropWithTableDesc behaves as prepareDrop, except it assumes the
-// table descriptor is already fetched. This is useful for DropDatabase,
-// as prepareDrop requires resolving a TableName when DropDatabase already
-// has it resolved.
-func (p *planner) prepareDropWithTableDesc(
-	ctx context.Context, tableDesc *tabledesc.Mutable,
-) error {
-	return p.CheckPrivilege(ctx, tableDesc, privilege.DROP)
+// canDropTable returns an error if the user cannot drop the table.
+func (p *planner) canDropTable(ctx context.Context, tableDesc *tabledesc.Mutable) error {
+	// Don't check for ownership if the table is a temp table.
+	// ResolveSchema currently doesn't work for temp schemas.
+	// Hack until #53163 is fixed.
+	hasOwnership := false
+	var err error
+	if !tableDesc.Temporary {
+		// If the user owns the schema the table is part of, they can drop the table.
+		hasOwnership, err = p.HasOwnershipOnSchema(ctx, tableDesc.GetParentSchemaID())
+		if err != nil {
+			return err
+		}
+	}
+	if !hasOwnership {
+		return p.CheckPrivilege(ctx, tableDesc, privilege.DROP)
+	}
+
+	return nil
 }
 
 // canRemoveFKBackReference returns an error if the input backreference isn't

pkg/sql/logictest/testdata/logic_test/drop_table

@@ -56,3 +56,22 @@ CREATE TABLE a (id INT PRIMARY KEY)
 query I
 SELECT * FROM a
 ----
+
+user testuser
+
+statement ok
+SET experimental_enable_user_defined_schemas = true
+
+statement ok
+CREATE SCHEMA s
+
+user root
+
+statement ok
+CREATE TABLE s.t()
+
+user testuser
+
+# Being the owner of schema s should allow testuser to drop table s.t.
+statement ok
+DROP TABLE s.t