Fix short-circuit inside a mapped task group not skipping downstream (#73959)

Author: kaxil

airflow-core/src/airflow/ti_deps/deps/not_previously_skipped_dep.py

@@ -60,8 +60,10 @@ def _get_dep_statuses(self, ti, dep_context, *, session):
 
                 # Use the parent's map context to look up the XCom. An unmapped parent
                 # (e.g. LatestOnlyOperator) writes XCom with map_index=-1, so we must
-                # query with -1 instead of the child's map_index.
-                xcom_map_index = ti.map_index if parent.is_mapped else -1
+                # query with -1 instead of the child's map_index. A parent inside a
+                # mapped task group is expanded like a mapped task, so it writes XCom
+                # per map index even though ``is_mapped`` is False.
+                xcom_map_index = ti.map_index if parent.get_needs_expansion() else -1
                 prev_result = ti.xcom_pull(
                     task_ids=parent.task_id,
                     key=XCOM_SKIPMIXIN_KEY,

airflow-core/tests/unit/ti_deps/deps/test_not_previously_skipped_dep.py

@@ -27,11 +27,13 @@
 from airflow.models.xcom import XComModel
 from airflow.providers.standard.operators.empty import EmptyOperator
 from airflow.providers.standard.operators.python import BranchPythonOperator
+from airflow.sdk import task, task_group
 from airflow.sdk.bases.xcom import BaseXCom
 from airflow.ti_deps.dep_context import DepContext
 from airflow.ti_deps.deps.not_previously_skipped_dep import (
     XCOM_SKIPMIXIN_FOLLOWED,
     XCOM_SKIPMIXIN_KEY,
+    XCOM_SKIPMIXIN_SKIPPED,
     NotPreviouslySkippedDep,
 )
 from airflow.utils.state import State
@@ -219,6 +221,48 @@ def test_unmapped_parent_skip_mapped_downstream(session, dag_maker):
     assert tis["op2"].state == State.SKIPPED
 
 
+def test_parent_in_mapped_task_group_skips_same_map_index(session, dag_maker):
+    """
+    A SkipMixin parent inside a mapped task group writes XCom per map index, so
+    each child TI in the group must read the decision for its own map index.
+    """
+    with dag_maker("test_mapped_group_skip_dag", schedule=None, session=session):
+
+        @task.short_circuit(task_id="gate")
+        def gate(value):
+            return value
+
+        @task_group
+        def group(value):
+            gate(value) >> EmptyOperator(task_id="child")
+
+        group.expand(value=[True, False])
+
+    dr = dag_maker.create_dagrun(run_type=DagRunType.MANUAL, state=State.RUNNING)
+    tis = {(ti.task_id, ti.map_index): ti for ti in dr.task_instances}
+    for map_index in (0, 1):
+        tis[("group.gate", map_index)].state = State.SUCCESS
+        session.merge(tis[("group.gate", map_index)])
+    # Only the map index 1 gate short-circuited, as SkipMixin.skip records it.
+    XComModel.set(
+        key=XCOM_SKIPMIXIN_KEY,
+        value={XCOM_SKIPMIXIN_SKIPPED: ["group.child"]},
+        dag_id=dr.dag_id,
+        task_id="group.gate",
+        run_id=dr.run_id,
+        map_index=1,
+        session=session,
+    )
+    session.flush()
+
+    dep = NotPreviouslySkippedDep()
+
+    assert not dep.is_met(tis[("group.child", 1)], session=session)
+    assert tis[("group.child", 1)].state == State.SKIPPED
+    assert dep.is_met(tis[("group.child", 0)], session=session)
+    assert tis[("group.child", 0)].state != State.SKIPPED
+
+
 def test_branch_skip_decision_bypasses_custom_xcom_backend(session, dag_maker):
     """
     A value-externalizing custom XCom backend must not break branch-skip of