sql: implement CREATE SCHEMA ... AUTHORIZATION

Fixes #53559.

This commit adds the `CREATE SCHEMA ... AUTHORIZATION` command. When
authorization is provided, the target user is given ownership of the
schema. If the schema name is not provided, then the schema is named the
same name as the target role.

Release justification: low risk updates to new functionality
Release note (sql change): Support the `CREATE SCHEMA ... AUTHORIZATION`
command.

docs/generated/sql/bnf/stmt_block.bnf

@@ -739,7 +739,6 @@ unreserved_keyword ::=
 	| 'AT'
 	| 'ATTRIBUTE'
 	| 'AUTOMATIC'
-	| 'AUTHORIZATION'
 	| 'BACKUP'
 	| 'BACKUPS'
 	| 'BEFORE'
@@ -1215,6 +1214,8 @@ create_index_stmt ::=
 create_schema_stmt ::=
 	'CREATE' 'SCHEMA' schema_name
 	| 'CREATE' 'SCHEMA' 'IF' 'NOT' 'EXISTS' schema_name
+	| 'CREATE' 'SCHEMA' opt_schema_name 'AUTHORIZATION' role_spec
+	| 'CREATE' 'SCHEMA' 'IF' 'NOT' 'EXISTS' opt_schema_name 'AUTHORIZATION' role_spec
 
 create_table_stmt ::=
 	'CREATE' opt_persistence_temp_table 'TABLE' table_name '(' opt_table_elem_list ')' opt_interleave opt_partition_by
@@ -1690,6 +1691,9 @@ opt_partition_by ::=
 	partition_by
 	| 
 
+opt_schema_name ::=
+	opt_name
+
 opt_persistence_temp_table ::=
 	opt_temp
 	| 'LOCAL' 'TEMPORARY'
@@ -2217,7 +2221,8 @@ row_source_extension_stmt ::=
 	| upsert_stmt
 
 type_func_name_no_crdb_extra_keyword ::=
-	'COLLATION'
+	'AUTHORIZATION'
+	| 'COLLATION'
 	| 'CROSS'
 	| 'FULL'
 	| 'INNER'

pkg/sql/create_schema.go

@@ -50,8 +50,13 @@ func (p *planner) createUserDefinedSchema(params runParams, n *tree.CreateSchema
 		return pgerror.New(pgcode.InvalidObjectDefinition, "cannot create schemas in the system database")
 	}
 
+	schemaName := n.Schema
+	if n.Schema == "" {
+		schemaName = n.AuthRole
+	}
+
 	// Ensure there aren't any name collisions.
-	exists, err := p.schemaExists(params.ctx, db.ID, n.Schema)
+	exists, err := p.schemaExists(params.ctx, db.ID, schemaName)
 	if err != nil {
 		return err
 	}
@@ -60,11 +65,11 @@ func (p *planner) createUserDefinedSchema(params runParams, n *tree.CreateSchema
 		if n.IfNotExists {
 			return nil
 		}
-		return pgerror.Newf(pgcode.DuplicateSchema, "schema %q already exists", n.Schema)
+		return pgerror.Newf(pgcode.DuplicateSchema, "schema %q already exists", schemaName)
 	}
 
 	// Check validity of the schema name.
-	if err := schemadesc.IsSchemaNameValid(n.Schema); err != nil {
+	if err := schemadesc.IsSchemaNameValid(schemaName); err != nil {
 		return err
 	}
 
@@ -89,12 +94,24 @@ func (p *planner) createUserDefinedSchema(params runParams, n *tree.CreateSchema
 
 	// Inherit the parent privileges.
 	privs := db.GetPrivileges()
-	privs.SetOwner(params.SessionData().User)
+
+	if n.AuthRole != "" {
+		exists, err := p.RoleExists(params.ctx, n.AuthRole)
+		if err != nil {
+			return err
+		}
+		if !exists {
+			return pgerror.Newf(pgcode.UndefinedObject, "role/user %q does not exist", n.AuthRole)
+		}
+		privs.SetOwner(n.AuthRole)
+	} else {
+		privs.SetOwner(params.SessionData().User)
+	}
 
 	// Create the SchemaDescriptor.
 	desc := schemadesc.NewCreatedMutable(descpb.SchemaDescriptor{
 		ParentID:   db.ID,
-		Name:       n.Schema,
+		Name:       schemaName,
 		ID:         id,
 		Privileges: privs,
 		Version:    1,
@@ -119,7 +136,7 @@ func (p *planner) createUserDefinedSchema(params runParams, n *tree.CreateSchema
 	// Finally create the schema on disk.
 	return p.createDescriptorWithID(
 		params.ctx,
-		catalogkeys.NewSchemaKey(db.ID, n.Schema).Key(p.ExecCfg().Codec),
+		catalogkeys.NewSchemaKey(db.ID, schemaName).Key(p.ExecCfg().Codec),
 		id,
 		desc,
 		params.ExecCfg().Settings,

pkg/sql/logictest/testdata/logic_test/schema

@@ -348,3 +348,38 @@ COMMENT ON COLUMN privs.usage_tbl.x IS 'foo'
 
 statement error pq: user testuser does not have USAGE privilege on schema privs
 ALTER TYPE privs.usage_typ ADD VALUE 'denied'
+
+subtest authorization
+
+user root
+# Test the AUTHORIZATION argument to CREATE SCHEMA.
+
+# Create a user to create a schema for.
+statement ok
+CREATE USER user1;
+
+# Creates a schema for named with user1 as the owner.
+statement ok
+CREATE SCHEMA AUTHORIZATION user1
+
+statement error pq: schema "user1" already exists
+CREATE SCHEMA AUTHORIZATION user1
+
+statement ok
+CREATE SCHEMA IF NOT EXISTS AUTHORIZATION user1
+
+statement ok
+CREATE SCHEMA user1_schema AUTHORIZATION user1
+
+# The created schemas should both be owned by user1.
+query TT
+SELECT
+  nspname, usename
+FROM
+  pg_catalog.pg_namespace
+  LEFT JOIN pg_catalog.pg_user ON pg_namespace.nspowner = pg_user.usesysid
+WHERE
+  nspname LIKE 'user1%';
+----
+user1         user1
+user1_schema  user1

pkg/sql/parser/parse_test.go

@@ -81,6 +81,10 @@ func TestParse(t *testing.T) {
 		{`CREATE DATABASE IF NOT EXISTS a TEMPLATE = 'template0' ENCODING = 'UTF8' LC_COLLATE = 'C.UTF-8' LC_CTYPE = 'INVALID'`},
 		{`CREATE SCHEMA IF NOT EXISTS foo`},
 		{`CREATE SCHEMA foo`},
+		{`CREATE SCHEMA IF NOT EXISTS foo AUTHORIZATION foobar`},
+		{`CREATE SCHEMA foo AUTHORIZATION foobar`},
+		{`CREATE SCHEMA IF NOT EXISTS AUTHORIZATION foobar`},
+		{`CREATE SCHEMA AUTHORIZATION foobar`},
 
 		{`CREATE INDEX a ON b (c)`},
 		{`CREATE INDEX CONCURRENTLY a ON b (c)`},

pkg/sql/parser/sql.y

@@ -934,7 +934,7 @@ func (u *sqlSymUnion) refreshDataOption() tree.RefreshDataOption {
 %type <str> db_object_name_component
 %type <*tree.UnresolvedObjectName> table_name standalone_index_name sequence_name type_name view_name db_object_name simple_db_object_name complex_db_object_name
 %type <[]*tree.UnresolvedObjectName> type_name_list
-%type <str> schema_name
+%type <str> schema_name opt_schema_name
 %type <[]string> schema_name_list
 %type <*tree.UnresolvedName> table_pattern complex_table_pattern
 %type <*tree.UnresolvedName> column_path prefixed_column_path column_path_with_star
@@ -5294,7 +5294,7 @@ pause_schedules_stmt:
 // %Help: CREATE SCHEMA - create a new schema
 // %Category: DDL
 // %Text:
-// CREATE SCHEMA [IF NOT EXISTS] <schemaname>
+// CREATE SCHEMA [IF NOT EXISTS] { <schemaname> | [<schemaname>] AUTHORIZATION <rolename> }
 create_schema_stmt:
   CREATE SCHEMA schema_name
   {
@@ -5309,6 +5309,21 @@ create_schema_stmt:
       IfNotExists: true,
     }
   }
+| CREATE SCHEMA opt_schema_name AUTHORIZATION role_spec
+  {
+    $$.val = &tree.CreateSchema{
+      Schema: $3,
+      AuthRole: $5,
+    }
+  }
+| CREATE SCHEMA IF NOT EXISTS opt_schema_name AUTHORIZATION role_spec
+  {
+    $$.val = &tree.CreateSchema{
+      Schema: $6,
+      IfNotExists: true,
+      AuthRole: $8,
+    }
+  }
 | CREATE SCHEMA error // SHOW HELP: CREATE SCHEMA
 
 // %Help: ALTER SCHEMA - alter an existing schema
@@ -11104,6 +11119,8 @@ sequence_name:         db_object_name
 
 schema_name:           name
 
+opt_schema_name:       opt_name
+
 table_name:            db_object_name
 
 standalone_index_name: db_object_name
@@ -11317,7 +11334,6 @@ unreserved_keyword:
 | AT
 | ATTRIBUTE
 | AUTOMATIC
-| AUTHORIZATION
 | BACKUP
 | BACKUPS
 | BEFORE
@@ -11688,7 +11704,8 @@ type_func_name_keyword:
 //
 // See type_func_name_crdb_extra_keyword below.
 type_func_name_no_crdb_extra_keyword:
-  COLLATION
+  AUTHORIZATION
+| COLLATION
 | CROSS
 | FULL
 | INNER

pkg/sql/sem/tree/create.go

@@ -1216,17 +1216,26 @@ func (node *CreateTable) HoistConstraints() {
 type CreateSchema struct {
 	IfNotExists bool
 	Schema      string
+	AuthRole    string
 }
 
 // Format implements the NodeFormatter interface.
 func (node *CreateSchema) Format(ctx *FmtCtx) {
-	ctx.WriteString("CREATE SCHEMA ")
+	ctx.WriteString("CREATE SCHEMA")
 
 	if node.IfNotExists {
-		ctx.WriteString("IF NOT EXISTS ")
+		ctx.WriteString(" IF NOT EXISTS")
 	}
 
-	ctx.WriteString(node.Schema)
+	if node.Schema != "" {
+		ctx.WriteString(" ")
+		ctx.WriteString(node.Schema)
+	}
+
+	if node.AuthRole != "" {
+		ctx.WriteString(" AUTHORIZATION ")
+		ctx.WriteString(node.AuthRole)
+	}
 }
 
 // CreateSequence represents a CREATE SEQUENCE statement.