[v3-3-test] Allow filtering the Dags list by failed and success runs in any run state filter

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)
 
 

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

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`

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`

airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts

@@ -680,7 +680,7 @@ export const useDagServiceGetDagTags = <TData = Common.DagServiceGetDagTagsDefau
 * @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`

airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts

@@ -680,7 +680,7 @@ export const useDagServiceGetDagTagsSuspense = <TData = Common.DagServiceGetDagT
 * @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`

airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts

@@ -1955,7 +1955,7 @@ export class DagService {
      * @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`

airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts

@@ -3438,7 +3438,7 @@ export type GetDagsUiData = {
     dagIds?: Array<(string)> | 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;

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<BooleanFilterValue> = ["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}
       />
       <RunStateSelect
         dataTestId="dags-any-run-state-filter"
         label={translate("filters.anyRunState")}
         onChange={handleActiveRunChange}
-        states={anyRunStates}
+        states={runStates}
         value={activeRunState ?? undefined}
       />
       <RequiredActionFilter needsReview={needsReview === "true"} onToggle={handleNeedsReviewToggle} />

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."""