Add missing index on task_instance.dag_version_id

[Vamsi-klu]

Summary

• Add ti_dag_version_id index on task_instance.dag_version_id column
• This column is queried by the DAG processor but had no index, causing full table scans on large tables (27M+ rows reported in Dag Processor performance issue querying task_instance table by dag_version_id #61894)
• Include Alembic migration 0107_3_2_0_add_dag_version_id_index_to_ti.py

Co-contributors : @codingrealitylabs @girlcoder-gaming

Test plan

• Verify Alembic migration applies cleanly: airflow db migrate
• Verify Alembic migration downgrades cleanly: airflow db downgrade
• Verify index exists after migration: SELECT * FROM pg_indexes WHERE indexname = 'ti_dag_version_id'
• Monitor scheduler performance improvement on large deployments

Closes: #61894

Note: On large task_instance tables, the index creation may take several minutes during migration.

🤖 Generated with Claude Code

[Vamsi-klu]

cc @ephraimbuddy @potiuk @jason810496 — Would appreciate your review. This adds a missing index on task_instance.dag_version_id to prevent full table scans during DAG processing on large tables.

[Copilot]

Pull request overview

Adds a missing database index on task_instance.dag_version_id to avoid full-table scans that significantly slow down the DAG processor on large task_instance tables.

Changes:

• Add ORM-level index definition ti_dag_version_id on TaskInstance.dag_version_id.
• Add Alembic migration 0061_3_0_0_add_dag_version_id_index_to_ti.py to create/drop the index in existing DBs.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated no comments.

File	Description
airflow/models/taskinstance.py	Declares the new ti_dag_version_id index in the TaskInstance table metadata.
airflow/migrations/versions/0061_3_0_0_add_dag_version_id_index_to_ti.py	Creates the ti_dag_version_id index on upgrade and drops it on downgrade.

Comments suppressed due to low confidence (1)

airflow/migrations/versions/0061_3_0_0_add_dag_version_id_index_to_ti.py:42

• On PostgreSQL, a plain CREATE INDEX will hold stronger locks than CONCURRENTLY and can block writes to task_instance for the duration of the build (which can be minutes+ on very large tables). Consider using a Postgres-specific concurrent index build (via an autocommit block + postgresql_concurrently=True) to reduce downtime risk, or otherwise document/handle the expected locking behavior explicitly in the migration code path.

def upgrade():
    """Add index on dag_version_id to task_instance table."""
    op.create_index("ti_dag_version_id", "task_instance", ["dag_version_id"], unique=False)

[girlcoder-gaming]

@codex please review this PR.