[BUG] Subworkflow chaining does not work when run locally

[d-tw]

Describe the bug

When the order of two subworkflows is controlled via the chaining operator, the order of execution is not applied when the code is run locally (locally as in "via direct invocation of python", not locally as in "via a local cluster").

When the workflow is published to Flyte, the DAG and subsequent execution are correct.

Expected behavior

The order of execution of subworkflows should be identical between local runs and remote execution, especially in the case of unit testing.

Additional context to reproduce

The version of flytekit is 1.5.0.

Sample code:

from flytekit import workflow, task
import logging

logging.basicConfig(level="DEBUG")

logger = logging.getLogger("wf")
logger.setLevel("DEBUG")

@task
def task_1():
    logger.info("task_1")

@task
def task_2():
    logger.info("task_2")

@workflow
def sub_workflow_1():
    task_1()

@workflow
def sub_workflow_2():
    task_2()

@workflow
def parent_workflow():
    s1 = sub_workflow_1()
    s2 = sub_workflow_2()

    s2 >> s1

if __name__ == "__main__":
    parent_workflow()

Output is:

INFO:wf:task_1
INFO:wf:task_2

Expected output is:

INFO:wf:task_2
INFO:wf:task_1

Screenshots

When registered with a Flyte cluster, the DAG is correct.

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

• Yes

Have you read the Code of Conduct?

• Yes

[welcome]

Thank you for opening your first issue here! 🛠

[kumare3]

@d-tw this is an excellent catch. you are right, the order is not really controlled locally, as the current local execution engine is a simple python interpreter and does not re-interpret the code. This is indeed a bug, but the fix might be more involved than a quick fix.
cc @wild-endeavor

I would love to know if this is a blocker in some way?

[d-tw]

@kumare3 Thanks for the response!

Looking at the flytekit code I realised that the local execution is fundamentally different from what happens on Flyte - it doesn’t use the DAG at all.

I hacked together a contextmanager that can be used in unit tests that patches PythonFunctionWorkflow.execute, replacing the execution of the workflow function with code that executes the nodes in DAG order, and binds inputs/outputs. It’s not pretty, and I’m sure there are some utility functions within flytekit that could simplify some of the binding parts, but it seems to work.

I think it brings us closer to on-Flyte behaviour, i.e. workflow code is only executed during compile, but I’m figuring it out as I go along!

So no, it’s not a blocker, but if ever you decide to implement it, I’ll be able to scrap my workaround :)

[eapolinario]

@d-tw , would it be possible for you to share the PR? One concern is how this applies to tasks, but I'm sure we can iterate on that as well once we have some code to handle sub-workflows.

[ericwudayi]

@eapolinario I think this issue is highly correlate to this one? #4080