Fix short circuit in mapped tasks #44925
Conversation
shahar1
added
type:bug-fix
| @@ -49,46 +49,45 @@ def _get_dep_statuses(self, ti, session, dep_context): | |||
| finished_task_ids = {t.task_id for t in finished_tis} | |||
|
|
|||
| for parent in upstream: | |||
| if isinstance(parent, SkipMixin): | |||
There was a problem hiding this comment.
Initially I though of making it:
if isinstance(parent, (SkipMixin, MappedOpeartor))
But then I saw that tests pass without this if - so I decided going with Occam's razor and simply remove it.
| op.downstream_task_ids = self.downstream_task_ids | ||
| op.upstream_task_ids = self.upstream_task_ids |
There was a problem hiding this comment.
@uranusjr following up #43883 (comment) - it seems that these attributes were missing here, so that's why PythonOperator printed misleading logs
| # The following could be applied only for non-mapped tasks | ||
| if map_index == -1: | ||
| SkipMixin._set_state_to_skipped(dag_run, task_ids_list, session) | ||
| session.commit() |
There was a problem hiding this comment.
This block could theoretically be removed, and then setting the skipped states will be done exclusively by NotPreviouslySkippedDep for both mapped and non-mapped.
Not sure regarding effects on performance, so I left it as-is for now (if we decide to remove it - some other tests will have to be adjusted).
shahar1
force-pushed
the
fix-dtm-short-circuit-main
branch
from
b13c1dc to
9d0aee7
Compare
shahar1
force-pushed
the
fix-dtm-short-circuit-main
branch
4 times, most recently
from
dd54028 to
f0ad8dc
Compare
shahar1
force-pushed
the
fix-dtm-short-circuit-main
branch
from
f0ad8dc to
a2e62b3
Compare
closes: #43883
^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an 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 a newsfragment file, named
{pr_number}.significant.rstor{issue_number}.significant.rst, in newsfragments.