You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by bb...@apache.org on 2026/10/14 22:43:50 UTC
(airflow) branch main updated: Upgrade Task Loop UX (#74671)
This is an automated email from the ASF dual-hosted git repository.
bbovenzi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 42945d62b1f Upgrade Task Loop UX (#74671)
42945d62b1f is described below
commit 42945d62b1f6ecbb8255b306c9becf7c2900dfef
Author: Brent Bovenzi <br...@astronomer.io>
AuthorDate: Wed Oct 14 18:43:41 2026 -0400
Upgrade Task Loop UX (#74671)
* Filter task instances by loop and iteration
Regions are how the server locates an execution; they are not something a user should
have to know to narrow a table. The loop filter is expressed the way the Dag was
written -- the loop's task group id and the pass number -- and the server resolves that
to the regions behind it.
Loop and iteration are one control rather than two. An iteration index means nothing
until you say which loop counts it, so the two could previously be set independently
into combinations that matched nothing: a pass without a loop, or a pass the loop never
reached. Both now sit behind a single pill whose options come from the loop's own
summary, so only passes that exist are offered, and removing the pill clears both
params. This follows the Run state pill, which already spreads one value over several
URL params.
The same options render as a strip above the table, because a loop's shape -- how many
passes it took and how they went -- is what you came to the page for, and a menu you
have to open is a poor place for it. Each chip is also the control that narrows to it,
and clicking the active one clears it. Past ten passes the chips collapse into the rich
dropdown, the threshold task tries already settled on.
A looped group no longer has a tab of its own. Its page is its task instances, the same
as any other group, with the strip and filter carrying the loop vocabulary. The strip
is scoped to the loop the group belongs to rather than every loop in the Dag; group ids
are dotted paths, so a group nested inside a loop reports the loop enclosing it.
Grid links now carry the newest region coordinate. A task inside a loop has one row per
iteration, so an unscoped link was ambiguous and the API rejected it with "a loop
producer requires an explicit scope".
isRegional answers whether a task instance carries a region; it is not the same question
as whether a loop produced it, and a mapped expansion is regional without being in a
loop. Cross-run scope and the mapped-tasks tab ask isLoopTaskInstance instead, backed by
the in_loop the schema already reports.
Affected-task selection goes back to being keyed by coordinate. Keying it by row id
meant an exclusion the user ticked was lost when a retry came back under a new id -- the
case its own test describes.
The filter registry no longer fetches a Dag structure for every filter bar in the app;
the page that knows which Dag is in view supplies the loops.
Drops the translation keys the rewrite orphaned.
Co-Authored-By: Claude <no...@anthropic.com>
* Select a loop iteration from the breadcrumb
A pass is part of where you are rather than a filter you applied -- the same task id means a
different run of it in every pass -- so it becomes its own breadcrumb level, built on the
switcher the Dag and run levels already use. A task group can be read whole and keeps an
"all" entry; a task instance is one row and must name a pass. Picking one moves the region
coordinate as well as the filter, so the page shows that pass rather than merely narrowing
to it, and a pass carried over from a different loop is ignored instead of being read as a
position in this one.
The task instances list loses its iteration strip. The loop pill already offers every loop
in scope with its passes, so the strip was a second way to say the same thing.
The loop filter now survives navigation. Coordinates pin one execution and must not follow
you to another, but a pass is where you are in a loop, so it stays as you move between the
loop's tasks; a page outside that loop ignores it. Task instances can be ordered by
iteration, and carry a Loop column so a list spanning loops says which one each row is from.
The gantt outlines the pinned pass, over the whole group or the single row depending on
which is being read, and clicking a bar selects the pass it was drawn for. Graph links carry
the newest region coordinate for the same reason grid links do: a task inside a loop has one
row per pass, so an unscoped link is ambiguous and the API rejects it.
Checkbox options move to the left of a dialog footer, away from Cancel and Confirm. This
replaces a margin hack in the clear dialog and gives the other dialogs the same shape.
Also fixes clearing a task instance, which had been failing silently: the dialog closed
itself as soon as the request was sent, and unmounting took its mutation observer with it,
so neither the error toast nor the query invalidation ever ran. Closing is left to the
mutation's success path.
Co-Authored-By: Claude <no...@anthropic.com>
* Address review: drop the dead execution changes and the broken sort
The UI no longer calls getExecution, so the route keeps the shape it had on main rather
than gaining order_by, in_loop and region nesting that nothing reaches. Its translations
go with it.
Ordering by iteration is withdrawn. A row's own region index is its map slot once a mapped
task runs inside a loop, so it does not order by what the column shows, and the enclosing
pass is an ancestry walk rather than a column. Sorting on region_index while the cursor
reported the pass would page wrongly, which is worse than the raise it replaced.
Clearing is fixed on main by #74725: the confirmation dialog was dismissed twice over by
the layer stack, not by its mutation observer being torn down. This drops the change that
guessed otherwise, which also reopened the main dialog mid-clear on the gated path.
A link that moves the coordinate now clears the pass it leaves behind, so the breadcrumb
and the Gantt cannot name a pass the page is not on. The loop filter is sent only with a
run, since the endpoint rejects it without one. The grid structure is not requested
without a Dag, and the task instance stream is only materialised for a Dag with loops.
Also removes two imports of a module this branch deleted, which were failing static checks.
Co-Authored-By: Claude <no...@anthropic.com>
* Address review: structure-based loop scope, polling, and an exact region filter
Loop membership is read from the Dag structure rather than the task id. A loop declared
with prefix_group_id=False does not name its members after itself, so matching on the
prefix lost them the breadcrumb, the filter and the Gantt highlight.
region_id goes back to naming one region exactly, and stops rejecting the cross-run form.
The logical scope of a pass -- the loop's own region plus the expansions nested in it --
is what loop_id and iteration are for, so the two filters keep one meaning each. The
coverage for that pair comes back alongside the region case, including the two guards.
The loop options follow the run the way the loop summary does, so a running loop's passes
and their states no longer sit stale behind a five minute staleTime.
Also fixes the task instance stream being drained before the summaries were built: without
loops it is walked once, and the pass that looks for the newest iteration has nothing to
find, so it is skipped rather than consuming it.
Co-Authored-By: Claude <no...@anthropic.com>
* Take a row's loop iteration from the server, and count every pass
The breadcrumb worked out which pass a page was on by reading region_index off the url.
For a mapped expansion that index is the map slot rather than the pass, so it searched the
pass list for a slot number, found nothing, and dropped the iteration level entirely --
leaving a mapped task inside a loop with no way to see or change its pass. A row already
knows its own pass: loop_iteration is resolved server side and holds for an expansion and
for a pass a clear forked into a successor region alike. Only a group, which is not one
row, still reads the url.
Map index and iteration are separate levels and now stay that way, rather than one url
field being asked to mean both.
State counts cover every pass. A looped task's cell stands for the work the loop did and
the UI says how many iterations that covers, so an earlier pass put back to work by a
clear is visible instead of hidden behind a finished later one. The coordinate still comes
from the newest pass, since a link has to address a row.
Co-Authored-By: Claude <no...@anthropic.com>
* Resolve a loop pass to the row it became, instead of guessing a coordinate
Switching iteration on a task instance page wrote the loop's region and the pass number as
if they addressed a row. They do not: a mapped expansion lives in its own region, and a
clear forks later passes into a successor one. The pair named no row, and asking for a
looped task instance that way is rejected rather than answered empty, so navigating
between a loop's tasks could fail outright.
The switcher now writes only what it knows -- the loop and the pass. The page resolves
that to a coordinate through the list endpoint, which already takes the pair and returns
each row's own region, and writes the answer back into the url. Everything downstream
keeps reading the url unchanged, and the resolved link is the one worth sharing.
The lookup sits on the page rather than in the shared coordinate hook, which nineteen
components use: a query there would have every one of them make a request.
The page holds its children while the lookup is in flight. The request has to be withheld
rather than made and handled, since there is no scope to make it with.
Co-Authored-By: Claude <no...@anthropic.com>
* Open the Gantt highlight where one run's segments meet
An outline rings all four sides, so a run drawn as a queued stretch followed by its state
showed the highlight down the seam between them -- an edge inside the thing being
highlighted rather than its boundary. The sides are drawn individually now, and the two
that already know they are touching a neighbour are left off.
Co-Authored-By: Claude <no...@anthropic.com>
* Address review: group membership, paging, and the loop a summary is asked for
The Gantt group highlight decided membership from the task id, so a group declared with
prefix_group_id=False lost it. It reads the structure now, the way the loop scope already
does.
Switching pass restarts paging. A page of one pass says nothing about where the next one
begins, so carrying its cursor across left the table on rows from the pass before.
A task instance page no longer offers its own row's region as the loop to summarise. Only
a group page's region names an invocation: a row's region is an expansion, or a pass a
clear forked, and asking for a run summary by one of those is rejected -- which left the
header falling back on a count of itself.
Co-Authored-By: Claude <no...@anthropic.com>
* Highlight only a pass the reader chose in the Gantt
The highlight fell back to the region index when no iteration was named. That index is the
map slot on a mapped expansion, so it rang bars of whatever pass happened to share the
number -- and resolving a pass to a row now writes a region index into the url far more
often, which made the fallback fire where it used to be rare.
Co-Authored-By: Claude <no...@anthropic.com>
* Keep earlier attempts on the Gantt chart after a clear
Every attempt is its own row and only the newest stays live, and ORM selects read live rows
unless they ask otherwise. The chart did not, so clearing a task took its earlier tries off
the chart entirely -- including the attempt that the pending-retry filter just above
deliberately drops the live row in favour of, which could never have been there to find.
Co-Authored-By: Claude <no...@anthropic.com>
* Name the try in the Gantt tooltip
The try sat inside a branch keyed on start_date, but the chart reports min_start_date, so
it never reached a bar's tooltip at all. It is read off the shape a caller passes now,
rather than from one of the branches below it.
Shown on a first try too. Clearing a loop pass regenerates the later ones as fresh
executions that start again at try 1, so two neighbouring bars can both be a first try and
the number is what tells them apart.
Co-Authored-By: Claude <no...@anthropic.com>
* Drop the iteration label from the mapped task instances header
The counts beside it are the task's across the whole loop, so naming one pass above them
said that pass had every expansion of every pass. This page lists the expansions of a
mapped task, which is an aggregate view, and the pass it belongs to is already named in
the breadcrumb directly above. Narrowing to a single pass is what the task instances
table and its loop filter are for.
Co-Authored-By: Claude <no...@anthropic.com>
* Regenerate the UI client and reformat after the review changes
The spec changed when the execution route went back to its shape on main and the iteration
sort was withdrawn, but the generated client still carried both, so the UI lint hook
regenerated it and failed on the difference.
Two Python files were left unformatted by the same round of edits.
Co-Authored-By: Claude <no...@anthropic.com>
* Fix the type errors the UI static check compiles for
The checked-in tsconfig does not typecheck src; the hook builds its own project extending
tsconfig.app.json, which turns on noUnusedLocals among other things, so running tsc
against the root config said clean while the hook did not.
Two unused locals left behind by earlier edits, a key typed as a plain string where its
neighbours are cast, and a tooltip reading try_number off a union whose grid summary
member does not carry one -- lifting it out of the start_date branch lost the narrowing
that had been doing that work.
Co-Authored-By: Claude <no...@anthropic.com>
---------
Co-authored-by: Claude <no...@anthropic.com>
---
.../api_fastapi/core_api/datamodels/ui/grid.py | 3 +
.../api_fastapi/core_api/openapi/_private_ui.yaml | 15 +
.../core_api/openapi/v2-rest-api-generated.yaml | 32 +-
.../core_api/routes/public/task_instances.py | 19 +-
.../api_fastapi/core_api/routes/ui/gantt.py | 10 +-
.../airflow/api_fastapi/core_api/routes/ui/grid.py | 46 +-
.../core_api/services/public/task_coordinates.py | 23 +-
.../api_fastapi/core_api/services/ui/grid.py | 21 +-
.../ui/openapi-gen/queries/ensureQueryData.ts | 4 +-
.../src/airflow/ui/openapi-gen/queries/prefetch.ts | 4 +-
.../src/airflow/ui/openapi-gen/queries/queries.ts | 4 +-
.../src/airflow/ui/openapi-gen/queries/suspense.ts | 4 +-
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 28 ++
.../ui/openapi-gen/requests/services.gen.ts | 6 +-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 9 +
.../airflow/ui/public/i18n/locales/en/common.json | 7 +
.../src/airflow/ui/public/i18n/locales/en/dag.json | 18 +-
.../ui/src/components/ActionAccordion/columns.tsx | 2 +
.../ui/src/components/Breadcrumb/CrumbSwitcher.tsx | 18 +-
.../ui/src/components/Clear/Run/ClearRunDialog.tsx | 68 +--
.../TaskInstance/ClearGroupTaskInstanceDialog.tsx | 50 ++-
.../Clear/TaskInstance/ClearTaskInstanceDialog.tsx | 17 +-
.../components/FilterBar/filters/LoopFilter.tsx | 167 +++++++
.../FilterBar/filters/loopParams.test.ts | 73 +++
.../src/components/FilterBar/filters/loopParams.ts | 60 +++
.../airflow/ui/src/components/Graph/TaskLink.tsx | 17 +-
.../airflow/ui/src/components/Graph/TaskNode.tsx | 3 +
.../airflow/ui/src/components/LoopOptionLabel.tsx | 76 ++++
.../TaskInstance/MarkTaskInstanceAsDialog.tsx | 14 +-
.../ui/src/components/TaskInstanceTooltip.test.tsx | 25 ++
.../ui/src/components/TaskInstanceTooltip.tsx | 18 +-
.../src/airflow/ui/src/components/TaskName.tsx | 22 +-
.../src/airflow/ui/src/constants/filterConfigs.tsx | 12 +
.../src/airflow/ui/src/constants/searchParams.ts | 3 +-
.../ui/src/hooks/useResolvePassCoordinate.test.ts | 110 +++++
.../ui/src/hooks/useResolvePassCoordinate.ts | 87 ++++
.../ui/src/hooks/useTaskInstanceCoordinates.ts | 1 +
.../ui/src/layouts/Details/DagBreadcrumb.test.tsx | 50 ++-
.../ui/src/layouts/Details/DagBreadcrumb.tsx | 60 ++-
.../layouts/Details/Gantt/GanttTimeline.test.tsx | 139 +++++-
.../ui/src/layouts/Details/Gantt/GanttTimeline.tsx | 43 +-
.../airflow/ui/src/layouts/Details/Gantt/utils.ts | 10 +
.../airflow/ui/src/layouts/Details/Grid/Grid.tsx | 9 +-
.../ui/src/layouts/Details/Grid/GridTI.test.tsx | 4 +-
.../airflow/ui/src/layouts/Details/Grid/GridTI.tsx | 8 +-
.../ui/src/layouts/Details/Grid/TaskNames.tsx | 25 +-
.../layouts/Details/IterationSwitcherButton.tsx | 121 +++++
.../src/airflow/ui/src/layouts/Details/NavTabs.tsx | 87 ++--
.../pages/GroupTaskInstance/GroupTaskInstance.tsx | 1 -
.../GroupTaskInstance/GroupTaskInstances.test.tsx | 93 ----
.../pages/GroupTaskInstance/GroupTaskInstances.tsx | 29 +-
.../LoopIterations/IterationSelect.test.tsx | 107 -----
.../LoopIterations/IterationSelect.tsx | 155 -------
.../GroupTaskInstance/LoopIterations/index.ts | 19 -
.../LoopIterations/loopUtils.test.ts | 24 +-
.../GroupTaskInstance/LoopIterations/loopUtils.ts | 17 +-
.../GroupTaskInstance/useLoopOutcomeStats.tsx | 4 +-
.../airflow/ui/src/pages/Run/Execution.test.tsx | 490 ---------------------
.../src/airflow/ui/src/pages/Run/Execution.tsx | 325 --------------
airflow-core/src/airflow/ui/src/pages/Run/Run.tsx | 1 -
.../pages/Task/Overview/LoopHistoryChart.test.tsx | 2 +-
.../src/pages/Task/Overview/LoopHistoryChart.tsx | 2 +-
.../airflow/ui/src/pages/TaskInstance/Header.tsx | 15 +
.../src/pages/TaskInstance/TaskInstance.test.tsx | 2 +-
.../ui/src/pages/TaskInstance/TaskInstance.tsx | 10 +-
.../src/pages/TaskInstances/TaskInstances.test.tsx | 42 +-
.../ui/src/pages/TaskInstances/TaskInstances.tsx | 58 +--
.../pages/TaskInstances/TaskInstancesFilter.tsx | 33 +-
.../src/airflow/ui/src/queries/useGridStructure.ts | 32 +-
.../src/airflow/ui/src/queries/useIsLoopGroup.ts | 72 +++
.../ui/src/queries/useLoopFilterOptions.test.ts | 109 +++++
.../airflow/ui/src/queries/useLoopFilterOptions.ts | 180 ++++++++
.../src/airflow/ui/src/queries/useLoopSummary.ts | 9 +-
airflow-core/src/airflow/ui/src/router.tsx | 2 -
.../ui/src/system-components/Modal/Modal.test.tsx | 29 +-
.../ui/src/system-components/Modal/Modal.tsx | 33 +-
.../airflow/ui/src/utils/executionParams.test.ts | 35 +-
.../src/airflow/ui/src/utils/executionParams.ts | 32 +-
.../src/airflow/ui/src/utils/useFiltersHandler.ts | 1 +
.../core_api/routes/public/test_task_instances.py | 17 +
.../api_fastapi/core_api/routes/ui/test_gantt.py | 29 ++
.../api_fastapi/core_api/routes/ui/test_grid.py | 32 ++
82 files changed, 2108 insertions(+), 1585 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/grid.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/grid.py
index c916e615452..3e2c391df65 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/grid.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/grid.py
@@ -37,6 +37,8 @@ class LightGridTaskInstanceSummary(BaseModel):
dag_version_number: int | None = None
has_note: bool = False
loop_iterations_count: int | None = None
+ latest_region_id: UUID | None = None
+ latest_region_index: int | None = None
class GridTISummaries(BaseModel):
@@ -51,6 +53,7 @@ class LoopIterationSummary(BaseModel):
"""Execution state for one existing iteration."""
index: int
+ task_count: int = 0
state: TaskInstanceState | None = None
start_date: datetime | None = None
end_date: datetime | None = None
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
index 3b988ffb102..f80dd4b4f4b 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
+++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
@@ -4878,6 +4878,17 @@ components:
- type: integer
- type: 'null'
title: Loop Iterations Count
+ latest_region_id:
+ anyOf:
+ - type: string
+ format: uuid
+ - type: 'null'
+ title: Latest Region Id
+ latest_region_index:
+ anyOf:
+ - type: integer
+ - type: 'null'
+ title: Latest Region Index
type: object
required:
- task_id
@@ -4938,6 +4949,10 @@ components:
index:
type: integer
title: Index
+ task_count:
+ type: integer
+ title: Task Count
+ default: 0
state:
anyOf:
- $ref: '#/components/schemas/TaskInstanceState'
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
index 9fba36f2126..cd8df5b8534 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
+++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
@@ -9196,20 +9196,6 @@ paths:
minimum: -1
- type: 'null'
title: Region Index
- - name: cursor
- in: query
- required: false
- schema:
- anyOf:
- - type: string
- - type: 'null'
- description: Cursor for keyset-based pagination. Pass an empty string for
- the first page, then use ``next_cursor`` from the response. When ``cursor``
- is provided, ``offset`` is ignored.
- title: Cursor
- description: Cursor for keyset-based pagination. Pass an empty string for
- the first page, then use ``next_cursor`` from the response. When ``cursor``
- is provided, ``offset`` is ignored.
- name: loop_id
in: query
required: false
@@ -9217,7 +9203,9 @@ paths:
anyOf:
- type: string
- type: 'null'
+ description: Task group id of a loop; matches every instance it produced.
title: Loop Id
+ description: Task group id of a loop; matches every instance it produced.
- name: iteration
in: query
required: false
@@ -9226,7 +9214,23 @@ paths:
- type: integer
minimum: 0
- type: 'null'
+ description: Narrow ``loop_id`` to a single pass of that loop.
title: Iteration
+ description: Narrow ``loop_id`` to a single pass of that loop.
+ - name: cursor
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ - type: 'null'
+ description: Cursor for keyset-based pagination. Pass an empty string for
+ the first page, then use ``next_cursor`` from the response. When ``cursor``
+ is provided, ``offset`` is ignored.
+ title: Cursor
+ description: Cursor for keyset-based pagination. Pass an empty string for
+ the first page, then use ``next_cursor`` from the response. When ``cursor``
+ is provided, ``offset`` is ignored.
- name: task_id
in: query
required: false
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py
index 88715c445b2..ba7290eb707 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py
@@ -636,14 +636,20 @@ def get_task_instances(
resolver: CoordinateResolverDep,
region_id: Annotated[UUID | None, Query()] = None,
region_index: Annotated[int | None, Query(ge=-1)] = None,
+ loop_id: Annotated[
+ str | None,
+ Query(description="Task group id of a loop; matches every instance it produced."),
+ ] = None,
+ iteration: Annotated[
+ int | None,
+ Query(ge=0, description="Narrow ``loop_id`` to a single pass of that loop."),
+ ] = None,
cursor: str | None = Query(
None,
description="Cursor for keyset-based pagination. "
"Pass an empty string for the first page, then use ``next_cursor`` from the response. "
"When ``cursor`` is provided, ``offset`` is ignored.",
),
- loop_id: Annotated[str | None, Query()] = None,
- iteration: Annotated[int | None, Query(ge=0)] = None,
) -> TaskInstanceCollectionResponse:
"""
Get list of task instances.
@@ -664,7 +670,7 @@ def get_task_instances(
use_cursor = cursor is not None
dag_run = None
query = add_public_map_index(eager_load_task_instance_for_validation(select(TI)))
- if loop_id is None and iteration is not None:
+ if iteration is not None and loop_id is None:
raise HTTPException(status.HTTP_400_BAD_REQUEST, "iteration requires loop_id")
if loop_id is not None:
if dag_id == "~" or dag_run_id == "~":
@@ -681,9 +687,12 @@ def get_task_instances(
if region_index is not None and region_id is None:
raise HTTPException(status.HTTP_400_BAD_REQUEST, "region_index requires region_id")
if region_id is not None:
+ # One meaning per filter: ``region_id`` names an exact region, and the logical scope of
+ # a loop pass -- its own region plus the expansions nested in it -- is what ``loop_id``
+ # and ``iteration`` are for.
query = query.where(TI.region_id == region_id)
- if region_index is not None:
- query = query.where(TI.region_index == region_index)
+ if region_index is not None:
+ query = query.where(TI.region_index == region_index)
if dag_run_id != "~":
if dag_id == "~":
raise HTTPException(
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/gantt.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/gantt.py
index ba3f7c9bfba..9d125b5e934 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/gantt.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/gantt.py
@@ -89,8 +89,14 @@ def get_gantt_data(
)
query = current_tis.order_by(
- TaskInstance.task_id, TaskInstance.region_id, TaskInstance.region_index, TaskInstance.try_number
- )
+ TaskInstance.task_id,
+ TaskInstance.region_id,
+ TaskInstance.region_index,
+ TaskInstance.try_number,
+ # Every attempt is its own row and only the newest stays live, so without this the chart
+ # loses each earlier try the moment a clear archives it -- including the attempt the
+ # filter above drops the pending retry in favour of.
+ ).execution_options(include_all_attempts=True)
results = session.execute(query).fetchall()
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py
index cfd6bd2bb97..6eca4ae979a 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py
@@ -442,21 +442,55 @@ def _build_ti_summaries(
if isinstance(group, SerializedLoopTaskGroup)
}
regions = DynamicRegion.load_for_run(dag_id, run_id, session=session) if loop_ids else {}
- for ti in chain([first], rows):
+ streamed = chain([first], rows)
+ # Only a Dag with loops needs two passes over the rows, one to find each invocation's newest
+ # iteration and one to build the summaries. Without loops the stream is consumed once, so a
+ # run of any size is not held in memory to learn that none of it was looped.
+ positions_by_ti: Iterable[tuple[TaskInstance, dict[str, tuple[UUID, int]]]]
+ if loop_ids:
+ positions_by_ti = [
+ (
+ ti,
+ {
+ loop_id: position
+ for loop_id in loop_ids
+ if (position := loop_position(regions, ti.region_id, ti.region_index, loop_id))
+ is not None
+ },
+ )
+ for ti in streamed
+ ]
+ else:
+ positions_by_ti = ((ti, {}) for ti in streamed)
+ # A looped task has one row per iteration. Counting them all would report the work done
+ # across the whole loop rather than the state the task is in, so the state counts come from
+ # the newest iteration of each invocation. Every iteration still feeds loop_iterations,
+ # which is what reports how many ran.
+ latest_iteration: dict[tuple[str, str, UUID], int] = {}
+ # Guarded because without loops ``positions_by_ti`` streams: walking it here would leave
+ # nothing for the summaries below, and there is no iteration to find anyway.
+ if loop_ids:
+ for ti, positions in positions_by_ti:
+ for loop_id, (family_id, iteration) in positions.items():
+ key = (ti.task_id, loop_id, family_id)
+ if iteration > latest_iteration.get(key, -1):
+ latest_iteration[key] = iteration
+ for ti, positions in positions_by_ti:
summary = ti_details.get(ti.task_id)
if summary is None:
summary = ti_details[ti.task_id] = GridNodeAgg()
- loop_positions = {}
- for loop_id in loop_ids:
- if position := loop_position(regions, ti.region_id, ti.region_index, loop_id):
- loop_positions[loop_id] = position
summary.add_ti(
state=ti.state,
start_date=ti.start_date,
end_date=ti.end_date,
dag_version_number=getattr(ti, "version_number", None),
has_note=bool(getattr(ti, "has_note", False)),
- loop_positions=loop_positions,
+ loop_positions=positions,
+ region=(ti.region_id, ti.region_index) if positions else None,
+ is_latest_pass=all(
+ iteration == latest_iteration[(ti.task_id, loop_id, family_id)]
+ for loop_id, (family_id, iteration) in positions.items()
+ ),
)
def get_node_summaries() -> Iterable[dict[str, Any]]:
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/services/public/task_coordinates.py b/airflow-core/src/airflow/api_fastapi/core_api/services/public/task_coordinates.py
index b76b7f3b35f..86dff4b3984 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/services/public/task_coordinates.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/services/public/task_coordinates.py
@@ -113,27 +113,34 @@ def loop_iteration_filter(
iteration: int | None,
session: Session,
) -> ColumnElement[bool]:
+ """
+ Match the task instances a named loop produced, optionally narrowed to one iteration.
+
+ Keyed by the loop's node id rather than a region, so it spans every invocation of that loop
+ in the run: a filter names a loop the author wrote, not one of its executions. Regions stay
+ an internal coordinate -- nothing here asks the caller to know one.
+ """
regions = {
region.id: region
for region in session.scalars(
select(DynamicRegion).where(DynamicRegion.dag_id == dag_id, DynamicRegion.run_id == run_id)
)
}
- any_index: list[UUID] = []
- loop_node: list[UUID] = []
+ nested: list[UUID] = []
+ own_node: list[UUID] = []
for region in regions.values():
position = loop_position(regions, region.id, -1, loop_id)
if position is None:
continue
if iteration is not None and region.node_id == loop_id:
- loop_node.append(region.id)
+ own_node.append(region.id)
elif iteration is None or position[1] == iteration:
- any_index.append(region.id)
+ nested.append(region.id)
predicates: list[ColumnElement[bool]] = []
- if any_index:
- predicates.append(TaskInstance.region_id.in_(any_index))
- if loop_node:
- predicates.append(and_(TaskInstance.region_id.in_(loop_node), TaskInstance.region_index == iteration))
+ if nested:
+ predicates.append(TaskInstance.region_id.in_(nested))
+ if own_node:
+ predicates.append(and_(TaskInstance.region_id.in_(own_node), TaskInstance.region_index == iteration))
return or_(*predicates) if predicates else false()
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/services/ui/grid.py b/airflow-core/src/airflow/api_fastapi/core_api/services/ui/grid.py
index 0ee737037c4..10becd3a2f1 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/services/ui/grid.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/services/ui/grid.py
@@ -66,6 +66,7 @@ class GridNodeAgg:
dag_version_number: int | None = None
has_note: bool = False
loop_iterations: dict[str, set[tuple[UUID, int]]] = field(default_factory=dict)
+ latest_region: tuple[UUID, int] | None = None
def add_ti(
self,
@@ -76,9 +77,24 @@ class GridNodeAgg:
dag_version_number: int | None,
has_note: bool = False,
loop_positions: dict[str, tuple[UUID, int]] | None = None,
+ is_latest_pass: bool = True,
+ region: tuple[UUID, int] | None = None,
) -> None:
- """Merge one task instance row into the summary."""
+ """
+ Merge one task instance row into the summary.
+
+ Every pass counts towards the states. A looped task's cell stands for the work the loop
+ did rather than one run of it, and the UI says how many iterations that covers, so an
+ earlier pass that is re-running is visible instead of hidden behind a finished later one.
+
+ ``is_latest_pass`` is false for a superseded iteration. Its state still counts, but it
+ does not supply the coordinate: a link has to address the newest row.
+ """
self.child_states[state] += 1
+ # A looped task has one row per iteration, so a link without a coordinate cannot say
+ # which one it means. Carry the newest so the grid can address it.
+ if is_latest_pass and region is not None:
+ self.latest_region = region
if has_started(state, start_date):
for loop_id, position in (loop_positions or {}).items():
self.loop_iterations.setdefault(loop_id, set()).add(position)
@@ -166,6 +182,8 @@ def _get_aggs_for_node(summary: GridNodeAgg) -> dict[str, Any]:
"child_states": _serialize_child_states(summary.child_states),
"dag_version_number": summary.dag_version_number,
"has_note": summary.has_note,
+ "latest_region_id": None if summary.latest_region is None else summary.latest_region[0],
+ "latest_region_index": None if summary.latest_region is None else summary.latest_region[1],
}
@@ -413,6 +431,7 @@ def summarize_loop_run(
rows.append(
LoopIterationSummary(
index=index,
+ task_count=len(tasks),
state=agg_state(states),
start_date=min(starts) if starts else None,
end_date=max(ends) if ends and all(s in State.finished for s in states) else None,
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
index 718f1947310..2fb80f04de7 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
@@ -1379,9 +1379,9 @@ export const ensureUseTaskInstanceServiceGetMappedTaskInstanceData = (queryClien
* @param data.dagRunId
* @param data.regionId
* @param data.regionIndex
+* @param data.loopId Task group id of a loop; matches every instance it produced.
+* @param data.iteration Narrow ``loop_id`` to a single pass of that loop.
* @param data.cursor Cursor for keyset-based pagination. Pass an empty string for the first page, then use ``next_cursor`` from the response. When ``cursor`` is provided, ``offset`` is ignored.
-* @param data.loopId
-* @param data.iteration
* @param data.taskId
* @param data.runAfterGte
* @param data.runAfterGt
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
index bcd34c714af..b54bb5e4416 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
@@ -1379,9 +1379,9 @@ export const prefetchUseTaskInstanceServiceGetMappedTaskInstance = (queryClient:
* @param data.dagRunId
* @param data.regionId
* @param data.regionIndex
+* @param data.loopId Task group id of a loop; matches every instance it produced.
+* @param data.iteration Narrow ``loop_id`` to a single pass of that loop.
* @param data.cursor Cursor for keyset-based pagination. Pass an empty string for the first page, then use ``next_cursor`` from the response. When ``cursor`` is provided, ``offset`` is ignored.
-* @param data.loopId
-* @param data.iteration
* @param data.taskId
* @param data.runAfterGte
* @param data.runAfterGt
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
index 11b7e9f5851..fb85ecf5053 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
@@ -1379,9 +1379,9 @@ export const useTaskInstanceServiceGetMappedTaskInstance = <TData = Common.TaskI
* @param data.dagRunId
* @param data.regionId
* @param data.regionIndex
+* @param data.loopId Task group id of a loop; matches every instance it produced.
+* @param data.iteration Narrow ``loop_id`` to a single pass of that loop.
* @param data.cursor Cursor for keyset-based pagination. Pass an empty string for the first page, then use ``next_cursor`` from the response. When ``cursor`` is provided, ``offset`` is ignored.
-* @param data.loopId
-* @param data.iteration
* @param data.taskId
* @param data.runAfterGte
* @param data.runAfterGt
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
index 84c37145799..4665742bd3b 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
@@ -1379,9 +1379,9 @@ export const useTaskInstanceServiceGetMappedTaskInstanceSuspense = <TData = Comm
* @param data.dagRunId
* @param data.regionId
* @param data.regionIndex
+* @param data.loopId Task group id of a loop; matches every instance it produced.
+* @param data.iteration Narrow ``loop_id`` to a single pass of that loop.
* @param data.cursor Cursor for keyset-based pagination. Pass an empty string for the first page, then use ``next_cursor`` from the response. When ``cursor`` is provided, ``offset`` is ignored.
-* @param data.loopId
-* @param data.iteration
* @param data.taskId
* @param data.runAfterGte
* @param data.runAfterGt
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
index 236c239ff9f..e9ad0a3f236 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -12111,6 +12111,29 @@ export const $LightGridTaskInstanceSummary = {
}
],
title: 'Loop Iterations Count'
+ },
+ latest_region_id: {
+ anyOf: [
+ {
+ type: 'string',
+ format: 'uuid'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Latest Region Id'
+ },
+ latest_region_index: {
+ anyOf: [
+ {
+ type: 'integer'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Latest Region Index'
}
},
type: 'object',
@@ -12163,6 +12186,11 @@ export const $LoopIterationSummary = {
type: 'integer',
title: 'Index'
},
+ task_count: {
+ type: 'integer',
+ title: 'Task Count',
+ default: 0
+ },
state: {
anyOf: [
{
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
index 4b799d58b9d..1ea77170ad0 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
@@ -3135,9 +3135,9 @@ export class TaskInstanceService {
* @param data.dagRunId
* @param data.regionId
* @param data.regionIndex
+ * @param data.loopId Task group id of a loop; matches every instance it produced.
+ * @param data.iteration Narrow ``loop_id`` to a single pass of that loop.
* @param data.cursor Cursor for keyset-based pagination. Pass an empty string for the first page, then use ``next_cursor`` from the response. When ``cursor`` is provided, ``offset`` is ignored.
- * @param data.loopId
- * @param data.iteration
* @param data.taskId
* @param data.runAfterGte
* @param data.runAfterGt
@@ -3204,9 +3204,9 @@ export class TaskInstanceService {
query: {
region_id: data.regionId,
region_index: data.regionIndex,
- cursor: data.cursor,
loop_id: data.loopId,
iteration: data.iteration,
+ cursor: data.cursor,
task_id: data.taskId,
run_after_gte: data.runAfterGte,
run_after_gt: data.runAfterGt,
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
index 06ac2f54295..c2de73e2d05 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -3131,6 +3131,8 @@ export type LightGridTaskInstanceSummary = {
dag_version_number?: number | null;
has_note?: boolean;
loop_iterations_count?: number | null;
+ latest_region_id?: string | null;
+ latest_region_index?: number | null;
};
/**
@@ -3154,6 +3156,7 @@ export type LoopInvocationResponse = {
*/
export type LoopIterationSummary = {
index: number;
+ task_count?: number;
state?: TaskInstanceState | null;
start_date?: string | null;
end_date?: string | null;
@@ -4640,12 +4643,18 @@ export type GetTaskInstancesData = {
endDateLt?: string | null;
endDateLte?: string | null;
executor?: Array<(string)>;
+ /**
+ * Narrow ``loop_id`` to a single pass of that loop.
+ */
iteration?: number | null;
limit?: number;
logicalDateGt?: string | null;
logicalDateGte?: string | null;
logicalDateLt?: string | null;
logicalDateLte?: string | null;
+ /**
+ * Task group id of a loop; matches every instance it produced.
+ */
loopId?: string | null;
mapIndex?: Array<(number)>;
offset?: number;
diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json b/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json
index 745a8b54a79..77deb1cb1fc 100644
--- a/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json
+++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json
@@ -144,13 +144,19 @@
"filter": "Filter",
"filters": {
"addFilter": "Add Filter",
+ "allIterations": "All",
"completedAt": "Completed At",
"date": "Date",
"durationFrom": "Duration From",
"durationTo": "Duration To",
"endTime": "End Time",
+ "iteration": "Iteration",
"logicalDateFrom": "Logical Date From",
"logicalDateTo": "Logical Date To",
+ "loop": "Loop",
+ "loopAllIterations": "{{loop}} all",
+ "loopIteration": "{{loop}} {{index}}",
+ "loopPlaceholder": "Select a loop",
"maxActiveRunsFrom": "Max Active Runs From",
"maxActiveRunsTo": "Max Active Runs To",
"removeFilter": "Remove Filter",
@@ -160,6 +166,7 @@
"selectDateRange": "Select Date Range",
"selectDateTime": "Select date and time",
"startTime": "Start Time",
+ "switchIteration": "Switch iteration",
"time": "Time"
},
"fullscreen": {
diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/dag.json b/airflow-core/src/airflow/ui/public/i18n/locales/en/dag.json
index 654c52203c0..6ef678ed46c 100644
--- a/airflow-core/src/airflow/ui/public/i18n/locales/en/dag.json
+++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/dag.json
@@ -68,17 +68,6 @@
"upcomingCount_one": "{{count, number}} Upcoming Deadline",
"upcomingCount_other": "{{count, number}} Upcoming Deadlines"
},
- "execution": {
- "clearSelected": "Clear selected executions",
- "empty": "No executions in this run.",
- "iteration": "{{node}} · Iteration {{index}}",
- "mapped_one": "{{task}} ({{count, number}} mapped task on this page)",
- "mapped_other": "{{task}} ({{count, number}} mapped tasks on this page)",
- "page": "Showing {{start, number}}–{{end, number}} of {{total, number}} executions",
- "runTasks": "Run tasks",
- "select": "Select {{task}}",
- "title": "Execution"
- },
"extraLinks": "Extra Links",
"grid": {
"buttons": {
@@ -120,11 +109,6 @@
"warning": "WARNING"
},
"loop": {
- "filter": {
- "all": "All iterations",
- "invocation": "Invocation",
- "label": "Iteration"
- },
"history": {
"capLabel": "Limit: {{max}}",
"iterationsAxis": "Iterations",
@@ -134,7 +118,7 @@
},
"title": "Iterations per run"
},
- "iteration": "Iteration {{index}}",
+ "iterationLabel": "Iteration {{index}}",
"iterations": "Iterations",
"outcome": "Outcome",
"reason": {
diff --git a/airflow-core/src/airflow/ui/src/components/ActionAccordion/columns.tsx b/airflow-core/src/airflow/ui/src/components/ActionAccordion/columns.tsx
index 2fef63d5cdf..380461e9a11 100644
--- a/airflow-core/src/airflow/ui/src/components/ActionAccordion/columns.tsx
+++ b/airflow-core/src/airflow/ui/src/components/ActionAccordion/columns.tsx
@@ -27,6 +27,8 @@ import type { DataTableFeatures } from "src/components/DataTable/features";
import type { MetaColumn } from "src/components/DataTable/types";
import { StateBadge } from "src/components/StateBadge";
+// Keyed by coordinate rather than row id: a retry of the same execution comes back with a new
+// id, and an exclusion the user ticked should survive that.
export const taskInstanceKey = (ti: TaskInstanceResponse): string =>
`${ti.dag_run_id}:${ti.task_id}:${ti.map_index}:${ti.region_id}:${ti.region_index}`;
diff --git a/airflow-core/src/airflow/ui/src/components/Breadcrumb/CrumbSwitcher.tsx b/airflow-core/src/airflow/ui/src/components/Breadcrumb/CrumbSwitcher.tsx
index 573c893e7a3..e1783786c9e 100644
--- a/airflow-core/src/airflow/ui/src/components/Breadcrumb/CrumbSwitcher.tsx
+++ b/airflow-core/src/airflow/ui/src/components/Breadcrumb/CrumbSwitcher.tsx
@@ -18,7 +18,7 @@
*/
import type { ReactNode } from "react";
-import { Box } from "@chakra-ui/react";
+import { Box, type BoxProps } from "@chakra-ui/react";
import { BsChevronExpand } from "react-icons/bs";
import { IconButton, Popover, Tooltip } from "src/system-components";
@@ -28,6 +28,8 @@ import { CUT_PADDING, type CrumbShape, crumbButtonStyles, getWedgePadding } from
type Props = {
readonly children: ReactNode;
+ /** Width of the panel. The default suits a search; a short fixed list wants to shrink to fit. */
+ readonly contentWidth?: BoxProps["width"];
/** Names the chevron for the tooltip and for assistive technology. */
readonly label: string;
readonly onOpenChange: (open: boolean) => void;
@@ -44,7 +46,17 @@ type Props = {
* current one is named. Open state stays at the call site, which owns the keyboard shortcut (if
* the level has one) and the search the panel holds.
*/
-export const CrumbSwitcher = ({ children, label, onOpenChange, open, search, shape, testId, to }: Props) => (
+export const CrumbSwitcher = ({
+ children,
+ contentWidth = "sm",
+ label,
+ onOpenChange,
+ open,
+ search,
+ shape,
+ testId,
+ to,
+}: Props) => (
<Popover.Root
lazyMount
onOpenChange={(event) => onOpenChange(event.open)}
@@ -78,7 +90,7 @@ export const CrumbSwitcher = ({ children, label, onOpenChange, open, search, sha
</Tooltip>
</CrumbGroup>
</Popover.Anchor>
- <Popover.Content data-testid={`${testId}-popover`} width="sm">
+ <Popover.Content data-testid={`${testId}-popover`} width={contentWidth}>
<Box p={2}>{search}</Box>
</Popover.Content>
</Popover.Root>
diff --git a/airflow-core/src/airflow/ui/src/components/Clear/Run/ClearRunDialog.tsx b/airflow-core/src/airflow/ui/src/components/Clear/Run/ClearRunDialog.tsx
index e13db005672..dde0eb8db49 100644
--- a/airflow-core/src/airflow/ui/src/components/Clear/Run/ClearRunDialog.tsx
+++ b/airflow-core/src/airflow/ui/src/components/Clear/Run/ClearRunDialog.tsx
@@ -113,41 +113,41 @@ const ClearRunDialog = ({ dagRun, onClose, open }: Props) => {
return (
<Modal
footerActions={
- <>
- <Button
- disabled={affectedTasks.total_entries === 0}
- loading={isPending}
- onClick={() => {
- mutate({
- dagId,
- dagRunId,
- requestBody: {
- dry_run: false,
- note: note === dagRun.note ? undefined : note,
- only_failed: onlyFailed,
- only_new: onlyNew,
- run_on_latest_version: runOnLatestVersion,
- },
- });
- }}
+ <Button
+ disabled={affectedTasks.total_entries === 0}
+ loading={isPending}
+ onClick={() => {
+ mutate({
+ dagId,
+ dagRunId,
+ requestBody: {
+ dry_run: false,
+ note: note === dagRun.note ? undefined : note,
+ only_failed: onlyFailed,
+ only_new: onlyNew,
+ run_on_latest_version: runOnLatestVersion,
+ },
+ });
+ }}
+ >
+ <CgRedo /> {translate("modal.confirm")}
+ </Button>
+ }
+ footerOptions={
+ shouldShowBundleVersionOption ? (
+ <Checkbox
+ checked={runOnLatestVersionForced || runOnLatestVersion}
+ disabled={runOnLatestVersionForced}
+ onCheckedChange={(event) => setRunOnLatestVersion(Boolean(event.checked))}
+ title={
+ runOnLatestVersionForced
+ ? translate("dags:runAndTaskActions.options.runOnLatestVersionForced")
+ : undefined
+ }
>
- <CgRedo /> {translate("modal.confirm")}
- </Button>
- {shouldShowBundleVersionOption ? (
- <Checkbox
- checked={runOnLatestVersionForced || runOnLatestVersion}
- disabled={runOnLatestVersionForced}
- onCheckedChange={(event) => setRunOnLatestVersion(Boolean(event.checked))}
- title={
- runOnLatestVersionForced
- ? translate("dags:runAndTaskActions.options.runOnLatestVersionForced")
- : undefined
- }
- >
- {translate("dags:runAndTaskActions.options.runOnLatestVersion")}
- </Checkbox>
- ) : undefined}
- </>
+ {translate("dags:runAndTaskActions.options.runOnLatestVersion")}
+ </Checkbox>
+ ) : undefined
}
lazyMount
onOpenChange={(details) => {
diff --git a/airflow-core/src/airflow/ui/src/components/Clear/TaskInstance/ClearGroupTaskInstanceDialog.tsx b/airflow-core/src/airflow/ui/src/components/Clear/TaskInstance/ClearGroupTaskInstanceDialog.tsx
index 455bf93b527..645c6538d0f 100644
--- a/airflow-core/src/airflow/ui/src/components/Clear/TaskInstance/ClearGroupTaskInstanceDialog.tsx
+++ b/airflow-core/src/airflow/ui/src/components/Clear/TaskInstance/ClearGroupTaskInstanceDialog.tsx
@@ -136,31 +136,33 @@ export const ClearGroupTaskInstanceDialog = ({ onClose, open, taskInstance }: Pr
return (
<Modal
footerActions={
+ <Button
+ disabled={affectedTasks.total_entries === 0}
+ loading={isPending}
+ onClick={() => {
+ mutate({
+ dagId,
+ requestBody: {
+ dag_run_id: runId,
+ dry_run: false,
+ include_downstream: downstream,
+ include_future: future,
+ include_past: past,
+ include_upstream: upstream,
+ ...(note === null ? {} : { note }),
+ ...(keepTaskState ? { keep_task_state: true } : {}),
+ only_failed: onlyFailed,
+ run_on_latest_version: runOnLatestVersion,
+ task_group_id: groupId,
+ },
+ });
+ }}
+ >
+ <CgRedo /> {translate("modal.confirm")}
+ </Button>
+ }
+ footerOptions={
<>
- <Button
- disabled={affectedTasks.total_entries === 0}
- loading={isPending}
- onClick={() => {
- mutate({
- dagId,
- requestBody: {
- dag_run_id: runId,
- dry_run: false,
- include_downstream: downstream,
- include_future: future,
- include_past: past,
- include_upstream: upstream,
- ...(note === null ? {} : { note }),
- ...(keepTaskState ? { keep_task_state: true } : {}),
- only_failed: onlyFailed,
- run_on_latest_version: runOnLatestVersion,
- task_group_id: groupId,
- },
- });
- }}
- >
- <CgRedo /> {translate("modal.confirm")}
- </Button>
<Checkbox
checked={keepTaskState}
onCheckedChange={(event) => setKeepTaskState(Boolean(event.checked))}
diff --git a/airflow-core/src/airflow/ui/src/components/Clear/TaskInstance/ClearTaskInstanceDialog.tsx b/airflow-core/src/airflow/ui/src/components/Clear/TaskInstance/ClearTaskInstanceDialog.tsx
index c9c1de43558..ddc6d22acd3 100644
--- a/airflow-core/src/airflow/ui/src/components/Clear/TaskInstance/ClearTaskInstanceDialog.tsx
+++ b/airflow-core/src/airflow/ui/src/components/Clear/TaskInstance/ClearTaskInstanceDialog.tsx
@@ -374,18 +374,19 @@ const ClearTaskInstanceDialog = (props: Props) => {
<>
<Modal
footerActions={
+ <Button
+ disabled={data.total_entries === 0 || keptTaskInstances.length === 0}
+ loading={isPending}
+ onClick={gateRequest === undefined ? confirmClear : onOpen}
+ >
+ <CgRedo /> {translate("modal.confirm")}
+ </Button>
+ }
+ footerOptions={
<>
- <Button
- disabled={data.total_entries === 0 || keptTaskInstances.length === 0}
- loading={isPending}
- onClick={gateRequest === undefined ? confirmClear : onOpen}
- >
- <CgRedo /> {translate("modal.confirm")}
- </Button>
<Checkbox
checked={keepTaskState}
onCheckedChange={(event) => setKeepTaskState(Boolean(event.checked))}
- style={{ marginRight: "auto" }}
>
{translate("dags:runAndTaskActions.options.keepTaskState")}
</Checkbox>
diff --git a/airflow-core/src/airflow/ui/src/components/FilterBar/filters/LoopFilter.tsx b/airflow-core/src/airflow/ui/src/components/FilterBar/filters/LoopFilter.tsx
new file mode 100644
index 00000000000..0e45a6a9aa2
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/components/FilterBar/filters/LoopFilter.tsx
@@ -0,0 +1,167 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useRef } from "react";
+
+import { Box, HStack, createListCollection } from "@chakra-ui/react";
+import { MdLoop } from "react-icons/md";
+import { useParams } from "react-router-dom";
+
+import { Select } from "src/system-components";
+
+import { LoopOptionLabel } from "src/components/LoopOptionLabel";
+
+import {
+ decodeLoopOption,
+ encodeLoopOption,
+ useLoopFilterOptions,
+ type LoopOption,
+} from "src/queries/useLoopFilterOptions";
+
+import { FilterPill } from "../FilterPill";
+import type { FilterPluginProps } from "../types";
+import { isLoopFilterValue } from "./loopParams";
+
+const collapsePill = () => {
+ setTimeout(() => {
+ const activeElement = document.activeElement as HTMLElement;
+
+ activeElement.blur();
+ }, 0);
+};
+
+/**
+ * One pill, one list: every loop in the run and the passes it ran, as single choices.
+ *
+ * A pass is only ever offered under the loop that counts it, so the two halves cannot be set
+ * independently -- an iteration without a loop, or one the loop never reached, both used to
+ * leave the table empty with nothing to say why.
+ */
+export const LoopFilter = ({ filter, onChange, onRemove }: FilterPluginProps) => {
+ const { dagId, groupId, runId } = useParams();
+ const value = isLoopFilterValue(filter.value) ? filter.value : undefined;
+ const { options } = useLoopFilterOptions({ dagId, groupId, runId });
+
+ const hasJustSelected = useRef(false);
+ const selectedValue =
+ value === undefined
+ ? undefined
+ : encodeLoopOption(value.loopId, value.iteration === undefined ? undefined : Number(value.iteration));
+
+ // ``label`` is what the collection's default itemToString reads, so typeahead has something
+ // to match; everything on screen renders through LoopOptionLabel instead.
+ const items = options.map((option) => ({
+ ...option,
+ label: option.iteration === undefined ? option.loopId : `${option.loopId} ${option.iteration}`,
+ }));
+ const collection = createListCollection({ items });
+ const current = options.find((option) => option.value === selectedValue);
+
+ const handleChange = ({ value: picked }: { value: Array<string> }) => {
+ const [choice] = picked;
+
+ if (choice === undefined) {
+ return;
+ }
+ hasJustSelected.current = true;
+ onChange(decodeLoopOption(choice));
+ collapsePill();
+ };
+
+ return (
+ <FilterPill
+ displayValue={current === undefined ? "" : <LoopOptionLabel compact option={current} />}
+ filter={filter}
+ hasValue={value !== undefined}
+ onRemove={onRemove}
+ // ``onKeyDown`` is deliberately not forwarded, for the reason given on RunStateFilter: the
+ // select owns Enter and Escape, and the pill acting on Enter too tears the filter down
+ // before the chosen value commits.
+ renderInput={({ onBlur, onFocus }, { onRequestClose }) => (
+ <Box
+ alignItems="center"
+ bg="bg"
+ border="0.5px solid"
+ borderColor="border"
+ borderRadius="full"
+ display="flex"
+ h="full"
+ onBlur={onBlur}
+ onFocus={onFocus}
+ overflow="hidden"
+ tabIndex={0}
+ width="480px"
+ >
+ <Box
+ alignItems="center"
+ bg="gray.muted"
+ borderLeftRadius="full"
+ display="flex"
+ fontSize="sm"
+ fontWeight="medium"
+ h="full"
+ px={4}
+ py={2}
+ whiteSpace="nowrap"
+ >
+ {filter.config.label}:
+ </Box>
+ <Select.Root
+ border="none"
+ collection={collection}
+ // A filter added from the menu has nothing to show until it is given a value, so
+ // open straight onto the options instead of making the user click again.
+ defaultOpen={value === undefined}
+ flex="1"
+ h="full"
+ onOpenChange={({ open }) => {
+ if (!open && !hasJustSelected.current) {
+ onRequestClose();
+ }
+ hasJustSelected.current = false;
+ }}
+ onValueChange={handleChange}
+ value={selectedValue === undefined ? [] : [selectedValue]}
+ >
+ <Select.Trigger dataTestId={`${filter.config.key}-filter`} triggerProps={{ border: "none" }}>
+ <HStack gap={2}>
+ <MdLoop />
+ <Select.ValueText placeholder={filter.config.placeholder}>
+ {(picked: Array<LoopOption>) =>
+ picked[0] === undefined ? undefined : <LoopOptionLabel compact option={picked[0]} />
+ }
+ </Select.ValueText>
+ </HStack>
+ </Select.Trigger>
+ <Select.Content>
+ {items.map((item) => (
+ <Select.Item
+ data-testid={`${filter.config.key}-filter-${item.value}`}
+ item={item}
+ key={item.value}
+ >
+ <LoopOptionLabel option={item} />
+ </Select.Item>
+ ))}
+ </Select.Content>
+ </Select.Root>
+ </Box>
+ )}
+ />
+ );
+};
diff --git a/airflow-core/src/airflow/ui/src/components/FilterBar/filters/loopParams.test.ts b/airflow-core/src/airflow/ui/src/components/FilterBar/filters/loopParams.test.ts
new file mode 100644
index 00000000000..ed20af39a26
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/components/FilterBar/filters/loopParams.test.ts
@@ -0,0 +1,73 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { describe, expect, it } from "vitest";
+
+import { decodeLoopOption, encodeLoopOption } from "src/queries/useLoopFilterOptions";
+
+import { loopFromSearchParams, loopToSearchParams } from "./loopParams";
+
+describe("loopFromSearchParams", () => {
+ it("drops an iteration that has no loop to count it", () => {
+ expect(loopFromSearchParams(new URLSearchParams("iteration=2"))).toBeUndefined();
+ });
+
+ it("reads a loop on its own as every iteration", () => {
+ expect(loopFromSearchParams(new URLSearchParams("loop_id=batch"))).toEqual({
+ iteration: undefined,
+ loopId: "batch",
+ });
+ });
+
+ it("reads a loop and iteration together", () => {
+ expect(loopFromSearchParams(new URLSearchParams("loop_id=batch&iteration=2"))).toEqual({
+ iteration: "2",
+ loopId: "batch",
+ });
+ });
+});
+
+describe("loopToSearchParams", () => {
+ it("clears both params when the pill is removed", () => {
+ expect(loopToSearchParams(undefined)).toEqual({ iteration: undefined, loop_id: undefined });
+ });
+
+ it("clears a stale iteration when the whole loop is chosen", () => {
+ expect(loopToSearchParams({ loopId: "batch" })).toEqual({ iteration: undefined, loop_id: "batch" });
+ });
+
+ it("writes both when one pass is chosen", () => {
+ expect(loopToSearchParams({ iteration: "2", loopId: "batch" })).toEqual({
+ iteration: "2",
+ loop_id: "batch",
+ });
+ });
+});
+
+describe("loop option encoding", () => {
+ it("round-trips a dotted loop id that contains the separator's characters", () => {
+ expect(decodeLoopOption(encodeLoopOption("outer.inner", 3))).toEqual({
+ iteration: "3",
+ loopId: "outer.inner",
+ });
+ });
+
+ it("round-trips a whole loop", () => {
+ expect(decodeLoopOption(encodeLoopOption("outer.inner"))).toEqual({ loopId: "outer.inner" });
+ });
+});
diff --git a/airflow-core/src/airflow/ui/src/components/FilterBar/filters/loopParams.ts b/airflow-core/src/airflow/ui/src/components/FilterBar/filters/loopParams.ts
new file mode 100644
index 00000000000..92beb51a7c2
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/components/FilterBar/filters/loopParams.ts
@@ -0,0 +1,60 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { SearchParamsKeys } from "src/constants/searchParams";
+
+import type { FilterValue } from "../types";
+
+// Sentinel for "every iteration of this loop". It is not written to the URL -- the absence of
+// an iteration param is what means "all" -- but the select needs a value to render as chosen.
+export const ITERATION_ALL = "all";
+
+export type LoopFilterValue = {
+ // Absent means every iteration; otherwise the index as it appears in the loop summary.
+ iteration?: string;
+ loopId: string;
+};
+
+export const isLoopFilterValue = (value: FilterValue): value is LoopFilterValue =>
+ typeof value === "object" && value !== null && !Array.isArray(value) && "loopId" in value;
+
+// One pill over the two params the task instance endpoint takes. A loop is always part of the
+// value, so an iteration can never be set on its own -- which the server answers with nothing,
+// since an iteration index means nothing until you say which loop it counts.
+export const loopToSearchParams = (value: FilterValue): Record<string, string | undefined> => {
+ const selection = isLoopFilterValue(value) ? value : undefined;
+
+ return {
+ [SearchParamsKeys.ITERATION]: selection?.iteration,
+ [SearchParamsKeys.LOOP_ID]: selection?.loopId,
+ };
+};
+
+export const loopFromSearchParams = (params: URLSearchParams): LoopFilterValue | undefined => {
+ const loopId = params.get(SearchParamsKeys.LOOP_ID);
+
+ // An iteration without a loop is not a filter anyone can act on, so it is dropped rather
+ // than shown as a pill that matches nothing.
+ if (loopId === null || loopId === "") {
+ return undefined;
+ }
+
+ const iteration = params.get(SearchParamsKeys.ITERATION);
+
+ return { iteration: iteration === null || iteration === "" ? undefined : iteration, loopId };
+};
diff --git a/airflow-core/src/airflow/ui/src/components/Graph/TaskLink.tsx b/airflow-core/src/airflow/ui/src/components/Graph/TaskLink.tsx
index d263e26d5ee..0b2082470c9 100644
--- a/airflow-core/src/airflow/ui/src/components/Graph/TaskLink.tsx
+++ b/airflow-core/src/airflow/ui/src/components/Graph/TaskLink.tsx
@@ -23,16 +23,20 @@ import { useParams, useSearchParams, Link as RouterLink } from "react-router-dom
import { TaskName, type TaskNameProps } from "src/components/TaskName";
import { taskNodeSeparator } from "src/utils/assetGraph";
-import { stripExecutionParams } from "src/utils/executionParams";
+import { setExecutionCoordinates, stripExecutionParams } from "src/utils/executionParams";
type Props = {
readonly dagId?: string;
readonly hasTaskInstance?: boolean;
readonly id: string;
+ // A task inside a loop has one row per pass, so a link with no coordinate is ambiguous and
+ // the API rejects it. The graph draws the task once, so it points at the newest pass.
+ readonly latestRegionId?: string | null;
+ readonly latestRegionIndex?: number | null;
} & TaskNameProps;
export const TaskLink = forwardRef<HTMLAnchorElement, Props>(
- ({ hasTaskInstance = true, id, isGroup, isMapped, ...rest }, ref) => {
+ ({ hasTaskInstance = true, id, isGroup, isMapped, latestRegionId, latestRegionIndex, ...rest }, ref) => {
const { dagId: urlDagId = "", groupId, runId, taskId: urlTaskId } = useParams();
const [searchParams] = useSearchParams();
@@ -62,7 +66,14 @@ export const TaskLink = forwardRef<HTMLAnchorElement, Props>(
return (
<RouterLink
ref={ref}
- to={{ pathname: basePath + taskPath, search: stripExecutionParams(searchParams).toString() }}
+ to={{
+ pathname: basePath + taskPath,
+ search: setExecutionCoordinates(
+ stripExecutionParams(searchParams),
+ latestRegionId,
+ latestRegionIndex,
+ ).toString(),
+ }}
>
<TaskName isGroup={isGroup} isMapped={isMapped} {...rest} />
</RouterLink>
diff --git a/airflow-core/src/airflow/ui/src/components/Graph/TaskNode.tsx b/airflow-core/src/airflow/ui/src/components/Graph/TaskNode.tsx
index 38bc82048dc..714308a5997 100644
--- a/airflow-core/src/airflow/ui/src/components/Graph/TaskNode.tsx
+++ b/airflow-core/src/airflow/ui/src/components/Graph/TaskNode.tsx
@@ -159,6 +159,9 @@ export const TaskNode = ({
isMapped={isMapped}
isOpen={isOpen}
label={displayLabel}
+ latestRegionId={taskInstance?.latest_region_id}
+ latestRegionIndex={taskInstance?.latest_region_index}
+ loopIterationsRan={taskInstance?.loop_iterations_count}
loopMaxIterations={loopMaxIterations}
setupTeardownType={setupTeardownType}
/>
diff --git a/airflow-core/src/airflow/ui/src/components/LoopOptionLabel.tsx b/airflow-core/src/airflow/ui/src/components/LoopOptionLabel.tsx
new file mode 100644
index 00000000000..45b5d8e0f5b
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/components/LoopOptionLabel.tsx
@@ -0,0 +1,76 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { HStack, Icon, Text } from "@chakra-ui/react";
+import { useTranslation } from "react-i18next";
+import { MdLoop } from "react-icons/md";
+
+import { StateBadge } from "src/components/StateBadge";
+
+import type { LoopOption } from "src/queries/useLoopFilterOptions";
+
+// One renderer for the collapsed pill and the open list, so what you picked reads the same as
+// what you picked it from.
+export const LoopOptionLabel = ({
+ compact = false,
+ hideLoopName = false,
+ option,
+}: {
+ readonly compact?: boolean;
+ // Set where the loop is already named alongside, so the rows read "All", "0", "1".
+ readonly hideLoopName?: boolean;
+ readonly option: LoopOption;
+}) => {
+ const { t: translate } = useTranslation(["common"]);
+ const isWholeLoop = option.iteration === undefined;
+ const named = isWholeLoop
+ ? translate("common:filters.loopAllIterations", { loop: option.loopId })
+ : translate("common:filters.loopIteration", { index: option.iteration, loop: option.loopId });
+ const bare = isWholeLoop ? translate("common:filters.allIterations") : String(option.iteration);
+ const label = hideLoopName ? bare : named;
+
+ return (
+ // A list row spreads its name and state to the edges; a collapsed pill or trigger sits
+ // inline and shrinks to what it holds, so it must not claim the full width.
+ <HStack
+ // A list row is a block that spreads its name and state to the edges. A collapsed pill or
+ // trigger sits inside a line of text, so it has to stay inline or it breaks that line.
+ display={compact ? "inline-flex" : "flex"}
+ gap={2}
+ justifyContent={compact ? "flex-start" : "space-between"}
+ width={compact ? "auto" : "full"}
+ >
+ <Text whiteSpace="nowrap">{label}</Text>
+ <HStack gap={1}>
+ {isWholeLoop && option.iterationsRan !== undefined ? (
+ <HStack color={option.state === "failed" ? "fg.error" : "fg.muted"} gap={1}>
+ <Icon aria-hidden as={MdLoop} boxSize={3.5} />
+ <Text fontSize="xs">
+ {option.iterationsRan}/{option.maxIterations}
+ </Text>
+ </HStack>
+ ) : undefined}
+ {option.state === undefined ? undefined : (
+ <StateBadge fontSize="xs" state={option.state}>
+ {compact ? undefined : translate(`common:states.${option.state}`)}
+ </StateBadge>
+ )}
+ </HStack>
+ </HStack>
+ );
+};
diff --git a/airflow-core/src/airflow/ui/src/components/MarkAs/TaskInstance/MarkTaskInstanceAsDialog.tsx b/airflow-core/src/airflow/ui/src/components/MarkAs/TaskInstance/MarkTaskInstanceAsDialog.tsx
index 751882812fd..80c3098f8de 100644
--- a/airflow-core/src/airflow/ui/src/components/MarkAs/TaskInstance/MarkTaskInstanceAsDialog.tsx
+++ b/airflow-core/src/airflow/ui/src/components/MarkAs/TaskInstance/MarkTaskInstanceAsDialog.tsx
@@ -46,7 +46,7 @@ const MarkTaskInstanceAsDialog = ({ onClose, open, state, taskInstance }: Props)
const dagRunId = taskInstance.dag_run_id;
const taskId = taskInstance.task_id;
const mapIndex = taskInstance.map_index;
- const isRegional = isLoopTaskInstance(taskInstance);
+ const regional = isLoopTaskInstance(taskInstance);
const { t: translate } = useTranslation();
const [markTaskInstanceDefaultOptions] = useMarkTaskInstanceDefaultOptions();
@@ -54,8 +54,8 @@ const MarkTaskInstanceAsDialog = ({ onClose, open, state, taskInstance }: Props)
// Skipping is limited to this task instance, so the options affecting other task instances are ignored.
const isSkipped = state === "skipped";
- const past = !isSkipped && !isRegional && selectedOptions.includes("past");
- const future = !isSkipped && !isRegional && selectedOptions.includes("future");
+ const past = !isSkipped && !regional && selectedOptions.includes("past");
+ const future = !isSkipped && !regional && selectedOptions.includes("future");
const upstream = !isSkipped && selectedOptions.includes("upstream");
const downstream = !isSkipped && selectedOptions.includes("downstream");
@@ -88,7 +88,7 @@ const MarkTaskInstanceAsDialog = ({ onClose, open, state, taskInstance }: Props)
refetchOnMount: "always",
},
requestBody: {
- ...(isRegional ? { region_id: taskInstance.region_id, region_index: taskInstance.region_index } : {}),
+ ...(regional ? { region_id: taskInstance.region_id, region_index: taskInstance.region_index } : {}),
include_downstream: downstream,
include_future: future,
include_past: past,
@@ -115,7 +115,7 @@ const MarkTaskInstanceAsDialog = ({ onClose, open, state, taskInstance }: Props)
dagRunId,
mapIndex,
requestBody: {
- ...(isRegional
+ ...(regional
? { region_id: taskInstance.region_id, region_index: taskInstance.region_index }
: {}),
include_downstream: downstream,
@@ -158,12 +158,12 @@ const MarkTaskInstanceAsDialog = ({ onClose, open, state, taskInstance }: Props)
onChange={setSelectedOptions}
options={[
{
- disabled: isSkipped || isRegional || taskInstance.logical_date === null,
+ disabled: isSkipped || regional || taskInstance.logical_date === null,
label: translate("dags:runAndTaskActions.options.past"),
value: "past",
},
{
- disabled: isSkipped || isRegional || taskInstance.logical_date === null,
+ disabled: isSkipped || regional || taskInstance.logical_date === null,
label: translate("dags:runAndTaskActions.options.future"),
value: "future",
},
diff --git a/airflow-core/src/airflow/ui/src/components/TaskInstanceTooltip.test.tsx b/airflow-core/src/airflow/ui/src/components/TaskInstanceTooltip.test.tsx
index 521661e7586..e667c59f86f 100644
--- a/airflow-core/src/airflow/ui/src/components/TaskInstanceTooltip.test.tsx
+++ b/airflow-core/src/airflow/ui/src/components/TaskInstanceTooltip.test.tsx
@@ -266,4 +266,29 @@ describe("TaskInstanceTooltip", () => {
expect(screen.getByText(/runId/iu)).toBeInTheDocument();
expect(screen.getByText(/manual__2025-01-01T00:00:00\+00:00/iu)).toBeInTheDocument();
});
+
+ it("names the try even on a first attempt", () => {
+ render(
+ <TaskInstanceTooltip
+ open
+ taskInstance={{
+ child_states: null,
+ loop_iteration: { iteration: 1, loop_id: "process_batch" },
+ max_end_date: null,
+ min_start_date: null,
+ state: "success",
+ task_display_name: "batch_ready",
+ task_id: "process_batch.batch_ready",
+ try_number: 1,
+ }}
+ >
+ <span>trigger</span>
+ </TaskInstanceTooltip>,
+ { wrapper: Wrapper },
+ );
+
+ // Clearing a pass regenerates the later ones as fresh executions that start at try 1, so
+ // two neighbouring bars can both be a first try and the number is what tells them apart.
+ expect(screen.getByText(/tryNumber/iu).textContent).toContain("1");
+ });
});
diff --git a/airflow-core/src/airflow/ui/src/components/TaskInstanceTooltip.tsx b/airflow-core/src/airflow/ui/src/components/TaskInstanceTooltip.tsx
index b1891784dfd..8df0332b122 100644
--- a/airflow-core/src/airflow/ui/src/components/TaskInstanceTooltip.tsx
+++ b/airflow-core/src/airflow/ui/src/components/TaskInstanceTooltip.tsx
@@ -94,8 +94,7 @@ const TaskInstanceTooltip = ({
taskInstance.loop_iteration !== null &&
taskInstance.loop_iteration !== undefined ? (
<Text>
- {translate("taskInstance.iteration")}: {taskInstance.loop_iteration.iteration} (
- {taskInstance.loop_iteration.loop_id})
+ {translate("taskInstance.iteration")}: {taskInstance.loop_iteration.iteration}
</Text>
) : undefined}
{"loop_iterations_count" in taskInstance &&
@@ -122,13 +121,18 @@ const TaskInstanceTooltip = ({
{translate("taskInstance.queuedWhen")}: <Time datetime={taskInstance.queued_when} />
</Text>
) : undefined}
+ {/* Outside the branches below, which key off the shape a caller passes: the chart
+ reports min_start_date rather than start_date, so a try nested in that branch
+ never reached it. Shown for a first try too, since clearing a loop pass
+ regenerates the later ones as fresh executions that start at try 1 -- two
+ neighbouring bars can both be a first try, and this is what tells them apart. */}
+ {"try_number" in taskInstance && typeof taskInstance.try_number === "number" && (
+ <Text>
+ {translate("tryNumber")}: {taskInstance.try_number}
+ </Text>
+ )}
{"start_date" in taskInstance ? (
<>
- {taskInstance.try_number > 1 && (
- <Text>
- {translate("tryNumber")}: {taskInstance.try_number}
- </Text>
- )}
<Text>
{translate("startDate")}: <Time datetime={taskInstance.start_date} />
</Text>
diff --git a/airflow-core/src/airflow/ui/src/components/TaskName.tsx b/airflow-core/src/airflow/ui/src/components/TaskName.tsx
index adae8b479e8..9d113614be5 100644
--- a/airflow-core/src/airflow/ui/src/components/TaskName.tsx
+++ b/airflow-core/src/airflow/ui/src/components/TaskName.tsx
@@ -20,7 +20,8 @@ import type { CSSProperties } from "react";
import { Box, type TextProps } from "@chakra-ui/react";
import { useTranslation } from "react-i18next";
-import { FiArrowDownRight, FiArrowUpRight, FiRepeat } from "react-icons/fi";
+import { FiArrowDownRight, FiArrowUpRight } from "react-icons/fi";
+import { MdLoop } from "react-icons/md";
import type { NodeResponse } from "openapi/requests/types.gen";
@@ -34,6 +35,8 @@ export type TaskNameProps = {
readonly isOpen?: boolean;
readonly isZoomedOut?: boolean;
readonly label: string;
+ /** Iterations that actually ran. Omitted where only the definition is known. */
+ readonly loopIterationsRan?: number | null;
readonly loopMaxIterations?: number | null;
readonly setupTeardownType?: NodeResponse["setup_teardown_type"];
} & TextProps;
@@ -52,11 +55,18 @@ export const TaskName = ({
isOpen = false,
isZoomedOut,
label,
+ loopIterationsRan,
loopMaxIterations,
setupTeardownType,
...rest
}: TaskNameProps) => {
const { i18n } = useTranslation();
+ // The cap alone reads as a count, so a loop that stopped at 4 of 10 would claim 10. Show what
+ // ran against what was allowed, and fall back to the cap only where no run is in view.
+ const loopIterationsLabel =
+ loopIterationsRan === null || loopIterationsRan === undefined
+ ? (loopMaxIterations ?? undefined)
+ : `${loopIterationsRan}/${loopMaxIterations ?? "?"}`;
if (isGroup) {
return (
@@ -69,16 +79,14 @@ export const TaskName = ({
{...rest}
>
{label}
+ {/* A group can be both: a loop whose body is a mapped expansion carries both markers. */}
{isLoop ? (
<>
- <FiRepeat size={14} style={{ ...iconStyle, marginLeft: 4 }} />
- {loopMaxIterations === null || loopMaxIterations === undefined
- ? undefined
- : ` ${loopMaxIterations}`}
+ <MdLoop size={14} style={{ ...iconStyle, marginLeft: 4 }} />
+ {loopIterationsLabel === undefined ? undefined : ` ${loopIterationsLabel}`}
</>
- ) : isMapped ? (
- " [ ]"
) : undefined}
+ {isMapped ? " [ ]" : undefined}
</Box>
);
}
diff --git a/airflow-core/src/airflow/ui/src/constants/filterConfigs.tsx b/airflow-core/src/airflow/ui/src/constants/filterConfigs.tsx
index 041af1d640c..969abe1b089 100644
--- a/airflow-core/src/airflow/ui/src/constants/filterConfigs.tsx
+++ b/airflow-core/src/airflow/ui/src/constants/filterConfigs.tsx
@@ -30,6 +30,7 @@ import {
MdDateRange,
MdHistory,
MdHourglassEmpty,
+ MdLoop,
MdPause,
MdPendingActions,
MdPlayArrow,
@@ -43,9 +44,11 @@ import { useTeamsServiceListTeams } from "openapi/queries";
import type { DagRunState, DagRunType, TaskInstanceState } from "openapi/requests/types.gen";
import type { FilterConfig } from "src/components/FilterBar";
+import { LoopFilter } from "src/components/FilterBar/filters/LoopFilter";
import { RunStateFilter } from "src/components/FilterBar/filters/RunStateFilter";
import { TagsFilter } from "src/components/FilterBar/filters/TagsFilter";
import { TimetableTypeFilter } from "src/components/FilterBar/filters/TimetableTypeFilter";
+import { loopFromSearchParams, loopToSearchParams } from "src/components/FilterBar/filters/loopParams";
import {
pausedStateFromSearchParams,
pausedStateToSearchParams,
@@ -306,6 +309,15 @@ export const useFilterConfigs = () => {
startKey: SearchParamsKeys.LOGICAL_DATE_GTE,
type: FilterTypes.DATERANGE,
},
+ [SearchParamsKeys.LOOP]: {
+ EditorComponent: LoopFilter,
+ fromSearchParams: loopFromSearchParams,
+ icon: <MdLoop />,
+ label: translate("common:filters.loop"),
+ placeholder: translate("common:filters.loopPlaceholder"),
+ toSearchParams: loopToSearchParams,
+ type: FilterTypes.SELECT,
+ },
[SearchParamsKeys.MAP_INDEX]: {
icon: <LuBrackets />,
label: translate("common:mapIndex"),
diff --git a/airflow-core/src/airflow/ui/src/constants/searchParams.ts b/airflow-core/src/airflow/ui/src/constants/searchParams.ts
index ada46fa5246..66f6f788ec3 100644
--- a/airflow-core/src/airflow/ui/src/constants/searchParams.ts
+++ b/airflow-core/src/airflow/ui/src/constants/searchParams.ts
@@ -86,7 +86,8 @@ export enum SearchParamsKeys {
LOGICAL_DATE_GTE = "logical_date_gte",
LOGICAL_DATE_LTE = "logical_date_lte",
LOGICAL_DATE_RANGE = "logical_date_range",
- LOOP_REGION_ID = "loop_region_id",
+ LOOP = "loop",
+ LOOP_ID = "loop_id",
MAP_INDEX = "map_index",
MAPPED = "mapped",
MAX_ACTIVE_RUNS_GTE = "max_active_runs_gte",
diff --git a/airflow-core/src/airflow/ui/src/hooks/useResolvePassCoordinate.test.ts b/airflow-core/src/airflow/ui/src/hooks/useResolvePassCoordinate.test.ts
new file mode 100644
index 00000000000..df4dbab309e
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/hooks/useResolvePassCoordinate.test.ts
@@ -0,0 +1,110 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { renderHook, waitFor } from "@testing-library/react";
+import type * as ReactRouterDom from "react-router-dom";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { useTaskInstanceServiceGetTaskInstances } from "openapi/queries";
+
+import { Wrapper } from "src/utils/Wrapper";
+
+import { useResolvePassCoordinate } from "./useResolvePassCoordinate";
+
+let mockParams: Record<string, string | undefined> = {};
+let mockSearchParams = new URLSearchParams();
+const setSearchParams = vi.fn();
+
+vi.mock("react-router-dom", async (importOriginal) => {
+ const actual = await importOriginal<typeof ReactRouterDom>();
+
+ return {
+ ...actual,
+ useParams: () => mockParams,
+ useSearchParams: () => [mockSearchParams, setSearchParams] as const,
+ };
+});
+
+vi.mock("openapi/queries", () => ({
+ useTaskInstanceServiceGetTaskInstances: vi.fn(),
+}));
+
+const lookup = (row?: { region_id: string; region_index: number }, isFetching = false) =>
+ ({ data: { task_instances: row === undefined ? [] : [row] }, isFetching }) as ReturnType<
+ typeof useTaskInstanceServiceGetTaskInstances
+ >;
+
+describe("useResolvePassCoordinate", () => {
+ beforeEach(() => {
+ setSearchParams.mockClear();
+ mockParams = { dagId: "d", runId: "r", taskId: "batch.work" };
+ mockSearchParams = new URLSearchParams();
+ vi.mocked(useTaskInstanceServiceGetTaskInstances).mockReturnValue(lookup());
+ });
+
+ it("does nothing when the url already names a row", () => {
+ mockSearchParams = new URLSearchParams("region_id=own®ion_index=3");
+
+ const { result } = renderHook(() => useResolvePassCoordinate(), { wrapper: Wrapper });
+
+ expect(result.current).toBe(false);
+ const [, , options] = vi.mocked(useTaskInstanceServiceGetTaskInstances).mock.calls[0] ?? [];
+
+ expect(options).toMatchObject({ enabled: false });
+ expect(setSearchParams).not.toHaveBeenCalled();
+ });
+
+ it("asks for the row of that pass, narrowed to the mapped slot", () => {
+ mockParams = { ...mockParams, mapIndex: "1" };
+ mockSearchParams = new URLSearchParams("loop_id=batch&iteration=2");
+
+ renderHook(() => useResolvePassCoordinate(), { wrapper: Wrapper });
+
+ const [request] = vi.mocked(useTaskInstanceServiceGetTaskInstances).mock.calls[0] ?? [];
+
+ expect(request).toMatchObject({ iteration: 2, loopId: "batch", mapIndex: [1], taskId: "batch.work" });
+ });
+
+ it("writes back the row's own coordinate, which the pass does not name", async () => {
+ mockParams = { ...mockParams, mapIndex: "1" };
+ mockSearchParams = new URLSearchParams("loop_id=batch&iteration=0");
+ // A mapped expansion lives in its own region, so neither value follows from the pass.
+ vi.mocked(useTaskInstanceServiceGetTaskInstances).mockReturnValue(
+ lookup({ region_id: "expansion", region_index: 1 }),
+ );
+
+ renderHook(() => useResolvePassCoordinate(), { wrapper: Wrapper });
+
+ await waitFor(() => expect(setSearchParams).toHaveBeenCalled());
+ const [written, options] = setSearchParams.mock.calls[0] as [URLSearchParams, { replace: boolean }];
+
+ expect(written.get("region_id")).toBe("expansion");
+ expect(written.get("region_index")).toBe("1");
+ // The same place, named properly, so it should not add a history entry.
+ expect(options).toEqual({ replace: true });
+ });
+
+ it("holds while the row is unknown, so a looped task is never asked for unscoped", () => {
+ mockSearchParams = new URLSearchParams("loop_id=batch&iteration=0");
+ vi.mocked(useTaskInstanceServiceGetTaskInstances).mockReturnValue(lookup(undefined, true));
+
+ const { result } = renderHook(() => useResolvePassCoordinate(), { wrapper: Wrapper });
+
+ expect(result.current).toBe(true);
+ });
+});
diff --git a/airflow-core/src/airflow/ui/src/hooks/useResolvePassCoordinate.ts b/airflow-core/src/airflow/ui/src/hooks/useResolvePassCoordinate.ts
new file mode 100644
index 00000000000..bedb1c0bcb1
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/hooks/useResolvePassCoordinate.ts
@@ -0,0 +1,87 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useEffect } from "react";
+
+import { useParams, useSearchParams } from "react-router-dom";
+
+import { useTaskInstanceServiceGetTaskInstances } from "openapi/queries";
+
+import { SearchParamsKeys } from "src/constants/searchParams";
+
+/**
+ * Turn a loop and pass in the url into the coordinate that names one row.
+ *
+ * A pass is what a reader can choose -- a number they can see -- but it does not address a row:
+ * a mapped expansion lives in its own region, and a clear forks later passes into a successor
+ * one. Only the server knows which row a pass became, and the list endpoint already answers
+ * that, so the pair is resolved there and written back as a coordinate. Everything downstream
+ * keeps reading the url, and the resolved link is the one worth sharing.
+ *
+ * Returns whether the lookup is still in flight. A looped task instance cannot be asked for
+ * without a coordinate -- the request is rejected rather than answered empty -- so the caller
+ * must hold its children until this clears.
+ */
+export const useResolvePassCoordinate = (): boolean => {
+ const [searchParams, setSearchParams] = useSearchParams();
+ const { dagId, mapIndex, runId, taskId } = useParams();
+ const loopId = searchParams.get(SearchParamsKeys.LOOP_ID);
+ const iteration = searchParams.get(SearchParamsKeys.ITERATION);
+ const parsedMapIndex = mapIndex === undefined ? Number.NaN : Number.parseInt(mapIndex, 10);
+
+ const needsLookup =
+ searchParams.get(SearchParamsKeys.REGION_ID) === null &&
+ loopId !== null &&
+ loopId !== "" &&
+ iteration !== null &&
+ iteration !== "" &&
+ dagId !== undefined &&
+ runId !== undefined &&
+ taskId !== undefined;
+
+ const { data, isFetching } = useTaskInstanceServiceGetTaskInstances(
+ {
+ dagId: dagId ?? "",
+ dagRunId: runId ?? "",
+ iteration: needsLookup ? Number(iteration) : undefined,
+ limit: 1,
+ loopId: loopId ?? undefined,
+ mapIndex: Number.isInteger(parsedMapIndex) ? [parsedMapIndex] : undefined,
+ taskId: taskId ?? "",
+ },
+ undefined,
+ { enabled: needsLookup },
+ );
+ const [row] = data?.task_instances ?? [];
+ const regionId = row?.region_id;
+ const regionIndex = row?.region_index;
+
+ useEffect(() => {
+ if (!needsLookup || regionId === undefined || regionIndex === undefined) {
+ return;
+ }
+ const next = new URLSearchParams(searchParams);
+
+ next.set(SearchParamsKeys.REGION_ID, regionId);
+ next.set(SearchParamsKeys.REGION_INDEX, String(regionIndex));
+ // Replaced rather than pushed: this is the same place, named properly.
+ setSearchParams(next, { replace: true });
+ }, [needsLookup, regionId, regionIndex, searchParams, setSearchParams]);
+
+ return needsLookup && (isFetching || regionId === undefined);
+};
diff --git a/airflow-core/src/airflow/ui/src/hooks/useTaskInstanceCoordinates.ts b/airflow-core/src/airflow/ui/src/hooks/useTaskInstanceCoordinates.ts
index 55fe3b488b9..26c9975b1ba 100644
--- a/airflow-core/src/airflow/ui/src/hooks/useTaskInstanceCoordinates.ts
+++ b/airflow-core/src/airflow/ui/src/hooks/useTaskInstanceCoordinates.ts
@@ -20,6 +20,7 @@ import { useSearchParams } from "react-router-dom";
import { SearchParamsKeys } from "src/constants/searchParams";
+/** The coordinate naming the task instance a page shows, as the url carries it. */
export const useTaskInstanceCoordinates = () => {
const [searchParams] = useSearchParams();
const regionIndex = searchParams.get(SearchParamsKeys.REGION_INDEX);
diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/DagBreadcrumb.test.tsx b/airflow-core/src/airflow/ui/src/layouts/Details/DagBreadcrumb.test.tsx
index f104faf5cba..f940ca33cf0 100644
--- a/airflow-core/src/airflow/ui/src/layouts/Details/DagBreadcrumb.test.tsx
+++ b/airflow-core/src/airflow/ui/src/layouts/Details/DagBreadcrumb.test.tsx
@@ -38,6 +38,22 @@ vi.mock("react-router-dom", async (importOriginal) => {
return { ...(original as object), useMatches: vi.fn() };
});
+const LOOP_PASSES = [
+ { iteration: undefined, iterationsRan: 3, loopId: "batch", loopRegionId: "r", maxIterations: 5 },
+ { iteration: 2, loopId: "batch", loopRegionId: "r" },
+ { iteration: 1, loopId: "batch", loopRegionId: "r" },
+ { iteration: 0, loopId: "batch", loopRegionId: "r" },
+];
+
+// Most levels have no loop in scope; the loop tests opt in by setting this.
+let mockLoopOptions: Array<unknown> = [];
+
+vi.mock("src/queries/useLoopFilterOptions", async (importOriginal) => {
+ const original = await importOriginal();
+
+ return { ...(original as object), useLoopFilterOptions: () => ({ options: mockLoopOptions }) };
+});
+
// The msw dag-details handler is pinned to this dag id.
const DAG_ID = "tutorial_taskflow_api";
@@ -63,7 +79,10 @@ const createWrapper =
</BaseWrapper>
);
-beforeEach(() => vi.mocked(useMatches).mockReturnValue([]));
+beforeEach(() => {
+ mockLoopOptions = [];
+ vi.mocked(useMatches).mockReturnValue([]);
+});
afterEach(() => cleanup());
@@ -237,4 +256,33 @@ describe("DagBreadcrumb", () => {
),
);
});
+
+ it("names the newest pass when the url pins none", async () => {
+ mockLoopOptions = LOOP_PASSES;
+ render(<DagBreadcrumb />, {
+ wrapper: createWrapper(
+ `/dags/${DAG_ID}/runs/run_1/tasks/batch.member`,
+ "/dags/:dagId/runs/:runId/tasks/:taskId",
+ ),
+ });
+
+ // Passes arrive newest first, so the highest index leads.
+ expect(await screen.findByTestId("switch-iteration")).toBeInTheDocument();
+ expect(screen.getByTestId("dag-breadcrumb")).toHaveTextContent("2");
+ });
+
+ it("falls back to this loop's newest when the url carries another loop's pass", async () => {
+ mockLoopOptions = LOOP_PASSES;
+ render(<DagBreadcrumb />, {
+ wrapper: createWrapper(
+ `/dags/${DAG_ID}/runs/run_1/tasks/batch.member?loop_id=other&iteration=0`,
+ "/dags/:dagId/runs/:runId/tasks/:taskId",
+ ),
+ });
+
+ await screen.findByTestId("switch-iteration");
+
+ // Reading "0" here would name a pass of a loop that never counted it.
+ expect(screen.getByTestId("dag-breadcrumb")).toHaveTextContent("2");
+ });
});
diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/DagBreadcrumb.tsx b/airflow-core/src/airflow/ui/src/layouts/Details/DagBreadcrumb.tsx
index 63c4284bba9..62da7adb37a 100644
--- a/airflow-core/src/airflow/ui/src/layouts/Details/DagBreadcrumb.tsx
+++ b/airflow-core/src/airflow/ui/src/layouts/Details/DagBreadcrumb.tsx
@@ -17,7 +17,7 @@
* under the License.
*/
import { useTranslation } from "react-i18next";
-import { useLocation, useParams } from "react-router-dom";
+import { useLocation, useParams, useSearchParams } from "react-router-dom";
import {
useDagRunServiceGetDagRun,
@@ -30,11 +30,15 @@ import type { TaskInstanceState } from "openapi/requests/types.gen";
import { BreadcrumbRow, CrumbLink, CrumbStack, CrumbText, type CrumbShape } from "src/components/Breadcrumb";
+import { SearchParamsKeys } from "src/constants/searchParams";
import { useTaskInstanceCoordinates } from "src/hooks/useTaskInstanceCoordinates";
+import { useGridTiSummariesStream } from "src/queries/useGridTISummaries.ts";
+import { selectActivePass, useLoopFilterOptions, type LoopOption } from "src/queries/useLoopFilterOptions";
import { isStatePending, useAutoRefresh } from "src/utils";
import { DagRunSwitcherButton } from "./DagRunSwitcherButton";
import { DagSwitcherButton } from "./DagSwitcherButton";
+import { IterationSwitcherButton } from "./IterationSwitcherButton";
/** Levels that stand for a Dag run — both offer the run search. */
const RUN_LEVEL_KEYS = new Set(["allRuns", "dagRun"]);
@@ -52,14 +56,18 @@ type Crumb = {
const BreadcrumbItem = ({
crumb,
dagId,
+ isGroupLevel,
isLast,
isMapped,
+ loopOptions,
shape,
}: {
readonly crumb: Crumb;
readonly dagId: string;
+ readonly isGroupLevel: boolean;
readonly isLast: boolean;
readonly isMapped: boolean;
+ readonly loopOptions: Array<LoopOption>;
readonly shape: CrumbShape;
}) => {
const content = (
@@ -72,6 +80,14 @@ const BreadcrumbItem = ({
/>
);
+ if (crumb.key === "iteration") {
+ return (
+ <IterationSwitcherButton includeWholeLoop={isGroupLevel} options={loopOptions} shape={shape}>
+ {content}
+ </IterationSwitcherButton>
+ );
+ }
+
if (crumb.key === "dag") {
return (
<DagSwitcherButton dagId={dagId} shape={shape}>
@@ -102,6 +118,9 @@ const BreadcrumbItem = ({
export const DagBreadcrumb = () => {
const { t: translate } = useTranslation(["common", "dag"]);
const { dagId = "", groupId, mapIndex = "-1", runId, taskId } = useParams();
+ const [searchParams] = useSearchParams();
+ // A task inside a loop is named for it, so either level resolves to the same loop.
+ const { options: loopOptions } = useLoopFilterOptions({ dagId, groupId: groupId ?? taskId, runId });
const { pathname } = useLocation();
const coordinates = useTaskInstanceCoordinates();
const refetchInterval = useAutoRefresh({ dagId });
@@ -125,6 +144,16 @@ export const DagBreadcrumb = () => {
const { data: task } = useTaskServiceGetTask({ dagId, taskId }, undefined, { enabled: Boolean(taskId) });
+ // A group has no instance of its own, so its state is the aggregate the grid already computes.
+ const { summariesByRunId } = useGridTiSummariesStream({
+ dagId,
+ runIds: groupId !== undefined && runId !== undefined ? [runId] : [],
+ });
+ const groupSummary =
+ groupId === undefined || runId === undefined
+ ? undefined
+ : summariesByRunId.get(runId)?.task_instances.find((summary) => summary.task_id === groupId);
+
const { data: mappedTaskInstance } = useTaskInstanceServiceGetMappedTaskInstance(
{ ...coordinates, dagId, dagRunId: runId ?? "", mapIndex: parsedMapIndex, taskId: taskId ?? "" },
undefined,
@@ -180,7 +209,9 @@ export const DagBreadcrumb = () => {
if (groupId !== undefined) {
crumbs.push({
caption: translate("taskGroup_one"),
+ hasState: true,
key: "group",
+ state: groupSummary?.state,
to: `/dags/${dagId}/groups/${groupId}`,
value: groupId,
});
@@ -214,6 +245,31 @@ export const DagBreadcrumb = () => {
});
}
+ // Only a level a loop encloses has a pass to name, so the crumb follows the options rather
+ // than the route: no loop in scope means no options and no crumb.
+ // A pass is a level inside a group or task; the run above them has no single loop to name.
+ if (loopOptions.length > 0 && (groupId !== undefined || taskId !== undefined)) {
+ // A group with no pass pinned is being looked at whole, which is a state only it can be in:
+ // a task instance is always one row, so it falls back to the newest pass instead.
+ const pinnedPass = searchParams.get(SearchParamsKeys.ITERATION);
+ const isWholeLoop = taskId === undefined && (pinnedPass === null || pinnedPass === "");
+ // A loaded row knows its own pass; only a group has to read it off the url.
+ const resolvedPass = (mappedTaskInstance ?? taskInstance)?.loop_iteration?.iteration;
+ const selected = isWholeLoop ? undefined : selectActivePass(loopOptions, searchParams, resolvedPass);
+
+ if (isWholeLoop || selected !== undefined) {
+ crumbs.push({
+ caption: translate("common:filters.iteration"),
+ hasState: true,
+ key: "iteration",
+ // Looking at the loop whole means its own outcome; looking at one pass means that pass's.
+ state: isWholeLoop ? loopOptions[0]?.state : selected?.state,
+ // How far the loop got belongs to the header's stat; the crumb names the pass alone.
+ value: isWholeLoop ? translate("common:filters.allIterations") : String(selected?.iteration),
+ });
+ }
+ }
+
if (mapIndex !== "-1") {
crumbs.push({
caption: translate("mapIndex"),
@@ -230,9 +286,11 @@ export const DagBreadcrumb = () => {
<BreadcrumbItem
crumb={crumb}
dagId={dagId}
+ isGroupLevel={taskId === undefined}
isLast={index === crumbs.length - 1}
isMapped={hasExpandedInstances}
key={crumb.key}
+ loopOptions={loopOptions}
shape={{ hasNotch: index > 0, hasPoint: index < crumbs.length - 1 }}
/>
))}
diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/GanttTimeline.test.tsx b/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/GanttTimeline.test.tsx
index fad5f9d4ad3..7d5b4daf2ec 100644
--- a/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/GanttTimeline.test.tsx
+++ b/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/GanttTimeline.test.tsx
@@ -46,6 +46,25 @@ vi.mock("@tanstack/react-virtual", () => ({
}),
}));
+let mockParams: Record<string, string | undefined> = {};
+let mockSearchParams = new URLSearchParams();
+
+vi.mock("react-router-dom", async (importOriginal) => {
+ const original = await importOriginal();
+
+ return {
+ ...(original as object),
+ useParams: () => mockParams,
+ useSearchParams: () => [mockSearchParams, vi.fn()] as const,
+ };
+});
+
+// Group membership is read from the Dag structure, which this test does not stand up.
+vi.mock("src/queries/useIsLoopGroup", () => ({
+ useGroupMembers: (groupId?: string) =>
+ groupId === "refine" ? new Set(["refine", "refine.improve", "refine.evaluate"]) : new Set<string>(),
+}));
+
const TestWrapper = ({ children }: PropsWithChildren) => <Wrapper>{children}</Wrapper>;
// Shared time range: 10:00 → 10:10 UTC on 2024-03-14
@@ -87,6 +106,18 @@ const buildSegment = (taskInstanceId: string, start: number, end: number): Gantt
y: "task_1",
});
+const passSegment = (taskId: string, iteration: number): GanttDataItem => ({
+ loopIteration: { iteration, loop_id: "refine" },
+ state: "success",
+ taskId,
+ taskInstanceId: `${taskId}-${iteration}`,
+ tryNumber: 1,
+ x: [MIN_MS, MAX_MS],
+ y: taskId,
+});
+
+const PASS_MEMBERS = ["refine.improve", "refine.evaluate"];
+
describe("GanttTimeline segment bars", () => {
afterEach(() => {
vi.unstubAllGlobals();
@@ -112,7 +143,8 @@ describe("GanttTimeline segment bars", () => {
fireEvent.pointerMove(screen.getByRole("link"), { pointerType: "mouse" });
- expect(await screen.findByText(`taskInstance.iteration: ${iteration} (loop)`)).toBeInTheDocument();
+ // The loop's name is not repeated here: the group it belongs to is already on screen.
+ expect(await screen.findByText(`taskInstance.iteration: ${iteration}`)).toBeInTheDocument();
});
it.each([
@@ -447,4 +479,109 @@ describe("GanttTimeline segment bars", () => {
// Regular task bar does not.
expect(bars[1]?.querySelector('[style*="repeating-linear-gradient"]')).toBeNull();
});
+
+ it.each([
+ { expected: 2, name: "every task of the group", params: { groupId: "refine" } },
+ { expected: 1, name: "only the row itself", params: { taskId: "refine.improve" } },
+ ])("lights up $name at the pinned pass", ({ expected, params }) => {
+ mockParams = params;
+ mockSearchParams = new URLSearchParams("iteration=2");
+ const nodes = PASS_MEMBERS.map((id) => ({ ...BASE_NODE, id, label: id }));
+
+ render(
+ <GanttTimeline
+ {...defaultProps}
+ flatNodes={nodes}
+ rowSegments={PASS_MEMBERS.map((id) => [passSegment(id, 2)])}
+ scrollContainerRef={makeScrollRef()}
+ />,
+ { wrapper: TestWrapper },
+ );
+
+ expect(document.querySelectorAll("[data-pinned-pass]")).toHaveLength(expected);
+ });
+
+ it("leaves other passes alone", () => {
+ mockParams = { groupId: "refine" };
+ mockSearchParams = new URLSearchParams("iteration=2");
+ const nodes = PASS_MEMBERS.map((id) => ({ ...BASE_NODE, id, label: id }));
+
+ render(
+ <GanttTimeline
+ {...defaultProps}
+ flatNodes={nodes}
+ rowSegments={PASS_MEMBERS.map((id) => [passSegment(id, 5)])}
+ scrollContainerRef={makeScrollRef()}
+ />,
+ { wrapper: TestWrapper },
+ );
+
+ expect(document.querySelectorAll("[data-pinned-pass]")).toHaveLength(0);
+ });
+
+ it("leaves the highlight open where one run's segments meet", () => {
+ mockParams = { taskId: "refine.improve" };
+ mockSearchParams = new URLSearchParams("iteration=2");
+ // Two segments of the same attempt, as a queued stretch followed by the run itself.
+ const shared = { loopIteration: { iteration: 2, loop_id: "refine" }, taskInstanceId: "one-try" };
+ const half = MIN_MS + (MAX_MS - MIN_MS) / 2;
+ const segments: Array<GanttDataItem> = [
+ {
+ ...shared,
+ state: "queued",
+ taskId: "refine.improve",
+ tryNumber: 1,
+ x: [MIN_MS, half],
+ y: "refine.improve",
+ },
+ {
+ ...shared,
+ state: "success",
+ taskId: "refine.improve",
+ tryNumber: 1,
+ x: [half, MAX_MS],
+ y: "refine.improve",
+ },
+ ];
+
+ render(
+ <GanttTimeline
+ {...defaultProps}
+ flatNodes={[{ ...BASE_NODE, id: "refine.improve", label: "refine.improve" }]}
+ rowSegments={[segments]}
+ scrollContainerRef={makeScrollRef()}
+ />,
+ { wrapper: TestWrapper },
+ );
+
+ const [first, second] = [...document.querySelectorAll("[data-pinned-pass]")];
+ const styleOf = (node: Element) => globalThis.getComputedStyle(node);
+
+ // The run is ringed at its ends and along its length, but not down the seam between the
+ // two states, which is inside the thing being highlighted rather than its edge.
+ expect(styleOf(first as Element).borderLeftWidth).toBe("2px");
+ expect(styleOf(first as Element).borderRightWidth).not.toBe("2px");
+ expect(styleOf(second as Element).borderLeftWidth).not.toBe("2px");
+ expect(styleOf(second as Element).borderRightWidth).toBe("2px");
+ expect(styleOf(first as Element).borderTopWidth).toBe("2px");
+ });
+
+ it("does not read a region index as a pass", () => {
+ mockParams = { taskId: "refine.improve" };
+ // A mapped expansion's region index is its map slot. Ringing bars because slot 2 happens to
+ // look like pass 2 would highlight an unrelated pass.
+ mockSearchParams = new URLSearchParams("region_index=2®ion_id=expansion");
+
+ render(
+ <GanttTimeline
+ {...defaultProps}
+ flatNodes={[{ ...BASE_NODE, id: "refine.improve", label: "refine.improve" }]}
+ rowSegments={[[passSegment("refine.improve", 2)]]}
+ scrollContainerRef={makeScrollRef()}
+ />,
+ { wrapper: TestWrapper },
+ );
+
+ expect(document.querySelectorAll("[data-pinned-pass]")).toHaveLength(0);
+ });
});
diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/GanttTimeline.tsx b/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/GanttTimeline.tsx
index 76119281c38..9e1205d1761 100644
--- a/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/GanttTimeline.tsx
+++ b/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/GanttTimeline.tsx
@@ -22,7 +22,7 @@ import { Fragment, useLayoutEffect, useRef, useState } from "react";
import { Badge, Box, Flex, Text } from "@chakra-ui/react";
import { useVirtualizer } from "@tanstack/react-virtual";
import dayjs from "dayjs";
-import { Link, useLocation, useParams } from "react-router-dom";
+import { Link, useLocation, useParams, useSearchParams } from "react-router-dom";
import type { LightGridTaskInstanceSummary } from "openapi/requests/types.gen";
@@ -30,6 +30,7 @@ import {
GANTT_AXIS_HEIGHT_PX,
GANTT_TOP_PADDING_PX,
ROW_HEIGHT,
+ SELECTED_TASK_OUTLINE_COLOR,
TASK_BAR_HEIGHT_PX,
} from "src/layouts/Details/Grid/constants";
import type { GridTask } from "src/layouts/Details/Grid/utils";
@@ -37,6 +38,9 @@ import type { GridTask } from "src/layouts/Details/Grid/utils";
import { StateIcon } from "src/components/StateIcon";
import TaskInstanceTooltip from "src/components/TaskInstanceTooltip";
+import { SearchParamsKeys } from "src/constants/searchParams";
+import { useColorMode } from "src/context/colorMode";
+import { useGroupMembers } from "src/queries/useIsLoopGroup";
import { useDurationFormat } from "src/utils";
import {
@@ -121,6 +125,17 @@ export const GanttTimeline = ({
const { locale } = useDurationFormat();
const location = useLocation();
const { groupId: selectedGroupId, taskId: selectedTaskId } = useParams();
+ const [searchParams] = useSearchParams();
+ const { colorMode = "light" } = useColorMode();
+ // Read from the structure rather than the id: a group declared with prefix_group_id=False
+ // does not name its children after itself.
+ const groupMembers = useGroupMembers(selectedGroupId);
+ // The pass the page is pinned to, named either as a filter or as a region coordinate. Bars
+ // from that pass are outlined the way the grid outlines the instance you are looking at.
+ // Only an explicitly chosen pass highlights. A region index cannot stand in for one: on a
+ // mapped expansion it is the map slot, so reading it here rings bars of an unrelated pass.
+ const pinnedPass = searchParams.get(SearchParamsKeys.ITERATION);
+ const pinnedOutline = SELECTED_TASK_OUTLINE_COLOR[colorMode];
const [bodyWidthPx, setBodyWidthPx] = useState(0);
const bodyRef = useRef<HTMLDivElement | null>(null);
@@ -295,6 +310,11 @@ export const GanttTimeline = ({
: allSegments;
const taskId = node.id;
const isSelected = selectedTaskId === taskId || selectedGroupId === taskId;
+ // What a pinned pass covers follows what is being looked at: one task instance is a
+ // single row, while a task group is every task it holds, so the whole pass lights up.
+ // Group ids are dotted paths, so a member is named for the group enclosing it.
+ const inPinnedScope =
+ selectedTaskId === undefined ? groupMembers.has(taskId) : taskId === selectedTaskId;
const gridSummary = summaryByTaskId.get(taskId);
return (
@@ -347,6 +367,14 @@ export const GanttTimeline = ({
segments[segIndex - 1]?.taskInstanceId === segment.taskInstanceId &&
segments[segIndex - 1]?.tryNumber === tryNumber;
+ const isPinnedPass =
+ inPinnedScope &&
+ pinnedPass !== null &&
+ pinnedPass !== "" &&
+ segment.loopIteration !== null &&
+ segment.loopIteration !== undefined &&
+ String(segment.loopIteration.iteration) === pinnedPass;
+
return (
<TaskInstanceTooltip
key={`${segment.taskInstanceId ?? taskId}-${tryNumber ?? -1}-${state ?? "none"}-${x[0]}`}
@@ -377,11 +405,24 @@ export const GanttTimeline = ({
>
<Badge
alignItems="center"
+ borderBottomColor={isPinnedPass ? pinnedOutline : undefined}
borderBottomLeftRadius={touchesPrev ? 0 : barRadius}
borderBottomRightRadius={touchesNext ? 0 : barRadius}
+ borderBottomStyle={isPinnedPass ? "solid" : undefined}
+ borderBottomWidth={isPinnedPass ? "2px" : undefined}
+ borderLeftColor={isPinnedPass ? pinnedOutline : undefined}
+ borderLeftStyle={isPinnedPass && !touchesPrev ? "solid" : undefined}
+ borderLeftWidth={isPinnedPass && !touchesPrev ? "2px" : undefined}
+ borderRightColor={isPinnedPass ? pinnedOutline : undefined}
+ borderRightStyle={isPinnedPass && !touchesNext ? "solid" : undefined}
+ borderRightWidth={isPinnedPass && !touchesNext ? "2px" : undefined}
+ borderTopColor={isPinnedPass ? pinnedOutline : undefined}
borderTopLeftRadius={touchesPrev ? 0 : barRadius}
borderTopRightRadius={touchesNext ? 0 : barRadius}
+ borderTopStyle={isPinnedPass ? "solid" : undefined}
+ borderTopWidth={isPinnedPass ? "2px" : undefined}
colorPalette={state ?? "none"}
+ data-pinned-pass={isPinnedPass || undefined}
display="flex"
h="100%"
justifyContent="center"
diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/utils.ts b/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/utils.ts
index a6ae1199da3..5303455a2c4 100644
--- a/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/utils.ts
+++ b/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/utils.ts
@@ -344,6 +344,16 @@ export const getGanttSegmentTo = ({
searchParams.set(SearchParamsKeys.REGION_INDEX, item.regionIndex.toString());
}
+ // The coordinate addresses the row; the loop and pass are what the breadcrumb and the filter
+ // read, so a bar carries both and clicking one lands on the pass it was drawn for.
+ if (item.loopIteration !== null && item.loopIteration !== undefined) {
+ searchParams.set(SearchParamsKeys.LOOP_ID, item.loopIteration.loop_id);
+ searchParams.set(SearchParamsKeys.ITERATION, item.loopIteration.iteration.toString());
+ } else {
+ searchParams.delete(SearchParamsKeys.LOOP_ID);
+ searchParams.delete(SearchParamsKeys.ITERATION);
+ }
+
if (tryNumber !== undefined && tryNumber < (maxTryByKey.get(getGanttTryKey(item)) ?? 1)) {
searchParams.set(SearchParamsKeys.TRY_NUMBER, tryNumber.toString());
}
diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/Grid/Grid.tsx b/airflow-core/src/airflow/ui/src/layouts/Details/Grid/Grid.tsx
index d0e58004c0e..f33d98c74a4 100644
--- a/airflow-core/src/airflow/ui/src/layouts/Details/Grid/Grid.tsx
+++ b/airflow-core/src/airflow/ui/src/layouts/Details/Grid/Grid.tsx
@@ -91,7 +91,7 @@ export const Grid = ({
const usesSharedScroll = Boolean(sharedScrollContainerRef && showGantt);
const { openGroupIds, toggleGroupId } = useGroups();
- const { dagId = "", groupId: selectedGroupId, taskId: selectedTaskId } = useParams();
+ const { dagId = "", groupId: selectedGroupId, runId: selectedRunId, taskId: selectedTaskId } = useParams();
const [searchParams] = useSearchParams();
const filterRoot = searchParams.get("root") ?? undefined;
@@ -270,7 +270,12 @@ export const Grid = ({
{/* Grid body */}
<Flex bg="bg" height={`${rowVirtualizer.getTotalSize()}px`} position="relative">
<Box left={0} position="sticky" zIndex={1} {...taskNameColumnStyles}>
- <TaskNames nodes={flatNodes} onRowClick={handleRowClick} virtualItems={virtualItems} />
+ <TaskNames
+ nodes={flatNodes}
+ onRowClick={handleRowClick}
+ tiSummaries={selectedRunId === undefined ? undefined : summariesByRunId.get(selectedRunId)}
+ virtualItems={virtualItems}
+ />
</Box>
<Flex flexDirection="row-reverse" flexShrink={0} style={{ isolation: "isolate" }}>
{!showGantt && <ScrollbarSpacer width={scrollbarSpacerWidth} />}
diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/Grid/GridTI.test.tsx b/airflow-core/src/airflow/ui/src/layouts/Details/Grid/GridTI.test.tsx
index fdba1a47861..87104d168f3 100644
--- a/airflow-core/src/airflow/ui/src/layouts/Details/Grid/GridTI.test.tsx
+++ b/airflow-core/src/airflow/ui/src/layouts/Details/Grid/GridTI.test.tsx
@@ -83,7 +83,7 @@ const renderGridTI = (
describe("GridTI", () => {
it("drops another execution's exact region when following an aggregate cell", () => {
renderGridTI(
- `/dags/example_dag/runs/${SELECTED_RUN_ID}/tasks/selected_task?region_id=11111111-1111-4111-8111-111111111111®ion_index=3&try_number=2&log_level=error&iteration=2&loop_region_id=another-loop`,
+ `/dags/example_dag/runs/${SELECTED_RUN_ID}/tasks/selected_task?region_id=11111111-1111-4111-8111-111111111111®ion_index=3&try_number=2&log_level=error`,
{ taskId: "other_task" },
);
const link = screen.getByRole("link");
@@ -91,8 +91,6 @@ describe("GridTI", () => {
expect(link).toHaveAttribute("href", expect.stringContaining("log_level=error"));
expect(link).not.toHaveAttribute("href", expect.stringContaining("region_"));
expect(link).not.toHaveAttribute("href", expect.stringContaining("try_number="));
- expect(link).not.toHaveAttribute("href", expect.stringContaining("iteration="));
- expect(link).not.toHaveAttribute("href", expect.stringContaining("loop_region_id="));
});
beforeEach(() => {
colorModeMock.mockReturnValue({ colorMode: "light" });
diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/Grid/GridTI.tsx b/airflow-core/src/airflow/ui/src/layouts/Details/Grid/GridTI.tsx
index a3ba4a0a882..5fc44cdb436 100644
--- a/airflow-core/src/airflow/ui/src/layouts/Details/Grid/GridTI.tsx
+++ b/airflow-core/src/airflow/ui/src/layouts/Details/Grid/GridTI.tsx
@@ -25,7 +25,7 @@ import { StateIcon } from "src/components/StateIcon";
import TaskInstanceTooltip from "src/components/TaskInstanceTooltip";
import { useColorMode } from "src/context/colorMode";
-import { stripExecutionParams } from "src/utils/executionParams";
+import { setExecutionCoordinates, stripExecutionParams } from "src/utils/executionParams";
import { buildTaskInstanceUrl } from "src/utils/links";
import { NOTE_GRADIENT, SELECTED_TASK_OUTLINE_COLOR } from "./constants";
@@ -72,7 +72,11 @@ export const GridTI = ({
})
: `/dags/${dagId}/tasks/${isGroup ? "group/" : ""}${taskId}`;
- const redirectionSearch = stripExecutionParams(searchParams).toString();
+ const redirectionSearch = setExecutionCoordinates(
+ stripExecutionParams(searchParams),
+ instance.latest_region_id,
+ instance.latest_region_index,
+ ).toString();
const isSelectedRow = selectedTaskId === taskId || selectedGroupId === taskId;
const isSelectedTaskInstance = selectedRunId === runId && isSelectedRow;
diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/Grid/TaskNames.tsx b/airflow-core/src/airflow/ui/src/layouts/Details/Grid/TaskNames.tsx
index 1b91b9e00ec..abe099d28ed 100644
--- a/airflow-core/src/airflow/ui/src/layouts/Details/Grid/TaskNames.tsx
+++ b/airflow-core/src/airflow/ui/src/layouts/Details/Grid/TaskNames.tsx
@@ -24,10 +24,12 @@ import { useTranslation } from "react-i18next";
import { FiChevronUp } from "react-icons/fi";
import { Link as RouterLink, useParams, useSearchParams } from "react-router-dom";
+import type { GridTISummaries } from "openapi/requests";
+
import { TaskName } from "src/components/TaskName";
import { useGroups } from "src/context/groups";
-import { stripExecutionParams } from "src/utils/executionParams";
+import { setExecutionCoordinates, stripExecutionParams } from "src/utils/executionParams";
import { ROW_HEIGHT } from "./constants";
import type { GridTask } from "./utils";
@@ -36,12 +38,13 @@ type Props = {
readonly depth?: number;
readonly nodes: Array<GridTask>;
readonly onRowClick?: () => void;
+ readonly tiSummaries?: GridTISummaries;
readonly virtualItems?: Array<VirtualItem>;
};
const indent = (depth: number) => `${depth * 0.75 + 0.5}rem`;
-export const TaskNames = ({ nodes, onRowClick, virtualItems }: Props) => {
+export const TaskNames = ({ nodes, onRowClick, tiSummaries, virtualItems }: Props) => {
const { t: translate } = useTranslation("dag");
const { toggleGroupId } = useGroups();
const { dagId = "", groupId, taskId } = useParams();
@@ -73,7 +76,18 @@ export const TaskNames = ({ nodes, onRowClick, virtualItems }: Props) => {
onRowClick?.();
};
- const search = stripExecutionParams(searchParams).toString();
+ const baseSearchParams = stripExecutionParams(searchParams);
+ // A task inside a loop has one row per iteration, so its link needs a coordinate or the API
+ // cannot say which one is meant. The grid shows the task once, so point at the newest.
+ const searchFor = (nodeId: string) => {
+ const summary = tiSummaries?.task_instances.find((ti) => ti.task_id === nodeId);
+
+ return setExecutionCoordinates(
+ new URLSearchParams(baseSearchParams),
+ summary?.latest_region_id,
+ summary?.latest_region_index,
+ ).toString();
+ };
// If virtualItems is provided, use virtualization; otherwise render all items
const itemsToRender =
@@ -126,7 +140,7 @@ export const TaskNames = ({ nodes, onRowClick, virtualItems }: Props) => {
style={{ outline: "none" }}
to={{
pathname: `/dags/${dagId}/tasks/group/${node.id}`,
- search,
+ search: searchFor(node.id),
}}
>
<Flex alignItems="center" width="100%">
@@ -137,7 +151,6 @@ export const TaskNames = ({ nodes, onRowClick, virtualItems }: Props) => {
isLoop={Boolean(node.is_loop)}
isMapped={Boolean(node.is_mapped)}
label={node.label}
- loopMaxIterations={node.loop_max_iterations}
paddingLeft={indent(node.depth)}
setupTeardownType={node.setup_teardown_type}
/>
@@ -170,7 +183,7 @@ export const TaskNames = ({ nodes, onRowClick, virtualItems }: Props) => {
replace
to={{
pathname: `/dags/${dagId}/tasks/${node.id}`,
- search,
+ search: searchFor(node.id),
}}
>
<TaskName
diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/IterationSwitcherButton.tsx b/airflow-core/src/airflow/ui/src/layouts/Details/IterationSwitcherButton.tsx
new file mode 100644
index 00000000000..17259809b8b
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/layouts/Details/IterationSwitcherButton.tsx
@@ -0,0 +1,121 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { type ReactNode, useState } from "react";
+
+import { Button, VStack } from "@chakra-ui/react";
+import { useTranslation } from "react-i18next";
+import { useLocation, useSearchParams } from "react-router-dom";
+
+import { CrumbSwitcher, type CrumbShape } from "src/components/Breadcrumb";
+import { LoopOptionLabel } from "src/components/LoopOptionLabel";
+
+import { SearchParamsKeys } from "src/constants/searchParams";
+import type { LoopOption } from "src/queries/useLoopFilterOptions";
+
+type Props = {
+ readonly children: ReactNode;
+ /** Offer the whole loop as a choice. A group can show every pass at once; one row cannot. */
+ readonly includeWholeLoop?: boolean;
+ readonly options: Array<LoopOption>;
+ readonly shape: CrumbShape;
+};
+
+/**
+ * The loop iteration level of the breadcrumb, with the loop's passes behind its chevron.
+ *
+ * A pass is part of where you are rather than a filter you applied -- the same task id means a
+ * different run of it in every pass -- so it switches where the current one is named. Picking one
+ * rewrites the query rather than the path, so the level links to itself.
+ */
+export const IterationSwitcherButton = ({ children, includeWholeLoop = false, options, shape }: Props) => {
+ const { t: translate } = useTranslation("common");
+ const [open, setOpen] = useState(false);
+ const [searchParams, setSearchParams] = useSearchParams();
+ const { pathname } = useLocation();
+
+ // A task group can be looked at whole, so it keeps the loop-wide entry; a task instance is
+ // one row and must name a pass. Passes arrive newest first, so the newest leads.
+ const passes = includeWholeLoop ? options : options.filter((option) => option.iteration !== undefined);
+ // A table page carries the pass as a filter, a task instance page as a region coordinate.
+ // Either names the same pass, so whichever is present decides what the level is showing.
+ // A pass belongs to the loop that counted it, so one carried in from a different loop is
+ // dropped rather than read as a position in this one.
+ const carriedLoop = searchParams.get(SearchParamsKeys.LOOP_ID);
+ const sameLoop = carriedLoop === null || carriedLoop === "" || carriedLoop === passes[0]?.loopId;
+ const active = sameLoop
+ ? (searchParams.get(SearchParamsKeys.ITERATION) ?? searchParams.get(SearchParamsKeys.REGION_INDEX))
+ : null;
+ const isActive = (option: LoopOption) =>
+ active === null || active === "" ? option === passes[0] : String(option.iteration) === active;
+
+ const choose = (option: LoopOption) => {
+ const next = new URLSearchParams(searchParams);
+
+ next.set(SearchParamsKeys.LOOP_ID, option.loopId);
+ if (option.iteration === undefined) {
+ next.delete(SearchParamsKeys.ITERATION);
+ } else {
+ next.set(SearchParamsKeys.ITERATION, String(option.iteration));
+ }
+ // The pass is named, never addressed. A coordinate points at one row, and this control does
+ // not know which row a pass became: a mapped expansion lives in its own region, and a clear
+ // forks later passes into a successor one. Dropping the old coordinate lets the page resolve
+ // the pass against the server instead of being sent to a row that may not exist.
+ next.delete(SearchParamsKeys.REGION_ID);
+ next.delete(SearchParamsKeys.REGION_INDEX);
+ // A page of one pass says nothing about where the next one starts, so paging restarts
+ // rather than carrying a cursor into a different set of rows.
+ next.delete(SearchParamsKeys.CURSOR);
+ next.delete(SearchParamsKeys.OFFSET);
+ setSearchParams(next);
+ setOpen(false);
+ };
+
+ return (
+ <CrumbSwitcher
+ // A pass is a number and a badge, so the panel shrinks to them instead of taking the
+ // width a search field needs.
+ contentWidth="auto"
+ label={translate("common:filters.switchIteration")}
+ onOpenChange={setOpen}
+ open={open}
+ search={
+ <VStack align="stretch" gap={0} maxHeight="320px" minWidth="32" overflowY="auto">
+ {passes.map((option) => (
+ <Button
+ aria-current={isActive(option)}
+ justifyContent="flex-start"
+ key={option.value}
+ onClick={() => choose(option)}
+ size="sm"
+ variant={isActive(option) ? "subtle" : "ghost"}
+ >
+ <LoopOptionLabel hideLoopName option={option} />
+ </Button>
+ ))}
+ </VStack>
+ }
+ shape={shape}
+ testId="switch-iteration"
+ to={`${pathname}?${searchParams.toString()}`}
+ >
+ {children}
+ </CrumbSwitcher>
+ );
+};
diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/NavTabs.tsx b/airflow-core/src/airflow/ui/src/layouts/Details/NavTabs.tsx
index 0659828caa4..6dec35c0bdb 100644
--- a/airflow-core/src/airflow/ui/src/layouts/Details/NavTabs.tsx
+++ b/airflow-core/src/airflow/ui/src/layouts/Details/NavTabs.tsx
@@ -19,7 +19,7 @@
import { useRef, type ReactNode } from "react";
import { Center, Flex } from "@chakra-ui/react";
-import { NavLink, useLocation } from "react-router-dom";
+import { NavLink, useLocation, useResolvedPath } from "react-router-dom";
import { useContainerWidth } from "src/utils";
@@ -38,12 +38,51 @@ type Props = {
const INDICATOR_HEIGHT = "2px";
+const trimTrailingSlash = (path: string) => path.replace(/\/$/u, "");
+
+const NavTabLink = ({
+ containerWidth,
+ icon,
+ label,
+ matchPaths,
+ search,
+ value,
+}: { readonly containerWidth: number } & NavTab) => {
+ const { pathname } = useLocation();
+ const resolved = useResolvedPath({ pathname: value, search });
+ // NavLink compares the raw location against its resolved target, and a run id carries ":" and
+ // "+". Landing on a link where those arrived percent-encoded leaves the two spellings of the
+ // same path looking different, so the tab for the page you are on reads as inactive.
+ const current = trimTrailingSlash(decodeURIComponent(pathname));
+ const target = trimTrailingSlash(decodeURIComponent(resolved.pathname));
+ const lastSegment = current.split("/").pop() ?? "";
+ const active = current === target || (matchPaths ?? []).includes(lastSegment);
+
+ return (
+ <NavLink end title={label} to={{ pathname: value, search }}>
+ <Center
+ _focus={{ color: active ? "fg" : "brand.solid" }}
+ _hover={{ color: active ? "fg" : "brand.solid" }}
+ aria-current={active ? "page" : undefined}
+ borderBottomColor={active ? "brand.solid" : "transparent"}
+ borderBottomWidth={INDICATOR_HEIGHT}
+ color={active ? "fg" : "fg.muted"}
+ fontSize="md"
+ fontWeight={active ? "bold" : "medium"}
+ height="40px"
+ mb={`-${INDICATOR_HEIGHT}`}
+ px={4}
+ transition="all 0.2s ease"
+ >
+ {containerWidth > 600 || !Boolean(icon) ? label : icon}
+ </Center>
+ </NavLink>
+ );
+};
+
export const NavTabs = ({ tabs }: Props) => {
const containerRef = useRef<HTMLDivElement>(null);
const containerWidth = useContainerWidth(containerRef);
- const { pathname } = useLocation();
- // Last path segment, e.g. "task-store" or "xcom"
- const lastSegment = pathname.split("/").pop() ?? "";
return (
<Flex
@@ -53,43 +92,9 @@ export const NavTabs = ({ tabs }: Props) => {
mb={2}
ref={containerRef}
>
- {tabs.map(({ icon, label, matchPaths, search, value }) => {
- const isPathMatch = (matchPaths ?? []).includes(lastSegment);
-
- return (
- <NavLink
- end
- key={value}
- title={label}
- to={{
- pathname: value,
- search,
- }}
- >
- {({ isActive }) => {
- const active = isActive || isPathMatch;
-
- return (
- <Center
- _focus={{ color: active ? "fg" : "brand.solid" }}
- _hover={{ color: active ? "fg" : "brand.solid" }}
- borderBottomColor={active ? "brand.solid" : "transparent"}
- borderBottomWidth={INDICATOR_HEIGHT}
- color={active ? "fg" : "fg.muted"}
- fontSize="md"
- fontWeight={active ? "bold" : "medium"}
- height="40px"
- mb={`-${INDICATOR_HEIGHT}`} // Show the border on top of its parent's border
- px={4}
- transition="all 0.2s ease"
- >
- {containerWidth > 600 || !Boolean(icon) ? label : icon}
- </Center>
- );
- }}
- </NavLink>
- );
- })}
+ {tabs.map((tab) => (
+ <NavTabLink containerWidth={containerWidth} key={tab.value} {...tab} />
+ ))}
</Flex>
);
};
diff --git a/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/GroupTaskInstance.tsx b/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/GroupTaskInstance.tsx
index 96822165b62..1b7110d0a23 100644
--- a/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/GroupTaskInstance.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/GroupTaskInstance.tsx
@@ -38,7 +38,6 @@ export const GroupTaskInstance = () => {
const gridTISummaries = summariesByRunId.get(runId);
const taskInstance = gridTISummaries?.task_instances.find((ti) => ti.task_id === groupId);
- // A looped group's iteration view lives on this same tab -- see GroupTaskInstances.
const tabs = [{ icon: <MdOutlineTask />, label: translate("tabs.taskInstances"), value: "" }];
return (
diff --git a/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/GroupTaskInstances.test.tsx b/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/GroupTaskInstances.test.tsx
deleted file mode 100644
index 02766459091..00000000000
--- a/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/GroupTaskInstances.test.tsx
+++ /dev/null
@@ -1,93 +0,0 @@
-/*!
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-import { render } from "@testing-library/react";
-import { MemoryRouter, Route, Routes } from "react-router-dom";
-import { beforeEach, describe, expect, it, vi } from "vitest";
-
-import { useIsLoopGroup } from "src/queries/useIsLoopGroup";
-import { useLoopSummary } from "src/queries/useLoopSummary";
-
-import { TaskInstances } from "../TaskInstances";
-import { GroupTaskInstances } from "./GroupTaskInstances";
-
-vi.mock("src/queries/useIsLoopGroup", () => ({ useIsLoopGroup: vi.fn() }));
-vi.mock("src/queries/useLoopSummary", () => ({ useLoopSummary: vi.fn() }));
-vi.mock("../TaskInstances", () => ({ TaskInstances: vi.fn(() => undefined) }));
-
-const renderPage = () =>
- render(
- <MemoryRouter initialEntries={["/dags/dag/runs/run/tasks/group"]}>
- <Routes>
- <Route element={<GroupTaskInstances />} path="/dags/:dagId/runs/:runId/tasks/:groupId" />
- </Routes>
- </MemoryRouter>,
- );
-
-describe("GroupTaskInstances", () => {
- beforeEach(() => {
- vi.mocked(TaskInstances).mockClear();
- });
-
- it.each([
- { label: "while the summary is loading", summary: { data: undefined } },
- { label: "when the summary failed to load", summary: { data: undefined, isError: true } },
- ])("filters a loop group's list by its id $label", ({ summary }) => {
- vi.mocked(useIsLoopGroup).mockReturnValue(true);
- vi.mocked(useLoopSummary).mockReturnValue(summary as ReturnType<typeof useLoopSummary>);
-
- renderPage();
-
- expect(TaskInstances).toHaveBeenCalledWith(
- expect.objectContaining({ extraFilter: undefined, loopGroupId: "group" }),
- undefined,
- );
- });
-
- it.each([
- { expected: undefined, label: "holds the list while the summary loads", summary: { data: undefined } },
- { expected: "all", label: "lists all iterations when the summary failed", summary: { isError: true } },
- {
- expected: "1",
- label: "defaults to the newest iteration",
- summary: { data: { iterations: [{ index: 0 }, { index: 1 }] } },
- },
- ])("$label without an iteration in the URL", ({ expected, summary }) => {
- vi.mocked(useIsLoopGroup).mockReturnValue(true);
- vi.mocked(useLoopSummary).mockReturnValue(summary as ReturnType<typeof useLoopSummary>);
-
- renderPage();
-
- expect(TaskInstances).toHaveBeenCalled();
- expect(vi.mocked(TaskInstances).mock.calls.map(([props]) => props?.iteration)).toEqual(
- vi.mocked(TaskInstances).mock.calls.map(() => expected),
- );
- });
-
- it("does not filter a plain group's list by loop", () => {
- vi.mocked(useIsLoopGroup).mockReturnValue(false);
- vi.mocked(useLoopSummary).mockReturnValue({ data: undefined } as ReturnType<typeof useLoopSummary>);
-
- renderPage();
-
- expect(TaskInstances).toHaveBeenCalledWith(
- expect.objectContaining({ loopGroupId: undefined }),
- undefined,
- );
- });
-});
diff --git a/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/GroupTaskInstances.tsx b/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/GroupTaskInstances.tsx
index e17543bc386..4494262dafb 100644
--- a/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/GroupTaskInstances.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/GroupTaskInstances.tsx
@@ -16,32 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-import { useParams, useSearchParams } from "react-router-dom";
-
-import { SearchParamsKeys } from "src/constants/searchParams";
-import { useIsLoopGroup } from "src/queries/useIsLoopGroup";
-import { useLoopSummary } from "src/queries/useLoopSummary";
-
import { TaskInstances } from "../TaskInstances";
-import { IterationSelect } from "./LoopIterations";
-import { getSelectedIteration, ITERATION_ALL } from "./LoopIterations/loopUtils";
-
-export const GroupTaskInstances = () => {
- const { dagId = "", groupId = "", runId = "" } = useParams();
- const [searchParams] = useSearchParams();
- const isLoopGroup = useIsLoopGroup(groupId);
- const { data: summary, isError } = useLoopSummary({ dagId, groupId, runId });
- const param = searchParams.get(SearchParamsKeys.ITERATION);
- const iteration =
- summary === undefined
- ? (param ?? (isError ? ITERATION_ALL : undefined))
- : getSelectedIteration(param, summary);
- return (
- <TaskInstances
- extraFilter={summary === undefined ? undefined : <IterationSelect summary={summary} />}
- iteration={iteration}
- loopGroupId={isLoopGroup ? groupId : undefined}
- />
- );
-};
+/** The group's task instances; the loop strip and filter above the table scope them to a pass. */
+export const GroupTaskInstances = () => <TaskInstances />;
diff --git a/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/LoopIterations/IterationSelect.test.tsx b/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/LoopIterations/IterationSelect.test.tsx
deleted file mode 100644
index dcd75072dac..00000000000
--- a/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/LoopIterations/IterationSelect.test.tsx
+++ /dev/null
@@ -1,107 +0,0 @@
-/*!
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-import "@testing-library/jest-dom/vitest";
-import { fireEvent, render, screen, waitFor } from "@testing-library/react";
-import { MemoryRouter, useLocation } from "react-router-dom";
-import { describe, expect, it } from "vitest";
-
-import type { LoopSummaryResponse } from "openapi/requests/types.gen";
-
-import { BaseWrapper } from "src/utils/Wrapper";
-
-import { IterationSelect } from "./IterationSelect";
-
-const summary: LoopSummaryResponse = {
- dag_id: "dag",
- group_id: "loop",
- iterations: [0, 1].map((index) => ({ end_date: null, index, start_date: null, state: "success" })),
- iterations_ran: 2,
- loop_region_id: "region-one",
- loop_regions: [{ region_id: "region-one" }],
- max_iterations: 5,
- run_id: "run",
- status: "running",
-};
-
-const Location = () => <output data-testid="location">{useLocation().search}</output>;
-const View = ({ data }: { readonly data: LoopSummaryResponse }) => (
- <BaseWrapper>
- <MemoryRouter initialEntries={["/?iteration=all&map_index=7"]}>
- <IterationSelect summary={data} />
- <Location />
- </MemoryRouter>
- </BaseWrapper>
-);
-
-describe("IterationSelect", () => {
- it("shows the newest iteration by default without writing it into the URL", async () => {
- render(
- <BaseWrapper>
- <MemoryRouter initialEntries={["/?map_index=7"]}>
- <IterationSelect summary={summary} />
- <Location />
- </MemoryRouter>
- </BaseWrapper>,
- );
- fireEvent.click(screen.getByTestId("loop-iteration-select"));
- const options = await screen.findAllByRole("option");
-
- expect(options.map((option) => option.getAttribute("aria-selected"))).toEqual(["false", "false", "true"]);
- expect(screen.getByTestId("location").textContent).toBe("?map_index=7");
- });
-
- it("keeps All selected when a new iteration is created", () => {
- const { rerender } = render(<View data={summary} />);
-
- rerender(
- <View
- data={{ ...summary, iterations: [...summary.iterations, { ...summary.iterations[0], index: 2 }] }}
- />,
- );
- expect(screen.getByTestId("location")).toHaveTextContent("iteration=all&map_index=7");
- });
-
- it("selects all iterations without removing the mapped slot filter", async () => {
- render(
- <BaseWrapper>
- <MemoryRouter initialEntries={["/?iteration=1&map_index=7&cursor=old"]}>
- <IterationSelect summary={summary} />
- <Location />
- </MemoryRouter>
- </BaseWrapper>,
- );
- fireEvent.click(screen.getByTestId("loop-iteration-select"));
- fireEvent.click(await screen.findByRole("option", { name: "loop.filter.all" }));
- await waitFor(() => expect(screen.getByTestId("location")).toHaveTextContent("iteration=all"));
- expect(screen.getByTestId("location")).toHaveTextContent("map_index=7");
- expect(screen.getByTestId("location")).not.toHaveTextContent("cursor");
- });
- it("replaces a cleared later iteration with the newest existing iteration and resets paging", async () => {
- render(
- <BaseWrapper>
- <MemoryRouter initialEntries={["/?iteration=9&cursor=old"]}>
- <IterationSelect summary={summary} />
- <Location />
- </MemoryRouter>
- </BaseWrapper>,
- );
- await waitFor(() => expect(screen.getByTestId("location")).toHaveTextContent("iteration=1"));
- expect(screen.getByTestId("location")).not.toHaveTextContent("cursor");
- });
-});
diff --git a/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/LoopIterations/IterationSelect.tsx b/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/LoopIterations/IterationSelect.tsx
deleted file mode 100644
index b2bfa2ac2bb..00000000000
--- a/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/LoopIterations/IterationSelect.tsx
+++ /dev/null
@@ -1,155 +0,0 @@
-/*!
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-import { useEffect } from "react";
-
-import { createListCollection, HStack, Text, VStack } from "@chakra-ui/react";
-import { useTranslation } from "react-i18next";
-import { useSearchParams } from "react-router-dom";
-
-import type { LoopSummaryResponse } from "openapi/requests/types.gen";
-
-import { Select } from "src/system-components";
-
-import { StateBadge } from "src/components/StateBadge";
-
-import { SearchParamsKeys } from "src/constants/searchParams";
-import { useDurationFormat } from "src/utils";
-
-import { getSelectedIteration, ITERATION_ALL } from "./loopUtils";
-
-type Props = {
- readonly summary: LoopSummaryResponse;
-};
-
-export const IterationSelect = ({ summary }: Props) => {
- const { t: translate } = useTranslation(["dag", "common"]);
- const [searchParams, setSearchParams] = useSearchParams();
- const { formatElapsed } = useDurationFormat();
- const param = searchParams.get(SearchParamsKeys.ITERATION);
- const selected = getSelectedIteration(param, summary);
-
- // Only a stale iteration (e.g. cleared away) is rewritten; its cursor paged a list that no longer exists.
- useEffect(() => {
- const regionId = searchParams.get(SearchParamsKeys.LOOP_REGION_ID);
-
- if (param === null || param === selected || (regionId !== null && regionId !== summary.loop_region_id)) {
- return;
- }
- const next = new URLSearchParams(searchParams);
-
- next.set(SearchParamsKeys.ITERATION, selected);
- next.delete(SearchParamsKeys.CURSOR);
- setSearchParams(next, { replace: true });
- }, [param, searchParams, selected, setSearchParams, summary.loop_region_id]);
-
- const options = [
- { detail: undefined, label: translate("loop.filter.all"), state: undefined, value: ITERATION_ALL },
- ...summary.iterations.map((iteration) => ({
- detail: formatElapsed(iteration.start_date, iteration.end_date),
- label: translate("loop.iteration", { index: iteration.index }),
- state: iteration.state,
- value: String(iteration.index),
- })),
- ];
- const collection = createListCollection({ items: options });
- const invocations = createListCollection({
- items: (summary.loop_regions ?? []).map((region, index) => ({
- label: `${summary.group_id} (${index + 1})`,
- value: region.region_id,
- })),
- });
-
- return (
- <HStack gap={2}>
- {invocations.items.length > 1 ? (
- <Select.Root
- collection={invocations}
- onValueChange={({ value }) => {
- const params = new URLSearchParams(searchParams);
- const [regionId] = value;
-
- if (regionId === undefined) {
- return;
- }
- params.set(SearchParamsKeys.LOOP_REGION_ID, regionId);
- params.delete(SearchParamsKeys.ITERATION);
- params.delete(SearchParamsKeys.CURSOR);
- setSearchParams(params);
- }}
- value={
- summary.loop_region_id === null || summary.loop_region_id === undefined
- ? []
- : [summary.loop_region_id]
- }
- width="260px"
- >
- <Select.Trigger>
- <Select.ValueText placeholder={translate("loop.filter.invocation")} />
- </Select.Trigger>
- <Select.Content>
- {invocations.items.map((item) => (
- <Select.Item item={item} key={item.value}>
- {item.label}
- </Select.Item>
- ))}
- </Select.Content>
- </Select.Root>
- ) : undefined}
- <Text color="fg.muted" fontSize="sm" whiteSpace="nowrap">
- {translate("loop.filter.label")}
- </Text>
- <Select.Root
- collection={collection}
- onValueChange={(event) => {
- const [next] = event.value;
- const params = new URLSearchParams(searchParams);
-
- params.set(SearchParamsKeys.ITERATION, next ?? ITERATION_ALL);
- params.delete(SearchParamsKeys.CURSOR);
- setSearchParams(params);
- }}
- value={[selected]}
- width="260px"
- >
- <Select.Trigger dataTestId="loop-iteration-select">
- <Select.ValueText />
- </Select.Trigger>
- <Select.Content>
- {options.map((option) => (
- <Select.Item item={option} key={option.value}>
- <VStack align="start" gap={0}>
- <HStack gap={2}>
- <Text>{option.label}</Text>
- {option.state === null || option.state === undefined ? undefined : (
- <StateBadge state={option.state}>{translate(`common:states.${option.state}`)}</StateBadge>
- )}
- </HStack>
- {option.detail === undefined || option.detail === "" ? undefined : (
- <Text color="fg.muted" fontSize="xs">
- {option.detail}
- </Text>
- )}
- </VStack>
- </Select.Item>
- ))}
- </Select.Content>
- </Select.Root>
- </HStack>
- );
-};
diff --git a/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/LoopIterations/index.ts b/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/LoopIterations/index.ts
deleted file mode 100644
index 66118cea7c7..00000000000
--- a/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/LoopIterations/index.ts
+++ /dev/null
@@ -1,19 +0,0 @@
-/*!
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-export { IterationSelect } from "./IterationSelect";
diff --git a/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/LoopIterations/loopUtils.test.ts b/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/LoopIterations/loopUtils.test.ts
index f18de434a19..287bb05990c 100644
--- a/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/LoopIterations/loopUtils.test.ts
+++ b/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/LoopIterations/loopUtils.test.ts
@@ -23,7 +23,7 @@ import type { LoopSummaryResponse } from "openapi/requests/types.gen";
import commonLocale from "../../../../public/i18n/locales/en/common.json";
import dagLocale from "../../../../public/i18n/locales/en/dag.json";
-import { getSelectedIteration, reasonSentence } from "./loopUtils";
+import { reasonSentence } from "./loopUtils";
const i18n = i18next.createInstance();
@@ -72,15 +72,12 @@ describe("loop locale keys", () => {
it.each([
"dag:loop.outcome",
"dag:loop.iterations",
- "dag:loop.iteration",
+ "dag:loop.iterationLabel",
"dag:loop.rules.behaviour",
"dag:loop.rules.boundedFor",
"dag:loop.rules.whileUntil",
"dag:loop.rules.maxIterations",
"dag:loop.rules.exitCriteria",
- "dag:loop.filter.all",
- "dag:loop.filter.invocation",
- "dag:loop.filter.label",
"dag:loop.history.title",
"dag:loop.history.iterationsAxis",
"dag:loop.history.capLabel",
@@ -91,20 +88,3 @@ describe("loop locale keys", () => {
expect(i18n.exists(key)).toBe(true);
});
});
-
-const ran = (...indexes: Array<number>) =>
- buildSummary({
- iterations: indexes.map((index) => ({ end_date: null, index, start_date: null, state: "success" })),
- });
-
-describe("getSelectedIteration", () => {
- it.each([
- { expected: "1", param: null, summary: ran(0, 1) },
- { expected: "0", param: "0", summary: ran(0, 1) },
- { expected: "all", param: "all", summary: ran(0, 1) },
- { expected: "1", param: "9", summary: ran(0, 1) },
- { expected: "all", param: null, summary: ran() },
- ])("selects $expected for iteration=$param", ({ expected, param, summary }) => {
- expect(getSelectedIteration(param, summary)).toBe(expected);
- });
-});
diff --git a/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/LoopIterations/loopUtils.ts b/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/LoopIterations/loopUtils.ts
index fbfedc15824..1d2833138e6 100644
--- a/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/LoopIterations/loopUtils.ts
+++ b/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/LoopIterations/loopUtils.ts
@@ -19,27 +19,12 @@
import type { LoopSummaryResponse } from "openapi/requests/types.gen";
/** Value of the iteration URL parameter that selects every iteration. */
-export const ITERATION_ALL = "all";
-
-/** The URL's iteration while the loop still has it, otherwise the newest iteration (or all when none ran). */
-export const getSelectedIteration = (param: string | null, summary: LoopSummaryResponse): string => {
- if (
- param !== null &&
- (param === ITERATION_ALL || summary.iterations.some((iteration) => String(iteration.index) === param))
- ) {
- return param;
- }
- const latest = summary.iterations.at(-1)?.index;
-
- return latest === undefined ? ITERATION_ALL : String(latest);
-};
-
export const reasonSentence = (
translate: (key: string, options?: Record<string, unknown>) => string,
summary: LoopSummaryResponse,
): string => {
const options = {
- index: summary.stopped_at_iteration ?? summary.failed_at_iteration ?? 0,
+ index: summary.stopped_at_iteration ?? summary.failed_at_iteration,
max: summary.max_iterations,
ran: summary.iterations_ran,
taskId: summary.reason_task_id,
diff --git a/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/useLoopOutcomeStats.tsx b/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/useLoopOutcomeStats.tsx
index 3a2bcf13f54..4983c94d744 100644
--- a/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/useLoopOutcomeStats.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/GroupTaskInstance/useLoopOutcomeStats.tsx
@@ -20,7 +20,7 @@ import type { ReactNode } from "react";
import { HStack, Icon, Text } from "@chakra-ui/react";
import { useTranslation } from "react-i18next";
-import { FiRepeat } from "react-icons/fi";
+import { MdLoop } from "react-icons/md";
import { useLoopSummary } from "src/queries/useLoopSummary";
@@ -55,7 +55,7 @@ export const useLoopOutcomeStats = ({
label: translate("loop.iterations"),
value: (
<HStack gap={1}>
- <Icon aria-hidden as={FiRepeat} boxSize={3.5} />
+ <Icon aria-hidden as={MdLoop} boxSize={3.5} />
<Text>
{summary.iterations_ran}/{summary.max_iterations}
</Text>
diff --git a/airflow-core/src/airflow/ui/src/pages/Run/Execution.test.tsx b/airflow-core/src/airflow/ui/src/pages/Run/Execution.test.tsx
deleted file mode 100644
index 08c512e59b3..00000000000
--- a/airflow-core/src/airflow/ui/src/pages/Run/Execution.test.tsx
+++ /dev/null
@@ -1,490 +0,0 @@
-/*!
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-import "@testing-library/jest-dom/vitest";
-import { fireEvent, render, screen, waitFor } from "@testing-library/react";
-import { Link, MemoryRouter, Route, Routes } from "react-router-dom";
-import { afterEach, beforeAll, describe, expect, it, vi } from "vitest";
-
-import { DagRunService, DagService, TaskInstanceService } from "openapi/requests";
-import type {
- DAGDetailsResponse,
- DAGRunResponse,
- ExecutionCollectionResponse,
- ExecutionRegionResponse,
- ExecutionTaskResponse,
- TaskInstanceResponse,
-} from "openapi/requests/types.gen";
-
-import i18n from "src/i18n/config";
-import type * as Utils from "src/utils";
-import { BaseWrapper } from "src/utils/Wrapper";
-
-import commonTranslations from "../../../public/i18n/locales/en/common.json";
-import dagTranslations from "../../../public/i18n/locales/en/dag.json";
-import dagsTranslations from "../../../public/i18n/locales/en/dags.json";
-import { Execution } from "./Execution";
-
-vi.mock("src/router", () => ({ taskInstanceRoutes: [] }));
-
-beforeAll(() => {
- i18n.addResourceBundle("en", "dag", dagTranslations, true, true);
- i18n.addResourceBundle("en", "common", commonTranslations, true, true);
- i18n.addResourceBundle("en", "dags", dagsTranslations, true, true);
-});
-
-const refresh = vi.hoisted<{ interval: number | false }>(() => ({ interval: false }));
-
-vi.mock("src/utils", async (importOriginal) => ({
- ...(await importOriginal<typeof Utils>()),
- useAutoRefresh: () => refresh.interval,
-}));
-
-const ROOT = "11111111-1111-4111-8111-111111111111";
-const FORK = "22222222-2222-4222-8222-222222222222";
-const MAPPING = "33333333-3333-4333-8333-333333333333";
-
-const region = (id: string, changes: Partial<ExecutionRegionResponse> = {}): ExecutionRegionResponse => ({
- forked_from_region_id: null,
- id,
- node_id: "body",
- parent_region_id: null,
- parent_region_index: null,
- resumes_from_index: 0,
- ...changes,
-});
-
-const task = (id: string, changes: Partial<ExecutionTaskResponse> = {}): ExecutionTaskResponse => ({
- dag_id: "dag",
- dag_run_id: "run",
- dag_version_id: null,
- duration: null,
- end_date: null,
- id,
- map_index: -1,
- operator: "EmptyOperator",
- region_id: ROOT,
- region_index: 0,
- start_date: null,
- state: "success",
- task_display_name: id,
- task_id: id,
- try_number: 1,
- ...changes,
-});
-
-const affectedTask = {
- dag_id: "dag",
- dag_run_id: "run",
- id: "work",
- map_index: -1,
- region_id: ROOT,
- region_index: 0,
- state: "success",
- task_id: "work",
-} as TaskInstanceResponse;
-
-const renderExecution = (search = "") =>
- render(
- <BaseWrapper>
- <MemoryRouter initialEntries={[`/dags/dag/runs/run/execution${search}`]}>
- <Routes>
- <Route element={<Execution />} path="/dags/:dagId/runs/:runId/execution" />
- </Routes>
- </MemoryRouter>
- </BaseWrapper>,
- );
-
-afterEach(() => {
- vi.restoreAllMocks();
- refresh.interval = false;
-});
-
-const mockRun = (state: DAGRunResponse["state"]) =>
- vi.spyOn(DagRunService, "getDagRun").mockResolvedValue({
- dag_id: "dag",
- dag_versions: [{ id: "version", version_number: 1 }],
- state,
- } as DAGRunResponse);
-
-const sentRealClear = (clear: { mock: { calls: Array<Array<unknown>> } }) =>
- (clear.mock.calls as Array<[{ requestBody: { dry_run?: boolean } }]>).some(
- ([request]) => request.requestBody.dry_run === false,
- );
-
-describe("Run Execution", () => {
- it("resets selected UUIDs when navigating to another run", async () => {
- vi.spyOn(DagRunService, "getExecution").mockResolvedValue({
- regions: [region(ROOT)],
- task_instances: [task("work")],
- total_entries: 1,
- });
- render(
- <BaseWrapper>
- <MemoryRouter initialEntries={["/dags/dag/runs/first/execution"]}>
- <Link to="/dags/dag/runs/second/execution">Second run</Link>
- <Routes>
- <Route element={<Execution />} path="/dags/:dagId/runs/:runId/execution" />
- </Routes>
- </MemoryRouter>
- </BaseWrapper>,
- );
- fireEvent.click(await screen.findByRole("checkbox", { name: "Select work (body · Iteration 0)" }));
- await waitFor(() =>
- expect(screen.getByRole("button", { name: "Clear selected executions" })).toBeEnabled(),
- );
- fireEvent.click(screen.getByRole("link", { name: "Second run" }));
- expect(screen.getByRole("button", { name: "Clear selected executions" })).toBeDisabled();
- });
- it("clears selected UUIDs separately when public task coordinates collide", async () => {
- vi.spyOn(DagRunService, "getExecution").mockResolvedValue({
- regions: [region(ROOT), region(FORK)],
- task_instances: [
- task("first", { task_display_name: "work", task_id: "work" }),
- task("second", { region_id: FORK, region_index: 1, task_display_name: "work", task_id: "work" }),
- ],
- total_entries: 2,
- });
- const clear = vi
- .spyOn(TaskInstanceService, "postClearTaskInstances")
- .mockResolvedValue({ task_instances: [affectedTask], total_entries: 1 });
-
- vi.spyOn(DagService, "getDagDetails").mockResolvedValue({ dag_id: "dag" } as DAGDetailsResponse);
- mockRun("success");
- renderExecution();
- fireEvent.click(await screen.findByRole("checkbox", { name: "Select work (body · Iteration 0)" }));
- fireEvent.click(screen.getByRole("checkbox", { name: "Select work (body · Iteration 1)" }));
- await waitFor(() =>
- expect(screen.getByRole("button", { name: "Clear selected executions" })).toBeEnabled(),
- );
- fireEvent.click(screen.getByRole("button", { name: "Clear selected executions" }));
- const confirm = () => screen.getByRole("button", { name: "Confirm" });
-
- await waitFor(() => expect(confirm()).toBeEnabled());
- fireEvent.click(confirm());
- await waitFor(() => expect(sentRealClear(clear)).toBe(true));
- expect(clear.mock.lastCall?.[0].requestBody).toMatchObject({
- dry_run: false,
- task_instance_ids: ["first", "second"],
- });
- });
- it("keeps running branches and retained regions visible beside newer loop passes", async () => {
- const response: ExecutionCollectionResponse = {
- regions: [region(ROOT), region(FORK, { forked_from_region_id: ROOT, resumes_from_index: 2 })],
- task_instances: [
- task("early", { state: "running" }),
- task("retained", { region_index: 3 }),
- task("gate", { operator: "LoopGateOperator", region_id: FORK, region_index: 2 }),
- task("setup", { region_id: undefined, region_index: undefined }),
- ],
- total_entries: 4,
- };
-
- vi.spyOn(DagRunService, "getExecution").mockResolvedValue(response);
- renderExecution();
-
- expect(await screen.findByRole("heading", { name: "body · Iteration 0" })).toBeVisible();
- expect(screen.getByRole("heading", { name: "body · Iteration 2" })).toBeVisible();
- expect(screen.getByRole("heading", { name: "body · Iteration 3" })).toBeVisible();
- const gate = screen.getByRole("link", { name: "gate" });
-
- expect(gate).toHaveAttribute("href", expect.stringContaining(`region_id=${FORK}`));
- expect(gate).toHaveAttribute("href", expect.stringContaining("region_index=2"));
- expect(gate).toHaveAttribute("href", expect.not.stringContaining("try_number"));
- expect(screen.getByRole("link", { name: "setup" })).toHaveAttribute(
- "href",
- "/dags/dag/runs/run/tasks/setup/logs",
- );
- expect(screen.getByRole("link", { name: "early" })).toBeVisible();
- expect(screen.queryByRole("heading", { name: "body · Iteration 4" })).not.toBeInTheDocument();
- });
-
- it.each(["body.map", "body.mapped.member"])(
- "expands %s positions within loop iteration 3 and preserves skipped empty mapping",
- async (taskId) => {
- vi.spyOn(DagRunService, "getExecution").mockResolvedValue({
- regions: [
- region(ROOT),
- region(MAPPING, { node_id: taskId, parent_region_id: ROOT, parent_region_index: 3 }),
- region("empty-mapping", { node_id: "empty", parent_region_id: ROOT, parent_region_index: 3 }),
- ],
- task_instances: [
- task("map-zero", {
- map_index: 0,
- region_id: MAPPING,
- region_index: 0,
- task_display_name: "map",
- task_id: taskId,
- }),
- task("map-one", {
- map_index: 1,
- region_id: MAPPING,
- region_index: 1,
- task_display_name: "map",
- task_id: taskId,
- }),
- task("empty", { region_id: "empty-mapping", region_index: -1, state: "skipped" }),
- ],
- total_entries: 3,
- });
- renderExecution();
-
- fireEvent.click(await screen.findByRole("button", { name: "map (2 mapped tasks on this page)" }));
- expect(screen.getByRole("link", { name: "map [0]" })).toHaveAttribute(
- "href",
- expect.stringContaining("/mapped/0/logs"),
- );
- expect(screen.getByRole("link", { name: "map [1]" })).toHaveAttribute(
- "href",
- expect.stringContaining("region_index=1"),
- );
- expect(screen.getByRole("link", { name: "empty" })).toBeVisible();
- expect(screen.getByRole("heading", { name: "body · Iteration 3" })).toBeVisible();
- },
- );
-
- it("pages using the API total without presenting a page as the complete loop", async () => {
- const fetch = vi
- .spyOn(DagRunService, "getExecution")
- .mockResolvedValueOnce({
- regions: [region(ROOT)],
- task_instances: [task("first")],
- total_entries: 101,
- })
- .mockResolvedValueOnce({
- regions: [region(ROOT)],
- task_instances: [task("second")],
- total_entries: 101,
- });
-
- renderExecution();
- expect(await screen.findByText("Showing 1–1 of 101 executions")).toBeVisible();
- fireEvent.click(screen.getByRole("button", { name: /next page/iu }));
- await waitFor(() =>
- expect(fetch).toHaveBeenLastCalledWith(expect.objectContaining({ limit: 100, offset: 100 })),
- );
- expect(await screen.findByRole("link", { name: "second" })).toBeVisible();
- expect(screen.queryByRole("link", { name: "first" })).not.toBeInTheDocument();
- expect(screen.getByText("Showing 101–101 of 101 executions")).toBeVisible();
- });
-
- it("translates the state badge, including the no-state fallback", async () => {
- vi.spyOn(DagRunService, "getExecution").mockResolvedValue({
- regions: [region(ROOT)],
- task_instances: [task("done"), task("unstarted", { state: null })],
- total_entries: 2,
- });
- renderExecution();
-
- expect(await screen.findByText("Success")).toBeVisible();
- expect(screen.getByText("No Status")).toBeVisible();
- });
-
- it("shows an empty state instead of a zero range for an empty run", async () => {
- vi.spyOn(DagRunService, "getExecution").mockResolvedValue({
- regions: [],
- task_instances: [],
- total_entries: 0,
- });
- renderExecution();
-
- expect(await screen.findByText("No executions in this run.")).toBeVisible();
- expect(screen.queryByText(/Showing/u)).not.toBeInTheDocument();
- });
-
- it("clamps an offset past the end of the live set to the last page", async () => {
- const fetch = vi
- .spyOn(DagRunService, "getExecution")
- .mockResolvedValueOnce({ regions: [region(ROOT)], task_instances: [], total_entries: 101 })
- .mockResolvedValue({ regions: [region(ROOT)], task_instances: [task("last")], total_entries: 101 });
-
- renderExecution("?execution_offset=500");
-
- expect(await screen.findByRole("link", { name: "last" })).toBeVisible();
- expect(fetch).toHaveBeenLastCalledWith(expect.objectContaining({ offset: 100 }));
- });
-
- it("keeps polling a pending run but not a finished one", async () => {
- refresh.interval = 20;
- const fetch = vi
- .spyOn(DagRunService, "getExecution")
- .mockResolvedValue({ regions: [region(ROOT)], task_instances: [task("work")], total_entries: 1 });
-
- mockRun("success");
- const finished = renderExecution();
-
- await screen.findByRole("link", { name: "work" });
- await new Promise((resolve) => {
- setTimeout(resolve, 200);
- });
- expect(fetch).toHaveBeenCalledTimes(1);
- finished.unmount();
- fetch.mockClear();
-
- mockRun("running");
- renderExecution();
- await waitFor(() => expect(fetch.mock.calls.length).toBeGreaterThan(2));
- });
-
- it("keeps polling a finished run until its last pending task has settled", async () => {
- refresh.interval = 20;
- const fetch = vi.spyOn(DagRunService, "getExecution").mockResolvedValue({
- regions: [region(ROOT)],
- task_instances: [task("work", { state: "running" })],
- total_entries: 1,
- });
-
- mockRun("success");
- renderExecution();
-
- await screen.findByRole("link", { name: "work" });
- await waitFor(() => expect(fetch.mock.calls.length).toBeGreaterThan(2));
- });
-
- it("keeps the selection when the clear dialog is cancelled and drops it once the clear succeeds", async () => {
- vi.spyOn(DagRunService, "getExecution").mockResolvedValue({
- regions: [region(ROOT)],
- task_instances: [task("work")],
- total_entries: 1,
- });
- const clear = vi
- .spyOn(TaskInstanceService, "postClearTaskInstances")
- .mockResolvedValue({ task_instances: [affectedTask], total_entries: 1 });
-
- vi.spyOn(DagService, "getDagDetails").mockResolvedValue({ dag_id: "dag" } as DAGDetailsResponse);
- mockRun("success");
- renderExecution();
- fireEvent.click(await screen.findByRole("checkbox", { name: "Select work (body · Iteration 0)" }));
- await waitFor(() =>
- expect(screen.getByRole("button", { name: "Clear selected executions" })).toBeEnabled(),
- );
- fireEvent.click(screen.getByRole("button", { name: "Clear selected executions" }));
- fireEvent.click(await screen.findByRole("button", { name: "Cancel" }));
- await waitFor(() => expect(screen.queryByRole("button", { name: "Cancel" })).not.toBeInTheDocument());
- expect(screen.getByRole("button", { name: "Clear selected executions" })).toBeEnabled();
-
- fireEvent.click(screen.getByRole("button", { name: "Clear selected executions" }));
- await screen.findByRole("button", { name: "Cancel" });
- const confirm = () => screen.getByRole("button", { name: "Confirm" });
-
- await waitFor(() => expect(confirm()).toBeEnabled());
- fireEvent.click(confirm());
- await waitFor(() => expect(sentRealClear(clear)).toBe(true));
- await waitFor(() =>
- expect(screen.getByRole("button", { name: "Clear selected executions" })).toBeDisabled(),
- );
- });
-
- it("offers the loop options for a selected loop execution and addresses it by id", async () => {
- vi.spyOn(DagRunService, "getExecution").mockResolvedValue({
- regions: [region(ROOT)],
- task_instances: [task("work")],
- total_entries: 1,
- });
- const clear = vi
- .spyOn(TaskInstanceService, "postClearTaskInstances")
- .mockResolvedValue({ task_instances: [affectedTask], total_entries: 1 });
-
- renderExecution();
- fireEvent.click(await screen.findByRole("checkbox", { name: "Select work (body · Iteration 0)" }));
- await waitFor(() =>
- expect(screen.getByRole("button", { name: "Clear selected executions" })).toBeEnabled(),
- );
- fireEvent.click(screen.getByRole("button", { name: "Clear selected executions" }));
-
- expect(await screen.findByRole("button", { name: "Later iterations" })).toBeVisible();
- await waitFor(() =>
- expect(clear.mock.lastCall?.[0].requestBody).toMatchObject({
- include_later_loop_iterations: true,
- task_instance_ids: ["work"],
- }),
- );
- });
-
- it("clears a plain mapped execution by task and map index without loop options", async () => {
- vi.spyOn(DagRunService, "getExecution").mockResolvedValue({
- regions: [region(MAPPING, { node_id: "plain" })],
- task_instances: [
- task("plain-0", { map_index: 0, region_id: MAPPING, region_index: 0, task_id: "plain" }),
- ],
- total_entries: 1,
- });
- const clear = vi
- .spyOn(TaskInstanceService, "postClearTaskInstances")
- .mockResolvedValue({ task_instances: [affectedTask], total_entries: 1 });
-
- renderExecution();
- fireEvent.click(await screen.findByRole("button", { name: "plain-0 (1 mapped task on this page)" }));
- fireEvent.click(await screen.findByRole("checkbox", { name: "Select plain-0 [0]" }));
- await waitFor(() =>
- expect(screen.getByRole("button", { name: "Clear selected executions" })).toBeEnabled(),
- );
- fireEvent.click(screen.getByRole("button", { name: "Clear selected executions" }));
-
- await waitFor(() =>
- expect(clear.mock.lastCall?.[0].requestBody).toMatchObject({ task_ids: [["plain", 0]] }),
- );
- expect(clear.mock.lastCall?.[0].requestBody).not.toHaveProperty("task_instance_ids");
- expect(screen.queryByRole("button", { name: "Later iterations" })).not.toBeInTheDocument();
- });
-
- it("selects the current execution once a selected one has been retried", async () => {
- refresh.interval = 20;
- vi.spyOn(DagRunService, "getExecution")
- .mockResolvedValueOnce({
- regions: [region(ROOT)],
- task_instances: [task("old-id", { state: "running", task_id: "work" })],
- total_entries: 1,
- })
- .mockResolvedValue({
- regions: [region(ROOT)],
- task_instances: [task("new-id", { state: "running", task_id: "work" })],
- total_entries: 1,
- });
- const clear = vi
- .spyOn(TaskInstanceService, "postClearTaskInstances")
- .mockResolvedValue({ task_instances: [], total_entries: 1 });
-
- mockRun("running");
- renderExecution();
- fireEvent.click(await screen.findByRole("checkbox", { name: "Select old-id (body · Iteration 0)" }));
- await waitFor(() =>
- expect(screen.getByRole("checkbox", { name: "Select new-id (body · Iteration 0)" })).toBeChecked(),
- );
- fireEvent.click(screen.getByRole("button", { name: "Clear selected executions" }));
-
- await waitFor(() =>
- expect(clear.mock.lastCall?.[0].requestBody).toMatchObject({ task_instance_ids: ["new-id"] }),
- );
- });
-
- it("keeps a selection made on another page when this page does not list it", async () => {
- vi.spyOn(DagRunService, "getExecution")
- .mockResolvedValueOnce({ regions: [region(ROOT)], task_instances: [task("first")], total_entries: 101 })
- .mockResolvedValue({ regions: [region(ROOT)], task_instances: [task("second")], total_entries: 101 });
-
- renderExecution();
- fireEvent.click(await screen.findByRole("checkbox", { name: "Select first (body · Iteration 0)" }));
- fireEvent.click(screen.getByRole("button", { name: /next page/iu }));
- expect(
- await screen.findByRole("checkbox", { name: "Select second (body · Iteration 0)" }),
- ).not.toBeChecked();
-
- expect(screen.getByRole("button", { name: "Clear selected executions" })).toBeEnabled();
- });
-});
diff --git a/airflow-core/src/airflow/ui/src/pages/Run/Execution.tsx b/airflow-core/src/airflow/ui/src/pages/Run/Execution.tsx
deleted file mode 100644
index f757eaf3b06..00000000000
--- a/airflow-core/src/airflow/ui/src/pages/Run/Execution.tsx
+++ /dev/null
@@ -1,325 +0,0 @@
-/*!
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-import { useEffect, useState } from "react";
-
-import { Box, Button, Heading, HStack, Link, Stack, Text } from "@chakra-ui/react";
-import { useTranslation } from "react-i18next";
-import { Link as RouterLink, useParams, useSearchParams } from "react-router-dom";
-
-import { useDagRunServiceGetDagRun, useDagRunServiceGetExecution } from "openapi/queries";
-import type { ExecutionRegionResponse, ExecutionTaskResponse } from "openapi/requests/types.gen";
-
-import { Checkbox, Pagination, ProgressBar } from "src/system-components";
-
-import ClearTaskInstanceDialog from "src/components/Clear/TaskInstance/ClearTaskInstanceDialog";
-import { ErrorAlert } from "src/components/ErrorAlert";
-import { StateBadge } from "src/components/StateBadge";
-
-import { SearchParamsKeys } from "src/constants/searchParams";
-import { isStatePending, useAutoRefresh } from "src/utils";
-import { getTaskInstanceLink } from "src/utils/links";
-
-const PAGE_SIZE = 100;
-
-type Group = {
- index?: number;
- nodeId?: string;
- tasks: Array<ExecutionTaskResponse>;
-};
-
-type SelectedExecution = { in_loop: boolean } & ExecutionTaskResponse;
-
-const getExecutionCoordinates = (task: ExecutionTaskResponse) =>
- JSON.stringify([task.dag_run_id, task.task_id, task.map_index, task.region_id, task.region_index]);
-
-const locateExecution = (task: ExecutionTaskResponse, byRegion: Map<string, ExecutionRegionResponse>) => {
- const region = task.region_id === undefined ? undefined : byRegion.get(task.region_id);
- const mapped = region?.node_id === task.task_id;
- const parentId = region?.parent_region_id;
- const parent = parentId === undefined || parentId === null ? undefined : byRegion.get(parentId);
- const nodeId = mapped ? parent?.node_id : region?.node_id;
- const index =
- nodeId === undefined ? undefined : mapped ? (region.parent_region_index ?? undefined) : task.region_index;
-
- return { index, nodeId };
-};
-
-const groupExecutions = (tasks: Array<ExecutionTaskResponse>, regions: Array<ExecutionRegionResponse>) => {
- const byRegion = new Map(regions.map((region) => [region.id, region]));
- const groups = new Map<string, Group>();
-
- for (const task of tasks) {
- const { index, nodeId } = locateExecution(task, byRegion);
- const key = JSON.stringify([nodeId, index]);
- const group = groups.get(key) ?? { index, nodeId, tasks: [] };
-
- group.tasks.push(task);
- groups.set(key, group);
- }
-
- return [...groups.entries()].sort(
- ([, first], [, second]) =>
- (first.nodeId ?? "").localeCompare(second.nodeId ?? "") || (first.index ?? -1) - (second.index ?? -1),
- );
-};
-
-const toSelectedExecution = (
- task: ExecutionTaskResponse,
- byRegion: Map<string, ExecutionRegionResponse>,
-): SelectedExecution => ({ ...task, in_loop: locateExecution(task, byRegion).nodeId !== undefined });
-
-const getTaskName = (task: ExecutionTaskResponse) =>
- `${task.task_display_name}${task.map_index >= 0 ? ` [${task.map_index}]` : ""}`;
-
-const TaskRow = ({
- onSelect,
- selected,
- selectLabel,
- task,
-}: {
- readonly onSelect: () => void;
- readonly selected: boolean;
- readonly selectLabel: string;
- readonly task: ExecutionTaskResponse;
-}) => {
- const { t: translate } = useTranslation();
-
- return (
- <HStack justify="space-between" py={1}>
- <Checkbox aria-label={selectLabel} checked={selected} onCheckedChange={onSelect} />
- <Link asChild>
- <RouterLink to={getTaskInstanceLink(task, "logs")}>{getTaskName(task)}</RouterLink>
- </Link>
- <StateBadge state={task.state}>{translate(`common:states.${task.state ?? "none"}`)}</StateBadge>
- </HStack>
- );
-};
-
-const ExecutionView = () => {
- const { dagId = "", runId = "" } = useParams();
- const { t: translate } = useTranslation("dag");
- const [searchParams, setSearchParams] = useSearchParams();
- const parsedOffset = Number(searchParams.get(SearchParamsKeys.EXECUTION_OFFSET) ?? 0);
- const offset = Number.isInteger(parsedOffset) && parsedOffset >= 0 ? parsedOffset : 0;
- const refresh = useAutoRefresh({ dagId });
- const { data: dagRun } = useDagRunServiceGetDagRun({ dagId, dagRunId: runId }, undefined, {
- refetchInterval: (query) => isStatePending(query.state.data?.state) && refresh,
- });
- const { data, error, isLoading } = useDagRunServiceGetExecution(
- { dagId, dagRunId: runId, limit: PAGE_SIZE, offset },
- undefined,
- {
- refetchInterval: (query) =>
- (isStatePending(dagRun?.state) ||
- Boolean(query.state.data?.task_instances.some((task) => isStatePending(task.state)))) &&
- refresh,
- },
- );
- const totalEntries = data?.total_entries ?? 0;
-
- useEffect(() => {
- if (data !== undefined && offset > 0 && offset >= totalEntries) {
- setSearchParams(
- (previous) => {
- const updated = new URLSearchParams(previous);
- const lastOffset = Math.max(0, Math.ceil(totalEntries / PAGE_SIZE) - 1) * PAGE_SIZE;
-
- if (lastOffset === 0) {
- updated.delete(SearchParamsKeys.EXECUTION_OFFSET);
- } else {
- updated.set(SearchParamsKeys.EXECUTION_OFFSET, String(lastOffset));
- }
-
- return updated;
- },
- { replace: true },
- );
- }
- }, [data, offset, setSearchParams, totalEntries]);
- const [expanded, setExpanded] = useState(new Set<string>());
- const [selected, setSelected] = useState(new Map<string, SelectedExecution>());
- const [clearing, setClearing] = useState(false);
-
- useEffect(() => {
- if (data === undefined) {
- return;
- }
- const byRegion = new Map(data.regions.map((region) => [region.id, region]));
- const current = new Map(
- data.task_instances.map((task) => [getExecutionCoordinates(task), toSelectedExecution(task, byRegion)]),
- );
-
- setSelected((previous) => {
- let replaced = false;
- const next = new Map<string, SelectedExecution>();
-
- for (const [id, task] of previous) {
- const replacement = current.get(getExecutionCoordinates(task));
-
- if (replacement !== undefined && replacement.id !== id) {
- replaced = true;
- next.set(replacement.id, replacement);
- } else {
- next.set(id, task);
- }
- }
-
- return replaced ? next : previous;
- });
- }, [data]);
- const regionsById = new Map((data?.regions ?? []).map((region) => [region.id, region]));
- const row = (group: Group, task: ExecutionTaskResponse) => (
- <TaskRow
- key={task.id}
- onSelect={() =>
- setSelected((previous) => {
- const updated = new Map(previous);
-
- if (updated.has(task.id)) {
- updated.delete(task.id);
- } else {
- updated.set(task.id, toSelectedExecution(task, regionsById));
- }
-
- return updated;
- })
- }
- selected={selected.has(task.id)}
- selectLabel={translate("execution.select", {
- task:
- group.nodeId === undefined
- ? getTaskName(task)
- : `${getTaskName(task)} (${translate("execution.iteration", { index: group.index, node: group.nodeId })})`,
- })}
- task={task}
- />
- );
- const groups = groupExecutions(data?.task_instances ?? [], data?.regions ?? []);
- const page = (next: number) =>
- setSearchParams((previous) => {
- const updated = new URLSearchParams(previous);
-
- updated.set(SearchParamsKeys.EXECUTION_OFFSET, String((next - 1) * PAGE_SIZE));
-
- return updated;
- });
-
- return (
- <Stack gap={4} p={4}>
- <Heading size="lg">{translate("execution.title")}</Heading>
- <Button alignSelf="start" disabled={selected.size === 0} onClick={() => setClearing(true)}>
- {translate("execution.clearSelected")}
- </Button>
- {clearing ? (
- <ClearTaskInstanceDialog
- onCleared={() => setSelected(new Map())}
- onClose={() => setClearing(false)}
- open
- taskInstances={[...selected.values()]}
- />
- ) : undefined}
- <ErrorAlert error={error} />
- {isLoading ? <ProgressBar size="xs" /> : undefined}
- {data !== undefined && totalEntries === 0 ? <Text>{translate("execution.empty")}</Text> : undefined}
- {totalEntries > 0 ? (
- <Text>
- {translate("execution.page", {
- end: Math.min(offset + (data?.task_instances.length ?? 0), totalEntries),
- start: (data?.task_instances.length ?? 0) > 0 ? offset + 1 : 0,
- total: totalEntries,
- })}
- </Text>
- ) : undefined}
- {groups.map(([key, group]) => {
- const tasksById = new Map<string, Array<ExecutionTaskResponse>>();
-
- for (const task of group.tasks) {
- const tasks = tasksById.get(task.task_id) ?? [];
-
- tasks.push(task);
- tasksById.set(task.task_id, tasks);
- }
-
- return (
- <Box borderRadius="md" borderWidth="1px" key={key} p={3}>
- <Heading size="md">
- {group.nodeId === undefined
- ? translate("execution.runTasks")
- : translate("execution.iteration", { index: group.index, node: group.nodeId })}
- </Heading>
- {[...tasksById].map(([taskId, tasks]) => {
- const mapped = tasks.every((task) => task.map_index >= 0);
- const expansionKey = `${key}:${taskId}`;
-
- return mapped ? (
- <Box key={taskId}>
- <Button
- aria-expanded={expanded.has(expansionKey)}
- onClick={() =>
- setExpanded((previous) => {
- const updated = new Set(previous);
-
- if (updated.has(expansionKey)) {
- updated.delete(expansionKey);
- } else {
- updated.add(expansionKey);
- }
-
- return updated;
- })
- }
- variant="ghost"
- >
- {translate("execution.mapped", {
- count: tasks.length,
- task: tasks[0]?.task_display_name,
- })}
- </Button>
- {expanded.has(expansionKey) ? tasks.map((task) => row(group, task)) : undefined}
- </Box>
- ) : (
- tasks.map((task) => row(group, task))
- );
- })}
- </Box>
- );
- })}
- {totalEntries > PAGE_SIZE ? (
- <Pagination.Root
- count={totalEntries}
- onPageChange={(event) => page(event.page)}
- page={Math.floor(offset / PAGE_SIZE) + 1}
- pageSize={PAGE_SIZE}
- >
- <HStack justify="center">
- <Pagination.PrevTrigger />
- <Pagination.Items />
- <Pagination.NextTrigger />
- </HStack>
- </Pagination.Root>
- ) : undefined}
- </Stack>
- );
-};
-
-export const Execution = () => {
- const { dagId, runId } = useParams();
-
- return <ExecutionView key={JSON.stringify([dagId, runId])} />;
-};
diff --git a/airflow-core/src/airflow/ui/src/pages/Run/Run.tsx b/airflow-core/src/airflow/ui/src/pages/Run/Run.tsx
index cff8b60d950..62966082512 100644
--- a/airflow-core/src/airflow/ui/src/pages/Run/Run.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/Run/Run.tsx
@@ -42,7 +42,6 @@ export const Run = () => {
const tabs = [
{ icon: <MdOutlineTask />, label: translate("tabs.taskInstances"), value: "" },
- { icon: <MdOutlineTask />, label: translate("execution.title"), value: "execution" },
{ icon: <FiDatabase />, label: translate("tabs.assetEvents"), value: "asset_events" },
{ icon: <MdOutlineEventNote />, label: translate("tabs.auditLog"), value: "events" },
{ icon: <FiCode />, label: translate("tabs.code"), value: "code" },
diff --git a/airflow-core/src/airflow/ui/src/pages/Task/Overview/LoopHistoryChart.test.tsx b/airflow-core/src/airflow/ui/src/pages/Task/Overview/LoopHistoryChart.test.tsx
index f64b5f50bab..48bafa140c1 100644
--- a/airflow-core/src/airflow/ui/src/pages/Task/Overview/LoopHistoryChart.test.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/Task/Overview/LoopHistoryChart.test.tsx
@@ -124,6 +124,6 @@ describe("LoopHistoryChart", () => {
</BaseWrapper>,
);
fireEvent.click(screen.getByRole("button", { name: "bar" }));
- expect(screen.getByTestId("location")).toHaveTextContent("loop_region_id=selected-nested-region");
+ expect(screen.getByTestId("location")).toHaveTextContent("region_id=selected-nested-region");
});
});
diff --git a/airflow-core/src/airflow/ui/src/pages/Task/Overview/LoopHistoryChart.tsx b/airflow-core/src/airflow/ui/src/pages/Task/Overview/LoopHistoryChart.tsx
index 43c514880de..82e3fe8ec51 100644
--- a/airflow-core/src/airflow/ui/src/pages/Task/Overview/LoopHistoryChart.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/Task/Overview/LoopHistoryChart.tsx
@@ -94,7 +94,7 @@ export const LoopHistoryChart = ({ groupId }: { readonly groupId: string }) => {
const params = new URLSearchParams();
if (run.loop_region_id !== null && run.loop_region_id !== undefined) {
- params.set(SearchParamsKeys.LOOP_REGION_ID, run.loop_region_id);
+ params.set(SearchParamsKeys.REGION_ID, run.loop_region_id);
}
void navigate(`/dags/${dagId}/runs/${run.run_id}/tasks/group/${groupId}?${params}`);
}
diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.tsx
index 85c55143ee1..436f6f7e99b 100644
--- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/Header.tsx
@@ -37,6 +37,7 @@ import { TeamName } from "src/components/TeamName";
import Time from "src/components/Time";
import { useShowTeam } from "src/hooks/useShowTeam";
+import { useLoopSummary } from "src/queries/useLoopSummary";
import { useTaskInstanceNote } from "src/queries/useTaskInstanceNote";
import { useDurationFormat } from "src/utils";
@@ -68,6 +69,12 @@ export const Header = ({ taskInstance }: { readonly taskInstance: TaskInstanceRe
const { formatElapsed, renderDuration } = useDurationFormat();
const { isPending, note, onOpen, onSave, setNote } = useTaskInstanceNote(taskInstance);
const showTeam = useShowTeam(taskInstance.team_name);
+ const loopIteration = taskInstance.loop_iteration;
+ const { data: loopSummary } = useLoopSummary({
+ dagId: taskInstance.dag_id,
+ groupId: loopIteration?.loop_id ?? "",
+ runId: taskInstance.dag_run_id,
+ });
const stats = [
{ label: translate("task.operator"), value: taskInstance.operator_name },
@@ -77,6 +84,14 @@ export const Header = ({ taskInstance }: { readonly taskInstance: TaskInstanceRe
...(taskInstance.try_number > 1
? [{ label: translate("tryNumber"), value: taskInstance.try_number }]
: []),
+ ...(loopIteration === null || loopIteration === undefined
+ ? []
+ : [
+ {
+ label: translate("common:filters.iteration"),
+ value: `${loopIteration.iteration}/${loopSummary?.iterations_ran ?? loopIteration.iteration}`,
+ },
+ ]),
{ label: translate("startDate"), value: <Time datetime={taskInstance.start_date} /> },
{ label: translate("endDate"), value: <Time datetime={taskInstance.end_date} /> },
...(Boolean(taskInstance.start_date)
diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/TaskInstance.test.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstance/TaskInstance.test.tsx
index 8979ca91f2a..b4d18fa47f5 100644
--- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/TaskInstance.test.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/TaskInstance.test.tsx
@@ -405,7 +405,7 @@ describe("TaskInstance", () => {
{ inLoop: false, visible: true },
{ inLoop: true, visible: false },
])(
- "shows the mapped task instances tab only outside a loop (in_loop=$inLoop)",
+ "shows the mapped task instances tab only outside a loop (inLoop=$inLoop)",
async ({ inLoop, visible }) => {
const queryClient = new QueryClient({ defaultOptions: { queries: { retry: false } } });
const regionId = "11111111-1111-4111-8111-111111111111";
diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/TaskInstance.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstance/TaskInstance.tsx
index 75295218e39..4f80be99d26 100644
--- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/TaskInstance.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/TaskInstance.tsx
@@ -38,6 +38,7 @@ import { SearchParamsKeys } from "src/constants/searchParams";
import { useHITLReviewTabs } from "src/hooks/useHITLReviewTabs";
import { usePluginTabs } from "src/hooks/usePluginTabs";
import { type TabItem, useRequiredActionTabs } from "src/hooks/useRequiredActionTabs";
+import { useResolvePassCoordinate } from "src/hooks/useResolvePassCoordinate";
import { useTaskInstanceCoordinates } from "src/hooks/useTaskInstanceCoordinates";
import { useTaskInstanceView } from "src/hooks/useTaskInstanceView";
import { useDefaultTaskInstanceTab } from "src/hooks/useUserSettings";
@@ -54,9 +55,10 @@ export const TaskInstance = () => {
const [searchParams] = useSearchParams();
const location = useLocation();
const coordinates = useTaskInstanceCoordinates();
+ const isResolvingCoordinate = useResolvePassCoordinate();
const { error, historical, historicalTaskInstance, isLoading, liveTaskInstance, taskInstance } =
useTaskInstanceView();
- const isRegional = liveTaskInstance !== undefined && isLoopTaskInstance(liveTaskInstance);
+ const regional = liveTaskInstance !== undefined && isLoopTaskInstance(liveTaskInstance);
const tryNumber = searchParams.get(SearchParamsKeys.TRY_NUMBER);
const coordinateSearch = new URLSearchParams();
@@ -131,7 +133,7 @@ export const TaskInstance = () => {
search: tab.search ?? (coordinateSearch.toString() || undefined),
}));
const newTabs =
- taskInstance && taskInstance.map_index > -1 && !isRegional && !historical
+ taskInstance && taskInstance.map_index > -1 && !regional && !historical
? [
...scopedTabs.slice(0, 1),
{
@@ -174,7 +176,9 @@ export const TaskInstance = () => {
return (
<ReactFlowProvider>
- <DetailsLayout error={error} isLoading={isLoading} tabs={displayTabs}>
+ {/* Held rather than rendered empty: a looped task instance cannot be asked for without a
+ coordinate, so the children must not run their queries until the pass resolves. */}
+ <DetailsLayout error={error} isLoading={isLoading || isResolvingCoordinate} tabs={displayTabs}>
{taskInstance === undefined ? (
isLoading ? undefined : (
<Heading p={2} size="lg">
diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.test.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.test.tsx
index a334d3f77ff..3b0e26d1773 100644
--- a/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.test.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.test.tsx
@@ -70,6 +70,15 @@ vi.mock("openapi/queries", async (importOriginal) => {
};
});
+// The loop filter now only applies where the page is inside that loop, so the table has to
+// know which loops the Dag declares.
+vi.mock("src/queries/useIsLoopGroup", () => ({
+ // Membership comes from the structure now, so the mock answers per node rather than
+ // claiming every page is inside the loop.
+ useLoopGroupIds: (nodeId?: string) =>
+ nodeId === undefined || nodeId === "batch" || nodeId.startsWith("batch.") ? ["batch"] : [],
+}));
+
vi.mock("./TaskInstancesFilter", () => ({
TaskInstancesFilter: () => null,
}));
@@ -238,33 +247,34 @@ describe("TaskInstances", () => {
expect(screen.getByText("eu-west")).toBeInTheDocument();
});
- it("keeps loop iteration filters separate from mapped slot filters", () => {
- mockParams.groupId = "loop";
- mockSearchParams = new URLSearchParams("map_index=1");
+ it("drops a stale iteration that no loop in the URL counts", () => {
+ mockSearchParams = new URLSearchParams("iteration=3");
vi.mocked(useTaskInstanceServiceGetTaskInstances).mockReturnValue(getTaskInstancesResponse([]));
- render(<TaskInstances iteration="3" loopGroupId="loop" />, { wrapper: Wrapper });
+ render(<TaskInstances />, { wrapper: Wrapper });
expect(useTaskInstanceServiceGetTaskInstances).toHaveBeenLastCalledWith(
- expect.objectContaining({
- iteration: 3,
- loopId: "loop",
- mapIndex: [1],
- }),
+ expect.objectContaining({ iteration: undefined, loopId: undefined }),
undefined,
expect.any(Object),
);
});
- it("does not fetch a loop group's list before its iteration is known", () => {
- mockParams.groupId = "loop";
+
+ it("sends the pair once a loop names the pass", () => {
+ // Dotted ids mean a task inside the loop is named for it, which is what puts it in scope.
+ mockParams.taskId = "batch.member";
+ mockSearchParams = new URLSearchParams("loop_id=batch&iteration=3");
vi.mocked(useTaskInstanceServiceGetTaskInstances).mockReturnValue(getTaskInstancesResponse([]));
- render(<TaskInstances loopGroupId="loop" />, { wrapper: Wrapper });
+ render(<TaskInstances />, { wrapper: Wrapper });
expect(useTaskInstanceServiceGetTaskInstances).toHaveBeenLastCalledWith(
- expect.objectContaining({ iteration: undefined, loopId: "loop" }),
+ expect.objectContaining({ iteration: 3, loopId: "batch" }),
undefined,
- expect.objectContaining({ enabled: false }),
+ expect.any(Object),
);
});
- it("ignores stale loop selectors outside a loop group", () => {
- mockSearchParams = new URLSearchParams("iteration=3");
+
+ it("ignores a loop carried in from a page outside it", () => {
+ // The filter now survives navigation, so an unrelated task must not inherit the scope.
+ mockParams.taskId = "unrelated_task";
+ mockSearchParams = new URLSearchParams("loop_id=batch&iteration=3");
vi.mocked(useTaskInstanceServiceGetTaskInstances).mockReturnValue(getTaskInstancesResponse([]));
render(<TaskInstances />, { wrapper: Wrapper });
expect(useTaskInstanceServiceGetTaskInstances).toHaveBeenLastCalledWith(
diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.tsx
index c67159ecd1c..48e33005e3d 100644
--- a/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.tsx
@@ -16,9 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-import type { ReactNode } from "react";
-
-import { Flex, HStack } from "@chakra-ui/react";
+import { Flex } from "@chakra-ui/react";
import type { ColumnDef } from "@tanstack/react-table";
import type { TFunction } from "i18next";
import { useTranslation } from "react-i18next";
@@ -29,8 +27,6 @@ import type { TaskInstanceResponse } from "openapi/requests/types.gen";
import { RouterLink, ActionBar } from "src/system-components";
-import { ITERATION_ALL } from "src/pages/GroupTaskInstance/LoopIterations/loopUtils";
-
import { ClearTaskInstanceButton } from "src/components/Clear";
import { DagVersion } from "src/components/DagVersion";
import { DataTable } from "src/components/DataTable";
@@ -54,6 +50,7 @@ import { TruncatedText } from "src/components/TruncatedText";
import { SearchParamsKeys, type SearchParamsKeysType } from "src/constants/searchParams";
import { useAdvancedSearchArg } from "src/hooks/useAdvancedSearch";
import { useConfig } from "src/queries/useConfig";
+import { useLoopGroupIds } from "src/queries/useIsLoopGroup";
import { formatNumber, useAutoRefresh, isStatePending, useDocumentTitle } from "src/utils";
import { getTaskInstanceLink } from "src/utils/links";
@@ -187,9 +184,20 @@ const taskInstanceColumns = ({
},
...(hasLoopIteration
? [
+ {
+ accessorKey: "loop_iteration.loop_id",
+ cell: ({ row: { original } }: TaskInstanceRow) => original.loop_iteration?.loop_id,
+ // The endpoint has no sort for the loop a row belongs to, only for the pass within it.
+ enableSorting: false,
+ header: translate("common:filters.loop"),
+ id: "loop",
+ },
{
accessorKey: "loop_iteration",
cell: ({ row: { original } }: TaskInstanceRow) => original.loop_iteration?.iteration,
+ // The endpoint has no sort for the pass: a row's own region index is its map slot
+ // once a mapped task runs inside a loop, and the enclosing pass is an ancestry walk
+ // rather than a column.
enableSorting: false,
header: translate("taskInstance.iteration"),
},
@@ -294,16 +302,10 @@ const taskInstanceColumns = ({
},
];
-type TaskInstancesProps = {
- readonly extraFilter?: ReactNode;
- /** Selected iteration of the loop group; undefined holds the list until the default is known. */
- readonly iteration?: string;
- readonly loopGroupId?: string;
-};
-
-export const TaskInstances = ({ extraFilter, iteration, loopGroupId }: TaskInstancesProps = {}) => {
+export const TaskInstances = () => {
const { i18n, t: translate } = useTranslation();
const { dagId, groupId, runId, taskId } = useParams();
+ const loopsHere = useLoopGroupIds(groupId ?? taskId);
// Only the standalone list page owns the tab title; nested tabs inherit their parent page's title.
useDocumentTitle(dagId === undefined ? translate("common:taskInstance_other") : undefined);
@@ -334,7 +336,13 @@ export const TaskInstances = ({ extraFilter, iteration, loopGroupId }: TaskInsta
const logicalDateLte = searchParams.get(LOGICAL_DATE_LTE_PARAM);
const tryNumberFilter = searchParams.get(TRY_NUMBER_PARAM);
const mapIndexFilter = searchParams.get(MAP_INDEX_PARAM);
- const inLoopRun = loopGroupId !== undefined && runId !== undefined;
+ const loopFilter = searchParams.get(SearchParamsKeys.LOOP_ID);
+ // The filter now survives navigation, so a loop carried in from elsewhere is ignored unless
+ // this page is actually inside it. A loop lives in one run, and the endpoint rejects the
+ // filter without one, so the cross-run Task page drops it rather than asking for a 400.
+ const loopInScope = loopFilter !== null && runId !== undefined && loopsHere.includes(loopFilter);
+ const iterationFilter = searchParams.get(SearchParamsKeys.ITERATION);
+ const parsedIteration = iterationFilter === null ? Number.NaN : Number.parseInt(iterationFilter, 10);
const startDate = searchParams.get(START_DATE_PARAM);
const endDate = searchParams.get(END_DATE_PARAM);
const poolNamePattern = searchParams.get(POOL_NAME_PATTERN_PARAM);
@@ -401,12 +409,14 @@ export const TaskInstances = ({ extraFilter, iteration, loopGroupId }: TaskInsta
durationGte: durationGte !== null && durationGte !== "" ? Number(durationGte) : undefined,
durationLte: durationLte !== null && durationLte !== "" ? Number(durationLte) : undefined,
endDateLte: endDate ?? undefined,
- iteration:
- inLoopRun && iteration !== undefined && iteration !== ITERATION_ALL ? Number(iteration) : undefined,
+ // The loop is named by the author; the server turns that into the regions behind it.
+ // A pass number is meaningless without the loop that counts it, and the endpoint rejects
+ // the pair outright, so a stale iteration left in the URL is dropped rather than sent.
+ iteration: loopInScope && Number.isInteger(parsedIteration) ? parsedIteration : undefined,
limit: pagination.pageSize,
logicalDateGte: logicalDateGte ?? undefined,
logicalDateLte: logicalDateLte ?? undefined,
- loopId: runId === undefined ? undefined : loopGroupId,
+ loopId: loopInScope ? loopFilter : undefined,
mapIndex: mapIndexFilter !== null && mapIndexFilter !== "" ? [Number(mapIndexFilter)] : undefined,
...operatorNameArg,
orderBy,
@@ -426,7 +436,6 @@ export const TaskInstances = ({ extraFilter, iteration, loopGroupId }: TaskInsta
},
undefined,
{
- enabled: !inLoopRun || iteration !== undefined,
placeholderData: (prev) => prev,
refetchInterval: (query) =>
query.state.data?.task_instances.some((ti) => isStatePending(ti.state)) ? refetchInterval : false,
@@ -466,18 +475,9 @@ export const TaskInstances = ({ extraFilter, iteration, loopGroupId }: TaskInsta
columns={columns}
data={data?.task_instances ?? []}
errorMessage={<ErrorAlert error={error} />}
- filterActions={
- extraFilter === undefined ? (
- <TaskInstancesFilter />
- ) : (
- <HStack alignItems="flex-start" gap={2} wrap="wrap">
- {extraFilter}
- <TaskInstancesFilter />
- </HStack>
- )
- }
+ filterActions={<TaskInstancesFilter />}
initialState={tableURLState}
- isLoading={isLoading || (inLoopRun && iteration === undefined)}
+ isLoading={isLoading}
modelName="common:taskInstance"
nextCursor={nextCursor}
onStateChange={setTableURLState}
diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstancesFilter.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstancesFilter.tsx
index 72340d12671..7abfba112b1 100644
--- a/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstancesFilter.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstancesFilter.tsx
@@ -16,9 +16,9 @@
* specific language governing permissions and limitations
* under the License.
*/
-import { useSearchParams, useParams } from "react-router-dom";
+import { useParams } from "react-router-dom";
-import { FilterBar, type FilterValue } from "src/components/FilterBar";
+import { FilterBar } from "src/components/FilterBar";
import { SearchParamsKeys, type SearchParamsKeysType } from "src/constants/searchParams";
import { useConfig } from "src/queries/useConfig";
@@ -31,6 +31,7 @@ const {
DURATION_GTE: DURATION_GTE_PARAM,
DURATION_LTE: DURATION_LTE_PARAM,
LOGICAL_DATE_RANGE: LOGICAL_DATE_RANGE_PARAM,
+ LOOP: LOOP_PARAM,
MAP_INDEX: MAP_INDEX_PARAM,
NAME_PATTERN: NAME_PATTERN_PARAM,
OPERATOR_NAME_PATTERN: OPERATOR_NAME_PATTERN_PARAM,
@@ -62,6 +63,11 @@ export const TaskInstancesFilter = () => {
STATE_PARAM as FilterableSearchParamsKeys,
];
+ // A loop lives in one run, so this only makes sense once a run is in scope.
+ if (runId !== undefined) {
+ paramKeys.push(LOOP_PARAM as FilterableSearchParamsKeys);
+ }
+
if (multiTeamEnabled) {
paramKeys.push(TEAMS_PARAM as FilterableSearchParamsKeys);
}
@@ -74,25 +80,10 @@ export const TaskInstancesFilter = () => {
paramKeys.unshift(DAG_ID_PATTERN_PARAM as FilterableSearchParamsKeys);
}
- const [searchParams] = useSearchParams();
-
- const { filterConfigs, handleFiltersChange } = useFiltersHandler(paramKeys);
-
- const initialValues: Record<string, FilterValue> = {};
-
- filterConfigs.forEach((config) => {
- const value = searchParams.get(config.key);
-
- if (value !== null && value !== "") {
- if (config.type === "number") {
- const parsedValue = Number(value);
-
- initialValues[config.key] = isNaN(parsedValue) ? value : parsedValue;
- } else {
- initialValues[config.key] = value;
- }
- }
- });
+ // ``useFiltersHandler`` already reads each filter's value, including composite ones whose key
+ // is not a url param of its own -- the loop pill spans ``loop_id`` and ``iteration``, so
+ // reading its key alone would leave it blank whenever a pass is selected elsewhere.
+ const { filterConfigs, handleFiltersChange, initialValues } = useFiltersHandler(paramKeys);
return (
<FilterBar
diff --git a/airflow-core/src/airflow/ui/src/queries/useGridStructure.ts b/airflow-core/src/airflow/ui/src/queries/useGridStructure.ts
index dc0d47d3a03..28a559424d9 100644
--- a/airflow-core/src/airflow/ui/src/queries/useGridStructure.ts
+++ b/airflow-core/src/airflow/ui/src/queries/useGridStructure.ts
@@ -66,19 +66,25 @@ export const useGridStructure = ({
value: triggeringUser,
});
- const { data: dagStructure, ...rest } = useGridServiceGetDagStructure({
- dagId,
- depth,
- includeDownstream,
- includeUpstream,
- limit,
- orderBy: ["-run_after"],
- root,
- ...runIdPatternArg,
- runType: runType ? [runType] : undefined,
- state: dagRunState ? [dagRunState] : undefined,
- ...triggeringUserArg,
- });
+ const { data: dagStructure, ...rest } = useGridServiceGetDagStructure(
+ {
+ dagId,
+ depth,
+ includeDownstream,
+ includeUpstream,
+ limit,
+ orderBy: ["-run_after"],
+ root,
+ ...runIdPatternArg,
+ runType: runType ? [runType] : undefined,
+ state: dagRunState ? [dagRunState] : undefined,
+ ...triggeringUserArg,
+ },
+ undefined,
+ // The global Task Instances page has no Dag, and asking for the structure of an empty id
+ // is a request that cannot answer anything.
+ { enabled: Boolean(dagId) },
+ );
return { data: dagStructure, ...rest };
};
diff --git a/airflow-core/src/airflow/ui/src/queries/useIsLoopGroup.ts b/airflow-core/src/airflow/ui/src/queries/useIsLoopGroup.ts
index 993b25e7fc3..2a0e25c9606 100644
--- a/airflow-core/src/airflow/ui/src/queries/useIsLoopGroup.ts
+++ b/airflow-core/src/airflow/ui/src/queries/useIsLoopGroup.ts
@@ -30,3 +30,75 @@ export const useLoopGroupNode = (groupId: string): GridNodeResponse | undefined
/** Whether the given Task Group is a looped group (``is_loop``), from the cached grid structure. */
export const useIsLoopGroup = (groupId: string): boolean => Boolean(useLoopGroupNode(groupId)?.is_loop);
+
+/** Every loop declared in the Dag, so a filter can offer them by the name the author gave. */
+/**
+ * The loops enclosing `nodeId`, or every loop in the Dag when no node is named.
+ *
+ * Membership comes from the structure rather than the id, because a loop declared with
+ * `prefix_group_id=False` does not name its members after itself.
+ */
+export const loopsForNode = (nodes: Array<GridNodeResponse> | undefined, nodeId?: string): Array<string> => {
+ const declared: Array<string> = [];
+ let enclosing: Array<string> | undefined;
+ const walk = (list: Array<GridNodeResponse> | undefined, ancestors: Array<string>) => {
+ for (const node of list ?? []) {
+ const isLoop = node.is_loop === true;
+ const scope = isLoop ? [...ancestors, node.id] : ancestors;
+
+ if (isLoop) {
+ declared.push(node.id);
+ }
+ if (node.id === nodeId) {
+ enclosing = scope;
+ }
+ walk(node.children ?? undefined, scope);
+ }
+ };
+
+ walk(nodes, []);
+
+ return nodeId === undefined ? declared : (enclosing ?? []);
+};
+
+/**
+ * Every task id inside `groupId`, including the group itself.
+ *
+ * Read from the structure rather than the id, because a group declared with
+ * `prefix_group_id=False` does not name its children after itself.
+ */
+export const membersOfGroup = (nodes: Array<GridNodeResponse> | undefined, groupId: string): Set<string> => {
+ const members = new Set<string>();
+ const collect = (node: GridNodeResponse) => {
+ members.add(node.id);
+ for (const child of node.children ?? []) {
+ collect(child);
+ }
+ };
+ const find = (list: Array<GridNodeResponse> | undefined): boolean =>
+ (list ?? []).some((node) => {
+ if (node.id === groupId) {
+ collect(node);
+
+ return true;
+ }
+
+ return find(node.children ?? undefined);
+ });
+
+ find(nodes);
+
+ return members;
+};
+
+export const useGroupMembers = (groupId?: string): Set<string> => {
+ const { data } = useGridStructure({ limit: 1 });
+
+ return groupId === undefined ? new Set<string>() : membersOfGroup(data, groupId);
+};
+
+export const useLoopGroupIds = (nodeId?: string): Array<string> => {
+ const { data } = useGridStructure({ limit: 1 });
+
+ return loopsForNode(data, nodeId);
+};
diff --git a/airflow-core/src/airflow/ui/src/queries/useLoopFilterOptions.test.ts b/airflow-core/src/airflow/ui/src/queries/useLoopFilterOptions.test.ts
new file mode 100644
index 00000000000..c4693477494
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/queries/useLoopFilterOptions.test.ts
@@ -0,0 +1,109 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { describe, expect, it } from "vitest";
+
+import type { GridNodeResponse } from "openapi/requests/types.gen";
+
+import { loopsForNode, membersOfGroup } from "./useIsLoopGroup";
+import { selectActivePass, type LoopOption } from "./useLoopFilterOptions";
+
+const node = (id: string, extra: Partial<GridNodeResponse> = {}): GridNodeResponse =>
+ ({ children: null, id, is_loop: false, label: id, ...extra }) as GridNodeResponse;
+
+// A loop declared with prefix_group_id=False does not name its members after itself, so the
+// structure is the only thing that says what is inside it.
+const STRUCTURE = [
+ node("refine", {
+ children: [node("improve"), node("evaluate")],
+ is_loop: true,
+ }),
+ node("outer", {
+ children: [node("outer.plain"), node("batch", { children: [node("batch.member")], is_loop: true })],
+ }),
+ node("unrelated"),
+];
+
+describe("loopsForNode", () => {
+ it("lists every declared loop when no node is named", () => {
+ expect(loopsForNode(STRUCTURE)).toEqual(["refine", "batch"]);
+ });
+
+ it("finds the loop enclosing a member that is not named after it", () => {
+ expect(loopsForNode(STRUCTURE, "improve")).toEqual(["refine"]);
+ });
+
+ it("treats a loop as being in its own scope", () => {
+ expect(loopsForNode(STRUCTURE, "refine")).toEqual(["refine"]);
+ });
+
+ it("finds a loop nested inside a plain group", () => {
+ expect(loopsForNode(STRUCTURE, "batch.member")).toEqual(["batch"]);
+ });
+
+ it("gives nothing for a node no loop encloses", () => {
+ expect(loopsForNode(STRUCTURE, "unrelated")).toEqual([]);
+ expect(loopsForNode(STRUCTURE, "outer.plain")).toEqual([]);
+ });
+});
+
+const pass = (iteration: number | undefined): LoopOption => ({ iteration, loopId: "batch" }) as LoopOption;
+
+// [whole loop, then the passes newest first]
+const OPTIONS = [pass(undefined), pass(2), pass(1), pass(0)];
+
+describe("selectActivePass", () => {
+ it("takes the pass the server resolved for the row", () => {
+ // A mapped expansion's region index is its map slot, so the url cannot be read as a pass.
+ // Slot 7 names no pass, and before this the level simply disappeared.
+ const params = new URLSearchParams("region_index=7");
+
+ expect(selectActivePass(OPTIONS, params, 1)?.iteration).toBe(1);
+ // Reading the slot off the url is what used to hide the level.
+ expect(OPTIONS.find((option) => option.iteration === 7)).toBeUndefined();
+ });
+
+ it("falls back to the newest pass when nothing pins one", () => {
+ expect(selectActivePass(OPTIONS, new URLSearchParams())?.iteration).toBe(2);
+ });
+
+ it("reads a pass pinned as a filter", () => {
+ expect(selectActivePass(OPTIONS, new URLSearchParams("iteration=0"))?.iteration).toBe(0);
+ });
+
+ it("ignores a pass belonging to a different loop", () => {
+ const params = new URLSearchParams("loop_id=other&iteration=0");
+
+ expect(selectActivePass(OPTIONS, params)?.iteration).toBe(2);
+ });
+});
+
+describe("membersOfGroup", () => {
+ it("collects a group's tasks even when they are not named after it", () => {
+ // prefix_group_id=False: "improve" is inside "refine" without carrying its name.
+ expect(membersOfGroup(STRUCTURE, "refine")).toEqual(new Set(["refine", "improve", "evaluate"]));
+ });
+
+ it("reaches a group nested inside another", () => {
+ expect(membersOfGroup(STRUCTURE, "batch")).toEqual(new Set(["batch", "batch.member"]));
+ });
+
+ it("gives nothing for a group that is not there", () => {
+ expect(membersOfGroup(STRUCTURE, "absent")).toEqual(new Set());
+ });
+});
diff --git a/airflow-core/src/airflow/ui/src/queries/useLoopFilterOptions.ts b/airflow-core/src/airflow/ui/src/queries/useLoopFilterOptions.ts
new file mode 100644
index 00000000000..0de9e4b1f19
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/queries/useLoopFilterOptions.ts
@@ -0,0 +1,180 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useQueries } from "@tanstack/react-query";
+
+import { useDagRunServiceGetDagRun } from "openapi/queries";
+import { UseGridServiceGetLoopSummaryKeyFn } from "openapi/queries/common";
+import { GridService } from "openapi/requests/services.gen";
+import type { TaskInstanceState } from "openapi/requests/types.gen";
+
+import { SearchParamsKeys } from "src/constants/searchParams";
+import { useLoopGroupIds } from "src/queries/useIsLoopGroup";
+import { isStatePending, useAutoRefresh } from "src/utils";
+
+// Loop ids are dotted task group paths, so "::" cannot occur inside one and can join the two
+// halves of an option value.
+const SEPARATOR = "::";
+
+export const encodeLoopOption = (loopId: string, iteration?: number): string =>
+ iteration === undefined ? loopId : `${loopId}${SEPARATOR}${iteration}`;
+
+export const decodeLoopOption = (value: string): { iteration?: string; loopId: string } => {
+ const index = value.lastIndexOf(SEPARATOR);
+
+ return index === -1
+ ? { loopId: value }
+ : { iteration: value.slice(index + SEPARATOR.length), loopId: value.slice(0, index) };
+};
+
+/**
+ * The pass a page is showing, given the loop's options and the url that addressed it.
+ *
+ * A table carries the pass as a filter and a task instance as a region coordinate; either names
+ * the same pass. A pass belongs to the loop that counted it, so one carried over from a different
+ * loop is ignored rather than read as a position in this one. With nothing pinned the newest
+ * pass is what the page means, since that is the one being read.
+ */
+export const selectActivePass = (
+ options: Array<LoopOption>,
+ searchParams: URLSearchParams,
+ resolvedPass?: number | null,
+): LoopOption | undefined => {
+ const [, ...passes] = options;
+
+ // The server resolves a row's pass for us, and it is the only answer that holds when the row
+ // is a mapped expansion -- there the region index is the map slot, not the pass -- or when a
+ // clear forked the later passes into a successor region.
+ if (resolvedPass !== null && resolvedPass !== undefined) {
+ return passes.find((pass) => pass.iteration === resolvedPass);
+ }
+
+ const carriedLoop = searchParams.get(SearchParamsKeys.LOOP_ID);
+
+ if (carriedLoop !== null && carriedLoop !== "" && carriedLoop !== passes[0]?.loopId) {
+ return passes[0];
+ }
+
+ const pinned = searchParams.get(SearchParamsKeys.ITERATION);
+
+ return pinned === null || pinned === ""
+ ? passes[0]
+ : passes.find((pass) => String(pass.iteration) === pinned);
+};
+
+export type LoopOption = {
+ // Set only on the whole-loop entry, which reports the run rather than one pass.
+ iteration: number | undefined;
+ iterationsRan: number | undefined;
+ loopId: string;
+ // The loop's own region. With a pass index it addresses that pass's row exactly, which is
+ // what a task instance page needs to show the pass rather than merely filter by it.
+ loopRegionId: string | undefined;
+ maxIterations: number | undefined;
+ state: TaskInstanceState | undefined;
+ value: string;
+};
+
+// A loop's outcome is its own vocabulary, but it lands on the same badge as a task instance so
+// the whole-loop entry reads like the passes beneath it. Both terminal-but-unremarkable endings
+// -- it converged, or it used up its allowance -- are a loop that did its job.
+const LOOP_STATE: Record<string, TaskInstanceState> = {
+ failed: "failed",
+ ran_to_cap: "success",
+ removed: "removed",
+ running: "running",
+ skipped: "skipped",
+ stopped_early: "success",
+};
+
+/**
+ * Every loop in the run paired with the passes it actually ran, as one flat list.
+ *
+ * Each loop reports its own iterations, so this fans out over them; a Dag normally declares one.
+ */
+export const useLoopFilterOptions = ({
+ dagId,
+ groupId,
+ runId,
+}: {
+ dagId?: string;
+ groupId?: string;
+ runId?: string;
+}) => {
+ const loopGroupIds = useLoopGroupIds(groupId);
+ const enabled = dagId !== undefined && runId !== undefined;
+ const refetchInterval = useAutoRefresh({ dagId });
+ const { data: dagRun } = useDagRunServiceGetDagRun(
+ { dagId: dagId ?? "", dagRunId: runId ?? "" },
+ undefined,
+ { enabled },
+ );
+ const isRunPending = isStatePending(dagRun?.state);
+
+ return useQueries({
+ combine: (results) => ({
+ isFetching: results.some((result) => result.isFetching),
+ options: results.flatMap((result, position): Array<LoopOption> => {
+ const loopId = loopGroupIds[position] ?? "";
+ // The whole-loop entry stands on its own, so a loop still filters before any
+ // iteration has been reported.
+ const all: LoopOption = {
+ iteration: undefined,
+ iterationsRan: result.data?.iterations_ran,
+ loopId,
+ loopRegionId: result.data?.loop_region_id ?? undefined,
+ maxIterations: result.data?.max_iterations,
+ state: result.data === undefined ? undefined : LOOP_STATE[result.data.status],
+ value: encodeLoopOption(loopId),
+ };
+
+ return [
+ all,
+ // The service reports passes oldest-first, but the newest is the one being read, so
+ // it leads the list and sits first after the whole-loop entry.
+ ...[...(result.data?.iterations ?? [])].reverse().map((iteration) => ({
+ iteration: iteration.index,
+ iterationsRan: undefined,
+ loopId,
+ loopRegionId: result.data?.loop_region_id ?? undefined,
+ maxIterations: undefined,
+ state: iteration.state ?? undefined,
+ value: encodeLoopOption(loopId, iteration.index),
+ })),
+ ];
+ }),
+ }),
+ // A pass belongs to one run, so with no run in scope there is nothing to offer. Without this
+ // the whole-loop entry is still built from an unresolved query, and callers that key off
+ // ``options.length`` -- the breadcrumb's iteration level among them -- would show it anyway.
+ queries: (enabled ? loopGroupIds : []).map((loopId) => ({
+ enabled,
+ queryFn: () => GridService.getLoopSummary({ dagId: dagId ?? "", groupId: loopId, runId: runId ?? "" }),
+ queryKey: UseGridServiceGetLoopSummaryKeyFn({
+ dagId: dagId ?? "",
+ groupId: loopId,
+ runId: runId ?? "",
+ }),
+ // A running loop gains passes and changes their states, so the switcher, breadcrumb and
+ // filter options follow the run the way the loop summary does. Polling stops as soon as
+ // the run is terminal.
+ refetchInterval: isRunPending && refetchInterval,
+ retry: false,
+ })),
+ });
+};
diff --git a/airflow-core/src/airflow/ui/src/queries/useLoopSummary.ts b/airflow-core/src/airflow/ui/src/queries/useLoopSummary.ts
index 70e37552bb0..33f1a75414d 100644
--- a/airflow-core/src/airflow/ui/src/queries/useLoopSummary.ts
+++ b/airflow-core/src/airflow/ui/src/queries/useLoopSummary.ts
@@ -18,7 +18,7 @@
*/
import { useEffect, useRef } from "react";
-import { useSearchParams } from "react-router-dom";
+import { useParams, useSearchParams } from "react-router-dom";
import { useDagRunServiceGetDagRun, useGridServiceGetLoopSummary } from "openapi/queries";
@@ -39,6 +39,7 @@ export const useLoopSummary = ({
}) => {
const refetchInterval = useAutoRefresh({ dagId });
const [searchParams] = useSearchParams();
+ const { taskId } = useParams();
const isLoopGroup = useIsLoopGroup(groupId);
const { data: dagRun } = useDagRunServiceGetDagRun({ dagId, dagRunId: runId }, undefined, {
enabled: Boolean(dagId) && Boolean(runId),
@@ -51,7 +52,11 @@ export const useLoopSummary = ({
{
dagId,
groupId,
- loopRegionId: searchParams.get(SearchParamsKeys.LOOP_REGION_ID) ?? undefined,
+ // Only a group page's region names a loop invocation. A task instance page carries its
+ // own row's region, and summarising a run by one of those is rejected -- an expansion or
+ // a pass a clear forked is not a family root -- so the server picks the invocation there.
+ loopRegionId:
+ taskId === undefined ? (searchParams.get(SearchParamsKeys.REGION_ID) ?? undefined) : undefined,
runId,
},
undefined,
diff --git a/airflow-core/src/airflow/ui/src/router.tsx b/airflow-core/src/airflow/ui/src/router.tsx
index 591e2e652c1..2b15dcd00aa 100644
--- a/airflow-core/src/airflow/ui/src/router.tsx
+++ b/airflow-core/src/airflow/ui/src/router.tsx
@@ -60,7 +60,6 @@ import { Providers } from "src/pages/Providers";
import { Run } from "src/pages/Run";
import { AssetEvents as DagRunAssetEvents } from "src/pages/Run/AssetEvents";
import { Details as DagRunDetails } from "src/pages/Run/Details";
-import { Execution } from "src/pages/Run/Execution";
import { Security } from "src/pages/Security";
import { Settings } from "src/pages/Settings";
import { Task } from "src/pages/Task";
@@ -244,7 +243,6 @@ export const routerConfig = [
{
children: [
{ element: <TaskInstances />, index: true },
- { element: <Execution />, path: "execution" },
// The Required Actions tab is now a button + modal; this keeps old /required_actions
// deep links alive by rendering the task instances, where the route sync opens the modal.
{ element: <TaskInstances />, path: "required_actions" },
diff --git a/airflow-core/src/airflow/ui/src/system-components/Modal/Modal.test.tsx b/airflow-core/src/airflow/ui/src/system-components/Modal/Modal.test.tsx
index 9ef95e7c1b8..2e18b372f5b 100644
--- a/airflow-core/src/airflow/ui/src/system-components/Modal/Modal.test.tsx
+++ b/airflow-core/src/airflow/ui/src/system-components/Modal/Modal.test.tsx
@@ -102,9 +102,34 @@ describe("Modal", () => {
expect([...(footer?.querySelectorAll("button") ?? [])].map((button) => button.textContent)).toStrictEqual(
["Save", "modal.cancel"],
);
- expect(globalThis.getComputedStyle(footer?.firstElementChild as HTMLElement).flexDirection).toBe(
- "row-reverse",
+ // ``row-reverse`` sits on the group holding the actions, not the footer's own first child:
+ // the footer leads with the options group, which is left-aligned and empty here.
+ const actionsGroup = screen.getByRole("button", { name: "Save" }).parentElement;
+
+ expect(globalThis.getComputedStyle(actionsGroup as HTMLElement).flexDirection).toBe("row-reverse");
+ });
+
+ it("puts footerOptions at the left of the footer, ahead of the actions", () => {
+ render(
+ <Modal
+ footerActions={<Button>Save</Button>}
+ footerOptions={<span>Keep state</span>}
+ open
+ title="Modal Heading"
+ >
+ Lorem Ipsum Content
+ </Modal>,
+ { wrapper: BaseWrapper },
);
+
+ const option = screen.getByText("Keep state");
+
+ // Compared against the Save button rather than the footer's first child, which wraps both
+ // groups and would contain the option whichever order they were in.
+ const save = screen.getByRole("button", { name: "Save" });
+
+ // eslint-disable-next-line no-bitwise
+ expect(option.compareDocumentPosition(save) & Node.DOCUMENT_POSITION_FOLLOWING).toBeTruthy();
});
it("lets headerProps and bodyProps children override the defaults", () => {
diff --git a/airflow-core/src/airflow/ui/src/system-components/Modal/Modal.tsx b/airflow-core/src/airflow/ui/src/system-components/Modal/Modal.tsx
index f7b36abecd7..41c68b94ebe 100644
--- a/airflow-core/src/airflow/ui/src/system-components/Modal/Modal.tsx
+++ b/airflow-core/src/airflow/ui/src/system-components/Modal/Modal.tsx
@@ -29,6 +29,9 @@ export type ModalProps = {
readonly children?: ReactNode;
readonly contentProps?: ComponentProps<typeof Dialog.Content>;
readonly footerActions?: ReactNode;
+ // Toggles that change what the action does, rather than actions themselves. They sit at the
+ // far left of the footer so the buttons stay together at the right, where they are looked for.
+ readonly footerOptions?: ReactNode;
readonly footerProps?: ChakraDialog.FooterProps;
readonly headerProps?: ChakraDialog.HeaderProps;
readonly hideCancelAction?: boolean;
@@ -53,6 +56,7 @@ export const Modal = ({
children,
contentProps,
footerActions,
+ footerOptions,
footerProps,
headerProps,
hideCancelAction = false,
@@ -67,18 +71,23 @@ export const Modal = ({
const titleContent = title === undefined ? undefined : <Dialog.Title>{title}</Dialog.Title>;
const headerContent = headerProps?.children ?? titleContent;
const actionsContent =
- footerActions === undefined ? undefined : (
- // `row-reverse` renders the first action rightmost, so the primary action is
- // written first and therefore reached first when tabbing.
- <Flex flexDirection="row-reverse" gap={2} w="100%">
- {footerActions}
- {hideCancelAction ? undefined : (
- <Dialog.ActionTrigger asChild>
- <Button variant="outline" {...cancelActionProps}>
- {cancelActionProps?.children ?? translate("modal.cancel")}
- </Button>
- </Dialog.ActionTrigger>
- )}
+ footerActions === undefined && footerOptions === undefined ? undefined : (
+ <Flex alignItems="center" gap={4} justifyContent="space-between" w="100%">
+ <Flex alignItems="center" flexWrap="wrap" gap={4}>
+ {footerOptions}
+ </Flex>
+ {/* `row-reverse` renders the first action rightmost, so the primary action is
+ written first and therefore reached first when tabbing. */}
+ <Flex flexDirection="row-reverse" flexShrink={0} gap={2}>
+ {footerActions}
+ {hideCancelAction ? undefined : (
+ <Dialog.ActionTrigger asChild>
+ <Button variant="outline" {...cancelActionProps}>
+ {cancelActionProps?.children ?? translate("modal.cancel")}
+ </Button>
+ </Dialog.ActionTrigger>
+ )}
+ </Flex>
</Flex>
);
const footerContent = footerProps?.children ?? actionsContent;
diff --git a/airflow-core/src/airflow/ui/src/utils/executionParams.test.ts b/airflow-core/src/airflow/ui/src/utils/executionParams.test.ts
index 222022d0962..164d1a412d7 100644
--- a/airflow-core/src/airflow/ui/src/utils/executionParams.test.ts
+++ b/airflow-core/src/airflow/ui/src/utils/executionParams.test.ts
@@ -18,17 +18,42 @@
*/
import { describe, expect, it } from "vitest";
-import { stripExecutionParams } from "./executionParams";
+import { setExecutionCoordinates, stripExecutionParams } from "./executionParams";
describe("stripExecutionParams", () => {
- it("drops the params that pin one execution, try or loop iteration and keeps the rest", () => {
- const source = new URLSearchParams(
- "try_number=2®ion_id=r®ion_index=1&iteration=3&loop_region_id=l&dag_run_state=failed",
- );
+ it("drops the params that pin one execution or try and keeps the rest", () => {
+ const source = new URLSearchParams("try_number=2®ion_id=r®ion_index=1&dag_run_state=failed");
const stripped = stripExecutionParams(source);
expect(stripped.toString()).toBe("dag_run_state=failed");
expect(source.has("try_number")).toBe(true);
});
+
+ it("keeps the loop scope, which follows you between a loop's tasks", () => {
+ const stripped = stripExecutionParams(new URLSearchParams("region_id=r&loop_id=batch&iteration=3"));
+
+ expect(stripped.toString()).toBe("loop_id=batch&iteration=3");
+ });
+});
+
+describe("setExecutionCoordinates", () => {
+ it("drops a pass the new coordinate does not name", () => {
+ const params = setExecutionCoordinates(new URLSearchParams("loop_id=batch&iteration=0"), "r", 4);
+
+ // The row shown is now the newest pass, so keeping "0" would label it wrongly.
+ expect(params.get("region_index")).toBe("4");
+ expect(params.has("iteration")).toBe(false);
+ expect(params.has("loop_id")).toBe(false);
+ });
+
+ it("leaves the pass alone when there is no coordinate to move to", () => {
+ const params = setExecutionCoordinates(
+ new URLSearchParams("loop_id=batch&iteration=0"),
+ undefined,
+ undefined,
+ );
+
+ expect(params.get("iteration")).toBe("0");
+ });
});
diff --git a/airflow-core/src/airflow/ui/src/utils/executionParams.ts b/airflow-core/src/airflow/ui/src/utils/executionParams.ts
index ee254ea95f3..aa1f160d94c 100644
--- a/airflow-core/src/airflow/ui/src/utils/executionParams.ts
+++ b/airflow-core/src/airflow/ui/src/utils/executionParams.ts
@@ -22,11 +22,14 @@ const EXECUTION_PARAMS = [
SearchParamsKeys.TRY_NUMBER,
SearchParamsKeys.REGION_ID,
SearchParamsKeys.REGION_INDEX,
- SearchParamsKeys.ITERATION,
- SearchParamsKeys.LOOP_REGION_ID,
];
-/** A copy of `searchParams` without the params that pin one execution, try or loop iteration. */
+/**
+ * A copy of `searchParams` without the params that pin one execution or try.
+ *
+ * The loop filter is deliberately not among them: a pass is where you are in a loop, not a row,
+ * so it survives moving between the loop's tasks. Pages drop a loop that is out of their scope.
+ */
export const stripExecutionParams = (searchParams: string | URLSearchParams): URLSearchParams => {
const stripped = new URLSearchParams(searchParams);
@@ -36,3 +39,26 @@ export const stripExecutionParams = (searchParams: string | URLSearchParams): UR
return stripped;
};
+
+/**
+ * Point a link at one execution. A task inside a loop has a row per iteration, so a link with
+ * no coordinate is ambiguous and the API rejects it; the newest iteration is the sensible
+ * landing place from a view that shows the task once.
+ */
+export const setExecutionCoordinates = (
+ searchParams: URLSearchParams,
+ regionId: string | null | undefined,
+ regionIndex: number | null | undefined,
+): URLSearchParams => {
+ if (regionId !== null && regionId !== undefined && regionIndex !== null && regionIndex !== undefined) {
+ searchParams.set(SearchParamsKeys.REGION_ID, regionId);
+ searchParams.set(SearchParamsKeys.REGION_INDEX, String(regionIndex));
+ // The coordinate decides which row is shown, and it is the newest pass here rather than
+ // whichever one was being read. Carrying the old pass label over would leave the breadcrumb
+ // and the Gantt naming a pass the page is not on, and Clear acting on a row nothing names.
+ searchParams.delete(SearchParamsKeys.ITERATION);
+ searchParams.delete(SearchParamsKeys.LOOP_ID);
+ }
+
+ return searchParams;
+};
diff --git a/airflow-core/src/airflow/ui/src/utils/useFiltersHandler.ts b/airflow-core/src/airflow/ui/src/utils/useFiltersHandler.ts
index 79d3651e272..18d43432178 100644
--- a/airflow-core/src/airflow/ui/src/utils/useFiltersHandler.ts
+++ b/airflow-core/src/airflow/ui/src/utils/useFiltersHandler.ts
@@ -85,6 +85,7 @@ export type FilterableSearchParamsKeys =
| SearchParamsKeys.KEY_PATTERN
| SearchParamsKeys.LAST_ASSET_EVENT_TIMESTAMP_RANGE
| SearchParamsKeys.LOGICAL_DATE_RANGE
+ | SearchParamsKeys.LOOP
| SearchParamsKeys.MAP_INDEX
| SearchParamsKeys.MAX_ACTIVE_RUNS_GTE
| SearchParamsKeys.MAX_ACTIVE_RUNS_LTE
diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
index 952ed417fd9..1a917017bff 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
@@ -1503,6 +1503,8 @@ class TestGetTaskInstances(TestTaskInstanceEndpoint):
)
session.commit()
url = f"/dags/{dr.dag_id}/dagRuns/{dr.run_id}/taskInstances"
+ # The UI filters by the loop the author named, not by a region. Keyed that way it spans
+ # every invocation, so a clear that forked the later passes does not lose them.
params = {"loop_id": "body", "limit": 1}
if cursor is not None:
params["cursor"] = cursor
@@ -1510,17 +1512,32 @@ class TestGetTaskInstances(TestTaskInstanceEndpoint):
assert response.status_code == 200, response.text
assert response.json()["total_entries"] == 4
assert len(response.json()["task_instances"]) == 1
+ # One pass covers the loop's own region at that index *and* the mapped expansion nested
+ # inside it, which sits at an index of its own.
response = test_client.get(url, params={**params, "iteration": 2, "limit": 100})
assert response.status_code == 200, response.text
assert response.json()["total_entries"] == 2
assert {ti["region_index"] for ti in response.json()["task_instances"]} == {2, 7}
response = test_client.get(url, params={**params, "iteration": 0, "limit": 100})
assert response.json()["total_entries"] == 2
+ # region_id names one region exactly; the logical scope of a pass is what loop_id and
+ # iteration are for, so this does not reach the expansion nested inside it.
+ exact = test_client.get(url, params={"region_id": str(replacement.id), "limit": 100})
+ assert exact.status_code == 200, exact.text
+ assert {ti["region_index"] for ti in exact.json()["task_instances"]} == {2}
def test_iteration_filter_requires_loop(self, test_client):
response = test_client.get("/dags/~/dagRuns/~/taskInstances", params={"iteration": 1})
assert response.status_code == 400
+ def test_loop_filter_requires_a_specific_dag_run(self, test_client):
+ response = test_client.get("/dags/~/dagRuns/~/taskInstances", params={"loop_id": "body"})
+ assert response.status_code == 400
+
+ def test_region_index_requires_region_id(self, test_client):
+ response = test_client.get("/dags/~/dagRuns/~/taskInstances", params={"region_index": 1})
+ assert response.status_code == 400
+
@pytest.mark.parametrize(
("task_instances", "update_extras", "url", "params", "expected_ti", "expected_queries_number"),
[
diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_gantt.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_gantt.py
index a2d65ec133d..b03710137fa 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_gantt.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_gantt.py
@@ -344,6 +344,35 @@ class TestGetGanttDataEndpoint:
assert tis[TASK_ID_3]["scheduled_dttm"] is None
assert tis[TASK_ID_3]["queued_dttm"] is None
+ def test_archived_attempts_stay_on_the_chart(self, test_client, dag_maker, session):
+ with dag_maker("gantt_cleared", serialized=True):
+ EmptyOperator(task_id="work")
+ dr = dag_maker.create_dagrun()
+ first = next(ti for ti in dr.task_instances if ti.task_id == "work")
+ first.try_number = 1
+ first.state = TaskInstanceState.SUCCESS
+ session.flush()
+ # Clearing keeps each attempt as its own row and leaves only the newest live, so the
+ # earlier try is only on the chart if archived rows are read.
+ first.archive(reason="cleared", session=session)
+ retry = TaskInstance(
+ task=dag_maker.serialized_dag.get_task("work"),
+ run_id=dr.run_id,
+ dag_version_id=first.dag_version_id,
+ region_id=first.region_id,
+ region_index=first.region_index,
+ )
+ retry.try_number = 2
+ retry.state = TaskInstanceState.SUCCESS
+ session.add(retry)
+ session.commit()
+
+ response = test_client.get(f"/gantt/{dr.dag_id}/{dr.run_id}")
+
+ assert response.status_code == 200, response.text
+ rows = [row for row in response.json()["task_instances"] if row["task_id"] == "work"]
+ assert {row["try_number"] for row in rows} == {1, 2}
+
def test_loop_passes_keep_exact_identity(self, test_client, dag_maker, session):
@task_group
def body():
diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py
index 450b586134b..80158ac5839 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py
@@ -886,6 +886,8 @@ class TestGetGridDataEndpoint:
}
for ti in expected["task_instances"]:
ti["loop_iterations_count"] = None
+ ti["latest_region_id"] = None
+ ti["latest_region_index"] = None
for obj in actual, expected:
tis = obj["task_instances"]
tis[:] = sorted(tis, key=lambda x: x["task_id"])
@@ -993,6 +995,8 @@ class TestGetGridDataEndpoint:
]
for ti in expected:
ti["loop_iterations_count"] = None
+ ti["latest_region_id"] = None
+ ti["latest_region_index"] = None
expected = sort_dict(expected)
actual = sort_dict(actual)
assert actual == expected
@@ -1527,6 +1531,34 @@ class TestGetLoopSummaryEndpoint:
assert next(ti for ti in tasks if ti["task_id"] == "body")["loop_iterations_count"] == 2
assert summary["iterations_ran"] == 2
+ def test_grid_counts_every_pass_but_links_to_the_newest(self, test_client, executed_fixed_loop, session):
+ run = executed_fixed_loop
+ # Clearing one pass without its later iterations puts an earlier pass back to work while
+ # a later one stays finished.
+ for ti in session.scalars(
+ select(TaskInstance).where(
+ TaskInstance.dag_id == run.dag_id,
+ TaskInstance.run_id == run.run_id,
+ TaskInstance.task_id == "body.work",
+ TaskInstance.region_index == 0,
+ )
+ ):
+ ti.state = TaskInstanceState.RUNNING
+ session.commit()
+
+ response = test_client.get(f"/grid/ti_summaries/{run.dag_id}", params={"run_ids": [run.run_id]})
+ by_task_id = {ti["task_id"]: ti for ti in json.loads(response.text)["task_instances"]}
+
+ # States are aggregated onto the group. The cell stands for the loop's work, so the
+ # re-running pass shows rather than being hidden behind the finished later ones.
+ assert by_task_id["body"]["child_states"]["running"] == 1
+ # The later passes still count; a floor rather than a total, so the loop's own gate
+ # tasks do not pin this.
+ assert by_task_id["body"]["child_states"]["success"] >= 2
+ # The coordinate is recorded per row, and a link still has to address one, so it stays
+ # on the newest pass even though an earlier one is the thing now running.
+ assert by_task_id["body.work"]["latest_region_index"] == 2
+
def test_loop_history_reads_runs_in_a_fixed_number_of_queries(self, test_client, dag_maker, session):
@task
def work():