Nataneljpwd commented on code in PR #64109:
URL: https://github.com/apache/airflow/pull/64109#discussion_r3645485707
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -735,16 +735,33 @@ 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
+ ),
+ nulls_first(running_drs.c.num_running,
session=session),
+ DagRun.run_after,
+ ],
+ )
+ .label("rn"),
+ )
+ .where(DagRun.state == DagRunState.QUEUED)
.join(
- DagModel,
+ running_drs,
and_(
- DagModel.dag_id == cls.dag_id,
- DagModel.is_paused == false(),
- DagModel.is_stale == false(),
+ running_drs.c.dag_id == DagRun.dag_id,
+ running_drs.c.backfill_id == DagRun.backfill_id,
Review Comment:
Not here, in the next query part, if it had run id then all dags with no
runs would be ignored, this step produces a result that looks like so:
-----------------------------------------------------------
Dag id | Run id | num currently running | serial count of DR (later used for
filtering) |
Where the serial count is given and ordered by the dagrun prioritization
(the 3 order bys in the row number func)
Later in the next query part I select the amount of available dagrun slots
for given dag (max active runs - current run count) by the row number (where
the serially increasing number is less than the available slots)
And because there are row locks for the dagrun, only those dagruns are
available to run and it selects exactly max dagruns per loop to schedule
dagruns (or less if there aren't any) which can be scheduled by filtering those
which cannot move to running in the SQL query
I will give an example:
If we have dags A, B and C where A has aax active runs of 1, B has 1 and C
has 3, the current implementation selects the first 5 (for this example)
dagruns ordered by their respective "priority"
Assume we have the following situations, there is already 2 DR running for
C and many more in queued
The queued DR select looks like so:
CCCCCCCAAABBACCCAB
The current implementation will select 5 of the C DR, schedule 1 and go on,
the next iteration same thing happens and 1 DR from a Is set to running
The implementation proposed changes the result of the query by only
selecting DRs which can be set to running, hence the DR queue will look like so
(and the query result as well)
CAB
And nothing else gets returned
This is a simplified example, with more DRs this issue cam become worse and
DRs are not moved to running and stay queued for a long time
If something isn't clear let me know and I'll try to clarify
Thank you
--
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]
