This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 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 7cdb9ad47af Fix 500 when combining last-run and any-run Dag state
filters (#71366)
7cdb9ad47af is described below
commit 7cdb9ad47aff1168a6de06363066184dd029d8b9
Author: Rahul Vats <[email protected]>
AuthorDate: Mon Aug 10 16:39:57 2026 +0530
Fix 500 when combining last-run and any-run Dag state filters (#71366)
---
.../src/airflow/api_fastapi/common/parameters.py | 10 +++++++---
.../unit/api_fastapi/core_api/routes/ui/test_dags.py | 18 ++++++++++++++++++
2 files changed, 25 insertions(+), 3 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/common/parameters.py
b/airflow-core/src/airflow/api_fastapi/common/parameters.py
index 76fe643fd6b..dfc255f4683 100644
--- a/airflow-core/src/airflow/api_fastapi/common/parameters.py
+++ b/airflow-core/src/airflow/api_fastapi/common/parameters.py
@@ -39,6 +39,7 @@ from pydantic import AfterValidator, BaseModel, NonNegativeInt
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
@@ -1480,10 +1481,13 @@ class _AnyDagRunStateFilter(BaseParam[DagRunState |
None]):
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)
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 14c7c73f657..33f4a5441d5 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
@@ -175,6 +175,24 @@ class TestGetDagRuns(TestPublicDagEndpoint):
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."""
