You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "dstandish (via GitHub)" <gi...@apache.org> on 2023/05/31 05:41:51 UTC
[GitHub] [airflow] dstandish opened a new pull request, #31628: Small cleanups / optimizations in DagRun.update_state
dstandish opened a new pull request, #31628:
URL: https://github.com/apache/airflow/pull/31628
1. no longer need getattr since mappedoperator has the attributes
2. the line `leaf_tis = list(set(leaf_tis) - set(teardown_tis))` drew my attention to the question, why weren't we using sets in the first place? So i switched to use sets. since there's a lot of `in` logic, good chance it's at least not worse, and it's a tiny bit cleaner this way.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on pull request #31628: Small cleanups / optimizations in DagRun.update_state
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on PR #31628:
URL: https://github.com/apache/airflow/pull/31628#issuecomment-1569754252
looks like you broke mypy @uranusjr :) https://github.com/apache/airflow/actions/runs/5130612005/jobs/9229712657?pr=31628#step:6:154
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a diff in pull request #31628: Small cleanups / optimizations in DagRun.update_state
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #31628:
URL: https://github.com/apache/airflow/pull/31628#discussion_r1211119030
##########
airflow/models/dagrun.py:
##########
@@ -596,19 +596,18 @@ def recalculate(self) -> _UnfinishedStates:
unfinished = unfinished.recalculate()
leaf_task_ids = {t.task_id for t in dag.leaves}
- leaf_tis = [ti for ti in tis if ti.task_id in leaf_task_ids if ti.state != TaskInstanceState.REMOVED]
- # TODO: Remove 'getattr' if setup/teardown is available for mapped task
+ leaf_tis = {ti for ti in tis if ti.task_id in leaf_task_ids if ti.state != TaskInstanceState.REMOVED}
if dag.teardowns:
# when on_failure_fail_dagrun is `False`, the final state of the DagRun
# will be computed as if the teardown task simply didn't exist.
- teardown_task_ids = [t.task_id for t in dag.teardowns]
- upstream_of_teardowns = [t.task_id for t in dag.tasks_upstream_of_teardowns]
- teardown_tis = [ti for ti in tis if ti.task_id in teardown_task_ids]
- on_failure_fail_tis = [ti for ti in teardown_tis if getattr(ti.task, "on_failure_fail_dagrun")]
- tis_upstream_of_teardowns = [ti for ti in tis if ti.task_id in upstream_of_teardowns]
- leaf_tis = list(set(leaf_tis) - set(teardown_tis))
- leaf_tis.extend(on_failure_fail_tis)
- leaf_tis.extend(tis_upstream_of_teardowns)
+ teardown_task_ids = {t.task_id for t in dag.teardowns}
+ upstream_of_teardowns = {t.task_id for t in dag.tasks_upstream_of_teardowns}
+ teardown_tis = {ti for ti in tis if ti.task_id in teardown_task_ids}
+ on_failure_fail_tis = {ti for ti in teardown_tis if ti.task.on_failure_fail_dagrun}
+ tis_upstream_of_teardowns = {ti for ti in tis if ti.task_id in upstream_of_teardowns}
+ leaf_tis -= teardown_tis
+ leaf_tis |= on_failure_fail_tis
+ leaf_tis |= tis_upstream_of_teardowns
Review Comment:
```suggestion
leaf_tis.difference_update(ti for ti in tis if ti.task_id in teardown_task_ids)
leaf_tis.update(
(ti for ti in teardown_tis if ti.task.on_failure_fail_dagrun),
(ti for ti in tis if ti.task_id in upstream_of_teardowns),
)
```
Should be even faster by avoid creating extra sets.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on a diff in pull request #31628: Small cleanups / optimizations in DagRun.update_state
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #31628:
URL: https://github.com/apache/airflow/pull/31628#discussion_r1212306753
##########
airflow/models/dagrun.py:
##########
@@ -596,19 +596,18 @@ def recalculate(self) -> _UnfinishedStates:
unfinished = unfinished.recalculate()
leaf_task_ids = {t.task_id for t in dag.leaves}
- leaf_tis = [ti for ti in tis if ti.task_id in leaf_task_ids if ti.state != TaskInstanceState.REMOVED]
- # TODO: Remove 'getattr' if setup/teardown is available for mapped task
+ leaf_tis = {ti for ti in tis if ti.task_id in leaf_task_ids if ti.state != TaskInstanceState.REMOVED}
if dag.teardowns:
# when on_failure_fail_dagrun is `False`, the final state of the DagRun
# will be computed as if the teardown task simply didn't exist.
- teardown_task_ids = [t.task_id for t in dag.teardowns]
- upstream_of_teardowns = [t.task_id for t in dag.tasks_upstream_of_teardowns]
- teardown_tis = [ti for ti in tis if ti.task_id in teardown_task_ids]
- on_failure_fail_tis = [ti for ti in teardown_tis if getattr(ti.task, "on_failure_fail_dagrun")]
- tis_upstream_of_teardowns = [ti for ti in tis if ti.task_id in upstream_of_teardowns]
- leaf_tis = list(set(leaf_tis) - set(teardown_tis))
- leaf_tis.extend(on_failure_fail_tis)
- leaf_tis.extend(tis_upstream_of_teardowns)
+ teardown_task_ids = {t.task_id for t in dag.teardowns}
+ upstream_of_teardowns = {t.task_id for t in dag.tasks_upstream_of_teardowns}
+ teardown_tis = {ti for ti in tis if ti.task_id in teardown_task_ids}
+ on_failure_fail_tis = {ti for ti in teardown_tis if ti.task.on_failure_fail_dagrun}
+ tis_upstream_of_teardowns = {ti for ti in tis if ti.task_id in upstream_of_teardowns}
+ leaf_tis -= teardown_tis
+ leaf_tis |= on_failure_fail_tis
+ leaf_tis |= tis_upstream_of_teardowns
Review Comment:
ok yeah, turns out a combination was required
`teardown_tis` was used two times, so it needed to be a set
the other two can be generator (and i did assign both of them to a variable which i think is helpful from readability perspective and should have same performance)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on a diff in pull request #31628: Small cleanups / optimizations in DagRun.update_state
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #31628:
URL: https://github.com/apache/airflow/pull/31628#discussion_r1211281469
##########
airflow/models/dagrun.py:
##########
@@ -596,19 +596,18 @@ def recalculate(self) -> _UnfinishedStates:
unfinished = unfinished.recalculate()
leaf_task_ids = {t.task_id for t in dag.leaves}
- leaf_tis = [ti for ti in tis if ti.task_id in leaf_task_ids if ti.state != TaskInstanceState.REMOVED]
- # TODO: Remove 'getattr' if setup/teardown is available for mapped task
+ leaf_tis = {ti for ti in tis if ti.task_id in leaf_task_ids if ti.state != TaskInstanceState.REMOVED}
if dag.teardowns:
# when on_failure_fail_dagrun is `False`, the final state of the DagRun
# will be computed as if the teardown task simply didn't exist.
- teardown_task_ids = [t.task_id for t in dag.teardowns]
- upstream_of_teardowns = [t.task_id for t in dag.tasks_upstream_of_teardowns]
- teardown_tis = [ti for ti in tis if ti.task_id in teardown_task_ids]
- on_failure_fail_tis = [ti for ti in teardown_tis if getattr(ti.task, "on_failure_fail_dagrun")]
- tis_upstream_of_teardowns = [ti for ti in tis if ti.task_id in upstream_of_teardowns]
- leaf_tis = list(set(leaf_tis) - set(teardown_tis))
- leaf_tis.extend(on_failure_fail_tis)
- leaf_tis.extend(tis_upstream_of_teardowns)
+ teardown_task_ids = {t.task_id for t in dag.teardowns}
+ upstream_of_teardowns = {t.task_id for t in dag.tasks_upstream_of_teardowns}
+ teardown_tis = {ti for ti in tis if ti.task_id in teardown_task_ids}
+ on_failure_fail_tis = {ti for ti in teardown_tis if ti.task.on_failure_fail_dagrun}
+ tis_upstream_of_teardowns = {ti for ti in tis if ti.task_id in upstream_of_teardowns}
+ leaf_tis -= teardown_tis
+ leaf_tis |= on_failure_fail_tis
+ leaf_tis |= tis_upstream_of_teardowns
Review Comment:
ah yes makes sense.... maybe i will keep the existing variables (for readability) but convert them to generators and use the non-operators
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on a diff in pull request #31628: Small cleanups / optimizations in DagRun.update_state
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #31628:
URL: https://github.com/apache/airflow/pull/31628#discussion_r1211284972
##########
airflow/models/dagrun.py:
##########
@@ -596,19 +596,18 @@ def recalculate(self) -> _UnfinishedStates:
unfinished = unfinished.recalculate()
leaf_task_ids = {t.task_id for t in dag.leaves}
- leaf_tis = [ti for ti in tis if ti.task_id in leaf_task_ids if ti.state != TaskInstanceState.REMOVED]
- # TODO: Remove 'getattr' if setup/teardown is available for mapped task
+ leaf_tis = {ti for ti in tis if ti.task_id in leaf_task_ids if ti.state != TaskInstanceState.REMOVED}
if dag.teardowns:
# when on_failure_fail_dagrun is `False`, the final state of the DagRun
# will be computed as if the teardown task simply didn't exist.
- teardown_task_ids = [t.task_id for t in dag.teardowns]
- upstream_of_teardowns = [t.task_id for t in dag.tasks_upstream_of_teardowns]
- teardown_tis = [ti for ti in tis if ti.task_id in teardown_task_ids]
- on_failure_fail_tis = [ti for ti in teardown_tis if getattr(ti.task, "on_failure_fail_dagrun")]
- tis_upstream_of_teardowns = [ti for ti in tis if ti.task_id in upstream_of_teardowns]
- leaf_tis = list(set(leaf_tis) - set(teardown_tis))
- leaf_tis.extend(on_failure_fail_tis)
- leaf_tis.extend(tis_upstream_of_teardowns)
+ teardown_task_ids = {t.task_id for t in dag.teardowns}
+ upstream_of_teardowns = {t.task_id for t in dag.tasks_upstream_of_teardowns}
+ teardown_tis = {ti for ti in tis if ti.task_id in teardown_task_ids}
+ on_failure_fail_tis = {ti for ti in teardown_tis if ti.task.on_failure_fail_dagrun}
+ tis_upstream_of_teardowns = {ti for ti in tis if ti.task_id in upstream_of_teardowns}
+ leaf_tis -= teardown_tis
+ leaf_tis |= on_failure_fail_tis
+ leaf_tis |= tis_upstream_of_teardowns
Review Comment:
mrrr cancel that
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on a diff in pull request #31628: Small cleanups / optimizations in DagRun.update_state
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #31628:
URL: https://github.com/apache/airflow/pull/31628#discussion_r1211189523
##########
airflow/models/dagrun.py:
##########
@@ -596,19 +596,18 @@ def recalculate(self) -> _UnfinishedStates:
unfinished = unfinished.recalculate()
leaf_task_ids = {t.task_id for t in dag.leaves}
- leaf_tis = [ti for ti in tis if ti.task_id in leaf_task_ids if ti.state != TaskInstanceState.REMOVED]
- # TODO: Remove 'getattr' if setup/teardown is available for mapped task
+ leaf_tis = {ti for ti in tis if ti.task_id in leaf_task_ids if ti.state != TaskInstanceState.REMOVED}
if dag.teardowns:
# when on_failure_fail_dagrun is `False`, the final state of the DagRun
# will be computed as if the teardown task simply didn't exist.
- teardown_task_ids = [t.task_id for t in dag.teardowns]
- upstream_of_teardowns = [t.task_id for t in dag.tasks_upstream_of_teardowns]
- teardown_tis = [ti for ti in tis if ti.task_id in teardown_task_ids]
- on_failure_fail_tis = [ti for ti in teardown_tis if getattr(ti.task, "on_failure_fail_dagrun")]
- tis_upstream_of_teardowns = [ti for ti in tis if ti.task_id in upstream_of_teardowns]
- leaf_tis = list(set(leaf_tis) - set(teardown_tis))
- leaf_tis.extend(on_failure_fail_tis)
- leaf_tis.extend(tis_upstream_of_teardowns)
+ teardown_task_ids = {t.task_id for t in dag.teardowns}
+ upstream_of_teardowns = {t.task_id for t in dag.tasks_upstream_of_teardowns}
+ teardown_tis = {ti for ti in tis if ti.task_id in teardown_task_ids}
+ on_failure_fail_tis = {ti for ti in teardown_tis if ti.task.on_failure_fail_dagrun}
+ tis_upstream_of_teardowns = {ti for ti in tis if ti.task_id in upstream_of_teardowns}
+ leaf_tis -= teardown_tis
+ leaf_tis |= on_failure_fail_tis
+ leaf_tis |= tis_upstream_of_teardowns
Review Comment:
thanks @uranusjr . i assumed that the operators did the same thing under the hood.
i looked at the docs [here](https://docs.python.org/3.8/library/stdtypes.html#set-types-set-frozenset) and that's at least what it looks like... they have -= and difference_update defined together, same with |= and update. is there some underlying implementation difference? (can't see the source code unfortunately since it looks like it's in C)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a diff in pull request #31628: Small cleanups / optimizations in DagRun.update_state
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #31628:
URL: https://github.com/apache/airflow/pull/31628#discussion_r1211267102
##########
airflow/models/dagrun.py:
##########
@@ -596,19 +596,18 @@ def recalculate(self) -> _UnfinishedStates:
unfinished = unfinished.recalculate()
leaf_task_ids = {t.task_id for t in dag.leaves}
- leaf_tis = [ti for ti in tis if ti.task_id in leaf_task_ids if ti.state != TaskInstanceState.REMOVED]
- # TODO: Remove 'getattr' if setup/teardown is available for mapped task
+ leaf_tis = {ti for ti in tis if ti.task_id in leaf_task_ids if ti.state != TaskInstanceState.REMOVED}
if dag.teardowns:
# when on_failure_fail_dagrun is `False`, the final state of the DagRun
# will be computed as if the teardown task simply didn't exist.
- teardown_task_ids = [t.task_id for t in dag.teardowns]
- upstream_of_teardowns = [t.task_id for t in dag.tasks_upstream_of_teardowns]
- teardown_tis = [ti for ti in tis if ti.task_id in teardown_task_ids]
- on_failure_fail_tis = [ti for ti in teardown_tis if getattr(ti.task, "on_failure_fail_dagrun")]
- tis_upstream_of_teardowns = [ti for ti in tis if ti.task_id in upstream_of_teardowns]
- leaf_tis = list(set(leaf_tis) - set(teardown_tis))
- leaf_tis.extend(on_failure_fail_tis)
- leaf_tis.extend(tis_upstream_of_teardowns)
+ teardown_task_ids = {t.task_id for t in dag.teardowns}
+ upstream_of_teardowns = {t.task_id for t in dag.tasks_upstream_of_teardowns}
+ teardown_tis = {ti for ti in tis if ti.task_id in teardown_task_ids}
+ on_failure_fail_tis = {ti for ti in teardown_tis if ti.task.on_failure_fail_dagrun}
+ tis_upstream_of_teardowns = {ti for ti in tis if ti.task_id in upstream_of_teardowns}
+ leaf_tis -= teardown_tis
+ leaf_tis |= on_failure_fail_tis
+ leaf_tis |= tis_upstream_of_teardowns
Review Comment:
The operators _do_ the same under the hood, the difference here is I use generators instead of sets. But the operators don’t support those, only functions do.
> Note, the non-operator versions of the [update()](https://docs.python.org/3.8/library/stdtypes.html#frozenset.update), [intersection_update()](https://docs.python.org/3.8/library/stdtypes.html#frozenset.intersection_update), [difference_update()](https://docs.python.org/3.8/library/stdtypes.html#frozenset.difference_update), and [symmetric_difference_update()](https://docs.python.org/3.8/library/stdtypes.html#frozenset.symmetric_difference_update) methods will accept any iterable as an argument.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish merged pull request #31628: Small cleanups / optimizations in DagRun.update_state
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish merged PR #31628:
URL: https://github.com/apache/airflow/pull/31628
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org