API - Improve public get_task_instances endpoint

Similarly to #62025

On big installation the "get_task_instances" endpoint to list all the dagruns in the UI is taking a long time to return a response.

The critical part of the code is the actual db query. Most likely due to the number of joins creating rows explosion. We need to optimize the query, most likely by double checking the eager loading options and verify that there is no row explosion cause by a wrong option (joinedload vs selectinload). Also we can add load_only to limit the number of columns selected in each joins.

airflow/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py

Lines 483 to 533 in 4e75919

    
           query = ( 
        
               select(TI).join(TI.dag_run).outerjoin(TI.dag_version).options(*eager_load_TI_and_TIH_for_validation()) 
        
           ) 
        
           if dag_run_id != "~": 
        
               dag_run = session.scalar(select(DagRun).filter_by(run_id=dag_run_id)) 
        
               if not dag_run: 
        
                   raise HTTPException( 
        
                       status.HTTP_404_NOT_FOUND, 
        
                       f"DagRun with run_id: `{dag_run_id}` was not found", 
        
                   ) 
        
               query = query.where(TI.run_id == dag_run_id) 
        
           if dag_id != "~": 
        
               dag = get_dag_for_run_or_latest_version(dag_bag, dag_run, dag_id, session) 
        
               query = query.where(TI.dag_id == dag_id) 
        
               if dag: 
        
                   task_group_id.dag = dag 
        
           task_instance_select, total_entries = paginated_select( 
        
               statement=query, 
        
               filters=[ 
        
                   run_after_range, 
        
                   logical_date_range, 
        
                   start_date_range, 
        
                   end_date_range, 
        
                   update_at_range, 
        
                   duration_range, 
        
                   state, 
        
                   pool, 
        
                   pool_name_pattern, 
        
                   queue, 
        
                   queue_name_pattern, 
        
                   executor, 
        
                   task_id, 
        
                   task_display_name_pattern, 
        
                   task_group_id, 
        
                   dag_id_pattern, 
        
                   run_id_pattern, 
        
                   version_number, 
        
                   readable_ti_filter, 
        
                   try_number, 
        
                   operator, 
        
                   operator_name_pattern, 
        
                   map_index, 
        
               ], 
        
               order_by=order_by, 
        
               offset=offset, 
        
               limit=limit, 
        
               session=session, 
        
           ) 
        
           task_instances = session.scalars(task_instance_select)

Also needs investigation to check if an extra index could help.

Screen.Recording.2026-02-16.at.16.28.21.mov

Committer

I acknowledge that I am a maintainer/committer of the Apache Airflow project.

Committer

I acknowledge that I am a maintainer/committer of the Apache Airflow project.

This will also fix #61212 which is similar where are multiple joins to the same query. Some of the things we noticed in our testing but yet to find any solution in our setup using mysql 8.x.

TaskInstance.start_date is not indexed by default but the UI by default sorts by it on load http://localhost:8000/api/v2/dags/~/dagRuns/~/taskInstances?limit=10&offset=0&order_by=-start_date&order_by=-run_after .
TaskInstance.id is a primary key which is a UUID. UUID is not a native type in mysql but is stored by varchar(32). Postgres has native UUID support. Ref https://stackoverflow.com/questions/2365132/uuid-performance-in-mysql .
Removing the nested joins of same table and removing sorting had greatly improved performance but I was not able to update the relevant SQLAlchemy parts in the API endpoint to get there. Also sorting is an important element where users want to know the recent task instances which we are yet to tackle.

I'd like to work on this issue!

@kevinhongzl

@kevinhongzl Please go ahead, you don't need to be assigned to start working on an issue or open a PR.

Optimize DagRun.get_task_instances to avoid redundant joinedload

fix: eliminate duplicate JOINs in get_task_instances endpoint (apache…

tests: replace query-count test with SQL-structure regression test for …

fix: eliminate duplicate JOINs in get_task_instances endpoint (apache…

tests: replace query-count test with SQL-structure regression test for …

fix: eliminate duplicate JOINs in get_task_instances endpoint (#62910)

fix: eliminate duplicate JOINs in get_task_instances endpoint (apache…

Closing for now, PR above did a significant improvement. The bottleneck is now on the Count(*) required for pagination. If we need to go further we will have to probably change the pagination to 'offset' pagination or do some bigger change.

fix: eliminate duplicate JOINs in get_task_instances endpoint (apache…

Navigation Menu

Repository navigation

API - Improve public get_task_instances endpoint #62027

Committer

Committer

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	query = (
	select(TI).join(TI.dag_run).outerjoin(TI.dag_version).options(*eager_load_TI_and_TIH_for_validation())
	)
	if dag_run_id != "~":
	dag_run = session.scalar(select(DagRun).filter_by(run_id=dag_run_id))
	if not dag_run:
	raise HTTPException(
	status.HTTP_404_NOT_FOUND,
	f"DagRun with run_id: `{dag_run_id}` was not found",
	)
	query = query.where(TI.run_id == dag_run_id)
	if dag_id != "~":
	dag = get_dag_for_run_or_latest_version(dag_bag, dag_run, dag_id, session)
	query = query.where(TI.dag_id == dag_id)
	if dag:
	task_group_id.dag = dag

	task_instance_select, total_entries = paginated_select(
	statement=query,
	filters=[
	run_after_range,
	logical_date_range,
	start_date_range,
	end_date_range,
	update_at_range,
	duration_range,
	state,
	pool,
	pool_name_pattern,
	queue,
	queue_name_pattern,
	executor,
	task_id,
	task_display_name_pattern,
	task_group_id,
	dag_id_pattern,
	run_id_pattern,
	version_number,
	readable_ti_filter,
	try_number,
	operator,
	operator_name_pattern,
	map_index,
	],
	order_by=order_by,
	offset=offset,
	limit=limit,
	session=session,
	)

	task_instances = session.scalars(task_instance_select)

API - Improve public get_task_instances endpoint #62027

Description

Committer

Committer

Activity

tirkarthi commented on Feb 16, 2026

kevinhongzl commented on Feb 16, 2026

pierrejeambrun commented on Feb 17, 2026

Arunodoy18 commented on Feb 18, 2026

pierrejeambrun commented on Mar 30, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions