Fix: Make 'flytectl compile' consider launchplans used within workflows

[fg91]

Why are the changes needed?

from flytekit import LaunchPlan, task, workflow

@task
def my_task():
    pass


@workflow
def inner_workflow():
    my_task()


@workflow
def outer_workflow():
    LaunchPlan.get_or_create(inner_workflow, "name_override")()

This workflow can be registered with pyflyte register and executed in a cluster. However, pyflyte --pkgs <module> package + flytectl compile gives the following error:

...
Compiling workflow: wfs.wf.outer_workflow
:( Error Compiling workflow: wfs.wf.outer_workflow
Error: Collected Errors: 1
        Error 0: Code: WorkflowReferenceNotFound, Node Id: start-node, Description: Referenced Workflow [resource_type:LAUNCH_PLAN  name:"name_override"] not found.

What changes were proposed in this pull request?

The error occurs since flytectl compile does not consider all launchplans but only the default launchplan for the currently compiled workflow.

How was this patch tested?

Added a unit test based on the example above.

Note to reviewers: We couldn't find whether the source files for the testdata/*.tgz should be checked in as well - should they?

Check all the applicable boxes

• I updated the documentation accordingly.
• All new and existing tests passed.
• All commits are signed-off.

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 61.11%. Comparing base (d04cf66) to head (0ffe186).
Report is 164 commits behind head on master.

Additional details and impacted files

@@           Coverage Diff           @@
##           master    #5437   +/-   ##
=======================================
  Coverage   61.10%   61.11%           
=======================================
  Files         793      793           
  Lines       51164    51171    +7     
=======================================
+ Hits        31265    31272    +7     
  Misses      17027    17027           
  Partials     2872     2872           

Flag	Coverage Δ
unittests-datacatalog	69.31% <ø> (ø)
unittests-flyteadmin	58.90% <ø> (ø)
unittests-flytecopilot	17.79% <ø> (ø)
unittests-flytectl	68.35% <100.00%> (+0.04%)	⬆️
unittests-flyteidl	79.30% <ø> (ø)
unittests-flyteplugins	61.94% <ø> (ø)
unittests-flytepropeller	57.32% <ø> (ø)
unittests-flytestdlib	65.82% <ø> (ø)

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[fellhorn]

We noticed that this PR still needs some more work as it still fails with LaunchPlan outputs used for other nodes. Here is a minimal example:

from flytekit import LaunchPlan, task, workflow

@task
def my_task(inp: int) -> int:
    return 42 + inp


@workflow
def inner_workflow(inp: int) -> int:
    return my_task(inp=inp)


@workflow
def outer_workflow() -> int:
    res = LaunchPlan.get_or_create(inner_workflow, "name_override", default_inputs={"inp": 3})()
    return my_task(inp=res)

It fails with:

Compiling tasks...

Compiling workflow: wfs.wf.inner_workflow

Compiling workflow: wfs.wf.outer_workflow
:( Error Compiling workflow: wfs.wf.outer_workflow
Error: Collected Errors: 3
        Error 0: Code: ParameterNotBound, Node Id: n1, Description: Parameter not bound [inp].
        Error 1: Code: VariableNameNotFound, Node Id: n0, Description: Variable [inp] not found on node [n0].
        Error 2: Code: VariableNameNotFound, Node Id: n0, Description: Variable [o0] not found on node [n0].

We tried to understand what is missing/different in flytectl compile to how the admin handles workflow registration but have not found the root cause yet. Generally wondering if we need to compile the workflows in a certain order and somehow also compile/populate the LaunchPlans as they currently don't have the input and output infos.

@dav009 do you maybe have some time to look at this with us or give us some hints where to start?

[fg91]

Closing in favour of #5463.