xBis7 commented on code in PR #71737:
URL: https://github.com/apache/airflow/pull/71737#discussion_r3864458639
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -1355,15 +1383,15 @@ def recalculate(self) -> _UnfinishedStates:
self.notify_dagrun_state_changed(msg="success")
if dag.has_on_success_callback:
- last_succeeded_ti: TI | None = max(
+ last_succeeded_ti: TI | FinishedTI | None = max(
Review Comment:
In all places where we accept a union `TI | FinishedTI` and there is the
same question, `since this is a FinishedTI why don't we just accept that`, if
the caller is the scheduler, then yes it will be a `FinishedTI`. But there are
other callers as well, e.g. the UI and in these cases we need an actual
`TaskInstance`.
Not to mention that enforcing all callers to use a `FinishedTI` would blow
the scope of this PR and it would also make it very hard to ensure the same
behavior as now. It would require extensive testing and even after that, there
would be some edge case with a bug.
IMO, it's not worth it because these paths aren't hot like the scheduler
loop.
Here are some examples, where we call these functions with an actual TI
outside of the scheduler loop.
1. REST API, `get_task_instance_dependencies`
line 352 it calls `get_failed_dep_statuses`
https://github.com/xBis7/airflow/blob/edd3176d01674ecc917ec4270a9a0b5c8fa06eb8/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py#L352
`get_failed_dep_statuses` calls `get_dep_statuses` for every context
while iterating
https://github.com/xBis7/airflow/blob/edd3176d01674ecc917ec4270a9a0b5c8fa06eb8/airflow-core/src/airflow/models/taskinstance.py#L1180
`_get_dep_statuses` for `TriggerRuleDep` calls `_evaluate_trigger_rule`
https://github.com/xBis7/airflow/blob/edd3176d01674ecc917ec4270a9a0b5c8fa06eb8/airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py#L117
which eventually calls `ensure_finished_tis` in multiple places and in
these tis are full ORM objects.
https://github.com/xBis7/airflow/blob/edd3176d01674ecc917ec4270a9a0b5c8fa06eb8/airflow-core/src/airflow/ti_deps/dep_context.py#L110-L129
2. CLI, `task_failed_deps`
https://github.com/xBis7/airflow/blob/edd3176d01674ecc917ec4270a9a0b5c8fa06eb8/airflow-core/src/airflow/cli/commands/task_command.py#L284
It ends down the same path as above
3. Before executing a task, `_check_and_change_state_before_execution`
These 2 sets
https://github.com/xBis7/airflow/blob/edd3176d01674ecc917ec4270a9a0b5c8fa06eb8/airflow-core/src/airflow/models/taskinstance.py#L1370-L1378
https://github.com/xBis7/airflow/blob/edd3176d01674ecc917ec4270a9a0b5c8fa06eb8/airflow-core/src/airflow/models/taskinstance.py#L1402-L1410
don't contain FinishedTIs.
--
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]
