[AIRFLOW-6869] Bulk fetch DAGRuns for _process_task_instances

[mik-laj]

Another performance optimization.
When a DAG file contains 199 DAGs, and each DAG contains 5 tasks

from datetime import timedelta

from airflow.models import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.dummy_operator import DummyOperator
from airflow.utils.dates import days_ago

args = {
    'owner': 'airflow',
    'start_date': days_ago(3),
}


def create_dag(dag_number):
    dag = DAG(
        dag_id=f'perf_50_dag_dummy_tasks_{dag_number}_of_50', default_args=args,
        dagrun_timeout=timedelta(minutes=60),
        is_paused_upon_creation=False,
    )

    for j in range(1, 10):
        DummyOperator(
            task_id='task_{}_of_5'.format(j),
            dag=dag
        )

    return dag


for i in range(1, 200):
    globals()[f"dag_{i}"] = create_dag(i)

I ran the following code:

from airflow.jobs.scheduler_job import DagFileProcessor

processor = DagFileProcessor([], log)
processor.process_file(DAG_FILE, None, pickle_dags=False)

I got the following values.
Before:
Query count: 1792
Average time: 4568.156 ms
After:
Query count: 1594
Average time: 3964.916 ms
Diff:
Query count: -200 (-11%)
Average time: -603 ms (-13%)

If I didn't make a mistake, when we combine the following changes:

• [AIRFLOW-6862] Do not check the freshness of fresh DAG #7481
• [AIRFLOW-6857] Bulk sync DAGs #7477
• [AIRFLOW-6856] Bulk fetch paused_dag_ids #7476

we get only... 5 queries per file instead of ....1792

Thanks for support to @evgenyshulman from Databand!

---

Issue link: AIRFLOW-6869

Make sure to mark the boxes below before creating PR: [x]

• Description above provides context of the change
• Commit message/PR title starts with [AIRFLOW-NNNN]. AIRFLOW-NNNN = JIRA ID^*
• Unit tests coverage for changes (not needed for documentation changes)
• Commits follow "How to write a good git commit message"
• Relevant documentation is updated including usage instructions.
• I will engage committers as explained in Contribution Workflow Example.

^* For document-only changes commit message can start with [AIRFLOW-XXXX].

---

In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.
Read the Pull Request Guidelines for more information.

[codecov-io]

Codecov Report

Merging #7489 into master will decrease coverage by 0.25%.
The diff coverage is 96.66%.

@@            Coverage Diff             @@
##           master    #7489      +/-   ##
==========================================
- Coverage   86.79%   86.54%   -0.26%     
==========================================
  Files         887      891       +4     
  Lines       41976    42110     +134     
==========================================
+ Hits        36432    36442      +10     
- Misses       5544     5668     +124

Impacted Files	Coverage Δ
airflow/jobs/scheduler_job.py	90.11% <100%> (-0.13%)	⬇️
airflow/models/dagrun.py	96.2% <83.33%> (-0.37%)	⬇️
airflow/kubernetes/volume_mount.py	44.44% <0%> (-55.56%)	⬇️
airflow/kubernetes/volume.py	52.94% <0%> (-47.06%)	⬇️
airflow/kubernetes/pod_launcher.py	47.18% <0%> (-45.08%)	⬇️
...viders/cncf/kubernetes/operators/kubernetes_pod.py	69.38% <0%> (-25.52%)	⬇️
airflow/kubernetes/refresh_config.py	50.98% <0%> (-23.53%)	⬇️
airflow/utils/file.py	87.83% <0%> (-1.36%)	⬇️
airflow/utils/dag_processing.py	87.93% <0%> (-0.2%)	⬇️
airflow/models/baseoperator.py	96.52% <0%> (ø)	⬆️
... and 27 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 4e0e2f0...f94e1fc. Read the comment docs.

[ashb]

The behaviour is different on py3.6 and 3.7, but is still wrong on both when a more than a single item is in items: 3.6 would return the last item only.

The docs for groupby say:

Because the source is shared, when the groupby() object is advanced, the previous group is no longer visible. So, if that data is needed later, it should be stored as a list.

Since the behaviour without list is broken in otherways on 3.6 too I think we can just replace this comment with:

         # As per the docs of groupby (https://docs.python.org/3/library/itertools.html#itertools.groupby)
         # we need to use `list()` otherwise the result will be wrong/incomplete

[houqp]

+1 on the comment change recommended by @ashb. Based the official doc, behavior from both 3.6 and 3.7 are expected and within the API spec. Returned group items are transient with each iteration and should be manually persisted if needed.

[mik-laj]

I will update the description during the next rebase.

[mik-laj]

Done.

[saguziel]

d.get(dag_id, []) instead pls

[mik-laj]

Done

[Fokko]

LGTM, small suggestions :)