xBis7 commented on code in PR #71737:
URL: https://github.com/apache/airflow/pull/71737#discussion_r3864468948
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -1737,10 +1812,28 @@ def _expand_mapped_task_if_needed(ti: TI) ->
Iterable[TI] | None:
return ready_tis, changed_tis, expansion_happened
+ def _ensure_type_task_instance(self, ti: TI | FinishedTI | None, *,
session: Session) -> TI | None:
+ """
+ Return ``ti`` as a full TaskInstance, re-fetching it when it is a
FinishedTI view.
+
+ Full instances (and ``None``) pass through unchanged. Only the
dag-callback path
+ needs the full row, and only when a dag_run reaches a terminal state,
so the
+ re-fetch costs one query per finished dag_run rather than one per loop.
+ """
+ if not isinstance(ti, FinishedTI):
+ return ti
+ return DagRun.fetch_task_instance(
+ dag_id=self.dag_id,
+ dag_run_id=self.run_id,
+ task_id=ti.task_id,
+ map_index=ti.map_index,
+ session=session,
+ )
Review Comment:
This actually checks the type and if it's a `FinishedTI`, it converts it
back to a `TaskInstance` ORM object.
It doesn't make sure that we still have a `FinishedTI` but the opposite.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
