Repository navigation
control state of individual taskflow in mapped task-group #40543
Description
Discussed in #33556
Originally posted by ntnhaatj August 20, 2023
Description
Hi, as my issue was raised here which also included depth-first execution model context, I would like to open an issue to have more discussion on this matter.
Use case/motivation
motivated from the depth-first execution model implemented in mapped TaskGroup:
┌──────────────┐ ┌────────────────┐ ┌───────────┐
┌──►│extract_file_1│►│transform_file_1│►│load_file_1│
│ └──────────────┘ └────────────────┘ └───────────┘
│
│ ┌──────────────┐ ┌────────────────┐ ┌───────────┐
┌─────────────┐ ├──►│extract_file_2│►│transform_file_2│►│load_file_2│
│get_file_list├─┤ └──────────────┘ └────────────────┘ └───────────┘
└─────────────┘ │
│ ... ... ...
│
│ ┌──────────────┐ ┌────────────────┐ ┌───────────┐
└──►│extract_file_N│►│transform_file_n│►│load_file_N│
└──────────────┘ └────────────────┘ └───────────┘
At present, the upstream / downstream list dependencies now only applied on DAGNode (which is the whole TaskGroup or Operator)
There might be better control over desired TaskFlow if we could enhance support for deeper upstream/downstream dependencies at the granularity of mapped task instances, instead of applying them to the entire TaskGroup.
In the model above, extract_file_1 task downstream list should be only transform_file_1 and load_file_1 in order, rather than encompassing the whole mapped group transform_file[] and load_file[] as in the current implementation.
For instance, I scheduled my test DAG on Airflow 2.7.0:
from airflow.decorators import dag, task_group, task
from airflow.operators.empty import EmptyOperator
from pendulum import datetime
files = ["a", "b", "c"]
@dag(start_date=datetime(2022, 12, 1), schedule=None, catchup=False)
def task_group_mapping_example():
@task_group(group_id="etl")
def etl_pipeline(file):
e = EmptyOperator(task_id="e")
t = EmptyOperator(task_id="t")
l = EmptyOperator(task_id="l")
e >> t >> l
etl = etl_pipeline.expand(file=files)
etl
task_group_mapping_example()Clear etl.e[1] state with downstream will trigger all mapped task in etl.t[] and etl.l[] group.
Thanks,
Related issues
No response
Are you willing to submit a PR?
- Yes I am willing to submit a PR!
Code of Conduct
- I agree to follow this project's Code of Conduct
Activity
boraberke commentedon Jul 2, 2024
I would like to give this one a try. Could you please assign me?
potiuk commentedon Jul 2, 2024
Sure
gavinhonl commentedon Jul 15, 2024
@potiuk - Thanks for raising this issue. I was about to create a new issue myself. IMO - the behaviour should be as follows:
9 remaining items
sharpnife commentedon Dec 23, 2024
@harjeevanmaan @shahar1 Any progress on this issue? 🙂
shahar1 commentedon Dec 23, 2024
Currently this issue is unassigned, so if it's important for you - you're welcome to join the efforts.
potiuk commentedon Dec 23, 2024
Indeed. FYI. @sharpnife -> the best way to influence speed of implementing task is to roll sleeves up and implement it, rather than ask for startus. That's how open-source works. Generally the status is always as seen in the issue - if no-one works on it, well, no-one works on it.
Shekharrajak commentedon Dec 31, 2025
In current implementation - when clearing etl.e[1] with downstream dependencies:
I think we need a function to get the map_index for a relative task that's in the same mapped task group.
Please assign the task to me, let me share the PR.
Fix clearing mapped task instances in same group to respect depth-fir…
potiuk commentedon Mar 2, 2026
@Shekharrajak We are unassigning you from this issue as part of our updated assignment policy.
This is not meant to discourage your contribution — quite the opposite! You are still very welcome to work on this issue and submit a PR for it. Simply comment that you are working on it and open a PR when ready.
We found that formal assignments were not working well, as they often prevented others from contributing when the assignee was not actively working on the issue.
adithya2754 commentedon Jul 17, 2026
Hey @potiuk could you please assign me? I would like to give it a try
shahar1 commentedon Jul 17, 2026
We don't assign users to GitHub issues anymore by policy (with some exceptions that don't apply for this case).
You're welcome to work on it and create a Pull Request.
gitedmond commentedon Sep 5, 2026
I tested the original scenario from this issue against current main (9f978d3), and I could not reproduce the reported behavior.
The regression test creates a serialized Dag with a three-item mapped TaskGroup containing extract >> transform >> load, creates a real DagRun, and verifies that all nine TaskInstance rows exist. It then calls the public clearTaskInstances API using the same mapped-task marker sent by the UI.
I tested indexes 0, 1, and 2 in both directions:
• Clearing etl.extract[index] with downstream tasks included returned and cleared only extract[index], transform[index], and load[index].
• Clearing etl.load[index] with upstream tasks included returned and cleared only extract[index], transform[index], and load[index].
• Preview mode left all states unchanged.
• Actual clearing changed only the three TaskInstance rows associated with the selected index.
All six parameterized cases passed against unchanged production code.
Since the original report used Airflow 2.7.0, the behavior appears to have been corrected by subsequent mapped-task handling changes, although I have not identified the exact fixing commit. This result does not imply that the original report was user error.
Could a maintainer confirm whether this issue can be closed as already fixed on current main? I can also provide the regression test if retaining explicit coverage for this behavior would be useful.
kevinhongzl commentedon Oct 13, 2026
@potiuk I believe we've fixed this issue in #57758 and we can close it.