diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index be537bed3df..db92f3f64e8 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -40,6 +40,7 @@ from sqlalchemy.orm import ( joinedload, object_session, Query, + reconstructor, ) from sqlalchemy.schema import UniqueConstraint @@ -1757,11 +1758,33 @@ class History(HasTags, Dictifiable, UsesAnnotations, HasName, RepresentById): self.datasets = [] self.galaxy_sessions = [] self.tags = [] + # Objects to eventually add to history + self._pending_additions = [] + + @reconstructor + def init_on_load(self): + # Restores properties that are not tracked in the database + self._pending_additions = [] + + def stage_addition(self, items): + history_id = self.id + for item in listify(items): + if history_id: + item.history_id = history_id + else: + item.history = self + self._pending_additions.append(item) @property def empty(self): return self.hid_counter == 1 + def add_pending_datasets(self, set_output_hid=True): + # These are assumed to be either copies of existing datasets or new, empty datasets, + # so we don't need to set the quota. + self.add_datasets(object_session(self), self._pending_additions, set_hid=set_output_hid, quota=False, flush=False) + self._pending_additions = [] + def _next_hid(self, n=1): # this is overriden in mapping.py db_next_hid() method if len(self.datasets) == 0: diff --git a/lib/galaxy/tools/__init__.py b/lib/galaxy/tools/__init__.py index 33bc0df6993..f6517f45f80 100644 --- a/lib/galaxy/tools/__init__.py +++ b/lib/galaxy/tools/__init__.py @@ -1646,9 +1646,8 @@ class Tool(Dictifiable): ) job = rval[0] out_data = rval[1] - if len(rval) == 4: - execution_slice.datasets_to_persist = rval[2] - execution_slice.history = rval[3] + if len(rval) > 2: + execution_slice.history = rval[2] except (webob.exc.HTTPFound, exceptions.MessageException) as e: # if it's a webob redirect exception, pass it up the stack raise e @@ -2777,14 +2776,10 @@ class DatabaseOperationTool(Tool): check_dataset_state(state) def _add_datasets_to_history(self, history, elements, datasets_visible=False): - datasets = [] for element_object in elements: if getattr(element_object, "history_content_type", None) == "dataset": element_object.visible = datasets_visible - datasets.append(element_object) - - if datasets: - history.add_datasets(self.sa_session, datasets, set_hid=True) + history.stage_addition(element_object) def produce_outputs(self, trans, out_data, output_collections, incoming, history, **kwds): return self._outputs_dict() @@ -3156,25 +3151,23 @@ class ApplyRulesTool(DatabaseOperationTool): def produce_outputs(self, trans, out_data, output_collections, incoming, history, tag_handler, **kwds): hdca = incoming["input"] rule_set = RuleSet(incoming["rules"]) - datasets_to_persist = [] + copied_datasets = [] def copy_dataset(dataset, tags): copied_dataset = dataset.copy(flush=False) if tags is not None: tag_handler.set_tags_from_list(trans.get_user(), copied_dataset, tags, flush=False) copied_dataset.history_id = history.id - datasets_to_persist.append(copied_dataset) + copied_datasets.append(copied_dataset) return copied_dataset new_elements = self.app.dataset_collections_service.apply_rules( hdca, rule_set, copy_dataset ) - hdca = output_collections.create_collection( - next(iter(self.outputs.values())), "output", collection_type=rule_set.collection_type, elements=new_elements, flush=False, set_hid=False, + 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, ) - if hdca: - datasets_to_persist.append(hdca) - return datasets_to_persist class TagFromFileTool(DatabaseOperationTool): @@ -3270,10 +3263,10 @@ class FilterFromFileTool(DatabaseOperationTool): discarded_elements[element_identifier] = copied_value self._add_datasets_to_history(history, filtered_elements.values()) - self._add_datasets_to_history(history, discarded_elements.values()) output_collections.create_collection( self.outputs["output_filtered"], "output_filtered", elements=filtered_elements ) + self._add_datasets_to_history(history, discarded_elements.values()) output_collections.create_collection( self.outputs["output_discarded"], "output_discarded", elements=discarded_elements ) diff --git a/lib/galaxy/tools/actions/__init__.py b/lib/galaxy/tools/actions/__init__.py index 6d517914c78..138c503268d 100644 --- a/lib/galaxy/tools/actions/__init__.py +++ b/lib/galaxy/tools/actions/__init__.py @@ -459,7 +459,6 @@ class DefaultToolAction: # Flush all datasets at once. return data - datasets_to_persist = [] for name, output in tool.outputs.items(): if not filter_output(output, incoming): handle_output_timer = ExecutionTimer() @@ -496,7 +495,7 @@ class DefaultToolAction: effective_output_name = output_part_def.effective_output_name element = handle_output(effective_output_name, output_part_def.output_def, hidden=True) - datasets_to_persist.append(element) + history.stage_addition(element) # TODO: this shouldn't exist in the top-level of the history at all # but for now we are still working around that by hiding the contents # there. @@ -513,17 +512,13 @@ class DefaultToolAction: element_kwds = dict(elements=collections_manager.ELEMENTS_UNINITIALIZED) else: element_kwds = dict(element_identifiers=element_identifiers) - hdca = output_collections.create_collection( + output_collections.create_collection( output=output, name=name, - set_hid=True if flush_job else False, - flush=flush_job, completed_job=completed_job, **element_kwds ) - if hdca: - datasets_to_persist.append(hdca) - log.info(f"Handled collection output named {name} for tool {tool.id} {handle_output_timer}") + log.info("Handled collection output named {} for tool {} {}".format(name, tool.id, handle_output_timer)) else: handle_output(name, output) log.info(f"Handled output named {name} for tool {tool.id} {handle_output_timer}") @@ -535,7 +530,7 @@ class DefaultToolAction: # Add all the top-level (non-child) datasets to the history unless otherwise specified for name, data in out_data.items(): if name not in child_dataset_names and name not in incoming: # don't add children; or already existing datasets, i.e. async created - datasets_to_persist.append(data) + history.stage_addition(data) # Add all the children to their parents for parent_name, child_name in parent_to_child_pairs: @@ -560,14 +555,13 @@ class DefaultToolAction: if app.config.track_jobs_in_database and rerun_remap_job_id is not None: # We need a flush here and get hids in order to rewrite jobs parameter, # but remapping jobs should only affect single jobs anyway, so this is not too costly. + history.add_pending_datasets(set_output_hid=set_output_hid) trans.sa_session.flush() - history.add_datasets(trans.sa_session, datasets_to_persist, set_hid=set_output_hid, quota=False, flush=False) self._remap_job_on_rerun(trans=trans, galaxy_session=galaxy_session, rerun_remap_job_id=rerun_remap_job_id, current_job=job, out_data=out_data) - datasets_to_persist = [] log.info(f"Setup for job {job.log_str()} complete, ready to be enqueued {job_setup_timer}") # Some tools are not really executable, but jobs are still created for them ( for record keeping ). @@ -593,8 +587,7 @@ class DefaultToolAction: else: if flush_job: # Set HID and add to history. - # This is brand new and certainly empty so don't worry about quota. - history.add_datasets(trans.sa_session, datasets_to_persist, set_hid=set_output_hid, quota=False, flush=False) + history.add_pending_datasets(set_output_hid=set_output_hid) job_flush_timer = ExecutionTimer() trans.sa_session.flush() log.info(f"Flushed transaction for job {job.log_str()} {job_flush_timer}") @@ -602,7 +595,7 @@ class DefaultToolAction: # Dispatch to a job handler. enqueue() is responsible for flushing the job app.job_manager.enqueue(job, tool=tool) trans.log_event("Added job to the job queue, id: %s" % str(job.id), tool_id=job.tool_id) - return job, out_data, datasets_to_persist, history + return job, out_data, history def _remap_job_on_rerun(self, trans, galaxy_session, rerun_remap_job_id, current_job, out_data): """ @@ -825,7 +818,7 @@ class OutputCollections: self.out_collection_instances = {} self.tags = tags - def create_collection(self, output, name, collection_type=None, set_hid=True, flush=True, completed_job=None, **element_kwds): + def create_collection(self, output, name, collection_type=None, completed_job=None, **element_kwds): input_collections = self.input_collections collections_manager = self.trans.app.dataset_collections_service collection_type = collection_type or output.structure.collection_type @@ -904,16 +897,16 @@ class OutputCollections: collection_type=collection_type, trusted_identifiers=True, tags=self.tags, - set_hid=set_hid, - flush=flush, + set_hid=False, + flush=False, completed_job=completed_job, output_name=name, **element_kwds ) # name here is name of the output element - not name # of the hdca. + self.history.stage_addition(hdca) self.out_collection_instances[name] = hdca - return hdca def on_text_for_names(input_names): diff --git a/lib/galaxy/tools/actions/model_operations.py b/lib/galaxy/tools/actions/model_operations.py index 81f545240ca..f45dd4c6baa 100644 --- a/lib/galaxy/tools/actions/model_operations.py +++ b/lib/galaxy/tools/actions/model_operations.py @@ -57,7 +57,7 @@ class ModelOperationToolAction(DefaultToolAction): # Create job. # job, galaxy_session = self._new_job_for_session(trans, tool, history) - datasets_to_persist = self._produce_outputs(trans, tool, out_data, output_collections, incoming=incoming, history=history, tags=preserved_tags) + self._produce_outputs(trans, tool, out_data, output_collections, incoming=incoming, history=history, tags=preserved_tags) self._record_inputs(trans, tool, job, incoming, inp_data, inp_dataset_collections) self._record_outputs(job, out_data, output_collections) job.state = job.states.OK @@ -67,11 +67,11 @@ class ModelOperationToolAction(DefaultToolAction): # trans.app.job_manager.job_queue.put( job.id, tool.id ) # trans.log_event( "Added database job action to the job queue, id: %s" % str(job.id), tool_id=job.tool_id ) log.info("Calling produce_outputs, tool is %s" % tool) - return job, out_data, datasets_to_persist, history + return job, out_data, history def _produce_outputs(self, trans, tool, out_data, output_collections, incoming, history, tags): tag_handler = trans.app.tag_handler.create_tag_handler_session() - datasets_to_persist = tool.produce_outputs(trans, out_data, output_collections, incoming, history=history, tags=tags, tag_handler=tag_handler) + tool.produce_outputs(trans, out_data, output_collections, incoming, history=history, tags=tags, tag_handler=tag_handler) mapped_over_elements = output_collections.dataset_collection_elements if mapped_over_elements: for name, value in out_data.items(): @@ -80,4 +80,3 @@ class ModelOperationToolAction(DefaultToolAction): mapped_over_elements[name].hda = value trans.sa_session.add_all(out_data.values()) - return datasets_to_persist diff --git a/lib/galaxy/tools/execute.py b/lib/galaxy/tools/execute.py index 86008f91544..c8f0a1cff3c 100644 --- a/lib/galaxy/tools/execute.py +++ b/lib/galaxy/tools/execute.py @@ -90,7 +90,7 @@ def execute(trans, tool, mapping_params, history, rerun_remap_job_id=None, colle jobs_executed = 0 has_remaining_jobs = False - datasets_to_persist = [] + execution_slice = None for i, execution_slice in enumerate(execution_tracker.new_execution_slices()): if max_num_jobs and jobs_executed >= max_num_jobs: @@ -100,12 +100,10 @@ def execute(trans, tool, mapping_params, history, rerun_remap_job_id=None, colle execute_single_job(execution_slice, completed_jobs[i]) history = execution_slice.history or history jobs_executed += 1 - if execution_slice.datasets_to_persist: - datasets_to_persist.extend(execution_slice.datasets_to_persist) - if datasets_to_persist: - history.add_datasets(trans.sa_session, datasets_to_persist, set_hid=True, quota=False, flush=False) - # a side effect of history.add_datasets is a commit within db_next_hid (even with flush=False). + if execution_slice: + # a side effect of adding datasets to a history is a commit within db_next_hid (even with flush=False). + history.add_pending_datasets() else: # Make sure collections, implicit jobs etc are flushed even if there are no precreated output datasets trans.sa_session.flush() @@ -130,7 +128,6 @@ class ExecutionSlice: self.job_index = job_index self.param_combination = param_combination self.dataset_collection_elements = dataset_collection_elements - self.datasets_to_persist = None self.history = None diff --git a/test/unit/tools/test_actions.py b/test/unit/tools/test_actions.py index d8d6e69bbf8..3f7347ee286 100644 --- a/test/unit/tools/test_actions.py +++ b/test/unit/tools/test_actions.py @@ -134,7 +134,7 @@ class DefaultToolActionTestCase(unittest.TestCase, tools_support.UsesApp, tools_ if incoming is None: incoming = dict(param1="moo") self._init_tool(contents) - job, out_data, _, _ = self.action.execute( + job, out_data, _, = self.action.execute( tool=self.tool, trans=self.trans, history=self.history,