Redshift batch inserts using COPY FROM operation

[brendanstennett]

Description

Fixes #24546

This PR aims to allow the use of COPY FROM statements when sinking data into Redshift.

The Redshift connector inherits BaseJdbcConnector which uses batched INSERT statements to execute sink operations. Even when using non transactional mode, this can only push about 1000 rows per second. This change stages the rows to a parquet file first, then issues a COPY FROM statement to load the table. We are noticing 250K rows per second or more using this method.

This has been running in production for 2+ months on our own branch.

This functionality needs to be enabled by specifying the following config option:

redshift.batched-inserts-copy-location=s3://my-bucket/my-prefix

The following options are also required when specifying the above:

redshift.batched-inserts-copy-iam-role=arn:aws:iam::123456789000:role/redshift_iam_role
s3.region=region
s3.aws-access-key=KEY
s3.aws-secret-key=SECRET

A suggested IAM Policy to for this role and user:

{
	"Version": "2012-10-17",
	"Statement": [
		{
			"Effect": "Allow",
			"Action": [
				"s3:ListBucket"
			],
			"Resource": "arn:aws:s3:::my-bucket"
		},
		{
			"Effect": "Allow",
			"Action": [
				"s3:GetObject",
				"s3:PutObject",
				"s3:DeleteObject"
			],
			"Resource": "arn:aws:s3:::my-bucket/*"
		}
	]
}

Additional context and related issues

• This is designed to work alongside and compliments the new Redshift UNLOAD feature Add support for fetching Redshift query results using Redshift unload command #24117
• When the above configuration is not supplied, it will default to base JDBC behaviour.
• There needs to be a new GHA var added to the repo REDSHIFT_S3_COPY_ROOT similar to REDSHIFT_S3_UNLOAD_ROOT

Release notes

( ) This is not user-visible or is docs only, and no release notes are required.
( ) Release notes are required. Please propose a release note for me.
(x) Release notes are required, with the following suggested text:

## Redshift
* Add support for Redshift COPY FROM statements for batch insert operations ({issue}`24546`)

[brendanstennett]

@ebyhr Added requested documentation. We need the ENV var set on the repo for the tests to pass.

REDSHIFT_S3_COPY_ROOT which would be a similar location to the ENV var already set of REDSHIFT_S3_UNLOAD_ROOT