MatrixManAtYrService opened a new issue, #25389:
URL: https://github.com/apache/airflow/issues/25389
### Apache Airflow version
main (development)
### What happened
Here's a DAG:
```python3
with DAG(
dag_id="expand_kwargs",
schedule_interval=None,
start_date=datetime(1970, 1, 1),
) as dag:
@task
def data():
return [
("hello {}", "foo"),
("goodbye {}", "bar"),
]
def mapper(entry):
if entry[1] != "bar":
return {"template": entry[0], "user": entry[1]}
else:
return None
@task
def printer(template, user):
print(template.format(user))
printer.expand_kwargs(data().map(mapper))
```
My understanding (from [this
test](https://github.com/apache/airflow/blob/main/tests/models/test_xcom_arg_map.py#L110))
is that I should be able to return `None` and prevent a task from being
created for that item. But when I try to run this I get an error:
```
{base_executor.py:91} INFO - Adding to queue: ['<TaskInstance:
expand_kwargs_mod_tf.one_task backfill__2022-07-28T00:00:00+00:00 map_index=1
[queued]>']
hello foo
{debug_executor.py:84} ERROR - Failed to execute task: 'NoneType' object is
not iterable.
Traceback (most recent call last):
File "/home/matt/src/airflow/airflow/executors/debug_executor.py", line
78, in _run_task
ti.run(job_id=ti.job_id, **params)
File "/home/matt/src/airflow/airflow/utils/session.py", line 71, in wrapper
return func(*args, session=session, **kwargs)
File "/home/matt/src/airflow/airflow/models/taskinstance.py", line 1782,
in run
self._run_raw_task(
File "/home/matt/src/airflow/airflow/utils/session.py", line 68, in wrapper
return func(*args, **kwargs)
File "/home/matt/src/airflow/airflow/models/taskinstance.py", line 1445,
in _run_raw_task
self._execute_task_with_callbacks(context, test_mode)
File "/home/matt/src/airflow/airflow/models/taskinstance.py", line 1580,
in _execute_task_with_callbacks
task_orig = self.render_templates(context=context)
File "/home/matt/src/airflow/airflow/models/taskinstance.py", line 2202,
in render_templates
rendered_task = self.task.render_template_fields(context)
File "/home/matt/src/airflow/airflow/models/mappedoperator.py", line 751,
in render_template_fields
unmapped_task = self.unmap(mapped_kwargs)
File "/home/matt/src/airflow/airflow/models/mappedoperator.py", line 592,
in unmap
kwargs = self._get_unmap_kwargs(kwargs,
strict=self._disallow_kwargs_override)
File "/home/matt/src/airflow/airflow/decorators/base.py", line 460, in
_get_unmap_kwargs
prevent_duplicates(
File "/home/matt/src/airflow/airflow/models/mappedoperator.py", line 127,
in prevent_duplicates
duplicated_keys = set(kwargs1).intersection(kwargs2)
TypeError: 'NoneType' object is not iterable
```
### What you think should happen instead
Since one of the two mapped items returned None, this task should end up
with just one instance "foo" and there should be no errors while queuing it.
Or, if I'm misunderstanding how this should work, then the error that I see
should do more to point me in the right direction.
### How to reproduce
Run the dag given above
### Operating System
NixOS 21.11 (linux kernel 5.10.101)
### Versions of Apache Airflow Providers
n/a
### Deployment
Virtualenv installation
### Deployment details
_No response_
### Anything else
I thought that this might be only a problem in the `prevent_duplicates`
function, so I made the following change:
```python3
# added 'or set()' below
duplicated_keys = set(kwargs1 or set()).intersection(kwargs2 or set())
```
Following that I got a different error:
```
[2022-07-28 16:17:59,391] {debug_executor.py:84} ERROR - Failed to execute
task: 'NoneType' object is not a mapping.
Traceback (most recent call last):
File "/home/matt/src/airflow/airflow/executors/debug_executor.py", line
78, in _run_task
ti.run(job_id=ti.job_id, **params)
File "/home/matt/src/airflow/airflow/utils/session.py", line 71, in wrapper
return func(*args, session=session, **kwargs)
File "/home/matt/src/airflow/airflow/models/taskinstance.py", line 1782,
in run
self._run_raw_task(
File "/home/matt/src/airflow/airflow/utils/session.py", line 68, in wrapper
return func(*args, **kwargs)
File "/home/matt/src/airflow/airflow/models/taskinstance.py", line 1445,
in _run_raw_task
self._execute_task_with_callbacks(context, test_mode)
File "/home/matt/src/airflow/airflow/models/taskinstance.py", line 1580,
in _execute_task_with_callbacks
task_orig = self.render_templates(context=context)
File "/home/matt/src/airflow/airflow/models/taskinstance.py", line 2202,
in render_templates
rendered_task = self.task.render_template_fields(context)
File "/home/matt/src/airflow/airflow/models/mappedoperator.py", line 751,
in render_template_fields
unmapped_task = self.unmap(mapped_kwargs)
File "/home/matt/src/airflow/airflow/models/mappedoperator.py", line 592,
in unmap
kwargs = self._get_unmap_kwargs(kwargs,
strict=self._disallow_kwargs_override)
File "/home/matt/src/airflow/airflow/decorators/base.py", line 467, in
_get_unmap_kwargs
self._combined_op_kwargs = {**self.partial_kwargs["op_kwargs"],
**mapped_kwargs["op_kwargs"]}
TypeError: 'NoneType' object is not a mapping
```
### 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]
