Fix DetachedInstanceError when dag_run attrs are accessed from ti

[ephraimbuddy]

Fix DetachedInstanceError when dag_run attrs are accessed from ti

Loading a taskinstance doesn't load the corresponding dag_run with
the effect that when the dag_run attr is accessed from the ti it gives
a DetachedInstanceError, dag_run is not bound to a session.

This PR fixes it by making an extra query to get the dagrun using the
relationship between the two objects

cc: @ash @uranusjr
I think with this, we can also get rid of ti.get_dagrun method?

---

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
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.

[ashb]

Hmmmmm, the selectin to the relationship is probably safer, but in cases where we know we are going to do this it's better to add options(eagerload(TI.dag_run)) etc to avoid the extra query.

No we can't remove it -- we can deprecate it but it can't be removed until 3.0

[ashb]

If possible we don't want these query counts to change.

[ashb]

@ephraimbuddy A fix like #18359 might also fix this problem, as might #18503

I think having this "fallback" to loading is better than an error, but by doing it like one of these other two it should avoid an extra query.

[ephraimbuddy]

It's not obvious that ti.dag_run is a detached object

@ephraimbuddy A fix like #18359 might also fix this problem, as might #18503

I think having this "fallback" to loading is better than an error, but by doing it like one of these other two it should avoid an extra query.

Yeah. You're right.

[ephraimbuddy]

The cause can easily be fixed by using ti.get_dagrun instead of ti.dag_run, see

airflow/airflow/providers/elasticsearch/log/es_task_handler.py

Lines 104 to 113 in 18e91bc

	dag_run = ti.dag_run
	if self.json_format:
	data_interval_start = self._clean_date(dag_run.data_interval_start)
	data_interval_end = self._clean_date(dag_run.data_interval_end)
	execution_date = self._clean_date(dag_run.execution_date)
	else:
	data_interval_start = dag_run.data_interval_start.isoformat()
	data_interval_end = dag_run.data_interval_end.isoformat()
	execution_date = dag_run.execution_date.isoformat()

Should I fix it at the source of the error?
EDIT:
Working on a solution now

[ephraimbuddy]

After not finding a way I could work this to load dag_run, this fix worked. I think it's better to use this instead of an extra query each time a task instance is loaded.

[ephraimbuddy]

After not finding a way I could work this to load dag_run, this fix worked. I think it's better to use this instead of an extra query each time a task instance is loaded.

If this fix is Ok, I will modify the title and commit message

[github-actions]

The PR is likely OK to be merged with just subset of tests for default Python and Database versions without running the full matrix of tests, because it does not modify the core of Airflow. If the committers decide that the full tests matrix is needed, they will add the label 'full tests needed'. Then you should rebase to the latest main or amend the last commit of the PR, and push it with --force-with-lease.

[ashb]

Much more direct - any we want to deprecate uses of ti.executipn_date.

That said: does this mean that the old version of the ES provider won't work with Airflow 2.2?

We can fix that by setting the deps somewhere in provider_manager; or add a eagerload(TI.dag_run) in airflow core before hitting this code

[ephraimbuddy]

@ashb, I applied this fix #18503 and it worked.
I also applied a fix where I attached dag_run to this ti:

airflow/airflow/cli/commands/task_command.py

Line 88 in 2643345

ti = dag_run.get_task_instance(task.task_id)

ti.dag_run=dag_run did not work but ti.dag_run = ti.get_dagrun() worked.
I would go with #18503. What do you suggest?

[ephraimbuddy]

@ashb, I applied this fix #18503 and it worked.
I also applied a fix where I attached dag_run to this ti:

airflow/airflow/cli/commands/task_command.py

Line 88 in 2643345

ti = dag_run.get_task_instance(task.task_id)

ti.dag_run=dag_run did not work but ti.dag_run = ti.get_dagrun() worked.
I would go with #18503. What do you suggest?

I found that there's an eager loading option in the relationship(used for an attribute that's guaranteed to have an element) and that reduced the query!

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.