Merge pull request #22970 from jmchilton/intermediate_extract

[26.1] Include hidden intermediates in workflow extraction summary (#22967)
This commit is contained in:
Marius van den Beek
2026-06-19 16:42:39 +02:00
committed by GitHub
5 changed files with 138 additions and 3 deletions
+10
View File
@@ -4241,6 +4241,16 @@ class History(Base, HasTags, UsesAnnotations, HasName, Serializable, UsesCreateA
def visible_contents(self):
return self.contents_iter(types=["dataset", "dataset_collection"], visible=True)
@property
def all_contents(self):
"""Return every content (hidden, deleted and purged included) ordered by hid.
Workflow extraction reconstructs provenance from the jobs behind history
items, so it needs the hidden intermediates IWC-style workflows produce -
not just the visible outputs.
"""
return self.contents_iter(types=["dataset", "dataset_collection"])
def contents_iter(self, **kwds):
"""
Fetch filtered list of contents of history.
+26 -1
View File
@@ -11,6 +11,8 @@ from typing import (
Optional,
)
from sqlalchemy import select
from galaxy import (
exceptions,
model,
@@ -18,6 +20,7 @@ from galaxy import (
from galaxy.managers.context import ProvidesHistoryContext
from galaxy.managers.jobs import JobManager
from galaxy.model import (
DatasetCollectionElement,
History,
HistoryDatasetAssociation,
HistoryDatasetCollectionAssociation,
@@ -314,8 +317,25 @@ class WorkflowSummary(BaseWorkflowSummary):
self.hda_hid_in_history: dict[int, int] = {}
self.hdca_hid_in_history: dict[int, int] = {}
self.collection_element_hda_ids: set[int] = self.__collection_element_hda_ids()
self.__summarize()
def __collection_element_hda_ids(self) -> set[int]:
"""Ids of this history's HDAs that are elements of some collection.
Hidden contents now reach summarization (see History.all_contents), but a
collection's element datasets are represented by their collection during
extraction - "the collection or nothing" - so the hidden ones must be
skipped to avoid minting spurious per-element steps (e.g. one per map-over
element). Fetched once rather than per-dataset."""
stmt = (
select(HistoryDatasetAssociation.id)
.join(DatasetCollectionElement, DatasetCollectionElement.hda_id == HistoryDatasetAssociation.id)
.where(HistoryDatasetAssociation.history_id == self.history.id)
)
return set(self.trans.sa_session.scalars(stmt).all())
def hid(self, content: HistoryItem) -> int:
if content.history_content_type == "dataset_collection":
if content.id in self.hdca_hid_in_history:
@@ -343,7 +363,7 @@ class WorkflowSummary(BaseWorkflowSummary):
# just grab the implicitly mapped jobs and handle in second pass. Second pass is
# needed because cannot allow selection of individual datasets from an implicit
# mapping during extraction - you get the collection or nothing.
for content in self.history.visible_contents:
for content in self.history.all_contents:
self.__summarize_content(content)
def __summarize_content(self, content: HistoryItem) -> None:
@@ -416,6 +436,11 @@ class WorkflowSummary(BaseWorkflowSummary):
self.jobs[DatasetCollectionCreationJob(dataset_collection)] = [(None, dataset_collection)]
def __summarize_dataset(self, dataset: HistoryDatasetAssociation) -> None:
if not dataset.visible and dataset.id in self.collection_element_hda_ids:
# Hidden element of a collection - represented by its collection, not
# as a standalone step. Visible collection members are left alone so
# behavior matches the prior visible-only scan.
return
if not self._check_state(dataset):
return
@@ -1379,5 +1379,48 @@ class TestWorkflowExtractionSummaryApi(_ExtractionHelpersMixin, BaseWorkflowsApi
assert isinstance(job["checked"], bool)
assert isinstance(job["outputs"], list)
@skip_without_tool("cat1")
def test_extraction_summary_includes_hidden_intermediate(self):
# Histories produced by IWC-style workflows hide their intermediate
# datasets. The summary must still surface the jobs behind those hidden
# intermediates so the whole provenance graph can be extracted - not
# just the chain of visible outputs.
with self.dataset_populator.test_history() as history_id:
hda1 = self.dataset_populator.new_dataset(history_id, content="foo\nbar", wait=True)
first_run = self.dataset_populator.run_tool(
"cat1", {"input1": {"src": "hda", "id": hda1["id"]}}, history_id
)
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
intermediate = first_run["outputs"][0]
self.dataset_populator.hide_dataset(intermediate["id"])
self.dataset_populator.run_tool("cat1", {"input1": {"src": "hda", "id": intermediate["id"]}}, history_id)
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
summary = self._get_extraction_summary(history_id)
tool_jobs = [j for j in summary["jobs"] if j["step_type"] == "tool"]
# Both cat1 jobs must appear even though the dataset bridging them is
# hidden; before the fix only the job behind the visible output did.
assert len(tool_jobs) == 2, summary["jobs"]
assert all(j["checked"] for j in tool_jobs), summary["jobs"]
@skip_without_tool("random_lines1")
def test_extraction_summary_no_spurious_rows_for_mapover_elements(self):
# Map-over hides each per-element output dataset. Surfacing hidden contents
# must not turn those elements into their own job cards - the mapped step
# (its implicit collection) represents them; otherwise a pair yields a
# spurious extra card per element.
with self.dataset_populator.test_history() as history_id:
hdca = self.dataset_collection_populator.create_pair_in_history(
history_id, contents=["1 2 3\n4 5 6", "7 8 9\n10 11 10"], wait=True
).json()["outputs"][0]
inputs = {"input": {"batch": True, "values": [{"src": "hdca", "id": hdca["id"]}]}, "num_lines": 1}
self._run_tool_get_collection_and_job_id(history_id, "random_lines1", inputs)
summary = self._get_extraction_summary(history_id)
tool_jobs = [j for j in summary["jobs"] if j["step_type"] == "tool"]
assert len(tool_jobs) == 1, summary["jobs"]
assert tool_jobs[0]["implicit_collection_jobs_id"] is not None, tool_jobs[0]
RunJobsSummary = namedtuple("RunJobsSummary", ["history_id", "workflow_id", "inputs", "jobs"])
+15
View File
@@ -226,6 +226,21 @@ def test_history_contents(session, make_history, make_hda):
assert contents_iter_names(ids=[d1.id, d3.id]) == ["1", "3"]
def test_history_content_views(session, make_history, make_hda):
h1 = make_history()
make_hda(history=h1, name="1") # visible, active
make_hda(history=h1, name="2", visible=False, create_dataset=True, sa_session=session) # hidden, active
make_hda(history=h1, name="3", deleted=True, create_dataset=True, sa_session=session) # visible, deleted
make_hda(history=h1, name="4", visible=False, deleted=True) # hidden, deleted
history = session.get(m.History, h1.id)
assert [c.name for c in history.active_contents] == ["1"]
assert [c.name for c in history.visible_contents] == ["1", "3"]
# all_contents (used by workflow extraction) must include the hidden
# intermediate "2" that visible_contents drops.
assert [c.name for c in history.all_contents] == ["1", "2", "3", "4"]
def test_current_galaxy_session(make_user, make_galaxy_session):
user = make_user()
galaxy_session = make_galaxy_session(user=user)
+44 -2
View File
@@ -122,6 +122,25 @@ class TestWorkflowExtractSummary(TestCase):
job = next(iter(job_dict.keys()))
assert job is creating_job
def test_includes_hidden_standalone_intermediate(self):
# A hidden dataset that is not a collection element (e.g. an intermediate
# an IWC workflow hid) must still yield a job - the regression this fixes.
hda = MockHda(visible=False)
self.history.active_datasets.append(hda)
job_dict, warnings = self._summarize()
assert not warnings
assert len(job_dict) == 1
def test_skips_hidden_collection_element(self):
# A hidden dataset that IS a collection element is represented by its
# collection, so it must not become a standalone job.
element = MockHda(visible=False, id=555)
self.history.active_datasets.append(element)
self.trans.sa_session.collection_element_hda_ids = [element.id]
job_dict, warnings = self._summarize()
assert not warnings
assert len(job_dict) == 0
def test_warns_and_skips_datasets_if_not_finished(self):
hda = MockHda(state="queued")
self.history.active_datasets.append(hda)
@@ -140,6 +159,7 @@ class MockJobToOutputDatasetAssociation:
class MockHistory:
def __init__(self):
self.id = 1
self.active_datasets = []
@property
@@ -150,19 +170,41 @@ class MockHistory:
def visible_contents(self):
return self.active_contents
@property
def all_contents(self):
return self.active_contents
class MockScalarResult:
def __init__(self, values):
self._values = values
def all(self):
return self._values
class MockSession:
def __init__(self):
self.collection_element_hda_ids: list[int] = []
def scalars(self, statement):
return MockScalarResult(self.collection_element_hda_ids)
class MockTrans:
def __init__(self, history):
self.history = history
self.sa_session = MockSession()
def get_history(self):
return self.history
class MockHda:
def __init__(self, state="ok", output_name="out1", job=None):
def __init__(self, state="ok", output_name="out1", job=None, visible=True, id=123):
self.hid = 1
self.id = 123
self.id = id
self.visible = visible
self.state = state
self.copied_from_history_dataset_association = None
self.copied_from_library_dataset_dataset_association = None