def METHOD_NAME(ds, **kwargs):
    conn_id = kwargs.get('conn_id')
    file_format = kwargs.get('file_format')
    task = kwargs.get('task')
    pg_hook = PostgresHook(conn_id)
    table_partition_keys = {
        'blocks': 'number',
        'contracts': 'address',
        'logs': 'block_number',
        'receipts': 'block_number',
        'token_transfers': 'block_number',
        'tokens': 'address',
        'traces': 'block_number',
        'transactions': 'block_number'
    }
    sql = """
        DROP TABLE IF EXISTS {schema}.{table}_copy_tmp;
        CREATE TABLE {schema}.{table}_copy_tmp
        (LIKE {schema}.{table});
    """
    if file_format == 'csv':
        sql += """
            COPY {schema}.{table}_copy_tmp
            FROM 's3://{output_bucket}/export/{table}/block_date={date}/{table}.{file_format}'
            WITH CREDENTIALS
            'aws_access_key_id={aws_access_key_id};aws_secret_access_key={aws_secret_access_key}'
            TRUNCATECOLUMNS BLANKSASNULL EMPTYASNULL IGNOREHEADER 1 CSV;
        """
    elif file_format == 'json':
        sql += """
            COPY {schema}.{table}_copy_tmp
            FROM 's3://{output_bucket}/export/{table}/block_date={date}/{table}.{file_format}'
            WITH CREDENTIALS
            'aws_access_key_id={aws_access_key_id};aws_secret_access_key={aws_secret_access_key}'
            TRUNCATECOLUMNS BLANKSASNULL EMPTYASNULL JSON 'auto';
        """
    else:
        raise ValueError('Only json and csv file formats are supported.')
    sql += """
        BEGIN TRANSACTION;
        DELETE FROM {schema}.{table}
        USING {schema}.{table}_copy_tmp
        WHERE
          {schema}.{table}.{partition_key} = {schema}.{table}_copy_tmp.{partition_key};
        INSERT INTO {schema}.{table}
        SELECT * FROM {schema}.{table}_copy_tmp;
        END TRANSACTION;
        DROP TABLE {schema}.{table}_copy_tmp;
    """
    formatted_sql = sql.format(
        schema=chain,
        table=task,
        partition_key=table_partition_keys[task],
        output_bucket=output_bucket,
        date=ds,
        file_format=file_format,
        aws_access_key_id=aws_access_key_id,
        aws_secret_access_key=aws_secret_access_key
    )
    pg_hook.run(formatted_sql)