leonsmith edited a comment on issue #18501: URL: https://github.com/apache/airflow/issues/18501#issuecomment-933369645
I think the fix here is to use a window function over [next_dagruns_to_examine](https://github.com/apache/airflow/blob/main/airflow/models/dagrun.py#L211) so it only returns dagruns up to the dags maximum concurrency. I know this is the core loop so performance is a concern here. Is there any objection to something like the following approach before I put effort in writing some tests? Are there any concerns for the other backends with this type of approach? ```python if state == State.QUEUED: # For dag runs in the queued state, # We only return the number of rows required to take the dag overs # it's max_active_runs limit running_drs = ( session.query(cls.dag_id, func.count(cls.state).label('num_running')) .filter(cls.state == DagRunState.RUNNING) .group_by(cls.dag_id) .subquery() ) query = query.outerjoin(running_drs, running_drs.c.dag_id == DagRun.dag_id) row_number = func.row_number().over(partition_by=DagRun.dag_id) open_slots = DagModel.max_active_runs - func.coalesce(running_drs.c.num_running, 0) query = query.filter(row_number < open_slots) query = query.order_by( nulls_first(cls.last_scheduling_decision, session=session), cls.execution_date, ) ``` -- 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]
