#3787 [BUG] Map (array) task cache miss Issue created by bstadlbauer Describe the bug When re-running the same map task with different inputs, the subsequent executions won't be written to cache even though Flyteconsole indicates so. Expected behavior Tasks should also be cached on subsequent runs of the same task Additional context to reproduce Workflow:

from flytekit import task, workflow, map_task


@task(
    cache=True,
    cache_version="1",
)
def my_task(a: int) -> bool:
    return True


@workflow
def my_workflow(inputs: list[int]):
    return map_task(my_task)(a=inputs)

Steps to reproduce: 1. Make sure you've restarted all flyte services (and don't have any previous cached data) 2. Run

my_workflow

with input

inputs=[0]

-> Artifact is written to cache as expected 3. Re-run

my_workflow

with input

inputs=[0]

-> Artifact is read from cache as expected 4. Run

my_workflow

with input

inputs=[1]

-> Task runs as expected. Artifact is NOT written to cache 5. Run

my_workflow

with input

inputs=[1]

-> Task is executed again as there has been no cache-hit Debugging showed that this is happening because the

workItemID

(here) of the of the

Writer

queue (which connects propeller to catalog) does not include any information about the input values (as

Key.String()

does not include the input here) This leads to subsequent runs creating the same

workItemID

, which will prevent them to being added to the writer queue (here) Screenshots No response Are you sure this issue hasn't been raised already? ☑︎ Yes Have you read the Code of Conduct? ☑︎ Yes flyteorg/flyte