You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "hussein-awala (via GitHub)" <gi...@apache.org> on 2023/08/20 20:26:27 UTC
[GitHub] [airflow] hussein-awala opened a new pull request, #33570: Fix waiting setup tasks when they are not a direct upstream
hussein-awala opened a new pull request, #33570:
URL: https://github.com/apache/airflow/pull/33570
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
closes: #33561
This PR force the TI to wait all the upstream setup tasks event if they are not a direct upstream. Please check #33561 for more details.
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1306720836
##########
airflow/models/dag.py:
##########
@@ -717,6 +718,16 @@ def validate_setup_teardown(self):
:meta private:
"""
for task in self.tasks:
+ if task.is_setup:
+ for down_task in task.downstream_list:
+ if not down_task.is_teardown and down_task.trigger_rule not in [
+ TriggerRule.ALL_SUCCESS,
+ TriggerRule.ONE_SUCCESS,
+ ]:
+ # this is required to ensure consistent clearing behavior when upstream
+ raise ValueError(
+ "Setup tasks must be followed with trigger rule ALL_SUCCESS or ONE_SUCCESS."
Review Comment:
```suggestion
"Setup tasks must be followed with trigger rule ALL_SUCCESS."
```
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1304620347
##########
airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -234,14 +250,19 @@ def _iter_upstream_conditions() -> Iterator[ColumnOperators]:
.group_by(TaskInstance.task_id)
).all()
upstream = sum(count for _, count in task_id_counts)
- upstream_setup = sum(c for t, c in task_id_counts if upstream_tasks[t].is_setup)
upstream_done = done >= upstream
+ setup_done = (success_setup + skipped_setup + failed_setup) >= upstream_setup
+ is_tear_down = task.is_teardown or trigger_rule == TR.ALL_DONE_SETUP_SUCCESS
Review Comment:
```suggestion
is_teardown = task.is_teardown or trigger_rule == TR.ALL_DONE_SETUP_SUCCESS
```
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1304628815
##########
airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -234,14 +250,19 @@ def _iter_upstream_conditions() -> Iterator[ColumnOperators]:
.group_by(TaskInstance.task_id)
).all()
upstream = sum(count for _, count in task_id_counts)
- upstream_setup = sum(c for t, c in task_id_counts if upstream_tasks[t].is_setup)
upstream_done = done >= upstream
+ setup_done = (success_setup + skipped_setup + failed_setup) >= upstream_setup
+ is_tear_down = task.is_teardown or trigger_rule == TR.ALL_DONE_SETUP_SUCCESS
Review Comment:
i don't think you need to check both is_teardown and trigger rule.
it should be sufficient to check the trigger rule.
and with that, it may be unnecessary to create this variable
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1309930933
##########
airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -34,10 +36,13 @@
from sqlalchemy.orm import Session
from sqlalchemy.sql.expression import ColumnOperators
+ from airflow.models.dagrun import DagRun
+ from airflow.models.operator import Operator
from airflow.models.taskinstance import TaskInstance
-class _UpstreamTIStates(NamedTuple):
+@dataclass
+class _UpstreamTIStates:
Review Comment:
Is this just to have the default? Dataclass is pretty significantly slower and not really worthwhile here since the class is unpacked pretty much immediately. Better to stick to a named tuple.
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1306476108
##########
airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -51,6 +52,7 @@ class _UpstreamTIStates(NamedTuple):
done: int
success_setup: int
skipped_setup: int
+ failed_setup: int
Review Comment:
until now, everything in this class is *direct* upstreams. is failed_setup direct only or does it include indirect too? if it includes indirect too, it should probably be clarified through a more precise variable name. but perhaps better would be to avoid mixing direct and indirect in the same class if it can be avoided. perhaps we can just add the information through an optional argument in `calculate` or something. this would also make the diff easier to deal with.
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1309949531
##########
airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -121,6 +132,13 @@ def _evaluate_trigger_rule(
from airflow.models.operator import needs_expansion
from airflow.models.taskinstance import TaskInstance
+ if ti.task.is_teardown:
+ setup_upstream_tasks = [task for task in ti.task.upstream_list if task.is_setup]
+ else:
+ setup_upstream_tasks = list(ti.task.get_upstreams_only_setups())
Review Comment:
```suggestion
if ti.task.is_teardown:
setup_upstream_tasks = (task for task in ti.task.upstream_list if task.is_setup)
else:
setup_upstream_tasks = ti.task.get_upstreams_only_setups()
```
No need to build a list here as far as I can tell
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1306720790
##########
airflow/models/dag.py:
##########
@@ -717,6 +718,16 @@ def validate_setup_teardown(self):
:meta private:
"""
for task in self.tasks:
+ if task.is_setup:
+ for down_task in task.downstream_list:
+ if not down_task.is_teardown and down_task.trigger_rule not in [
+ TriggerRule.ALL_SUCCESS,
+ TriggerRule.ONE_SUCCESS,
Review Comment:
```suggestion
```
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1306472441
##########
tests/models/test_taskinstance.py:
##########
@@ -1270,20 +1338,142 @@ def test_depends_on_past(self, dag_maker):
param(
"all_done_setup_success",
1,
- _UpstreamTIStates(3, 1, 0, 0, 0, 4, 1, 0),
+ 0,
+ _UpstreamTIStates(3, 1, 0, 0, 0, 4, 1, 0, 0),
True,
None,
False,
id="not all done, one skipped",
),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(7, 0, 0, 0, 0, 7, 2, 0, 0),
+ True,
+ None,
+ True,
+ id="indirect upstream setups - all success",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(7, 0, 0, 0, 0, 7, 1, 0, 0),
+ True,
+ None,
+ False,
+ id="indirect upstream setups - one not done and one success",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(7, 0, 0, 0, 0, 7, 0, 1, 0),
+ True,
+ None,
+ False,
+ id="indirect upstream setups - one not done and one skipped",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(7, 0, 0, 0, 0, 7, 1, 1, 0),
+ True,
+ TaskInstanceState.SKIPPED,
+ True,
+ id="indirect upstream setups - all done: one skipped and one success",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(7, 0, 0, 0, 0, 7, 1, 0, 1),
+ True,
+ TaskInstanceState.UPSTREAM_FAILED,
+ False,
+ id="indirect upstream setups - all done: one success and one failed",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(7, 0, 0, 0, 0, 7, 0, 1, 1),
+ True,
+ TaskInstanceState.UPSTREAM_FAILED,
+ False,
+ id="indirect upstream setups - all done: one skipped and one failed",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(7, 0, 0, 0, 0, 7, 0, 0, 1),
+ True,
+ TaskInstanceState.UPSTREAM_FAILED,
+ False,
+ id="indirect upstream setups - one not done and one failed",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(0, 0, 1, 0, 0, 1, 2, 0, 0),
+ True,
+ TaskInstanceState.UPSTREAM_FAILED,
+ False,
+ id="indirect upstream setups - all setup success but upstream failed",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(5, 0, 0, 0, 0, 5, 2, 0, 0),
+ True,
+ None,
+ False,
+ id="indirect upstream setups - all setup success but upstream still running",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(5, 0, 0, 0, 0, 5, 0, 0, 1),
+ True,
+ TaskInstanceState.UPSTREAM_FAILED,
+ False,
+ id="indirect upstream setups - one setup failed but upstream still running",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(4, 0, 1, 0, 0, 5, 0, 2, 0),
+ True,
+ TaskInstanceState.UPSTREAM_FAILED,
+ False,
+ id="indirect upstream setups - all setup skipped but upstream failed",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(0, 7, 0, 0, 0, 7, 0, 0, 1),
+ True,
+ TaskInstanceState.UPSTREAM_FAILED,
+ False,
+ id="indirect upstream setups - one setup failed but upstream skipped",
+ ),
],
)
def test_check_task_dependencies(
Review Comment:
i think that some of these tests, where you've made substantial changes, i think we should just make a new test.
it's very important that we're confident that we're not breaking anything here, so rather than make such significant changes to the test, i think it's preferable to add a new one alongside. i'll try to do this right now.
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1306474630
##########
tests/models/test_taskinstance.py:
##########
@@ -1103,59 +1103,116 @@ def test_depends_on_past(self, dag_maker):
# Numeric fields are in order:
# successes, skipped, failed, upstream_failed, removed, done
@pytest.mark.parametrize(
- "trigger_rule, upstream_setups, upstream_states, flag_upstream_failed, expect_state, expect_passed",
+ "trigger_rule, direct_upstream_setups, indirect_upstream_setups, upstream_states,"
Review Comment:
there are so many changes to tests it makes it a bit hard to review.
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on PR #33570:
URL: https://github.com/apache/airflow/pull/33570#issuecomment-1692446901
Cool, so one thing @hussein-awala ... I think in order to ensure consistent behavior, we must also add something like this to function `validate_setup_teardown`
```
if task.is_setup:
for down_task in task.downstream_list:
if not down_task.is_teardown and down_task.trigger_rule not in one_success_rules:
# this is required to ensure consistent clearing behavior when upstream
raise ValueError(
"Setup tasks must be followed with trigger rule ALL_SUCCESS or ONE_SUCCESS."
)
```
or perhaps better, do the check when setting upstream / downstream. The reason is that, what you are doing (and needfully so) is essentially to check that upstream setups are done and successful before the downstream (which is not directly connected) may run. BUT, if a user uses a trigger rule such as `one_failed` following a setup task, then this can create odd results where, the scenario where the task runs is when the setup fails, but we're clearing it and running it, and the setup succeeds, so it shouldn't run. It's very odd and nonsensical, but we should guard against it and for consistency. If it's not clear I can try and create an example. We could perhaps relax this a bit to allow other trigger rules for work tasks not inside the scope of the setup.
Other point... on the topic of scope of the setup.... I think your logic for `get_setups_only` may be incorrect because it gets, i think, all upstream setups, but note that the scope of the setup can be constrained by adding a teardown, which means that anything that is not between the setup and its teardown is not assumed to require that setup. i think the easiest way for you to obtain just the "relevant" setups is to just use `get_upstreams_only_setups_and_teardowns` and filter out the teardowns. This method restricts to just setups and teardowns for which the object task is "in scope".
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on PR #33570:
URL: https://github.com/apache/airflow/pull/33570#issuecomment-1692109417
@hussein-awala i noticed there are some failing tests. is this ready for review or are you still working through some edge cases?
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1306476712
##########
airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -94,33 +98,41 @@ def _get_dep_statuses(
session: Session,
dep_context: DepContext,
) -> Iterator[TIDepStatus]:
+ # get all the setup tasks upstream of this task
+ setup_upstream_tasks = list(ti.task.get_upstreams_only_setups())
# Checking that all upstream dependencies have succeeded.
if not ti.task.upstream_task_ids:
yield self._passing_status(reason="The task instance did not have any upstream tasks.")
return
- if ti.task.trigger_rule == TR.ALWAYS:
+ if ti.task.trigger_rule == TR.ALWAYS and not setup_upstream_tasks:
+ # even with ALWAYS trigger rule, we still need to check setup tasks
Review Comment:
i think always should probably keep the same behavior. it says in the docs `No dependencies at all, run this task at any time`. it used to be called "dummy" trigger rule. so i think we just keep the absolute short circuit here.
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1306476962
##########
airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -94,33 +98,41 @@ def _get_dep_statuses(
session: Session,
dep_context: DepContext,
) -> Iterator[TIDepStatus]:
+ # get all the setup tasks upstream of this task
+ setup_upstream_tasks = list(ti.task.get_upstreams_only_setups())
# Checking that all upstream dependencies have succeeded.
if not ti.task.upstream_task_ids:
yield self._passing_status(reason="The task instance did not have any upstream tasks.")
return
- if ti.task.trigger_rule == TR.ALWAYS:
+ if ti.task.trigger_rule == TR.ALWAYS and not setup_upstream_tasks:
+ # even with ALWAYS trigger rule, we still need to check setup tasks
yield self._passing_status(reason="The task had a always trigger rule set.")
return
- yield from self._evaluate_trigger_rule(ti=ti, dep_context=dep_context, session=session)
+ yield from self._evaluate_trigger_rule(
Review Comment:
given that there's no need to evaluate whether there are upstream setups for the purpose of TR.ALWAYS, i think we no longer need this new param `setup_upstream_tasks` in `_evaluate_trigger_rule`
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on PR #33570:
URL: https://github.com/apache/airflow/pull/33570#issuecomment-1694385121
i'm gonna collaborate on this with you if that's alright since time is short.
i will rebase and add a couple changes which of course you can revert if you have concerns. i'l try to make it obvious what each commit does.
--
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] hussein-awala commented on pull request #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on PR #33570:
URL: https://github.com/apache/airflow/pull/33570#issuecomment-1712778174
Fixed by an alternative solution by #33903
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1306477061
##########
airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -94,33 +98,41 @@ def _get_dep_statuses(
session: Session,
dep_context: DepContext,
) -> Iterator[TIDepStatus]:
+ # get all the setup tasks upstream of this task
+ setup_upstream_tasks = list(ti.task.get_upstreams_only_setups())
# Checking that all upstream dependencies have succeeded.
if not ti.task.upstream_task_ids:
yield self._passing_status(reason="The task instance did not have any upstream tasks.")
return
- if ti.task.trigger_rule == TR.ALWAYS:
+ if ti.task.trigger_rule == TR.ALWAYS and not setup_upstream_tasks:
+ # even with ALWAYS trigger rule, we still need to check setup tasks
Review Comment:
it's really a non-sensical rule. with it, the deps are just for show. no one should ever really use this trigger rule. perhaps it's there for testing 🤷
##########
airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -94,33 +98,41 @@ def _get_dep_statuses(
session: Session,
dep_context: DepContext,
) -> Iterator[TIDepStatus]:
+ # get all the setup tasks upstream of this task
+ setup_upstream_tasks = list(ti.task.get_upstreams_only_setups())
# Checking that all upstream dependencies have succeeded.
if not ti.task.upstream_task_ids:
yield self._passing_status(reason="The task instance did not have any upstream tasks.")
return
- if ti.task.trigger_rule == TR.ALWAYS:
+ if ti.task.trigger_rule == TR.ALWAYS and not setup_upstream_tasks:
+ # even with ALWAYS trigger rule, we still need to check setup tasks
Review Comment:
it's really a non-sensical rule. with it, the deps are just for show. no one should ever really use this trigger rule. perhaps it's there for testing 🤷
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1306476226
##########
airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -51,6 +52,7 @@ class _UpstreamTIStates(NamedTuple):
done: int
success_setup: int
skipped_setup: int
+ failed_setup: int
Review Comment:
the other issue is, there are more states than just "failed" that we need to account for. there is also upstream_failed, skipped, etc -- essentially anything other than success.
--
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] hussein-awala commented on a diff in pull request #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1306480924
##########
tests/models/test_taskinstance.py:
##########
@@ -1270,20 +1338,142 @@ def test_depends_on_past(self, dag_maker):
param(
"all_done_setup_success",
1,
- _UpstreamTIStates(3, 1, 0, 0, 0, 4, 1, 0),
+ 0,
+ _UpstreamTIStates(3, 1, 0, 0, 0, 4, 1, 0, 0),
True,
None,
False,
id="not all done, one skipped",
),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(7, 0, 0, 0, 0, 7, 2, 0, 0),
+ True,
+ None,
+ True,
+ id="indirect upstream setups - all success",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(7, 0, 0, 0, 0, 7, 1, 0, 0),
+ True,
+ None,
+ False,
+ id="indirect upstream setups - one not done and one success",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(7, 0, 0, 0, 0, 7, 0, 1, 0),
+ True,
+ None,
+ False,
+ id="indirect upstream setups - one not done and one skipped",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(7, 0, 0, 0, 0, 7, 1, 1, 0),
+ True,
+ TaskInstanceState.SKIPPED,
+ True,
+ id="indirect upstream setups - all done: one skipped and one success",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(7, 0, 0, 0, 0, 7, 1, 0, 1),
+ True,
+ TaskInstanceState.UPSTREAM_FAILED,
+ False,
+ id="indirect upstream setups - all done: one success and one failed",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(7, 0, 0, 0, 0, 7, 0, 1, 1),
+ True,
+ TaskInstanceState.UPSTREAM_FAILED,
+ False,
+ id="indirect upstream setups - all done: one skipped and one failed",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(7, 0, 0, 0, 0, 7, 0, 0, 1),
+ True,
+ TaskInstanceState.UPSTREAM_FAILED,
+ False,
+ id="indirect upstream setups - one not done and one failed",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(0, 0, 1, 0, 0, 1, 2, 0, 0),
+ True,
+ TaskInstanceState.UPSTREAM_FAILED,
+ False,
+ id="indirect upstream setups - all setup success but upstream failed",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(5, 0, 0, 0, 0, 5, 2, 0, 0),
+ True,
+ None,
+ False,
+ id="indirect upstream setups - all setup success but upstream still running",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(5, 0, 0, 0, 0, 5, 0, 0, 1),
+ True,
+ TaskInstanceState.UPSTREAM_FAILED,
+ False,
+ id="indirect upstream setups - one setup failed but upstream still running",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(4, 0, 1, 0, 0, 5, 0, 2, 0),
+ True,
+ TaskInstanceState.UPSTREAM_FAILED,
+ False,
+ id="indirect upstream setups - all setup skipped but upstream failed",
+ ),
+ param(
+ "all_success",
+ 0,
+ 2,
+ _UpstreamTIStates(0, 7, 0, 0, 0, 7, 0, 0, 1),
+ True,
+ TaskInstanceState.UPSTREAM_FAILED,
+ False,
+ id="indirect upstream setups - one setup failed but upstream skipped",
+ ),
],
)
def test_check_task_dependencies(
Review Comment:
It was my first thought, but I tried to avoid duplicating the code. But yeah +1
--
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] hussein-awala closed pull request #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala closed pull request #33570: Fix waiting setup tasks when they are not a direct upstream
URL: https://github.com/apache/airflow/pull/33570
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1309953671
##########
airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -234,59 +296,72 @@ def _iter_upstream_conditions() -> Iterator[ColumnOperators]:
.group_by(TaskInstance.task_id)
).all()
upstream = sum(count for _, count in task_id_counts)
- upstream_setup = sum(c for t, c in task_id_counts if upstream_tasks[t].is_setup)
+ if ti.task.is_teardown:
+ upstream_setup = sum(c for t, c in task_id_counts if upstream_tasks[t].is_setup)
upstream_done = done >= upstream
+ setup_done = (success_setup + skipped_setup + failed_setup) >= upstream_setup
changed = False
new_state = None
if dep_context.flag_upstream_failed:
- if trigger_rule == TR.ALL_SUCCESS:
- if upstream_failed or failed:
- new_state = TaskInstanceState.UPSTREAM_FAILED
- elif skipped:
- new_state = TaskInstanceState.SKIPPED
- elif removed and success and ti.map_index > -1:
- if ti.map_index >= success:
- new_state = TaskInstanceState.REMOVED
- elif trigger_rule == TR.ALL_FAILED:
- if success or skipped:
- new_state = TaskInstanceState.SKIPPED
- elif trigger_rule == TR.ONE_SUCCESS:
- if upstream_done and done == skipped:
- # if upstream is done and all are skipped mark as skipped
- new_state = TaskInstanceState.SKIPPED
- elif upstream_done and success <= 0:
- # if upstream is done and there are no success mark as upstream failed
- new_state = TaskInstanceState.UPSTREAM_FAILED
- elif trigger_rule == TR.ONE_FAILED:
- if upstream_done and not (failed or upstream_failed):
- new_state = TaskInstanceState.SKIPPED
- elif trigger_rule == TR.ONE_DONE:
- if upstream_done and not (failed or success):
- new_state = TaskInstanceState.SKIPPED
- elif trigger_rule == TR.NONE_FAILED:
- if upstream_failed or failed:
- new_state = TaskInstanceState.UPSTREAM_FAILED
- elif trigger_rule == TR.NONE_FAILED_MIN_ONE_SUCCESS:
- if upstream_failed or failed:
- new_state = TaskInstanceState.UPSTREAM_FAILED
- elif skipped == upstream:
- new_state = TaskInstanceState.SKIPPED
- elif trigger_rule == TR.NONE_SKIPPED:
- if skipped:
- new_state = TaskInstanceState.SKIPPED
- elif trigger_rule == TR.ALL_SKIPPED:
- if success or failed:
- new_state = TaskInstanceState.SKIPPED
- elif trigger_rule == TR.ALL_DONE_SETUP_SUCCESS:
- if upstream_done and upstream_setup and skipped_setup >= upstream_setup:
- # when there is an upstream setup and they have all skipped, then skip
- new_state = TaskInstanceState.SKIPPED
- elif upstream_done and upstream_setup and success_setup == 0:
- # when there is an upstream setup, if none succeeded, mark upstream failed
- # if at least one setup ran, we'll let it run
- new_state = TaskInstanceState.UPSTREAM_FAILED
+ if not task.is_teardown and failed_setup:
+ # we should exclude the teardown tasks from this check,
+ # because they should be run even if there is only one success setup task
+ new_state = TaskInstanceState.UPSTREAM_FAILED
+ elif not task.is_teardown and upstream_setup and setup_done and skipped_setup > 0:
+ # when there are upstream setup tasks and at least one of them is skipped, then skip
+ new_state = TaskInstanceState.SKIPPED
+ elif not upstream_setup or setup_done:
+ # if there are no upstream setup tasks or all of them are done,
+ # and we haven't set a new state, then we can check the upstream tasks
Review Comment:
I’d do
```python
elif upstream_setup and not setup_down:
pass
```
and dedent the entire block below. (Not sure if I got the boolean negation right)
##########
airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -234,59 +296,72 @@ def _iter_upstream_conditions() -> Iterator[ColumnOperators]:
.group_by(TaskInstance.task_id)
).all()
upstream = sum(count for _, count in task_id_counts)
- upstream_setup = sum(c for t, c in task_id_counts if upstream_tasks[t].is_setup)
+ if ti.task.is_teardown:
+ upstream_setup = sum(c for t, c in task_id_counts if upstream_tasks[t].is_setup)
upstream_done = done >= upstream
+ setup_done = (success_setup + skipped_setup + failed_setup) >= upstream_setup
changed = False
new_state = None
if dep_context.flag_upstream_failed:
- if trigger_rule == TR.ALL_SUCCESS:
- if upstream_failed or failed:
- new_state = TaskInstanceState.UPSTREAM_FAILED
- elif skipped:
- new_state = TaskInstanceState.SKIPPED
- elif removed and success and ti.map_index > -1:
- if ti.map_index >= success:
- new_state = TaskInstanceState.REMOVED
- elif trigger_rule == TR.ALL_FAILED:
- if success or skipped:
- new_state = TaskInstanceState.SKIPPED
- elif trigger_rule == TR.ONE_SUCCESS:
- if upstream_done and done == skipped:
- # if upstream is done and all are skipped mark as skipped
- new_state = TaskInstanceState.SKIPPED
- elif upstream_done and success <= 0:
- # if upstream is done and there are no success mark as upstream failed
- new_state = TaskInstanceState.UPSTREAM_FAILED
- elif trigger_rule == TR.ONE_FAILED:
- if upstream_done and not (failed or upstream_failed):
- new_state = TaskInstanceState.SKIPPED
- elif trigger_rule == TR.ONE_DONE:
- if upstream_done and not (failed or success):
- new_state = TaskInstanceState.SKIPPED
- elif trigger_rule == TR.NONE_FAILED:
- if upstream_failed or failed:
- new_state = TaskInstanceState.UPSTREAM_FAILED
- elif trigger_rule == TR.NONE_FAILED_MIN_ONE_SUCCESS:
- if upstream_failed or failed:
- new_state = TaskInstanceState.UPSTREAM_FAILED
- elif skipped == upstream:
- new_state = TaskInstanceState.SKIPPED
- elif trigger_rule == TR.NONE_SKIPPED:
- if skipped:
- new_state = TaskInstanceState.SKIPPED
- elif trigger_rule == TR.ALL_SKIPPED:
- if success or failed:
- new_state = TaskInstanceState.SKIPPED
- elif trigger_rule == TR.ALL_DONE_SETUP_SUCCESS:
- if upstream_done and upstream_setup and skipped_setup >= upstream_setup:
- # when there is an upstream setup and they have all skipped, then skip
- new_state = TaskInstanceState.SKIPPED
- elif upstream_done and upstream_setup and success_setup == 0:
- # when there is an upstream setup, if none succeeded, mark upstream failed
- # if at least one setup ran, we'll let it run
- new_state = TaskInstanceState.UPSTREAM_FAILED
+ if not task.is_teardown and failed_setup:
+ # we should exclude the teardown tasks from this check,
+ # because they should be run even if there is only one success setup task
+ new_state = TaskInstanceState.UPSTREAM_FAILED
+ elif not task.is_teardown and upstream_setup and setup_done and skipped_setup > 0:
+ # when there are upstream setup tasks and at least one of them is skipped, then skip
+ new_state = TaskInstanceState.SKIPPED
+ elif not upstream_setup or setup_done:
+ # if there are no upstream setup tasks or all of them are done,
+ # and we haven't set a new state, then we can check the upstream tasks
Review Comment:
I’d do
```python
elif upstream_setup and not setup_down:
pass
```
and dedent the entire block below. (Not sure if I got the boolean negation right)
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1304620347
##########
airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -234,14 +250,19 @@ def _iter_upstream_conditions() -> Iterator[ColumnOperators]:
.group_by(TaskInstance.task_id)
).all()
upstream = sum(count for _, count in task_id_counts)
- upstream_setup = sum(c for t, c in task_id_counts if upstream_tasks[t].is_setup)
upstream_done = done >= upstream
+ setup_done = (success_setup + skipped_setup + failed_setup) >= upstream_setup
+ is_tear_down = task.is_teardown or trigger_rule == TR.ALL_DONE_SETUP_SUCCESS
Review Comment:
```suggestion
is_teardown = task.is_teardown or trigger_rule == TR.ALL_DONE_SETUP_SUCCESS
```
##########
airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -234,14 +250,19 @@ def _iter_upstream_conditions() -> Iterator[ColumnOperators]:
.group_by(TaskInstance.task_id)
).all()
upstream = sum(count for _, count in task_id_counts)
- upstream_setup = sum(c for t, c in task_id_counts if upstream_tasks[t].is_setup)
upstream_done = done >= upstream
+ setup_done = (success_setup + skipped_setup + failed_setup) >= upstream_setup
+ is_tear_down = task.is_teardown or trigger_rule == TR.ALL_DONE_SETUP_SUCCESS
changed = False
new_state = None
if dep_context.flag_upstream_failed:
- if trigger_rule == TR.ALL_SUCCESS:
+ if not is_tear_down and failed_setup:
Review Comment:
```suggestion
if not is_teardown and failed_setup:
```
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on PR #33570:
URL: https://github.com/apache/airflow/pull/33570#issuecomment-1692073049
good catch. bummer we missed this one. will take a look here.
--
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] hussein-awala commented on a diff in pull request #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1306478613
##########
airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -51,6 +52,7 @@ class _UpstreamTIStates(NamedTuple):
done: int
success_setup: int
skipped_setup: int
+ failed_setup: int
Review Comment:
For me there is 4 different cases:
- running (all the unfinished states) -> we should wait
- success -> we can run the task
- failed (failed or upstream_failed) -> fail
- skipped with any fail -> we should skipp
So failed here includes failed and upstream_failed:
```python
return _UpstreamTIStates(
success=counter.get(TaskInstanceState.SUCCESS, 0),
skipped=counter.get(TaskInstanceState.SKIPPED, 0),
failed=counter.get(TaskInstanceState.FAILED, 0),
upstream_failed=counter.get(TaskInstanceState.UPSTREAM_FAILED, 0),
removed=counter.get(TaskInstanceState.REMOVED, 0),
done=sum(counter.values()),
success_setup=setup_counter.get(TaskInstanceState.SUCCESS, 0),
skipped_setup=setup_counter.get(TaskInstanceState.SKIPPED, 0),
failed_setup=setup_counter.get(TaskInstanceState.FAILED, 0)
+ setup_counter.get(TaskInstanceState.UPSTREAM_FAILED, 0),
)
```
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1304619901
##########
airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -234,14 +250,19 @@ def _iter_upstream_conditions() -> Iterator[ColumnOperators]:
.group_by(TaskInstance.task_id)
).all()
upstream = sum(count for _, count in task_id_counts)
- upstream_setup = sum(c for t, c in task_id_counts if upstream_tasks[t].is_setup)
upstream_done = done >= upstream
+ setup_done = (success_setup + skipped_setup + failed_setup) >= upstream_setup
+ is_tear_down = task.is_teardown or trigger_rule == TR.ALL_DONE_SETUP_SUCCESS
changed = False
new_state = None
if dep_context.flag_upstream_failed:
- if trigger_rule == TR.ALL_SUCCESS:
+ if not is_tear_down and failed_setup:
Review Comment:
```suggestion
if not is_teardown and failed_setup:
```
--
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] hussein-awala commented on pull request #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on PR #33570:
URL: https://github.com/apache/airflow/pull/33570#issuecomment-1692391115
> @hussein-awala i noticed there are some failing tests. is this ready for review or are you still working through some edge cases?
It was supposed to be ready (I tested it and it worked perfectly as expected), but there seems to be a problem with the mapped operators (which I didn't test), and I didn't have time to check it. I'll resume working on it this weekend, and I'll draft it in the meantime.
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1306476380
##########
airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -94,33 +98,41 @@ def _get_dep_statuses(
session: Session,
dep_context: DepContext,
) -> Iterator[TIDepStatus]:
+ # get all the setup tasks upstream of this task
+ setup_upstream_tasks = list(ti.task.get_upstreams_only_setups())
# Checking that all upstream dependencies have succeeded.
if not ti.task.upstream_task_ids:
yield self._passing_status(reason="The task instance did not have any upstream tasks.")
return
- if ti.task.trigger_rule == TR.ALWAYS:
+ if ti.task.trigger_rule == TR.ALWAYS and not setup_upstream_tasks:
+ # even with ALWAYS trigger rule, we still need to check setup tasks
yield self._passing_status(reason="The task had a always trigger rule set.")
return
- yield from self._evaluate_trigger_rule(ti=ti, dep_context=dep_context, session=session)
+ yield from self._evaluate_trigger_rule(
+ ti=ti, dep_context=dep_context, setup_upstream_tasks=setup_upstream_tasks, session=session
+ )
def _evaluate_trigger_rule(
self,
*,
ti: TaskInstance,
dep_context: DepContext,
+ setup_upstream_tasks: list[Operator] | None = None,
session: Session,
) -> Iterator[TIDepStatus]:
"""Evaluate whether ``ti``'s trigger rule was met.
:param ti: Task instance to evaluate the trigger rule of.
:param dep_context: The current dependency context.
+ setup_upstream_tasks: The setup tasks upstream of the current task.
Review Comment:
should probably clarify that it includes indirect
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on PR #33570:
URL: https://github.com/apache/airflow/pull/33570#issuecomment-1694392987
ok yeah that worked... ok so yeah, pushed a few changes, mainly the only changes i made were with respect to making the diff easier to follow. added a few comments. will continue to review.
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on PR #33570:
URL: https://github.com/apache/airflow/pull/33570#issuecomment-1694385376
oh darn cannot rebase let's see if i can just add
--
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] hussein-awala commented on a diff in pull request #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1306477735
##########
airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -51,6 +52,7 @@ class _UpstreamTIStates(NamedTuple):
done: int
success_setup: int
skipped_setup: int
+ failed_setup: int
Review Comment:
> is failed_setup direct only or does it include indirect too?
yes it includes indirect too. The same for success_setup and skipped_setup
--
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 #33570: Fix waiting setup tasks when they are not a direct upstream
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #33570:
URL: https://github.com/apache/airflow/pull/33570#discussion_r1309928233
##########
airflow/models/dag.py:
##########
@@ -717,6 +718,11 @@ def validate_setup_teardown(self):
:meta private:
"""
for task in self.tasks:
+ if task.is_setup:
+ for down_task in task.downstream_list:
+ if not down_task.is_teardown and down_task.trigger_rule != TriggerRule.ALL_SUCCESS:
+ # this is required to ensure consistent clearing behavior when upstream
+ raise ValueError("Setup tasks must be followed with trigger rule ALL_SUCCESS.")
Review Comment:
Since we’ve drilled this deep, can this show what the offending task is for clarity?
##########
airflow/models/dag.py:
##########
@@ -717,6 +718,11 @@ def validate_setup_teardown(self):
:meta private:
"""
for task in self.tasks:
+ if task.is_setup:
+ for down_task in task.downstream_list:
+ if not down_task.is_teardown and down_task.trigger_rule != TriggerRule.ALL_SUCCESS:
+ # this is required to ensure consistent clearing behavior when upstream
+ raise ValueError("Setup tasks must be followed with trigger rule ALL_SUCCESS.")
Review Comment:
Since we’ve drilled this deep, can this show what the offending task is for clarity?
--
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