diff --git a/redshift/helpers.go b/redshift/helpers.go
index f8df5dc..ffaf007 100644
--- a/redshift/helpers.go
+++ b/redshift/helpers.go
@@ -194,6 +194,8 @@ func validatePrivileges(privileges []string, objectType string) bool {
 			default:
 				return false
 			}
+		case "ROLE", "SYSTEM":
+			continue
 		default:
 			return false
 		}
diff --git a/redshift/provider.go b/redshift/provider.go
index b9a1c87..a3f7394 100644
--- a/redshift/provider.go
+++ b/redshift/provider.go
@@ -129,6 +129,7 @@ func Provider() *schema.Provider {
 		ResourcesMap: map[string]*schema.Resource{
 			"redshift_user":                redshiftUser(),
 			"redshift_group":               redshiftGroup(),
+			"redshift_role":                redshiftRole(),
 			"redshift_schema":              redshiftSchema(),
 			"redshift_default_privileges":  redshiftDefaultPrivileges(),
 			"redshift_grant":               redshiftGrant(),
diff --git a/redshift/resource_redshift_grant.go b/redshift/resource_redshift_grant.go
index 9179338..e1f9cf5 100644
--- a/redshift/resource_redshift_grant.go
+++ b/redshift/resource_redshift_grant.go
@@ -15,6 +15,7 @@ import (
 const (
 	grantUserAttr       = "user"
 	grantGroupAttr      = "group"
+	grantRoleAttr       = "role"
 	grantSchemaAttr     = "schema"
 	grantObjectTypeAttr = "object_type"
 	grantObjectsAttr    = "objects"
@@ -30,6 +31,8 @@ var grantAllowedObjectTypes = []string{
 	"function",
 	"procedure",
 	"language",
+	"role",
+	"system",
 }
 
 var grantObjectTypesCodes = map[string][]string{
@@ -41,7 +44,7 @@ var grantObjectTypesCodes = map[string][]string{
 func redshiftGrant() *schema.Resource {
 	return &schema.Resource{
 		Description: `
-Defines access privileges for users and  groups. Privileges include access options such as being able to read data in tables and views, write data, create tables, and drop tables. Use this command to give specific privileges for a table, database, schema, function, procedure, language, or column.
+		Defines access permissions for a user or user group. Permissions include access options such as being able to read data in tables and views, write data, create tables, and drop tables. Use this command to give specific permissions for a table, database, schema, function, procedure, language, or column.
 `,
 		Read: RedshiftResourceFunc(resourceRedshiftGrantRead),
 		Create: RedshiftResourceFunc(
@@ -61,16 +64,16 @@ Defines access privileges for users and  groups. Privileges include access optio
 				Type:         schema.TypeString,
 				Optional:     true,
 				ForceNew:     true,
-				ExactlyOneOf: []string{grantUserAttr, grantGroupAttr},
-				Description:  "The name of the user to grant privileges on. Either `user` or `group` parameter must be set.",
+				ExactlyOneOf: []string{grantUserAttr, grantGroupAttr, grantRoleAttr},
+				Description:  "The name of the user to grant privileges on. Either `user` or `group` or `role` parameter must be set.",
 				ValidateFunc: validation.StringDoesNotMatch(regexp.MustCompile("^(?i)public$"), "User name cannot be 'public'. To use GRANT ... TO PUBLIC set the group name to 'public' instead."),
 			},
 			grantGroupAttr: {
 				Type:         schema.TypeString,
 				Optional:     true,
 				ForceNew:     true,
-				ExactlyOneOf: []string{grantUserAttr, grantGroupAttr},
-				Description:  "The name of the group to grant privileges on. Either `group` or `user` parameter must be set. Settings the group name to `public` or `PUBLIC` (it is case insensitive in this case) will result in a `GRANT ... TO PUBLIC` statement.",
+				ExactlyOneOf: []string{grantUserAttr, grantGroupAttr, grantRoleAttr},
+				Description:  "The name of the group to grant privileges on. Either `user` or `group` or `role` parameter must be set. Settings the group name to `public` or `PUBLIC` (it is case insensitive in this case) will result in a `GRANT ... TO PUBLIC` statement.",
 				StateFunc: func(val interface{}) string {
 					name := val.(string)
 					if strings.ToLower(name) == grantToPublicName {
@@ -79,6 +82,13 @@ Defines access privileges for users and  groups. Privileges include access optio
 					return name
 				},
 			},
+			grantRoleAttr: {
+				Type:         schema.TypeString,
+				Optional:     true,
+				ForceNew:     true,
+				ExactlyOneOf: []string{grantUserAttr, grantGroupAttr, grantRoleAttr},
+				Description:  "The name of the role to grant privileges on. Either `user` or `group` or `role` parameter must be set.",
+			},
 			grantSchemaAttr: {
 				Type:        schema.TypeString,
 				Optional:    true,
@@ -143,6 +153,46 @@ func resourceRedshiftGrantCreate(db *DBConnection, d *schema.ResourceData) error
 		return fmt.Errorf("parameter `%s` is required for objects of type language", grantObjectsAttr)
 	}
 
+	if objectType == "role" {
+		if len(objects) > 0 {
+			return fmt.Errorf("parameter `%s` is not allowed for type role", grantObjectsAttr)
+		}
+
+		_, isSchema := d.GetOk(grantSchemaAttr)
+		if isSchema {
+			return fmt.Errorf("parameter `%s` is not allowed. Instead use `%s` or `%s` for type role", grantSchemaAttr, grantRoleAttr, grantUserAttr)
+		}
+
+		if len(privileges) > 1 {
+			return fmt.Errorf("parameter `%s` must be a single privilege for type role", grantPrivilegesAttr)
+		}
+	}
+
+	if objectType == "system" {
+		_, isSchema := d.GetOk(grantSchemaAttr)
+		_, isUser := d.GetOk(grantUserAttr)
+
+		if isSchema || isUser {
+			return fmt.Errorf("parameter `%s` or `%s` are not allowed. Instead use `%s` for type system", grantSchemaAttr, grantUserAttr, grantRoleAttr)
+		}
+
+		if len(privileges) > 1 {
+			contains := func(s []string, str string) bool {
+				for _, v := range s {
+					if v == str {
+						return true
+					}
+				}
+
+				return false
+			}
+
+			if contains(privileges, "all privileges") {
+				return fmt.Errorf("parameter `%s` specifying `all privileges` is not supported", grantPrivilegesAttr)
+			}
+		}
+	}
+
 	if !validatePrivileges(privileges, objectType) {
 		return fmt.Errorf("Invalid privileges list %v for object of type %s", privileges, objectType)
 	}
@@ -206,16 +256,108 @@ func resourceRedshiftGrantReadImpl(db *DBConnection, d *schema.ResourceData) err
 		return readCallableGrants(db, d)
 	case "language":
 		return readLanguageGrants(db, d)
+	case "role":
+		return readRoleGrants(db, d)
+	case "system":
+		return readSystemGrants(db, d)
 	default:
 		return fmt.Errorf("Unsupported %s %s", grantObjectTypeAttr, objectType)
 	}
 }
 
+func readSystemGrants(db *DBConnection, d *schema.ResourceData) error {
+	roleName := d.Get(grantRoleAttr).(string)
+
+	queryArgs := []interface{}{
+		roleName,
+	}
+
+	query := `
+	SELECT lower(system_privilege) FROM svv_system_privileges WHERE identity_name = $1 and identity_type = 'role'
+	`
+
+	rows, err := db.Query(query, queryArgs...)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+
+	privilegesSet := schema.NewSet(schema.HashString, nil)
+	for rows.Next() {
+		var system_privilege string
+
+		if err := rows.Scan(&system_privilege); err != nil {
+			return err
+		}
+
+		// filter out buggy privilege name redshift support indicated this is a bug
+		if system_privilege == "unkown" {
+			continue
+		}
+
+		privilegesSet.Add(system_privilege)
+	}
+
+	if err := d.Set(grantPrivilegesAttr, privilegesSet); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func readRoleGrants(db *DBConnection, d *schema.ResourceData) error {
+	var queryArgs []interface{}
+	var query, granted string
+
+	_, isRole := d.GetOk(grantRoleAttr)
+
+	privileges := []string{}
+	for _, p := range d.Get(grantPrivilegesAttr).(*schema.Set).List() {
+		privileges = append(privileges, p.(string))
+	}
+
+	if isRole {
+		roleName := d.Get(grantRoleAttr).(string)
+
+		queryArgs = []interface{}{
+			roleName, privileges[0],
+		}
+
+		query = `
+		SELECT granted_role_name FROM svv_role_grants WHERE role_name = $1 AND granted_role_name = $2
+		`
+	} else {
+		userName := d.Get(grantUserAttr).(string)
+
+		queryArgs = []interface{}{
+			userName, privileges[0],
+		}
+
+		query = `
+		SELECT role_name FROM svv_user_grants WHERE user_name = $1 AND role_name = $2
+		`
+	}
+
+	if err := db.QueryRow(query, queryArgs...).Scan(&granted); err != nil {
+		return err
+	}
+
+	privilegesSet := schema.NewSet(schema.HashString, nil)
+	privilegesSet.Add(granted)
+
+	if err := d.Set(grantPrivilegesAttr, privilegesSet); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func readDatabaseGrants(db *DBConnection, d *schema.ResourceData) error {
 	var entityName, query string
 	var databaseCreate, databaseTemp bool
 
 	_, isUser := d.GetOk(grantUserAttr)
+	_, isRole := d.GetOk(grantRoleAttr)
 
 	if isUser {
 		entityName = d.Get(grantUserAttr).(string)
@@ -227,6 +369,15 @@ func readDatabaseGrants(db *DBConnection, d *schema.ResourceData) error {
   WHERE
     db.datname=$1 
     AND u.usename=$2
+`
+	} else if isRole {
+		entityName = d.Get(grantRoleAttr).(string)
+		query = `
+	SELECT
+		SUM(CASE WHEN privilege_type = 'CREATE' THEN 1 ELSE 0 END) AS "create",
+		SUM(CASE WHEN privilege_type = 'TEMP' THEN 1 ELSE 0 END) AS "temporary"
+	FROM SVV_DATABASE_PRIVILEGES WHERE database_name = $1 AND identity_name = $2 AND identity_type = 'role'
+	GROUP BY identity_name
 `
 	} else {
 		entityName = d.Get(grantGroupAttr).(string)
@@ -276,6 +427,7 @@ func readSchemaGrants(db *DBConnection, d *schema.ResourceData) error {
 	var schemaCreate, schemaUsage bool
 
 	_, isUser := d.GetOk(grantUserAttr)
+	_, isRole := d.GetOk(grantRoleAttr)
 	schemaName := d.Get(grantSchemaAttr).(string)
 
 	if isUser {
@@ -289,6 +441,15 @@ func readSchemaGrants(db *DBConnection, d *schema.ResourceData) error {
 		ns.nspname=$1 
 		AND u.usename=$2
 	`
+	} else if isRole {
+		entityName = d.Get(grantRoleAttr).(string)
+		query = `
+	SELECT
+		SUM(CASE WHEN privilege_type = 'CREATE' THEN 1 ELSE 0 END) AS create,
+		SUM(CASE WHEN privilege_type = 'USAGE' THEN 1 ELSE 0 END) AS usage
+	FROM SVV_SCHEMA_PRIVILEGES WHERE namespace_name = $1 AND identity_name = $2 AND identity_type = 'role'
+	GROUP BY identity_name		
+`
 	} else {
 		entityName = d.Get(grantGroupAttr).(string)
 		query = `
@@ -467,8 +628,10 @@ func readCallableGrants(db *DBConnection, d *schema.ResourceData) error {
 	log.Printf("[DEBUG] Reading callable grants")
 
 	var entityName, query string
+	var queryArgs []interface{}
 
 	_, isUser := d.GetOk(grantUserAttr)
+	_, isRole := d.GetOk(grantRoleAttr)
 	schemaName := d.Get(grantSchemaAttr).(string)
 	objectType := d.Get(grantObjectTypeAttr).(string)
 
@@ -486,6 +649,23 @@ func readCallableGrants(db *DBConnection, d *schema.ResourceData) error {
 		AND u.usename=$2
 		AND pr.prokind=ANY($3)
 `
+		queryArgs = []interface{}{
+			schemaName, entityName, pq.Array(grantObjectTypesCodes[objectType]),
+		}
+
+	} else if isRole {
+		entityName = d.Get(grantRoleAttr).(string)
+		query = `
+		SELECT
+			function_name,
+			SUM(CASE WHEN privilege_type = 'EXECUTE' THEN 1 ELSE 0 END) AS execute
+		FROM svv_function_privileges WHERE namespace_name = $1 AND identity_name = $2 AND identity_type = 'role'
+		GROUP BY function_name, argument_types, identity_name
+	`
+		queryArgs = []interface{}{
+			schemaName, entityName,
+		}
+
 	} else {
 		entityName = d.Get(grantGroupAttr).(string)
 		query = `
@@ -500,12 +680,13 @@ func readCallableGrants(db *DBConnection, d *schema.ResourceData) error {
     AND gr.groname=$2
 		AND pr.prokind=ANY($3)
 `
+		queryArgs = []interface{}{
+			schemaName, entityName, pq.Array(grantObjectTypesCodes[objectType]),
+		}
 	}
 
-	callables := stripArgumentsFromCallablesDefinitions(d.Get(grantObjectsAttr).(*schema.Set))
-	queryArgs := []interface{}{
-		schemaName, entityName, pq.Array(grantObjectTypesCodes[objectType]),
-	}
+	objs := d.Get(grantObjectsAttr).(*schema.Set)
+	callables := stripArgumentsFromCallablesDefinitions(objs)
 
 	if isGrantToPublic(d) {
 		query = `
@@ -663,6 +844,9 @@ func createGrantsRevokeQuery(d *schema.ResourceData, databaseName string) string
 	if groupName, isGroup := d.GetOk(grantGroupAttr); isGroup {
 		toWhomIndicator = "GROUP"
 		entityName = groupName.(string)
+	} else if roleName, isRole := d.GetOk(grantRoleAttr); isRole {
+		toWhomIndicator = "ROLE"
+		entityName = roleName.(string)
 	} else if userName, isUser := d.GetOk(grantUserAttr); isUser {
 		entityName = userName.(string)
 	}
@@ -734,6 +918,23 @@ func createGrantsRevokeQuery(d *schema.ResourceData, databaseName string) string
 			toWhomIndicator,
 			fromEntityName,
 		)
+	case "ROLE":
+		privileges := []string{}
+		for _, p := range d.Get(grantPrivilegesAttr).(*schema.Set).List() {
+			privileges = append(privileges, p.(string))
+		}
+
+		query = fmt.Sprintf(
+			"REVOKE ROLE %s FROM %s %s",
+			privileges[0],
+			toWhomIndicator,
+			fromEntityName,
+		)
+	case "SYSTEM":
+		query = fmt.Sprintf(
+			"REVOKE ALL PRIVILEGES FROM ROLE %s",
+			fromEntityName,
+		)
 	}
 	log.Printf("[DEBUG] Created REVOKE query: %s", query)
 	return query
@@ -749,6 +950,9 @@ func createGrantsQuery(d *schema.ResourceData, databaseName string) string {
 	if groupName, isGroup := d.GetOk(grantGroupAttr); isGroup {
 		toWhomIndicator = "GROUP"
 		entityName = groupName.(string)
+	} else if roleName, isRole := d.GetOk(grantRoleAttr); isRole {
+		toWhomIndicator = "ROLE"
+		entityName = roleName.(string)
 	} else if userName, isUser := d.GetOk(grantUserAttr); isUser {
 		entityName = userName.(string)
 	}
@@ -818,6 +1022,19 @@ func createGrantsQuery(d *schema.ResourceData, databaseName string) string {
 				toEntityName,
 			)
 		}
+	case "ROLE":
+		query = fmt.Sprintf(
+			"GRANT ROLE %s TO %s %s",
+			privileges[0],
+			toWhomIndicator,
+			toEntityName,
+		)
+	case "SYSTEM":
+		query = fmt.Sprintf(
+			"GRANT %s TO ROLE %s",
+			strings.Join(privileges, ","),
+			toEntityName,
+		)
 	}
 
 	log.Printf("[DEBUG] Created GRANT query: %s", query)
@@ -846,6 +1063,10 @@ func generateGrantID(d *schema.ResourceData) string {
 		parts = append(parts, fmt.Sprintf("gn:%s", name))
 	}
 
+	if _, isRole := d.GetOk(grantRoleAttr); isRole {
+		parts = append(parts, fmt.Sprintf("rn:%s", d.Get(grantRoleAttr).(string)))
+	}
+
 	if _, isUser := d.GetOk(grantUserAttr); isUser {
 		parts = append(parts, fmt.Sprintf("un:%s", d.Get(grantUserAttr).(string)))
 	}
@@ -853,10 +1074,19 @@ func generateGrantID(d *schema.ResourceData) string {
 	objectType := fmt.Sprintf("ot:%s", d.Get(grantObjectTypeAttr).(string))
 	parts = append(parts, objectType)
 
-	if objectType != "ot:database" && objectType != "ot:language" {
+	if objectType != "ot:database" && objectType != "ot:language" && objectType != "ot:role" && objectType != "ot:system" {
 		parts = append(parts, d.Get(grantSchemaAttr).(string))
 	}
 
+	if objectType == "ot:role" || objectType == "ot:system" {
+		privileges := []string{}
+		for _, p := range d.Get(grantPrivilegesAttr).(*schema.Set).List() {
+			privileges = append(privileges, p.(string))
+		}
+
+		parts = append(parts, fmt.Sprintf("pv:%s", strings.Join(privileges, ",")))
+	}
+
 	for _, object := range d.Get(grantObjectsAttr).(*schema.Set).List() {
 		parts = append(parts, object.(string))
 	}
diff --git a/redshift/resource_redshift_grant_test.go b/redshift/resource_redshift_grant_test.go
index 09f5846..d17c446 100644
--- a/redshift/resource_redshift_grant_test.go
+++ b/redshift/resource_redshift_grant_test.go
@@ -164,9 +164,15 @@ func TestAccRedshiftGrant_BasicDatabase(t *testing.T) {
 		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_user"), "-", "_"),
 		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_user@tf_acc_domain.tld"), "-", "_"),
 	}
+	roleNames := []string{
+		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_role"), "-", "_"),
+		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_role@tf_acc_domain.tld"), "-", "_"),
+	}
 
 	for i, groupName := range groupNames {
 		userName := userNames[i]
+		roleName := roleNames[i]
+
 		config := fmt.Sprintf(`
 		resource "redshift_group" "group" {
 		  name = %[1]q
@@ -176,6 +182,10 @@ func TestAccRedshiftGrant_BasicDatabase(t *testing.T) {
 		  name = %[2]q
 		  password = "TestPassword123"
 		}
+
+		resource "redshift_role" "role" {
+			name = %[3]q
+		}
 		
 		resource "redshift_grant" "grant" {
 		  group = redshift_group.group.name
@@ -188,7 +198,14 @@ func TestAccRedshiftGrant_BasicDatabase(t *testing.T) {
 		  object_type = "database"
 		  privileges = ["temporary"]
 		}
-		`, groupName, userName)
+
+		resource "redshift_grant" "grant_role" {
+			role = redshift_role.role.name
+			object_type = "database"
+			privileges = ["create", "temporary"]
+		  }		
+		`, groupName, userName, roleName)
+
 		resource.Test(t, resource.TestCase{
 			PreCheck:     func() { testAccPreCheck(t) },
 			Providers:    testAccProviders,
@@ -209,6 +226,13 @@ func TestAccRedshiftGrant_BasicDatabase(t *testing.T) {
 						resource.TestCheckResourceAttr("redshift_grant.grant_user", "object_type", "database"),
 						resource.TestCheckResourceAttr("redshift_grant.grant_user", "privileges.#", "1"),
 						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_user", "privileges.*", "temporary"),
+
+						resource.TestCheckResourceAttr("redshift_grant.grant_role", "id", fmt.Sprintf("rn:%s_ot:database", roleName)),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role", "role", roleName),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role", "object_type", "database"),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role", "privileges.#", "2"),
+						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_role", "privileges.*", "create"),
+						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_role", "privileges.*", "temporary"),
 					),
 				},
 			},
@@ -216,6 +240,133 @@ func TestAccRedshiftGrant_BasicDatabase(t *testing.T) {
 	}
 }
 
+func TestAccRedshiftGrant_BasicRole(t *testing.T) {
+	roleName := strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_role"), "-", "_")
+	anotherRoleName := strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_another_role"), "-", "_")
+	userName := strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_user"), "-", "_")
+
+	config := fmt.Sprintf(`
+	resource "redshift_role" "role" {
+		name = %[1]q
+	}
+
+	resource "redshift_role" "another_role" {
+		name = %[2]q
+	}
+
+	resource "redshift_user" "user" {
+		name = %[3]q
+		password = "TestPassword123"
+	}
+
+	// grant role sys:secadmin to role some_role
+	resource "redshift_grant" "sys_role_to_role" {
+	  role       = redshift_role.role.name
+	  object_type = "role"
+	  privileges  = ["sys:secadmin"]
+	}
+	
+	// grant role sys:secadmin to some_user
+	resource "redshift_grant" "sys_role_to_user" {
+	  user = redshift_user.user.name
+	  object_type = "role"
+	  privileges  = ["sys:secadmin"]
+	}
+	
+	// grant role some_role to user some_user
+	resource "redshift_grant" "role_to_user" {
+	  user = redshift_user.user.name
+	  object_type = "role"
+	  privileges  = [redshift_role.role.name]
+	}
+	
+	// grant role some_role to role another_role
+	resource "redshift_grant" "role_to_role" {
+	  role = redshift_role.another_role.name
+	  object_type = "role"
+	  privileges  = [redshift_role.role.name]
+	}	
+	`, roleName, anotherRoleName, userName)
+
+	resource.Test(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: func(s *terraform.State) error { return nil },
+		Steps: []resource.TestStep{
+			{
+				Config: config,
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("redshift_grant.sys_role_to_role", "id", fmt.Sprintf("rn:%s_ot:role_pv:sys:secadmin", roleName)),
+					resource.TestCheckResourceAttr("redshift_grant.sys_role_to_role", "role", roleName),
+					resource.TestCheckResourceAttr("redshift_grant.sys_role_to_role", "object_type", "role"),
+					resource.TestCheckResourceAttr("redshift_grant.sys_role_to_role", "privileges.#", "1"),
+					resource.TestCheckTypeSetElemAttr("redshift_grant.sys_role_to_role", "privileges.*", "sys:secadmin"),
+
+					resource.TestCheckResourceAttr("redshift_grant.sys_role_to_user", "id", fmt.Sprintf("un:%s_ot:role_pv:sys:secadmin", userName)),
+					resource.TestCheckResourceAttr("redshift_grant.sys_role_to_user", "user", userName),
+					resource.TestCheckResourceAttr("redshift_grant.sys_role_to_user", "object_type", "role"),
+					resource.TestCheckResourceAttr("redshift_grant.sys_role_to_user", "privileges.#", "1"),
+					resource.TestCheckTypeSetElemAttr("redshift_grant.sys_role_to_user", "privileges.*", "sys:secadmin"),
+
+					resource.TestCheckResourceAttr("redshift_grant.role_to_role", "id", fmt.Sprintf("rn:%s_ot:role_pv:%s", anotherRoleName, roleName)),
+					resource.TestCheckResourceAttr("redshift_grant.role_to_role", "role", anotherRoleName),
+					resource.TestCheckResourceAttr("redshift_grant.role_to_role", "object_type", "role"),
+					resource.TestCheckResourceAttr("redshift_grant.role_to_role", "privileges.#", "1"),
+					resource.TestCheckTypeSetElemAttr("redshift_grant.role_to_role", "privileges.*", roleName),
+
+					resource.TestCheckResourceAttr("redshift_grant.role_to_user", "id", fmt.Sprintf("un:%s_ot:role_pv:%s", userName, roleName)),
+					resource.TestCheckResourceAttr("redshift_grant.role_to_user", "user", userName),
+					resource.TestCheckResourceAttr("redshift_grant.role_to_user", "object_type", "role"),
+					resource.TestCheckResourceAttr("redshift_grant.role_to_user", "privileges.#", "1"),
+					resource.TestCheckTypeSetElemAttr("redshift_grant.role_to_user", "privileges.*", roleName),
+				),
+			},
+		},
+	})
+}
+
+func TestAccRedshiftGrant_BasicSystem(t *testing.T) {
+	roleName := strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_role"), "-", "_")
+	anotherRoleName := strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_role"), "-", "_")
+
+	config := fmt.Sprintf(`
+	// grant role sys:secadmin to role some_role
+	resource "redshift_role" "role" {
+		name = %[1]q
+	}
+
+	resource "redshift_role" "another_role" {
+		name = %[2]q
+	}
+
+	// grant alter default privileges, drop table to role some_role
+	resource "redshift_grant" "some_privs_to_role" {
+	  role       = redshift_role.role.name
+	  object_type = "system"
+	  privileges  = ["alter default privileges", "drop table"]
+	}
+	`, roleName, anotherRoleName)
+
+	resource.Test(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: func(s *terraform.State) error { return nil },
+		Steps: []resource.TestStep{
+			{
+				Config: config,
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("redshift_grant.some_privs_to_role", "id", fmt.Sprintf("rn:%s_ot:system_pv:alter default privileges,drop table", roleName)),
+					resource.TestCheckResourceAttr("redshift_grant.some_privs_to_role", "role", roleName),
+					resource.TestCheckResourceAttr("redshift_grant.some_privs_to_role", "object_type", "system"),
+					resource.TestCheckResourceAttr("redshift_grant.some_privs_to_role", "privileges.#", "2"),
+					resource.TestCheckTypeSetElemAttr("redshift_grant.some_privs_to_role", "privileges.*", "alter default privileges"),
+					resource.TestCheckTypeSetElemAttr("redshift_grant.some_privs_to_role", "privileges.*", "drop table"),
+				),
+			},
+		},
+	})
+}
+
 func TestAccRedshiftGrant_BasicSchema(t *testing.T) {
 	groupNames := []string{
 		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_group"), "-", "_"),
@@ -225,10 +376,15 @@ func TestAccRedshiftGrant_BasicSchema(t *testing.T) {
 		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_user"), "-", "_"),
 		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_user@tf_acc_domain.tld"), "-", "_"),
 	}
+	roleNames := []string{
+		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_role"), "-", "_"),
+		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_role@tf_acc_domain.tld"), "-", "_"),
+	}
 	schemaName := strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_schema_basic"), "-", "_")
 
 	for i, groupName := range groupNames {
 		userName := userNames[i]
+		roleName := roleNames[i]
 		config := fmt.Sprintf(`
 		resource "redshift_user" "user" {
 		  name = %[1]q
@@ -260,6 +416,33 @@ func TestAccRedshiftGrant_BasicSchema(t *testing.T) {
 		  privileges = ["create", "usage"]
 		}
 		`, userName, groupName, schemaName)
+
+		// combining the user/group tests with role resulted in the below error; instead break out the roles separately
+		// Error: pq: duplicate key violates unique constraint "pg_permission_perm_index" (possibly caused by concurrent transaction conflict)
+		configRole := fmt.Sprintf(`
+		resource "redshift_user" "user" {
+			name = %[1]q
+		}
+
+		resource "redshift_schema" "schema" {
+		  name = %[2]q
+		
+		  owner = redshift_user.user.name
+		}
+		
+		resource "redshift_role" "role" {
+			name = %[3]q
+		}
+
+		resource "redshift_grant" "grant_role" {
+		  role = redshift_role.role.name
+		  schema = redshift_schema.schema.name
+		  
+		  object_type = "schema"
+		  privileges = ["create", "usage"]
+		}
+		`, userName, schemaName, roleName)
+
 		resource.Test(t, resource.TestCase{
 			PreCheck:     func() { testAccPreCheck(t) },
 			Providers:    testAccProviders,
@@ -283,9 +466,21 @@ func TestAccRedshiftGrant_BasicSchema(t *testing.T) {
 						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_user", "privileges.*", "usage"),
 					),
 				},
+				{
+					Config: configRole,
+					Check: resource.ComposeAggregateTestCheckFunc(
+						resource.TestCheckResourceAttr("redshift_grant.grant_role", "id", fmt.Sprintf("rn:%s_ot:schema_%s", roleName, schemaName)),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role", "role", roleName),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role", "object_type", "schema"),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role", "privileges.#", "2"),
+						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_role", "privileges.*", "create"),
+						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_role", "privileges.*", "usage"),
+					),
+				},
 			},
 		})
 	}
+
 }
 
 func TestAccRedshiftGrant_BasicTable(t *testing.T) {
@@ -297,9 +492,14 @@ func TestAccRedshiftGrant_BasicTable(t *testing.T) {
 		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_user"), "-", "_"),
 		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_user@tf_acc_domain.tld"), "-", "_"),
 	}
+	roleNames := []string{
+		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_role"), "-", "_"),
+		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_role@tf_acc_domain.tld"), "-", "_"),
+	}
 
 	for i, groupName := range groupNames {
 		userName := userNames[i]
+		roleName := roleNames[i]
 		config := fmt.Sprintf(`
 		resource "redshift_group" "group" {
 		  name = %[1]q
@@ -328,6 +528,23 @@ func TestAccRedshiftGrant_BasicTable(t *testing.T) {
 		  privileges = ["select", "update", "insert", "delete", "drop", "references", "rule", "trigger"]
 		}
 		`, groupName, userName)
+
+		// combining the user/group tests with role resulted in the below error; instead break out the roles separately
+		// Error: pq: duplicate key violates unique constraint "pg_permission_perm_index" (possibly caused by concurrent transaction conflict)
+		configRole := fmt.Sprintf(`
+		resource "redshift_role" "role" {
+			name = %[1]q
+		}
+
+		resource "redshift_grant" "grant_role" {
+		  role = redshift_role.role.name
+		  schema = "pg_catalog"
+		  
+		  object_type = "table"
+		  objects = ["pg_user_info"]
+		  privileges = ["select", "update", "insert", "delete", "drop", "references", "rule", "trigger"]
+		}
+		`, roleName)
 		resource.Test(t, resource.TestCase{
 			PreCheck:     func() { testAccPreCheck(t) },
 			Providers:    testAccProviders,
@@ -369,6 +586,26 @@ func TestAccRedshiftGrant_BasicTable(t *testing.T) {
 						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_user", "privileges.*", "trigger"),
 					),
 				},
+				{
+					Config: configRole,
+					Check: resource.ComposeTestCheckFunc(
+						resource.TestCheckResourceAttr("redshift_grant.grant_role", "id", fmt.Sprintf("rn:%s_ot:table_pg_catalog_pg_user_info", roleName)),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role", "role", roleName),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role", "schema", "pg_catalog"),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role", "object_type", "table"),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role", "objects.#", "1"),
+						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_role", "objects.*", "pg_user_info"),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role", "privileges.#", "8"),
+						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_role", "privileges.*", "select"),
+						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_role", "privileges.*", "update"),
+						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_role", "privileges.*", "insert"),
+						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_role", "privileges.*", "delete"),
+						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_role", "privileges.*", "drop"),
+						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_role", "privileges.*", "references"),
+						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_role", "privileges.*", "rule"),
+						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_role", "privileges.*", "trigger"),
+					),
+				},
 			},
 		})
 	}
@@ -383,17 +620,22 @@ func TestAccRedshiftGrant_BasicCallables(t *testing.T) {
 		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_user"), "-", "_"),
 		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_user@tf_acc_domain.tld"), "-", "_"),
 	}
+	roleNames := []string{
+		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_role"), "-", "_"),
+		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_role@tf_acc_domain.tld"), "-", "_"),
+	}
 	schema := strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_schema_basic"), "-", "_")
 
 	for i, groupName := range groupNames {
 		userName := userNames[i]
+		roleName := roleNames[i]
 		resource.Test(t, resource.TestCase{
 			PreCheck:     func() { testAccPreCheck(t) },
 			Providers:    testAccProviders,
 			CheckDestroy: func(s *terraform.State) error { return nil },
 			Steps: []resource.TestStep{
 				{
-					Config: testAccRedshiftGrant_basicCallables_configUserGroup(userName, groupName, schema),
+					Config: testAccRedshiftGrant_basicCallables_configUserGroup(userName, groupName, roleName),
 				},
 				{
 					PreConfig: func() {
@@ -433,10 +675,29 @@ func TestAccRedshiftGrant_BasicCallables(t *testing.T) {
 						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_user_proc", "privileges.*", "execute"),
 					),
 				},
+				{
+					Config: testAccRedshiftGrant_basicCallables_configRoleWithGrants(roleName, schema),
+					Check: resource.ComposeTestCheckFunc(
+						resource.TestCheckResourceAttr("redshift_grant.grant_role_func", "id", fmt.Sprintf("rn:%s_ot:function_%s_test_call(int,int)_test_call(float,float)", roleName, schema)),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role_func", "role", roleName),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role_func", "object_type", "function"),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role_func", "privileges.#", "1"),
+						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_role_func", "privileges.*", "execute"),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role_proc", "id", fmt.Sprintf("rn:%s_ot:procedure_%s_test_call()", roleName, schema)),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role_proc", "role", roleName),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role_proc", "object_type", "procedure"),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role_proc", "privileges.#", "1"),
+						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_role_proc", "privileges.*", "execute"),
+					),
+				},
 				{
 					Config:  testAccRedshiftGrant_basicCallables_configUserGroupWithGrants(userName, groupName, schema),
 					Destroy: true,
 				},
+				{
+					Config:  testAccRedshiftGrant_basicCallables_configRoleWithGrants(roleName, schema),
+					Destroy: true,
+				},
 				// Creating additional dummy step as TestStep does not have PostConfig
 				// property, so clean up cannot be performed in the previous one.
 				{
@@ -470,11 +731,16 @@ func TestAccRedshiftGrant_BasicLanguage(t *testing.T) {
 		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_user"), "-", "_"),
 		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_user@tf_acc_domain.tld"), "-", "_"),
 	}
+	roleNames := []string{
+		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_role"), "-", "_"),
+		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_role@tf_acc_domain.tld"), "-", "_"),
+	}
 	addedLanguage := "plpythonu"
 	secondLanguage := "plpgsql"
 
 	for i, groupName := range groupNames {
 		userName := userNames[i]
+		roleName := roleNames[i]
 		config := fmt.Sprintf(`
 		resource "redshift_user" "user" {
 		  name = %[1]q
@@ -483,6 +749,10 @@ func TestAccRedshiftGrant_BasicLanguage(t *testing.T) {
 		resource "redshift_group" "group" {
 		  name = %[2]q
 		}
+
+		resource "redshift_role" "role" {
+			name = %[5]q
+		}
 		
 		resource "redshift_grant" "grant" {
 		  group  = redshift_group.group.name
@@ -499,7 +769,15 @@ func TestAccRedshiftGrant_BasicLanguage(t *testing.T) {
 		  object_type = "language"
 		  privileges = ["usage"]
 		}
-		`, userName, groupName, addedLanguage, secondLanguage)
+
+		resource "redshift_grant" "grant_role" {
+			role  = redshift_role.role.name
+			objects = [%[3]q, %[4]q]
+		  
+			object_type = "language"
+			privileges = ["usage"]
+		  }		
+		`, userName, groupName, addedLanguage, secondLanguage, roleName)
 		resource.Test(t, resource.TestCase{
 			PreCheck:     func() { testAccPreCheck(t) },
 			Providers:    testAccProviders,
@@ -519,6 +797,12 @@ func TestAccRedshiftGrant_BasicLanguage(t *testing.T) {
 						resource.TestCheckResourceAttr("redshift_grant.grant_user", "object_type", "language"),
 						resource.TestCheckResourceAttr("redshift_grant.grant_user", "privileges.#", "1"),
 						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_user", "privileges.*", "usage"),
+
+						resource.TestCheckResourceAttr("redshift_grant.grant_role", "id", fmt.Sprintf("rn:%s_ot:language_%s_%s", roleName, addedLanguage, secondLanguage)),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role", "role", roleName),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role", "object_type", "language"),
+						resource.TestCheckResourceAttr("redshift_grant.grant_role", "privileges.#", "1"),
+						resource.TestCheckTypeSetElemAttr("redshift_grant.grant_role", "privileges.*", "usage"),
 					),
 				},
 			},
@@ -657,7 +941,7 @@ func testAccRedshiftGrant_Regression_Issue_43_compare_ids(addr1 string, addr2 st
 	}
 }
 
-func testAccRedshiftGrant_basicCallables_configUserGroup(username, group, schema string) string {
+func testAccRedshiftGrant_basicCallables_configUserGroup(username, group, role string) string {
 	return fmt.Sprintf(`
 resource "redshift_user" "user" {
   name = %[1]q
@@ -666,7 +950,11 @@ resource "redshift_user" "user" {
 resource "redshift_group" "group" {
   name = %[2]q
 }
-`, username, group)
+
+resource "redshift_role" "role" {
+	name = %[3]q
+}
+`, username, group, role)
 }
 
 func testAccRedshiftGrant_basicCallables_configUserGroupWithGrants(username, group, schema string) string {
@@ -717,6 +1005,32 @@ resource "redshift_grant" "grant_user_proc" {
 `, username, group, schema)
 }
 
+func testAccRedshiftGrant_basicCallables_configRoleWithGrants(role, schema string) string {
+	return fmt.Sprintf(`
+resource "redshift_role" "role" {
+	name = %[1]q
+}
+
+resource "redshift_grant" "grant_role_func" {
+	schema = %[2]q
+	role  = redshift_role.role.name
+	objects = ["test_call(float,float)", "test_call(int,int)"]
+
+	object_type = "function"
+	privileges = ["execute"]
+}
+
+resource "redshift_grant" "grant_role_proc" {
+	schema = %[2]q
+	role  = redshift_role.role.name
+	objects = ["test_call()"]
+
+	object_type = "procedure"
+	privileges = ["execute"]
+}
+`, role, schema)
+}
+
 func testAccRedshiftGrant_basicCallables_createSchemaAndCallables(t *testing.T, db *DBConnection, schema string) error {
 	_, err := db.Exec(fmt.Sprintf("CREATE SCHEMA %s", pq.QuoteIdentifier(schema)))
 	if err != nil {
diff --git a/redshift/resource_redshift_role.go b/redshift/resource_redshift_role.go
new file mode 100644
index 0000000..32c11ac
--- /dev/null
+++ b/redshift/resource_redshift_role.go
@@ -0,0 +1,158 @@
+package redshift
+
+import (
+	"database/sql"
+	"fmt"
+
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/validation"
+	"github.com/lib/pq"
+)
+
+const (
+	roleNameAttr = "name"
+)
+
+func redshiftRole() *schema.Resource {
+	return &schema.Resource{
+		Description: `
+Roles are collections of permissions that you can assign to a user or another role. You can assign system or database permissions to a role. A user inherits permissions from an assigned role.
+`,
+		Create: RedshiftResourceFunc(resourceRedshiftRoleCreate),
+		Read:   RedshiftResourceFunc(resourceRedshiftRoleRead),
+		Update: RedshiftResourceFunc(resourceRedshiftRoleUpdate),
+		Delete: RedshiftResourceFunc(
+			RedshiftResourceRetryOnPQErrors(resourceRedshiftRoleDelete),
+		),
+		Exists: RedshiftResourceExistsFunc(resourceRedshiftRoleExists),
+		Importer: &schema.ResourceImporter{
+			State: schema.ImportStatePassthrough,
+		},
+
+		Schema: map[string]*schema.Schema{
+			roleNameAttr: {
+				Type:         schema.TypeString,
+				Required:     true,
+				Description:  "The name of the role (case sensitive). The role name must be unique and can't be the same as any user names. A role name can't be a reserved word.",
+				ValidateFunc: validation.StringNotInSlice(reservedWords, false),
+			},
+		},
+	}
+}
+
+func resourceRedshiftRoleExists(db *DBConnection, d *schema.ResourceData) (bool, error) {
+	var roleName string
+	err := db.QueryRow("SELECT role_name FROM svv_roles WHERE role_id = $1", d.Id()).Scan(&roleName)
+
+	switch {
+	case err == sql.ErrNoRows:
+		return false, nil
+	case err != nil:
+		return false, err
+	}
+
+	return true, nil
+}
+
+func resourceRedshiftRoleRead(db *DBConnection, d *schema.ResourceData) error {
+	return resourceRedshiftRoleReadImpl(db, d)
+}
+
+func resourceRedshiftRoleReadImpl(db *DBConnection, d *schema.ResourceData) error {
+	var (
+		roleName string
+	)
+
+	sql := `SELECT role_name from svv_roles WHERE role_id=$1`
+	if err := db.QueryRow(sql, d.Id()).Scan(&roleName); err != nil {
+		return err
+	}
+
+	d.Set(roleNameAttr, roleName)
+
+	return nil
+}
+
+func resourceRedshiftRoleCreate(db *DBConnection, d *schema.ResourceData) error {
+	roleName := d.Get(roleNameAttr).(string)
+
+	tx, err := startTransaction(db.client, "")
+	if err != nil {
+		return err
+	}
+	defer deferredRollback(tx)
+
+	create_role_sql := fmt.Sprintf("CREATE ROLE %s", pq.QuoteIdentifier(roleName))
+	if _, err := tx.Exec(create_role_sql); err != nil {
+		return fmt.Errorf("Could not create redshift role: %s", err)
+	}
+
+	var roleId string
+	find_role_id_sql := fmt.Sprintf("SELECT role_id FROM svv_roles WHERE role_name = %s", pq.QuoteLiteral(roleName))
+	if err := tx.QueryRow(find_role_id_sql).Scan(&roleId); err != nil {
+		return fmt.Errorf("Could not get redshift role id for '%s': %s", roleName, err)
+	}
+
+	d.SetId(roleId)
+
+	if err = tx.Commit(); err != nil {
+		return fmt.Errorf("could not commit transaction: %w", err)
+	}
+
+	return resourceRedshiftRoleReadImpl(db, d)
+}
+
+func resourceRedshiftRoleDelete(db *DBConnection, d *schema.ResourceData) error {
+	roleName := d.Get(roleNameAttr).(string)
+
+	tx, err := startTransaction(db.client, "")
+	if err != nil {
+		return err
+	}
+	defer deferredRollback(tx)
+
+	if _, err := tx.Exec(fmt.Sprintf("DROP ROLE %s", pq.QuoteIdentifier(roleName))); err != nil {
+		return err
+	}
+
+	return tx.Commit()
+}
+
+func resourceRedshiftRoleUpdate(db *DBConnection, d *schema.ResourceData) error {
+	tx, err := startTransaction(db.client, "")
+	if err != nil {
+		return err
+	}
+	defer deferredRollback(tx)
+
+	if err := setRoleName(tx, d); err != nil {
+		return err
+	}
+
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("could not commit transaction: %w", err)
+	}
+
+	return resourceRedshiftRoleReadImpl(db, d)
+}
+
+func setRoleName(tx *sql.Tx, d *schema.ResourceData) error {
+	if !d.HasChange(roleNameAttr) {
+		return nil
+	}
+
+	oldRaw, newRaw := d.GetChange(roleNameAttr)
+	oldValue := oldRaw.(string)
+	newValue := newRaw.(string)
+
+	if newValue == "" {
+		return fmt.Errorf("Error setting role name to an empty string")
+	}
+
+	sql := fmt.Sprintf("ALTER ROLE %s RENAME TO %s", pq.QuoteIdentifier(oldValue), pq.QuoteIdentifier(newValue))
+	if _, err := tx.Exec(sql); err != nil {
+		return fmt.Errorf("Error updating Role NAME: %w", err)
+	}
+
+	return nil
+}
diff --git a/redshift/resource_redshift_role_test.go b/redshift/resource_redshift_role_test.go
new file mode 100644
index 0000000..59356d5
--- /dev/null
+++ b/redshift/resource_redshift_role_test.go
@@ -0,0 +1,150 @@
+package redshift
+
+import (
+	"database/sql"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/acctest"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/terraform"
+)
+
+func TestAccRedshiftRole_Basic(t *testing.T) {
+	basicConfig := `
+	resource "redshift_role" "simple" {
+		name = "role_simple"
+	}
+	
+	resource "redshift_role" "fancy_name" {
+		name = "sOme_fancy_name-@www"
+	}
+`
+
+	resource.Test(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckRedshiftRoleDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: basicConfig,
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckRedshiftRoleExists("role_simple"),
+					resource.TestCheckResourceAttr("redshift_role.simple", "name", "role_simple"),
+
+					testAccCheckRedshiftRoleExists("sOme_fancy_name-@www"),
+					resource.TestCheckResourceAttr("redshift_role.fancy_name", "name", "sOme_fancy_name-@www"),
+				),
+			},
+		},
+	})
+}
+
+func TestAccRedshiftRole_Update(t *testing.T) {
+	roleNames := []string{
+		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_role"), "-", "_"),
+		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_role"), "-", "_"),
+		strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_role@tf_acc_domain.tld"), "-", "_"),
+	}
+	roleNameUpdated := strings.ReplaceAll(acctest.RandomWithPrefix("tf_acc_role_updated"), "-", "_")
+
+	for _, roleName := range roleNames {
+		configCreate := fmt.Sprintf(`
+		resource "redshift_role" "update_role" {
+		  name = %[1]q
+		}
+		`, roleName)
+
+		configUpdate := fmt.Sprintf(`
+		resource "redshift_role" "update_role" {
+		  name = %[1]q
+		}
+		`, roleNameUpdated)
+		resource.Test(t, resource.TestCase{
+			PreCheck:     func() { testAccPreCheck(t) },
+			Providers:    testAccProviders,
+			CheckDestroy: testAccCheckRedshiftRoleDestroy,
+			Steps: []resource.TestStep{
+				{
+					Config: configCreate,
+					Check: resource.ComposeTestCheckFunc(
+						testAccCheckRedshiftRoleExists(roleName),
+						resource.TestCheckResourceAttr("redshift_role.update_role", "name", strings.ToLower(roleName)),
+					),
+				},
+				{
+					Config: configUpdate,
+					Check: resource.ComposeTestCheckFunc(
+						testAccCheckRedshiftRoleExists(roleNameUpdated),
+						resource.TestCheckResourceAttr("redshift_role.update_role", "name", strings.ToLower(roleNameUpdated)),
+					),
+				},
+				// apply the first one again to check if all parameters roll back properly
+				{
+					Config: configCreate,
+					Check: resource.ComposeTestCheckFunc(
+						testAccCheckRedshiftRoleExists(roleName),
+						resource.TestCheckResourceAttr("redshift_role.update_role", "name", strings.ToLower(roleName)),
+					),
+				},
+			},
+		})
+	}
+}
+
+func testAccCheckRedshiftRoleDestroy(s *terraform.State) error {
+	client := testAccProvider.Meta().(*Client)
+
+	for _, rs := range s.RootModule().Resources {
+		if rs.Type != "redshift_role" {
+			continue
+		}
+
+		exists, err := checkRoleExists(client, rs.Primary.ID)
+
+		if err != nil {
+			return fmt.Errorf("Error checking role %s", err)
+		}
+
+		if exists {
+			return fmt.Errorf("Role still exists after destroy")
+		}
+	}
+
+	return nil
+}
+
+func testAccCheckRedshiftRoleExists(roleName string) resource.TestCheckFunc {
+	return func(s *terraform.State) error {
+		client := testAccProvider.Meta().(*Client)
+
+		exists, err := checkRoleExists(client, roleName)
+		if err != nil {
+			return fmt.Errorf("Error checking role %s", err)
+		}
+
+		if !exists {
+			return fmt.Errorf("Role %s not found", roleName)
+		}
+
+		return nil
+	}
+}
+
+func checkRoleExists(client *Client, roleName string) (bool, error) {
+	db, err := client.Connect()
+	if err != nil {
+		return false, err
+	}
+	var _rez int
+	err = db.QueryRow("SELECT 1 from svv_roles WHERE role_name=$1", roleName).Scan(&_rez)
+	switch {
+	case err == sql.ErrNoRows:
+		return false, nil
+	case err != nil:
+		return false, fmt.Errorf("Error reading info about role: %s", err)
+	}
+
+	return true, nil
+}
diff --git a/redshift/resource_redshift_schema_test.go b/redshift/resource_redshift_schema_test.go
index 61069ca..a927b19 100644
--- a/redshift/resource_redshift_schema_test.go
+++ b/redshift/resource_redshift_schema_test.go
@@ -625,7 +625,7 @@ resource "redshift_schema" "schema_configured" {
   name = "schema_configured"
   quota = 15
   cascade_on_delete = false
-  owner = upper(redshift_user.schema_test_user1.name)
+  owner = redshift_user.schema_test_user1.name
 }
 
 resource "redshift_schema" "fancy_name" {
diff --git a/redshift/resource_redshift_user_test.go b/redshift/resource_redshift_user_test.go
index 6553dd8..ae0e850 100644
--- a/redshift/resource_redshift_user_test.go
+++ b/redshift/resource_redshift_user_test.go
@@ -290,7 +290,7 @@ func TestAccRedshiftUser_SuperuserSyslogAccess(t *testing.T) {
 			resource "redshift_user" "superuser" {
 			  name = %[1]q
 			  superuser = local.is_superuser
-			  password  = "foobar12355#"
+			  password  = "Foobar12355#"
 			  syslog_access = %[3]q
 			}
 			`, userName, test.isSuperuser, test.syslogAccess)