Dynamic mapped tasks group arguments are interpreted as MappedArgument when provided to classic operators
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 10h
- Merged PRs (30d)
- 483
Description
### Discussed in https://github.com/apache/airflow/discussions/40728
Originally posted by **florian-guily** April 24, 2024
### Apache Airflow version
Other Airflow 2 version (please specify below)
### If "Other Airflow 2 version" selected, which one?
2.8.4
### What happened?
When using expand_kwargs on a task group, arguments of this task group will not get correctly interpreted when using them in classic operators inside this task group. They will be interpreted as MappedArgument instead of their real values.
### What you think should happen instead?
Real value of the mapped task group's argument should be passed to the operators.
### How to reproduce
This was originally done with a google cloud operator in a task group, but i managed to reproduce it with a bash operator.
```
from airflow import DAG
from airflow.decorators import task, task_group
from airflow.operators.bash import BashOperator
from pendulum import datetime
dag = DAG(
dag_id="airflow_issue_test",
start_date=datetime(2024, 1, 1, tz='UTC'),
catchup=False,
schedule="@daily",
default_args={"retries": 2},
)
with dag:
@task()
def list_dict_generator():
my_list = [
{
"project": "my_project",
"dataset": f"dataset_{number}",
"table": "my_table",
"partition_id": "my_partition_id",
}
for number in range(10)
]
return my_list
@task_group()
def my_tg(project, dataset, table_name, partition_id):
BashOperator(
task_id="bash_task",
bash_command=f"echo {project}.{dataset}.{table_name}${partition_id}",
env={"MY_VAR": "Hello World"}
)
partitions_to_delete = list_dict_generator()
my_tg.expand_kwargs(partitions_to_delete)
```
Here are the associated logs. You can clearly see that multiple MappedArguments are passed in the echo command, which should not happen.
[dag_id=airflow_issue_test_run_id=scheduled__2024-04-14T00_00_00+00_00_task_id=bq_to_gcs_tg.bash_task_map_index=0_attempt=2.log](https://github.com/apache/airflow/files/15091225/dag_id.airflow_issue_test_run_id.scheduled__2024-04-14T00_00_00%2B00_00_task_id.bq_to_gcs_tg.bash_task_map_index.0_attempt.2.log)
### Operating System
Mac OS Sonoma 14.2.1 (23C71)
### Versions of Apache Airflow Providers
apache-airflow-providers-google==10.16.0
apache-airflow-providers-common-sql>=1.11.0
### Deployment
Docker-Compose
### Deployment details
_No response_
### Anything else?
_No response_
### Are you willing to submit PR?
- [X] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
Contributor guide
Research direction
Start with the expand_kwargs task-group example and the BashOperator entry point, reproducing the issue with the provided DAG. Trace how mapped task-group arguments reach classic operators; done means the operators receive real values instead of MappedArgument instances, with a regression test covering the example.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- data-engineering
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100