Merge pull request #22154 from mvdbeek/fix-redundant-empty-collection-elements

[26.0] Fix redundant empty collection elements

lib/galaxy/job_execution/output_collect.py

@@ -189,6 +189,9 @@ def collect_dynamic_outputs(
 
         # We are adding dynamic collections, which may be precreated, but their actually state is still new!
         collection.populated_state = collection.populated_states.NEW
+        # Clear any existing elements to avoid duplicates when re-populating
+        collection.elements.clear()
+        collection.element_count = None
 
         try:
             collection_builder = builder.BoundCollectionBuilder(collection)

lib/galaxy/model/__init__.py

@@ -1840,8 +1840,21 @@ class Job(Base, JobLike, UsesCreateAndUpdateTime, Dictifiable, Serializable):
                 if obj.name not in out_data:
                     out_collections[obj.name] = obj.dataset_collection_instance
                 # else this is a mapped over output
-        if not exclude_implicit_outputs:
-            out_collections.update([(obj.name, obj.dataset_collection) for obj in self.output_dataset_collections])
+        if exclude_implicit_outputs:
+            # Include implicit output dataset collections only when they represent
+            # a tool's collection output (name not in out_data). Exclude shared DCs
+            # for mapped dataset outputs (name in out_data) which have N precreated
+            # elements where only the current job's element is initialized.
+            for implicit_obj in self.output_dataset_collections:
+                if implicit_obj.name not in out_data:
+                    out_collections[implicit_obj.name] = implicit_obj.dataset_collection
+        else:
+            out_collections.update(
+                [
+                    (implicit_obj.name, implicit_obj.dataset_collection)
+                    for implicit_obj in self.output_dataset_collections
+                ]
+            )
         return IoDicts(inp_data, out_data, out_collections)
 
     # TODO: Add accessors for members defined in SQL Alchemy for the Job table and

lib/galaxy/model/store/__init__.py

@@ -899,6 +899,11 @@ class ModelImportStore(metaclass=abc.ABCMeta):
                 for attribute in attributes:
                     if attribute in collection_attrs:
                         setattr(dc, attribute, collection_attrs.get(attribute))
+                # Clear existing elements to avoid duplicates when re-importing
+                if "elements" in collection_attrs:
+                    for element in list(dc.elements):
+                        self.sa_session.delete(element)
+                    dc.elements.clear()
                 materialize_elements(dc)
             else:
                 # create collection

test/integration/test_extended_metadata.py

@@ -1,6 +1,16 @@
 """Integration tests for the Pulsar embedded runner."""
 
+from sqlalchemy import (
+    func,
+    select,
+)
+
+from galaxy.model import (
+    DatasetCollectionElement,
+    HistoryDatasetCollectionAssociation,
+)
 from galaxy_test.base.populators import (
+    DatasetCollectionPopulator,
     DatasetPopulator,
     LibraryPopulator,
 )
@@ -56,6 +66,7 @@ class TestExtendedMetadataIntegration(integration_util.IntegrationTestCase):
     def setUp(self):
         super().setUp()
         self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+        self.dataset_collection_populator = DatasetCollectionPopulator(self.galaxy_interactor)
         self.library_populator = LibraryPopulator(self.galaxy_interactor)
 
     @classmethod
@@ -100,6 +111,184 @@ class TestExtendedMetadataIntegration(integration_util.IntegrationTestCase):
         assert dataset["file_ext"] == "bed", dataset
         assert dataset["created_from_basename"] == "4.bed"
 
+    def test_no_duplicate_elements_in_dynamic_list_output(self, history_id):
+        """Run a tool with dynamic collection output and verify no duplicate elements."""
+        response = self.dataset_populator.run_tool(
+            "collection_creates_dynamic_list_of_pairs",
+            {"foo": "bar"},
+            history_id,
+        )
+        job_api_id = response["jobs"][0]["id"]
+        self.dataset_populator.wait_for_job(job_api_id, assert_ok=True)
+
+        output_collections = response["output_collections"]
+        hdca_details = self.dataset_populator.get_history_collection_details(
+            history_id, content_id=output_collections[0]["id"]
+        )
+
+        sa_session = self._app.model.session
+        hdca_id = self._app.security.decode_id(hdca_details["id"])
+        hdca = sa_session.get(HistoryDatasetCollectionAssociation, hdca_id)
+        assert hdca is not None
+        dc = hdca.collection
+
+        # Verify exactly 3 outer elements (samp1, samp2, samp3) - no duplicates
+        outer_count = sa_session.scalar(
+            select(func.count()).where(DatasetCollectionElement.dataset_collection_id == dc.id)
+        )
+        assert outer_count == 3, f"Expected 3 outer elements but found {outer_count}. Duplicate elements detected!"
+
+        # Verify each inner pair has exactly 2 elements (forward, reverse)
+        for element in dc.elements:
+            assert element.child_collection is not None
+            inner_count = sa_session.scalar(
+                select(func.count()).where(
+                    DatasetCollectionElement.dataset_collection_id == element.child_collection.id
+                )
+            )
+            assert inner_count == 2, (
+                f"Expected 2 inner elements for '{element.element_identifier}' "
+                f"but found {inner_count}. Duplicate elements detected!"
+            )
+
+    def test_no_duplicate_elements_in_mapped_dynamic_collection(self, history_id):
+        """Map a tool with dynamic collection output over a list and verify no duplicate elements."""
+        # Create a list of 2 tabular datasets
+        fetch_response = self.dataset_collection_populator.create_list_in_history(
+            history_id,
+            contents=["101\t1\n101\t2\n105\t3\n", "201\t10\n201\t20\n205\t30\n"],
+            ext="tabular",
+            wait=True,
+        )
+        self.dataset_populator.wait_for_history(history_id, assert_ok=True)
+        hdca_id = fetch_response.json()["output_collections"][0]["id"]
+
+        # Map collection_split_on_column over the list
+        inputs = {
+            "input1": {"batch": True, "values": [{"src": "hdca", "id": hdca_id}]},
+        }
+        response = self.dataset_populator.run_tool(
+            "collection_split_on_column",
+            inputs,
+            history_id,
+        )
+
+        # Wait for all mapping jobs to complete
+        for job in response["jobs"]:
+            self.dataset_populator.wait_for_job(job["id"], assert_ok=True)
+
+        # The implicit output should be a list:list collection
+        implicit_collections = response["implicit_collections"]
+        assert len(implicit_collections) == 1
+        implicit_hdca_details = self.dataset_populator.get_history_collection_details(
+            history_id, content_id=implicit_collections[0]["id"]
+        )
+
+        sa_session = self._app.model.session
+        implicit_hdca_id = self._app.security.decode_id(implicit_hdca_details["id"])
+        implicit_hdca = sa_session.get(HistoryDatasetCollectionAssociation, implicit_hdca_id)
+        assert implicit_hdca is not None
+        outer_dc = implicit_hdca.collection
+
+        # The outer collection should have exactly 2 elements (one per input)
+        outer_count = sa_session.scalar(
+            select(func.count()).where(DatasetCollectionElement.dataset_collection_id == outer_dc.id)
+        )
+        assert (
+            outer_count == 2
+        ), f"Expected 2 outer elements but found {outer_count}. Duplicate elements detected in mapped output!"
+
+        # Each inner collection should have no duplicate elements
+        for element in outer_dc.elements:
+            inner_dc = element.child_collection
+            assert inner_dc is not None, f"Inner collection for '{element.element_identifier}' is None"
+            inner_count = sa_session.scalar(
+                select(func.count()).where(DatasetCollectionElement.dataset_collection_id == inner_dc.id)
+            )
+            # Each input has 2 unique first-column values, so 2 split files
+            assert (
+                inner_count is not None and inner_count > 0
+            ), f"Inner collection for '{element.element_identifier}' has no elements"
+            # Check for duplicates: element_identifiers should be unique
+            inner_elements = sa_session.scalars(
+                select(DatasetCollectionElement).where(DatasetCollectionElement.dataset_collection_id == inner_dc.id)
+            ).all()
+            identifiers = [e.element_identifier for e in inner_elements]
+            assert len(identifiers) == len(set(identifiers)), (
+                f"Duplicate element identifiers found in inner collection "
+                f"for '{element.element_identifier}': {identifiers}"
+            )
+
+    def test_no_duplicate_elements_in_mapped_static_collection(self, history_id):
+        """Map a tool with a static collection output over a list and verify no duplicate elements.
+
+        This covers the case where a tool declares a static collection output
+        (e.g. paired with forward/reverse) and is mapped over a list input,
+        producing a list:paired output. The per-job DC contains fully
+        initialized elements and must be included in io_dicts for metadata
+        serialization.
+        """
+        hdca = self.dataset_collection_populator.create_list_in_history(
+            history_id,
+            contents=["line1\nline2\nline3\nline4\n", "lineA\nlineB\nlineC\nlineD\n"],
+            ext="txt",
+            wait=True,
+        )
+        self.dataset_populator.wait_for_history(history_id, assert_ok=True)
+        hdca_id = hdca.json()["output_collections"][0]["id"]
+
+        inputs = {
+            "input1": {"batch": True, "values": [{"src": "hdca", "id": hdca_id}]},
+        }
+        response = self.dataset_populator.run_tool(
+            "collection_creates_pair",
+            inputs,
+            history_id,
+        )
+
+        for job in response["jobs"]:
+            self.dataset_populator.wait_for_job(job["id"], assert_ok=True)
+
+        implicit_collections = response["implicit_collections"]
+        assert len(implicit_collections) == 1
+        implicit_hdca_details = self.dataset_populator.get_history_collection_details(
+            history_id, content_id=implicit_collections[0]["id"]
+        )
+
+        sa_session = self._app.model.session
+        implicit_hdca_id = self._app.security.decode_id(implicit_hdca_details["id"])
+        implicit_hdca = sa_session.get(HistoryDatasetCollectionAssociation, implicit_hdca_id)
+        assert implicit_hdca is not None
+        outer_dc = implicit_hdca.collection
+
+        # The outer collection should have exactly 2 elements (one per input)
+        outer_count = sa_session.scalar(
+            select(func.count()).where(DatasetCollectionElement.dataset_collection_id == outer_dc.id)
+        )
+        assert (
+            outer_count == 2
+        ), f"Expected 2 outer elements but found {outer_count}. Duplicate elements detected in mapped output!"
+
+        # Each inner pair should have exactly 2 elements (forward, reverse) with no duplicates
+        for element in outer_dc.elements:
+            inner_dc = element.child_collection
+            assert inner_dc is not None, f"Inner collection for '{element.element_identifier}' is None"
+            inner_count = sa_session.scalar(
+                select(func.count()).where(DatasetCollectionElement.dataset_collection_id == inner_dc.id)
+            )
+            assert inner_count == 2, (
+                f"Expected 2 inner elements for '{element.element_identifier}' "
+                f"but found {inner_count}. Duplicate elements detected!"
+            )
+            inner_elements = sa_session.scalars(
+                select(DatasetCollectionElement).where(DatasetCollectionElement.dataset_collection_id == inner_dc.id)
+            ).all()
+            identifiers = [e.element_identifier for e in inner_elements]
+            assert set(identifiers) == {
+                "forward",
+                "reverse",
+            }, f"Expected forward/reverse but got {identifiers} for '{element.element_identifier}'"
+
     def test_purge_while_job_running(self):
         # pass extra_sleep, since templating the command line will fail if the output
         # is deleted before remote_tool_eval runs.

test/unit/data/model/test_model.py

@@ -23,22 +23,36 @@ def test_permitted_actions():
 def test_io_dicts_excludes_implicit_output_collections():
     """Regression test for https://github.com/galaxyproject/galaxy/issues/22015
 
-    When a tool with an explicit collection output is mapped over a list,
-    each job gets a JobToImplicitOutputDatasetCollectionAssociation pointing
-    to a DatasetCollection with precreated (unpopulated) elements.
-    io_dicts(exclude_implicit_outputs=True) must exclude these to avoid
-    crashes during metadata serialization on unpopulated elements.
+    When a tool with a dataset output is mapped over a list, each job gets
+    both a JobToOutputDatasetAssociation and a
+    JobToImplicitOutputDatasetCollectionAssociation with the same name.
+    The implicit DC has precreated (unpopulated) elements; only the current
+    job's element is initialized. io_dicts(exclude_implicit_outputs=True)
+    must exclude these shared DCs (name in out_data) to avoid crashes
+    during metadata serialization, but must include implicit DCs for
+    collection outputs (name not in out_data) so set_metadata.py can
+    discover and populate them.
     """
     job = model.Job()
     dc = model.DatasetCollection(collection_type="paired")
     assoc = model.JobToImplicitOutputDatasetCollectionAssociation(name="paired_output", dataset_collection=dc)
     job.output_dataset_collections.append(assoc)
 
-    # With exclude_implicit_outputs=True, output_dataset_collections must be excluded
+    # When the name is NOT in out_data (collection output), the implicit DC
+    # should be included even with exclude_implicit_outputs=True
+    io = job.io_dicts(exclude_implicit_outputs=True)
+    assert "paired_output" in io.out_collections
+    assert io.out_collections["paired_output"] is dc
+
+    # Now simulate a mapped dataset output: same name in both out_data and
+    # output_dataset_collections. The shared DC must be excluded.
+    hda = model.HistoryDatasetAssociation()
+    out_assoc = model.JobToOutputDatasetAssociation(name="paired_output", dataset=hda)
+    job.output_datasets.append(out_assoc)
+
     io = job.io_dicts(exclude_implicit_outputs=True)
     assert "paired_output" not in io.out_collections
 
     # With exclude_implicit_outputs=False (default), they should be included
     io = job.io_dicts(exclude_implicit_outputs=False)
     assert "paired_output" in io.out_collections
-    assert io.out_collections["paired_output"] is dc