sql: implement CREATE SCHEMA ... AUTHORIZATION

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.