jgoedeke opened a new issue, #59008:
URL: https://github.com/apache/airflow/issues/59008
### Apache Airflow version
3.1.3
### If "Other Airflow 2/3 version" selected, which one?
_No response_
### What happened?
When using dynamic task mapping within @task_group and expand_kwargs(), XCom
pulls from mapped tasks behave inconsistently based on the number of mapped
instances. If n > 1, ti.xcom_pull(task_ids="task_group.task") returns a list of
results; if n = 1, it returns a dict/object directly, not wrapped in a list.
This inconsistency is not expected and makes template usage unpredictable.
### What you think should happen instead?
XCom pulls should return a consistent data type regardless of the number of
mapped instances. Either always return a list—even with a single element for n
= 1—or ensure that mapping always returns per-instance data without extra index
logic. Currently, logic in Jinja templates must defensively handle both
structures, e.g. using complex conditional checks on the data type (see below).
### How to reproduce
```
from airflow.providers.standard.operators.bash import BashOperator
from airflow.sdk import dag, task, task_group
@dag(schedule=None, tags=["test"])
def test_mapped_taskgroup_xcom():
@task
def generate_inputs():
# The inconsistency occurs when this task returns a list with only
one item
return [
{"name": "item1", "value": 100},
{"name": "item2", "value": 200},
{"name": "item3", "value": 300},
]
@task_group
def process_item(name: str, value: int):
@task
def get_metadata(name: str, value: int) -> dict:
return {
"name": name,
"value": value,
"doubled": value * 2,
"message": f"Processing {name}",
}
def metadata_output(key: str) -> str:
return (
f'{{{{
ti.xcom_pull(task_ids="process_item.get_metadata")["{key}"] }}}}'
)
use_metadata = BashOperator(
task_id="use_metadata",
bash_command=f'echo "Name: {metadata_output("name")}, Value:
{metadata_output("value")}, Doubled: {metadata_output("doubled")}, Message:
{metadata_output("message")}"',
)
@task
def final_task(metadata: dict) -> None:
print(f"Final processing: {metadata}")
metadata = get_metadata(name, value)
metadata >> use_metadata >> final_task(metadata)
inputs = generate_inputs()
process_item.expand_kwargs(inputs)
test_mapped_taskgroup_xcom()
```
### Operating System
Docker Compose
### Versions of Apache Airflow Providers
_No response_
### Deployment
Other Docker-based deployment
### Deployment details
_No response_
### Anything else?
Workaround:
```python
def metadata_output(key: str) -> str:
# When task group is mapped with multiple items, XCom returns a
list
# When mapped with single item, XCom returns the dict directly
# Use |list test to check if it's a list (sequence but not
string/mapping)
return f'{{% set xcom_data =
ti.xcom_pull(task_ids="process_item.get_metadata") %}}{{{{
(xcom_data[ti.map_index] if xcom_data is sequence and xcom_data is not string
and xcom_data is not mapping else xcom_data)["{key}"] }}}}'
````
### 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
