diff --git a/command/base_predict_test.go b/command/base_predict_test.go
index de8e8610aafe..45093d09cc84 100644
--- a/command/base_predict_test.go
+++ b/command/base_predict_test.go
@@ -386,6 +386,7 @@ func TestPredict_Plugins(t *testing.T) {
 				"postgresql-database-plugin",
 				"rabbitmq",
 				"radius",
+				"redshift-database-plugin",
 				"ssh",
 				"totp",
 				"transit",
diff --git a/helper/builtinplugins/registry.go b/helper/builtinplugins/registry.go
index 4732ebbfb0ee..5d3b51679fb2 100644
--- a/helper/builtinplugins/registry.go
+++ b/helper/builtinplugins/registry.go
@@ -28,6 +28,7 @@ import (
 	dbMssql "github.com/hashicorp/vault/plugins/database/mssql"
 	dbMysql "github.com/hashicorp/vault/plugins/database/mysql"
 	dbPostgres "github.com/hashicorp/vault/plugins/database/postgresql"
+	dbRedshift "github.com/hashicorp/vault/plugins/database/redshift"
 	"github.com/hashicorp/vault/sdk/database/helper/credsutil"
 	"github.com/hashicorp/vault/sdk/helper/consts"
 	"github.com/hashicorp/vault/sdk/logical"
@@ -97,6 +98,7 @@ func newRegistry() *registry {
 			"mysql-legacy-database-plugin": dbMysql.New(credsutil.NoneLength, dbMysql.LegacyMetadataLen, dbMysql.LegacyUsernameLen),
 
 			"postgresql-database-plugin":    dbPostgres.New,
+			"redshift-database-plugin":      dbRedshift.New(true),
 			"mssql-database-plugin":         dbMssql.New,
 			"cassandra-database-plugin":     dbCass.New,
 			"mongodb-database-plugin":       dbMongo.New,
diff --git a/plugins/database/redshift/redshift-database-plugin/main.go b/plugins/database/redshift/redshift-database-plugin/main.go
new file mode 100644
index 000000000000..d7abf300c48f
--- /dev/null
+++ b/plugins/database/redshift/redshift-database-plugin/main.go
@@ -0,0 +1,20 @@
+package main
+
+import (
+	"log"
+	"os"
+
+	"github.com/hashicorp/vault/api"
+	"github.com/hashicorp/vault/plugins/database/redshift"
+)
+
+func main() {
+	apiClientMeta := &api.PluginAPIClientMeta{}
+	flags := apiClientMeta.FlagSet()
+	flags.Parse(os.Args[1:])
+
+	if err := redshift.Run(apiClientMeta.GetTLSConfig()); err != nil {
+		log.Println(err)
+		os.Exit(1)
+	}
+}
diff --git a/plugins/database/redshift/redshift.go b/plugins/database/redshift/redshift.go
new file mode 100644
index 000000000000..d2c688c20b92
--- /dev/null
+++ b/plugins/database/redshift/redshift.go
@@ -0,0 +1,522 @@
+package redshift
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/hashicorp/errwrap"
+	"github.com/hashicorp/go-multierror"
+	"github.com/hashicorp/vault/api"
+	"github.com/hashicorp/vault/sdk/database/dbplugin"
+	"github.com/hashicorp/vault/sdk/database/helper/connutil"
+	"github.com/hashicorp/vault/sdk/database/helper/credsutil"
+	"github.com/hashicorp/vault/sdk/database/helper/dbutil"
+	"github.com/hashicorp/vault/sdk/helper/dbtxn"
+	"github.com/hashicorp/vault/sdk/helper/strutil"
+	"github.com/lib/pq"
+)
+
+const (
+	// This is how this plugin will be reflected in middleware
+	// such as metrics.
+	middlewareTypeName = "redshift"
+
+	// This allows us to use the postgres database driver.
+	sqlTypeName = "postgres"
+
+	defaultRenewSQL = `
+ALTER USER "{{name}}" VALID UNTIL '{{expiration}}';
+`
+	defaultRotateRootCredentialsSQL = `
+ALTER USER "{{username}}" WITH PASSWORD '{{password}}';
+`
+)
+
+// lowercaseUsername is the reason we wrote this plugin. Redshift implements (mostly)
+// a postgres 8 interface, and part of that is under the hood, it's lowercasing the
+// usernames.
+func New(lowercaseUsername bool) func() (interface{}, error) {
+	return func() (interface{}, error) {
+		db := newRedshift(lowercaseUsername)
+		// Wrap the plugin with middleware to sanitize errors
+		dbType := dbplugin.NewDatabaseErrorSanitizerMiddleware(db, db.SecretValues)
+		return dbType, nil
+	}
+}
+
+func newRedshift(lowercaseUsername bool) *RedShift {
+	connProducer := &connutil.SQLConnectionProducer{}
+	connProducer.Type = sqlTypeName
+
+	credsProducer := &credsutil.SQLCredentialsProducer{
+		DisplayNameLen:    8,
+		RoleNameLen:       8,
+		UsernameLen:       63,
+		Separator:         "-",
+		LowercaseUsername: lowercaseUsername,
+	}
+
+	db := &RedShift{
+		SQLConnectionProducer: connProducer,
+		CredentialsProducer:   credsProducer,
+	}
+
+	return db
+}
+
+// Run instantiates a RedShift object, and runs the RPC server for the plugin
+func Run(apiTLSConfig *api.TLSConfig) error {
+	dbType, err := New(true)()
+	if err != nil {
+		return err
+	}
+
+	dbplugin.Serve(dbType.(dbplugin.Database), api.VaultPluginTLSProvider(apiTLSConfig))
+
+	return nil
+}
+
+type RedShift struct {
+	*connutil.SQLConnectionProducer
+	credsutil.CredentialsProducer
+}
+
+func (r *RedShift) Type() (string, error) {
+	return middlewareTypeName, nil
+}
+
+// getConnection accepts a context and retuns a new pointer to a sql.DB object.
+// It's up to the caller to close the connection or handle reuse logic.
+func (r *RedShift) getConnection(ctx context.Context) (*sql.DB, error) {
+	db, err := r.Connection(ctx)
+	if err != nil {
+		return nil, err
+	}
+	return db.(*sql.DB), nil
+}
+
+// SetCredentials uses provided information to set/create a user in the
+// database. Unlike CreateUser, this method requires a username be provided and
+// uses the name given, instead of generating a name. This is used for creating
+// and setting the password of static accounts, as well as rolling back
+// passwords in the database in the event an updated database fails to save in
+// Vault's storage.
+func (r *RedShift) SetCredentials(ctx context.Context, statements dbplugin.Statements, staticUser dbplugin.StaticUserConfig) (username, password string, err error) {
+	if len(statements.Rotation) == 0 {
+		return "", "", errors.New("empty rotation statements")
+	}
+
+	username = staticUser.Username
+	password = staticUser.Password
+	if username == "" || password == "" {
+		return "", "", errors.New("must provide both username and password")
+	}
+
+	// Grab the lock
+	r.Lock()
+	defer r.Unlock()
+
+	// Get the connection
+	db, err := r.getConnection(ctx)
+	if err != nil {
+		return "", "", err
+	}
+	defer db.Close()
+
+	// Check if the role exists
+	var exists bool
+	err = db.QueryRowContext(ctx, "SELECT exists (SELECT usename FROM pg_user WHERE usename=$1);", username).Scan(&exists)
+	if err != nil && err != sql.ErrNoRows {
+		return "", "", err
+	}
+
+	// Vault requires the database user already exist, and that the credentials
+	// used to execute the rotation statements has sufficient privileges.
+	stmts := statements.Rotation
+
+	// Start a transaction
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		return "", "", err
+	}
+	defer func() {
+		tx.Rollback()
+	}()
+
+	// Execute each query
+	for _, stmt := range stmts {
+		for _, query := range strutil.ParseArbitraryStringSlice(stmt, ";") {
+			query = strings.TrimSpace(query)
+			if len(query) == 0 {
+				continue
+			}
+
+			m := map[string]string{
+				"name":     staticUser.Username,
+				"password": password,
+			}
+			if err := dbtxn.ExecuteTxQuery(ctx, tx, m, query); err != nil {
+				return "", "", err
+			}
+		}
+	}
+
+	// Commit the transaction
+	if err := tx.Commit(); err != nil {
+		return "", "", err
+	}
+	return username, password, nil
+}
+
+func (r *RedShift) CreateUser(ctx context.Context, statements dbplugin.Statements, usernameConfig dbplugin.UsernameConfig, expiration time.Time) (username string, password string, err error) {
+	statements = dbutil.StatementCompatibilityHelper(statements)
+
+	if len(statements.Creation) == 0 {
+		return "", "", dbutil.ErrEmptyCreationStatement
+	}
+
+	// Grab the lock
+	r.Lock()
+	defer r.Unlock()
+
+	username, err = r.GenerateUsername(usernameConfig)
+	if err != nil {
+		return "", "", err
+	}
+
+	password, err = r.GeneratePassword()
+	if err != nil {
+		return "", "", err
+	}
+
+	expirationStr, err := r.GenerateExpiration(expiration)
+	if err != nil {
+		return "", "", err
+	}
+
+	// Get the connection
+	db, err := r.getConnection(ctx)
+	if err != nil {
+		return "", "", err
+	}
+	defer db.Close()
+
+	// Start a transaction
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		return "", "", err
+
+	}
+	defer func() {
+		tx.Rollback()
+	}()
+
+	// Execute each query
+	for _, stmt := range statements.Creation {
+		for _, query := range strutil.ParseArbitraryStringSlice(stmt, ";") {
+			query = strings.TrimSpace(query)
+			if len(query) == 0 {
+				continue
+			}
+
+			m := map[string]string{
+				"name":       username,
+				"password":   password,
+				"expiration": expirationStr,
+			}
+			if err := dbtxn.ExecuteTxQuery(ctx, tx, m, query); err != nil {
+				return "", "", err
+			}
+		}
+	}
+
+	// Commit the transaction
+	if err := tx.Commit(); err != nil {
+		return "", "", err
+	}
+	return username, password, nil
+}
+
+func (r *RedShift) RenewUser(ctx context.Context, statements dbplugin.Statements, username string, expiration time.Time) error {
+	r.Lock()
+	defer r.Unlock()
+
+	statements = dbutil.StatementCompatibilityHelper(statements)
+
+	renewStmts := statements.Renewal
+	if len(renewStmts) == 0 {
+		renewStmts = []string{defaultRenewSQL}
+	}
+
+	db, err := r.getConnection(ctx)
+	if err != nil {
+		return err
+	}
+	defer db.Close()
+
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		tx.Rollback()
+	}()
+
+	expirationStr, err := r.GenerateExpiration(expiration)
+	if err != nil {
+		return err
+	}
+
+	for _, stmt := range renewStmts {
+		for _, query := range strutil.ParseArbitraryStringSlice(stmt, ";") {
+			query = strings.TrimSpace(query)
+			if len(query) == 0 {
+				continue
+			}
+
+			m := map[string]string{
+				"name":       username,
+				"expiration": expirationStr,
+			}
+			if err := dbtxn.ExecuteTxQuery(ctx, tx, m, query); err != nil {
+				return err
+			}
+		}
+	}
+
+	return tx.Commit()
+}
+
+func (r *RedShift) RevokeUser(ctx context.Context, statements dbplugin.Statements, username string) error {
+	// Grab the lock
+	r.Lock()
+	defer r.Unlock()
+
+	statements = dbutil.StatementCompatibilityHelper(statements)
+
+	if len(statements.Revocation) == 0 {
+		return r.defaultRevokeUser(ctx, username)
+	}
+
+	return r.customRevokeUser(ctx, username, statements.Revocation)
+}
+
+func (r *RedShift) customRevokeUser(ctx context.Context, username string, revocationStmts []string) error {
+	db, err := r.getConnection(ctx)
+	if err != nil {
+		return err
+	}
+	defer db.Close()
+
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		tx.Rollback()
+	}()
+
+	for _, stmt := range revocationStmts {
+		for _, query := range strutil.ParseArbitraryStringSlice(stmt, ";") {
+			query = strings.TrimSpace(query)
+			if len(query) == 0 {
+				continue
+			}
+
+			m := map[string]string{
+				"name": username,
+			}
+			if err := dbtxn.ExecuteTxQuery(ctx, tx, m, query); err != nil {
+				return err
+			}
+		}
+	}
+
+	return tx.Commit()
+}
+
+func (r *RedShift) defaultRevokeUser(ctx context.Context, username string) error {
+	db, err := r.getConnection(ctx)
+	if err != nil {
+		return err
+	}
+	defer db.Close()
+
+	// Check if the role exists
+	var exists bool
+	err = db.QueryRowContext(ctx, "SELECT exists (SELECT usename FROM pg_user WHERE usename=$1);", username).Scan(&exists)
+	if err != nil && err != sql.ErrNoRows {
+		return err
+	}
+
+	if !exists {
+		return nil
+	}
+
+	// Query for permissions; we need to revoke permissions before we can drop
+	// the role
+	// This isn't done in a transaction because even if we fail along the way,
+	// we want to remove as much access as possible
+	stmt, err := db.PrepareContext(ctx, "SELECT DISTINCT table_schema FROM information_schema.role_column_grants WHERE grantee=$1;")
+	if err != nil {
+		return err
+	}
+	defer stmt.Close()
+
+	rows, err := stmt.QueryContext(ctx, username)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+
+	const initialNumRevocations = 16
+	revocationStmts := make([]string, 0, initialNumRevocations)
+	for rows.Next() {
+		var schema string
+		err = rows.Scan(&schema)
+		if err != nil {
+			// keep going; remove as many permissions as possible right now
+			continue
+		}
+		revocationStmts = append(revocationStmts, fmt.Sprintf(
+			`REVOKE ALL PRIVILEGES ON ALL TABLES IN SCHEMA %s FROM %s;`,
+			pq.QuoteIdentifier(schema),
+			pq.QuoteIdentifier(username)))
+
+		revocationStmts = append(revocationStmts, fmt.Sprintf(
+			`REVOKE USAGE ON SCHEMA %s FROM %s;`,
+			pq.QuoteIdentifier(schema),
+			pq.QuoteIdentifier(username)))
+	}
+
+	// for good measure, revoke all privileges and usage on schema public
+	revocationStmts = append(revocationStmts, fmt.Sprintf(
+		`REVOKE ALL PRIVILEGES ON ALL TABLES IN SCHEMA public FROM %s;`,
+		pq.QuoteIdentifier(username)))
+
+	revocationStmts = append(revocationStmts, fmt.Sprintf(
+		"REVOKE USAGE ON SCHEMA public FROM %s;",
+		pq.QuoteIdentifier(username)))
+
+	// get the current database name so we can issue a REVOKE CONNECT for
+	// this username
+	var dbname sql.NullString
+	if err := db.QueryRowContext(ctx, "SELECT current_database();").Scan(&dbname); err != nil {
+		return err
+	}
+
+	if dbname.Valid {
+		/*
+			We create this stored procedure to ensure we can durably revoke users on Redshift. We do not
+			clean up since that can cause race conditions with other instances of Vault attempting to use
+			this SP at the same time.
+		*/
+		revocationStmts = append(revocationStmts, `CREATE OR REPLACE PROCEDURE terminateloop(dbusername varchar(100))
+LANGUAGE plpgsql
+AS $$
+DECLARE
+  currentpid int;
+  loopvar int;
+  qtyconns int;
+BEGIN
+SELECT COUNT(process) INTO qtyconns FROM stv_sessions WHERE user_name=dbusername;
+  FOR loopvar IN 1..qtyconns LOOP
+    SELECT INTO currentpid process FROM stv_sessions WHERE user_name=dbusername ORDER BY process ASC LIMIT 1;
+    SELECT pg_terminate_backend(currentpid);
+  END LOOP;
+END
+$$;`)
+
+		revocationStmts = append(revocationStmts, fmt.Sprintf(`call terminateloop('%s');`, username))
+	}
+
+	// again, here, we do not stop on error, as we want to remove as
+	// many permissions as possible right now
+	var lastStmtError *multierror.Error //error
+	for _, query := range revocationStmts {
+		if err := dbtxn.ExecuteDBQuery(ctx, db, nil, query); err != nil {
+			lastStmtError = multierror.Append(lastStmtError, err)
+		}
+	}
+
+	// can't drop if not all privileges are revoked
+	if rows.Err() != nil {
+		return errwrap.Wrapf("could not generate revocation statements for all rows: {{err}}", rows.Err())
+	}
+	if lastStmtError != nil {
+		return errwrap.Wrapf("could not perform all revocation statements: {{err}}", lastStmtError)
+	}
+
+	// Drop this user
+	stmt, err = db.PrepareContext(ctx, fmt.Sprintf(
+		`DROP USER IF EXISTS %s;`, pq.QuoteIdentifier(username)))
+	if err != nil {
+		return err
+	}
+	defer stmt.Close()
+	if _, err := stmt.ExecContext(ctx); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (r *RedShift) RotateRootCredentials(ctx context.Context, statements []string) (map[string]interface{}, error) {
+	r.Lock()
+	defer r.Unlock()
+
+	if len(r.Username) == 0 || len(r.Password) == 0 {
+		return nil, errors.New("username and password are required to rotate")
+	}
+
+	rotateStatements := statements
+	if len(rotateStatements) == 0 {
+		rotateStatements = []string{defaultRotateRootCredentialsSQL}
+	}
+
+	db, err := r.getConnection(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer db.Close()
+
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		return nil, err
+	}
+	defer func() {
+		tx.Rollback()
+	}()
+
+	password, err := r.GeneratePassword()
+	if err != nil {
+		return nil, err
+	}
+
+	for _, stmt := range rotateStatements {
+		for _, query := range strutil.ParseArbitraryStringSlice(stmt, ";") {
+			query = strings.TrimSpace(query)
+			if len(query) == 0 {
+				continue
+			}
+			m := map[string]string{
+				"username": r.Username,
+				"password": password,
+			}
+			if err := dbtxn.ExecuteTxQuery(ctx, tx, m, query); err != nil {
+				return nil, err
+			}
+		}
+	}
+
+	if err := tx.Commit(); err != nil {
+		return nil, err
+	}
+
+	r.RawConfig["password"] = password
+	return r.RawConfig, nil
+}
diff --git a/plugins/database/redshift/redshift_test.go b/plugins/database/redshift/redshift_test.go
new file mode 100644
index 000000000000..c8b6ffdd30ed
--- /dev/null
+++ b/plugins/database/redshift/redshift_test.go
@@ -0,0 +1,528 @@
+package redshift
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/go-uuid"
+	"github.com/hashicorp/vault/sdk/database/dbplugin"
+	"github.com/hashicorp/vault/sdk/helper/dbtxn"
+	"github.com/lib/pq"
+)
+
+/*
+To run these sets of acceptance tests, you must pre-configure a Redshift cluster
+in AWS and ensure the machine running these tests has network access to it.
+
+Once the redshift cluster is running, you can pass the admin username and password
+as environment variables to be used to run these tests. Note that these tests
+will create users on your redshift cluster and currently do not clean up after
+themselves.
+
+The RotateRoot test is potentially destructive in that it will rotate your root
+password on your Redshift cluster to an insecure, cleartext password defined in the
+test method. Because of this, you must pass TEST_ROTATE_ROOT=1 to enable it explicitly.
+
+Do not run this test suite against a production Redshift cluster.
+
+Configuration:
+
+		REDSHIFT_URL=my-redshift-url.region.redshift.amazonaws.com:5439/database-name
+		REDSHIFT_USER=my-redshift-admin-user
+		REDSHIFT_PASSWORD=my-redshift-admin-password
+		VAULT_ACC=<unset || 1> # This must be set to run any of the tests in this test suite
+		TEST_ROTATE_ROOT=<unset || 1> # This must be set to explicitly run the rotate root test
+*/
+
+var (
+	keyRedshiftURL      = "REDSHIFT_URL"
+	keyRedshiftUser     = "REDSHIFT_USER"
+	keyRedshiftPassword = "REDSHIFT_PASSWORD"
+
+	vaultACC = "VAULT_ACC"
+)
+
+func redshiftEnv() (url string, user string, password string, errEmpty error) {
+	errEmpty = errors.New("err: empty but required env value")
+
+	if url = os.Getenv(keyRedshiftURL); url == "" {
+		return "", "", "", errEmpty
+	}
+
+	if user = os.Getenv(keyRedshiftUser); url == "" {
+		return "", "", "", errEmpty
+	}
+
+	if password = os.Getenv(keyRedshiftPassword); url == "" {
+		return "", "", "", errEmpty
+	}
+
+	url = fmt.Sprintf("postgres://%s:%s@%s", user, password, url)
+
+	return url, user, password, nil
+}
+
+func TestPostgreSQL_Initialize(t *testing.T) {
+	if os.Getenv(vaultACC) != "1" {
+		t.SkipNow()
+	}
+
+	url, _, _, err := redshiftEnv()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	connectionDetails := map[string]interface{}{
+		"connection_url":       url,
+		"max_open_connections": 5,
+	}
+
+	db := newRedshift(true)
+	_, err = db.Init(context.Background(), connectionDetails, true)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	if !db.Initialized {
+		t.Fatal("Database should be initialized")
+	}
+
+	err = db.Close()
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	// Test decoding a string value for max_open_connections
+	connectionDetails = map[string]interface{}{
+		"connection_url":       url,
+		"max_open_connections": "5",
+	}
+
+	_, err = db.Init(context.Background(), connectionDetails, true)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+}
+
+func TestPostgreSQL_CreateUser(t *testing.T) {
+	if os.Getenv(vaultACC) != "1" {
+		t.SkipNow()
+	}
+
+	url, _, _, err := redshiftEnv()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	connectionDetails := map[string]interface{}{
+		"connection_url": url,
+	}
+
+	db := newRedshift(true)
+	_, err = db.Init(context.Background(), connectionDetails, true)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	usernameConfig := dbplugin.UsernameConfig{
+		DisplayName: "test",
+		RoleName:    "test",
+	}
+
+	// Test with no configured Creation Statement
+	_, _, err = db.CreateUser(context.Background(), dbplugin.Statements{}, usernameConfig, time.Now().Add(time.Minute))
+	if err == nil {
+		t.Fatal("Expected error when no creation statement is provided")
+	}
+
+	statements := dbplugin.Statements{
+		Creation: []string{testRedshiftRole},
+	}
+
+	username, password, err := db.CreateUser(context.Background(), statements, usernameConfig, time.Now().Add(time.Minute))
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	if err = testCredsExist(t, url, username, password); err != nil {
+		t.Fatalf("Could not connect with new credentials: %s\n%s:%s", err, username, password)
+	}
+
+	statements.Creation = []string{testRedshiftReadOnlyRole}
+	username, password, err = db.CreateUser(context.Background(), statements, usernameConfig, time.Now().Add(time.Minute))
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	// Sleep to make sure we haven't expired if granularity is only down to the second
+	time.Sleep(2 * time.Second)
+
+	if err = testCredsExist(t, url, username, password); err != nil {
+		t.Fatalf("Could not connect with new credentials: %s", err)
+	}
+}
+
+func TestPostgreSQL_RenewUser(t *testing.T) {
+	if os.Getenv(vaultACC) != "1" {
+		t.SkipNow()
+	}
+
+	url, _, _, err := redshiftEnv()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	connectionDetails := map[string]interface{}{
+		"connection_url": url,
+	}
+
+	db := newRedshift(true)
+	_, err = db.Init(context.Background(), connectionDetails, true)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	statements := dbplugin.Statements{
+		Creation: []string{testRedshiftRole},
+	}
+
+	usernameConfig := dbplugin.UsernameConfig{
+		DisplayName: "test",
+		RoleName:    "test",
+	}
+
+	username, password, err := db.CreateUser(context.Background(), statements, usernameConfig, time.Now().Add(2*time.Second))
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	if err = testCredsExist(t, url, username, password); err != nil {
+		t.Fatalf("Could not connect with new credentials: %s", err)
+	}
+
+	err = db.RenewUser(context.Background(), statements, username, time.Now().Add(time.Minute))
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	// Sleep longer than the initial expiration time
+	time.Sleep(2 * time.Second)
+
+	if err = testCredsExist(t, url, username, password); err != nil {
+		t.Fatalf("Could not connect with new credentials: %s", err)
+	}
+	statements.Renewal = []string{defaultRenewSQL}
+	username, password, err = db.CreateUser(context.Background(), statements, usernameConfig, time.Now().Add(2*time.Second))
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	if err = testCredsExist(t, url, username, password); err != nil {
+		t.Fatalf("Could not connect with new credentials: %s", err)
+	}
+
+	err = db.RenewUser(context.Background(), statements, username, time.Now().Add(time.Minute))
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	// Sleep longer than the initial expiration time
+	time.Sleep(2 * time.Second)
+
+	if err = testCredsExist(t, url, username, password); err != nil {
+		t.Fatalf("Could not connect with new credentials: %s", err)
+	}
+
+}
+
+func TestPostgreSQL_RevokeUser(t *testing.T) {
+	if os.Getenv(vaultACC) != "1" {
+		t.SkipNow()
+	}
+
+	url, _, _, err := redshiftEnv()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	connectionDetails := map[string]interface{}{
+		"connection_url": url,
+	}
+
+	db := newRedshift(true)
+	_, err = db.Init(context.Background(), connectionDetails, true)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	statements := dbplugin.Statements{
+		Creation: []string{testRedshiftRole},
+	}
+
+	usernameConfig := dbplugin.UsernameConfig{
+		DisplayName: "test",
+		RoleName:    "test",
+	}
+
+	username, password, err := db.CreateUser(context.Background(), statements, usernameConfig, time.Now().Add(2*time.Second))
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	if err = testCredsExist(t, url, username, password); err != nil {
+		t.Fatalf("Could not connect with new credentials: %s", err)
+	}
+
+	// Test default revoke statements
+	err = db.RevokeUser(context.Background(), statements, username)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	if err := testCredsExist(t, url, username, password); err == nil {
+		t.Fatal("Credentials were not revoked")
+	}
+
+	username, password, err = db.CreateUser(context.Background(), statements, usernameConfig, time.Now().Add(2*time.Second))
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	if err = testCredsExist(t, url, username, password); err != nil {
+		t.Fatalf("Could not connect with new credentials: %s", err)
+	}
+
+	// Test custom revoke statements
+	statements.Revocation = []string{defaultRedshiftRevocationSQL}
+	err = db.RevokeUser(context.Background(), statements, username)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	if err := testCredsExist(t, url, username, password); err == nil {
+		t.Fatal("Credentials were not revoked")
+	}
+}
+
+func TestPostgresSQL_SetCredentials(t *testing.T) {
+	if os.Getenv(vaultACC) != "1" {
+		t.SkipNow()
+	}
+
+	url, _, _, err := redshiftEnv()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	connectionDetails := map[string]interface{}{
+		"connection_url": url,
+	}
+
+	// create the database user
+	uid, err := uuid.GenerateUUID()
+	if err != nil {
+		t.Fatal(err)
+	}
+	dbUser := "vaultstatictest-" + fmt.Sprintf("%s", uid)
+	createTestPGUser(t, url, dbUser, "1Password", testRoleStaticCreate)
+
+	db := newRedshift(true)
+	_, err = db.Init(context.Background(), connectionDetails, true)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	password, err := db.GenerateCredentials(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	usernameConfig := dbplugin.StaticUserConfig{
+		Username: dbUser,
+		Password: password,
+	}
+
+	// Test with no configured Rotation Statement
+	username, password, err := db.SetCredentials(context.Background(), dbplugin.Statements{}, usernameConfig)
+	if err == nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	statements := dbplugin.Statements{
+		Rotation: []string{testRedshiftStaticRoleRotate},
+	}
+	// User should not exist, make sure we can create
+	username, password, err = db.SetCredentials(context.Background(), statements, usernameConfig)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	if err := testCredsExist(t, url, username, password); err != nil {
+		t.Fatalf("Could not connect with new credentials: %s", err)
+	}
+
+	// call SetCredentials again, password will change
+	newPassword, _ := db.GenerateCredentials(context.Background())
+	usernameConfig.Password = newPassword
+	username, password, err = db.SetCredentials(context.Background(), statements, usernameConfig)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	if password != newPassword {
+		t.Fatal("passwords should have changed")
+	}
+
+	if err := testCredsExist(t, url, username, password); err != nil {
+		t.Fatalf("Could not connect with new credentials: %s", err)
+	}
+}
+
+func TestPostgreSQL_RotateRootCredentials(t *testing.T) {
+	/*
+		   Extra precaution is taken for rotating root creds because it's assumed that this
+		   test will run against a live redshift cluster. This test must run last because
+		   it is destructive.
+
+			 To run this test you must pass TEST_ROTATE_ROOT=1
+	*/
+	if os.Getenv(vaultACC) != "1" || os.Getenv("TEST_ROTATE_ROOT") != "1" {
+		t.SkipNow()
+	}
+
+	url, adminUser, adminPassword, err := redshiftEnv()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	connectionDetails := map[string]interface{}{
+		"connection_url": url,
+		"username":       adminUser,
+		"password":       adminPassword,
+	}
+
+	db := newRedshift(true)
+
+	connProducer := db.SQLConnectionProducer
+
+	_, err = db.Init(context.Background(), connectionDetails, true)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	if !connProducer.Initialized {
+		t.Fatal("Database should be initialized")
+	}
+
+	newConf, err := db.RotateRootCredentials(context.Background(), nil)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	fmt.Printf("rotated root credentials, new user/pass:\nusername: %s\npassword: %s\n", newConf["username"], newConf["password"])
+
+	if newConf["password"] == adminPassword {
+		t.Fatal("password was not updated")
+	}
+
+	err = db.Close()
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+}
+
+func testCredsExist(t testing.TB, connURL, username, password string) error {
+	t.Helper()
+	_, adminUser, adminPassword, err := redshiftEnv()
+	if err != nil {
+		return err
+	}
+
+	connURL = strings.Replace(connURL, fmt.Sprintf("%s:%s", adminUser, adminPassword), fmt.Sprintf("%s:%s", username, password), 1)
+	db, err := sql.Open("postgres", connURL)
+	if err != nil {
+		return err
+	}
+	defer db.Close()
+	return db.Ping()
+}
+
+const testRedshiftRole = `
+CREATE USER "{{name}}" WITH PASSWORD '{{password}}' VALID UNTIL '{{expiration}}'; 
+GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA public TO "{{name}}";
+`
+
+const testRedshiftReadOnlyRole = `
+CREATE USER "{{name}}" WITH
+  PASSWORD '{{password}}'
+  VALID UNTIL '{{expiration}}';
+GRANT SELECT ON ALL TABLES IN SCHEMA public TO "{{name}}";
+`
+
+const defaultRedshiftRevocationSQL = `
+REVOKE ALL PRIVILEGES ON ALL TABLES IN SCHEMA public FROM "{{name}}";
+REVOKE USAGE ON SCHEMA public FROM "{{name}}";
+
+DROP USER IF EXISTS "{{name}}";
+`
+
+const testRedshiftStaticRole = `
+CREATE USER "{{name}}" WITH
+  PASSWORD '{{password}}';
+GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA public TO "{{name}}";
+`
+
+const testRoleStaticCreate = `
+CREATE USER "{{name}}" WITH
+  PASSWORD '{{password}}';
+`
+
+const testRedshiftStaticRoleRotate = `
+ALTER USER "{{name}}" WITH PASSWORD '{{password}}';
+`
+
+// This is a copy of a test helper method also found in
+// builtin/logical/database/rotation_test.go , and should be moved into a shared
+// helper file in the future.
+func createTestPGUser(t *testing.T, connURL string, username, password, query string) {
+	t.Helper()
+	conn, err := pq.ParseURL(connURL)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	db, err := sql.Open("postgres", conn)
+	defer db.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Start a transaction
+	ctx := context.Background()
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		_ = tx.Rollback()
+	}()
+
+	m := map[string]string{
+		"name":     username,
+		"password": password,
+	}
+	if err := dbtxn.ExecuteTxQuery(ctx, tx, m, query); err != nil {
+		t.Fatal(err)
+	}
+	// Commit the transaction
+	if err := tx.Commit(); err != nil {
+		t.Fatal(err)
+	}
+}
diff --git a/sdk/database/helper/connutil/sql.go b/sdk/database/helper/connutil/sql.go
index 796df8b16f8d..08f610ac7cc7 100644
--- a/sdk/database/helper/connutil/sql.go
+++ b/sdk/database/helper/connutil/sql.go
@@ -131,9 +131,9 @@ func (c *SQLConnectionProducer) Connection(ctx context.Context) (interface{}, er
 	// Ensure timezone is set to UTC for all the connections
 	if strings.HasPrefix(conn, "postgres://") || strings.HasPrefix(conn, "postgresql://") {
 		if strings.Contains(conn, "?") {
-			conn += "&timezone=utc"
+			conn += "&timezone=UTC"
 		} else {
-			conn += "?timezone=utc"
+			conn += "?timezone=UTC"
 		}
 	}
 
diff --git a/sdk/database/helper/credsutil/sql.go b/sdk/database/helper/credsutil/sql.go
index 748b504effb7..986631da94cb 100644
--- a/sdk/database/helper/credsutil/sql.go
+++ b/sdk/database/helper/credsutil/sql.go
@@ -3,6 +3,7 @@ package credsutil
 import (
 	"context"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/hashicorp/vault/sdk/database/dbplugin"
@@ -14,10 +15,11 @@ const (
 
 // SQLCredentialsProducer implements CredentialsProducer and provides a generic credentials producer for most sql database types.
 type SQLCredentialsProducer struct {
-	DisplayNameLen int
-	RoleNameLen    int
-	UsernameLen    int
-	Separator      string
+	DisplayNameLen    int
+	RoleNameLen       int
+	UsernameLen       int
+	Separator         string
+	LowercaseUsername bool
 }
 
 func (scp *SQLCredentialsProducer) GenerateCredentials(ctx context.Context) (string, error) {
@@ -64,6 +66,10 @@ func (scp *SQLCredentialsProducer) GenerateUsername(config dbplugin.UsernameConf
 		username = username[:scp.UsernameLen]
 	}
 
+	if scp.LowercaseUsername {
+		username = strings.ToLower(username)
+	}
+
 	return username, nil
 }
 
diff --git a/vault/testing.go b/vault/testing.go
index 54be9826fc3b..324508feaa20 100644
--- a/vault/testing.go
+++ b/vault/testing.go
@@ -1844,6 +1844,7 @@ func (m *mockBuiltinRegistry) Keys(pluginType consts.PluginType) []string {
 		"mongodbatlas-database-plugin",
 		"hana-database-plugin",
 		"influxdb-database-plugin",
+		"redshift-database-plugin",
 	}
 }
 
diff --git a/vendor/github.com/hashicorp/vault/sdk/database/helper/connutil/sql.go b/vendor/github.com/hashicorp/vault/sdk/database/helper/connutil/sql.go
index 796df8b16f8d..08f610ac7cc7 100644
--- a/vendor/github.com/hashicorp/vault/sdk/database/helper/connutil/sql.go
+++ b/vendor/github.com/hashicorp/vault/sdk/database/helper/connutil/sql.go
@@ -131,9 +131,9 @@ func (c *SQLConnectionProducer) Connection(ctx context.Context) (interface{}, er
 	// Ensure timezone is set to UTC for all the connections
 	if strings.HasPrefix(conn, "postgres://") || strings.HasPrefix(conn, "postgresql://") {
 		if strings.Contains(conn, "?") {
-			conn += "&timezone=utc"
+			conn += "&timezone=UTC"
 		} else {
-			conn += "?timezone=utc"
+			conn += "?timezone=UTC"
 		}
 	}
 
diff --git a/vendor/github.com/hashicorp/vault/sdk/database/helper/credsutil/sql.go b/vendor/github.com/hashicorp/vault/sdk/database/helper/credsutil/sql.go
index 748b504effb7..986631da94cb 100644
--- a/vendor/github.com/hashicorp/vault/sdk/database/helper/credsutil/sql.go
+++ b/vendor/github.com/hashicorp/vault/sdk/database/helper/credsutil/sql.go
@@ -3,6 +3,7 @@ package credsutil
 import (
 	"context"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/hashicorp/vault/sdk/database/dbplugin"
@@ -14,10 +15,11 @@ const (
 
 // SQLCredentialsProducer implements CredentialsProducer and provides a generic credentials producer for most sql database types.
 type SQLCredentialsProducer struct {
-	DisplayNameLen int
-	RoleNameLen    int
-	UsernameLen    int
-	Separator      string
+	DisplayNameLen    int
+	RoleNameLen       int
+	UsernameLen       int
+	Separator         string
+	LowercaseUsername bool
 }
 
 func (scp *SQLCredentialsProducer) GenerateCredentials(ctx context.Context) (string, error) {
@@ -64,6 +66,10 @@ func (scp *SQLCredentialsProducer) GenerateUsername(config dbplugin.UsernameConf
 		username = username[:scp.UsernameLen]
 	}
 
+	if scp.LowercaseUsername {
+		username = strings.ToLower(username)
+	}
+
 	return username, nil
 }
 
diff --git a/website/pages/api-docs/secret/databases/redshift.mdx b/website/pages/api-docs/secret/databases/redshift.mdx
new file mode 100644
index 000000000000..ea6404505df5
--- /dev/null
+++ b/website/pages/api-docs/secret/databases/redshift.mdx
@@ -0,0 +1,117 @@
+---
+layout: api
+page_title: Redshift - Database - Secrets Engines - HTTP API
+sidebar_title: Redshift
+description: >-
+  The Redshift plugin for Vault's database secrets engine generates database
+  credentials to access the AWS Redshift service.
+---
+
+# Redshift Database Plugin HTTP API
+
+The Redshift database plugin is one of the supported plugins for the database
+secrets engine. This plugin generates database credentials dynamically based on
+configured roles for the Redshift database.
+
+## Configure Connection
+
+In addition to the parameters defined by the [Database
+Backend](/api/secret/databases#configure-connection), this plugin
+has a number of parameters to further configure a connection.
+
+| Method | Path                     |
+| :----- | :----------------------- |
+| `POST` | `/database/config/:name` |
+
+### Parameters
+
+- `connection_url` `(string: <required>)` - Specifies the Redshift DSN. This field
+  can be templated and supports passing the username and password
+  parameters in the following format {{field_name}}. A templated connection URL is
+  required when using root credential rotation.
+
+- `max_open_connections` `(int: 4)` - Specifies the maximum number of open
+  connections to the database.
+
+- `max_idle_connections` `(int: 0)` - Specifies the maximum number of idle
+  connections to the database. A zero uses the value of `max_open_connections`
+  and a negative value disables idle connections. If larger than
+  `max_open_connections` it will be reduced to be equal.
+
+- `max_connection_lifetime` `(string: "0s")` - Specifies the maximum amount of
+  time a connection may be reused. If <= 0s connections are reused forever.
+
+- `username` `(string: "")` - The root credential username used in the connection URL.
+
+- `password` `(string: "")` - The root credential password used in the connection URL.
+
+### Sample Payload
+
+```json
+{
+  "plugin_name": "redshift-database-plugin",
+  "allowed_roles": "readonly",
+  "connection_url": "postgresql://{{username}}:{{password}}@localhost:5432/dev",
+  "max_open_connections": 5,
+  "max_connection_lifetime": "5s",
+  "username": "username",
+  "password": "password"
+}
+```
+
+### Sample Request
+
+```
+$ curl \
+    --header "X-Vault-Token: ..." \
+    --request POST \
+    --data @payload.json \
+    http://127.0.0.1:8200/v1/database/config/redshift
+```
+
+## Statements
+
+Statements are configured during role creation and are used by the plugin to
+determine what is sent to the database on user creation, renewing, and
+revocation. For more information on configuring roles see the [Role
+API](/api/secret/databases#create-role) in the database secrets engine docs.
+
+### Parameters
+
+The following are the statements used by this plugin. If not mentioned in this
+list the plugin does not support that statement type.
+
+- `creation_statements` `(list: <required>)` – Specifies the database
+  statements executed to create and configure a user. Must be a
+  semicolon-separated string, a base64-encoded semicolon-separated string, a
+  serialized JSON string array, or a base64-encoded serialized JSON string
+  array. The '{{name}}', '{{password}}' and '{{expiration}}' values will be
+  substituted. The generated password will be a random alphanumeric 20 character
+  string.
+
+- `revocation_statements` `(list: [])` – Specifies the database statements to
+  be executed to revoke a user. Must be a semicolon-separated string, a
+  base64-encoded semicolon-separated string, a serialized JSON string array, or
+  a base64-encoded serialized JSON string array. The '{{name}}' value will be
+  substituted. If not provided defaults to a generic drop user statement.
+
+- `rollback_statements` `(list: [])` – Specifies the database statements to be
+  executed rollback a create operation in the event of an error. Not every
+  plugin type will support this functionality. Must be a semicolon-separated
+  string, a base64-encoded semicolon-separated string, a serialized JSON string
+  array, or a base64-encoded serialized JSON string array. The '{{name}}' value
+  will be substituted.
+
+- `renew_statements` `(list: [])` – Specifies the database statements to be
+  executed to renew a user. Not every plugin type will support this
+  functionality. Must be a semicolon-separated string, a base64-encoded
+  semicolon-separated string, a serialized JSON string array, or a
+  base64-encoded serialized JSON string array. The '{{name}}' and
+  '{{expiration}}' values will be substituted.
+
+- `rotation_statements` `(list: [])` – Specifies the database statements to be
+  executed to rotate the password for a given username. Must be a
+  semicolon-separated string, a base64-encoded semicolon-separated string, a
+  serialized JSON string array, or a base64-encoded serialized JSON string
+  array. The '{{name}}' and '{{password}}' values will be substituted. The
+  generated password will be a random alphanumeric 20 character string.
diff --git a/website/pages/docs/secrets/databases/redshift.mdx b/website/pages/docs/secrets/databases/redshift.mdx
new file mode 100644
index 000000000000..5876bbd8f910
--- /dev/null
+++ b/website/pages/docs/secrets/databases/redshift.mdx
@@ -0,0 +1,82 @@
+---
+layout: docs
+page_title: Redshift - Database - Secrets Engines
+sidebar_title: Redshift
+description: |-
+  Redshift is a supported plugin for the database secrets engine.
+  This plugin generates database credentials dynamically based on configured
+  roles for the AWS Redshift database service.
+---
+
+# Redshift Database Secrets Engine
+
+Redshift is a supported plugin for the database secrets engine. This
+plugin generates database credentials dynamically based on configured roles for
+the AWS Redshift database service, and also supports [Static
+Roles](/docs/secrets/databases#static-roles).
+
+See the [database secrets engine](/docs/secrets/databases) docs for
+more information about setting up the database secrets engine.
+
+## Setup
+
+1.  Enable the database secrets engine if it is not already enabled:
+
+    ```text
+    $ vault secrets enable database
+    Success! Enabled the database secrets engine at: database/
+    ```
+
+    By default, the secrets engine will enable at the name of the engine. To
+    enable the secrets engine at a different path, use the `-path` argument.
+
+1.  Configure Vault with the proper plugin and connection information to access your Redshift database:
+
+    ```text
+    $ vault write database/config/my-redshift-database \
+        plugin_name=redshift-database-plugin \
+        allowed_roles="my-role" \
+        connection_url="postgresql://{{username}}:{{password}}@localhost:5432/<optional: db-name>" \
+        username="root" \
+        password="root"
+    ```
+
+1.  Configure a role that maps a name in Vault to a SQL statement to execute which 
+    creates the database credential:
+
+    ```text
+    $ vault write database/roles/my-role \
+        db_name=my-redshift-database \
+        creation_statements="CREATE USER \"{{name}}\" WITH PASSWORD '{{password}}' VALID UNTIL '{{expiration}}'; \
+            GRANT SELECT ON ALL TABLES IN SCHEMA public TO \"{{name}}\";" \
+        default_ttl="1h" \
+        max_ttl="24h"
+    Success! Data written to: database/roles/my-role
+    ```
+
+## Usage
+
+After the secrets engine is configured and a user/machine has a Vault token with
+the proper permission, it can generate credentials.
+
+1.  Generate a new credential by reading from the `/creds` endpoint with the name
+    of the role:
+
+    ```text
+    $ vault read database/creds/my-role
+    Key                Value
+    ---                -----
+    lease_id           database/creds/my-role/2f6a614c-4aa2-7b19-24b9-ad944a8d4de6
+    lease_duration     1h
+    lease_renewable    true
+    password           8cab931c-d62e-a73d-60d3-5ee85139cd66
+    username           v-root-e2978cd0-
+    ```
+
+## API
+
+The full list of configurable options can be seen in the [Redshift database
+plugin API](/api/secret/databases/redshift) page.
+
+For more information on the database secrets engine's HTTP API please see the
+[Database secrets engine API](/api/secret/databases) page.