Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@ class TaskInstanceResponse(OmitsMissingRegion, BaseModel):
map_index: int
region_id: RegionId = None
region_index: RegionIndex = None
in_loop: bool
logical_date: datetime | None
run_after: datetime
start_date: datetime | None
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5250,6 +5250,9 @@ components:
region_index:
type: integer
title: Region Index
in_loop:
type: boolean
title: In Loop
logical_date:
anyOf:
- type: string
Expand Down Expand Up @@ -5401,6 +5404,7 @@ components:
- dag_id
- dag_run_id
- map_index
- in_loop
- logical_date
- run_after
- start_date
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17920,6 +17920,9 @@ components:
region_index:
type: integer
title: Region Index
in_loop:
type: boolean
title: In Loop
logical_date:
anyOf:
- type: string
Expand Down Expand Up @@ -18071,6 +18074,7 @@ components:
- dag_id
- dag_run_id
- map_index
- in_loop
- logical_date
- run_after
- start_date
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,8 @@ def __getattr__(self, name: str) -> Any:
if self.projected_map_index is not None:
return self.projected_map_index
return self.resolver.public_map_index(self.value)
if name == "in_loop":
return self.value.region_id != SENTINEL_REGION_ID and self.resolver._is_in_loop(self.value)
return getattr(self.value, name)


Expand Down
12 changes: 11 additions & 1 deletion airflow-core/src/airflow/models/task_coordinates.py
Original file line number Diff line number Diff line change
Expand Up @@ -164,6 +164,7 @@ class TaskCoordinateResolver:
_dags: dict[UUID, SerializedDAG] = attrs.field(factory=dict, init=False)
_region_nodes: dict[UUID, str | None] = attrs.field(factory=dict, init=False)
_regional_tasks: dict[tuple[str, str | None, str], bool] = attrs.field(factory=dict, init=False)
_looped_tasks: dict[tuple[UUID, str], bool] = attrs.field(factory=dict, init=False)

@classmethod
def for_dag(cls, dag: SerializedDAG | None, session: Session) -> TaskCoordinateResolver:
Expand Down Expand Up @@ -201,7 +202,16 @@ def get_task(
raise ValueError(f"Pinned Dag for {dag_id}/{run_id} not found")
return dag.get_task(task_id)

def find_task(self, ti: TaskInstance) -> SerializedOperator | None:
def _is_in_loop(self, ti: TaskCoordinate) -> bool:
if ti.dag_version_id is None:
return False
key = (ti.dag_version_id, ti.task_id)
if key not in self._looped_tasks:
task = self.find_task(ti)
self._looped_tasks[key] = task is not None and enclosing_loop(task) is not None
return self._looped_tasks[key]

def find_task(self, ti: TaskCoordinate) -> SerializedOperator | None:
"""Return the task of ``ti`` from its pinned Dag, or None when that Dag or the task is gone."""
if ti.dag_version_id is None or (dag := self.get_dag(ti.dag_version_id)) is None:
return None
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -8067,6 +8067,10 @@ export const $TaskInstanceResponse = {
type: 'integer',
title: 'Region Index'
},
in_loop: {
type: 'boolean',
title: 'In Loop'
},
logical_date: {
anyOf: [
{
Expand Down Expand Up @@ -8351,7 +8355,7 @@ export const $TaskInstanceResponse = {
}
},
type: 'object',
required: ['id', 'task_id', 'dag_id', 'dag_run_id', 'map_index', 'logical_date', 'run_after', 'start_date', 'end_date', 'duration', 'state', 'try_number', 'max_tries', 'task_display_name', 'dag_display_name', 'hostname', 'unixname', 'pool', 'pool_slots', 'queue', 'priority_weight', 'operator', 'operator_name', 'queued_when', 'scheduled_when', 'pid', 'executor', 'executor_config', 'note', 'rendered_map_index', 'trigger', 'triggerer_job', 'dag_version'],
required: ['id', 'task_id', 'dag_id', 'dag_run_id', 'map_index', 'in_loop', 'logical_date', 'run_after', 'start_date', 'end_date', 'duration', 'state', 'try_number', 'max_tries', 'task_display_name', 'dag_display_name', 'hostname', 'unixname', 'pool', 'pool_slots', 'queue', 'priority_weight', 'operator', 'operator_name', 'queued_when', 'scheduled_when', 'pid', 'executor', 'executor_config', 'note', 'rendered_map_index', 'trigger', 'triggerer_job', 'dag_version'],
title: 'TaskInstanceResponse',
description: 'TaskInstance serializer for responses.'
} as const;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2199,6 +2199,7 @@ export type TaskInstanceResponse = {
map_index: number;
region_id?: string;
region_index?: number;
in_loop: boolean;
logical_date: string | null;
run_after: string;
start_date: string | null;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -466,6 +466,7 @@
"executorConfig": "Executor Config",
"hostname": "Hostname",
"id": "ID",
"iteration": "Iteration",
"maxTries": "Max Tries",
"pid": "PID",
"pool": "Pool",
Expand Down
11 changes: 11 additions & 0 deletions airflow-core/src/airflow/ui/public/i18n/locales/en/dag.json
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,17 @@
"upcomingCount_one": "{{count}} Upcoming Deadline",
"upcomingCount_other": "{{count}} 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": {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -92,13 +92,15 @@
"existingTasks": "Clear existing tasks",
"future": "Future",
"keepTaskState": "Keep task state and resume",
"laterIterations": "Later iterations",
"onlyFailed": "Clear only failed tasks",
"past": "Past",
"preventRunningTasks": "Prevent rerun if task is running",
"queueNew": "Queue up new tasks",
"runOnLatestVersion": "Run with latest bundle version",
"runOnLatestVersionForced": "Always uses the latest — there's no earlier version to go back to",
"upstream": "Upstream"
"upstream": "Upstream",
"wholeExpansion": "Whole expansion"
}
},
"runStateCounts": {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -27,10 +27,8 @@ import type { DataTableFeatures } from "src/components/DataTable/features";
import type { MetaColumn } from "src/components/DataTable/types";
import { StateBadge } from "src/components/StateBadge";

// Stable per-row key; dag_run_id keeps the same task distinct across runs (past/future
// expansion), and map_index disambiguates the mapped instances of one task.
export const taskInstanceKey = (ti: TaskInstanceResponse): string =>
`${ti.dag_run_id}:${ti.task_id}:${ti.map_index}`;
`${ti.dag_run_id}:${ti.task_id}:${ti.map_index}:${ti.region_id}:${ti.region_index}`;

// When provided, rows render a leading checkbox so the user can exclude individual
// task instances from the action. Excluded keys are tracked by the caller.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,6 @@ export const ClearGroupTaskInstanceDialog = ({ onClose, open, taskInstance }: Pr
}, [open, keepTaskStateDefault]);

const { isPending, mutate } = useClearTaskInstances({
dagId,
dagRunId: runId,
onSuccessConfirm: onCloseDialog,
});
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -30,17 +30,9 @@ import { useClearTaskInstancesDryRun } from "src/queries/useClearTaskInstancesDr
import { useDurationFormat } from "src/utils";

type Props = {
readonly dagDetails?: {
readonly dryRun?: {
dagId: string;
dagRunId: string;
downstream?: boolean;
future?: boolean;
mapIndex?: number;
onlyFailed?: boolean;
past?: boolean;
taskId: string;
taskIds?: ClearTaskInstancesBody["task_ids"];
upstream?: boolean;
requestBody: ClearTaskInstancesBody;
};
readonly onClose: () => void;
readonly onConfirm?: () => void;
Expand All @@ -49,37 +41,24 @@ type Props = {
};

const ClearTaskInstanceConfirmationDialog = ({
dagDetails,
dryRun,
onClose,
onConfirm,
open,
preventRunningTask,
}: Props) => {
const { t: translate } = useTranslation();
const { formatRelative } = useDurationFormat();
const useExplicitTaskIds = dagDetails?.taskIds !== undefined;
const { data, isFetching } = useClearTaskInstancesDryRun({
dagId: dagDetails?.dagId ?? "",
dagId: dryRun?.dagId ?? "",
options: {
enabled: open && Boolean(dagDetails),
enabled: open && Boolean(dryRun),
gcTime: 0,
refetchOnMount: "always",
refetchOnWindowFocus: false,
staleTime: 0,
},
requestBody: {
dag_run_id: dagDetails?.dagRunId ?? "",
include_downstream: !useExplicitTaskIds && dagDetails?.downstream,
include_future: dagDetails?.future,
include_past: dagDetails?.past,
include_upstream: !useExplicitTaskIds && dagDetails?.upstream,
only_failed: dagDetails?.onlyFailed,
task_ids: useExplicitTaskIds
? dagDetails.taskIds
: dagDetails?.mapIndex === undefined
? [dagDetails?.taskId ?? ""]
: [[dagDetails.taskId, dagDetails.mapIndex]],
},
requestBody: dryRun?.requestBody ?? {},
});

const [isReady, setIsReady] = useState(false);
Expand Down
Loading
Loading