From a701a718a0c7675f21a063b225aa31cc54a0d4b4 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Wed, 20 Jun 2018 20:34:22 -0400 Subject: [PATCH 1/4] More correct, more efficient collection operations. We were copying datasets and not assigning them new HIDs in some cases (the newer collection operations apply_rules, tag_from_file were doing the right thing here but it looks like we didn't go back and fix older operations). With the uniform handling of dataset copying in place - I used the much more efficient history.add_datasets instead of history.add_dataset. This also optimizes the actual HDA copying by eliminating some extra flushes. hda.copy() would flush to get an HDA before copying metadata but this is only needed if there are FileParameter metadata types (e.g. bams) so I made that flush conditional as well as the final one. So for most datatypes - this eliminates two extra flushes per dataset copied. Rebased with fix for issue pointed out by @nsoranzo - cleaned up _add_datasets_to_history to just consume values directly. Using itervalues since these methods have performance issues and these lists can be quite large. --- lib/galaxy/model/__init__.py | 24 ++++++++--- lib/galaxy/model/metadata.py | 8 ++++ lib/galaxy/tools/__init__.py | 81 ++++++++++++++++++++++++------------ 3 files changed, 80 insertions(+), 33 deletions(-) diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index dc0f76f6fa9..6adefb9fafb 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -2164,6 +2164,10 @@ class DatasetInstance(object): self._metadata_collection = galaxy.model.metadata.MetadataCollection(self) return self._metadata_collection + @property + def set_metadata_requires_flush(self): + return self.metadata.requires_dataset_id + def set_metadata(self, bunch): # Needs to accept a MetadataCollection, a bunch, or a dict self._metadata = self.metadata.make_dict_copy(bunch) @@ -2554,7 +2558,7 @@ class HistoryDatasetAssociation(DatasetInstance, HasTags, Dictifiable, UsesAnnot self.version = self.version + 1 if self.version else 1 session.add(past_hda) - def copy(self, parent_id=None, copy_tags=None): + def copy(self, parent_id=None, copy_tags=None, force_flush=True): """ Create a copy of this HDA. """ @@ -2574,15 +2578,23 @@ class HistoryDatasetAssociation(DatasetInstance, HasTags, Dictifiable, UsesAnnot # update init non-keywords as well hda.purged = self.purged hda.copy_tags_to(copy_tags) - object_session(self).add(hda) - object_session(self).flush() + # This next line seems unneeded. -John hda.set_size() - # Need to set after flushed, as MetadataFiles require dataset.id + object_session(self).add(hda) + flushed = False + # May need to set after flushed, as MetadataFiles require dataset.id + if hda.set_metadata_requires_flush: + object_session(self).flush() + flushed = True hda.metadata = self.metadata + # In some instances peek relies on dataset_id, i.e. gmaj.zip for viewing MAFs if not self.datatype.copy_safe_peek: - # In some instances peek relies on dataset_id, i.e. gmaj.zip for viewing MAFs + if not flushed: + object_session(self).flush() + hda.set_peek() - object_session(self).flush() + if force_flush: + object_session(self).flush() return hda def copy_tags_to(self, copy_tags=None): diff --git a/lib/galaxy/model/metadata.py b/lib/galaxy/model/metadata.py index 817a1a809e9..368ff7a8546 100644 --- a/lib/galaxy/model/metadata.py +++ b/lib/galaxy/model/metadata.py @@ -144,6 +144,14 @@ class MetadataCollection(object): rval[key] = self.spec[key].param.make_copy(value, target_context=self, source_context=to_copy) return rval + @property + def requires_dataset_id(self): + for key in self.spec: + if isinstance(self.spec[key].param, FileParameter): + return True + + return False + def from_JSON_dict(self, filename=None, path_rewriter=None, json_dict=None): dataset = self.parent if filename is not None: diff --git a/lib/galaxy/tools/__init__.py b/lib/galaxy/tools/__init__.py index 6ecb0adeee6..93da8e8c027 100755 --- a/lib/galaxy/tools/__init__.py +++ b/lib/galaxy/tools/__init__.py @@ -16,7 +16,7 @@ from xml.etree import ElementTree import packaging.version from mako.template import Template from paste import httpexceptions -from six import string_types +from six import itervalues, string_types from six.moves.urllib.parse import unquote_plus import tool_shed.util.repository_util as repository_util @@ -2378,6 +2378,15 @@ class DatabaseOperationTool(Tool): map(check_dataset_instance, input_dataset_collection.dataset_instances) + def _add_datasets_to_history(self, history, elements): + datasets = [] + for element_object in elements: + if getattr(element_object, "history_content_type", None) == "dataset": + datasets.append(element_object) + + if datasets: + history.add_datasets(self.sa_session, datasets, set_hid=True) + def produce_outputs(self, trans, out_data, output_collections, incoming, history): return self._outputs_dict() @@ -2400,10 +2409,7 @@ class UnzipCollectionTool(DatabaseOperationTool): assert collection.collection_type == "paired" forward_o, reverse_o = collection.dataset_instances forward, reverse = forward_o.copy(copy_tags=tags), reverse_o.copy(copy_tags=tags) - - # TODO: rename... - history.add_dataset(forward, set_hid=True) - history.add_dataset(reverse, set_hid=True) + self._add_datasets_to_history(history, [forward, reverse]) out_data["forward"] = forward out_data["reverse"] = reverse @@ -2420,9 +2426,7 @@ class ZipCollectionTool(DatabaseOperationTool): new_elements = odict() new_elements["forward"] = forward new_elements["reverse"] = reverse - history.add_dataset(forward, set_hid=False) - history.add_dataset(reverse, set_hid=False) - + self._add_datasets_to_history(history, [forward, reverse]) output_collections.create_collection( next(iter(self.outputs.values())), "output", elements=new_elements ) @@ -2500,11 +2504,14 @@ class MergeCollectionTool(DatabaseOperationTool): # Don't copy until we know everything is fine and we have the structure of the list ready to go. new_elements = odict() for key, value in new_element_structure.items(): - copied_value = value.copy() - if getattr(copied_value, "history_content_type", None) == "dataset": - history.add_dataset(copied_value, set_hid=False) + if getattr(value, "history_content_type", None) == "dataset": + copied_value = value.copy(force_flush=False) + copied_value.visible = False + else: + copied_value = value.copy() new_elements[key] = copied_value + self._add_datasets_to_history(history, itervalues(new_elements)) output_collections.create_collection( next(iter(self.outputs.values())), "output", elements=new_elements ) @@ -2517,8 +2524,11 @@ class FilterDatasetsTool(DatabaseOperationTool): for dce in elements_to_copy: element_identifier = dce.element_identifier copied_value = dce.element_object.copy() - if getattr(copied_value, "history_content_type", None) == "dataset": - history.add_dataset(copied_value, set_hid=False) + if getattr(dce.element_object, "history_content_type", None) == "dataset": + copied_value = dce.element_object.copy(force_flush=False) + copied_value.visible = False + else: + copied_value = dce.element_object.copy() new_elements[element_identifier] = copied_value return new_elements @@ -2550,7 +2560,7 @@ class FilterDatasetsTool(DatabaseOperationTool): elements_to_copy.append(element) new_elements = self._get_new_elements(history=history, elements_to_copy=elements_to_copy) - + self._add_datasets_to_history(history, itervalues(new_elements)) output_collections.create_collection( next(iter(self.outputs.values())), "output", @@ -2581,6 +2591,7 @@ class FlattenTool(DatabaseOperationTool): hdca = incoming["input"] join_identifier = incoming["join_identifier"] new_elements = odict() + copied_datasets = [] def add_elements(collection, prefix=""): for dce in collection.elements: @@ -2590,10 +2601,13 @@ class FlattenTool(DatabaseOperationTool): if dce.is_collection: add_elements(dce_object, prefix=identifier) else: - copied_dataset = dce_object.copy() - history.add_dataset(copied_dataset, set_hid=False) + copied_dataset = dce_object.copy(force_flush=False) + copied_dataset.visible = False new_elements[identifier] = copied_dataset + copied_datasets.append(copied_dataset) + add_elements(hdca.collection) + self._add_datasets_to_history(history, copied_datasets) output_collections.create_collection( next(iter(self.outputs.values())), "output", elements=new_elements ) @@ -2629,9 +2643,11 @@ class SortTool(DatabaseOperationTool): for dce in sorted_elements: dce_object = dce.element_object - copied_dataset = dce_object.copy() - history.add_dataset(copied_dataset, set_hid=False) + copied_dataset = dce_object.copy(force_flush=False) + copied_dataset.visible = False new_elements[dce.element_identifier] = copied_dataset + + self._add_datasets_to_history(history, itervalues(new_elements)) output_collections.create_collection( next(iter(self.outputs.values())), "output", elements=new_elements ) @@ -2651,9 +2667,11 @@ class RelabelFromFileTool(DatabaseOperationTool): new_label = new_label.strip() if new_label in new_elements: raise Exception("New identifier [%s] appears twice in resulting collection, these values must be unique." % new_label) - copied_value = dce_object.copy() - if getattr(copied_value, "history_content_type", None) == "dataset": - history.add_dataset(copied_value, set_hid=False) + if getattr(dce_object, "history_content_type", None) == "dataset": + copied_value = dce_object.copy(force_flush=False) + copied_value.visible = False + else: + copied_value = dce_object.copy() new_elements[new_label] = copied_value new_labels_path = new_labels_dataset_assoc.file_name @@ -2681,6 +2699,7 @@ class RelabelFromFileTool(DatabaseOperationTool): for key in new_elements.keys(): if not re.match("^[\w\-_]+$", key): raise Exception("Invalid new colleciton identifier [%s]" % key) + self._add_datasets_to_history(history, itervalues(new_elements)) output_collections.create_collection( next(iter(self.outputs.values())), "output", elements=new_elements ) @@ -2693,17 +2712,18 @@ class ApplyRulesTool(DatabaseOperationTool): log.info(incoming) hdca = incoming["input"] rule_set = RuleSet(incoming["rules"]) + copied_datasets = [] def copy_dataset(dataset): - copied_dataset = dataset.copy() + copied_dataset = dataset.copy(force_flush=False) copied_dataset.visible = False - history.add_dataset(copied_dataset, set_hid=True) + copied_datasets.append(copied_dataset) return copied_dataset new_elements = self.app.dataset_collections_service.apply_rules( hdca, rule_set, copy_dataset ) - log.info(new_elements) + self._add_datasets_to_history(history, copied_datasets) output_collections.create_collection( next(iter(self.outputs.values())), "output", collection_type=rule_set.collection_type, elements=new_elements ) @@ -2718,13 +2738,14 @@ class TagFromFileTool(DatabaseOperationTool): new_tags_dataset_assoc = incoming["tags"] new_elements = odict() tags_manager = GalaxyTagManager(trans.app.model.context) + new_datasets = [] def add_copied_value_to_new_elements(new_tags_dict, dce): if getattr(dce.element_object, "history_content_type", None) == "dataset": - copied_value = dce.element_object.copy() + copied_value = dce.element_object.copy(force_flush=False) # copy should never be visible, since part of a collection copied_value.visble = False - history.add_dataset(copied_value, copied_value, set_hid=False) + new_datasets.append(copied_value) new_tags = new_tags_dict.get(dce.element_identifier) if new_tags: if how in ('add', 'remove') and dce.element_object.tags: @@ -2763,6 +2784,7 @@ class TagFromFileTool(DatabaseOperationTool): new_tags_dict = {item[0]: item[1:] for item in source_new_tags} for i, dce in enumerate(hdca.collection.elements): add_copied_value_to_new_elements(new_tags_dict, dce) + self._add_datasets_to_history(history, new_datasets) output_collections.create_collection( next(iter(self.outputs.values())), "output", elements=new_elements ) @@ -2791,13 +2813,18 @@ class FilterFromFileTool(DatabaseOperationTool): copied_value = dce_object.copy() if getattr(copied_value, "history_content_type", None) == "dataset": - history.add_dataset(copied_value, set_hid=False) + copied_value = dce_object.copy(force_flush=False) + copied_value.visible = False + else: + copied_value = dce_object.copy() if passes_filter: filtered_elements[element_identifier] = copied_value else: discarded_elements[element_identifier] = copied_value + self._add_datasets_to_history(history, itervalues(filtered_elements)) + self._add_datasets_to_history(history, itervalues(discarded_elements)) output_collections.create_collection( self.outputs["output_filtered"], "output_filtered", elements=filtered_elements ) From db5b368668265a5fcbd147d9db76cc64156bb6f7 Mon Sep 17 00:00:00 2001 From: mvdbeek Date: Thu, 21 Jun 2018 13:12:33 +0200 Subject: [PATCH 2/4] Don't rely on the order of history contents in test_map_over_collection_structured_like --- test/api/test_tools.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/test/api/test_tools.py b/test/api/test_tools.py index 3036a59542e..f7b2f55fdf0 100644 --- a/test/api/test_tools.py +++ b/test/api/test_tools.py @@ -1464,14 +1464,14 @@ class ToolsTestCase(api.ApiTestCase): "input": {'values': [dict(src="hdca", id=hdca_id)]}, "how|filter_source": {'batch': True, 'values': [dict(src="hdca", id=hdca_id)]} } - self._run("__FILTER_FROM_FILE__", history_id, inputs, assert_ok=True) + implicit_collections = self._run("__FILTER_FROM_FILE__", history_id, inputs, assert_ok=True)['implicit_collections'] + discarded_collection, filtered_collection = implicit_collections self.dataset_populator.wait_for_history(history_id, assert_ok=True) history_contents = self.dataset_populator._get_contents_request(history_id).json() # We should have a final collection count of 3 (2 nested collections, plus the input collection) new_collections = len([c for c in history_contents if c['history_content_type'] == 'dataset_collection']) - 1 assert new_collections == 2, "Expected to generate 4 new, filtered collections, but got %d collections" % new_collections - filtered_collection = history_contents[7] - assert filtered_collection['collection_type'] == 'list:list', filtered_collection + assert filtered_collection['collection_type'] == discarded_collection['collection_type'] == 'list:list', filtered_collection collection_details = self.dataset_populator.get_history_collection_details(history_id, hid=filtered_collection['hid']) assert collection_details['element_count'] == 2 first_collection_level = collection_details['elements'][0] From a10aab8f4f59d12c0d97fd45f1625576287e4a48 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Fri, 22 Jun 2018 13:03:46 -0400 Subject: [PATCH 3/4] Remove copy-paste mistakes. --- lib/galaxy/tools/__init__.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/lib/galaxy/tools/__init__.py b/lib/galaxy/tools/__init__.py index 93da8e8c027..6830d632271 100755 --- a/lib/galaxy/tools/__init__.py +++ b/lib/galaxy/tools/__init__.py @@ -2523,7 +2523,6 @@ class FilterDatasetsTool(DatabaseOperationTool): new_elements = odict() for dce in elements_to_copy: element_identifier = dce.element_identifier - copied_value = dce.element_object.copy() if getattr(dce.element_object, "history_content_type", None) == "dataset": copied_value = dce.element_object.copy(force_flush=False) copied_value.visible = False @@ -2811,7 +2810,6 @@ class FilterFromFileTool(DatabaseOperationTool): in_filter_file = element_identifier in filtered_identifiers passes_filter = in_filter_file if how_filter == "remove_if_absent" else not in_filter_file - copied_value = dce_object.copy() if getattr(copied_value, "history_content_type", None) == "dataset": copied_value = dce_object.copy(force_flush=False) copied_value.visible = False From db036c6f51e53b537d6b88e872c2ecae74b72b44 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Fri, 22 Jun 2018 14:17:31 -0400 Subject: [PATCH 4/4] Another typo fix. --- lib/galaxy/tools/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/galaxy/tools/__init__.py b/lib/galaxy/tools/__init__.py index 6830d632271..77eb22e0e85 100755 --- a/lib/galaxy/tools/__init__.py +++ b/lib/galaxy/tools/__init__.py @@ -2810,7 +2810,7 @@ class FilterFromFileTool(DatabaseOperationTool): in_filter_file = element_identifier in filtered_identifiers passes_filter = in_filter_file if how_filter == "remove_if_absent" else not in_filter_file - if getattr(copied_value, "history_content_type", None) == "dataset": + if getattr(dce_object, "history_content_type", None) == "dataset": copied_value = dce_object.copy(force_flush=False) copied_value.visible = False else: