sql: support default privileges at the database level

Release note (sql change): Added support for ALTER DEFAULT PRIVILEGES
and default privileges stored on databases.

All objects created in a database will have the privilege set defined
by the default privileges for that type of object on the database.
The types of objects are TABLES, SEQUENCES, SCHEMAS, TYPES.

Example: ALTER DEFAULT PRIVILEGES GRANT SELECT ON TABLES TO foo
makes it such that all tables created by the user that executed the
ALTER DEFAULT PRIVILEGES command will have SELECT privilege on the table
for user foo.

Additionally, one can specify a role.
Example: ALTER DEFAULT PRIVILEGES FOR ROLE bar GRANT SELECT ON TABLES TO foo.
All tables created by bar will have SELECT privilege for foo.
If a role is not specified, it uses the current user.

See: https://www.postgresql.org/docs/current/sql-alterdefaultprivileges.html

Currently, default privileges are not supported on the schema.
Specifying a schema like ALTER DEFAULT PRIVILEGES IN SCHEMA s will error.

WITH GRANT OPTION is ignored.
GRANT OPTION FOR is also ignored.

docs/generated/sql/bnf/alter_default_privileges_stmt.bnf

@@ -1,3 +1,3 @@
 alter_default_privileges_stmt ::=
-	'ALTER' 'DEFAULT' 'PRIVILEGES' opt_for_role opt_in_schema abbreviated_grant_stmt
-	| 'ALTER' 'DEFAULT' 'PRIVILEGES' opt_for_role opt_in_schema abbreviated_revoke_stmt
+	'ALTER' 'DEFAULT' 'PRIVILEGES' opt_for_roles opt_in_schemas abbreviated_grant_stmt
+	| 'ALTER' 'DEFAULT' 'PRIVILEGES' opt_for_roles opt_in_schemas abbreviated_revoke_stmt

docs/generated/sql/bnf/stmt_block.bnf

@@ -1311,8 +1311,8 @@ alter_type_stmt ::=
 	| 'ALTER' 'TYPE' type_name 'OWNER' 'TO' role_spec
 
 alter_default_privileges_stmt ::=
-	'ALTER' 'DEFAULT' 'PRIVILEGES' opt_for_role opt_in_schema abbreviated_grant_stmt
-	| 'ALTER' 'DEFAULT' 'PRIVILEGES' opt_for_role opt_in_schema abbreviated_revoke_stmt
+	'ALTER' 'DEFAULT' 'PRIVILEGES' opt_for_roles opt_in_schemas abbreviated_grant_stmt
+	| 'ALTER' 'DEFAULT' 'PRIVILEGES' opt_for_roles opt_in_schemas abbreviated_revoke_stmt
 
 role_or_group_or_user ::=
 	'ROLE'
@@ -1785,12 +1785,12 @@ opt_add_val_placement ::=
 	| 'AFTER' 'SCONST'
 	| 
 
-opt_for_role ::=
-	'FOR' role_or_group_or_user role_spec
+opt_for_roles ::=
+	'FOR' role_or_group_or_user role_spec_list
 	| 
 
-opt_in_schema ::=
-	'IN' 'SCHEMA' qualifiable_schema_name
+opt_in_schemas ::=
+	'IN' 'SCHEMA' schema_name_list
 	| 
 
 abbreviated_grant_stmt ::=

pkg/ccl/backupccl/restore_job.go

@@ -2398,7 +2398,10 @@ func getRestoringPrivileges(
 			// SCHEMA}. But also like CREATE {TABLE,SCHEMA}, we set the owner to the
 			// user creating the table (the one running the restore).
 			// TODO(dt): Make this more configurable.
-			updatedPrivileges = sql.CreateInheritedPrivilegesFromDBDesc(parentDB, user)
+			updatedPrivileges = descpb.CreatePrivilegesFromDefaultPrivileges(
+				parentDB.GetID(), parentDB.GetDefaultPrivilegeDescriptor(), user, tree.Tables,
+				parentDB.GetPrivileges(),
+			)
 		}
 	case catalog.TypeDescriptor, catalog.DatabaseDescriptor:
 		if descCoverage == tree.RequestedDescriptors {

pkg/sql/BUILD.bazel

@@ -8,6 +8,7 @@ go_library(
         "add_column.go",
         "alter_column_type.go",
         "alter_database.go",
+        "alter_default_privileges.go",
         "alter_index.go",
         "alter_primary_key.go",
         "alter_role.go",

pkg/sql/alter_default_privileges.go

@@ -12,13 +12,149 @@ package sql
 
 import (
 	"context"
+	"fmt"
 
+	"github.com/cockroachdb/cockroach/pkg/security"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/dbdesc"
+	"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"
 )
 
+var targetObjectToPrivilegeObject = map[tree.AlterDefaultPrivilegesTargetObject]privilege.ObjectType{
+	tree.Tables:    privilege.Table,
+	tree.Sequences: privilege.Table,
+	tree.Types:     privilege.Type,
+	tree.Schemas:   privilege.Schema,
+}
+
+type alterDefaultPrivilegesNode struct {
+	n *tree.AlterDefaultPrivileges
+
+	// Only one of dbDesc or schemaDesc should be populated.
+	dbDesc     *dbdesc.Mutable
+	schemaDesc *schemadesc.Mutable
+}
+
+func (n *alterDefaultPrivilegesNode) Next(runParams) (bool, error) { return false, nil }
+func (n *alterDefaultPrivilegesNode) Values() tree.Datums          { return tree.Datums{} }
+func (n *alterDefaultPrivilegesNode) Close(context.Context)        {}
+
 func (p *planner) alterDefaultPrivileges(
 	ctx context.Context, n *tree.AlterDefaultPrivileges,
 ) (planNode, error) {
-	// TODO: implement this.
-	return nil, nil
+	// ALTER DEFAULT PRIVILEGES without specifying a schema alters the privileges
+	// for the current database.
+	database := p.CurrentDatabase()
+	dbDesc, err := p.Descriptors().GetMutableDatabaseByName(ctx, p.txn, database,
+		tree.DatabaseLookupFlags{Required: true})
+	if err != nil {
+		return nil, err
+	}
+
+	if len(n.Schemas) > 0 {
+		return nil, unimplemented.NewWithIssue(
+			67376, "ALTER DEFAULT PRIVILEGES IN SCHEMA not implemented",
+		)
+	}
+
+	return &alterDefaultPrivilegesNode{
+		n:      n,
+		dbDesc: dbDesc,
+	}, err
+}
+
+func (n *alterDefaultPrivilegesNode) startExec(params runParams) error {
+	var targetRoles []security.SQLUsername
+	if len(n.n.Roles) == 0 {
+		targetRoles = append(targetRoles, params.p.User())
+	} else {
+		targetRoles = n.n.Roles
+	}
+
+	users, err := params.p.GetAllRoles(params.ctx)
+	if err != nil {
+		return err
+	}
+
+	for _, targetRole := range targetRoles {
+		if _, found := users[targetRole]; !found {
+			return pgerror.Newf(pgcode.UndefinedObject,
+				"role %s does not exist", targetRole.Normalized())
+		}
+	}
+
+	privileges := n.n.Grant.Privileges
+	grantees := n.n.Grant.Grantees
+	objectType := n.n.Grant.Target
+	if !n.n.IsGrant {
+		privileges = n.n.Revoke.Privileges
+		grantees = n.n.Revoke.Grantees
+		objectType = n.n.Revoke.Target
+	}
+
+	granteesSQLUsername := make([]security.SQLUsername, len(grantees))
+	for i, grantee := range grantees {
+		granteesSQLUsername[i] = security.MakeSQLUsernameFromPreNormalizedString(string(grantee))
+	}
+
+	for _, grantee := range granteesSQLUsername {
+		if _, found := users[grantee]; !found {
+			return pgerror.Newf(pgcode.UndefinedObject,
+				"role %s does not exist", grantee.Normalized())
+		}
+	}
+
+	// You can change default privileges only for objects that will be created
+	// by yourself or by roles that you are a member of.
+	for _, targetRole := range targetRoles {
+		if targetRole != params.p.User() {
+			memberOf, err := params.p.MemberOfWithAdminOption(params.ctx, params.p.User())
+			if err != nil {
+				return err
+			}
+
+			if _, found := memberOf[targetRole]; !found {
+				return pgerror.Newf(pgcode.InsufficientPrivilege,
+					"must be a member of %s", targetRole.Normalized())
+			}
+		}
+	}
+
+	if err := privilege.ValidatePrivileges(
+		privileges,
+		targetObjectToPrivilegeObject[objectType],
+	); err != nil {
+		return err
+	}
+
+	defaultPrivs := n.dbDesc.GetDefaultPrivilegeDescriptor()
+
+	for _, targetRole := range targetRoles {
+		if n.n.IsGrant {
+			defaultPrivs.GrantDefaultPrivileges(
+				targetRole, privileges, grantees, objectType,
+			)
+		} else {
+			defaultPrivs.RevokeDefaultPrivileges(
+				targetRole, privileges, grantees, objectType,
+			)
+		}
+	}
+
+	err = params.p.writeNonDropDatabaseChange(
+		params.ctx, n.dbDesc, tree.AsStringWithFQNames(n.n, params.Ann()),
+	)
+	if err != nil {
+		return err
+	}
+
+	if err := params.p.createNonDropDatabaseChangeJob(params.ctx, n.dbDesc.ID,
+		fmt.Sprintf("updating privileges for database %d", n.dbDesc.ID)); err != nil {
+		return err
+	}
+	return nil
 }

pkg/sql/catalog/dbdesc/database_desc.go

@@ -214,6 +214,9 @@ func (desc *immutable) ValidateSelf(vea catalog.ValidationErrorAccumulator) {
 	// Validate the privilege descriptor.
 	vea.Report(desc.Privileges.Validate(desc.GetID(), privilege.Database))
 
+	// Validate the default privilege descriptor.
+	vea.Report(desc.GetDefaultPrivilegeDescriptor().Validate())
+
 	if desc.IsMultiRegion() {
 		desc.validateMultiRegion(vea)
 	}
@@ -335,6 +338,17 @@ func (desc *Mutable) ImmutableCopy() catalog.Descriptor {
 	return imm
 }
 
+// GetDefaultPrivilegeDescriptor gets or creates the DefaultPrivilegeDescriptor
+// for the database.
+func (desc *immutable) GetDefaultPrivilegeDescriptor() *descpb.DefaultPrivilegeDescriptor {
+	if desc.DefaultPrivileges == nil {
+		defaultPrivileges := descpb.DefaultPrivilegeDescriptor{}
+		desc.DefaultPrivileges = &defaultPrivileges
+	}
+
+	return desc.DefaultPrivileges
+}
+
 // IsNew implements the MutableDescriptor interface.
 func (desc *Mutable) IsNew() bool {
 	return desc.ClusterVersion == nil

pkg/sql/catalog/descpb/BUILD.bazel

@@ -8,6 +8,7 @@ go_library(
     srcs = [
         "column.go",
         "constraint.go",
+        "default_privilege.go",
         "descriptor.go",
         "index.go",
         "join_type.go",
@@ -36,6 +37,7 @@ go_library(
         "//pkg/util/errorutil/unimplemented",
         "//pkg/util/hlc",
         "//pkg/util/log",
+        "//pkg/util/protoutil",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_redact//:redact",
     ],