Refactor multiple input handling.

Fixes common-workflow-lab#54. Implements count-lines7-wf.cwl conformance test.

Author: jmchilton

lib/galaxy/dataset_collections/matching.py

@@ -16,8 +16,10 @@ class CollectionsToMatch( object ):
 
     def __init__( self ):
         self.collections = {}
+        self.uses_ephemeral_collections = False
 
     def add( self, input_name, hdca, subcollection_type=None, linked=True ):
+        self.uses_ephemeral_collections = self.uses_ephemeral_collections or not hasattr( hdca, "hid" )
         self.collections[ input_name ] = bunch.Bunch(
             hdca=hdca,
             subcollection_type=subcollection_type,
@@ -45,6 +47,7 @@ def __init__( self ):
         self.linked_structure = None
         self.unlinked_structures = []
         self.collections = {}
+        self.uses_ephemeral_collections = False
 
     def __attempt_add_to_linked_match( self, input_name, hdca, collection_type_description, subcollection_type ):
         structure = get_structure( hdca, collection_type_description, leaf_subcollection_type=subcollection_type )
@@ -71,14 +74,19 @@ def structure( self ):
 
     @property
     def implicit_inputs( self ):
-        return list( self.collections.items() )
+        if not self.uses_ephemeral_collections:
+            # Consider doing something smarter here.
+            return list( self.collections.items() )
+        else:
+            return []
 
     @staticmethod
     def for_collections( collections_to_match, collection_type_descriptions ):
         if not collections_to_match.has_collections():
             return None
 
         matching_collections = MatchingCollections()
+        matching_collections.uses_ephemeral_collections = collections_to_match.uses_ephemeral_collections
         for input_key, to_match in collections_to_match.items():
             hdca = to_match.hdca
             collection_type_description = collection_type_descriptions.for_collection_type( hdca.collection.collection_type )

lib/galaxy/tools/actions/__init__.py

@@ -560,6 +560,9 @@ def _record_inputs( self, trans, tool, job, incoming, inp_data, inp_dataset_coll
                         reductions[name] = []
                     reductions[name].append(dataset_collection)
 
+                if getattr( dataset_collection, "ephemeral", False ):
+                    dataset_collection = dataset_collection.persistent_object
+
                 # TODO: verify can have multiple with same name, don't want to loose tracability
                 job.add_input_dataset_collection( name, dataset_collection )
 

lib/galaxy/tools/cwl/parser.py

@@ -963,7 +963,7 @@ def to_dict(self, itemwise=True):
             if self.input_type == INPUT_TYPE.FLOAT:
                 as_dict["value"] = "0.0"
             elif self.input_type == INPUT_TYPE.DATA_COLLECTON:
-                as_dict["collection_type"] = "record"
+                as_dict["collection_type"] = self.collection_type
 
         return as_dict
 

lib/galaxy/tools/cwl/representation.py

@@ -130,6 +130,12 @@ def dataset_wrapper_to_file_json(dataset_wrapper):
         elif type_representation.name == "json":
             raw_value = param_dict_value.value
             return json.loads(raw_value)
+        elif type_representation.name == "array":
+            # TODO: generalize to lists of lists and lists of non-files...
+            rval = []
+            for value in param_dict_value:
+                rval.append(dataset_wrapper_to_file_json(value))
+            return rval
         elif type_representation.name == "record":
             rval = dict()  # TODO: THIS NEEDS TO BE ORDERED BUT odict not json serializable!
             for key, value in param_dict_value.items():

lib/galaxy/tools/execute.py

@@ -138,7 +138,7 @@ def create_output_collections( self, trans, history, params ):
 
         structure = self.collection_info.structure
 
-        if hasattr( self.collection_info, "collections" ):
+        if not self.collection_info.uses_ephemeral_collections:
             # params is just one sample tool param execution with parallelized
             # collection replaced with a specific dataset. Need to replace this
             # with the collection and wrap everything up so can evaluate output
@@ -148,7 +148,7 @@ def create_output_collections( self, trans, history, params ):
             collection_names = ["collection %d" % c.hid for c in self.collection_info.collections.values()]
             on_text = on_text_for_names( collection_names )
         else:
-            on_text = "implicitly create collection for inputs"
+            on_text = "implicitly created collection from inputs"
 
         collections = {}
 

lib/galaxy/tools/parameters/basic.py

@@ -1472,6 +1472,9 @@ def get_initial_value( self, trans, other_values ):
                         return hdca
 
     def to_json( self, value, app, use_security ):
+        if getattr( value, "ephemeral", False ):
+            value = value.persistent_object
+
         def single_to_json( value ):
             src = None
             if isinstance( value, dict ) and 'src' in value and 'id' in value:

lib/galaxy/tools/wrappers.py

@@ -384,7 +384,7 @@ def __init__( self, job_working_directory, has_collection, dataset_paths=[], **k
         else:
             self.__input_supplied = True
 
-        if hasattr( has_collection, "name" ):
+        if hasattr( has_collection, "history_content_type" ):
             # It is a HistoryDatasetCollectionAssociation
             collection = has_collection.collection
             self.name = has_collection.name

lib/galaxy/workflow/modules.py

@@ -14,7 +14,6 @@
     web
 )
 from galaxy.dataset_collections import matching
-from galaxy.dataset_collections.structure import leaf, Tree
 from galaxy.exceptions import ToolMissingException
 from galaxy.jobs.actions.post import ActionBox
 from galaxy.model import PostJobAction
@@ -861,25 +860,6 @@ def decode_runtime_state( self, runtime_state ):
         else:
             raise ToolMissingException( "Tool %s missing. Cannot recover runtime state." % self.tool_id )
 
-    def _check_for_scatters( self, step, tool, progress, tool_state ):
-        scatter_collector = ScatterOverCollector(
-            self.app
-        )
-
-        def callback( input, prefixed_name, **kwargs ):
-            replacement = progress.replacement_for_tool_input( step, input, prefixed_name )
-            log.info("replacement for %s is %s" % (prefixed_name, replacement))
-            if replacement:
-                if isinstance(replacement, ScatterOver):
-                    scatter_collector.add_scatter(replacement)
-
-            return NO_REPLACEMENT
-
-        visit_input_values( tool.inputs, tool_state, callback, no_replacement_value=NO_REPLACEMENT )
-
-        # TODO: num slices is bad - what about empty arrays.
-        return None if scatter_collector.num_slices == 0 else scatter_collector
-
     def execute( self, trans, progress, invocation, step ):
         tool = trans.app.toolbox.get_tool( step.tool_id, tool_version=step.tool_version, tool_hash=step.tool_hash )
         tool_state = step.state
@@ -890,10 +870,10 @@ def execute( self, trans, progress, invocation, step ):
         collections_to_match = self._find_collections_to_match( tool, progress, step )
         # Have implicit collections...
         if collections_to_match.has_collections():
-            # Is a MatchingCollections
             collection_info = self.trans.app.dataset_collections_service.match_collections( collections_to_match )
         else:
-            collection_info = self._check_for_scatters( step, tool, progress, make_dict_copy( tool_state.inputs ) )
+            collection_info = None
+
         param_combinations = []
         if collection_info:
             iteration_elements_iter = collection_info.slice_collections()
@@ -1005,16 +985,17 @@ def callback( input, prefixed_name, **kwargs ):
             is_data_param = isinstance( input, DataToolParameter )
             if is_data_param and not input.multiple:
                 data = progress.replacement_for_tool_input( step, input, prefixed_name )
-                if isinstance( data, model.HistoryDatasetCollectionAssociation ):
+                if hasattr( data, "collection" ):
                     collections_to_match.add( prefixed_name, data )
 
             is_data_collection_param = isinstance( input, DataCollectionToolParameter )
             if is_data_collection_param and not input.multiple:
                 data = progress.replacement_for_tool_input( step, input, prefixed_name )
                 history_query = input._history_query( self.trans )
-                subcollection_type_description = history_query.can_map_over( data )
-                if subcollection_type_description:
-                    collections_to_match.add( prefixed_name, data, subcollection_type=subcollection_type_description.collection_type )
+                if hasattr( data, "collection" ):
+                    subcollection_type_description = history_query.can_map_over( data )
+                    if subcollection_type_description:
+                        collections_to_match.add( prefixed_name, data, subcollection_type=subcollection_type_description.collection_type )
 
         visit_input_values( tool.inputs, step.state.inputs, callback )
         return collections_to_match
@@ -1142,57 +1123,28 @@ def load_module_sections( trans ):
     return module_sections
 
 
-class ScatterOverCollector(object):
+class EphemeralCollection(object):
+    """Interface for collecting datasets together in workflows and treating as collections.
 
-    def __init__(self, app):
-        self.inputs_per_name = {}
-        self.num_slices = 0
-        self.app = app
-
-    def add_scatter(self, scatter_over):
-        inputs = scatter_over.inputs
-        self.inputs_per_name[scatter_over.prefixed_name] = inputs
-        if self.num_slices > 0:
-            assert len(inputs) == self.num_slices
-        else:
-            self.num_slices = len(inputs)
-
-    def slice_collections(self):
-        slices = []
-        for i in range(self.num_slices):
-            this_slice = {}
-            for prefixed_name, inputs in self.inputs_per_name.items():
-                this_slice[prefixed_name] = SliceElement(inputs[i], str(i))
-            slices.append(this_slice)
-        return slices
-
-    @property
-    def structure(self):
-        collection_type_descriptions = self.app.dataset_collections_service.collection_type_descriptions
-        collection_type_description = collection_type_descriptions.for_collection_type("list")
-        children = []
-        for input in self.inputs_per_name.values()[0]:
-            children.append((input.element_identifier, leaf))
-
-        return Tree(children, collection_type_description)
-
-    @property
-    def implicit_inputs(self):
-        return []
-
-
-class SliceElement(object):
-
-    def __init__(self, dataset_instance, element_identifier):
-        self.dataset_instance = dataset_instance
-        self.element_identifier = element_identifier
+    These aren't real collections in the database - just datasets groupped together
+    in someway by workflows for passing data around as collections.
+    """
 
+    # Used to distinguish between datasets and collections frequently.
+    ephemeral = True
+    history_content_type = "dataset_collection"
+    name = "Dynamically generated collection"
 
-class ScatterOver(object):
+    def __init__(self, collection, history):
+        self.collection = collection
+        self.history = history
 
-    def __init__(self, prefixed_name, inputs):
-        self.prefixed_name = prefixed_name
-        self.inputs = inputs
+        hdca = model.HistoryDatasetCollectionAssociation(
+            collection=collection,
+            history=history,
+        )
+        hdca.history.add_dataset_collection( hdca )
+        self.persistent_object = hdca
 
 
 class DelayedWorkflowEvaluation(Exception):

lib/galaxy/workflow/run.py

@@ -301,14 +301,67 @@ def replacement_for_tool_input( self, step, input, prefixed_name ):
                     # We've mapped multiple individual inputs to a single parameter,
                     # promote output to a collection.
                     inputs = []
-                    for c in connection:
+                    input_history_content_type = None
+                    input_collection_type = None
+                    for i, c in enumerate(connection):
                         input_from_connection = self.replacement_for_connection( c, is_data=is_data )
+                        input_history_content_type = input_from_connection.history_content_type
+                        if i == 0:
+                            if input_history_content_type == "dataset_collection":
+                                input_collection_type = input_from_connection.collection.collection_type
+                            else:
+                                input_collection_type = None
+                        else:
+                            if input_collection_type is None:
+                                if input_history_content_type != "dataset":
+                                    raise Exception("Cannot map over a combination of datasets and collections.")
+                            else:
+                                if input_history_content_type != "dataset_collection":
+                                    raise Exception("Cannot merge over combinations of datasets and collections.")
+                                elif input_from_connection.collection.collection_type != input_collection_type:
+                                    raise Exception("Cannot merge collections of different collection types.")
+
                         inputs.append(input_from_connection)
 
-                    replacement = modules.ScatterOver(
-                        prefixed_name,
-                        inputs,
+
+                    if input.type == "data_collection":
+                        # TODO: Implement more nested types here...
+                        assert input.collection_types == ["list"], input.collection_types
+
+                    collection = model.DatasetCollection()
+                    # If individual datasets provided (type is None) - premote to a list.
+                    collection.collection_type = input_collection_type or "list"
+                    elements = []
+
+                    next_index = 0
+                    for input in inputs:
+                        if input_collection_type is None:
+                            element = model.DatasetCollectionElement(
+                                element=input,
+                                element_index=next_index,
+                                element_identifier=str(next_index),
+                            )
+                            elements.append(element)
+                            next_index += 1
+                        elif input_collection_type == "list":
+                            for dataset_instance in input.dataset_instances:
+                                element = model.DatasetCollectionElement(
+                                    element=dataset_instance,
+                                    element_index=next_index,
+                                    element_identifier=str(next_index),
+                                )
+                                elements.append(element)
+                                next_index += 1
+                        else:
+                            raise NotImplementedError()
+
+                    collection.elements = elements
+
+                    ephemeral_collection = modules.EphemeralCollection(
+                        collection=collection,
+                        history=self.workflow_invocation.history,
                     )
+                    return ephemeral_collection
 
         return replacement
 

test/unit/tools/test_cwl.py

@@ -184,6 +184,14 @@ def test_workflow_scatter_multiple_input():
     assert len(galaxy_workflow_dict["steps"]) == 3
 
 
+def test_workflow_multiple_input_merge_flattened():
+    version = "v1.0"
+    proxy = workflow_proxy(_cwl_tool_path("%s/count-lines7-wf.cwl" % version))
+
+    galaxy_workflow_dict = proxy.to_dict()
+    assert len(galaxy_workflow_dict["steps"]) == 3
+
+
 def test_load_proxy_simple():
     cat3 = _cwl_tool_path("draft3/cat3-tool.cwl")
     tool_source = get_tool_source(cat3)