sql: support default privileges at the database level

docs/generated/sql/bnf/alter_ddl_stmt.bnf

@@ -8,3 +8,4 @@ alter_ddl_stmt ::=
 	| alter_partition_stmt
 	| alter_schema_stmt
 	| alter_type_stmt
+	| alter_default_privileges_stmt

docs/generated/sql/bnf/alter_default_privileges_stmt.bnf

@@ -0,0 +1,3 @@
+alter_default_privileges_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

@@ -379,6 +379,7 @@ alter_ddl_stmt ::=
 	| alter_partition_stmt
 	| alter_schema_stmt
 	| alter_type_stmt
+	| alter_default_privileges_stmt
 
 alter_role_stmt ::=
 	'ALTER' role_or_group_or_user string_or_placeholder opt_role_options
@@ -887,6 +888,7 @@ unreserved_keyword ::=
 	| 'FORCE'
 	| 'FORCE_INDEX'
 	| 'FUNCTION'
+	| 'FUNCTIONS'
 	| 'GENERATED'
 	| 'GEOMETRYM'
 	| 'GEOMETRYZ'
@@ -1048,6 +1050,7 @@ unreserved_keyword ::=
 	| 'ROLES'
 	| 'ROLLBACK'
 	| 'ROLLUP'
+	| 'ROUTINES'
 	| 'ROWS'
 	| 'RULE'
 	| 'RUNNING'
@@ -1309,6 +1312,10 @@ alter_type_stmt ::=
 	| 'ALTER' 'TYPE' type_name 'SET' 'SCHEMA' schema_name
 	| 'ALTER' 'TYPE' type_name 'OWNER' 'TO' role_spec
 
+alter_default_privileges_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'
 	| 'USER'
@@ -1780,6 +1787,21 @@ opt_add_val_placement ::=
 	| 'AFTER' 'SCONST'
 	| 
 
+opt_for_roles ::=
+	'FOR' role_or_group_or_user name_list
+	| 
+
+opt_in_schemas ::=
+	'IN' 'SCHEMA' schema_name_list
+	| 
+
+abbreviated_grant_stmt ::=
+	'GRANT' privileges 'ON' alter_default_privileges_target_object 'TO' name_list opt_with_grant_option
+
+abbreviated_revoke_stmt ::=
+	'REVOKE' privileges 'ON' alter_default_privileges_target_object 'FROM' name_list opt_drop_behavior
+	| 'REVOKE' 'GRANT' 'OPTION' 'FOR' privileges 'ON' alter_default_privileges_target_object 'FROM' name_list opt_drop_behavior
+
 role_options ::=
 	( role_option ) ( ( role_option ) )*
 
@@ -2250,6 +2272,16 @@ survival_goal_clause ::=
 primary_region_clause ::=
 	'PRIMARY' 'REGION' opt_equal region_name
 
+alter_default_privileges_target_object ::=
+	'TABLES'
+	| 'SEQUENCES'
+	| 'TYPES'
+	| 'SCHEMAS'
+
+opt_with_grant_option ::=
+	'WITH' 'GRANT' 'OPTION'
+	| 
+
 role_option ::=
 	'CREATEROLE'
 	| 'NOCREATEROLE'

pkg/ccl/backupccl/restore_job.go

@@ -2406,11 +2406,11 @@ func getRestoringPrivileges(
 				return nil, errors.Wrapf(err, "failed to lookup parent DB %d", errors.Safe(desc.GetParentID()))
 			}
 
-			// Default is to copy privs from restoring parent db, like CREATE {TABLE,
-			// 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.GetDefaultPrivileges(), user, tree.Tables,
+				parentDB.GetPrivileges(),
+			)
 		}
 	case catalog.TypeDescriptor, catalog.DatabaseDescriptor:
 		if descCoverage == tree.RequestedDescriptors {

pkg/ccl/importccl/import_stmt_test.go

@@ -5561,7 +5561,6 @@ func TestImportPgDumpIgnoredStmts(t *testing.T) {
 				SELECT $1 ~ '^[0-9]+$'
 				$_$;
 				ALTER FUNCTION public.isnumeric(text) OWNER TO roland;
-        ALTER DEFAULT PRIVILEGES FOR ROLE rolename IN SCHEMA "schemaname" REVOKE ALL ON TABLES FROM rolename;
 
         ALTER TABLE "database"."table" ALTER COLUMN "Id" ADD GENERATED BY DEFAULT AS IDENTITY (
             SEQUENCE NAME "database"."sequencename"
@@ -5675,7 +5674,6 @@ alter domain: could not be parsed
 create function: could not be parsed
 `,
 			`alter function: could not be parsed
-alter default privileges: could not be parsed
 alter table alter column add: could not be parsed
 copy from unsupported format: could not be parsed
 grant privileges on schema with: could not be parsed

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

@@ -0,0 +1,151 @@
+// Copyright 2021 The Cockroach Authors.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package sql
+
+import (
+	"context"
+
+	"github.com/cockroachdb/cockroach/pkg/security"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/dbdesc"
+	"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/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
+
+	dbDesc *dbdesc.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) {
+	// 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 {
+		for _, role := range n.n.Roles {
+			user, err := security.MakeSQLUsernameFromUserInput(string(role), security.UsernameValidation)
+			if err != nil {
+				return err
+			}
+			targetRoles = append(targetRoles, user)
+		}
+	}
+
+	if err := params.p.validateRoles(params.ctx, targetRoles, false /* isPublicValid */); err != nil {
+		return err
+	}
+
+	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
+	}
+
+	granteeSQLUsernames := make([]security.SQLUsername, len(grantees))
+	for i, grantee := range grantees {
+		user, err := security.MakeSQLUsernameFromUserInput(string(grantee), security.UsernameValidation)
+		if err != nil {
+			return err
+		}
+		granteeSQLUsernames[i] = user
+	}
+
+	if err := params.p.validateRoles(params.ctx, granteeSQLUsernames, true /* isPublicValid */); err != nil {
+		return err
+	}
+
+	// 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
+	}
+
+	if n.dbDesc.GetDefaultPrivileges() == nil {
+		n.dbDesc.SetDefaultPrivilegeDescriptor(descpb.InitDefaultPrivilegeDescriptor())
+	}
+
+	defaultPrivs := n.dbDesc.GetDefaultPrivileges()
+
+	for _, targetRole := range targetRoles {
+		if n.n.IsGrant {
+			defaultPrivs.GrantDefaultPrivileges(
+				targetRole, privileges, grantees, objectType,
+			)
+		} else {
+			defaultPrivs.RevokeDefaultPrivileges(
+				targetRole, privileges, grantees, objectType,
+			)
+		}
+	}
+
+	return params.p.writeNonDropDatabaseChange(
+		params.ctx, n.dbDesc, tree.AsStringWithFQNames(n.n, params.Ann()),
+	)
+}

pkg/sql/catalog/dbdesc/database_desc.go

@@ -214,6 +214,12 @@ func (desc *immutable) ValidateSelf(vea catalog.ValidationErrorAccumulator) {
 	// Validate the privilege descriptor.
 	vea.Report(desc.Privileges.Validate(desc.GetID(), privilege.Database))
 
+	// The DefaultPrivilegeDescriptor may be nil.
+	if desc.GetDefaultPrivileges() != nil {
+		// Validate the default privilege descriptor.
+		vea.Report(desc.GetDefaultPrivileges().Validate())
+	}
+
 	if desc.IsMultiRegion() {
 		desc.validateMultiRegion(vea)
 	}
@@ -404,3 +410,11 @@ func (desc *Mutable) SetRegionConfig(cfg *descpb.DatabaseDescriptor_RegionConfig
 func (desc *Mutable) HasPostDeserializationChanges() bool {
 	return desc.changed
 }
+
+// SetDefaultPrivilegeDescriptor sets the default privilege descriptor
+// for the database.
+func (desc *Mutable) SetDefaultPrivilegeDescriptor(
+	defaultPrivilegeDescriptor *descpb.DefaultPrivilegeDescriptor,
+) {
+	desc.DefaultPrivileges = defaultPrivilegeDescriptor
+}

pkg/sql/catalog/dbdesc/database_desc_builder.go

@@ -141,20 +141,26 @@ func NewInitial(
 		id,
 		name,
 		descpb.NewDefaultPrivilegeDescriptor(owner),
+		descpb.InitDefaultPrivilegeDescriptor(),
 		options...,
 	)
 }
 
 // NewInitialWithPrivileges constructs a new Mutable for an initial version
 // from an id and name and custom privileges.
 func NewInitialWithPrivileges(
-	id descpb.ID, name string, privileges *descpb.PrivilegeDescriptor, options ...NewInitialOption,
+	id descpb.ID,
+	name string,
+	privileges *descpb.PrivilegeDescriptor,
+	defaultPrivileges *descpb.DefaultPrivilegeDescriptor,
+	options ...NewInitialOption,
 ) *Mutable {
 	ret := descpb.DatabaseDescriptor{
-		Name:       name,
-		ID:         id,
-		Version:    1,
-		Privileges: privileges,
+		Name:              name,
+		ID:                id,
+		Version:           1,
+		Privileges:        privileges,
+		DefaultPrivileges: defaultPrivileges,
 	}
 	for _, option := range options {
 		option(&ret)

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",
     ],