Skip to content

Reinstate tih.id as pkey and auto-increment on downgrade #58149

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Chais wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Reinstate tih.id as pkey and auto-increment on downgrade #58149

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0071913
Reinstate tih.id as pkey and auto-increment on downgrade
Nov 10, 2025
6cb636f
Merge branch 'main' into 7645189f3479-downgrade-fix
Chais Nov 17, 2025
77097d1
add_column does not support primary_key so execute instead
Jan 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 7 additions & 27 deletions airflow-core/src/airflow/migrations/versions/0060_3_0_0_add_try_id_to_ti_and_tih.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,34 +127,14 @@ def upgrade():

def downgrade():
"""Unapply Add try_id to TI and TIH."""
dialect_name = op.get_bind().dialect.name
with op.batch_alter_table("task_instance_history", schema=None) as batch_op:
batch_op.drop_constraint(batch_op.f("task_instance_history_pkey"), type_="primary")
batch_op.add_column(sa.Column("id", sa.INTEGER, nullable=True))
match batch_op.get_bind().dialect:
case "mysql":
batch_op.execute("ALTER TABLE task_instance_history ADD COLUMN id INTEGER PRIMARY KEY AUTO_INCREMENT;")
case "postgresql":
batch_op.execute("ALTER TABLE task_instance_history ADD COLUMN id SERIAL PRIMARY KEY;")
case "sqlite":
batch_op.execute("ALTER TABLE task_instance_history ADD COLUMN id INTEGER PRIMARY KEY AUTOINCREMENT;")
batch_op.drop_column("task_instance_id")
if dialect_name == "postgresql":
op.execute(
"""
ALTER TABLE task_instance_history ADD COLUMN row_num SERIAL;
UPDATE task_instance_history SET id = row_num;
ALTER TABLE task_instance_history DROP COLUMN row_num;
"""
)
elif dialect_name == "mysql":
op.execute(
"""
SET @row_number = 0;
UPDATE task_instance_history
SET id = (@row_number := @row_number + 1)
ORDER BY try_id;
"""
)
else:
op.execute("""
UPDATE task_instance_history
SET id = (SELECT COUNT(*) FROM task_instance_history t2 WHERE t2.id <= task_instance_history.id);
""")
with op.batch_alter_table("task_instance_history", schema=None) as batch_op:
batch_op.alter_column("id", nullable=False, existing_type=sa.INTEGER)
batch_op.drop_column("try_id")
batch_op.create_primary_key("task_instance_history_pkey", ["id"])