kaxil commented on code in PR #75417:
URL: https://github.com/apache/airflow/pull/75417#discussion_r4359652482
##########
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:
This gates on the total count, so it only fires when every direct upstream
is inside the group. Add one upstream outside the group that fails (the case
`ALL_DONE_MIN_ONE_SUCCESS` exists to tolerate) and `upstream` is 1, so the -1
ti still goes to `upstream_failed` even though every expanded ti would see one
success and run. That's the first known limitation in the description.
Checking whether an upstream task_id is missing from `task_id_counts`,
instead of `upstream == 0`, fixed that shape in a local run (`{0, 1, 2:
success}`, and test_trigger_rule_dep, test_dagrun, test_mappedoperator and
test_taskinstance still pass). For other rules I think that condition would
also expand the task before the outside upstream finishes, e.g. an
`all_success` task with a failed outside upstream would get N `upstream_failed`
tis instead of one, so it probably wants scoping to `ALL_DONE_MIN_ONE_SUCCESS`.
Worth doing here, or as a follow-up with an issue?
##########
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:
With this guard in place, `NONE_FAILED_MIN_ONE_SUCCESS` doesn't seem to need
its `is_fast_triggered` entry (L211) anymore. With that line removed locally,
both #39801 tests still pass, and your per-index test switched to that rule
gives `{0: success, 1: upstream_failed, 2: success}` for the failed index
instead of `{-1: upstream_failed}`, which is the second known limitation.
test_trigger_rule_dep, test_dagrun and test_mappedoperator all pass. Could go
in this PR along with the comment above the tuple, since the backport touches
the same file anyway.
##########
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:
Since this line is being rewritten anyway, could the docstring describe the
behaviour rather than point at #39801? The review instructions ask to keep
issue numbers out of test docstrings. Something like ``A ``*_MIN_ONE_SUCCESS``
task in a mapped task group expands and runs once its in-group upstream
succeeds.``
--
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]
