Alter columns change type (redshift-dialect) with auto generate command .

[Steff94190]

Hi All,

I have some use case when I have to alter type of one or many column on table (with redshift dialect).

In redshift we can't directly user something like this ALTER TABLE table_name ALTER COLUMN column_name datatype;.
We need to do something like

ALTER TABLE <TABLE_NAME> RENAME TO <TABLE_NAME>_OLD;
CREATE TABLE <TABLE_NAME> ( <NEW_COLUMN_DEFINITION> );
INSERT INTO <TABLE_NAME> (<NEW_COLUMN_DEFINITION>)
SELECT <COLUMNS>
FROM <TABLE_NAME>_OLD;
DROP TABLE <TABLE_NAME>_OLD;

I wrote a script that works when one column change (modify_nullabe or modify_type)

python Script:

@writer.rewrites(ops.AlterTableOp)
def alter_table(context, revision, op):
    nb_tbl_occur = TABLE_ALREADY_MANAGE.count(op.table_name)
    if (nb_tbl_occur < 2) and ((op.modify_nullable is not None) or (op.modify_type is not None)):
        old_table_name = op.table_name + "_old"
        rename_table = ops.ExecuteSQLOp(f"""ALTER TABLE {op.schema}.{op.table_name} RENAME TO {old_table_name}""")
        if nb_tbl_occur == 1:  
            METADATA_FROM_DB.reflect(DB_ENGINE, op.schema, False, [op.table_name], False, False)
            metadata_table = METADATA_FROM_DB.tables[f"{op.schema}.{op.table_name}"]
        else: 
            metadata_table = dict(TARGET_METADATA.get(op.schema).tables).get(f"{op.schema}.{op.table_name}")

        metadata_column = metadata_table._columns
        options = redshift_table_option(metadata_table.dialect_options.get("redshift"))

        create_tbl = ops.CreateTableOp(
            table_name=op.table_name,
            columns=metadata_column,
            schema=op.schema,
            comment=metadata_table.comment,
            redshift_distkey=options["distkey"],
            redshift_diststyle=options["diststyle"],
            redshift_ignore_search_path=options["ignore_search_path"],
            redshift_interleaved_sortkey=options["interleaved_sortkey"],
            redshift_sortkey=options["sortkey"],
        )

        str_cols = ", ".join([col.key for col in metadata_column])

        # insert data
        insert_data = ops.ExecuteSQLOp(
            f"""INSERT INTO "{op.schema}"."{op.table_name}" ({str_cols}) 
                    SELECT {str_cols} 
                FROM "{op.schema}"."{old_table_name}" 
            """
        )

        # Drop old table
        drop_table_old = ops.ExecuteSQLOp(f"""DROP TABLE {op.schema}.{old_table_name}""")
        TABLE_ALREADY_MANAGE.append(op.table_name)
        return [
            rename_table,
            create_tbl,
            insert_data,
            drop_table_old
        ]
   elif nb_tbl_occur == 2:
        return

    return op

Result with one column (id) change:

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.execute('ALTER TABLE table_name RENAME TO table_name_old')
    op.create_table('table_name',
    sa.Column('id', sa.SmallInteger(), nullable=False),
    sa.Column('api', sa.String(length=256), nullable=False),
    sa.Column('comsumption_type', sa.String(length=64), nullable=False),
    sa.Column('source', sa.String(length=64), nullable=False),
    sa.Column('market', sa.String(length=64), nullable=True),
    sa.PrimaryKeyConstraint('id'),
    schema='common_layer',
    redshift_distkey=None,
    redshift_diststyle='ALL',
    redshift_ignore_search_path=False,
    redshift_interleaved_sortkey=None,
    redshift_sortkey='id'
    )
    op.execute('INSERT INTO "common_layer"."table_name" (id, api, comsumption_type, source, market) \n                    SELECT id, api, comsumption_type, source, market \n                FROM "common_layer"."table_name_old" \n            ')
    op.execute('DROP TABLE table_name_old')
    # ### end Alembic commands ###


def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.execute('ALTER TABLE table_name RENAME TO table_name_old')
    op.create_table('table_name',
    sa.Column('id', sa.INTEGER(), autoincrement=False, nullable=False),
    sa.Column('api', sa.VARCHAR(length=256), autoincrement=False, nullable=False),
    sa.Column('comsumption_type', sa.VARCHAR(length=64), autoincrement=False, nullable=False),
    sa.Column('source', sa.VARCHAR(length=64), autoincrement=False, nullable=False),
    sa.Column('market', sa.VARCHAR(length=64), autoincrement=False, nullable=True),
    sa.PrimaryKeyConstraint('id'),
    schema='common_layer',
    redshift_distkey=None,
    redshift_diststyle='ALL',
    redshift_ignore_search_path=False,
    redshift_interleaved_sortkey=None,
    redshift_sortkey=('id',)
    )
    op.execute('INSERT INTO "common_layer"."table_name" (id, api, comsumption_type, source, market) \n                    SELECT id, api, comsumption_type, source, market \n                FROM "common_layer"."table_name_old" \n            ')
    op.execute('DROP TABLE table_name_old')
    # ### end Alembic commands ###

Result is good. but when I change many column (id and api) I have something like this

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.execute('ALTER TABLE table_name RENAME TO table_name_old')
    op.create_table('table_name',
    sa.Column('id', sa.SmallInteger(), nullable=False),
    sa.Column('api', sa.String(length=3), nullable=False),
    sa.Column('comsumption_type', sa.String(length=64), nullable=False),
    sa.Column('source', sa.String(length=64), nullable=False),
    sa.Column('market', sa.String(length=64), nullable=True),
    sa.PrimaryKeyConstraint('id'),
    schema='common_layer',
    comment='Matching between Argus user application and cleaned API names',
    redshift_distkey=None,
    redshift_diststyle='ALL',
    redshift_ignore_search_path=False,
    redshift_interleaved_sortkey=None,
    redshift_sortkey='id'
    )
    op.execute('INSERT INTO "common_layer"."table_name" (id, api, comsumption_type, source, market) \n                    SELECT id, api, comsumption_type, source, market \n                FROM "common_layer"."table_name_old" \n            ')
    op.execute('DROP TABLE table_name_old')
    op.execute('ALTER TABLE table_name RENAME TO table_name_old')
    op.create_table('table_name',
    sa.Column('id', sa.INTEGER(), autoincrement=False, nullable=False),
    sa.Column('api', sa.VARCHAR(length=256), autoincrement=False, nullable=False),
    sa.Column('comsumption_type', sa.VARCHAR(length=64), autoincrement=False, nullable=False),
    sa.Column('source', sa.VARCHAR(length=64), autoincrement=False, nullable=False),
    sa.Column('market', sa.VARCHAR(length=64), autoincrement=False, nullable=True),
    sa.PrimaryKeyConstraint('id'),
    schema='common_layer',
    comment='Matching between Argus user application and cleaned API names',
    redshift_distkey=None,
    redshift_diststyle='ALL',
    redshift_ignore_search_path=False,
    redshift_interleaved_sortkey=None,
    redshift_sortkey=('id',)
    )
    op.execute('INSERT INTO "common_layer"."table_name" (id, api, comsumption_type, source, market) \n                    SELECT id, api, comsumption_type, source, market \n                FROM "common_layer"."table_name_old" \n            ')
    op.execute('DROP TABLE table_name_old')
    # ### end Alembic commands ###


def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    pass
    # ### end Alembic commands ###

I want the same result like previous result.

Do you have an idea ?
thank you

[zzzeek]

hi -

the sequence you deseribe is supported by Alembic batch mode . All of the "rename table" / "select into" stuff is fully implemented already. hope this helps.

[franziskakuehn]

is redshift column order change with alembic migrations in a stable version?

with op.batch_alter_table("unified_la_stellen", schema=latotal_consumption_schema_name, recreate="always") as batch_op:
    batch_op.add_column(sa.Column('v_ist_test', sa.String()), insert_after='region')

with sqlalchemy<2.0.0 for
sqlalchemy-redshift==0.8.14
and alembic>1.10 I get

c8984929f584_test_add_col.py:32: in upgrade
    batch_op.add_column(sa.Column('v_ist_test', sa.String()), insert_after='region')
../../.pyenv/versions/3.9.18/lib/python3.9/contextlib.py:126: in __exit__
    next(self.gen)
.venv/lib/python3.9/site-packages/alembic/operations/base.py:398: in batch_alter_table
    impl.flush()
.venv/lib/python3.9/site-packages/alembic/operations/batch.py:164: in flush
    batch_impl._create(self.impl)
.venv/lib/python3.9/site-packages/alembic/operations/batch.py:446: in _create
    op_impl.prep_table_for_batch(self, self.table)
.venv/lib/python3.9/site-packages/alembic/ddl/postgresql.py:103: in prep_table_for_batch
    self.drop_constraint(constraint)
.venv/lib/python3.9/site-packages/alembic/ddl/impl.py:350: in drop_constraint
    self._exec(schema.DropConstraint(const))
.venv/lib/python3.9/site-packages/alembic/ddl/impl.py:207: in _exec
    return conn.execute(construct, multiparams)
.venv/lib/python3.9/site-packages/sqlalchemy/engine/base.py:1385: in execute
    return meth(self, multiparams, params, _EMPTY_EXECUTION_OPTS)
.venv/lib/python3.9/site-packages/sqlalchemy/sql/ddl.py:80: in _execute_on_connection
    return connection._execute_ddl(
.venv/lib/python3.9/site-packages/sqlalchemy/engine/base.py:1474: in _execute_ddl
    compiled = ddl.compile(
.venv/lib/python3.9/site-packages/sqlalchemy/sql/elements.py:503: in compile
    return self._compiler(dialect, **kw)
.venv/lib/python3.9/site-packages/sqlalchemy/sql/ddl.py:32: in _compiler
    return dialect.ddl_compiler(dialect, self, **kw)
.venv/lib/python3.9/site-packages/sqlalchemy/sql/compiler.py:464: in __init__
    self.string = self.process(self.statement, **compile_kwargs)
.venv/lib/python3.9/site-packages/sqlalchemy/sql/compiler.py:499: in process
    return obj._compiler_dispatch(self, **kwargs)
.venv/lib/python3.9/site-packages/sqlalchemy/ext/compiler.py:548: in <lambda>
    lambda *arg, **kw: existing(*arg, **kw),
.venv/lib/python3.9/site-packages/sqlalchemy/ext/compiler.py:604: in __call__
    expr = fn(element, compiler, **kw)
.venv/lib/python3.9/site-packages/sqlalchemy/ext/compiler.py:530: in _wrap_existing_dispatch
    return existing_dispatch(element, compiler, **kw)
.venv/lib/python3.9/site-packages/sqlalchemy/sql/visitors.py:82: in _compiler_dispatch
    return meth(self, **kw)
.venv/lib/python3.9/site-packages/sqlalchemy/sql/compiler.py:4814: in visit_drop_constraint
    formatted_name = self.preparer.format_constraint(constraint)
.venv/lib/python3.9/site-packages/sqlalchemy/sql/compiler.py:5515: in format_constraint
    return self.truncate_and_render_constraint_name(
.venv/lib/python3.9/site-packages/sqlalchemy/sql/compiler.py:5539: in truncate_and_render_constraint_name
    return self._truncate_and_render_maxlen_name(
.venv/lib/python3.9/site-packages/sqlalchemy/sql/compiler.py:5553: in _truncate_and_render_maxlen_name
    return self.quote(name)
.venv/lib/python3.9/site-packages/sqlalchemy/sql/compiler.py:5452: in quote
    if self._requires_quotes(ident):
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <sqlalchemy_redshift.dialect.RedshiftIdentifierPreparer object at 0x10537b220>
value = ''

    def _requires_quotes(self, value):
        """Return True if the given identifier requires quoting."""
        lc_value = value.lower()
        return (
            lc_value in self.reserved_words
>           or value[0] in self.illegal_initial_characters
            or not self.legal_characters.match(util.text_type(value))
            or (lc_value != value)
        )
E       IndexError: string index out of range

.venv/lib/python3.9/site-packages/sqlalchemy/sql/compiler.py:5361: IndexError

[franziskakuehn]

if a table is created with alembic operations in redshift without primary key, alembic set a primary key constraint per default without recognising naming conventions set in target metadata in env.py
due to this the above error occurs by trying to drop the primary key constraint without a name during batchoperation.add_column and order recognition. constraint.name = ''

if a table with a primary key constraint gets another column over batch operation with column order recognition with or without naming conventions (because redshift sets a constraint name) this error is not thrown because name is not ''

If I just want to alter the order of an existing column over batchoperations.alter_column('col', insert_after='col1')) this is ignored completely.