tanelk opened a new issue, #37073:
URL: https://github.com/apache/airflow/issues/37073
### Apache Airflow version
main (development)
### If "Other Airflow 2 version" selected, which one?
_No response_
### What happened?
When returning multiple outputs from a task and mapping task groups over
these values, then the number of expanded tasks is not the number of returned
values, but instead the number of returned keys in the dictionary.
### What you think should happen instead?
_No response_
### How to reproduce
Minimal DAG to reproduce:
```from airflow import DAG
from airflow.decorators import task_group, task
from datetime import datetime
@task(multiple_outputs=True)
def list_values():
return {
'values1': [1],
'values2': [2, 3, 4],
}
@task
def subtask(value):
print(value)
@task_group
def process(value):
subtask(value=value)
with DAG(
dag_id='test_dag',
schedule_interval="*/1 * * * *",
start_date=datetime(2024, 1, 29),
max_active_runs=1,
) as dag:
values = list_values()
process.override(group_id='process1').expand(value=values['values1'])
process.override(group_id='process2').expand(value=values['values2'])
```
The `subtask` in `process1` should have 1 mapped task and in `process2` it
should have 3 mapped tasks.
Instead there are 2 mapped tasks in both task groups.
The 2 tasks in `process2` print correct values (2 and 3), but third value is
not accessed.
First task in `process1` print correct value (1) and the second one gets
exception:
```
[2024-01-29, 14:54:11 UTC] {abstractoperator.py:707} ERROR - Exception
rendering Jinja template for task 'process1.subtask', field 'op_kwargs'.
Template: {'value':
MappedArgument(_input=DictOfListsExpandInput(value={'value': XComArg(,
'values1')}), _key='value')}
Traceback (most recent call last):
File "/opt/airflow/airflow/models/abstractoperator.py", line 699, in
_do_render_template_fields
rendered_content = self.render_template(
File "/opt/airflow/airflow/template/templater.py", line 186, in
render_template
return {k: self.render_template(v, context, jinja_env, oids) for k, v in
value.items()}
File "/opt/airflow/airflow/template/templater.py", line 186, in
return {k: self.render_template(v, context, jinja_env, oids) for k, v in
value.items()}
File "/opt/airflow/airflow/template/templater.py", line 176, in
render_template
return value.resolve(context)
File "/opt/airflow/airflow/utils/session.py", line 79, in wrapper
return func(*args, session=session, **kwargs)
File "/opt/airflow/airflow/models/expandinput.py", line 70, in resolve
data, _ = self._input.resolve(context, session=session)
File "/opt/airflow/airflow/models/expandinput.py", line 201, in resolve
data = {k: self._expand_mapped_field(k, v, context, session=session) for
k, v in self.value.items()}
File "/opt/airflow/airflow/models/expandinput.py", line 201, in
data = {k: self._expand_mapped_field(k, v, context, session=session) for
k, v in self.value.items()}
File "/opt/airflow/airflow/models/expandinput.py", line 185, in
_expand_mapped_field
return value[found_index]
IndexError: list index out of range
```
A screenshot of the DAG:
### Operating System
Arch linux
### Versions of Apache Airflow Providers
_No response_
### Deployment
Other
### Deployment details
Tried with virtualenv install of
[v2.8.1](https://pypi.python.org/pypi/apache-airflow/2.8.1) and also with
breeze on main branch.
### Anything else?
_No response_
### Are you willing to submit PR?
- [ ] 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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
