[Core feature] Flytekit should support `unsafe` mode for types

[Mecoli1219]

Tracking issue

Related to flyteorg/flyte#5319

Why are the changes needed?

Use Case: When the user wants to apply Flytekit to their Python codebase, but annotating the type is too hard or time-consuming, we should support unsafe mode for easy usage.

What changes were proposed in this pull request?

1. I add unsafe kwarg to both task and workflow, and the default value of unsafe is False
2. If the unsafe kwarg is True and the types of inputs or outputs are not specified (which is None), replace the type from None to Optional[Any]. This will trigger FlytePickleTransformer to handle each unspecified type.

How was this patch tested?

Two tests are created to examine the ability of unsafe kwarg.

1. test_unsafe_input_wf_and_task: This test mainly examines the unspecified input types. If the unsafe is False, the workflow or task should throw errors. Otherwise, it should succeed.
2. test_unsafe_wf_and_task: This test mimics the use case mentioned above, where the user might not specify all the types in their codebase. The workflow should succeed if the unsafe kwarg of all tasks and workflow are set to True.

Setup process

Screenshots

Check all the applicable boxes

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

Related PRs

https://github.com/flyteorg/flyte/issues/5319

Docs link

[Mecoli1219]

@pingsutw I created 2 new unit tests, and both of them passed.

[codecov[bot]]

Codecov Report

All modified and coverable lines are covered by tests :white_check_mark:

Project coverage is 74.99%. Comparing base (4bef161) to head (7db0869). Report is 5 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff             @@
##           master    #2419       +/-   ##
===========================================
+ Coverage   46.48%   74.99%   +28.51%     
===========================================
  Files         199      279       +80     
  Lines       20707    24265     +3558     
  Branches     2665     2679       +14     
===========================================
+ Hits         9625    18197     +8572     
+ Misses      10610     5341     -5269     
- Partials      472      727      +255     

:umbrella: View full report in Codecov by Sentry.
:loudspeaker: Have feedback on the report? Share it here.

[Mecoli1219]

NVM, it is still the same problem as mentioned in the comment above 6 months ago.

@task(pickle_untyped=True, container_image=image)
def t1(a) -> int:
    if type(a) == int:
        return a + 1
    return 0

@workflow
def wf1_with_unsafe() -> Tuple[int, int, int]:
    a1 = t1(a=1)
    a2 = t1(a="1")
    a3 = t1(a=None)    #! This will fail
    return a1, a2, a3

Currently, the input with a value None will fail

[Mecoli1219]

First example

@task(pickle_untyped=True, container_image=image)
def t1(a) -> int:
    if type(a) == int:
        return a + 1
    return 0


@workflow
def wf1_with_unsafe() -> Tuple[int, int, int]:
    a1 = t1(a=1)
    a2 = t1(a="1")
    # a3 = t1(a=None)  #! This will fail
    return a1, a2, a2

Second example

@task(pickle_untyped=True, container_image=image)
def t1(a):
    if type(a) == int:
        return a + 1
    return 1


@workflow
def wf1_with_unsafe() -> Tuple[Any, Any, Any]:
    a1 = t1(a=1)
    a2 = t1(a="1")
    # a3 = t1(a=None)  #! This will fail
    return a1, a2, a2

[Mecoli1219]

I revert the support for None type in pickle. I think we could discuss that later and solve that in another PR if finally decided to do that.