PrefectHQ / prefect

Prefect is a workflow orchestration tool empowering developers to build, observe, and react to data pipelines
https://prefect.io
Apache License 2.0
15.29k stars 1.5k forks source link

Tasks need `persist_result=True` to persist transaction records #14087

Closed desertaxle closed 1 week ago

desertaxle commented 1 week ago

Tasks must have persist_result=True for cache policies to take effect. Users should not need to explicitly set persist_result=True and Prefect should persists records when necessary.

MRE:

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

@task(cache_policy=INPUTS)
def foo(x):
    return x

@flow(result_storage='gcs-bucket/gcs-test-record-store')
def bar():
    return foo(1, return_state=True), foo(2, return_state=True)

first_state, second_state = bar()
assert first_state.name == "Completed"
assert second_state.name == "Completed"

third_state, fourth_state = bar()
assert third_state.name == "Cached"
assert fourth_state.name == "Cached"

Setting persist_result=True will allow this script to run successfully.

Related to #13800