[Flyte][2][Literal Type For Scalar] Binary IDL With MessagePack

[Future-Outlier]

Tracking issue

#5318

Why are the changes needed?

When using Scalar_Binary with a non-empty Tag (i.e., Tag length > 0), we need to return core.SimpleType_STRUCT as the literal type.
This allows for the correct handling of structured types in Flyte tasks, such as the following example:

@task
def dict_task(input: dict) -> dict:
    return input

flytekit branch: flyteorg/flytekit#2757

This change ensures that tasks like the one above can properly return structured data, such as dictionaries.

The following image illustrates the call stack showing where the literalTypeForScalar method is invoked:

What changes were proposed in this pull request?

1. Added logic to handle Scalar_Binary when the Tag is non-empty, returning core.SimpleType_STRUCT as the literal type.
2. tests

How was this patch tested?

Unit tests and remote execution.

from flytekit import task, workflow, ImageSpec

flytekit_hash = "b1bf20c2358db2a189e5e33beb492794d23d5e5b"

flytekit = f"git+https://github.com/flyteorg/flytekit.git@{flytekit_hash}"

image = ImageSpec(
    packages=[flytekit],
    apt_packages=["git"],
    registry="localhost:30000",
)

@task(container_image=image)
def dict_task(input: dict) -> dict:
    return input


# Generate more complex dict inputs with lists and dicts as keys and values
dict_inputs = [
    # Basic key-value combinations with int, str, bool, float
    {1: "a", "key": 2.5, True: False, 3.14: 100},
    {"a": 1, 2: "b", 3.5: True, False: 3.1415},

    # Lists as values, mixed types
    {
        1: [1, "a", 2.5, False],
        "key_list": ["str", 3.14, True, 42],
        True: [False, 2.718, "test"],
    },

    # Nested dicts with basic types
    {
        "nested_dict": {1: 2, "key": "value", True: 3.14, False: "string"},
        3.14: {"pi": 3.14, "e": 2.718, 42: True},
    },

    # Nested lists and dicts as values
    {
        "list_in_dict": [
            {"inner_dict_1": [1, 2.5, "a"], "inner_dict_2": [True, False, 3.14]},
            [1, 2, 3, {"nested_list_dict": [False, "test"]}],
        ]
    },

    # More complex nested structures
    {
        "complex_nested": {
            1: {"nested_dict": {True: [1, "a", 2.5]}},  # Nested dict with list as value
            "string_key": {False: {3.14: {"deep": [1, "deep_value"]}}},  # Deep nesting
        }
    },

    # Dict with list as keys (not typical, but valid in Python if list is hashable, here used only as values)
    {
        "list_of_dicts": [{"a": 1, "b": 2}, {"key1": "value1", "key2": "value2"}],
        10: [{"nested_list": [1, "value", 3.14]}, {"another_list": [True, False]}],
    },

    # More nested combinations of list and dict
    {
        "outer_list": [
            [1, 2, 3],
            {"inner_dict": {"key1": [True, "string", 3.14], "key2": [1, 2.5]}},  # Dict inside list
        ],
        "another_dict": {"key1": {"subkey": [1, 2, "str"]}, "key2": [False, 3.14, "test"]},
    },
]



if __name__ == "__main__":
    import os
    import json
    from flytekit.clis.sdk_in_container import pyflyte
    from click.testing import CliRunner

    runner = CliRunner()
    path = os.path.realpath(__file__)
    for i, input_dict in enumerate(dict_inputs):
        print(f"\n=== Running test {i + 1} ===")
        json_input = json.dumps(input_dict)  # Convert input to JSON string
        result = runner.invoke(pyflyte.main,
                               ["run",
                                path,
                                "dict_task",
                                "--input",
                                json_input])
        print(f"Local Execution {i}: ", result.output)

    for i, input_dict in enumerate(dict_inputs):
        print(f"\n=== Running test {i + 1} ===")
        json_input = json.dumps(input_dict)  # Convert input to JSON string
        result = runner.invoke(pyflyte.main,
                               ["run",
                                "--remote",
                                path,
                                "dict_task",
                                "--input",
                                json_input])
        print("Remote Execution: ", result.output)

Setup process

single binary.
flytekit branch: flyteorg/flytekit#2757

Screenshots

• 8 test cases compiled succesfully

Check all the applicable boxes

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

Related PRs

Docs link

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 36.30%. Comparing base (1502d23) to head (69767c6).
Report is 156 commits behind head on master.

Additional details and impacted files

@@           Coverage Diff           @@
##           master    #5761   +/-   ##
=======================================
  Coverage   36.29%   36.30%           
=======================================
  Files        1305     1305           
  Lines      109997   110000    +3     
=======================================
+ Hits        39927    39932    +5     
+ Misses      65914    65912    -2     
  Partials     4156     4156           

Flag	Coverage Δ
unittests-datacatalog	51.37% <ø> (ø)
unittests-flyteadmin	55.62% <ø> (ø)
unittests-flytecopilot	12.17% <ø> (ø)
unittests-flytectl	62.21% <ø> (ø)
unittests-flyteidl	7.12% <ø> (ø)
unittests-flyteplugins	53.35% <ø> (ø)
unittests-flytepropeller	41.89% <100.00%> (+0.01%)	⬆️
unittests-flytestdlib	55.21% <ø> (ø)

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.

[Future-Outlier]

1. update it to msgpack