#4021 [BUG] Cannot read cache task output returning Optional[str] Issue created by mickjermsurawong-stripe Describe the bug Re-running a cached task returning

Optional[str]

would fail. • Running the task below once is successful and cache is written. • Re-running (without cache overwrite) would fail because deserialized type from data catalog doesn't match the method signature. For example:

@task(cache=True, cache_version="1.0")
def return_optional(task_out: str) -> Optional[str]:
    return task_out

Msg on Flyte console:

Workflow[...] failed. RuntimeExecutionError: max number of system retry attempts [51/50] exhausted. Last known status message: 
Failed to check Catalog for previous results: 
unexpected artifactData: [o0] 
type: [union_type:<variants:<simple:STRING structure:<tag:"str" > > > ] does not match any task output 
type: [union_type:<variants:<simple:STRING > variants:<simple:NONE > > ]

Error from flytepropeller log:

DataCatalog failed to get outputs from artifact dc8dc0fc-a61d-4126-85bd-34b163d49257, 
err: unexpected artifactData: [o0] 
type: [union_type:<variants:<simple:STRING structure:<tag:"str" > > > ] does not match any task output 
type: [union_type:<variants:<simple:STRING > variants:<simple:NONE > > ]

Looking at underlying decoded proto, which is consistent with error message of deserialized type.

1 {
  9 {
    1 {
      1 {
        1 {
          3: "foo"
        }
      }
    }
    2 {
      11 {
        1: "str"
      }
      1: 3
    }
  }
}

Expected behavior Cached task returning optional string (or other types) should be able to get output from the cache. Optional type should be round-trip serializable. Specifically, it seems that the serialized proto data doesn't encode "optionality" of type here. Additional context to reproduce Task

return_optional

will fail after the first successful run caching its output

from flytekit import task, workflow
from typing import Optional

@task(cache=True, cache_version="2.0")
def return_optional(task_out: str) -> Optional[str]:
    return task_out

@workflow
def example() -> Optional[str]:
    return return_optional(task_out="foo")

Screenshots No response Are you sure this issue hasn't been raised already? ☑︎ Yes Have you read the Code of Conduct? ☑︎ Yes flyteorg/flyte