Function upload_arrow_table_to_redshift

sdk/python/feast/infra/utils/aws_utils.py:290–390

Uploads an Arrow Table to Redshift to a new or existing table. Here's how the upload process works: 1. PyArrow Table is serialized into a Parquet format on local disk 2. The Parquet file is uploaded to S3 3. The S3 file is uploaded to Redshift as a new table through COPY

(
    table: Union[pyarrow.Table, Path],
    redshift_data_client,
    cluster_id: Optional[str],
    workgroup: Optional[str],
    database: str,
    user: Optional[str],
    s3_resource,
    iam_role: str,
    s3_path: str,
    table_name: str,
    schema: Optional[pyarrow.Schema] = None,
    fail_if_exists: bool = True,
)

def upload_arrow_table_to_redshift(
    table: Union[pyarrow.Table, Path],
    redshift_data_client,
    cluster_id: Optional[str],
    workgroup: Optional[str],
    database: str,
    user: Optional[str],
    s3_resource,
    iam_role: str,
    s3_path: str,
    table_name: str,
    schema: Optional[pyarrow.Schema] = None,
    fail_if_exists: bool = True,
):
    """Uploads an Arrow Table to Redshift to a new or existing table.

    Here's how the upload process works:
        1. PyArrow Table is serialized into a Parquet format on local disk
        2. The Parquet file is uploaded to S3
        3. The S3 file is uploaded to Redshift as a new table through COPY command
        4. The local disk & s3 paths are cleaned up

    Args:
        redshift_data_client: Redshift Data API Service client
        cluster_id: Redshift Cluster Identifier
        workgroup: Redshift Serverless Workgroup
        database: Redshift Database Name
        user: Redshift username
        s3_resource: S3 Resource object
        s3_path: S3 path where the Parquet file is temporarily uploaded
        iam_role: IAM Role for Redshift to assume during the COPY command.
                  The role must grant permission to read the S3 location.
        table_name: The name of the new Redshift table where we copy the dataframe
        table: The Arrow Table or Path to parquet dataset to upload
        schema: (Optionally) client may provide arrow Schema which will be converted into redshift table schema
        fail_if_exists: fail if table with such name exists or append data to existing table

    Raises:
        RedshiftTableNameTooLong: The specified table name is too long.
    """
    if len(table_name) > REDSHIFT_TABLE_NAME_MAX_LENGTH:
        raise RedshiftTableNameTooLong(table_name)

    if isinstance(table, pyarrow.Table) and not schema:
        schema = table.schema

    if not schema:
        raise ValueError("Schema must be specified when data is passed as a Path")

    bucket, key = get_bucket_and_key(s3_path)

    column_query_list = ", ".join(
        [f"{field.name} {pa_to_redshift_value_type(field.type)}" for field in schema]
    )

    uploaded_files = []

    if isinstance(table, Path):