[v3-3-test] Fix 500 when combining last-run and any-run Dag state filters (#71366) (#71371)

(cherry picked from commit 7cdb9ad)

Co-authored-by: Rahul Vats <43964496+vatsrahul1001@users.noreply.github.com>

Author: github-actions[bot]

airflow-core/src/airflow/api_fastapi/common/parameters.py

@@ -38,6 +38,7 @@
 from sqlalchemy import Column, String, and_, func, not_, or_, select as sql_select, true as sql_true
 from sqlalchemy.ext.compiler import compiles
 from sqlalchemy.inspection import inspect
+from sqlalchemy.orm import aliased
 from sqlalchemy.sql.functions import FunctionElement
 
 from airflow._shared.timezones import timezone
@@ -1279,10 +1280,13 @@ def to_orm(self, select: Select) -> Select:
         if self.value is None and self.skip_none:
             return select
 
-        # EXISTS resolves each Dag via the (dag_id, state) index instead of scanning every run in the state.
+        # Alias DagRun so this EXISTS subquery cannot auto-correlate to a DagRun the outer query
+        # may already reference (e.g. the last_dag_run_state filter), which would strip the
+        # subquery's FROM and raise. EXISTS resolves each Dag via the (dag_id, state) index.
+        any_run = aliased(DagRun)
         has_run_in_state = (
-            sql_select(DagRun.dag_id)
-            .where(DagRun.dag_id == DagModel.dag_id, DagRun.state == self.value)
+            sql_select(any_run.dag_id)
+            .where(any_run.dag_id == DagModel.dag_id, any_run.state == self.value)
             .exists()
         )
         return select.where(has_run_in_state)

airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dags.py

@@ -160,6 +160,24 @@ def test_dag_run_state_matches_any_run_not_only_latest(self, test_client, sessio
         assert any_state.status_code == 200
         assert [dag["dag_id"] for dag in any_state.json()["dags"]] == [DAG1_ID]
 
+    @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+    def test_last_and_any_run_state_filters_combined(self, test_client, session):
+        # Regression: combining the last-run and any-run state filters must return the
+        # intersection, not raise. The any-run EXISTS subquery must not correlate to the
+        # DagRun the last-run filter joins into the outer query.
+        latest_run = session.scalar(
+            select(DagRun).where(DagRun.dag_id == DAG1_ID, DagRun.run_id == "run_id_5")
+        )
+        latest_run.state = DagRunState.FAILED
+        session.commit()
+
+        response = test_client.get(
+            "/dags",
+            params={"last_dag_run_state": "failed", "dag_run_state": "failed", "dag_ids": [DAG1_ID]},
+        )
+        assert response.status_code == 200
+        assert [dag["dag_id"] for dag in response.json()["dags"]] == [DAG1_ID]
+
     @pytest.fixture
     def setup_hitl_data(self, create_task_instance: TaskInstance, session: Session):
         """Setup HITL test data for parametrized tests."""