sql: support ALTER DEFAULT PRIVILEGES IN SCHEMA

Release note (sql change):
Support ALTER DEFAULT PRIVILEGES IN SCHEMA <schemas...>
Users can now in addition to specifying default privileges globally
(within a database), specify default privileges in a specific schema.

When creating an object that has default privileges specified at
the database (global) and at the schema level, the union of the
default privileges are taken.

docs/generated/eventlog.md

@@ -1559,6 +1559,7 @@ An event of type `alter_default_privileges` is recorded when default privileges
 | `DatabaseName` | The name of the affected database. | yes |
 | `RoleName` | Either role_name should be populated or for_all_roles should be true. The role having its default privileges altered. | yes |
 | `ForAllRoles` | Identifies if FOR ALL ROLES is used. | no |
+| `SchemaName` | The names of the affected schema. | yes |
 
 
 #### Common fields

pkg/ccl/backupccl/restore_job.go

@@ -2639,7 +2639,9 @@ func getRestorePrivilegesForTableOrSchema(
 
 		// TODO(dt): Make this more configurable.
 		immutableDefaultPrivileges := parentDB.GetDefaultPrivilegeDescriptor()
-		updatedPrivileges = immutableDefaultPrivileges.CreatePrivilegesFromDefaultPrivileges(
+		updatedPrivileges = catprivilege.CreatePrivilegesFromDefaultPrivileges(
+			immutableDefaultPrivileges,
+			nil, /* schemaDefaultPrivilegeDescriptor */
 			parentDB.GetID(), user, tree.Tables, parentDB.GetPrivileges())
 	}
 	return updatedPrivileges, nil

pkg/sql/alter_default_privileges.go

@@ -18,12 +18,13 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catprivilege"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/dbdesc"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/schemadesc"
 	"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/util/errorutil/unimplemented"
 	"github.com/cockroachdb/cockroach/pkg/util/log/eventpb"
+	"github.com/cockroachdb/errors"
 )
 
 var targetObjectToPrivilegeObject = map[tree.AlterDefaultPrivilegesTargetObject]privilege.ObjectType{
@@ -36,7 +37,8 @@ var targetObjectToPrivilegeObject = map[tree.AlterDefaultPrivilegesTargetObject]
 type alterDefaultPrivilegesNode struct {
 	n *tree.AlterDefaultPrivileges
 
-	dbDesc *dbdesc.Mutable
+	dbDesc      *dbdesc.Mutable
+	schemaDescs []*schemadesc.Mutable
 }
 
 func (n *alterDefaultPrivilegesNode) Next(runParams) (bool, error) { return false, nil }
@@ -64,15 +66,31 @@ func (p *planner) alterDefaultPrivileges(
 		return nil, err
 	}
 
-	if len(n.Schemas) > 0 {
-		return nil, unimplemented.NewWithIssue(
-			67376, "ALTER DEFAULT PRIVILEGES IN SCHEMA not implemented",
+	objectType := n.Grant.Target
+	if !n.IsGrant {
+		objectType = n.Revoke.Target
+	}
+
+	if len(n.Schemas) > 0 && objectType == tree.Schemas {
+		return nil, pgerror.WithCandidateCode(errors.New(
+			"cannot use IN SCHEMA clause when using GRANT/REVOKE ON SCHEMAS"),
+			pgcode.InvalidGrantOperation,
 		)
 	}
 
+	var schemaDescs []*schemadesc.Mutable
+	for _, sc := range n.Schemas {
+		schemaDesc, err := p.Descriptors().GetMutableSchemaByName(ctx, p.txn, dbDesc, sc.Schema(), tree.SchemaLookupFlags{Required: true})
+		if err != nil {
+			return nil, err
+		}
+		schemaDescs = append(schemaDescs, schemaDesc.(*schemadesc.Mutable))
+	}
+
 	return &alterDefaultPrivilegesNode{
-		n:      n,
-		dbDesc: dbDesc,
+		n:           n,
+		dbDesc:      dbDesc,
+		schemaDescs: schemaDescs,
 	}, err
 }
 
@@ -139,8 +157,99 @@ func (n *alterDefaultPrivilegesNode) startExec(params runParams) error {
 		return err
 	}
 
+	if len(n.schemaDescs) == 0 {
+		return n.alterDefaultPrivilegesForDatabase(params, targetRoles, objectType, grantees, privileges, grantOption)
+	}
+	return n.alterDefaultPrivilegesForSchemas(params, targetRoles, objectType, grantees, privileges, grantOption)
+}
+
+func (n *alterDefaultPrivilegesNode) alterDefaultPrivilegesForSchemas(
+	params runParams,
+	targetRoles []security.SQLUsername,
+	objectType tree.AlterDefaultPrivilegesTargetObject,
+	grantees tree.RoleSpecList,
+	privileges privilege.List,
+	grantOption bool,
+) error {
+	var events []eventLogEntry
+	for _, schemaDesc := range n.schemaDescs {
+		if schemaDesc.GetDefaultPrivileges() == nil {
+			schemaDesc.SetDefaultPrivilegeDescriptor(catprivilege.MakeDefaultPrivilegeDescriptor(descpb.DefaultPrivilegeDescriptor_SCHEMA))
+		}
+
+		defaultPrivs := schemaDesc.GetMutableDefaultPrivilegeDescriptor()
+
+		var roles []descpb.DefaultPrivilegesRole
+		if n.n.ForAllRoles {
+			roles = append(roles, descpb.DefaultPrivilegesRole{
+				ForAllRoles: true,
+			})
+		} else {
+			roles = make([]descpb.DefaultPrivilegesRole, len(targetRoles))
+			for i, role := range targetRoles {
+				roles[i] = descpb.DefaultPrivilegesRole{
+					Role: role,
+				}
+			}
+		}
+
+		granteeSQLUsernames, err := grantees.ToSQLUsernames(params.SessionData(), security.UsernameValidation)
+		if err != nil {
+			return err
+		}
+		for _, role := range roles {
+			if n.n.IsGrant {
+				defaultPrivs.GrantDefaultPrivileges(
+					role, privileges, granteeSQLUsernames, objectType, grantOption,
+				)
+			} else {
+				defaultPrivs.RevokeDefaultPrivileges(
+					role, privileges, granteeSQLUsernames, objectType, grantOption,
+				)
+			}
+
+			eventDetails := eventpb.CommonSQLPrivilegeEventDetails{}
+			if n.n.IsGrant {
+				eventDetails.GrantedPrivileges = privileges.SortedNames()
+			} else {
+				eventDetails.RevokedPrivileges = privileges.SortedNames()
+			}
+			event := eventpb.AlterDefaultPrivileges{
+				CommonSQLPrivilegeEventDetails: eventDetails,
+				SchemaName:                     schemaDesc.GetName(),
+			}
+			if n.n.ForAllRoles {
+				event.ForAllRoles = true
+			} else {
+				event.RoleName = role.Role.Normalized()
+			}
+
+			events = append(events, eventLogEntry{
+				targetID: int32(n.dbDesc.GetID()),
+				event:    &event,
+			})
+
+			if err := params.p.writeSchemaDescChange(
+				params.ctx, schemaDesc, tree.AsStringWithFQNames(n.n, params.Ann()),
+			); err != nil {
+				return err
+			}
+		}
+	}
+
+	return params.p.logEvents(params.ctx, events...)
+}
+
+func (n *alterDefaultPrivilegesNode) alterDefaultPrivilegesForDatabase(
+	params runParams,
+	targetRoles []security.SQLUsername,
+	objectType tree.AlterDefaultPrivilegesTargetObject,
+	grantees tree.RoleSpecList,
+	privileges privilege.List,
+	grantOption bool,
+) error {
 	if n.dbDesc.GetDefaultPrivileges() == nil {
-		n.dbDesc.SetDefaultPrivilegeDescriptor(catprivilege.MakeNewDefaultPrivilegeDescriptor())
+		n.dbDesc.SetDefaultPrivilegeDescriptor(catprivilege.MakeDefaultPrivilegeDescriptor(descpb.DefaultPrivilegeDescriptor_DATABASE))
 	}
 
 	defaultPrivs := n.dbDesc.GetMutableDefaultPrivilegeDescriptor()
@@ -160,7 +269,7 @@ func (n *alterDefaultPrivilegesNode) startExec(params runParams) error {
 	}
 
 	var events []eventLogEntry
-	granteeSQLUsernames, err = grantees.ToSQLUsernames(params.SessionData(), security.UsernameValidation)
+	granteeSQLUsernames, err := grantees.ToSQLUsernames(params.SessionData(), security.UsernameValidation)
 	if err != nil {
 		return err
 	}