From 7c160485c5719e769cc7d4360cf34a5ba4117058 Mon Sep 17 00:00:00 2001 From: Yuseok Jo Date: Thu, 23 Jul 2026 17:51:56 +0900 Subject: [PATCH] Allow filtering the Dags list by failed and success runs in any run state filter (#69875) (cherry picked from commit a990ce976e7ddba115e3d864d18a84ff58b7e7ed) --- .../airflow/api_fastapi/common/parameters.py | 19 ++++++-------- .../core_api/openapi/_private_ui.yaml | 6 ++--- .../ui/openapi-gen/queries/ensureQueryData.ts | 2 +- .../ui/openapi-gen/queries/prefetch.ts | 2 +- .../airflow/ui/openapi-gen/queries/queries.ts | 2 +- .../ui/openapi-gen/queries/suspense.ts | 2 +- .../ui/openapi-gen/requests/services.gen.ts | 2 +- .../ui/openapi-gen/requests/types.gen.ts | 2 +- .../DagsList/DagsFilters/DagsFilters.tsx | 8 +++--- .../core_api/routes/ui/test_dags.py | 25 ++++++++++--------- 10 files changed, 32 insertions(+), 38 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/common/parameters.py b/airflow-core/src/airflow/api_fastapi/common/parameters.py index a42cbf9df3a10..e033b777772b3 100644 --- a/airflow-core/src/airflow/api_fastapi/common/parameters.py +++ b/airflow-core/src/airflow/api_fastapi/common/parameters.py @@ -1287,29 +1287,26 @@ def depends(cls, has_pending_actions: bool | None = Query(None)) -> _PendingActi class _AnyDagRunStateFilter(BaseParam[DagRunState | None]): """Filter Dags that have any DagRun in the given state, not only the latest one.""" - # Only these states have a partial index on dag_run; others would force a full table scan. - SUPPORTED_STATES = (DagRunState.QUEUED, DagRunState.RUNNING) - def to_orm(self, select: Select) -> Select: if self.value is None and self.skip_none: return select - run_subquery = sql_select(DagRun.dag_id).where(DagRun.state == self.value).distinct() - return select.where(DagModel.dag_id.in_(run_subquery)) + # EXISTS resolves each Dag via the (dag_id, state) index instead of scanning every run in the state. + has_run_in_state = ( + sql_select(DagRun.dag_id) + .where(DagRun.dag_id == DagModel.dag_id, DagRun.state == self.value) + .exists() + ) + return select.where(has_run_in_state) @classmethod def depends( cls, dag_run_state: DagRunState | None = Query( None, - description="Filter Dags that have any DagRun in the given state. Only ``queued`` and ``running`` are supported.", + description="Filter Dags that have any DagRun in the given state.", ), ) -> _AnyDagRunStateFilter: - if dag_run_state is not None and dag_run_state not in cls.SUPPORTED_STATES: - raise HTTPException( - status.HTTP_400_BAD_REQUEST, - detail=f"dag_run_state only supports {[state.value for state in cls.SUPPORTED_STATES]}.", - ) return cls().set_value(dag_run_state) 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 e7eb12acfaa37..90e221c07035e 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 @@ -464,11 +464,9 @@ paths: anyOf: - $ref: '#/components/schemas/DagRunState' - type: 'null' - description: Filter Dags that have any DagRun in the given state. Only ``queued`` - and ``running`` are supported. + description: Filter Dags that have any DagRun in the given state. title: Dag Run State - description: Filter Dags that have any DagRun in the given state. Only ``queued`` - and ``running`` are supported. + description: Filter Dags that have any DagRun in the given state. - name: bundle_name in: query required: false 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 c4a9f18e8d8f3..7fc2b816ad0c8 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts @@ -680,7 +680,7 @@ export const ensureUseDagServiceGetDagTagsData = (queryClient: QueryClient, { li * @param data.paused * @param data.hasImportErrors Filter Dags by having import errors. Only Dags that have been successfully loaded before will be returned. * @param data.lastDagRunState -* @param data.dagRunState Filter Dags that have any DagRun in the given state. Only ``queued`` and ``running`` are supported. +* @param data.dagRunState Filter Dags that have any DagRun in the given state. * @param data.bundleName * @param data.bundleVersion * @param data.orderBy Attributes to order by, multi criteria sort is supported. Prefix with `-` for descending order. Supported attributes: `dag_id, dag_display_name, next_dagrun, state, start_date, last_run_state, last_run_start_date` 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 ce348567d74a7..1d55ca5203e4c 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts @@ -680,7 +680,7 @@ export const prefetchUseDagServiceGetDagTags = (queryClient: QueryClient, { limi * @param data.paused * @param data.hasImportErrors Filter Dags by having import errors. Only Dags that have been successfully loaded before will be returned. * @param data.lastDagRunState -* @param data.dagRunState Filter Dags that have any DagRun in the given state. Only ``queued`` and ``running`` are supported. +* @param data.dagRunState Filter Dags that have any DagRun in the given state. * @param data.bundleName * @param data.bundleVersion * @param data.orderBy Attributes to order by, multi criteria sort is supported. Prefix with `-` for descending order. Supported attributes: `dag_id, dag_display_name, next_dagrun, state, start_date, last_run_state, last_run_start_date` 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 8bc74b305aa17..a5fc9003bcd76 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts @@ -680,7 +680,7 @@ export const useDagServiceGetDagTags = | null; dagRunsLimit?: number; /** - * Filter Dags that have any DagRun in the given state. Only ``queued`` and ``running`` are supported. + * Filter Dags that have any DagRun in the given state. */ dagRunState?: DagRunState | null; excludeStale?: boolean; diff --git a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.tsx b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.tsx index 9f1863bc12048..5a0178177199b 100644 --- a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.tsx +++ b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.tsx @@ -45,9 +45,7 @@ const { type BooleanFilterValue = "all" | "false" | "true"; -const lastRunStates = ["failed", "queued", "running", "success"] as const; -// Mirrors the backend limit (see _AnyDagRunStateFilter): only running/queued are indexed on dag_run. -const anyRunStates = ["queued", "running"] as const; +const runStates = ["failed", "queued", "running", "success"] as const; const booleanFilterValues: ReadonlyArray = ["all", "true", "false"]; const toBooleanFilterValue = ( @@ -161,14 +159,14 @@ export const DagsFilters = () => { dataTestId="dags-last-run-state-filter" label={translate("filters.lastRunState")} onChange={handleStateChange} - states={lastRunStates} + states={runStates} value={state ?? undefined} /> diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dags.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dags.py index 25e6ac12ff38f..04fa6876d47f1 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dags.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dags.py @@ -135,30 +135,31 @@ def test_should_return_200(self, test_client, query_params, expected_ids, expect previous_run_after = dag_run["run_after"] @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle") - def test_dag_run_state_matches_any_run_not_only_latest(self, test_client, session): - # Backwards backfill: an older run is still running while the latest run already finished. + @pytest.mark.parametrize("state", ["queued", "running", "failed", "success"]) + def test_dag_run_state_matches_any_run_not_only_latest(self, test_client, session, state): + # Give DAG1 an older run in the probed state while its latest run ends in a + # different state, so a latest-run filter misses it but an any-run filter finds it + # (e.g. failure history hidden behind a green latest run). older_run = session.scalar( select(DagRun).where(DagRun.dag_id == DAG1_ID, DagRun.run_id == "run_id_1") ) - older_run.state = DagRunState.RUNNING + older_run.state = DagRunState(state) + latest_run = session.scalar( + select(DagRun).where(DagRun.dag_id == DAG1_ID, DagRun.run_id == "run_id_5") + ) + latest_run.state = DagRunState.FAILED if state == "success" else DagRunState.SUCCESS session.commit() - # last_dag_run_state only looks at the latest run, which is not running - last_state = test_client.get("/dags", params={"last_dag_run_state": "running"}) + # last_dag_run_state only looks at the latest run, which is in a different state + last_state = test_client.get("/dags", params={"last_dag_run_state": state, "dag_ids": [DAG1_ID]}) assert last_state.status_code == 200 assert [dag["dag_id"] for dag in last_state.json()["dags"]] == [] # dag_run_state matches a Dag that has any run in the state - any_state = test_client.get("/dags", params={"dag_run_state": "running"}) + any_state = test_client.get("/dags", params={"dag_run_state": state, "dag_ids": [DAG1_ID]}) assert any_state.status_code == 200 assert [dag["dag_id"] for dag in any_state.json()["dags"]] == [DAG1_ID] - @pytest.mark.parametrize("unsupported_state", ["success", "failed"]) - def test_dag_run_state_rejects_unsupported_states(self, test_client, unsupported_state): - # Only running/queued have a partial index; other states would force a full table scan. - response = test_client.get("/dags", params={"dag_run_state": unsupported_state}) - assert response.status_code == 400 - @pytest.fixture def setup_hitl_data(self, create_task_instance: TaskInstance, session: Session): """Setup HITL test data for parametrized tests."""