Vamsi-klu commented on code in PR #73086:
URL: https://github.com/apache/airflow/pull/73086#discussion_r4002487614
##########
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:
task_ids is the union of every representative version. _get_group_tasks then
does TI.task_id.in_(task_ids) and TI.run_id.in_(run_ids). A request that names
runs from two versions, or a group with prefix_group_id=False, can attribute a
task that belongs to version B's group to version A's run.
Resolve per run, or intersect. Do not union then filter.
--
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]
