xBis7 commented on code in PR #71737:
URL: https://github.com/apache/airflow/pull/71737#discussion_r3871061741
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -1465,10 +1512,38 @@ def _filter_tis_and_exclude_removed(dag: SerializedDAG,
tis: list[TI]) -> Iterab
else:
yield ti
- tis = list(_filter_tis_and_exclude_removed(self.get_dag(), tis))
+ def _build_finished_tis(dag: SerializedDAG, rows) -> list[FinishedTI]:
+ """Attach the serialized task, dropping (and marking REMOVED) rows
whose task is gone."""
+ finished: list[FinishedTI] = []
+ orphaned: list[str] = []
+ for row in rows:
+ try:
+ task = dag.get_task(row.task_id)
Review Comment:
I had a better look at this.
> we don't need to query the database again as it's the same task
We don't query the DB at all. We are looking it up in a cached dict.
https://github.com/xBis7/airflow/blob/edd3176d01674ecc917ec4270a9a0b5c8fa06eb8/airflow-core/src/airflow/serialization/definitions/dag.py#L264-L267
and then the `task_dict`
https://github.com/xBis7/airflow/blob/edd3176d01674ecc917ec4270a9a0b5c8fa06eb8/airflow-core/src/airflow/serialization/definitions/dag.py#L138
> We can also bulk update it on case a lot of mapped tasks are removed but I
don't think it's worth it as it does not happen a lot
This is how it works. If you look at the entire code snippet
https://github.com/xBis7/airflow/blob/edd3176d01674ecc917ec4270a9a0b5c8fa06eb8/airflow-core/src/airflow/models/dagrun.py#L1519C13-L1539C18
```python
for row in rows:
try:
task = dag.get_task(row.task_id)
except TaskNotFound:
if row.state != TaskInstanceState.REMOVED:
orphaned.append(row.task_id)
continue
finished.append(FinishedTI(**row._mapping, task=task))
if orphaned:
self.log.error("Failed to get task for finished tis %s. Marking them as
removed.", orphaned)
session.execute(
update(TI)
.where(
TI.dag_id == self.dag_id,
TI.run_id == self.run_id,
TI.task_id.in_(orphaned),
TI.state != TaskInstanceState.REMOVED,
)
.values(state=TaskInstanceState.REMOVED)
.execution_options(synchronize_session=False)
)
```
This is an iteration and for every element, we append to `orphaned` and then
do a bulk update query for all of them.
--
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]
