This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun 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 1f70ead81ef Fix /ui/dags recent-runs query degradation on large,
skewed dag_run tables (#74401)
1f70ead81ef is described below
commit 1f70ead81ef00ac8e0ddb338a862d4f0f226a78b
Author: Ivan Tham <[email protected]>
AuthorDate: Wed Oct 14 19:30:22 2026 +0800
Fix /ui/dags recent-runs query degradation on large, skewed dag_run tables
(#74401)
* Fix /ui/dags recent-runs query degradation on large, skewed dag_run tables
The per-Dag UNION ALL branches introduced in #67721 plan as backward index
scans of the global idx_dag_run_run_after with a Filter on dag_id. On a
large, skewed dag_run table this is catastrophic for "cold" Dags — those
whose last run is months old or that never ran: the scan must walk
essentially every row of every *other* Dag before reaching that branch's
rows (measured on a 3.8M-row metadata DB: 3,800,261 rows removed by filter,
~2.3s for a single cold branch, 30-60s per page — gateway 504s), because
at default statistics the planner misestimates cold dag_ids (they fall
outside the MCV list) and bets the backward scan hits LIMIT matches
quickly. Hot pages are affected too, bistably: the identical statement
was observed flipping 22.8ms <-> 141s across a single ANALYZE.
Replace the 50-branch union with one page-scoped window query:
row_number() OVER (PARTITION BY dag_id ORDER BY run_after DESC, id DESC)
over WHERE dag_id IN (page's dag ids), filtered to rn <= dag_runs_limit.
This restores the pre-#67721 access path (per-Dag probes of
idx_dag_run_dag_id + one sort + WindowAgg), bounded regardless of where
each Dag's runs sit in time or what the planner estimates, while keeping
#67721's exact-N-per-Dag semantics. On run_after ties the window's
(id DESC) tiebreak deterministically keeps the latest-inserted row,
where the old per-branch LIMIT picked arbitrarily; the outer ORDER BY
leaves tied rows in unspecified order, same as the union. All columns
are selected inline (no join-back). dag_runs_limit is now NonNegativeInt,
matching limit/offset.
Measured end-to-end on a dockerized copy of a 3.8M-row production
metadata DB (PostgreSQL 15.14, default statistics, 74%-cold first page):
GET /ui/dags?limit=50&dag_runs_limit=14: 31.2s -> 0.13-0.64s
same rows and order returned (291/291 id-set match, total_entries
unchanged); selection can differ from the union only on run_after ties,
where the new pick is the newer run
Trade-off: the window sorts all runs of the page's Dags, so a page of the
50 highest-run-count Dags (~619K rows) costs ~2.0s with a 58MB external
sort (vs the union's 0.25-12s bistable plan) — linear and deterministic
instead of the 30-60s+ cold-page catastrophe.
The window-function shape is portable (same ANSI syntax as the
pre-#67721 rank() query and the existing row_number usage in
common/db/assets.py), so the sqlite/mysql/mssql matrix is unaffected.
Co-Authored-By: Claude Code <[email protected]>
* Add tests for recent-runs limit and run_after tie-breaking
Per review: one test asserts a Dag with more runs than dag_runs_limit
returns exactly the N newest in descending run_after order; one test
pins the (run_after DESC, id DESC) tie-break — with two runs tied on
run_after, the later-inserted run (higher id) is the deterministic pick,
which is the one spot the window query's output differs from the old
per-branch union.
Co-Authored-By: Claude Code <[email protected]>
* Remove newsfragment per review
Co-Authored-By: Claude Code <[email protected]>
* Trim redundant comments in the recent-runs tests
Co-Authored-By: Claude Code <[email protected]>
* Keep dag_runs_limit as plain int
The NonNegativeInt validation is a contract change unrelated to the
regression this PR fixes — out of scope here.
Co-Authored-By: Claude Code <[email protected]>
* Drop the explanatory comment above the ranked query
Co-Authored-By: Claude Code <[email protected]>
---------
Co-authored-by: Claude Code <[email protected]>
---
.../airflow/api_fastapi/core_api/routes/ui/dags.py | 31 +++++++++----
.../api_fastapi/core_api/routes/ui/test_dags.py | 51 ++++++++++++++++++++++
2 files changed, 74 insertions(+), 8 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dags.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dags.py
index 6888ed886a5..23886e8840b 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dags.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dags.py
@@ -233,7 +233,7 @@ def get_dags(
recent_dag_runs: list = []
if dags:
- recent_runs_branches = [
+ ranked = (
select(
DagRun.id,
DagRun.dag_id,
@@ -243,16 +243,31 @@ def get_dags(
DagRun.run_after,
DagRun.start_date,
DagRun.state,
+ func.row_number()
+ .over(
+ partition_by=DagRun.dag_id,
+ order_by=(DagRun.run_after.desc(), DagRun.id.desc()),
+ )
+ .label("rn"),
)
- .where(DagRun.dag_id == dag.dag_id)
- .order_by(DagRun.run_after.desc())
- .limit(dag_runs_limit)
+ .where(DagRun.dag_id.in_([dag.dag_id for dag in dags]))
.subquery()
- for dag in dags
- ]
- recent_runs_union = union_all(*(select(branch) for branch in
recent_runs_branches)).subquery()
+ )
recent_dag_runs = list(
-
session.execute(select(recent_runs_union).order_by(recent_runs_union.c.run_after.desc()))
+ session.execute(
+ select(
+ ranked.c.id,
+ ranked.c.dag_id,
+ ranked.c.run_id,
+ ranked.c.end_date,
+ ranked.c.logical_date,
+ ranked.c.run_after,
+ ranked.c.start_date,
+ ranked.c.state,
+ )
+ .where(ranked.c.rn <= dag_runs_limit)
+ .order_by(ranked.c.run_after.desc())
+ )
)
# Fetch pending HITL actions for each Dag if we are not certain whether
some of the Dag might contain HITL actions
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 ed9a5f36ae0..dcee4b1a86b 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
@@ -207,6 +207,57 @@ class TestGetDagRuns(TestPublicDagEndpoint):
assert dags_by_id[DAG1_ID]["has_unfinished_runs"] is True
assert dags_by_id[DAG2_ID]["has_unfinished_runs"] is False
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ def test_recent_dag_runs_returns_exactly_n_newest_in_order(self,
test_client, session):
+ """A Dag with more runs than dag_runs_limit returns exactly the N
newest, newest first."""
+ response = test_client.get(
+ "/dags",
+ params={"dag_ids": [DAG1_ID], "dag_runs_limit": 3},
+ )
+
+ assert response.status_code == 200
+ dag_runs = response.json()["dags"][0]["latest_dag_runs"]
+ assert [run["run_id"] for run in dag_runs] == ["run_id_5", "run_id_4",
"run_id_3"]
+ run_afters = [run["run_after"] for run in dag_runs]
+ assert run_afters == sorted(run_afters, reverse=True)
+
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ def test_recent_dag_runs_tie_breaks_on_id_desc(self, test_client, session):
+ """Two runs tied on run_after: the later-inserted one (higher id)
wins, deterministically."""
+ tie_run_after = pendulum.datetime(2030, 6, 1, tz="UTC")
+ session.add_all(
+ [
+ DagRun(
+ dag_id=DAG1_ID,
+ run_id="tie_first_inserted",
+ run_type=DagRunType.MANUAL,
+ logical_date=tie_run_after - pendulum.duration(days=1),
+ run_after=tie_run_after,
+ state=DagRunState.SUCCESS,
+ triggered_by=DagRunTriggeredByType.TEST,
+ ),
+ DagRun(
+ dag_id=DAG1_ID,
+ run_id="tie_second_inserted",
+ run_type=DagRunType.MANUAL,
+ logical_date=tie_run_after - pendulum.duration(days=2),
+ run_after=tie_run_after,
+ state=DagRunState.SUCCESS,
+ triggered_by=DagRunTriggeredByType.TEST,
+ ),
+ ]
+ )
+ session.commit()
+
+ response = test_client.get(
+ "/dags",
+ params={"dag_ids": [DAG1_ID], "dag_runs_limit": 1},
+ )
+
+ assert response.status_code == 200
+ dag_runs = response.json()["dags"][0]["latest_dag_runs"]
+ assert dag_runs[0]["run_id"] == "tie_second_inserted"
+
@pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
@pytest.mark.parametrize(
("query_params", "expected_ids"),
