Make it easy to manually query store for result records written on disk

[TheoMathurin]

Result records written on disk can be read by Prefect in the context of tasks where caching is enabled. For a more flexible use of cache in complex workflows, I'd like to find an easy way to directly retrieve results written by previous flow runs given that I know the keys used to store them.

I tried to use the store, through get_result_store, but it looks like only results that have been cached in memory in the current flow run can be accessed. Here is a basic example:

from prefect import task, flow
from prefect.results import get_result_store


@task(cache_key_fn=lambda task, parameters: "my-key")
def my_task():
    return "foobar"


@flow(persist_result=True, log_prints=True)
def pipeline():
    store = get_result_store()
    # print(store.read("my-key").result)  # Will not work on second run
    result = my_task()
    print(store.read("my-key").result)  # Will work on first and second run

    return result


if __name__ == "__main__":
    pipeline()

The traceback when attempting to read from the store before calling the task is similar to this.

The reason is likely the same: in this context the store's metadata_storage is NullFilesystem. If I set store.metadata_storage to None myself, then it works. I don't know if this can have adverse consequences in the overall behaviour of the caching mechanism though. Presumably having NullFileSystem in the flow context is intended behaviour (unlike a transaction context).

As an aside, I noticed that if I set cache_result_in_memory to False at flow-level, the store's LRU cache will nonetheless contain the result of my_task, and store.cache_result_in_memory will automatically be changed to True. When reading the docs, I thought that it meant disabling memory storage for all its tasks (given that there is no caching at flow-level). Setting it to False at task-level does prevent it from being stored in memory.