kaxil commented on code in PR #64109:
URL: https://github.com/apache/airflow/pull/64109#discussion_r4243926254
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -811,16 +811,32 @@ def get_queued_dag_runs_to_set_running(cls, session:
Session) -> ScalarResult[Da
.subquery()
)
- query = (
- select(cls)
- .where(cls.state == DagRunState.QUEUED)
+ available_dagruns_rn = (
+ select(
+ DagRun.dag_id,
+ DagRun.id,
+ running_drs.c.num_running,
+ func.row_number()
+ .over(
+ partition_by=[DagRun.dag_id, DagRun.backfill_id],
+ order_by=[
+ nulls_first(cast("ColumnElement[Any]",
BackfillDagRun.sort_ordinal), session=session),
+ nulls_first(
+ cast("ColumnElement[Any]",
cls.last_scheduling_decision), session=session
+ ),
+ DagRun.run_after,
+ ],
+ )
+ .label("rn"),
+ )
+ .where(DagRun.state == DagRunState.QUEUED)
Review Comment:
The window ranks every queued run in the partition, but `run_after <= now()`
is only applied on the outer query (line 888), after `rn <= cap - num_running`.
So a run that isn't due yet can take a slot and then get dropped, and the ready
run ranked after it never comes back.
That's reachable on main today: a reverse backfill on a partitioned
timetable can have `to_date` in the future (validation in `backfill.py` only
rejects when both bounds are in the future, and `_get_info_list` keeps future
partitions for partitioned timetables). Reversing gives those future partitions
`sort_ordinal` 1..N, which is the first window key. With the default backfill
`max_active_runs=10` and ten or more future partitions, the backfill returns
nothing until they mature, where main starts the newest past partitions right
away.
Moving the `run_after` filter into this subquery fixes it, and the comment
at line 873 would then be true. Since you'd be touching the subquery anyway,
I'd move the paused/stale `DagModel` join in here as well. I don't think the
planner can push a join condition into a subquery that computes a window
function, so as written the ranking also runs over queued runs of paused Dags.
##########
airflow-core/tests/unit/jobs/test_scheduler_job.py:
##########
@@ -4477,6 +4480,94 @@ def
test_runs_are_created_after_max_active_runs_was_reached(self, dag_maker, ses
dag_runs = DagRun.find(dag_id=dag.dag_id, session=session)
assert len(dag_runs) == 2
+ def test_runs_are_not_starved_by_max_active_runs_limit(self, dag_maker,
session):
Review Comment:
This one also passes with `dagrun.py` reverted to main (checked in breeze),
same as ashb's point on the test below. The three Dags share `start_date` and
schedule, so their queued runs interleave by `run_after`, main's top 20 holds
about 7 per Dag, and the Python guard still starts 3 of each.
To fail on main, one Dag's backlog has to sort entirely ahead of a Dag that
has capacity. For example: Dag A with `max_active_runs=1` and 25 queued runs,
Dag B starting a day later with `max_active_runs=3` and 5 queued runs, then
assert `{A: 1, B: 3}` running. Main gives `{A: 1}` and this branch passes.
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -830,37 +846,42 @@ def get_queued_dag_runs_to_set_running(cls, session:
Session) -> ScalarResult[Da
),
isouter=True,
)
- .join(Backfill, isouter=True)
+ .subquery()
+ )
+
+ query = (
+ select(cls)
.join(
- running_drs,
+ available_dagruns_rn,
+ available_dagruns_rn.c.id == DagRun.id,
+ )
+ .join(
+ DagModel,
and_(
- running_drs.c.dag_id == DagRun.dag_id,
- coalesce(running_drs.c.backfill_id, text("-1"))
- == coalesce(DagRun.backfill_id, text("-1")),
+ DagModel.dag_id == cls.dag_id,
+ DagModel.is_paused == false(),
+ DagModel.is_stale == false(),
),
- isouter=True,
)
+ .order_by(
Review Comment:
On main the outer `ORDER BY` started with `sort_ordinal` NULLS FIRST so
backfill runs sort after every non-backfill run. That key now only lives in the
window, so it no longer orders anything across partitions. Backfill runs
usually have an older `run_after`, so they now win the 20-row limit over
scheduled runs. In a breeze run with one backfill (cap 3) and a regular Dag
with 30 queued runs, main's query started 20 scheduled runs and this one
started 3 backfill plus 17 scheduled.
Could you select `sort_ordinal` from the subquery and put it back as the
first key here?
`test_backfill_runs_are_started_with_lower_priority_catchup_true` can't catch
this as rewritten, since 1 + 3 + 16 startable runs exactly fill the limit. It
needs more startable runs than `DEFAULT_DAGRUNS_TO_EXAMINE`, and dag2's
`max_active_runs` set explicitly instead of relying on the 16 from config.
##########
airflow-core/tests/unit/jobs/test_scheduler_job.py:
##########
@@ -7752,41 +7845,55 @@ def _running_counts():
EmptyOperator(task_id="mytask")
dr = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED,
state=State.QUEUED)
- for _ in range(9):
+ for _ in range(29):
dr = dag_maker.create_dagrun_after(dr,
run_type=DagRunType.SCHEDULED, state=State.QUEUED)
# initial state -- nothing is running
assert dag1_non_b_running == 0
assert dag1_b_running == 0
assert total_running == 0
- assert session.scalar(select(func.count(DagRun.id))) == 46
+ assert session.scalar(select(func.count(DagRun.id))) == 66
assert session.scalar(select(func.count()).where(DagRun.dag_id ==
dag1_dag_id)) == 36
# now let's run it once
self.job_runner._start_queued_dagruns(session)
session.flush()
# after running the scheduler one time, observe that only one dag run
is started
Review Comment:
This still says only one dag run is started, but the assert below is now 20.
The `session.flush()` right after `session.commit()` on line 7883 doesn't do
anything either.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
