De-dupe constant tasks

[bnaul]

Probably not a high priority (and maybe not feasible for some reason I haven't thought of) but I find this behavior a bit irritating:

from prefect import task, Flow
from prefect.engine.executors import DaskExecutor

@task
def add(x, y):
    return x + y

with Flow('flow') as flow:
    for i in range(10):
        add(i, 0)

flow.run(executor=DaskExecutor('localhost:8786'))

Result: the constant 0 task is repeated every time, as is the corresponding log:

[2019-09-18 04:37:19,486] INFO - prefect.TaskRunner | Task '0': Starting task run...
[2019-09-18 04:37:19,486] INFO - prefect.TaskRunner | Task '0': Starting task run...
[2019-09-18 04:37:19,489] INFO - prefect.TaskRunner | Task '0': Starting task run...
[2019-09-18 04:37:19,492] INFO - prefect.TaskRunner | Task '0': finished task run for task with final state: 'Success'
[2019-09-18 04:37:19,493] INFO - prefect.TaskRunner | Task '0': finished task run for task with final state: 'Success'
[2019-09-18 04:37:19,497] INFO - prefect.TaskRunner | Task '0': finished task run for task with final state: 'Success'
...

Is there any kind of simple de-duping scheme that might work within a flow, maybe just for Python literals...?

Edit: possibly kind of a dupe of #891 which was already declined...? 🤷‍♂

[cicdw]

This is a really good call out; the original reason for this behavior is that we needed a place to track the values that are being used as Task inputs, and simply reusing Prefect's Task machinery was the most natural way to do that. In addition, if we see a non-Prefect Task as an input and simply stored its value without further introspection, we might lose dependency relationships such as:

with Flow("example") as flow:
    my_task(inputs=[task_a, task_b])

However, I fully agree that this is not ideal and something we should improve upon. In the meantime, one way of at least de-duping some constants is to wrap them in a Prefect Constant task:

from prefect import task, Flow
from prefect.tasks.core.constants import Constant
from prefect.engine.executors import DaskExecutor

@task
def add(x, y):
    return x + y

with Flow('flow') as flow:
    zero = Constant(0)
    for i in range(10):
        add(i, zero)

flow.run(executor=DaskExecutor('localhost:8786'))

I'll revisit the logic for this and see how we might improve it.

[bnaul]

Makes sense, thanks! One more note: the same behavior would be nice for parameters. Currently:

with Flow('flow') as flow:
    for i in range(10):
        add(i, Parameter('x'))

flow.run(parameters={'x': 1}, executor=DaskExecutor('localhost:8786'))


ValueError: A task with the slug "x" already exists in this flow.