#3922 [BUG] maptasks with optional output fail to validate with optional interfaces Issue created by hamersaw Describe the bug If

min_success_ratio

is set on a maptask and a subtask fails, Flyte will inject a

nil

value in the

List

output. In the case where the maptask has an

Optional

output, the resulting variables will be wrapped in a

union

to maintain type metadata. This may result in a list similar to:

{
  "o0": [
    {
      "union": 0
    },
    "(empty)",
    {
      "union": 2
    }
  ]
}

When executing launchplans Flyte attempts to validate input data with the input interface. To do this, it infers the literal type from the actual value. The aforementioned maptask will output the type:

collection_type:<
    union_type:<
        variants:<
            simple:NONE 
        > 
        variants:<
            union_type:<
                variants:<
                    simple:INTEGER 
                    structure:<
                        tag:"int" 
                    > 
                >
            >
        >
    >
>

whereas a input value of python type

List[Optional[int]]

will have type:

collection_type:<
    union_type:<
        variants:<
            simple:INTEGER 
            structure:<
                tag:"int" 
            > 
        > 
        variants:<
            simple:NONE 
            structure:<
                tag:"none" 
            > 
        > 
    > 
>

As you can see the integer variant is wrapped in an extra

union

. This results in Flyte throwing an exception that these types are not castable. Expected behavior This, like most things, should just work. Additional context to reproduce

@task
def debug3_foo(x: int) -> Optional[int]:
    if x == 1:
        raise RuntimeError(f"Failed on {x}")

    return x

@workflow
def debug3_sub(x: List[int]) -> List[Optional[int]]:
    return map_task(debug3_foo, min_success_ratio=0.5)(x=x)

@task
def debug3_bar(x: List[Optional[int]]) -> List[Optional[int]]:
    return x

@workflow
def debug3_subb(x: List[Optional[int]]) -> List[Optional[int]]:
    return debug3_bar(x=x)

@reference_launch_plan(
    project="flytesnacks",
    domain="development",
    name="launchplan_reference.debug3_subb",
    version="fjPtoDlcWGVo3ngRWgepOQ==",
)
def debug3_subb_ref(x: List[Optional[int]]) -> List[Optional[int]]:
    ...

@workflow
def debug3(x: List[int]) -> List[Optional[int]]:
    y = debug3_sub(x=x)
    #return debug3_subb(x=y)
    return debug3_subb_ref(x=y)

and run with inputs

[0,1,2]

so that the

1

subtask instance fails when we raise an exception explicitly. Screenshots No response Are you sure this issue hasn't been raised already? ☑︎ Yes Have you read the Code of Conduct? ☑︎ Yes flyteorg/flyte