Repository: apache/airflow
File: airflow-core/src/airflow/jobs/scheduler_job_runner.py (Python)

Code before:
                        "Not executing %s. Requested pool slots (%s) are greater than "
                        "total pool slots: '%s' for pool: %s.",
                        task_instance,
                        task_instance.pool_slots,
                        pool_total,
                        pool_name,
                    )

                    pool_num_starving_tasks[pool_name] += 1
                    num_starving_tasks_total += 1
                    starved_tasks.add((task_instance.dag_id, task_instance.task_id))
                    continue

                if task_instance.pool_slots > open_slots:
                    self.log.info(
                        "Not executing %s since it requires %s slots "
                        "but there are %s open slots in the pool %s.",
                        task_instance,
                        task_instance.pool_slots,
                        open_slots,
                        pool_name,
                    )
                    pool_num_starving_tasks[pool_name] += 1
                    num_starving_tasks_total += 1
                    starved_tasks.add((task_instance.dag_id, task_instance.task_id))
                    # Though we can execute tasks with lower priority if there's enough room
                    continue

                # Check to make sure that the task max_active_tasks of the DAG hasn't been
                # reached.
                dag_id = task_instance.dag_id
                dag_run_key = (dag_id, task_instance.run_id)
                current_active_tasks_per_dag_run = concurrency_map.dag_run_active_tasks_map[dag_run_key]
                dag_max_active_tasks = task_instance.dag_model.max_active_tasks
                self.log.info(
                    "DAG %s has %s/%s running and queued tasks",
                    dag_id,
                    current_active_tasks_per_dag_run,
                    dag_max_active_tasks,
                )
                if current_active_tasks_per_dag_run >= dag_max_active_tasks:
                    self.log.info(
                        "Not executing %s since the number of tasks running or queued "
                        "from DAG %s is >= to the DAG's max_active_tasks limit of %s",
                        task_instance,
                        dag_id,
                        dag_max_active_tasks,
                    )
                    starved_dags.add(dag_id)
                    continue


Review comment (suggestion):
Similar thing here: The window only partitions on `run_id`, which collides across DAGs. Once one DAG consumes its `max_active_tasks`, every other DAG with the same `run_id` is filtered out by `row_num <= dr_max_active_tasks`, even if they have zero active tasks. 

Better would be to partition by both `TI.dag_id` and `TI.run_id` (or the dag run PK) before applying the limit so each DAG run keeps its own quota.


Code after:
                    TI.map_index.label("map_index_for_ordering"),
                )
            ).subquery()

            # Select only rows where row_number <= max_active_tasks.
            query = (
                select(TI)
                .with_hint(TI, "USE INDEX (ti_state)", dialect_name="mysql")
                .select_from(ranked_query)
                .join(
                    TI,
                    (TI.dag_id == ranked_query.c.dag_id)
                    & (TI.task_id == ranked_query.c.task_id)
                    & (TI.run_id == ranked_query.c.run_id)
                    & (TI.map_index == ranked_query.c.map_index),
                )
                .where(ranked_query.c.row_num <= ranked_query.c.dr_max_active_tasks)
                # Add the order_by columns from the ranked query for sqlite.
                .order_by(
                    -ranked_query.c.priority_weight_for_ordering,
                    ranked_query.c.logical_date_for_ordering,
                    ranked_query.c.map_index_for_ordering,
                )
                .options(selectinload(TI.dag_model))
            )

            query = query.limit(max_tis)

            timer = Stats.timer("scheduler.critical_section_query_duration")
            timer.start()

            try:
                locked_query = with_row_locks(query, of=TI, session=session, skip_locked=True)
                task_instances_to_examine = session.scalars(locked_query).all()

                if self.log.isEnabledFor(logging.DEBUG):
                    self.log.debug("Length of the tis to examine is %d", len(task_instances_to_examine))
                    self.log.debug(
                        "TaskInstance selection is: %s",
                        dict(Counter(ti.dag_id for ti in task_instances_to_examine)),
                    )

                timer.stop(send=True)
            except OperationalError as e:
                timer.stop(send=False)
                raise e

            # TODO[HA]: This was wrong before anyway, as it only looked at a sub-set of dags, not everything.
            # Stats.gauge('scheduler.tasks.pending', len(task_instances_to_examine))

            if not task_instances_to_examine: