Prefect: Downstream mapped task attempts to run in skipped ifelse branch

[jdamiani27]

In the following flow, the print_item mapped task attempts to run even though the make_list task gets skipped:

from prefect import Flow, Parameter, task
from prefect.tasks.control_flow import ifelse


@task
def is_none(x):
    return x is None


@task
def make_list(): 
    return [3, 2, 1]


@task
def print_item(i):
    print(i)


@task
def none_printer():
    print('Param was none')


with Flow('mapped branch') as flow:
    param = Parameter('param', required=False)
    make_list = make_list()
    ifelse(is_none(param), none_printer(), make_list)
    print_item.map(i=make_list)


state = flow.run()

Output:

[2019-05-23 16:03:27,751] INFO - prefect.FlowRunner | Beginning Flow run for 'mapped branch'
[2019-05-23 16:03:27,752] INFO - prefect.FlowRunner | Starting flow run.
[2019-05-23 16:03:27,756] INFO - prefect.TaskRunner | Task 'param': Starting task run...
[2019-05-23 16:03:27,757] INFO - prefect.TaskRunner | Task 'param': finished task run for task with final state: 'Success'
[2019-05-23 16:03:27,757] INFO - prefect.TaskRunner | Task 'is_none': Starting task run...
[2019-05-23 16:03:27,757] INFO - prefect.TaskRunner | Task 'is_none': finished task run for task with final state: 'Success'
[2019-05-23 16:03:27,758] INFO - prefect.TaskRunner | Task 'CompareValue: "True"': Starting task run...
[2019-05-23 16:03:27,758] INFO - prefect.TaskRunner | Task 'CompareValue: "True"': finished task run for task with final state: 'Success'
[2019-05-23 16:03:27,758] INFO - prefect.TaskRunner | Task 'none_printer': Starting task run...
Param was none
[2019-05-23 16:03:27,759] INFO - prefect.TaskRunner | Task 'none_printer': finished task run for task with final state: 'Success'
[2019-05-23 16:03:27,759] INFO - prefect.TaskRunner | Task 'CompareValue: "False"': Starting task run...
[2019-05-23 16:03:27,759] INFO - prefect.TaskRunner | Task 'CompareValue: "False"': finished task run for task with final state: 'Skipped'
[2019-05-23 16:03:27,760] INFO - prefect.TaskRunner | Task 'make_list': Starting task run...
[2019-05-23 16:03:27,760] INFO - prefect.TaskRunner | Task 'make_list': finished task run for task with final state: 'Skipped'
[2019-05-23 16:03:27,760] INFO - prefect.TaskRunner | Task 'print_item': Starting task run...
[2019-05-23 16:03:27,760] INFO - prefect.TaskRunner | Unexpected error: TypeError("'NoResultType' object does not support indexing",)
[2019-05-23 16:03:27,760] INFO - prefect.TaskRunner | Task 'print_item': finished task run for task with final state: 'Failed'
[2019-05-23 16:03:27,762] INFO - prefect.FlowRunner | Flow run FAILED: some reference tasks failed.

[jdamiani27]

Viz for posterity

[cicdw]

@jdamiani27 great find! I've made a simple PR that addresses this. Once it is merged, you can optionally work off the master branch to avoid this bug. Additionally, we should be cutting a new release early next week which will include this as well!