bingqin2 commented on code in PR #73086:
URL: https://github.com/apache/airflow/pull/73086#discussion_r4005997668
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -1361,6 +1353,61 @@ def _get_group_tasks(
return group_tasks
+def _get_group_task_ids(
+ dag_id: str,
+ task_group_id: str,
+ session: SessionDep,
+ dag_bag: DagBagDep,
+ logical_dates=None,
+ run_ids=None,
+) -> set[str]:
+ """
+ Return the ids of the tasks that make up a task group.
+
+ When the request names Dag runs, the group is resolved against the Dag
version each of those
+ runs resolves to (the version a run of a versioned bundle was created
from, the latest version
+ otherwise), so a group renamed or removed after a run was created is still
found for that run,
+ and a group that only exists in a newer version is not. Without a named
run, or while none of
+ the named runs exists yet, the latest version answers.
+ """
+ dags: list[SerializedDAG] = []
+ if logical_dates or run_ids:
+ runs = session.scalars(
+ select(DR).where(
+ DR.dag_id == dag_id,
+ *([DR.logical_date.in_(logical_dates)] if logical_dates else
[]),
+ *([DR.run_id.in_(run_ids)] if run_ids else []),
+ )
+ ).all()
+ # One lookup per distinct version: a run of a versioned bundle
resolves to the version it
+ # was created from, every other run resolves to the latest one.
+ representatives: dict[UUID | None, DR] = {}
+ for run in runs:
+ key = run.created_dag_version_id if run.bundle_version and
run.created_dag_version_id else None
+ representatives.setdefault(key, run)
+ for run in representatives.values():
Review Comment:
Agreed, thanks. The group is now resolved per run:
`_get_dags_for_named_runs` maps each existing named run to the version it
resolves to (one deserialization per distinct version),
`_get_group_task_ids_by_run` keeps the task ids per run, and `_get_group_tasks`
filters with `or_(and_(TI.run_id == run_id, TI.task_id.in_(ids)))`, so a task
that is part of the group in one version is never attributed to a run of
another. The latest version only answers when no named run exists. The count
endpoint had the same leak one layer up: it reduced the group instances to
`(task_id, map_index)` pairs and matched them across every named run, so it now
matches `(run_id, task_id, map_index)`. New tests cover the
`prefix_group_id=False` case you describe on both endpoints (task2 outside the
group in v1 and inside it in v2: 3 instances for the two runs, not 4).
The same per-run resolution now also gives the provider side (#73085) the
answer it needs: `task-instances/states` returns 404 for a `task_ids` entry
that no named run's version defines and no named run has an instance of. A task
the version defines counts before its instance exists (the reparse window the
scheduler's `_verify_integrity_if_dag_changed` leaves), and an instance counts
on its own (a task removed in a newer version), so existing callers that ask
about their own task are unaffected. Runs that do not exist are skipped,
following the point on #67832 that nothing is known before a run exists.
`count` is left alone, since the task runner uses it for mapped-task counts.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
