[AIRFLOW-6590] Use batch db operations in jobs

[turbaszek]

The PR changes numerous single selects / updates in base, scheduler, and backfill jobs to bulk operations.

Measure

To measure the number of SQL queries I required two things:

• SQLAlchemy callbacks that log query information. This callback can be turned on / off by specifying sqlalchemy_stats in debug section of Airflow config.

@SQLALCHEMY 0.028488636016845703 |$ 'create_session':/opt/airflow/airflow/utils/session.py:32 |$ cli_action_loggers.py:default_action_log>contextlib.py:__exit__>session.py:create_session |$  INSERT INTO log (dttm,...

• Script to run successive SchedulerJob and to generate .csv with all queries per job. The script is located in scripts/perf/sql_queries.py and it works out of the box in the breeze environment.

I am using SequentialExecutor and scripts/perf/dags/sql_perf_dag.py and local Postgres.

Result

On the average, the number of queries was reduced by ~14% and time fo queries by ~10.9%.

Thanks for helping to @evgenyshulman from Databand!

---

Issue link: AIRFLOW-6590

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.

[potiuk]

Just one comment to Exception raised.

[codecov-io]

Codecov Report

❗ No coverage uploaded for pull request base (master@f3eea3e). Click here to learn what that means.
The diff coverage is 16.66%.

@@            Coverage Diff            @@
##             master    #7370   +/-   ##
=========================================
  Coverage          ?   86.31%           
=========================================
  Files             ?      873           
  Lines             ?    40769           
  Branches          ?        0           
=========================================
  Hits              ?    35188           
  Misses            ?     5581           
  Partials          ?        0

Impacted Files	Coverage Δ
airflow/cli/commands/task_command.py	72.12% <ø> (ø)
airflow/jobs/scheduler_job.py	89.35% <0%> (ø)
airflow/models/dagrun.py	96.47% <10%> (ø)
airflow/jobs/backfill_job.py	92.09% <13.33%> (ø)
airflow/jobs/base_job.py	91.6% <20%> (ø)
airflow/models/taskinstance.py	94.24% <28.57%> (ø)
airflow/utils/sqlalchemy.py	84.93% <30.76%> (ø)

---

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 f3eea3e...543a91a. Read the comment docs.

[ashb]

Nice work - taking a look now.

[potiuk]

Please rebase to latest master @nuclearpinguin - we had a six drama on Travis but it's solved now.

[ashb]

Nice change!

A couple of style points etc, and one question about the change to the tests.

[saguziel]

@nuclearpinguin I believe that this should be before session.commit().

Session.commit() should flush the session, and then sqlalchemy will need to requery each TI individually to reconstruct the STI