kaxil commented on code in PR #64109:
URL: https://github.com/apache/airflow/pull/64109#discussion_r3133307791
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -713,8 +731,12 @@ def get_queued_dag_runs_to_set_running(cls, session:
Session) -> ScalarResult[Da
# the one done in this query verifies that the dag is not
maxed out
# it could return many more dag runs than runnable if there is
even
# capacity for 1. this could be improved.
- coalesce(running_drs.c.num_running, text("0"))
- < coalesce(Backfill.max_active_runs, DagModel.max_active_runs),
+ available_dagruns_rn.c.rn
+ <= coalesce(
+ Backfill.max_active_runs,
+ DagModel.max_active_runs,
Review Comment:
`DagModel.max_active_runs` is nullable (see `dag.py:403`, with a TODO saying
"should not be nullable"). If a DagModel row has NULL,
`coalesce(Backfill.max_active_runs, DagModel.max_active_runs)` is NULL, `NULL -
X` is NULL, and the predicate silently filters out every queued run for that
DAG -- hard to diagnose. Worth adding a config-default fallback:
`coalesce(Backfill.max_active_runs, DagModel.max_active_runs, <config_max>)`.
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -679,9 +679,27 @@ def get_queued_dag_runs_to_set_running(cls, session:
Session) -> ScalarResult[Da
.subquery()
)
+ available_dagruns_rn = (
+ select(
+ DagRun.dag_id,
+ DagRun.id,
+ func.row_number()
+ .over(partition_by=[DagRun.dag_id, DagRun.backfill_id],
order_by=DagRun.logical_date)
Review Comment:
Partitioned backfills set `logical_date = NULL` (see `backfill.py:709-710`),
so `row_number() OVER (ORDER BY logical_date)` has no tie-breaker -- the NULL
rows are ranked in arbitrary order. The subsequent `rn <= max_active_runs -
num_running` then keeps an unpredictable subset, breaking backfill ordering
guarantees. Add a deterministic secondary sort (e.g. `sort_ordinal`,
`run_after`, or `id`) to the window's `order_by`.
##########
airflow-core/tests/unit/jobs/test_scheduler_job.py:
##########
@@ -5883,41 +5971,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
- # this is because there are 30 runs for dag 1 so neither the backfills
nor
- # any runs for dag2 get started
+ # and 3 backfill dagruns are started
+ # this is because there are 30 queued dagruns, many of which get
filtered because their DAGs
+ # have already reached max_active_runs
+ # and so due to the default dagruns-to-examine limit, we look at the
first 20 dagruns that CAN be run
+ # according to the max_active_runs parameter, meaning 3 backfill runs
will start, 1 non-backfill,
+ # and all runnable dagruns for dag2
assert DagRun.DEFAULT_DAGRUNS_TO_EXAMINE == 20
dag1_non_b_running, dag1_b_running, total_running = _running_counts()
assert dag1_non_b_running == 1
- assert dag1_b_running == 0
- assert total_running == 1
- assert session.scalar(select(func.count()).select_from(DagRun)) == 46
+ assert dag1_b_running == 3
+ assert total_running == 20
+ assert session.scalar(select(func.count()).select_from(DagRun)) == 66
assert session.scalar(select(func.count()).where(DagRun.dag_id ==
dag1_dag_id)) == 36
+ # now we finish all lower priority backfill tasks, and observe new
higher priority tasks are started
Review Comment:
Comment says "finish all lower priority backfill tasks" but the update below
targets `test_dag2` runs, which are created with
`run_type=DagRunType.SCHEDULED` (not backfill). Could you update the comment to
match what's actually being flipped?
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -679,9 +679,27 @@ def get_queued_dag_runs_to_set_running(cls, session:
Session) -> ScalarResult[Da
.subquery()
)
+ available_dagruns_rn = (
+ select(
+ DagRun.dag_id,
+ DagRun.id,
+ func.row_number()
+ .over(partition_by=[DagRun.dag_id, DagRun.backfill_id],
order_by=DagRun.logical_date)
+ .label("rn"),
+ )
+ .where(DagRun.state == DagRunState.QUEUED)
Review Comment:
The subquery runs `row_number()` over every queued DagRun in the DB,
including runs for paused or stale DAGs that the outer `JOIN DagModel ON
is_paused = false AND is_stale = false` filters out afterwards. Pushing those
predicates into this subquery's WHERE would shrink the window input before
ranking. Runs every scheduler loop, so worth it.
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -679,9 +679,27 @@ def get_queued_dag_runs_to_set_running(cls, session:
Session) -> ScalarResult[Da
.subquery()
)
+ available_dagruns_rn = (
+ select(
+ DagRun.dag_id,
+ DagRun.id,
+ func.row_number()
+ .over(partition_by=[DagRun.dag_id, DagRun.backfill_id],
order_by=DagRun.logical_date)
Review Comment:
The window orders by `logical_date` ASC, but for reverse backfills
`sort_ordinal=1` is the *latest* logical_date (see `backfill.py:524-525` where
`dagrun_info_list` is reversed before `enumerate(..., start=1)` in
`_create_runs_*`). So the subquery keeps earliest-by-logical_date runs, which
are the ones with the *highest* `sort_ordinal` (lowest priority for a reverse
backfill). The outer `order_by(sort_ordinal)` then cannot recover them since
they have already been filtered. Consider ordering the window by the same key
used for prioritization (e.g. `BackfillDagRun.sort_ordinal` joined into the
subquery) so the kept rows match the outer priority.
--
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]
