diff --git a/metaflow/plugins/argo/argo_workflows.py b/metaflow/plugins/argo/argo_workflows.py index b033c6d638f..f27454d603d 100644 --- a/metaflow/plugins/argo/argo_workflows.py +++ b/metaflow/plugins/argo/argo_workflows.py @@ -1380,19 +1380,35 @@ def _visit( ) else: # Every other node needs only input-paths - parameters = [ - Parameter("input-paths").value( - compress_list( - [ - "argo-{{workflow.name}}/%s/{{tasks.%s.outputs.parameters.task-id}}" - % (n, self._sanitize(n)) - for n in node.in_funcs - ], - # NOTE: We set zlibmin to infinite because zlib compression for the Argo input-paths breaks template value substitution. - zlibmin=inf, + def _build_input_path(n): + sanitized = self._sanitize(n) + parent_node = self.graph[n] + if self._is_conditional_node(parent_node): + return ( + "argo-{{workflow.name}}/%s/" + "{{=(tasks['%s'].status == 'Succeeded' " + "? tasks['%s'].outputs.parameters['task-id'] : '')}}" + % (n, sanitized, sanitized) ) + else: + return ( + "argo-{{workflow.name}}/%s/" + "{{tasks.%s.outputs.parameters.task-id}}" + % (n, sanitized) + ) + has_conditional_parent = any( + self._is_conditional_node(self.graph[n]) for n in node.in_funcs + ) + if has_conditional_parent: + input_paths_value = ",".join( + _build_input_path(n) for n in node.in_funcs ) - ] + else: + input_paths_value = compress_list( + [_build_input_path(n) for n in node.in_funcs], + zlibmin=inf, + ) + parameters = [Parameter("input-paths").value(input_paths_value)] # NOTE: Due to limitations with Argo Workflows Parameter size we # can not pass arbitrarily large lists of task id's to join tasks. # Instead we ensure that task id's for foreach tasks can be diff --git a/metaflow/plugins/argo/conditional_input_paths.py b/metaflow/plugins/argo/conditional_input_paths.py index fda796d1aac..c3ac8d05576 100644 --- a/metaflow/plugins/argo/conditional_input_paths.py +++ b/metaflow/plugins/argo/conditional_input_paths.py @@ -25,7 +25,16 @@ def generate_input_paths(input_paths, skippable_steps): # strip these out of the list. # all pathspecs of leading steps that executed. - trimmed = [path for path in paths if not "{{" in path] + def _is_resolved_path(path): + if "{{" in path: + return False + parts = path.split("/") + # Paths from skipped conditional steps resolve to "argo-//" + # (empty task-id segment), which must be filtered out. + if len(parts) < 3 or not parts[-1]: + return False + return True + trimmed = [path for path in paths if _is_resolved_path(path)] skippable_steps = [step for step in skippable_steps if step] skippable_step_set = set(skippable_steps)