Reduce task group query work in Execution API

[steveahnahn]

Reduce task-group query work in the Execution API.

The task-instance count and state endpoints now resolve task-group task IDs from the serialized Dag and apply those IDs to the task-instance query before executing it. This avoids loading task-instance rows just to build another filter, and avoids the /states task-group path fetching Dag-wide task instances before replacing that result with the group rows.

Why:

• /execution/task-instances/states?task_group_id=... executed one broad task_instance SELECT and one group-scoped task_instance SELECT.
• /execution/task-instances/count?task_group_id=... loaded group task-instance rows before executing the count query.
• The serialized Dag already contains the task IDs needed to apply the group filter directly.

What changed:

• Replace the row-loading task-group helper with a task-ID helper.
• Apply task-group task IDs directly in the count query.
• Build one task-id filter for /states, preserving the union of explicit task_ids and task-group tasks.
• Add regression assertions that each task-group request performs a single task_instance SELECT.

Evidence:

• Reproduction before fix:
  • breeze run pytest airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py::TestGetCount::test_get_count_with_task_group airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py::TestGetTaskStates::test_get_task_states_group_id_basic -q
  • Before the source change, both tests failed with AssertionError: assert 2 == 1.
• Regression test passing after fix:
  • breeze run pytest airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py::TestGetCount::test_get_count_with_task_group airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py::TestGetTaskStates::test_get_task_states_group_id_basic -q
  • 2 passed

Was generative AI tooling used to co-author this PR?

• Yes - Codex (GPT-5)

Generated-by: Codex (GPT-5) following the guidelines