Simplify query for orphaned tasks

[dstandish]

This query runs periodically (controlled by orphaned_tasks_check_interval setting).

We observed that it can take around a minute to run for very large TI tables. While looking into that issue, I noticed this query was more complicated than it needed to be. This PR simplifies the query, though it may not meaningfully help performance.

Two changes here.

First, previously we ended up with two joins to DagRun because the dag_run relationship attr is lazy="joined" and sqlalchemy was not using it. By setting to be lazy, we eliminate the extra join and we also don't ask for any columns in dag run (previously the generated query asked for all of them, even though we try to limit via options further down).

Second, we use inner join for queued by job. It seems that the outer join was there when this adoption logic was first added in #10729, and the comment added at the time indicated that it was only there to handle tasks in flight during upgrade to 2.0:

                   # outerjoin is because we didn't use to have queued_by_job
                   # set, so we need to pick up anything pre upgrade. This (and the
                   # "or queued_by_job_id IS NONE") can go as soon as scheduler HA is
                   # released.

Before:

SELECT
    task_instance.task_id,
    task_instance.dag_id,
    task_instance.run_id,
    task_instance.map_index,
    dag_run_1.state,
    dag_run_1.id,
    dag_run_1.dag_id AS dag_id_1,
    dag_run_1.queued_at,
    dag_run_1.execution_date,
    dag_run_1.start_date,
    dag_run_1.end_date,
    dag_run_1.run_id AS run_id_1,
    dag_run_1.creating_job_id,
    dag_run_1.external_trigger,
    dag_run_1.run_type,
    dag_run_1.conf,
    dag_run_1.data_interval_start,
    dag_run_1.data_interval_end,
    dag_run_1.last_scheduling_decision,
    dag_run_1.dag_hash,
    dag_run_1.log_template_id,
    dag_run_1.updated_at,
    dag_run_1.clear_number
FROM task_instance
LEFT OUTER JOIN job ON job.id = task_instance.queued_by_job_id
JOIN dag_run ON dag_run.dag_id = task_instance.dag_id AND dag_run.run_id = task_instance.run_id
JOIN dag_run AS dag_run_1 ON dag_run_1.dag_id = task_instance.dag_id AND dag_run_1.run_id = task_instance.run_id
WHERE
    task_instance.state IN (__[postcompile_state_1])
    AND (task_instance.queued_by_job_id IS NULL OR job.state != :state_2)
    AND dag_run.run_type != :run_type_1
    AND dag_run.state = :state_3

After:

SELECT
    task_instance.task_id,
    task_instance.dag_id,
    task_instance.run_id,
    task_instance.map_index
FROM task_instance
JOIN job ON job.id = task_instance.queued_by_job_id
JOIN dag_run ON dag_run.dag_id = task_instance.dag_id AND dag_run.run_id = task_instance.run_id
WHERE
    task_instance.state IN (__[postcompile_state_1])
    AND job.state != :state_2
    AND dag_run.run_type != :run_type_1
    AND dag_run.state = :state_3

[dirrao]

Good optimization. I see related test cases are failing. It would be great if you can fix them.

[dirrao]

LGTM

[dstandish]

can i get an amen @jedcunningham @uranusjr @ephraimbuddy