Caching Tasks with multiple return values

[andersy005]

I'm trying to enable caching for tasks that return multiple values, and I was wondering whether this is something that Prefect can handle and/or whether this would be an anti-pattern? Below is a reproducible example that demonstrates what I am trying to do.

import prefect
from prefect import task, Flow
import time

@task(target="baz", result=prefect.engine.results.LocalResult(dir="/tmp/testing"), nout=2)
def testing():
    time.sleep(5)
    return 9, 10

with Flow("xarray") as flow:
    a, b = testing()

When I run the flow for the first time, Prefect doesn't appear to complain or indicate that something is broken:

%%time
flow.run()

[2021-12-06 08:52:32-0700] INFO - prefect.FlowRunner | Beginning Flow run for 'xarray'
[2021-12-06 08:52:32-0700] INFO - prefect.TaskRunner | Task 'testing': Starting task run...
[2021-12-06 08:52:37-0700] INFO - prefect.TaskRunner | Task 'testing': Finished task run for task with final state: 'Success'
[2021-12-06 08:52:37-0700] INFO - prefect.TaskRunner | Task 'testing[0]': Starting task run...
[2021-12-06 08:52:37-0700] INFO - prefect.TaskRunner | Task 'testing[0]': Finished task run for task with final state: 'Success'
[2021-12-06 08:52:37-0700] INFO - prefect.TaskRunner | Task 'testing[1]': Starting task run...
[2021-12-06 08:52:37-0700] INFO - prefect.TaskRunner | Task 'testing[1]': Finished task run for task with final state: 'Success'
[2021-12-06 08:52:37-0700] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded
CPU times: user 113 ms, sys: 5.79 ms, total: 119 ms
Wall time: 5.12 s
<Success: "All reference tasks succeeded.">

However, the second time around I get this error:

%%time
flow.run()

[2021-12-06 08:52:40-0700] INFO - prefect.FlowRunner | Beginning Flow run for 'xarray'
[2021-12-06 08:52:40-0700] INFO - prefect.TaskRunner | Task 'testing': Starting task run...
[2021-12-06 08:52:40-0700] INFO - prefect.TaskRunner | Task 'testing': Finished task run for task with final state: 'Cached'
[2021-12-06 08:52:40-0700] INFO - prefect.TaskRunner | Task 'testing[0]': Starting task run...
[2021-12-06 08:52:40-0700] ERROR - prefect.TaskRunner | Task 'testing[0]': Exception encountered during task execution!
Traceback (most recent call last):
  File "/Users/abanihi/.mambaforge/envs/esds-funnel-dev/lib/python3.9/site-packages/prefect/engine/task_runner.py", line 876, in get_task_run_state
    value = prefect.utilities.executors.run_task_with_timeout(
  File "/Users/abanihi/.mambaforge/envs/esds-funnel-dev/lib/python3.9/site-packages/prefect/utilities/executors.py", line 454, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "/Users/abanihi/.mambaforge/envs/esds-funnel-dev/lib/python3.9/site-packages/prefect/utilities/tasks.py", line 456, in method
    return run_method(self, *args, **kwargs)
  File "/Users/abanihi/.mambaforge/envs/esds-funnel-dev/lib/python3.9/site-packages/prefect/tasks/core/operators.py", line 38, in run
    return task_result[key]
TypeError: 'int' object is not subscriptable
[2021-12-06 08:52:40-0700] INFO - prefect.TaskRunner | Task 'testing[0]': Finished task run for task with final state: 'Failed'
[2021-12-06 08:52:40-0700] INFO - prefect.TaskRunner | Task 'testing[1]': Starting task run...
[2021-12-06 08:52:40-0700] ERROR - prefect.TaskRunner | Task 'testing[1]': Exception encountered during task execution!
Traceback (most recent call last):
  File "/Users/abanihi/.mambaforge/envs/esds-funnel-dev/lib/python3.9/site-packages/prefect/engine/task_runner.py", line 876, in get_task_run_state
    value = prefect.utilities.executors.run_task_with_timeout(
  File "/Users/abanihi/.mambaforge/envs/esds-funnel-dev/lib/python3.9/site-packages/prefect/utilities/executors.py", line 454, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "/Users/abanihi/.mambaforge/envs/esds-funnel-dev/lib/python3.9/site-packages/prefect/utilities/tasks.py", line 456, in method
    return run_method(self, *args, **kwargs)
  File "/Users/abanihi/.mambaforge/envs/esds-funnel-dev/lib/python3.9/site-packages/prefect/tasks/core/operators.py", line 38, in run
    return task_result[key]
TypeError: 'int' object is not subscriptable
[2021-12-06 08:52:40-0700] INFO - prefect.TaskRunner | Task 'testing[1]': Finished task run for task with final state: 'Failed'
[2021-12-06 08:52:40-0700] INFO - prefect.FlowRunner | Flow run FAILED: some reference tasks failed.
CPU times: user 124 ms, sys: 6.22 ms, total: 130 ms
Wall time: 127 ms
<Failed: "Some reference tasks failed.">

[marcoBmota8]

I am having the same issue. I am using Prefect 1.0 and woudl love to see if what I am tryign to achieve can be done on Prefect 2.0.
To solve andersy005 issue I tried to compact both output variables a,b in his case into a dictionary and have a target in the @task decorator. However this only runs fine the first time saving the file as specified by LocalResult and target in the @task decorator.

The secont time it is run it retrieves and error similar as andersy005. Inspecting the saved file it does not contain the dictionary but the last array of the two it shoudl contain.

Why is this?

Is there a way to save the entire dictionary?

[marcoBmota8]

My code would look somethign like this

import prefect
from prefect import task, Flow
import time

@task(target="baz", result=prefect.engine.results.LocalResult(dir="/tmp/testing"), target = 'result.pkl)
def testing():
    time.sleep(5)
dict_result = {'a':a,'b':b}
    return dict_result

with Flow("xarray") as flow:
    result = testing()

[marcoBmota8]

See https://github.com/marcoBmota8/prefect_utils