sql: fix schema privileges on descriptor read

Release note (bug fix): Previously a schema's privilege descriptor
could become corrupted upon executing ALTER DATABASE ...
CONVERT TO SCHEMA due to privileges that are invalid on a schema
being copied over to the schema rendering the schema inusable due
to invalid privileges.

pkg/sql/catalog/descpb/privilege.go

@@ -249,6 +249,27 @@ func MaybeFixUsagePrivForTablesAndDBs(ptr **PrivilegeDescriptor) bool {
 	return modified
 }
 
+// MaybeFixSchemaPrivileges removes all invalid bits set on a schema's
+// PrivilegeDescriptor.
+// This is necessary due to ALTER DATABASE ... CONVERT TO SCHEMA originally
+// copying all database privileges to the schema. Not all database privileges
+// are valid for schemas thus after running ALTER DATABASE ... CONVERT TO SCHEMA,
+// the schema may become unusable.
+func MaybeFixSchemaPrivileges(ptr **PrivilegeDescriptor) {
+	if *ptr == nil {
+		*ptr = &PrivilegeDescriptor{}
+	}
+	p := *ptr
+
+	validPrivs := privilege.GetValidPrivilegesForObject(privilege.Schema).ToBitField()
+
+	for i := range p.Users {
+		// Users is a slice of values, we need pointers to make them mutable.
+		userPrivileges := &p.Users[i]
+		userPrivileges.Privileges &= validPrivs
+	}
+}
+
 // MaybeFixPrivileges fixes the privilege descriptor if needed, including:
 // * adding default privileges for the "admin" role
 // * fixing default privileges for the "root" user

pkg/sql/catalog/descpb/privilege_test.go

@@ -889,3 +889,107 @@ func TestMaybeFixUsageAndZoneConfigPrivilege(t *testing.T) {
 	}
 
 }
+
+// TestMaybeFixSchemaPrivileges ensures that invalid privileges are removed
+// from a schema's privilege descriptor.
+func TestMaybeFixSchemaPrivileges(t *testing.T) {
+	fooUser := security.MakeSQLUsernameFromPreNormalizedString("foo")
+	barUser := security.MakeSQLUsernameFromPreNormalizedString("bar")
+
+	type userPrivileges map[security.SQLUsername]privilege.List
+
+	testCases := []struct {
+		input  userPrivileges
+		output userPrivileges
+	}{
+		{
+			userPrivileges{
+				fooUser: privilege.List{
+					privilege.ALL,
+					privilege.CONNECT,
+					privilege.CREATE,
+					privilege.DROP,
+					privilege.GRANT,
+					privilege.SELECT,
+					privilege.INSERT,
+					privilege.DELETE,
+					privilege.UPDATE,
+					privilege.USAGE,
+					privilege.ZONECONFIG,
+				},
+				barUser: privilege.List{
+					privilege.CONNECT,
+					privilege.CREATE,
+					privilege.DROP,
+					privilege.GRANT,
+					privilege.SELECT,
+					privilege.INSERT,
+					privilege.DELETE,
+					privilege.UPDATE,
+					privilege.USAGE,
+					privilege.ZONECONFIG,
+				},
+			},
+			userPrivileges{
+				fooUser: privilege.List{privilege.ALL},
+				barUser: privilege.List{
+					privilege.GRANT,
+					privilege.CREATE,
+					privilege.USAGE,
+				},
+			},
+		},
+		{
+			userPrivileges{
+				fooUser: privilege.List{privilege.GRANT},
+			},
+			userPrivileges{
+				fooUser: privilege.List{privilege.GRANT},
+			},
+		},
+		{
+			userPrivileges{
+				fooUser: privilege.List{privilege.CREATE},
+			},
+			userPrivileges{
+				fooUser: privilege.List{privilege.CREATE},
+			},
+		},
+		{
+			userPrivileges{
+				fooUser: privilege.List{privilege.USAGE},
+			},
+			userPrivileges{
+				fooUser: privilege.List{privilege.USAGE},
+			},
+		},
+	}
+
+	for num, tc := range testCases {
+		desc := &PrivilegeDescriptor{}
+		for u, p := range tc.input {
+			desc.Grant(u, p)
+		}
+		MaybeFixSchemaPrivileges(&desc)
+
+		for u, p := range tc.output {
+			outputUser, ok := desc.findUser(u)
+			if !ok {
+				t.Errorf("#%d: expected user %s in output, but not found (%v)",
+					num, u, desc.Users,
+				)
+			}
+			if a, e := privilege.ListFromBitField(outputUser.Privileges, privilege.Any), p; a.ToBitField() != e.ToBitField() {
+				t.Errorf("#%d: user %s: expected privileges %v, got %v",
+					num, u, e, a,
+				)
+			}
+
+			err := privilege.ValidatePrivileges(p, privilege.Schema)
+			if err != nil {
+				t.Errorf("%s\n", err.Error())
+			}
+		}
+
+	}
+}

pkg/sql/catalog/schemadesc/schema_desc_builder.go

@@ -51,6 +51,8 @@ func (sdb *schemaDescriptorBuilder) DescriptorType() catalog.DescriptorType {
 func (sdb *schemaDescriptorBuilder) RunPostDeserializationChanges(
 	_ context.Context, _ catalog.DescGetter,
 ) error {
+	privDesc := sdb.original.GetPrivileges()
+	descpb.MaybeFixSchemaPrivileges(&privDesc)
 	return nil
 }