[Core feature] list support for partials (map_task)

[klekass]

Motivation: Why do you think this is important?

Currently partial functions with lists bound to an argument do not work in conjunction with map_tasks. See the following example:

@task
def demo_task(strings: List[str], num: int) -> str:
    return f"{num}".join(strings)

@workflow
def demo_wf(strings: List[str], nums: List[int]) -> None:
    map_task(functools.partial(demo_task, strings=strings))(num=nums)

This example works only if the length of strings is equal to the length of nums. This limitation seems arbitrary to the user and should either not be supported at all or support lists of different sizes aswell. This feature request aims for the latter.

Goal: What should the final outcome look like, ideally?

Given the example:

@task
def demo_task(strings: List[str], num: int) -> str:
    return f"{num}".join(strings)

@workflow
def demo_wf(strings: List[str], nums: List[int]) -> None:
    map_task(functools.partial(demo_task, strings=strings))(num=nums)

Input: ["a", "b", "c"], [1, 2]
Output: ["a1b1c", "a2b2c"]

Describe alternatives you've considered

As an alternative, the old pattern of input/output preparation tasks can be used, which the partial task aimed to eliminate.

Propose: Link/Inline OR Additional context

No response

Are you sure this issue hasn't been raised already?

• Yes

Have you read the Code of Conduct?

• Yes

[eapolinario]

We don't expose which arguments are partials in flyteidl (so the backend cannot know which ones are already evaluated).