How does task caching work across flow runs?

[steven-murray]

Bug summary

I have written a fairly simple workflow, and I am expecting the result of the task to be cached, but it is not being cached. Here's an MWE:

from prefect import task, flow
from prefect.futures import wait
from prefect.cache_policies import DEFAULT
from time import sleep

@task(persist_result=True, task_run_name="{year}-{day:>03}", cache_policy=DEFAULT)
def run_long_thing(
    year: int,
    day: int,
) -> str:
    # ... actually some long-running code that returns a Path
    sleep(1.5)
    return f"{year}-{day}"


@flow
def run_mini():
    ydays = [(2016, day) for day in range(50)]
    
    outputs = [] 
    for year, day in ydays:
        outputs.append(run_long_thing.submit(year, day))
    wait(outputs)
    print(outputs)


if __name__ == "__main__":
    run_mini()

When I look at the state of the tasks in the UI, they are shown as Completed, not Cached (and as further confirmation, in the real scenario, the task prints, and these prints are shown in the terminal when executing for a second time with the same settings).

I assume I'm doing something silly (just started with prefect yesterday), but I can't seem to figure out what it is!

Version info (prefect version output)

Version:             3.0.3
API version:         0.8.4
Python version:      3.11.10
Git commit:          d9c51bc3
Built:               Fri, Sep 20, 2024 8:52 AM
OS/Arch:             linux/x86_64
Profile:             default
Server type:         server
Pydantic version:    2.9.2
Integrations:
  prefect-dask:      0.3.1

Additional context

No response

[zzstoatzz]

hey @steven-murray - I converted this to a discussion, since I think its more of a question than a bug - hope thats okay with you

to be clear, here I'm assuming that you're asking about how to cache task results across flow runs, because it appears that you're submitting your task over range(50), which within a single flow run, would give unique task inputs for each invocation (i.e. all cache misses inside one flow run). If I'm mischaracterizing the ask, feel free to let me know and I can update the title here

consider this variant of your example, where importantly:

• i use INPUTS and not DEFAULT cache policy, since with DEFAULT we will miss the cache in subsequent flow runs, because the key is computed from the flow run id
• we can use .map and unmapped to clean up the outputs = []; outputs.append(item) pattern

from time import sleep

from prefect import flow, task, unmapped
from prefect.cache_policies import INPUTS


@task(persist_result=True, task_run_name="{year}-{day:>03}", cache_policy=INPUTS)
def run_long_thing(
    year: int,
    day: int,
) -> str:
    # ... actually some long-running code that returns a Path
    sleep(1.5)
    return f"{year}-{day}"


@flow
def run_mini():
    return run_long_thing.map(unmapped(2016), range(50))


if __name__ == "__main__":
    run_mini()

[zzstoatzz]

here the output using INPUTS (on the second run) is probably what you expected with the DEFAULT policy?

» python repros/15559.py
17:15:56.529 | INFO    | prefect.engine - Created flow run 'elite-rattlesnake' for flow 'run-mini'
....
17:15:57.363 | INFO    | Task run '2016-046' - Finished in state Cached(type=COMPLETED)
17:15:57.365 | INFO    | Task run '2016-049' - Finished in state Cached(type=COMPLETED)
17:15:57.365 | INFO    | Task run '2016-044' - Finished in state Cached(type=COMPLETED)
17:15:57.366 | INFO    | Task run '2016-047' - Finished in state Cached(type=COMPLETED)

[steven-murray]

Thank you @zzstoatzz you are totally correct -- somehow I had missed that the DEFAULT cache policy included the flow ID. Using INPUTS + TASK_SOURCE works as intended.

[zzstoatzz]

sure thing! glad you've got it working