[ColtenOuO]

Sumarry

GET /execution/task-instances/states (get_task_instance_states in airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py) is called by the worker/Task SDK whenever a task needs the states of sibling task instances (e.g. trigger-rule evaluation, wait_for_downstream, cross-task state checks). When the request is scoped with task_group_id, the handler was doing unnecessary DB work on every call:

1. It first ran a query selecting every TaskInstance row for the Dag (optionally narrowed by task_ids, logical_dates, run_ids, map_index), loading full ORM entities.
2. It then ran a second query to resolve the task group's members and fetch their TaskInstance rows.
3. If the caller had not also passed task_ids, the result of the first query was thrown away entirely and only the second query's rows were used — meaning step 1 was pure waste on every task-group-scoped call.

For Dags with many tasks and/or many Dag runs, that first discarded query could return and hydrate a large number of rows for nothing, adding an avoidable round trip and CPU/memory cost to a request that sits on the task execution hot path (workers call this while running tasks, so the extra latency is paid per task, not once).

This PR removes the wasted round trip:

• The task group's task IDs are now resolved from the already-parsed Dag definition (task_group.iter_tasks()), which requires no extra TaskInstance query — only the existing Dag-version lookup that was already being performed.
• Those IDs are unioned with any explicit task_ids from the request before building the query, so there is exactly one TaskInstance query per request, filtered with task_id.in_(...) — preserving the exact original semantics (task_ids alone, task_group_id alone, or the union of both).
• That single query now selects only the four columns actually used (run_id, task_id, map_index, state) instead of loading full TaskInstance ORM objects, cutting the amount of data hydrated per call further.
• The pre-existing _get_group_tasks helper (still used by the sibling /count endpoint, which needs actual map_index values from persisted rows for a different query shape) is left untouched aside from factoring out the shared "resolve the task group or 404" logic into _get_task_group.

Change

• get_task_instance_states: replaced the "query everything, then maybe discard it" flow with a single TaskInstance query. The task group's task IDs are resolved up front (from task_group.iter_tasks()) and unioned with any request task_ids into one task_id.in_(...) filter before the query runs, instead of running one query filtered by task_ids and a second, separate query for the group, then conditionally throwing the first result set away.
• That single query now does select(TI.run_id, TI.task_id, TI.map_index, TI.state) instead of select(TI), so it loads four scalar columns per row instead of hydrating full TaskInstance ORM objects.
• Added _get_task_group (Dag/task-group lookup + 404 handling) factored out of the pre-existing _get_group_tasks, and a new _get_group_task_ids that returns just the group's task IDs without touching the TaskInstance table. _get_group_tasks itself is unchanged in behavior — it's still used as-is by the /count endpoint, which needs real persisted (task_id, map_index) pairs for a different query shape.
• Added test_get_task_states_with_task_group_id_query_count (query-count regression test).

Impact

For any task_group_id-scoped call to this endpoint, DB round trips drop from 2 (one of which was always discarded) down to 1, and that query is narrower (4 scalar columns instead of full-row ORM hydration). This directly reduces the RTT the worker/Task SDK pays on this call, and removes the redundant row hydration/network cost on the API server and DB — proportional to Dag size and Dag-run count, so the win scales with the DAGs that would have hit this path hardest.

Testing

• Added test_get_task_states_with_task_group_id_query_count, a query-count regression test using assert_queries_count, which fails against the old code (3 queries) and passes against the fix (2 queries: one Dag-version lookup, one TaskInstance query).
• All existing tests in TestGetTaskStates and TestGetCount (the /count endpoint that shares the untouched helper) continue to pass, confirming the task_ids/task_group_id union semantics are unchanged, including the mixed task_ids + task_group_id case.

---

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

• Yes — Claude Code (Sonnet 5)

[ColtenOuO]

Resolved the conflict :D

[potiuk]

Hello @ColtenOuO - thank you for your contributions to Apache Airflow!

The Airflow community has introduced a limit of 5 open pull requests at a time for contributors without write access to the repository. You currently have 19 open pull requests, so - as a one-time step of introducing the limit - we closed the ones where maintainers have not engaged yet:

• [AIP-94] Add airflowctl Tasks List Command #70472 - [AIP-94] Add airflowctl Tasks List Command
• Let HookToolset retry a failed tool call instead of failing the task #71853 - Let HookToolset retry a failed tool call instead of failing the task
• Fix scheduler crash when the API server response is a redirect #71962 - Fix scheduler crash when the API server response is a redirect
• Fix manually triggered runs blocking a scheduled run at the same logical_date #72003 - Fix manually triggered runs blocking a scheduled run at the same logical_date
• Fix mismatched XCom values when a dynamically mapped task has a gap #72007 - Fix mismatched XCom values when a dynamically mapped task has a gap
• Refactor schema version checks to share target-branch and worktree logic #72070 - Refactor schema version checks to share target-branch and worktree logic
• Document custom OTel metrics endpoint path support #72073 - Document custom OTel metrics endpoint path support
• Batch XCom lookups when resolving a mapped task's expand() arguments #72107 - Batch XCom lookups when resolving a mapped task's expand() arguments
• Remove redundant full-Dag TaskInstance query from the execution API task-states endpoint #72270 - Remove redundant full-Dag TaskInstance query from the execution API task-states endpoint
• Fix asset dependency graph dropping outlets for tasks that both consume and produce #72272 - Fix asset dependency graph dropping outlets for tasks that both consume and produce
• Make supervisor split-line stdout test actually cover the split in Task SDK #72585 - Make supervisor split-line stdout test actually cover the split in Task SDK
• Fix GET /assets/{asset_id} returning 500 when asset events share a timestamp #72599 - Fix GET /assets/{asset_id} returning 500 when asset events share a timestamp
• Show detailed API error messages in UI toasters #72831 - Show detailed API error messages in UI toasters

These pull requests stay open because maintainers are already engaged in them - they count towards your limit:

• Fix output type loss in AgentOperator's human-in-the-loop review #70132 - Fix output type loss in AgentOperator's human-in-the-loop review
• Decode base64 for bytes-typed HookToolset parameters #70433 - Decode base64 for bytes-typed HookToolset parameters
• Allow providers to ship testing Dags separately from example Dags #70541 - Allow providers to ship testing Dags separately from example Dags
• Reject unknown update_mask fields instead of silently ignoring them #71003 - Reject unknown update_mask fields instead of silently ignoring them
• Speed up clearing downstream tasks on large Dags #71203 - Speed up clearing downstream tasks on large Dags
• Add deadline alert coverage to Dag endpoint tests #72296 - Add deadline alert coverage to Dag endpoint tests

This is not a judgement of you or of your changes. We never told contributors before that opening many pull requests at once was a problem, so there is nothing to feel bad about - and nothing is lost: your branches, commits and the review history stay where they are.

What we ask you to do is to make your first prioritization decision: choose which of the pull requests above matter most to you, and reopen them (up to 5 open at a time, including the ones still open) with the "Reopen pull request" button or gh pr reopen <PR_NUMBER> --repo apache/airflow. Reopen the ones you are ready to follow through - keep them rebased, respond to review comments and fix failing checks.

While your pull requests are waiting for review, the most valuable thing you can do is help in other ways - reviewing other contributors' pull requests, helping with issues, and taking part in the discussions on the devlist and Slack.

Why we introduced the limit, what it means for you and how to reopen or restore a pull request is explained in https://github.com/apache/airflow/blob/main/contributing-docs/32_open_pull_request_limit.rst.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting