Query TaskReschedule only if task is UP_FOR_RESCHEDULE

[turbaszek]

---

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

• Description above provides context of the change
• Unit tests coverage for changes (not needed for documentation changes)
• Target Github ISSUE in description if exists
• 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.

---

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]

Nice optimisation :)

[mik-laj]

@seelmann Can you look at it?

[seelmann]

Makes perfect sense, +1.

Another optimization could be to not fetch all TaskReschedule objects but only the first one. Similar in ready_to_reschedule.py only the last one is needed. I wanted to change that long time ago but didn't find the time and then forgot...

[turbaszek]

Another optimization could be to not fetch all TaskReschedule objects but only the first one. Similar in ready_to_reschedule.py only the last one is needed.

Sounds reasonable, thanks!

[turbaszek]

@seelmann would like to take another look?

[seelmann]

LGTM

[mik-laj]

Suggested change

	task_reschedule = TaskReschedule.query_for_task_instance(task_instance=ti, descending=True).first()
	task_reschedule = (
	TaskReschedule.query_for_task_instance(task_instance=ti, descending=True)
	.with_entities(TR.reschedule_date)
	.first()
	)

[ashb]

Oh does with_entites say "just select this one column? Nice!

[mik-laj]

This allows you to change the content of the parameter passed to the query method on an existing object. We already use it in the project. https://github.com/apache/airflow/pull/8729/files

[mik-laj]

Here is doc: https://docs.sqlalchemy.org/en/13/orm/query.html#sqlalchemy.orm.query.Query.with_entities

[ashb]

Please pass on the same session object.

Suggested change

	task_reschedule = TaskReschedule.query_for_task_instance(task_instance=ti, descending=True).first()
	task_reschedule = (
	TaskReschedule.query_for_task_instance(task_instance=ti, descending=True, session=session)
	.with_entities(TR.reschedule_date)
	.first()
	)

[ashb]

You missed Kamil's suggestion of .with_entites here.

[turbaszek]

I think it's there :)

[turbaszek]

@ashb @mik-laj your suggestions are added

[mik-laj]

@turbaszek Should we merge it?