[BUG] Workflows in conditionals require returned output.

[dansola]

Describe the bug

When a workflow is used in the then part of a conditional, there are failures if the workflow in the conditional doesn't return anything.

The child execution of the workflow actually succeeds, but it's the parent conditional that fails.

You need to return some dummy value in order to get the conditional to work.

Expected behavior

Ideally, like a task, the workflow shouldn't need to return anything to be used in a conditional.

Additional context to reproduce

Broken Example:

import flytekit as fl
from flytekit.core.task import Echo

echo = Echo(name="echo")

@fl.task
def t1():
    print('t1')

@fl.workflow
def wf1():
    t1()

@fl.workflow
def wf_conditional(val: str):
    conditional_wf = (
        fl.conditional("conditional")
        .if_(val != "")
        .then(
            wf1()
        )
        .else_()
        .then(echo())
    )

The conditional fails with

Failed to copy child node outputs from [s3://union-oc-production-demo/metadata/propeller/flytesnacks-development-a28459qp8kb8hvvqvgdw/n1/data/0/n1-n0/0/outputs.pb] to [s3://union-oc-production-demo/metadata/propeller/flytesnacks-development-a28459qp8kb8hvvqvgdw/n1/data/0/outputs.pb]

Mitigated Example:

import flytekit as fl
from flytekit.core.task import Echo

echo = Echo(name="echo", inputs={"dummy_input": bool})

@fl.task
def t1():
    print('t1')

@fl.workflow
def wf1() -> bool:
    t1()
    return True

@fl.workflow
def wf_conditional(val: str):
    conditional_wf = (
        fl.conditional("conditional")
        .if_(val != "")
        .then(
            wf1()
        )
        .else_()
        .then(echo())
    )

Screenshots

No response

Are you sure this issue hasn't been raised already?

• Yes

Have you read the Code of Conduct?

• Yes