I have a flow with the below structure that is resulting in a mess of a dependency graph that mixes dependencies between mapped instances of each task where for example: task3-1 will be dependent on task2-0. This only seems to be introduced when I add the

wait_for

parameter to my third mapped task. What am I missing here?

from prefect import flow, task

@task
def task0(input):
    pass

@task
def task1(input, task0_input):
    pass

@task
def task2(input, task0_input):
    pass

@task
def task3(input, task0_input):
    pass

@flow
def map_flow(input):
    task0_complete = task0.map(input)
    task1_complete = task1.map(input,task0_complete)
    task2_complete = task2.map(input,task0_complete)
    task3_complete = task3.map(input,task0_complete,wait_for=[task1_complete,task2_complete])

map_flow([0,1])

hi @Daniel Lomartra - based on your code, it does make sense to me that your dep graph would be strange when you pass in

task0_complete

to everything, you're telling prefect that there's a data dependency there, so each task0 instance will point into every downstream task instance if you add some sleeps, its a bit easier to see what is your actual use case?

Is it okay if I drop you a DM so I can show my actual code? It is difficult to explain the use case in only general terms.