Skip to content

Remove auto unroll for dict values from taskflow #27826

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
bolkedebruin wants to merge 2 commits into from
Closed

Remove auto unroll for dict values from taskflow #27826

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
eac885b
Remove auto unroll for dict values from taskflow
bolkedebruin Nov 22, 2022
05b51cb
Explicitly fail on implicit unroll
bolkedebruin Nov 22, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions airflow/decorators/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -300,6 +300,10 @@ def _infer_multiple_outputs(self):
except TypeError: # Can't evaluate return type.
return False
ttype = getattr(return_type, "__origin__", return_type)
if ttype == dict or ttype == Dict:
raise AttributeError(
"multiple_outputs was not set and will not implicitly unroll dict for return values"
)
Comment on lines +304 to +306

@uranusjr uranusjr Nov 23, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

AttributeError feels wrong; maybe something like RuntimeError or a custom exception class is more suitable. I’d also want the message to be clearer (tell the user “set multiple_outputs=True is to retain the old behaviour”).

@uranusjr uranusjr Nov 23, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I wonder if the new DagWarning mechanism is even better for this. It would still allow the DAG to parse (i.e. not breaking people’s existing setups) but instruct the user to fix things.

@bolkedebruin bolkedebruin Nov 23, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it is an AttributeError. The attribute multiple_outputs was not set, so we don't know what to do. An AirflowException is very generic.

If we are going for option 2 I don't think the DAG should parse and in that case I think the error message should not refer to the old behavior as that implies knowing what that is. What you are describing seems more like a deprecation warning though and thus option 3?

return ttype == dict or ttype == Dict

def __attrs_post_init__(self):
Expand Down
10 changes: 6 additions & 4 deletions docs/apache-airflow/tutorial/taskflow.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -377,19 +377,21 @@ section "Having sensors return XOM values" of :doc:`apache-airflow-providers:how

Multiple outputs inference
--------------------------
Tasks can also infer multiple outputs by using dict Python typing.
In case a task has multiple return values, Airflow can unroll these in separate values if using
Python dict typing. The values then become available as separate arguments to downstream tasks.

.. code-block:: python

@task
@task(multiple_outputs=True)
def identity_dict(x: int, y: int) -> dict[str, int]:
return {"x": x, "y": y}

By using the typing ``Dict`` for the function return type, the ``multiple_outputs`` parameter
is automatically set to true.

Note, If you manually set the ``multiple_outputs`` parameter the inference is disabled and
the parameter value is used.
Note, before Airflow 2.5, Airflow would automatically unroll ``dict`` return values and you would need to set
``multiple_outputs=False` explicitly. This was deemed confusing and now always defaults to ``False``.


Adding dependencies between decorated and traditional tasks
-----------------------------------------------------------
Expand Down
2 changes: 2 additions & 0 deletions newsfragments/27819.significant.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
Airflow no longer automatically unrolls dict values into separate arguments requiring an explicit
``multiple_outputs=True`` if unrolling is required.
17 changes: 6 additions & 11 deletions tests/decorators/test_python.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,19 +94,14 @@ def test_python_operator_python_callable_is_callable(self):
"Dict[str, int]",
],
)
def test_infer_multiple_outputs_using_dict_typing(self, resolve, annotation):
@task_decorator
def identity_dict(x: int, y: int) -> resolve(annotation):
return {"x": x, "y": y}

assert identity_dict(5, 5).operator.multiple_outputs is True
def test_infer_multiple_outputs_using_dict_typing_raises(self, resolve, annotation):
with pytest.raises(AttributeError) as e:

# Check invoking ``@task_decorator.__call__()`` yields the correct inference.
@task_decorator()
def identity_dict_with_decorator_call(x: int, y: int) -> resolve(annotation):
return {"x": x, "y": y}
@task_decorator()
def identity_dict(x: int, y: int) -> resolve(annotation):
return {"x": x, "y": y}

assert identity_dict_with_decorator_call(5, 5).operator.multiple_outputs is True
assert "not implicitly unroll" in str(e.value)

def test_infer_multiple_outputs_using_other_typing(self):
@task_decorator
Expand Down