Fix per-index evaluation of ONE_FAILED in mapped task groups

[shahar1]

closes: #50210

A task using a "fast triggered" trigger rule (ONE_FAILED, ONE_SUCCESS, ONE_DONE) inside a mapped task group was evaluated against every expanded instance of its upstream, instead of the upstream instance that shares its own map index. As a result, a single failed (or succeeded) upstream instance wrongly triggered the rule for every expanded instance of the task.

For example, in the reproduction from the issue, divide(0) fails while divide(1..3) succeed, yet the mapped report_failure task (ONE_FAILED) ran for all map indexes instead of only index 0.

Root cause

TriggerRuleDep._get_relevant_upstream_map_indexes short-circuits to "depend on every upstream instance" for the fast-triggered rules when the upstream is not an expansion dependency of the group (introduced in #44937 to fix #34023). That broad behavior is only needed for the not-yet-expanded summary task instance (map_index < 0), so the task is not prematurely skipped before the mapped task group expands. Applying it to the already-expanded instances broke per-index correspondence.

Fix

Restrict the special case to the summary instance (ti.map_index < 0). Expanded instances now use the normal per-map-index upstream resolution, so each report_failure[k] depends on divide[k] only.

Tests

• Added test_one_failed_trigger_rule_in_mapped_task_group_is_per_index (the issue's repro: report_failure is now {0: success, 1..3: skipped}).
• Added test_one_failed_trigger_rule_runs_on_indirect_failure_in_mapped_task_group (end-to-end guard for the Trigger Rule ONE_FAILED does not work in task group with mapped tasks #34023 behavior).
• The existing test_mapped_tasks_in_mapped_task_group_waits_for_upstreams_to_complete ([v2-10-test] Fix premature evaluation in mapped task group #44937, premature-skip guard) still passes.

---

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Opus 4.8)

Generated-by: Claude Code (Opus 4.8) following the guidelines

[shahar1]

Tested against the original repro. example in the issue (compiled into Airflow 3), seems to do the job:

from __future__ import annotations

import logging
from datetime import datetime

from airflow.sdk import DAG, task, task_group
from airflow.utils.trigger_rule import TriggerRule

logger = logging.getLogger(__name__)


@task
def divide(i):
    return 30 / i


@task(trigger_rule=TriggerRule.ONE_FAILED)
def report_failure(i):
    logger.info("%s failed", i)


@task
def report_success(i):
    logger.info("%s succeeded", i)


with DAG(
    dag_id="Test-Trigger-Rule",
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False,
    tags=["test", "trigger-rule"],
) as dag:

    @task
    def gen_examples():
        return [0, 1, 2, 3]

    @task_group
    def divide_and_report(i):
        divide(i) >> [report_success(i), report_failure(i)]

    divide_and_report.expand(i=gen_examples())

[kaxil]

Looks good to me

[github-actions]

Hi maintainer, this PR was merged without a milestone set.
We've automatically set the milestone to Airflow 3.2.3 based on: backport label targeting v3-2-test
If this milestone is not correct, please update it to the appropriate milestone.

This comment was generated by Milestone Tag Assistant.

[github-actions]

Backport successfully created: v3-2-test

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
✅	v3-2-test