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/07/12 00:39:47 UTC
[GitHub] [airflow] dstandish opened a new pull request, #32538: Fix short circuit operator re teardowns
dstandish opened a new pull request, #32538:
URL: https://github.com/apache/airflow/pull/32538
Short circuit operator should by default not skip teardowns for which the object task is in scope.
So for example if you have `s1 >> w1 >> w2 >> t1` and t1 is teardown for s1, then if w1 is short circuit, w2 should be skipped but not t1.
A more interesting scenario is
```
s1 >> op1 >> s2 >> op2 >> [t1, t2]
```
or
```
s1 >> op1 >> s2 >> op2 >> t2 >> t1
```
In both of these cases, if op1 is short circuit (and s1 >> t1, s2 >> t2), then tasks `s2, op2, t2` should all be skipped but t1 should not!
--
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 #32538: Fix short circuit operator re teardowns
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1260454176
##########
airflow/operators/python.py:
##########
@@ -251,24 +252,47 @@ def execute(self, context: Context) -> Any:
self.log.info("Proceeding with downstream tasks...")
return condition
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
+ scope_teardowns = {x.task_id for x in self.get_upstreams_only_setups_and_teardowns() if x.is_teardown}
+
+ def is_relevant_teardown(other):
+ if other.task_id in scope_teardowns:
+ return True
+ elif other.is_teardown and not any(x.is_setup for x in other.upstream_list):
+ return True
+ else:
+ return False
+
+ downstream_tasks = [
+ x for x in context["task"].get_flat_relatives(upstream=False) if not is_relevant_teardown(x)
+ ]
self.log.debug("Downstream task IDs %s", downstream_tasks)
if downstream_tasks:
dag_run = context["dag_run"]
execution_date = dag_run.execution_date
-
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
if self.ignore_downstream_trigger_rules is True:
self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ self.skip(
+ dag_run=dag_run,
+ execution_date=execution_date,
+ tasks=downstream_tasks,
+ map_index=context["ti"].map_index,
+ )
else:
self.log.info("Skipping downstream tasks while respecting trigger rules...")
# Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
# Scheduler handle the remaining downstream task(s) appropriately.
+ to_skip = [
+ x
+ for x in context["task"].get_direct_relatives(upstream=False)
+ if not is_relevant_teardown(x)
+ ]
Review Comment:
thanks @uranusjr ... this needs a bit more work found another edge case
--
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 #32538: Fix short circuit operator re teardowns
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1260567257
##########
airflow/operators/python.py:
##########
@@ -250,25 +251,35 @@ def execute(self, context: Context) -> Any:
if condition:
self.log.info("Proceeding with downstream tasks...")
return condition
-
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
+ downstream_tasks = [
+ x for x in context["task"].get_flat_relatives(upstream=False) if not x.is_teardown
+ ]
self.log.debug("Downstream task IDs %s", downstream_tasks)
if downstream_tasks:
dag_run = context["dag_run"]
execution_date = dag_run.execution_date
-
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
if self.ignore_downstream_trigger_rules is True:
self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ self.skip(
+ dag_run=dag_run,
+ execution_date=execution_date,
+ tasks=downstream_tasks,
+ map_index=context["ti"].map_index,
+ )
else:
self.log.info("Skipping downstream tasks while respecting trigger rules...")
# Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
# Scheduler handle the remaining downstream task(s) appropriately.
+ to_skip = [
+ x for x in context["task"].get_direct_relatives(upstream=False) if not x.is_teardown
+ ]
Review Comment:
OK @uranusjr i went with that optimization... PTAL
--
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 #32538: Fix short circuit operator re teardowns
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1260559655
##########
airflow/operators/python.py:
##########
@@ -250,25 +251,35 @@ def execute(self, context: Context) -> Any:
if condition:
self.log.info("Proceeding with downstream tasks...")
return condition
-
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
+ downstream_tasks = [
+ x for x in context["task"].get_flat_relatives(upstream=False) if not x.is_teardown
+ ]
self.log.debug("Downstream task IDs %s", downstream_tasks)
if downstream_tasks:
dag_run = context["dag_run"]
execution_date = dag_run.execution_date
-
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
if self.ignore_downstream_trigger_rules is True:
self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ self.skip(
+ dag_run=dag_run,
+ execution_date=execution_date,
+ tasks=downstream_tasks,
+ map_index=context["ti"].map_index,
+ )
else:
self.log.info("Skipping downstream tasks while respecting trigger rules...")
# Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
# Scheduler handle the remaining downstream task(s) appropriately.
+ to_skip = [
+ x for x in context["task"].get_direct_relatives(upstream=False) if not x.is_teardown
+ ]
Review Comment:
when you say not that trivial isn't it just doing `dag.task_dict[x] for x in t.upstream_task_ids` which is not much different?
--
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 #32538: Fix short circuit operator re teardowns
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1261585319
##########
airflow/operators/python.py:
##########
@@ -251,27 +253,37 @@ def execute(self, context: Context) -> Any:
self.log.info("Proceeding with downstream tasks...")
return condition
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
- self.log.debug("Downstream task IDs %s", downstream_tasks)
+ if not self.downstream_task_ids:
+ self.log.info("No downstream tasks; nothing to do.")
+ return
- if downstream_tasks:
- dag_run = context["dag_run"]
- execution_date = dag_run.execution_date
+ dag_run = context["dag_run"]
+ execution_date = dag_run.execution_date
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
Review Comment:
Logical date is typed as `datetime` too it seems.
And worse, it does not exist on DagRunPydantic, so mypy complains about that. i'm just keeping the insert
--
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 #32538: Fix short circuit operator re teardowns
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1260440104
##########
airflow/operators/python.py:
##########
@@ -251,24 +252,47 @@ def execute(self, context: Context) -> Any:
self.log.info("Proceeding with downstream tasks...")
return condition
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
+ scope_teardowns = {x.task_id for x in self.get_upstreams_only_setups_and_teardowns() if x.is_teardown}
+
+ def is_relevant_teardown(other):
+ if other.task_id in scope_teardowns:
+ return True
+ elif other.is_teardown and not any(x.is_setup for x in other.upstream_list):
+ return True
+ else:
+ return False
+
+ downstream_tasks = [
+ x for x in context["task"].get_flat_relatives(upstream=False) if not is_relevant_teardown(x)
+ ]
self.log.debug("Downstream task IDs %s", downstream_tasks)
if downstream_tasks:
dag_run = context["dag_run"]
execution_date = dag_run.execution_date
-
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
if self.ignore_downstream_trigger_rules is True:
self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ self.skip(
+ dag_run=dag_run,
+ execution_date=execution_date,
+ tasks=downstream_tasks,
+ map_index=context["ti"].map_index,
+ )
else:
self.log.info("Skipping downstream tasks while respecting trigger rules...")
# Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
# Scheduler handle the remaining downstream task(s) appropriately.
+ to_skip = [
+ x
+ for x in context["task"].get_direct_relatives(upstream=False)
+ if not is_relevant_teardown(x)
+ ]
Review Comment:
Need a comment around here to explain why relevant teardowns should not be skipped. Same for the call above.
--
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 #32538: Fix short circuit operator re teardowns
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1262910810
##########
airflow/operators/python.py:
##########
@@ -251,27 +253,37 @@ def execute(self, context: Context) -> Any:
self.log.info("Proceeding with downstream tasks...")
return condition
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
- self.log.debug("Downstream task IDs %s", downstream_tasks)
+ if not self.downstream_task_ids:
+ self.log.info("No downstream tasks; nothing to do.")
+ return
- if downstream_tasks:
- dag_run = context["dag_run"]
- execution_date = dag_run.execution_date
+ dag_run = context["dag_run"]
+ execution_date = dag_run.execution_date
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
+ def get_tasks_to_skip():
if self.ignore_downstream_trigger_rules is True:
- self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ tasks = context["task"].get_flat_relatives(upstream=False)
else:
- self.log.info("Skipping downstream tasks while respecting trigger rules...")
- # Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
- # Scheduler handle the remaining downstream task(s) appropriately.
- self.skip(
- dag_run,
- execution_date,
- context["task"].get_direct_relatives(upstream=False),
- map_index=context["ti"].map_index,
- )
-
+ tasks = context["task"].get_direct_relatives(upstream=False)
+ for t in tasks:
+ if not t.is_teardown:
+ yield t
+
+ to_skip = get_tasks_to_skip()
+
+ # this let's us avoid an intermediate list unless debug logging
+ if self.log.getEffectiveLevel() <= logging.DEBUG:
+ self.log.debug("Downstream task IDs %s", to_skip := list(get_tasks_to_skip()))
Review Comment:
> Ah OK get_tasks_to_skip() is an iterator. Maybe move the line to below this block instead then?
don't follow -- what would that do?
--
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 #32538: Fix short circuit operator re teardowns
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1260472292
##########
airflow/operators/python.py:
##########
@@ -250,25 +251,35 @@ def execute(self, context: Context) -> Any:
if condition:
self.log.info("Proceeding with downstream tasks...")
return condition
-
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
+ downstream_tasks = [
+ x for x in context["task"].get_flat_relatives(upstream=False) if not x.is_teardown
+ ]
self.log.debug("Downstream task IDs %s", downstream_tasks)
if downstream_tasks:
dag_run = context["dag_run"]
execution_date = dag_run.execution_date
-
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
if self.ignore_downstream_trigger_rules is True:
self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ self.skip(
+ dag_run=dag_run,
+ execution_date=execution_date,
+ tasks=downstream_tasks,
+ map_index=context["ti"].map_index,
+ )
else:
self.log.info("Skipping downstream tasks while respecting trigger rules...")
# Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
# Scheduler handle the remaining downstream task(s) appropriately.
+ to_skip = [
+ x for x in context["task"].get_direct_relatives(upstream=False) if not x.is_teardown
+ ]
Review Comment:
I wonder instead of getting the tasks again (which is surpringly not that trivial), it may make more sense to simply filter from `downstream_tasks`
```python
to_skip = [t for t in downstream_tasks if self.task_id in t.upstream_task_ids]
```
--
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 #32538: Fix short circuit operator re teardowns
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1260929248
##########
airflow/operators/python.py:
##########
@@ -251,27 +253,37 @@ def execute(self, context: Context) -> Any:
self.log.info("Proceeding with downstream tasks...")
return condition
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
- self.log.debug("Downstream task IDs %s", downstream_tasks)
+ if not self.downstream_task_ids:
+ self.log.info("No downstream tasks; nothing to do.")
+ return
- if downstream_tasks:
- dag_run = context["dag_run"]
- execution_date = dag_run.execution_date
+ dag_run = context["dag_run"]
+ execution_date = dag_run.execution_date
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
+ def get_tasks_to_skip():
if self.ignore_downstream_trigger_rules is True:
- self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ tasks = context["task"].get_flat_relatives(upstream=False)
else:
- self.log.info("Skipping downstream tasks while respecting trigger rules...")
- # Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
- # Scheduler handle the remaining downstream task(s) appropriately.
- self.skip(
- dag_run,
- execution_date,
- context["task"].get_direct_relatives(upstream=False),
- map_index=context["ti"].map_index,
- )
-
+ tasks = context["task"].get_direct_relatives(upstream=False)
+ for t in tasks:
+ if not t.is_teardown:
+ yield t
+
+ to_skip = get_tasks_to_skip()
+
+ # this let's us avoid an intermediate list unless debug logging
+ if self.log.getEffectiveLevel() <= logging.DEBUG:
+ self.log.debug("Downstream task IDs %s", to_skip := list(get_tasks_to_skip()))
Review Comment:
Pushing the boundary of the walrus I see…
But should line 274 be removed?
##########
airflow/operators/python.py:
##########
@@ -251,27 +253,37 @@ def execute(self, context: Context) -> Any:
self.log.info("Proceeding with downstream tasks...")
return condition
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
- self.log.debug("Downstream task IDs %s", downstream_tasks)
+ if not self.downstream_task_ids:
+ self.log.info("No downstream tasks; nothing to do.")
+ return
- if downstream_tasks:
- dag_run = context["dag_run"]
- execution_date = dag_run.execution_date
+ dag_run = context["dag_run"]
+ execution_date = dag_run.execution_date
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
+ def get_tasks_to_skip():
if self.ignore_downstream_trigger_rules is True:
- self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ tasks = context["task"].get_flat_relatives(upstream=False)
else:
- self.log.info("Skipping downstream tasks while respecting trigger rules...")
- # Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
- # Scheduler handle the remaining downstream task(s) appropriately.
- self.skip(
- dag_run,
- execution_date,
- context["task"].get_direct_relatives(upstream=False),
- map_index=context["ti"].map_index,
- )
-
+ tasks = context["task"].get_direct_relatives(upstream=False)
+ for t in tasks:
+ if not t.is_teardown:
+ yield t
+
+ to_skip = get_tasks_to_skip()
+
+ # this let's us avoid an intermediate list unless debug logging
+ if self.log.getEffectiveLevel() <= logging.DEBUG:
+ self.log.debug("Downstream task IDs %s", to_skip := list(get_tasks_to_skip()))
Review Comment:
Pushing the boundary of the walrus I see…
But should line 274 be removed?
--
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 #32538: Fix short circuit operator re teardowns
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1260568877
##########
airflow/operators/python.py:
##########
@@ -250,25 +251,35 @@ def execute(self, context: Context) -> Any:
if condition:
self.log.info("Proceeding with downstream tasks...")
return condition
-
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
+ downstream_tasks = [
+ x for x in context["task"].get_flat_relatives(upstream=False) if not x.is_teardown
+ ]
self.log.debug("Downstream task IDs %s", downstream_tasks)
if downstream_tasks:
dag_run = context["dag_run"]
execution_date = dag_run.execution_date
-
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
if self.ignore_downstream_trigger_rules is True:
self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ self.skip(
+ dag_run=dag_run,
+ execution_date=execution_date,
+ tasks=downstream_tasks,
+ map_index=context["ti"].map_index,
+ )
else:
self.log.info("Skipping downstream tasks while respecting trigger rules...")
# Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
# Scheduler handle the remaining downstream task(s) appropriately.
+ to_skip = [
+ x for x in context["task"].get_direct_relatives(upstream=False) if not x.is_teardown
+ ]
Review Comment:
note: i removed a debug log which emitted the full list of short circuited tasks so that way we could use generator but if you think that's no bueno LMK.
--
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 #32538: Fix short circuit operator re teardowns
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1261500185
##########
airflow/operators/python.py:
##########
@@ -251,27 +253,37 @@ def execute(self, context: Context) -> Any:
self.log.info("Proceeding with downstream tasks...")
return condition
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
- self.log.debug("Downstream task IDs %s", downstream_tasks)
+ if not self.downstream_task_ids:
+ self.log.info("No downstream tasks; nothing to do.")
+ return
- if downstream_tasks:
- dag_run = context["dag_run"]
- execution_date = dag_run.execution_date
+ dag_run = context["dag_run"]
+ execution_date = dag_run.execution_date
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
+ def get_tasks_to_skip():
if self.ignore_downstream_trigger_rules is True:
- self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ tasks = context["task"].get_flat_relatives(upstream=False)
else:
- self.log.info("Skipping downstream tasks while respecting trigger rules...")
- # Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
- # Scheduler handle the remaining downstream task(s) appropriately.
- self.skip(
- dag_run,
- execution_date,
- context["task"].get_direct_relatives(upstream=False),
- map_index=context["ti"].map_index,
- )
-
+ tasks = context["task"].get_direct_relatives(upstream=False)
+ for t in tasks:
+ if not t.is_teardown:
+ yield t
+
+ to_skip = get_tasks_to_skip()
+
+ # this let's us avoid an intermediate list unless debug logging
+ if self.log.getEffectiveLevel() <= logging.DEBUG:
+ self.log.debug("Downstream task IDs %s", to_skip := list(get_tasks_to_skip()))
Review Comment:
> Pushing the boundary of the walrus I see…
yes, hehe
> But should line 274 be removed?
No because on 274 it is a generator. only if we need to log.debug is there a need to "materialize" it in the form of a list, because then it needs to be logged and passed to `skip`, which makes another list out of it anyway.
So it's maybe a silly optimization but hey you started it 😛
##########
airflow/operators/python.py:
##########
@@ -251,27 +253,37 @@ def execute(self, context: Context) -> Any:
self.log.info("Proceeding with downstream tasks...")
return condition
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
- self.log.debug("Downstream task IDs %s", downstream_tasks)
+ if not self.downstream_task_ids:
+ self.log.info("No downstream tasks; nothing to do.")
+ return
- if downstream_tasks:
- dag_run = context["dag_run"]
- execution_date = dag_run.execution_date
+ dag_run = context["dag_run"]
+ execution_date = dag_run.execution_date
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
+ def get_tasks_to_skip():
if self.ignore_downstream_trigger_rules is True:
- self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ tasks = context["task"].get_flat_relatives(upstream=False)
else:
- self.log.info("Skipping downstream tasks while respecting trigger rules...")
- # Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
- # Scheduler handle the remaining downstream task(s) appropriately.
- self.skip(
- dag_run,
- execution_date,
- context["task"].get_direct_relatives(upstream=False),
- map_index=context["ti"].map_index,
- )
-
+ tasks = context["task"].get_direct_relatives(upstream=False)
+ for t in tasks:
+ if not t.is_teardown:
+ yield t
+
+ to_skip = get_tasks_to_skip()
+
+ # this let's us avoid an intermediate list unless debug logging
+ if self.log.getEffectiveLevel() <= logging.DEBUG:
+ self.log.debug("Downstream task IDs %s", to_skip := list(get_tasks_to_skip()))
Review Comment:
> Pushing the boundary of the walrus I see…
yes, hehe
> But should line 274 be removed?
No because on 274 it is a generator. only if we need to log.debug is there a need to "materialize" it in the form of a list, because then it needs to be logged and passed to `skip`, which makes another list out of it anyway.
So it's maybe a silly optimization but hey you started it 😛
--
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 #32538: Fix short circuit operator re teardowns
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1260440104
##########
airflow/operators/python.py:
##########
@@ -251,24 +252,47 @@ def execute(self, context: Context) -> Any:
self.log.info("Proceeding with downstream tasks...")
return condition
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
+ scope_teardowns = {x.task_id for x in self.get_upstreams_only_setups_and_teardowns() if x.is_teardown}
+
+ def is_relevant_teardown(other):
+ if other.task_id in scope_teardowns:
+ return True
+ elif other.is_teardown and not any(x.is_setup for x in other.upstream_list):
+ return True
+ else:
+ return False
+
+ downstream_tasks = [
+ x for x in context["task"].get_flat_relatives(upstream=False) if not is_relevant_teardown(x)
+ ]
self.log.debug("Downstream task IDs %s", downstream_tasks)
if downstream_tasks:
dag_run = context["dag_run"]
execution_date = dag_run.execution_date
-
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
if self.ignore_downstream_trigger_rules is True:
self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ self.skip(
+ dag_run=dag_run,
+ execution_date=execution_date,
+ tasks=downstream_tasks,
+ map_index=context["ti"].map_index,
+ )
else:
self.log.info("Skipping downstream tasks while respecting trigger rules...")
# Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
# Scheduler handle the remaining downstream task(s) appropriately.
+ to_skip = [
+ x
+ for x in context["task"].get_direct_relatives(upstream=False)
+ if not is_relevant_teardown(x)
+ ]
Review Comment:
Need a comment around here to explain why relevant teardowns should not be skipped.
--
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 #32538: Fix short circuit operator re teardowns
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1260465841
##########
airflow/operators/python.py:
##########
@@ -251,24 +252,47 @@ def execute(self, context: Context) -> Any:
self.log.info("Proceeding with downstream tasks...")
return condition
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
+ scope_teardowns = {x.task_id for x in self.get_upstreams_only_setups_and_teardowns() if x.is_teardown}
+
+ def is_relevant_teardown(other):
+ if other.task_id in scope_teardowns:
+ return True
+ elif other.is_teardown and not any(x.is_setup for x in other.upstream_list):
+ return True
+ else:
+ return False
+
+ downstream_tasks = [
+ x for x in context["task"].get_flat_relatives(upstream=False) if not is_relevant_teardown(x)
+ ]
self.log.debug("Downstream task IDs %s", downstream_tasks)
if downstream_tasks:
dag_run = context["dag_run"]
execution_date = dag_run.execution_date
-
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
if self.ignore_downstream_trigger_rules is True:
self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ self.skip(
+ dag_run=dag_run,
+ execution_date=execution_date,
+ tasks=downstream_tasks,
+ map_index=context["ti"].map_index,
+ )
else:
self.log.info("Skipping downstream tasks while respecting trigger rules...")
# Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
# Scheduler handle the remaining downstream task(s) appropriately.
+ to_skip = [
+ x
+ for x in context["task"].get_direct_relatives(upstream=False)
+ if not is_relevant_teardown(x)
+ ]
Review Comment:
ok... updated... simplified.... I was making it more complicated than it needed to be. We don't need to manually skip any teardowns because if they should be skipped, the scheduler will handle it.
--
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 #32538: Fix short circuit operator re teardowns
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1261500185
##########
airflow/operators/python.py:
##########
@@ -251,27 +253,37 @@ def execute(self, context: Context) -> Any:
self.log.info("Proceeding with downstream tasks...")
return condition
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
- self.log.debug("Downstream task IDs %s", downstream_tasks)
+ if not self.downstream_task_ids:
+ self.log.info("No downstream tasks; nothing to do.")
+ return
- if downstream_tasks:
- dag_run = context["dag_run"]
- execution_date = dag_run.execution_date
+ dag_run = context["dag_run"]
+ execution_date = dag_run.execution_date
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
+ def get_tasks_to_skip():
if self.ignore_downstream_trigger_rules is True:
- self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ tasks = context["task"].get_flat_relatives(upstream=False)
else:
- self.log.info("Skipping downstream tasks while respecting trigger rules...")
- # Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
- # Scheduler handle the remaining downstream task(s) appropriately.
- self.skip(
- dag_run,
- execution_date,
- context["task"].get_direct_relatives(upstream=False),
- map_index=context["ti"].map_index,
- )
-
+ tasks = context["task"].get_direct_relatives(upstream=False)
+ for t in tasks:
+ if not t.is_teardown:
+ yield t
+
+ to_skip = get_tasks_to_skip()
+
+ # this let's us avoid an intermediate list unless debug logging
+ if self.log.getEffectiveLevel() <= logging.DEBUG:
+ self.log.debug("Downstream task IDs %s", to_skip := list(get_tasks_to_skip()))
Review Comment:
> Pushing the boundary of the walrus I see…
yes, hehe
> But should line 274 be removed?
No because on 274 it is a generator. only if we need to log.debug is there a need to "materialize" it in the form of a list, because then it needs to be logged and passed to `skip`, which makes another list out of it anyway.
So it's maybe a silly optimization but hey you started it 😛
But so generally it will just be generator here, passed to skip as generator. But if user enables debug logging, then when we log it we render it into a list at that log line, and then pass the same list down to skip.
--
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 #32538: Fix short circuit operator re teardowns
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1261592086
##########
airflow/operators/python.py:
##########
@@ -251,27 +253,37 @@ def execute(self, context: Context) -> Any:
self.log.info("Proceeding with downstream tasks...")
return condition
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
- self.log.debug("Downstream task IDs %s", downstream_tasks)
+ if not self.downstream_task_ids:
+ self.log.info("No downstream tasks; nothing to do.")
+ return
- if downstream_tasks:
- dag_run = context["dag_run"]
- execution_date = dag_run.execution_date
+ dag_run = context["dag_run"]
+ execution_date = dag_run.execution_date
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
Review Comment:
Ok actually... at least we can use `cast("DateTime", dag_run.execution_date)` which is a little cleaner than the assert i suppose
--
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 #32538: Fix short circuit operator re teardowns
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1260579342
##########
airflow/operators/python.py:
##########
@@ -250,25 +251,35 @@ def execute(self, context: Context) -> Any:
if condition:
self.log.info("Proceeding with downstream tasks...")
return condition
-
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
+ downstream_tasks = [
+ x for x in context["task"].get_flat_relatives(upstream=False) if not x.is_teardown
+ ]
self.log.debug("Downstream task IDs %s", downstream_tasks)
if downstream_tasks:
dag_run = context["dag_run"]
execution_date = dag_run.execution_date
-
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
if self.ignore_downstream_trigger_rules is True:
self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ self.skip(
+ dag_run=dag_run,
+ execution_date=execution_date,
+ tasks=downstream_tasks,
+ map_index=context["ti"].map_index,
+ )
else:
self.log.info("Skipping downstream tasks while respecting trigger rules...")
# Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
# Scheduler handle the remaining downstream task(s) appropriately.
+ to_skip = [
+ x for x in context["task"].get_direct_relatives(upstream=False) if not x.is_teardown
+ ]
Review Comment:
ok pushed the optimization / refactor a little further here. hopefully i don't break something. lemme know what you think. restored the debug logging, and now it will debug log the actual lists in either case
--
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 #32538: Fix short circuit operator re teardowns
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1260562506
##########
airflow/operators/python.py:
##########
@@ -250,25 +251,35 @@ def execute(self, context: Context) -> Any:
if condition:
self.log.info("Proceeding with downstream tasks...")
return condition
-
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
+ downstream_tasks = [
+ x for x in context["task"].get_flat_relatives(upstream=False) if not x.is_teardown
+ ]
self.log.debug("Downstream task IDs %s", downstream_tasks)
if downstream_tasks:
dag_run = context["dag_run"]
execution_date = dag_run.execution_date
-
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
if self.ignore_downstream_trigger_rules is True:
self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ self.skip(
+ dag_run=dag_run,
+ execution_date=execution_date,
+ tasks=downstream_tasks,
+ map_index=context["ti"].map_index,
+ )
else:
self.log.info("Skipping downstream tasks while respecting trigger rules...")
# Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
# Scheduler handle the remaining downstream task(s) appropriately.
+ to_skip = [
+ x for x in context["task"].get_direct_relatives(upstream=False) if not x.is_teardown
+ ]
Review Comment:
a different way to optimize would to not create the `downstream_tasks` list at all in the `ignore_downstream_trigger_rules is False` case. then we could just pass to skip `x for x in context["task"].get_direct_relatives(upstream=False) if not x.is_teardown` as a generator since `skip` method can handle that -- i.e. there's not really a need for selectively calling skip.
but stepping back, i'm not adding any new call to either get_flat_relatives or get_direct_relatives -- both calls were there and i don't increase their likelihood of calling or frequency. i just happened to move to a variable cus it was handy to debug something.
##########
airflow/operators/python.py:
##########
@@ -250,25 +251,35 @@ def execute(self, context: Context) -> Any:
if condition:
self.log.info("Proceeding with downstream tasks...")
return condition
-
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
+ downstream_tasks = [
+ x for x in context["task"].get_flat_relatives(upstream=False) if not x.is_teardown
+ ]
self.log.debug("Downstream task IDs %s", downstream_tasks)
if downstream_tasks:
dag_run = context["dag_run"]
execution_date = dag_run.execution_date
-
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
if self.ignore_downstream_trigger_rules is True:
self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ self.skip(
+ dag_run=dag_run,
+ execution_date=execution_date,
+ tasks=downstream_tasks,
+ map_index=context["ti"].map_index,
+ )
else:
self.log.info("Skipping downstream tasks while respecting trigger rules...")
# Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
# Scheduler handle the remaining downstream task(s) appropriately.
+ to_skip = [
+ x for x in context["task"].get_direct_relatives(upstream=False) if not x.is_teardown
+ ]
Review Comment:
a different way to optimize would to not create the `downstream_tasks` list at all in the `ignore_downstream_trigger_rules is False` case. then we could just pass to skip `x for x in context["task"].get_direct_relatives(upstream=False) if not x.is_teardown` as a generator since `skip` method can handle that -- i.e. there's not really a need for selectively calling skip.
but stepping back, just to be clear, i'm not adding any new call to either get_flat_relatives or get_direct_relatives -- both calls were there and i don't increase their likelihood of calling or frequency. i just happened to move to a variable cus it was handy to debug something.
--
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 #32538: Fix short circuit operator re teardowns
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1260928029
##########
airflow/operators/python.py:
##########
@@ -251,27 +253,37 @@ def execute(self, context: Context) -> Any:
self.log.info("Proceeding with downstream tasks...")
return condition
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
- self.log.debug("Downstream task IDs %s", downstream_tasks)
+ if not self.downstream_task_ids:
+ self.log.info("No downstream tasks; nothing to do.")
+ return
- if downstream_tasks:
- dag_run = context["dag_run"]
- execution_date = dag_run.execution_date
+ dag_run = context["dag_run"]
+ execution_date = dag_run.execution_date
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
Review Comment:
I think you can do `execution_date = context["logical_date"]`, that one should be guaranteed to be a pendulum DateTime.
--
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 #32538: Fix short circuit operator re teardowns
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish merged PR #32538:
URL: https://github.com/apache/airflow/pull/32538
--
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 #32538: Fix short circuit operator re teardowns
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1262188392
##########
airflow/operators/python.py:
##########
@@ -251,27 +253,37 @@ def execute(self, context: Context) -> Any:
self.log.info("Proceeding with downstream tasks...")
return condition
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
- self.log.debug("Downstream task IDs %s", downstream_tasks)
+ if not self.downstream_task_ids:
+ self.log.info("No downstream tasks; nothing to do.")
+ return
- if downstream_tasks:
- dag_run = context["dag_run"]
- execution_date = dag_run.execution_date
+ dag_run = context["dag_run"]
+ execution_date = dag_run.execution_date
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
+ def get_tasks_to_skip():
if self.ignore_downstream_trigger_rules is True:
- self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ tasks = context["task"].get_flat_relatives(upstream=False)
else:
- self.log.info("Skipping downstream tasks while respecting trigger rules...")
- # Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
- # Scheduler handle the remaining downstream task(s) appropriately.
- self.skip(
- dag_run,
- execution_date,
- context["task"].get_direct_relatives(upstream=False),
- map_index=context["ti"].map_index,
- )
-
+ tasks = context["task"].get_direct_relatives(upstream=False)
+ for t in tasks:
+ if not t.is_teardown:
+ yield t
+
+ to_skip = get_tasks_to_skip()
+
+ # this let's us avoid an intermediate list unless debug logging
+ if self.log.getEffectiveLevel() <= logging.DEBUG:
+ self.log.debug("Downstream task IDs %s", to_skip := list(get_tasks_to_skip()))
Review Comment:
Ah OK `get_tasks_to_skip()` is an iterator. Maybe move the line to below this block instead then?
Personally I’d maybe rename the function or `to_skip` to be clearer there’s an iterator involved (either `iter_tasks_to_skip()` or `tasks_to_skip_iter`)
##########
airflow/operators/python.py:
##########
@@ -251,27 +253,37 @@ def execute(self, context: Context) -> Any:
self.log.info("Proceeding with downstream tasks...")
return condition
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
- self.log.debug("Downstream task IDs %s", downstream_tasks)
+ if not self.downstream_task_ids:
+ self.log.info("No downstream tasks; nothing to do.")
+ return
- if downstream_tasks:
- dag_run = context["dag_run"]
- execution_date = dag_run.execution_date
+ dag_run = context["dag_run"]
+ execution_date = dag_run.execution_date
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
+ def get_tasks_to_skip():
if self.ignore_downstream_trigger_rules is True:
- self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ tasks = context["task"].get_flat_relatives(upstream=False)
else:
- self.log.info("Skipping downstream tasks while respecting trigger rules...")
- # Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
- # Scheduler handle the remaining downstream task(s) appropriately.
- self.skip(
- dag_run,
- execution_date,
- context["task"].get_direct_relatives(upstream=False),
- map_index=context["ti"].map_index,
- )
-
+ tasks = context["task"].get_direct_relatives(upstream=False)
+ for t in tasks:
+ if not t.is_teardown:
+ yield t
+
+ to_skip = get_tasks_to_skip()
+
+ # this let's us avoid an intermediate list unless debug logging
+ if self.log.getEffectiveLevel() <= logging.DEBUG:
+ self.log.debug("Downstream task IDs %s", to_skip := list(get_tasks_to_skip()))
Review Comment:
Ah OK `get_tasks_to_skip()` is an iterator. Maybe move the line to below this block instead then?
Personally I’d maybe rename the function or `to_skip` to be clearer there’s an iterator involved (either `iter_tasks_to_skip()` or `tasks_to_skip_iter`)
--
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 #32538: Fix short circuit operator re teardowns
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1260472292
##########
airflow/operators/python.py:
##########
@@ -250,25 +251,35 @@ def execute(self, context: Context) -> Any:
if condition:
self.log.info("Proceeding with downstream tasks...")
return condition
-
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
+ downstream_tasks = [
+ x for x in context["task"].get_flat_relatives(upstream=False) if not x.is_teardown
+ ]
self.log.debug("Downstream task IDs %s", downstream_tasks)
if downstream_tasks:
dag_run = context["dag_run"]
execution_date = dag_run.execution_date
-
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
if self.ignore_downstream_trigger_rules is True:
self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ self.skip(
+ dag_run=dag_run,
+ execution_date=execution_date,
+ tasks=downstream_tasks,
+ map_index=context["ti"].map_index,
+ )
else:
self.log.info("Skipping downstream tasks while respecting trigger rules...")
# Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
# Scheduler handle the remaining downstream task(s) appropriately.
+ to_skip = [
+ x for x in context["task"].get_direct_relatives(upstream=False) if not x.is_teardown
+ ]
Review Comment:
I wonder instead of getting the tasks again (which is surpringly not that trivial), it may make more sense to simply filter from `downstream_tasks`
```python
to_skip = [t for t in get_direct_relatives if self.task_id in t.upstream_task_ids]
```
--
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 #32538: Fix short circuit operator re teardowns
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #32538:
URL: https://github.com/apache/airflow/pull/32538#discussion_r1263154303
##########
airflow/operators/python.py:
##########
@@ -251,27 +253,37 @@ def execute(self, context: Context) -> Any:
self.log.info("Proceeding with downstream tasks...")
return condition
- downstream_tasks = context["task"].get_flat_relatives(upstream=False)
- self.log.debug("Downstream task IDs %s", downstream_tasks)
+ if not self.downstream_task_ids:
+ self.log.info("No downstream tasks; nothing to do.")
+ return
- if downstream_tasks:
- dag_run = context["dag_run"]
- execution_date = dag_run.execution_date
+ dag_run = context["dag_run"]
+ execution_date = dag_run.execution_date
+ if TYPE_CHECKING:
+ assert isinstance(execution_date, DateTime)
+ def get_tasks_to_skip():
if self.ignore_downstream_trigger_rules is True:
- self.log.info("Skipping all downstream tasks...")
- self.skip(dag_run, execution_date, downstream_tasks, map_index=context["ti"].map_index)
+ tasks = context["task"].get_flat_relatives(upstream=False)
else:
- self.log.info("Skipping downstream tasks while respecting trigger rules...")
- # Explicitly setting the state of the direct, downstream task(s) to "skipped" and letting the
- # Scheduler handle the remaining downstream task(s) appropriately.
- self.skip(
- dag_run,
- execution_date,
- context["task"].get_direct_relatives(upstream=False),
- map_index=context["ti"].map_index,
- )
-
+ tasks = context["task"].get_direct_relatives(upstream=False)
+ for t in tasks:
+ if not t.is_teardown:
+ yield t
+
+ to_skip = get_tasks_to_skip()
+
+ # this let's us avoid an intermediate list unless debug logging
+ if self.log.getEffectiveLevel() <= logging.DEBUG:
+ self.log.debug("Downstream task IDs %s", to_skip := list(get_tasks_to_skip()))
Review Comment:
just going to merge this... if you want to golf it further, you can tag me in 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