https://flyte.org logo
Join Slack
Powered by
I need to run a workflow from another project thro...
# flyte-support
b
I need to run a workflow from another project through a launch plan, but I have issues with calling that lp with optional inputs inside a dynamic task. Any ideas? Is this expected?
Copy code
@task
def test_task(a: int, b: int | None) -> int:
    return a + b if b is not None else a

@workflow
def test_workflow(a: int, b: int | None) -> int:
    # Imagine this is a launch plan from another project
    return test_task(a=a, b=b)

@dynamic
def test_workflow_dynamic() -> list[int]:
    rs = []
    for a in range(3):
        # b will never be used in this dynamic workflow
        r = test_workflow(a=a, b=None)
        rs.append(r)
    return rs

@workflow
def multiple_workflow_runs() -> list[int]:
    return test_workflow_dynamic()
Gives us
Copy code
AssertionError: Failed to Bind variable b for function flyte.xyz.abc.test_workflow.
Yes, it might make sense to avoid optional altogether and build a new workflow per combination of non-none args, but let’s assume that is a lot of work and I don’t own that code. In general, I don’t see much explicit support for optional args and things happen to work for us.
✅ 1
f
.
Do we not provide a good error
But wait, workflow should also work
b
Correct, Imagine that the workflow is the launchplan. I just wanted to make a minimal thing to test because that also fails.
The whole launchplan point is mostly that it lives in another project and I would prefer to avoid going to to make any fixes.
f
Yoh should be able to call workflow, but sadly it should use default launchplan or bind the node inputs
I think this is a bug/ feature. We are working on even with tasks Cc @thankful-minister-83577 we should create node bindings from default values
Launchplan will actually work
b
I’ll try by referencing a pre-registered 
test_workflow
with a launchplan. I suspect it will not work because that is what caused me to investigate this in my actual setup.
Copy code
@task
def test_task(a: int, b: int | None) -> int:
    return a + b if b is not None else a


# let this be a stub from another package for example
@workflow
def test_workflow(a: int, b: int | None) -> int:
    return test_task(a=a, b=b)


from flytekit.core.launch_plan import reference_launch_plan

# reference launch plan for test_workflow above
@reference_launch_plan(
    project="my_project",
    domain="development",
    name="flyte.abc.def.test_workflow",
    version="abcdef",
)
def test_workflow_lp(a: int, b: int | None = None) -> int: ...


@dynamic
def test_workflow_dynamic() -> list[int]:
    rs = []
    for a in range(3):
        r = test_workflow_lp(a=a, b=None)
        rs.append(r)
    return rs


@workflow
def multiple_workflow_runs() -> list[int]:
    return test_workflow_dynamic()
@freezing-airport-6809 is this what you are expecting would work?
I see:
Copy code
Workflow[redwood:development:flyte.redwood.featurestore.multiple_workflow_runs] failed. RuntimeExecutionError: max number of system retry attempts [51/50] exhausted. Last known status message: 0: [User] malformed dynamic workflow, caused by: Collected Errors: 3
	Error 0: Code: MismatchingTypes, Node Id: dn0, Description: Variable [b] (type [union_type:{variants:{simple:NONE structure:{tag:"none"}}}]) doesn't match expected type [blob:{format:"PythonPickle"} metadata:{fields:{key:"python_class_name" value:{string_value:"int | None"}}}].
	Error 1: Code: MismatchingTypes, Node Id: dn1, Description: Variable [b] (type [union_type:{variants:{simple:NONE structure:{tag:"none"}}}]) doesn't match expected type [blob:{format:"PythonPickle"} metadata:{fields:{key:"python_class_name" value:{string_value:"int | None"}}}].
	Error 2: Code: MismatchingTypes, Node Id: dn2, Description: Variable [b] (type [union_type:{variants:{simple:NONE structure:{tag:"none"}}}]) doesn't match expected type [blob:{format:"PythonPickle"} metadata:{fields:{key:"python_class_name" value:{string_value:"int | None"}}}].

1: 0: [User] malformed dynamic workflow, caused by: Collected Errors: 3
	Error 0: Code: MismatchingTypes, Node Id: dn0, Description: Variable [b] (type [union_type:{variants:{simple:NONE structure:{tag:"none"}}}]) doesn't match expected type [blob:{format:"PythonPickle"} metadata:{fields:{key:"python_class_name" value:{string_value:"int | None"}}}].
	Error 1: Code: MismatchingTypes, Node Id: dn1, Description: Variable [b] (type [union_type:{variants:{simple:NONE structure:{tag:"none"}}}]) doesn't match expected type [blob:{format:"PythonPickle"} metadata:{fields:{key:"python_class_name" value:{string_value:"int | None"}}}].
	Error 2: Code: MismatchingTypes, Node Id: dn2, Description: Variable [b] (type [union_type:{variants:{simple:NONE structure:{tag:"none"}}}]) doesn't match expected type [blob:{format:"PythonPickle"} metadata:{fields:{key:"python_class_name" value:{string_value:"int | None"}}}].
Slightly modifying the above does work:
Copy code
@task
def test_task(a: int, b: int | None = None) -> int:
    return a + b if b is not None else a


# let this be a stub from another package for example
@workflow
def test_workflow(a: int, b: int | None = None) -> int:
    return test_task(a=a, b=b)


from flytekit.core.launch_plan import reference_launch_plan


@reference_launch_plan(
    project="redwood",
    domain="development",
    name="flyte.redwood.featurestore.test_workflow",
    version="multiscale_featurestore-fast-2024-05-13_16-02-37",
)
def test_workflow_lp(a: int, b: int | None = None) -> int: ...


@dynamic
def test_workflow_dynamic() -> list[int]:
    rs = []
    for a in range(3):
        r = test_workflow_lp(a=a)
        rs.append(r)
    return rs


@workflow
def multiple_workflow_runs() -> list[int]:
    return test_workflow_dynamic()
Where this version actually sets a default at the workflow level before building the launchplan, whereas there was no default in the previous example.
I also found that my actual use case was failing when the underlying task in the workflow we reference has types 
int | float | None
. We only need 
float | None
, which fixed things for us. So, in summary, • defaults must be set in the underlying workflow--it is not enough to set defaults in a reference launch plan • 
int | float | None
was giving us issues when resolving that single input.
Apologies for the noise here, but I think we are gtg.
89 Views
Open in Slack
PreviousNext
Powered by