From 3b428ec84a8f7b9a21ab9f053d40fafc69cfd0ae Mon Sep 17 00:00:00 2001 From: Shahar Epstein <60007259+shahar1@users.noreply.github.com> Date: Thu, 22 Oct 2026 11:55:03 +0300 Subject: [PATCH 1/2] Run all_done_min_one_success tasks inside mapped task groups A task using the all_done_min_one_success trigger rule inside a dynamically mapped task group never expanded: as soon as its in-group upstream expanded, the not-yet-expanded placeholder of the task saw no upstream task instances left to judge, treated the empty set as "all done, nothing succeeded" and was marked upstream_failed, failing the Dag run even when every upstream succeeded. The placeholder now defers to the expanded task instances, which then apply the trigger rule per map index, matching how all_success and the other "all done" rules already behave in mapped task groups. --- .../airflow/ti_deps/deps/trigger_rule_dep.py | 5 ++ .../tests/unit/models/test_mappedoperator.py | 69 ++++++++++++++++--- 2 files changed, 63 insertions(+), 11 deletions(-) diff --git a/airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py b/airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py index 9ccc4adcb133b..6b8656813d557 100644 --- a/airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py +++ b/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: + return + upstream_done = done >= upstream changed = False diff --git a/airflow-core/tests/unit/models/test_mappedoperator.py b/airflow-core/tests/unit/models/test_mappedoperator.py index a6b89e039780f..eb878bfb69479 100644 --- a/airflow-core/tests/unit/models/test_mappedoperator.py +++ b/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. +@pytest.mark.parametrize( + "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. - A task with the ``NONE_FAILED_MIN_ONE_SUCCESS`` trigger rule inside a dynamically + A task with a ``*_MIN_ONE_SUCCESS`` trigger rule inside a dynamically expanded task group must expand and run once its upstream (also inside the group) - succeeds, instead of being skipped prematurely at its unexpanded (map_index -1) - summary ti before the task group has expanded. + succeeds, instead of being skipped or marked upstream_failed prematurely at its + unexpanded (map_index -1) summary ti. """ - with dag_maker(dag_id="test_none_failed_min_one_success_in_mapped_task_group") as dag: + with dag_maker(dag_id=f"test_{trigger_rule.value}_in_mapped_task_group") as dag: @task def init(): @@ -1976,7 +1979,7 @@ def tg(message): def imsleepy(message): return message - @task(trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS) + @task(trigger_rule=trigger_rule) def imawake(message): pass @@ -1994,14 +1997,17 @@ def imawake(message): assert states["tg.imawake"] == {0: "success", 1: "success", 2: "success"} -def test_none_failed_min_one_success_trigger_rule_expands_in_nested_task_group(dag_maker): +@pytest.mark.parametrize( + "trigger_rule", [TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS, TriggerRule.ALL_DONE_MIN_ONE_SUCCESS] +) +def test_min_one_success_trigger_rule_expands_in_nested_task_group(dag_maker, trigger_rule): """Regression test for #39801, nested-group shape. - Same as above, but the ``NONE_FAILED_MIN_ONE_SUCCESS`` task lives in a plain task + Same as above, but the ``*_MIN_ONE_SUCCESS`` task lives in a plain task group nested inside the mapped one (the exact shape of the issue's reproducer), so its immediate ``task_group`` is not mapped — only an ancestor is. """ - with dag_maker(dag_id="test_none_failed_min_one_success_in_nested_task_group") as dag: + with dag_maker(dag_id=f"test_{trigger_rule.value}_in_nested_task_group") as dag: @task def init(): @@ -2015,7 +2021,7 @@ def imsleepy(message): @task_group(group_id="inner") def inner(message): - @task(trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS) + @task(trigger_rule=trigger_rule) def imawake(message): pass @@ -2035,6 +2041,47 @@ def imawake(message): assert states["tg.inner.imawake"] == {0: "success", 1: "success", 2: "success"} +@pytest.mark.parametrize( + ("upstream_outcome", "expected_state"), + [("skip", "skipped"), ("fail", "upstream_failed")], +) +def test_all_done_min_one_success_decides_per_index_in_mapped_task_group( + dag_maker, upstream_outcome, expected_state +): + """An ``ALL_DONE_MIN_ONE_SUCCESS`` task in a mapped group is evaluated per expanded index.""" + with dag_maker(dag_id=f"test_all_done_min_one_success_upstream_{upstream_outcome}") as dag: + + @task + def init(): + return ["seize", "the", "day"] + + @task_group(group_id="tg") + def tg(message): + @task + def imsleepy(message): + if message == "the": + if upstream_outcome == "skip": + raise AirflowSkipException + raise ValueError("boom") + return message + + @task(trigger_rule=TriggerRule.ALL_DONE_MIN_ONE_SUCCESS) + def imawake(message): + pass + + imawake(imsleepy(message)) + + tg.expand(message=init()) + + dr = dag.test() + + states: dict[str, dict[int, str | None]] = defaultdict(dict) + for ti in dr.get_task_instances(): + states[ti.task_id][ti.map_index] = ti.state + + assert states["tg.imawake"] == {0: "success", 1: expected_state, 2: "success"} + + def test_mapped_operator_retry_delay_default(dag_maker): """ Test that MappedOperator.retry_delay returns default value when not explicitly set. From f2d1f5efebe105fb4d5001a5ec46342e9c9dc093 Mon Sep 17 00:00:00 2001 From: Shahar Epstein <60007259+shahar1@users.noreply.github.com> Date: Fri, 23 Oct 2026 17:13:47 +0300 Subject: [PATCH 2/2] Decide *_min_one_success per index when an upstream is outside the group The placeholder of an all_done_min_one_success task in a mapped task group only deferred to its expanded instances when it saw no upstream at all. With an extra upstream outside the group, it judged that upstream alone: a failed outside task marked the whole task upstream_failed, although every index would have seen one success from its in-group upstream. The placeholder now defers as soon as any in-group upstream has expanded out of its view. This also covers none_failed_min_one_success, which drops its fast-trigger entry: counting every upstream index made one failed index fail the whole task. For these two rules, an outside failure seen after the expansion now gives one upstream_failed instance per index instead of one in total; the run outcome is the same. Other rules still decide at the placeholder, so they keep a single upstream_failed instance. --- .../airflow/ti_deps/deps/trigger_rule_dep.py | 33 +++++++----- .../tests/unit/models/test_mappedoperator.py | 51 +++++++++++++------ 2 files changed, 56 insertions(+), 28 deletions(-) diff --git a/airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py b/airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py index 6b8656813d557..8ed5d25e425a7 100644 --- a/airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py +++ b/airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py @@ -195,20 +195,18 @@ def _get_relevant_upstream_map_indexes(upstream_id: str) -> int | range | None: # Only the not-yet-expanded summary ti (map_index < 0) needs the broad # "depend on every upstream ti" behavior, so a rule that can be satisfied - # by a subset of upstreams (ONE_SUCCESS / ONE_FAILED / ONE_DONE / - # NONE_FAILED_MIN_ONE_SUCCESS) does not skip it before the mapped task - # group has expanded (see #34023, #39801). Once the ti is expanded, each - # instance must depend on the upstream instance(s) that share its map - # index, otherwise a single upstream failure would wrongly trigger every - # expanded instance (see #50210). The task may be nested in plain task - # groups inside the mapped one (see #39801), so check the closest mapped - # ancestor rather than only the immediate parent group. + # by a subset of upstreams (ONE_SUCCESS / ONE_FAILED / ONE_DONE) does not + # skip it before the mapped task group has expanded (see #34023). Once the + # ti is expanded, each instance must depend on the upstream instance(s) that + # share its map index, otherwise a single upstream failure would wrongly + # trigger every expanded instance (see #50210). The task may be nested in + # plain task groups inside the mapped one (see #39801), so check the closest + # mapped ancestor rather than only the immediate parent group. if map_index < 0 and task.get_closest_mapped_task_group() is not None: is_fast_triggered = task.trigger_rule in ( TR.ONE_SUCCESS, TR.ONE_FAILED, TR.ONE_DONE, - TR.NONE_FAILED_MIN_ONE_SUCCESS, ) if is_fast_triggered and upstream_id not in set( _iter_expansion_dependencies(task_group=task.task_group) @@ -422,6 +420,7 @@ def _evaluate_direct_relatives() -> Iterator[TIDepStatus]: success_setup = upstream_states.success_setup skipped_setup = upstream_states.skipped_setup + task_id_counts: list[tuple[str, int]] | None = None # Optimization: Don't need to hit the database if all upstreams are # "simple" tasks (no task or task group mapping involved). if not any(t.get_needs_expansion() for t in upstream_tasks.values()): @@ -431,7 +430,6 @@ def _evaluate_direct_relatives() -> Iterator[TIDepStatus]: # Regional selections share counts only for identical producer UUID sets; # legacy runs can share by task IDs. Expansion invalidates both forms. cache_key: tuple[str, str, frozenset[str | UUID]] | None = None - task_id_counts: list[tuple[str, int]] | None = None if task.get_closest_mapped_task_group() is None: scope: frozenset[str | UUID] = ( frozenset( @@ -459,9 +457,18 @@ 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: + # Once an in-group upstream has expanded, the summary ti of a task in a mapped task + # group no longer sees it and would judge every index by the outside upstreams alone. + # For these two rules the expanded tis decide instead. The cost: an outside failure seen + # after the expansion gives one upstream_failed ti per index instead of one in total, + # which is why other rules still decide here. + if ( + map_index < 0 + and trigger_rule in (TR.ALL_DONE_MIN_ONE_SUCCESS, TR.NONE_FAILED_MIN_ONE_SUCCESS) + and task_id_counts is not None + and len(task_id_counts) < len(upstream_tasks) + and task.get_closest_mapped_task_group() is not None + ): return upstream_done = done >= upstream diff --git a/airflow-core/tests/unit/models/test_mappedoperator.py b/airflow-core/tests/unit/models/test_mappedoperator.py index eb878bfb69479..807b0aa2587f7 100644 --- a/airflow-core/tests/unit/models/test_mappedoperator.py +++ b/airflow-core/tests/unit/models/test_mappedoperator.py @@ -1960,12 +1960,10 @@ def group(value): "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. + """A ``*_MIN_ONE_SUCCESS`` task in a mapped task group runs once its in-group upstream succeeds. - A task with a ``*_MIN_ONE_SUCCESS`` trigger rule inside a dynamically - expanded task group must expand and run once its upstream (also inside the group) - succeeds, instead of being skipped or marked upstream_failed prematurely at its - unexpanded (map_index -1) summary ti. + It must not be skipped or marked upstream_failed prematurely at its unexpanded + (map_index -1) summary ti. """ with dag_maker(dag_id=f"test_{trigger_rule.value}_in_mapped_task_group") as dag: @@ -2042,34 +2040,57 @@ def imawake(message): @pytest.mark.parametrize( - ("upstream_outcome", "expected_state"), - [("skip", "skipped"), ("fail", "upstream_failed")], + ("trigger_rule", "outside_outcome", "inner_outcome", "expected_state"), + [ + (TriggerRule.ALL_DONE_MIN_ONE_SUCCESS, None, "skip", "skipped"), + (TriggerRule.ALL_DONE_MIN_ONE_SUCCESS, None, "fail", "upstream_failed"), + (TriggerRule.ALL_DONE_MIN_ONE_SUCCESS, "fail", "success", "success"), + (TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS, None, "fail", "upstream_failed"), + (TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS, "skip", "fail", "upstream_failed"), + ], ) -def test_all_done_min_one_success_decides_per_index_in_mapped_task_group( - dag_maker, upstream_outcome, expected_state +def test_min_one_success_decides_per_index_in_mapped_task_group( + dag_maker, trigger_rule, outside_outcome, inner_outcome, expected_state ): - """An ``ALL_DONE_MIN_ONE_SUCCESS`` task in a mapped group is evaluated per expanded index.""" - with dag_maker(dag_id=f"test_all_done_min_one_success_upstream_{upstream_outcome}") as dag: + """A ``*_MIN_ONE_SUCCESS`` task in a mapped group is evaluated per expanded index. + + The middle index of its in-group upstream ends with ``inner_outcome``; an optional + upstream outside the group ends with ``outside_outcome``. + """ + with dag_maker( + dag_id=f"test_{trigger_rule.value}_outside_{outside_outcome}_inner_{inner_outcome}" + ) as dag: @task def init(): return ["seize", "the", "day"] + @task + def outside(): + if outside_outcome == "skip": + raise AirflowSkipException + raise ValueError("boom") + + outside_task = outside() if outside_outcome else None + @task_group(group_id="tg") def tg(message): @task def imsleepy(message): if message == "the": - if upstream_outcome == "skip": + if inner_outcome == "skip": raise AirflowSkipException - raise ValueError("boom") + if inner_outcome == "fail": + raise ValueError("boom") return message - @task(trigger_rule=TriggerRule.ALL_DONE_MIN_ONE_SUCCESS) + @task(trigger_rule=trigger_rule) def imawake(message): pass - imawake(imsleepy(message)) + awake = imawake(imsleepy(message)) + if outside_task is not None: + outside_task >> awake tg.expand(message=init())