Hi all, i have seen this odd error in dynamic task where • Dynamic task does not generate a node ◦ one of the condition is not met • Dynamic task returns a Map type • So we just return some default value Then we are getting hit with this error

Traceback (most recent call last):
  File "/fn/lib/venv/lib/python3.10/site-packages/flytekit/core/base_task.py", line 479, in dispatch_execute
    native_outputs = self.execute(**native_inputs)
  File "/fn/lib/venv/lib/python3.10/site-packages/flytekit/core/python_function_task.py", line 163, in execute
    return self.dynamic_execute(self._task_function, **kwargs)
  File "/fn/lib/venv/lib/python3.10/site-packages/flytekit/core/python_function_task.py", line 268, in dynamic_execute
    return self.compile_into_workflow(ctx, task_function, **kwargs)
  File "/fn/lib/venv/lib/python3.10/site-packages/flytekit/core/python_function_task.py", line 204, in compile_into_workflow
    literals={
  File "/fn/lib/venv/lib/python3.10/site-packages/flytekit/core/python_function_task.py", line 205, in <dictcomp>
    binding.var: binding.binding.to_literal_model() for binding in workflow_spec.template.outputs
  File "/fn/lib/venv/lib/python3.10/site-packages/flytekit/models/literals.py", line 467, in to_literal_model
    return Literal(map=LiteralMap(literals={k: binding.to_literal_model() for k, binding in self.map.bindings}))
  File "/fn/lib/venv/lib/python3.10/site-packages/flytekit/models/literals.py", line 467, in <dictcomp>
    return Literal(map=LiteralMap(literals={k: binding.to_literal_model() for k, binding in self.map.bindings}))
ValueError: too many values to unpack (expected 2)

So we do construct the dictionary in the dynamic task and return it though like

return in_fastq1, in_fastq2, {k: get_empty_flyte_file() for k in in_metrics_keys}

is this expected?

@Jay Ganbat, can you share the signature of the dynamic task?

@BASE_TASK_WORKER.run_dynamic_task_with_args()
def execute_umi_extraction_task(
    in_spec_path: FlyteFile,
    in_fastq1: FlyteFile,
    in_fastq2: FlyteFile,
    in_stage_name: str,
    out_gcs_fastq_dir: str,
    out_gcs_metrics_dir: str,
) -> tuple[FlyteFile, FlyteFile, dict[str, FlyteFile]]:

so if some condition in

in_spec_path

dont exists we just return

metrics_keys = [
        "out_umi_counts",
        "out_fuzzy_matching_umi_counts",
        "out_non_matching_umi_counts",
        "out_umi_conversion_metric",
    ]
return in_fastq1, in_fastq2, {k: get_empty_flyte_file() for k in metrics_keys}

@Jay Ganbat, I was able to find a minimal repro for this. flytekit has a bug right now where dictionaries cannot be returned by @dynamic workflows. I opened https://github.com/flyteorg/flyte/issues/3032 to track this and will prioritize this.

Thank you for taking care of this. We added a dummy task that does nothing before returning and that circumvents the issue

of course. I'll ping here when we have a fix out.