Reduce reschedule startup queries in execution API

The run endpoint was doing two TaskReschedule round trips to compute values that can be fetched together. Collapsing them into one aggregate query keeps the rescheduled-sensor context fix intact while trimming overhead on task startup, and the test now proves we return the earliest reschedule timestamp instead of depending on insertion order.

Author: hkc-8010

airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py

@@ -290,17 +290,12 @@ def ti_run(
                 xcom_query = xcom_query.where(XComModel.map_index == map_index)
 
             xcom_keys = list(session.scalars(xcom_query))
-        task_reschedule_count = (
-            session.scalar(
-                select(func.count(TaskReschedule.id)).where(TaskReschedule.ti_id == task_instance_id)
-            )
-            or 0
-        )
-        first_task_reschedule_start_date = None
-        if task_reschedule_count > 0:
-            first_task_reschedule_start_date = session.scalar(
-                select(func.min(TaskReschedule.start_date)).where(TaskReschedule.ti_id == task_instance_id)
-            )
+        task_reschedule_count, first_task_reschedule_start_date = session.execute(
+            select(
+                func.count(TaskReschedule.id),
+                func.min(TaskReschedule.start_date),
+            ).where(TaskReschedule.ti_id == task_instance_id)
+        ).one()
 
         dr.team_name = get_team_name_for_ti(task_instance_id, session)
 

airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py

@@ -344,18 +344,18 @@ def test_ti_run_state_includes_first_task_reschedule_start_date(
         )
         session.add_all(
             [
-                TaskReschedule(
-                    ti_id=ti.id,
-                    start_date=first_reschedule_start_date,
-                    end_date=timezone.datetime(2024, 9, 30, 10, 1),
-                    reschedule_date=timezone.datetime(2024, 9, 30, 10, 2),
-                ),
                 TaskReschedule(
                     ti_id=ti.id,
                     start_date=second_reschedule_start_date,
                     end_date=timezone.datetime(2024, 9, 30, 11, 1),
                     reschedule_date=timezone.datetime(2024, 9, 30, 11, 2),
                 ),
+                TaskReschedule(
+                    ti_id=ti.id,
+                    start_date=first_reschedule_start_date,
+                    end_date=timezone.datetime(2024, 9, 30, 10, 1),
+                    reschedule_date=timezone.datetime(2024, 9, 30, 10, 2),
+                ),
             ]
         )
         session.commit()