shahar1 commented on code in PR #75417:
URL: https://github.com/apache/airflow/pull/75417#discussion_r4360826933
##########
airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -459,6 +459,11 @@ def _evaluate_direct_relatives() -> Iterator[TIDepStatus]:
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)
+ # Once its in-group upstreams have expanded, the summary ti of a
task in a mapped
+ # task group has no upstream ti left to judge; the expanded tis
decide instead.
+ if map_index < 0 and upstream == 0 and
task.get_closest_mapped_task_group() is not None:
Review Comment:
Done in f2d1f5efeb. The guard now checks for an upstream task_id missing
from `task_id_counts` instead of `upstream == 0`. With a failed outside
upstream, `ALL_DONE_MIN_ONE_SUCCESS` now decides per index: `{0, 1, 2:
success}` when the in-group upstream succeeds, and `skipped` /
`upstream_failed` on a skipped / failed index.
I scoped it to both `ALL_DONE_MIN_ONE_SUCCESS` and
`NONE_FAILED_MIN_ONE_SUCCESS`, not to ADMOS alone, because of your other
comment. Once NFMOS leaves `is_fast_triggered`, an ADMOS-only guard regresses
NFMOS: with a skipped outside upstream and a successful in-group one, the
placeholder counts only the skipped task, `skipped == upstream`, and the whole
task ends `{-1: skipped}` (main gives `{0, 1, 2: success}`). Both #39801 tests
fail in that setup too. Other rules still decide at the placeholder.
The cost you described does apply to NFMOS: if the outside upstream fails
after the group has expanded, the task gets one `upstream_failed` ti per index
instead of one in total. The run outcome is the same, and the comment above the
guard states it. I did not add a test for that case, because the result depends
on whether the outside task fails before or after the expansion.
---
Drafted-by: Claude Code (Opus 5.5) (no human review before posting)
##########
airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -459,6 +459,11 @@ def _evaluate_direct_relatives() -> Iterator[TIDepStatus]:
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)
+ # Once its in-group upstreams have expanded, the summary ti of a
task in a mapped
Review Comment:
Done in f2d1f5efeb: `NONE_FAILED_MIN_ONE_SUCCESS` is out of
`is_fast_triggered`, and the comment above the tuple no longer lists it. It
only works together with the guard change in the other thread (see there for
why). The per-index test is now parametrized over both rules and covers NFMOS
with a failed in-group index (`{0: success, 1: upstream_failed, 2: success}`)
and with a skipped outside upstream plus a failed in-group index.
---
Drafted-by: Claude Code (Opus 5.5) (no human review before posting)
##########
airflow-core/tests/unit/models/test_mappedoperator.py:
##########
@@ -1956,15 +1956,18 @@ def group(value):
assert states[task_id] == {0: "success", 1: "skipped"}
-def
test_none_failed_min_one_success_trigger_rule_expands_in_mapped_task_group(dag_maker):
- """Regression test for #39801.
[email protected](
+ "trigger_rule", [TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS,
TriggerRule.ALL_DONE_MIN_ONE_SUCCESS]
+)
+def test_min_one_success_trigger_rule_expands_in_mapped_task_group(dag_maker,
trigger_rule):
+ """Regression test for #39801 and its ``ALL_DONE_MIN_ONE_SUCCESS``
counterpart.
Review Comment:
Done in f2d1f5efeb, the docstring now describes the behaviour.
---
Drafted-by: Claude Code (Opus 5.5) (no human review before posting)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
