sql: alter schema owner

docs/generated/sql/bnf/stmt_block.bnf

@@ -1128,6 +1128,7 @@ alter_partition_stmt ::=
 
 alter_schema_stmt ::=
 	'ALTER' 'SCHEMA' schema_name 'RENAME' 'TO' schema_name
+	| 'ALTER' 'SCHEMA' schema_name 'OWNER' 'TO' role_spec
 
 alter_type_stmt ::=
 	'ALTER' 'TYPE' type_name 'ADD' 'VALUE' 'SCONST' opt_add_val_placement
@@ -1547,6 +1548,11 @@ alter_zone_partition_stmt ::=
 schema_name ::=
 	name
 
+role_spec ::=
+	non_reserved_word_or_sconst
+	| 'CURRENT_USER'
+	| 'SESSION_USER'
+
 opt_add_val_placement ::=
 	'BEFORE' 'SCONST'
 	| 'AFTER' 'SCONST'
@@ -1953,11 +1959,6 @@ alter_index_cmds ::=
 sequence_option_list ::=
 	( sequence_option_elem ) ( ( sequence_option_elem ) )*
 
-role_spec ::=
-	non_reserved_word_or_sconst
-	| 'CURRENT_USER'
-	| 'SESSION_USER'
-
 role_option ::=
 	'CREATEROLE'
 	| 'NOCREATEROLE'

pkg/sql/alter_schema.go

@@ -17,6 +17,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
 	"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/log"
@@ -63,11 +64,47 @@ func (n *alterSchemaNode) startExec(params runParams) error {
 	switch t := n.n.Cmd.(type) {
 	case *tree.AlterSchemaRename:
 		return params.p.renameSchema(params.ctx, n.db, n.desc, t.NewName, tree.AsStringWithFQNames(n.n, params.Ann()))
+	case *tree.AlterSchemaOwner:
+		return params.p.alterSchemaOwner(params.ctx, n.db, n.desc, t.Owner, tree.AsStringWithFQNames(n.n, params.Ann()))
 	default:
 		return errors.AssertionFailedf("unknown schema cmd %T", t)
 	}
 }
 
+func (p *planner) alterSchemaOwner(
+	ctx context.Context,
+	db *sqlbase.MutableDatabaseDescriptor,
+	scDesc *sqlbase.MutableSchemaDescriptor,
+	newOwner string,
+	jobDesc string,
+) error {
+	privs := scDesc.GetPrivileges()
+
+	hasOwnership, err := p.HasOwnership(ctx, scDesc)
+	if err != nil {
+		return err
+	}
+
+	if err := p.checkCanAlterToNewOwner(ctx, scDesc, privs, newOwner, hasOwnership); err != nil {
+		return err
+	}
+
+	// The new owner must also have CREATE privilege on the schema's database.
+	if err := p.CheckPrivilegeForUser(ctx, db, privilege.CREATE, newOwner); err != nil {
+		return err
+	}
+
+	// If the owner we want to set to is the current owner, do a no-op.
+	if newOwner == privs.Owner {
+		return nil
+	}
+
+	// Update the owner of the schema.
+	privs.SetOwner(newOwner)
+
+	return p.writeSchemaDescChange(ctx, scDesc, jobDesc)
+}
+
 func (p *planner) renameSchema(
 	ctx context.Context,
 	db *sqlbase.MutableDatabaseDescriptor,

pkg/sql/logictest/testdata/logic_test/alter_schema_owner

@@ -0,0 +1,81 @@
+statement ok
+SET experimental_enable_user_defined_schemas = true
+
+statement ok
+CREATE SCHEMA s
+
+# Ensure user must exist for set owner.
+statement error pq: role/user "fake_user" does not exist
+ALTER SCHEMA s OWNER TO fake_user
+
+# Ensure the current user is a member of the role we're setting to.
+statement error pq: must be member of role "testuser"
+ALTER SCHEMA s OWNER TO testuser
+
+user testuser
+
+# Ensure the user has to be an owner to alter the owner.
+statement error pq: must be owner of schema s
+ALTER SCHEMA s OWNER TO testuser
+
+user root
+
+statement ok
+GRANT testuser TO root
+
+statement ok
+CREATE USER testuser2
+
+statement ok
+GRANT testuser2 TO root
+
+# Ensure the desired owner has CREATE privilege on the database.
+statement error pq: user testuser2 does not have CREATE privilege on database test
+ALTER SCHEMA s OWNER TO testuser2
+
+statement ok
+GRANT CREATE ON DATABASE test TO testuser, testuser2
+
+# testuser has the required privileges to become the new owner of schema s.
+statement ok
+ALTER SCHEMA s OWNER TO testuser
+
+# setup to allow testuser2 as a member of testuser to alter the owner.
+statement ok
+REVOKE testuser, testuser2 FROM root
+
+statement ok
+GRANT testuser TO testuser2
+
+statement ok
+GRANT root TO testuser
+
+user testuser2
+
+# testuser2 should be able to alter the owner since it is a member of testuser.
+statement ok
+ALTER SCHEMA s OWNER TO root
+
+# set the owner back to testuser.
+
+user root
+
+statement ok
+REVOKE root FROM testuser
+
+statement ok
+GRANT testuser TO root
+
+statement ok
+ALTER SCHEMA s OWNER TO testuser
+
+# setup to allow testuser2 to become the owner again.
+statement ok
+REVOKE testuser FROM testuser2
+
+statement ok
+GRANT testuser2 TO testuser
+
+# Ensure testuser is owner by dropping the schema.
+statement ok
+DROP SCHEMA s

pkg/sql/parser/parse_test.go

@@ -1279,6 +1279,7 @@ func TestParse(t *testing.T) {
 		{`ALTER INDEX IF EXISTS a@primary RENAME TO like`},
 
 		{`ALTER SCHEMA s RENAME TO s2`},
+		{`ALTER SCHEMA s OWNER TO foo`},
 
 		{`ALTER TABLE a RENAME TO b`},
 		{`EXPLAIN ALTER TABLE a RENAME TO b`},

pkg/sql/parser/sql.y

@@ -5232,7 +5232,12 @@ alter_schema_stmt:
   }
 | ALTER SCHEMA schema_name OWNER TO role_spec
   {
-    return unimplementedWithIssueDetail(sqllex, 50882, "ALTER SCHEMA OWNER TO")
+    $$.val = &tree.AlterSchema{
+      Schema: $3,
+      Cmd: &tree.AlterSchemaOwner{
+        Owner: $6,
+      },
+    }
   }
 | ALTER SCHEMA error // SHOW HELP: ALTER SCHEMA
 

pkg/sql/sem/tree/alter_schema.go

@@ -43,3 +43,16 @@ func (node *AlterSchemaRename) Format(ctx *FmtCtx) {
 	ctx.WriteString(" RENAME TO ")
 	ctx.FormatNameP(&node.NewName)
 }
+
+func (*AlterSchemaOwner) alterSchemaCmd() {}
+
+// AlterSchemaOwner represents an ALTER SCHEMA RENAME command.
+type AlterSchemaOwner struct {
+	Owner string
+}
+
+// Format implements the NodeFormatter interface.
+func (node *AlterSchemaOwner) Format(ctx *FmtCtx) {
+	ctx.WriteString(" OWNER TO ")
+	ctx.FormatNameP(&node.Owner)
+}