Merge pull request #20731 from mvdbeek/hierarchical_display_collection_dataset_states

Hierarchical display collection dataset states

client/src/api/datasets.ts

@@ -113,3 +113,13 @@ export async function fetchDatasetAttributes(datasetId: string) {
 
 export type HistoryContentType = components["schemas"]["HistoryContentType"];
 export type HistoryContentSource = components["schemas"]["HistoryContentSource"];
+
+/** Dataset state constants */
+// Non-terminal dataset states (dataset is still being processed)
+export const NON_TERMINAL_DATASET_STATES = ["new", "upload", "queued", "running", "setting_metadata"];
+
+// Error dataset states (dataset failed processing)
+export const ERROR_DATASET_STATES = ["error", "failed_metadata"];
+
+// Terminal dataset states (dataset processing is complete)
+export const TERMINAL_DATASET_STATES = ["ok", "empty", "deferred", "discarded", "paused"].concat(ERROR_DATASET_STATES);

client/src/api/schema/schema.ts

@@ -8958,6 +8958,18 @@ export interface components {
              * @default []
              */
             elements: components["schemas"]["DCESummary"][];
+            /**
+             * Elements Datatypes
+             * @description A set containing all the different element datatypes in the collection.
+             */
+            elements_datatypes: string[];
+            /**
+             * Datasets deleted
+             * @description The number of elements in the collection that are marked as deleted.
+             */
+            elements_deleted: number;
+            /** @description A dictionary containing counts for each dataset state in the collection. */
+            elements_states: components["schemas"]["ElementsStatesDict"];
             /**
              * Dataset Collection ID
              * @example 0123456789ABCDEF
@@ -10224,6 +10236,33 @@ export interface components {
          * @enum {string}
          */
         ElementsFromType: "archive" | "bagit" | "bagit_archive" | "directory";
+        /** ElementsStatesDict */
+        ElementsStatesDict: {
+            /** Deferred */
+            deferred?: number;
+            /** Discarded */
+            discarded?: number;
+            /** Empty */
+            empty?: number;
+            /** Error */
+            error?: number;
+            /** Failed Metadata */
+            failed_metadata?: number;
+            /** New */
+            new?: number;
+            /** Ok */
+            ok?: number;
+            /** Paused */
+            paused?: number;
+            /** Queued */
+            queued?: number;
+            /** Running */
+            running?: number;
+            /** Setting Metadata */
+            setting_metadata?: number;
+            /** Upload */
+            upload?: number;
+        };
         /** EmptyFieldParameterValidatorModel */
         EmptyFieldParameterValidatorModel: {
             /**
@@ -12281,6 +12320,13 @@ export interface components {
              * @description A set containing all the different element datatypes in the collection.
              */
             elements_datatypes?: string[] | null;
+            /**
+             * Datasets deleted
+             * @description The number of elements in the collection that are marked as deleted.
+             */
+            elements_deleted?: number | null;
+            /** @description A dictionary containing counts for each dataset state in the collection. */
+            elements_states?: components["schemas"]["ElementsStatesDict"] | null;
             /**
              * HID
              * @description The index position of this item in the History.
@@ -12427,6 +12473,13 @@ export interface components {
              * @description A set containing all the different element datatypes in the collection.
              */
             elements_datatypes: string[];
+            /**
+             * Datasets deleted
+             * @description The number of elements in the collection that are marked as deleted.
+             */
+            elements_deleted: number;
+            /** @description A dictionary containing counts for each dataset state in the collection. */
+            elements_states: components["schemas"]["ElementsStatesDict"];
             /**
              * HID
              * @description The index position of this item in the History.
@@ -12565,6 +12618,13 @@ export interface components {
              * @description A set containing all the different element datatypes in the collection.
              */
             elements_datatypes: string[];
+            /**
+             * Datasets deleted
+             * @description The number of elements in the collection that are marked as deleted.
+             */
+            elements_deleted: number;
+            /** @description A dictionary containing counts for each dataset state in the collection. */
+            elements_states: components["schemas"]["ElementsStatesDict"];
             /**
              * HID
              * @description The index position of this item in the History.

client/src/components/History/Content/Collection/CollectionDescription.test.ts

@@ -22,6 +22,8 @@ const defaultTestHDCA: HDCASummary = {
     deleted: false,
     visible: true,
     elements_datatypes: [],
+    elements_deleted: 0,
+    elements_states: {},
     history_id: "fake_history_id",
     model_class: "HistoryDatasetCollectionAssociation",
     populated_state: "ok",

client/src/components/History/Content/Collection/CollectionDescription.vue

@@ -1,11 +1,14 @@
 <script setup lang="ts">
+import { BBadge } from "bootstrap-vue";
 import { computed } from "vue";
 
 import type { HDCASummary } from "@/api";
 
+import { DatasetStateSummary } from "./DatasetStateSummary";
 import { JobStateSummary } from "./JobStateSummary";
 
 import CollectionProgress from "./CollectionProgress.vue";
+import DatasetProgress from "./DatasetProgress.vue";
 
 interface Props {
     hdca: HDCASummary;
@@ -28,6 +31,10 @@ const jobStateSummary = computed(() => {
     return new JobStateSummary(props.hdca);
 });
 
+const datasetStateSummary = computed(() => {
+    return new DatasetStateSummary(props.hdca);
+});
+
 const collectionLabel = computed(() => {
     const collectionType = props.hdca.collection_type;
     const isList = collectionType.startsWith("list");
@@ -89,15 +96,30 @@ function pluralize(word: string) {
 
 <template>
     <div>
-        <span v-if="hdca.collection_type == 'paired_or_unpaired'" class="description mt-1 mb-1">
-            a <b v-if="isHomogeneous">{{ homogeneousDatatype }}</b> {{ pluralizedItem }}
-        </span>
-        <span v-else class="description mt-1 mb-1">
-            a {{ collectionLabel }} with {{ hdca.element_count || 0
-            }}<b v-if="isHomogeneous">{{ homogeneousDatatype }}</b> {{ pluralizedItem }}
-        </span>
+        <div class="d-flex justify-content-between align-items-start">
+            <div>
+                <span v-if="hdca.collection_type == 'paired_or_unpaired'" class="description mt-1 mb-1">
+                    a <b v-if="isHomogeneous">{{ homogeneousDatatype }}</b> {{ pluralizedItem }}
+                </span>
+                <span v-else class="description mt-1 mb-1">
+                    a {{ collectionLabel }} with {{ hdca.element_count || 0
+                    }}<b v-if="isHomogeneous">{{ homogeneousDatatype }}</b> {{ pluralizedItem }}
+                </span>
+            </div>
+
+            <BBadge
+                v-if="datasetStateSummary.elements_deleted > 0"
+                variant="warning"
+                class="ml-2"
+                :title="`${datasetStateSummary.elements_deleted} deleted ${
+                    datasetStateSummary.elements_deleted === 1 ? 'dataset' : 'datasets'
+                } in collection`">
+                <icon icon="trash" /> {{ datasetStateSummary.elements_deleted }} deleted
+            </BBadge>
+        </div>
 
         <CollectionProgress v-if="jobStateSummary.size != 0" :summary="jobStateSummary" />
+        <DatasetProgress v-else-if="datasetStateSummary.datasetCount > 0" :summary="datasetStateSummary" />
     </div>
 </template>
 

client/src/components/History/Content/Collection/DatasetProgress.vue

@@ -0,0 +1,37 @@
+<!-- Dataset state progress bar for a collection. -->
+<script setup lang="ts">
+import type { DatasetStateSummary } from "./DatasetStateSummary";
+
+interface Props {
+    summary: DatasetStateSummary;
+}
+
+defineProps<Props>();
+</script>
+
+<template>
+    <div class="dataset-progress">
+        <BProgress v-if="!summary.isTerminal" :max="summary.datasetCount">
+            <BProgressBar
+                v-if="summary.errorCount"
+                v-b-tooltip.hover="`${summary.errorCount} Error`"
+                :value="summary.errorCount"
+                variant="danger" />
+            <BProgressBar
+                v-if="summary.okCount"
+                v-b-tooltip.hover="`${summary.okCount} OK`"
+                :value="summary.okCount"
+                variant="success" />
+            <BProgressBar
+                v-if="summary.runningCount"
+                v-b-tooltip.hover="`${summary.runningCount} Running`"
+                :value="summary.runningCount"
+                variant="warning" />
+            <BProgressBar
+                v-if="summary.waitingCount"
+                v-b-tooltip.hover="`${summary.waitingCount} Waiting`"
+                :value="summary.waitingCount"
+                variant="secondary" />
+        </BProgress>
+    </div>
+</template>

client/src/components/History/Content/Collection/DatasetStateSummary.ts

@@ -0,0 +1,87 @@
+/**
+ * Read-only class for dataset states in a collection.
+ * Similar to JobStateSummary but for dataset states.
+ */
+import { ERROR_DATASET_STATES, NON_TERMINAL_DATASET_STATES } from "@/api/datasets";
+
+interface HDCAWithDatasetStates {
+    elements_states?: Record<string, number>;
+    elements_deleted?: number;
+    populated_state?: string | null;
+}
+
+export class DatasetStateSummary {
+    elements_states: Record<string, number>;
+    elements_deleted: number;
+    populated_state: string | null;
+
+    constructor(hdca: HDCAWithDatasetStates = {}) {
+        this.elements_states = hdca.elements_states || {};
+        this.elements_deleted = hdca.elements_deleted || 0;
+        this.populated_state = hdca.populated_state || null;
+    }
+
+    get(state: string): number {
+        return this.elements_states[state] || 0;
+    }
+
+    get datasetCount(): number {
+        return Object.values(this.elements_states).reduce((sum, count) => sum + count, 0);
+    }
+
+    private getCount(states: string[]): number {
+        return states.reduce((sum, state) => sum + this.get(state), 0);
+    }
+
+    get okCount(): number {
+        return this.getCount(["ok", "empty", "deferred"]);
+    }
+
+    get errorCount(): number {
+        return this.getCount(["error", "failed_metadata"]);
+    }
+
+    get runningCount(): number {
+        return this.getCount(["running", "setting_metadata"]);
+    }
+
+    get waitingCount(): number {
+        return this.getCount(["queued", "new", "upload", "paused"]);
+    }
+
+    get deferredCount(): number {
+        return this.getCount(["deferred", "discarded"]);
+    }
+
+    get isTerminal(): boolean {
+        return !NON_TERMINAL_DATASET_STATES.some((state) => this.get(state) > 0);
+    }
+
+    get state(): string {
+        if (this.datasetCount === 0) {
+            return this.populated_state || "new";
+        }
+
+        if (this.errorCount > 0) {
+            return "error";
+        }
+
+        if (this.runningCount > 0) {
+            return "running";
+        }
+
+        if (this.waitingCount > 0) {
+            return "queued";
+        }
+
+        if (this.deferredCount > 0) {
+            return "deferred";
+        }
+
+        if (this.isTerminal) {
+            return "ok";
+        }
+
+        return "new";
+    }
+}

client/src/components/History/Content/ContentItem.vue

@@ -20,7 +20,7 @@ import { useEntryPointStore } from "@/stores/entryPointStore";
 import { useEventStore } from "@/stores/eventStore";
 import { clearDrag } from "@/utils/setDrag";
 
-import { getContentItemState, type StateMap, STATES } from "./model/states";
+import { getContentItemState, type State, STATES } from "./model/states";
 import type { RouterPushOptions } from "./router-push-options";
 
 import CollectionDescription from "./Collection/CollectionDescription.vue";
@@ -97,7 +97,7 @@ const itemIsRunningInteractiveTool = computed(() => {
 });
 
 const contentId = computed(() => {
-    return `dataset-${props.item.id}`;
+    return isCollection.value ? `collection-${props.item.id}` : `dataset-${props.item.id}`;
 });
 
 const contentCls = computed(() => {
@@ -133,7 +133,7 @@ const hasStateIcon = computed(() => {
     return contentState.value && contentState.value.icon;
 });
 
-const state = computed<keyof StateMap>(() => {
+const state = computed<State>(() => {
     if (props.isPlaceholder) {
         return "placeholder";
     }
@@ -159,7 +159,7 @@ const tagsDisabled = computed(() => {
 });
 
 const isCollection = computed(() => {
-    return "collection_type" in props.item;
+    return "collection_type" in props.item || props.item.element_type === "dataset_collection";
 });
 
 const itemUrls = computed<ItemUrls>(() => {

client/src/components/History/Content/model/states.ts

@@ -1,9 +1,28 @@
-import { isHDCA } from "@/api";
+import type { DCESummary, HDADetailed, HDCADetailed } from "@/api";
+import { ERROR_DATASET_STATES, NON_TERMINAL_DATASET_STATES } from "@/api/datasets";
+import type { HistoryContentsResult } from "@/api/histories";
 import type { components } from "@/api/schema";
 
+import { DatasetStateSummary } from "../Collection/DatasetStateSummary";
+
+type HistoryContentItem = HistoryContentsResult[number];
+
+function isHDCAItem(item: HistoryContentItem | HDADetailed | HDCADetailed | DCESummary): boolean {
+    return (
+        item &&
+        typeof item === "object" &&
+        "history_content_type" in item &&
+        item.history_content_type === "dataset_collection"
+    );
+}
+
+function isDCEWithCollection(item: HistoryContentItem | HDADetailed | HDCADetailed | DCESummary): boolean {
+    return item && typeof item === "object" && "collection_type" in item && item.collection_type !== undefined;
+}
+
 type DatasetState = components["schemas"]["DatasetState"];
 // The 'failed' state is for the collection job state summary, not a dataset state.
-type State =
+export type State =
     | DatasetState
     | "failed"
     | "placeholder"
@@ -166,24 +185,51 @@ export const HIERARCHICAL_COLLECTION_JOB_STATES = [
     "new",
 ] as const;
 
-export function getContentItemState(item: any) {
-    if (isHDCA(item)) {
-        if (item.populated_state === "failed") {
+/** Similar hierarchy for dataset states, ordered from highest to lowest priority. */
+export const HIERARCHICAL_COLLECTION_DATASET_STATES = [
+    ...(ERROR_DATASET_STATES as readonly DatasetState[]),
+    ...(NON_TERMINAL_DATASET_STATES as readonly DatasetState[]),
+    "deferred" as const,
+    "discarded" as const,
+] as const satisfies readonly DatasetState[];
+
+export function getContentItemState(item: HistoryContentItem | HDADetailed | HDCADetailed): State {
+    if (isHDCAItem(item) || isDCEWithCollection(item)) {
+        if ("populated_state" in item && item.populated_state === "failed") {
             return "failed_populated_state";
         }
-        if (item.populated_state === "new") {
+        if ("populated_state" in item && item.populated_state === "new") {
             return "new_populated_state";
         }
-        if (item.job_state_summary) {
+
+        // Check dataset states first (higher priority for actual data states)
+        if ("elements_states" in item && item.elements_states) {
+            const datasetSummary = new DatasetStateSummary(
+                item as {
+                    elements_states?: Record<string, number>;
+                    elements_deleted?: number;
+                    populated_state?: string | null;
+                }
+            );
+            for (const datasetState of HIERARCHICAL_COLLECTION_DATASET_STATES) {
+                if (datasetSummary.get(datasetState) > 0) {
+                    return datasetState;
+                }
+            }
+        }
+
+        // Fall back to job states if no dataset states
+        if ("job_state_summary" in item && item.job_state_summary) {
+            const jobStateSummary = item.job_state_summary as Record<string, number>;
             for (const jobState of HIERARCHICAL_COLLECTION_JOB_STATES) {
-                if (item.job_state_summary[jobState] > 0) {
+                if ((jobStateSummary[jobState] || 0) > 0) {
                     return jobState;
                 }
             }
         }
-    } else if (item.accessible === false) {
+    } else if ("accessible" in item && item.accessible === false) {
         return "inaccessible";
-    } else if (item.state) {
+    } else if ("state" in item && item.state) {
         return item.state;
     }
     return "ok";

client/src/stores/collectionElementsStore.test.ts

@@ -125,6 +125,8 @@ function mockCollection(id: string, numElements = 10): HDCASummary {
         id: id,
         element_count: numElements,
         elements_datatypes: ["txt"],
+        elements_deleted: 0,
+        elements_states: {},
         collection_type: "list",
         populated_state: "ok",
         populated_state_message: "",

lib/galaxy/jobs/__init__.py

@@ -1520,9 +1520,7 @@ class MinimalJobWrapper(HasResourceParameters):
                         dep_job_assoc.job,
                         "Execution of this dataset's job is paused because its input datasets are in an error state.",
                     )
-            job.set_final_state(
-                job.states.ERROR, supports_skip_locked=self.app.application_stack.supports_skip_locked()
-            )
+            job.set_final_state(job.states.ERROR)
             job.command_line = self.command_line
             job.info = message
             # TODO: Put setting the stdout, stderr, and exit code in one place
@@ -2292,7 +2290,7 @@ class MinimalJobWrapper(HasResourceParameters):
 
         # Finally set the job state.  This should only happen *after* all
         # dataset creation, and will allow us to eliminate force_history_refresh.
-        job.set_final_state(final_job_state, supports_skip_locked=self.app.application_stack.supports_skip_locked())
+        job.set_final_state(final_job_state)
         if not job.tasks:
             # If job was composed of tasks, don't attempt to recollect statistics
             self._collect_metrics(job, job_metrics_directory)

lib/galaxy/jobs/handler.py

@@ -1086,7 +1086,7 @@ class JobHandlerStopQueue(BaseJobHandlerQueue):
         if error_msg is not None:
             final_state = job.states.ERROR
             job.info = error_msg
-        job.set_final_state(final_state, supports_skip_locked=self.app.application_stack.supports_skip_locked())
+        job.set_final_state(final_state)
         self.sa_session.add(job)
         self.sa_session.flush()
 

lib/galaxy/managers/collections_util.py

@@ -182,6 +182,13 @@ def dictify_element_reference(
             object_details["element_count"] = element_object.element_count
             object_details["populated"] = element_object.populated_optimized
 
+            # Add hierarchical state information for intermediate collections
+            # Add elements_states and intermediate_states from DatasetCollection
+            dataset_summary = element_object.dataset_states_and_extensions_summary
+            object_details["elements_states"] = dataset_summary.states
+            object_details["elements_deleted"] = dataset_summary.deleted
+            object_details["elements_datatypes"] = dataset_summary.extensions
+
             # Recursively yield elements for each nested collection...
             if recursive:
                 elements, rest_fuzzy_counts = get_fuzzy_count_elements(element_object, rank_fuzzy_counts)

lib/galaxy/managers/hdcas.py

@@ -286,6 +286,9 @@ class HDCASerializer(DCASerializer, taggable.TaggableSerializerMixin, annotatabl
                 "populated_state",
                 "populated_state_message",
                 "element_count",
+                "elements_datatypes",
+                "elements_deleted",
+                "elements_states",
                 "job_source_id",
                 "job_source_type",
                 "job_state_summary",
@@ -305,7 +308,6 @@ class HDCASerializer(DCASerializer, taggable.TaggableSerializerMixin, annotatabl
             [
                 "populated",
                 "elements",
-                "elements_datatypes",
             ],
             include_keys_from="summary",
         )
@@ -333,6 +335,8 @@ class HDCASerializer(DCASerializer, taggable.TaggableSerializerMixin, annotatabl
             "contents_url": self.generate_contents_url,
             "job_state_summary": self.serialize_job_state_summary,
             "elements_datatypes": self.serialize_elements_datatypes,
+            "elements_states": lambda item, key, **context: item.dataset_dbkeys_and_extensions_summary[2],
+            "elements_deleted": lambda item, key, **context: item.dataset_dbkeys_and_extensions_summary[3],
             "collection_id": self.serialize_id,
         }
         self.serializers.update(serializers)

lib/galaxy/model/__init__.py

@@ -81,6 +81,7 @@ from sqlalchemy import (
     Integer,
     join,
     JSON,
+    literal,
     MetaData,
     not_,
     Numeric,
@@ -2235,44 +2236,15 @@ class Job(Base, JobLike, UsesCreateAndUpdateTime, Dictifiable, Serializable):
 
         return rval
 
-    def update_hdca_update_time_for_job(self, update_time, sa_session, supports_skip_locked):
-        subq = (
-            sa_session.query(HistoryDatasetCollectionAssociation.id)
-            .join(ImplicitCollectionJobs)
-            .join(ImplicitCollectionJobsJobAssociation)
-            .filter(ImplicitCollectionJobsJobAssociation.job_id == self.id)
-        )
-        if supports_skip_locked:
-            subq = subq.with_for_update(skip_locked=True).subquery()
-        implicit_statement = (
-            HistoryDatasetCollectionAssociation.table.update()
-            .where(HistoryDatasetCollectionAssociation.id.in_(select(subq)))
-            .values(update_time=update_time)
-        )
+    def update_hdca_update_time_for_job(self, update_time, sa_session):
         explicit_statement = (
             HistoryDatasetCollectionAssociation.table.update()
             .where(HistoryDatasetCollectionAssociation.job_id == self.id)
             .values(update_time=update_time)
         )
         sa_session.execute(explicit_statement)
-        if supports_skip_locked:
-            sa_session.execute(implicit_statement)
-        else:
-            conn = sa_session.connection(execution_options={"isolation_level": "SERIALIZABLE"})
-            with conn.begin() as trans:
-                try:
-                    conn.execute(implicit_statement)
-                    trans.commit()
-                except OperationalError as e:
-                    # If this is a serialization failure on PostgreSQL, then e.orig is a psycopg2 TransactionRollbackError
-                    # and should have attribute `code`. Other engines should just report the message and move on.
-                    if int(getattr(e.orig, "pgcode", -1)) != 40001:
-                        log.debug(
-                            f"Updating implicit collection uptime_time for job {self.id} failed (this is expected for large collections and not a problem): {unicodify(e)}"
-                        )
-                    trans.rollback()
 
-    def set_final_state(self, final_state, supports_skip_locked):
+    def set_final_state(self, final_state):
         self.set_state(final_state)
         # TODO: migrate to where-in subqueries?
         statement = text(
@@ -2284,9 +2256,7 @@ class Job(Base, JobLike, UsesCreateAndUpdateTime, Dictifiable, Serializable):
         )
         sa_session = required_object_session(self)
         update_time = now()
-        self.update_hdca_update_time_for_job(
-            update_time=update_time, sa_session=sa_session, supports_skip_locked=supports_skip_locked
-        )
+        self.update_hdca_update_time_for_job(update_time=update_time, sa_session=sa_session)
         params = {"job_id": self.id, "update_time": update_time}
         sa_session.execute(statement, params)
 
@@ -2350,9 +2320,22 @@ class Job(Base, JobLike, UsesCreateAndUpdateTime, Dictifiable, Serializable):
         ]
         sa_session = required_object_session(self)
         update_time = now()
+
+        # Update job-specific collection associations
         self.update_hdca_update_time_for_job(
-            update_time=update_time, sa_session=sa_session, supports_skip_locked=supports_skip_locked
+            update_time=update_time,
+            sa_session=sa_session,
         )
+        # Update general collection associations for all datasets in this job
+        # This handles collections that contain datasets from this job but aren't directly job-associated
+        job_datasets = sa_session.query(Dataset).filter(Dataset.job_id == self.id).all()
+        for dataset in job_datasets:
+            try:
+                dataset.touch_collection_update_time(supports_skip_locked=supports_skip_locked)
+            except Exception:
+                # Don't let collection update failures prevent job comion
+                log.error(f"Failed to update collection times for dataset {dataset.id} from job {self.id}")
+
         params = {"job_id": self.id, "state": self.state, "info": self.info, "update_time": update_time}
         for statement in statements:
             sa_session.execute(statement, params)
@@ -4670,6 +4653,178 @@ class Dataset(Base, StorableObject, Serializable):
         serialization_options.attach_identifier(id_encoder, self, rval)
         return rval
 
+    def touch_collection_update_time(self, supports_skip_locked=None):
+        """
+        Update time is important to determine for what history items we need to fetch
+        updated data. This method updates all collection associations that reference
+        this dataset through any of its dataset instances (HDAs, LDDAs, etc.).
+
+        Uses database-specific recursive CTEs for optimal performance.
+
+        Args:
+            supports_skip_locked: Boolean to indicate if database supports
+                FOR UPDATE SKIP LOCKED. Defaults to False.
+        """
+        session = object_session(self)
+        if not session:
+            return
+        dialect_name = session.bind.dialect.name
+
+        if dialect_name in ("postgresql", "sqlite"):
+            self._touch_collection_update_time_cte(session, supports_skip_locked)
+        else:
+            # Fallback for other databases
+            self._touch_collection_update_time_fallback(session)
+
+    def _touch_collection_update_time_cte(self, session, supports_skip_locked):
+        """CTE implementation using SQLAlchemy core constructs"""
+
+        # Base case columns
+        base_columns = [
+            DatasetCollectionElement.dataset_collection_id.label("collection_id"),
+            literal(0).label("depth_level"),
+        ]
+
+        # Create a single base query that covers both HDA and LDDA cases using OR conditions
+        base_query = (
+            select(*base_columns)
+            .select_from(
+                DatasetCollectionElement.__table__.outerjoin(
+                    HistoryDatasetAssociation.__table__, DatasetCollectionElement.hda_id == HistoryDatasetAssociation.id
+                ).outerjoin(
+                    LibraryDatasetDatasetAssociation.__table__,
+                    DatasetCollectionElement.ldda_id == LibraryDatasetDatasetAssociation.id,
+                )
+            )
+            .where(
+                or_(
+                    HistoryDatasetAssociation.dataset_id == self.id,
+                    LibraryDatasetDatasetAssociation.dataset_id == self.id,
+                )
+            )
+        )
+
+        # Create the recursive CTE from the single base query
+        collection_hierarchy_cte = base_query.cte(name="collection_hierarchy", recursive=True)
+
+        # Create aliases for the recursive part
+        ch_alias = aliased(collection_hierarchy_cte, name="ch")
+        parent_dce_alias = aliased(DatasetCollectionElement, name="parent_dce")
+
+        # Recursive case columns - conditional based on database support
+        recursive_columns = [
+            parent_dce_alias.dataset_collection_id.label("collection_id"),
+            (ch_alias.c.depth_level + 1).label("depth_level"),
+        ]
+
+        # Recursive case conditions
+        recursive_conditions = [
+            parent_dce_alias.child_collection_id == ch_alias.c.collection_id,
+            ch_alias.c.depth_level < 50,
+        ]
+
+        # Recursive case: Parent collections
+        recursive_query = select(*recursive_columns).where(and_(*recursive_conditions))
+
+        # Add the recursive part to the CTE
+        collection_hierarchy_cte = collection_hierarchy_cte.union_all(recursive_query)
+
+        # Create the final subquery that selects and orders collection IDs
+        collection_ids_subquery = select(collection_hierarchy_cte.c.collection_id).order_by(
+            collection_hierarchy_cte.c.collection_id
+        )
+
+        if supports_skip_locked:
+            # Use FOR UPDATE SKIP LOCKED for better concurrency when supported
+            # First select the HDCAs we want to update with FOR UPDATE SKIP LOCKED
+            hdca_ids_to_update = (
+                session.execute(
+                    select(HistoryDatasetCollectionAssociation.id)
+                    .where(HistoryDatasetCollectionAssociation.collection_id.in_(collection_ids_subquery))
+                    .order_by(HistoryDatasetCollectionAssociation.id)
+                    .with_for_update(skip_locked=True)
+                )
+                .scalars()
+                .all()
+            )
+
+            # Then update only the locked HDCAs
+            if hdca_ids_to_update:
+                update_stmt = (
+                    update(HistoryDatasetCollectionAssociation)
+                    .where(HistoryDatasetCollectionAssociation.id.in_(hdca_ids_to_update))
+                    .values(update_time=now())
+                )
+                session.execute(update_stmt)
+        else:
+            # Fallback to regular update without locking
+            update_stmt = (
+                update(HistoryDatasetCollectionAssociation)
+                .where(HistoryDatasetCollectionAssociation.collection_id.in_(collection_ids_subquery))
+                .values(update_time=now())
+            )
+            session.execute(update_stmt)
+
+    def _touch_collection_update_time_fallback(self, session):
+        """Python-based fallback for databases without good CTE support"""
+        visited_collections = set()
+        hdcas_to_update = []
+
+        def collect_parent_hdcas(collection_id, depth=0):
+            if collection_id in visited_collections or depth > 50:
+                return
+
+            visited_collections.add(collection_id)
+
+            # Find HDCAs for this collection - ordered by ID for consistency
+            hdca_stmt = (
+                select(HistoryDatasetCollectionAssociation)
+                .where(HistoryDatasetCollectionAssociation.collection_id == collection_id)
+                .order_by(HistoryDatasetCollectionAssociation.id)
+            )
+            hdcas = session.scalars(hdca_stmt).all()
+            hdcas_to_update.extend(hdcas)
+
+            # Find parent collections - ordered by ID for consistency
+            parent_stmt = (
+                select(DatasetCollectionElement.dataset_collection_id)
+                .where(DatasetCollectionElement.child_collection_id == collection_id)
+                .order_by(DatasetCollectionElement.dataset_collection_id)
+            )
+            parent_collection_ids = session.scalars(parent_stmt).all()
+
+            for parent_id in sorted(parent_collection_ids):
+                collect_parent_hdcas(parent_id, depth + 1)
+
+        # Start with collections containing this dataset - ordered by ID for consistency
+        initial_stmt = (
+            select(DatasetCollectionElement.dataset_collection_id)
+            .where(
+                or_(
+                    DatasetCollectionElement.hda_id.in_(
+                        select(HistoryDatasetAssociation.id).where(HistoryDatasetAssociation.dataset_id == self.id)
+                    ),
+                    DatasetCollectionElement.ldda_id.in_(
+                        select(LibraryDatasetDatasetAssociation.id).where(
+                            LibraryDatasetDatasetAssociation.dataset_id == self.id
+                        )
+                    ),
+                )
+            )
+            .order_by(DatasetCollectionElement.dataset_collection_id)
+        )
+
+        initial_collection_ids = session.scalars(initial_stmt).all()
+
+        for collection_id in sorted(initial_collection_ids):
+            collect_parent_hdcas(collection_id)
+
+        # Update all collected HDCAs - sort by ID for consistent ordering
+        update_time = now()
+        for hdca in sorted(hdcas_to_update, key=lambda x: x.id):
+            hdca.update_time = update_time
+            session.add(hdca)
+
 
 class DatasetSource(Base, Dictifiable, Serializable):
     __tablename__ = "dataset_source"
@@ -4915,19 +5070,29 @@ class DatasetInstance(RepresentById, UsesCreateAndUpdateTime, _HasTable):
         # self._state holds state that should only affect this particular dataset association, not the dataset state itself
         if self._state:
             return self._state
-        return self.dataset.state
+        return self.dataset.state if self.dataset else None
 
     @state.setter
-    def state(self, state: Optional[DatasetState]):
+    def state(self, state: DatasetState):
         if state != self.state:
             if state in (DatasetState.FAILED_METADATA, DatasetState.SETTING_METADATA):
                 self._state = state
             else:
                 self.set_metadata_success_state()
+                assert self.dataset, "Dataset must be set before setting state"
                 sa_session = object_session(self)
                 if sa_session:
                     sa_session.add(self.dataset)
                 self.dataset.state = state
+                self.touch_collection_update_time()
+
+    def touch_collection_update_time(self):
+        """
+        Update time is important to determine for what history items we need to fetch
+        updated data.
+        """
+        if self.dataset:
+            self.dataset.touch_collection_update_time()
 
     def set_metadata_success_state(self):
         self._state = None
@@ -6673,6 +6838,13 @@ class InnerCollectionFilter(NamedTuple):
         return self.operator_function(getattr(table, self.column), self.expected_value)
 
 
+class CollectionStateSummary(NamedTuple):
+    dbkeys: list[Union[str, None]]
+    extensions: list[str]
+    states: dict[str, int]
+    deleted: int
+
+
 class DatasetCollection(Base, Dictifiable, UsesAnnotations, Serializable):
     __tablename__ = "dataset_collection"
 
@@ -6694,8 +6866,14 @@ class DatasetCollection(Base, Dictifiable, UsesAnnotations, Serializable):
         order_by=lambda: DatasetCollectionElement.element_index,
     )
 
-    dict_collection_visible_keys = ["id", "collection_type"]
-    dict_element_visible_keys = ["id", "collection_type"]
+    dict_collection_visible_keys = [
+        "id",
+        "collection_type",
+        "elements_datatypes",
+        "elements_states",
+        "elements_deleted",
+    ]
+    dict_element_visible_keys = ["id", "collection_type", "elements_datatypes", "elements_states", "elements_deleted"]
 
     populated_states = DatasetCollectionPopulatedState
 
@@ -6807,12 +6985,8 @@ class DatasetCollection(Base, Dictifiable, UsesAnnotations, Serializable):
     @property
     def elements_deleted(self):
         if not hasattr(self, "_elements_deleted"):
-            if session := object_session(self):
-                stmt = self._build_nested_collection_attributes_stmt(
-                    hda_attributes=("deleted",), dataset_attributes=("deleted",)
-                )
-                stmt = stmt.exists().where(or_(HistoryDatasetAssociation.deleted == true(), Dataset.deleted == true()))
-                self._elements_deleted = session.execute(select(stmt)).scalar()
+            if object_session(self):
+                self._elements_deleted = self.dataset_states_and_extensions_summary.deleted
             else:
                 self._elements_deleted = False
                 for dataset_instance in self.dataset_instances:
@@ -6822,40 +6996,76 @@ class DatasetCollection(Base, Dictifiable, UsesAnnotations, Serializable):
         return self._elements_deleted
 
     @property
-    def dataset_states_and_extensions_summary(self):
-        if not hasattr(self, "_dataset_states_and_extensions_summary"):
-            stmt = self._build_nested_collection_attributes_stmt(
-                hda_attributes=("extension",), dataset_attributes=("state",)
-            )
-            # With DISTINCT, all columns that appear in the ORDER BY clause must appear in the SELECT clause.
-            stmt = stmt.add_columns(*stmt._order_by_clauses)
-            stmt = stmt.distinct()
+    def elements_datatypes(self):
+        if not hasattr(self, "_elements_datatypes"):
+            if object_session(self):
+                self._elements_datatypes = self.dataset_states_and_extensions_summary.extensions
+            else:
+                self._elements_datatypes = defaultdict(int)
+                for dataset_instance in self.dataset_instances:
+                    if dataset_instance.extension:
+                        self._elements_datatypes[dataset_instance.extension] += 1
+        return self._elements_datatypes
 
+    @property
+    def elements_states(self):
+        if not hasattr(self, "_elements_states"):
+            if object_session(self):
+                self._elements_states = self.dataset_states_and_extensions_summary.states
+            else:
+                self._elements_states = defaultdict(int)
+                for dataset_instance in self.dataset_instances:
+                    if dataset_instance.state:
+                        self._elements_states[dataset_instance.state] += 1
+        return self._elements_states
+
+    @property
+    def dataset_states_and_extensions_summary(self):
+        """
+        Calculate dataset states and extensions for this collection.
+        Returns (dbkeys, extensions, states, deleted) similar to HDCA method.
+        """
+        if not hasattr(self, "_dataset_states_and_extensions_summary"):
+
+            stmt = self._build_nested_collection_attributes_stmt(
+                hda_attributes=("_metadata", "extension", "deleted"), dataset_attributes=("state",)
+            )
             tuples = required_object_session(self).execute(stmt)
 
             extensions = set()
-            states = set()
-            for extension, state, *_ in tuples:  # we discard the added columns from the order-by clause
-                states.add(state)
-                extensions.add(extension)
-
-            self._dataset_states_and_extensions_summary = (states, extensions)
+            dbkeys = set()
+            states = defaultdict(int)
+            deleted = 0
+            for row in tuples:
+                if row is not None:
+                    dbkey_field = row._metadata.get("dbkey")
+                    if isinstance(dbkey_field, list):
+                        for dbkey in dbkey_field:
+                            dbkeys.add(dbkey)
+                    else:
+                        dbkeys.add(dbkey_field)
+                    if row.extension:
+                        extensions.add(row.extension)
+                    if row.deleted:
+                        deleted += 1
+                    if row.state:
+                        states[row.state] += 1
+            # Filter out None values before sorting
+            filtered_dbkeys = (dbkey for dbkey in dbkeys if dbkey is not None)
+            filtered_extensions = (ext for ext in extensions if ext is not None)
 
+            self._dataset_states_and_extensions_summary = CollectionStateSummary(
+                sorted(filtered_dbkeys), sorted(filtered_extensions), states, deleted
+            )
         return self._dataset_states_and_extensions_summary
 
     @property
     def has_deferred_data(self):
         if not hasattr(self, "_has_deferred_data"):
-            has_deferred_data = False
-            if session := object_session(self):
-                # TODO: Optimize by just querying without returning the states...
-                stmt = self._build_nested_collection_attributes_stmt(dataset_attributes=("state",))
-                tuples = session.execute(stmt)
-                for (state,) in tuples:
-                    if state == Dataset.states.DEFERRED:
-                        has_deferred_data = True
-                        break
+            if object_session(self):
+                has_deferred_data = Dataset.states.DEFERRED in self.dataset_states_and_extensions_summary.states
             else:
+                has_deferred_data = False
                 # This will be in a remote tool evaluation context, so can't query database
                 for dataset_element in self.dataset_elements_and_identifiers():
                     if dataset_element.hda.state == Dataset.states.DEFERRED:
@@ -7163,7 +7373,9 @@ class DatasetCollectionInstance(HasName, UsesCreateAndUpdateTime):
             populated_state=self.collection.populated_state,
             populated_state_message=self.collection.populated_state_message,
             element_count=self.collection.element_count,
-            elements_datatypes=list(self.dataset_dbkeys_and_extensions_summary[1]),
+            elements_datatypes=list(self.collection.elements_datatypes),
+            elements_deleted=self.collection.elements_deleted,
+            elements_states=self.collection.elements_states,
             type="collection",  # contents type (distinguished from file or folder (in case of library))
         )
 
@@ -7369,11 +7581,15 @@ class HistoryDatasetCollectionAssociation(
     @property
     def dataset_dbkeys_and_extensions_summary(self):
         if not hasattr(self, "_dataset_dbkeys_and_extensions_summary"):
-            stmt = self.collection._build_nested_collection_attributes_stmt(hda_attributes=("_metadata", "extension"))
+            stmt = self.collection._build_nested_collection_attributes_stmt(
+                hda_attributes=("_metadata", "extension", "deleted"), dataset_attributes=("state",)
+            )
             tuples = required_object_session(self).execute(stmt)
 
             extensions = set()
             dbkeys = set()
+            states = defaultdict(int)
+            deleted = 0
             for row in tuples:
                 if row is not None:
                     dbkey_field = row._metadata.get("dbkey")
@@ -7384,7 +7600,11 @@ class HistoryDatasetCollectionAssociation(
                         dbkeys.add(dbkey_field)
                     if row.extension:
                         extensions.add(row.extension)
-            self._dataset_dbkeys_and_extensions_summary = (dbkeys, extensions)
+                    if row.deleted:
+                        deleted += 1
+                    if row.state:
+                        states[row.state] += 1
+            self._dataset_dbkeys_and_extensions_summary = (dbkeys, extensions, states, deleted)
         return self._dataset_dbkeys_and_extensions_summary
 
     @property
@@ -7458,7 +7678,7 @@ class HistoryDatasetCollectionAssociation(
     def to_dict(self, view="collection"):
         original_dict_value = super().to_dict(view=view)
         if view == "dbkeysandextensions":
-            (dbkeys, extensions) = self.dataset_dbkeys_and_extensions_summary
+            (dbkeys, extensions, *_) = self.dataset_dbkeys_and_extensions_summary
             dict_value = dict(
                 dbkey=dbkeys.pop() if len(dbkeys) == 1 else "?",
                 extension=extensions.pop() if len(extensions) == 1 else "auto",

lib/galaxy/model/dataset_collections/adapters.py

@@ -1,3 +1,4 @@
+from collections import defaultdict
 from typing import (
     TYPE_CHECKING,
 )
@@ -5,6 +6,7 @@ from typing import (
 from pydantic import ValidationError
 from typing_extensions import Self
 
+from galaxy.model import CollectionStateSummary
 from galaxy.tool_util_models.parameters import (
     AdaptedDataCollectionPromoteCollectionElementToCollectionRequestInternal,
     AdaptedDataCollectionPromoteDatasetsToCollectionRequestInternal,
@@ -88,11 +90,11 @@ class DCECollectionAdapter(CollectionAdapter):
             return self._dce.child_collection.dataset_states_and_extensions_summary
         else:
             hda = self._dce.dataset_instance
-            extensions = set()
-            states = set()
-            states.add(hda.dataset.state)
-            extensions.add(hda.extension)
-            return (states, extensions)
+            dbkeys = [hda.dbkey] if hda.dbkey else []
+            extensions = [hda.extension] if hda.extension else []
+            states = {hda.dataset.state: 1} if hda.dataset.state else {}
+            deleted = 1 if hda.deleted or (hda.dataset and hda.dataset.deleted) else 0
+            return CollectionStateSummary(dbkeys=dbkeys, extensions=extensions, states=states, deleted=deleted)
 
     @property
     def dataset_instances(self):
@@ -161,11 +163,11 @@ class PromoteDatasetToCollection(CollectionAdapter):
     @property
     def dataset_states_and_extensions_summary(self):
         hda = self._hda
-        extensions = set()
-        states = set()
-        states.add(hda.dataset.state)
-        extensions.add(hda.extension)
-        return (states, extensions)
+        dbkeys = [hda.dbkey] if hda.dbkey else []
+        extensions = [hda.extension] if hda.extension else []
+        states = {hda.dataset.state: 1} if hda.dataset.state else {}
+        deleted = 1 if hda.deleted or (hda.dataset and hda.dataset.deleted) else 0
+        return CollectionStateSummary(dbkeys=dbkeys, extensions=extensions, states=states, deleted=deleted)
 
     @property
     def dataset_instances(self):
@@ -234,12 +236,22 @@ class PromoteDatasetsToCollection(CollectionAdapter):
 
     @property
     def dataset_states_and_extensions_summary(self):
+        dbkeys = set()
         extensions = set()
-        states = set()
+        states: dict[str, int] = defaultdict(int)
+        deleted = 0
         for hda in self.dataset_instances:
-            states.add(hda.dataset.state)
-            extensions.add(hda.extension)
-        return (states, extensions)
+            if hda.dbkey:
+                dbkeys.add(hda.dbkey)
+            if hda.extension:
+                extensions.add(hda.extension)
+            if hda.dataset.state:
+                states[hda.dataset.state] += 1
+            if hda.deleted or (hda.dataset and hda.dataset.deleted):
+                deleted += 1
+        return CollectionStateSummary(
+            dbkeys=sorted(dbkeys), extensions=sorted(extensions), states=states, deleted=deleted
+        )
 
     @property
     def adapting(self):

lib/galaxy/schema/schema.py

@@ -97,6 +97,23 @@ class DatasetState(str, Enum):
         return self.__members__.values()
 
 
+# Create dictionary for ElementsStatesDict using class syntax
+class ElementsStatesDict(TypedDict, total=False):
+    # Add fields for each DatasetState value
+    new: NotRequired[int]
+    upload: NotRequired[int]
+    queued: NotRequired[int]
+    running: NotRequired[int]
+    ok: NotRequired[int]
+    empty: NotRequired[int]
+    error: NotRequired[int]
+    paused: NotRequired[int]
+    setting_metadata: NotRequired[int]
+    failed_metadata: NotRequired[int]
+    deferred: NotRequired[int]
+    discarded: NotRequired[int]
+
+
 class JobState(str, Enum):
     NEW = "new"
     RESUBMITTED = "resubmitted"
@@ -1032,6 +1049,17 @@ class DCObject(Model, WithModelClass):
     element_count: ElementCountField
     contents_url: Optional[ContentsUrlField] = None
     elements: list["DCESummary"] = ElementsField
+    elements_states: ElementsStatesDict = Field(
+        ..., description="A dictionary containing counts for each dataset state in the collection."
+    )
+    elements_deleted: int = Field(
+        ...,
+        title="Datasets deleted",
+        description="The number of elements in the collection that are marked as deleted.",
+    )
+    elements_datatypes: set[str] = Field(
+        ..., description="A set containing all the different element datatypes in the collection."
+    )
 
 
 class DCESummary(Model, WithModelClass):
@@ -1180,6 +1208,14 @@ class HDCASummary(HDCACommon, WithModelClass):
     elements_datatypes: set[str] = Field(
         ..., description="A set containing all the different element datatypes in the collection."
     )
+    elements_states: ElementsStatesDict = Field(
+        ..., description="A dictionary containing counts for each dataset state in the collection."
+    )
+    elements_deleted: int = Field(
+        ...,
+        title="Datasets deleted",
+        description="The number of elements in the collection that are marked as deleted.",
+    )
     job_source_id: Optional[EncodedDatabaseIdField] = Field(
         None,
         title="Job Source ID",

lib/galaxy/tools/__init__.py

@@ -3619,8 +3619,9 @@ class DatabaseOperationTool(Tool):
                 if not input_dataset_collection.collection.populated_optimized:
                     raise ToolInputsNotReadyException("An input collection is not populated.")
 
-            states, _ = input_dataset_collection.collection.dataset_states_and_extensions_summary
-            for state in states:
+            summary = input_dataset_collection.collection.dataset_states_and_extensions_summary
+            states = summary.states
+            for state in states.keys():
                 check_dataset_state(input_key, state)
 
     def _add_datasets_to_history(self, history, elements, datasets_visible=False):

lib/galaxy/tools/actions/__init__.py

@@ -293,7 +293,8 @@ class DefaultToolAction(ToolAction):
                 for action, role_id in action_tuples:
                     record_permission(action, role_id)
 
-                _, extensions = collection.dataset_states_and_extensions_summary
+                summary = collection.dataset_states_and_extensions_summary
+                extensions = summary.extensions
                 conversion_required = False
                 for ext in extensions:
                     if ext:

lib/galaxy/tools/parameters/dataset_matcher.py

@@ -213,8 +213,10 @@ class SummaryDatasetCollectionMatcher:
         if not dataset_collection.populated_optimized:
             return False
 
-        (states, extensions) = dataset_collection.dataset_states_and_extensions_summary
-        for state in states:
+        summary = dataset_collection.dataset_states_and_extensions_summary
+        states = summary.states
+        extensions = summary.extensions
+        for state in states.keys():
             if state not in self.dataset_matcher_factory.valid_input_states:
                 return False
 

lib/galaxy_test/selenium/test_uploads.py

@@ -499,7 +499,7 @@ PRJDA60709  SAMD00016382    DRX000480   ftp.sra.ebi.ac.uk/vol1/fastq/DRR000/DRR0
         self.screenshot("rules_deferred_list_7_named")
         rule_builder.main_button_ok.wait_for_and_click()
         hid = 2
-        self.history_panel_wait_for_hid_ok(hid)
+        self.history_panel_wait_for_hid_state(hid, state="deferred", allowed_force_refreshes=1)
         self.screenshot("rules_deferred_list_7_download_complete")
 
     def _read_rules_test_data_file(self, name):

test/unit/app/managers/test_markdown_export.py

@@ -375,10 +375,15 @@ history_dataset_display(history_dataset_id=2)
 history_dataset_collection_display(history_dataset_collection_id=1)
 ```
 """
-        # Patch out url_for since we haven't initialized an app.
+        # Mock the HDCASerializer to return a dummy serialized view
         from galaxy.managers.hdcas import HDCASerializer
 
-        with mock.patch.object(HDCASerializer, "url_for", return_value="http://google.com"):
+        mock_hdca_view = {"id": "encoded_id_1", "name": "cool name", "collection_type": "paired"}
+
+        with (
+            mock.patch.object(HDCASerializer, "url_for", return_value="http://google.com"),
+            mock.patch.object(HDCASerializer, "serialize_to_view", return_value=mock_hdca_view),
+        ):
             export, extra_data = self._ready_export(example)
         assert "history_dataset_collections" in extra_data
         assert len(extra_data.get("history_dataset_collections")) == 1

test/unit/app/tools/test_select_parameters.py

@@ -9,17 +9,23 @@ from .util import BaseParameterTestCase
 
 
 class TestSelectToolParameter(BaseParameterTestCase):
+
+    def new_hda(self):
+        hda = model.HistoryDatasetAssociation()
+        hda._state = model.Dataset.states.OK
+        return hda
+
     def test_validated_values(self):
         self.options_xml = """<options><filter type="data_meta" ref="input_bam" key="dbkey"/></options>"""
         with pytest.raises(ValueError) as exc_info:
-            self.param.from_json("42", self.trans, {"input_bam": model.HistoryDatasetAssociation()})
-            assert str(exc_info.value) == "parameter 'my_name': requires a value, but no legal values defined"
+            self.param.from_json("42", self.trans, {"input_bam": self.new_hda()})
+        assert str(exc_info.value) == "Parameter 'my_name': requires a value, but no legal values defined"
 
     def test_validated_values_missing_dependency(self):
         self.options_xml = """<options><filter type="data_meta" ref="input_bam" key="dbkey"/></options>"""
         with pytest.raises(ValueError) as exc_info:
             self.param.from_json("42", self.trans)
-            assert str(exc_info.value) == "parameter 'my_name': requires a value, but no legal values defined"
+        assert str(exc_info.value) == "Parameter 'my_name': requires a value, but no legal values defined"
 
     def test_unvalidated_values(self):
         self.options_xml = """<options><filter type="data_meta" ref="input_bam" key="dbkey"/></options>"""
@@ -29,14 +35,14 @@ class TestSelectToolParameter(BaseParameterTestCase):
     def test_validated_datasets(self):
         self.options_xml = """<options><filter type="data_meta" ref="input_bam" key="dbkey"/></options>"""
         with pytest.raises(ValueError) as exc_info:
-            self.param.from_json(model.HistoryDatasetAssociation(), self.trans, {"input_bam": None})
-            assert str(exc_info.value) == "parameter 'my_name': requires a value, but no legal values defined"
+            self.param.from_json(self.new_hda(), self.trans, {"input_bam": None})
+        assert str(exc_info.value) == "Parameter 'my_name': requires a value, but no legal values defined"
 
     def test_unvalidated_datasets(self):
         self.options_xml = """<options><filter type="data_meta" ref="input_bam" key="dbkey"/></options>"""
         self.trans.workflow_building_mode = True
         assert isinstance(
-            self.param.from_json(model.HistoryDatasetAssociation(), self.trans, {"input_bam": RuntimeValue()}),
+            self.param.from_json(self.new_hda(), self.trans, {"input_bam": RuntimeValue()}),
             model.HistoryDatasetAssociation,
         )
 

test/unit/data/model/test_touch_collection_update_time.py

@@ -0,0 +1,356 @@
+"""
+Unit tests for Dataset.touch_collection_update_time() functionality.
+
+Tests both the recursive CTE approach and the Python fallback implementation
+across different database backends and collection hierarchy scenarios.
+"""
+
+import uuid
+from unittest.mock import patch
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from galaxy import model
+from galaxy.model.unittest_utils.model_testing_utils import persist
+
+
+@pytest.fixture(scope="module")
+def engine():
+    """Create test database engine."""
+    db_uri = "sqlite:///:memory:"
+    return create_engine(db_uri)
+
+
+@pytest.fixture(scope="module")
+def init_model(engine):
+    """Create model objects in the engine's database."""
+    model.mapper_registry.metadata.create_all(engine)
+
+
+@pytest.fixture
+def session(init_model, engine):
+    """Create test database session with all Galaxy model tables."""
+    with Session(engine) as s:
+        yield s
+
+
+@pytest.fixture
+def user(session):
+    """Create a test user."""
+    unique_id = uuid.uuid4().hex[:8]
+    user = model.User(username=f"test_user_{unique_id}", email=f"test_{unique_id}@example.com", password="password123")
+    user_id = persist(session, user)
+    return session.get(model.User, user_id)
+
+
+@pytest.fixture
+def history(session, user):
+    """Create a test history."""
+    history = model.History(name="Test History", user=user)
+    history_id = persist(session, history)
+    return session.get(model.History, history_id)
+
+
+@pytest.fixture
+def dataset(session):
+    """Create a test dataset."""
+    dataset = model.Dataset(state="ok")
+    dataset_id = persist(session, dataset)
+    return session.get(model.Dataset, dataset_id)
+
+
+@pytest.fixture
+def hda(session, history, dataset):
+    """Create a test HistoryDatasetAssociation."""
+    hda = model.HistoryDatasetAssociation(dataset=dataset, history=history, name="Test Dataset", hid=1)
+    hda_id = persist(session, hda)
+    return session.get(model.HistoryDatasetAssociation, hda_id)
+
+
+def create_simple_collection_hierarchy(session, hda):
+    """
+    Create a simple collection hierarchy:
+    HDCA -> Collection -> DCE -> HDA
+    """
+    # Create a collection
+    collection = model.DatasetCollection(collection_type="list")
+    collection_id = persist(session, collection)
+    collection = session.get(model.DatasetCollection, collection_id)
+
+    # Create collection element linking collection to HDA
+    dce = model.DatasetCollectionElement(
+        collection=collection, element=hda, element_identifier="item1", element_index=0
+    )
+    persist(session, dce)
+
+    # Create HDCA
+    hdca = model.HistoryDatasetCollectionAssociation(
+        collection_id=collection.id, history_id=hda.history.id, name="Test Collection", hid=2
+    )
+    hdca_id = persist(session, hdca)
+    return session.get(model.HistoryDatasetCollectionAssociation, hdca_id)
+
+
+def create_nested_collection_hierarchy(session, hda):
+    """
+    Create a nested collection hierarchy:
+    HDCA -> Collection1 -> DCE -> Collection2 -> DCE -> HDA
+    """
+    # Create inner collection (contains the HDA)
+    inner_collection = model.DatasetCollection(collection_type="list")
+    inner_collection_id = persist(session, inner_collection)
+    inner_collection = session.get(model.DatasetCollection, inner_collection_id)
+
+    # Create DCE linking inner collection to HDA
+    inner_dce = model.DatasetCollectionElement(
+        collection=inner_collection, element=hda, element_identifier="inner_item", element_index=0
+    )
+    persist(session, inner_dce)
+
+    # Create outer collection
+    outer_collection = model.DatasetCollection(collection_type="list:list")
+    outer_collection_id = persist(session, outer_collection)
+    outer_collection = session.get(model.DatasetCollection, outer_collection_id)
+
+    # Create DCE linking outer collection to inner collection
+    outer_dce = model.DatasetCollectionElement(
+        collection=outer_collection, element=inner_collection, element_identifier="outer_item", element_index=0
+    )
+    persist(session, outer_dce)
+
+    # Create HDCA for outer collection
+    hdca = model.HistoryDatasetCollectionAssociation(
+        collection_id=outer_collection.id, history_id=hda.history.id, name="Nested Collection", hid=3
+    )
+    hdca_id = persist(session, hdca)
+    return session.get(model.HistoryDatasetCollectionAssociation, hdca_id)
+
+
+def create_multiple_hdca_hierarchy(session, hda):
+    """
+    Create a hierarchy with multiple HDCAs referencing the same collection:
+    HDCA1 -> Collection -> DCE -> HDA
+    HDCA2 -> Collection (same as above)
+    """
+    # Create a collection
+    collection = model.DatasetCollection(collection_type="list")
+    collection_id = persist(session, collection)
+    collection = session.get(model.DatasetCollection, collection_id)
+
+    # Create collection element linking collection to HDA
+    dce = model.DatasetCollectionElement(
+        collection=collection, element=hda, element_identifier="shared_item", element_index=0
+    )
+    persist(session, dce)
+
+    # Create first HDCA
+    hdca1 = model.HistoryDatasetCollectionAssociation(
+        collection_id=collection.id, history_id=hda.history.id, name="Shared Collection 1", hid=4
+    )
+    hdca1_id = persist(session, hdca1)
+
+    # Create second HDCA referencing the same collection
+    hdca2 = model.HistoryDatasetCollectionAssociation(
+        collection_id=collection.id, history_id=hda.history.id, name="Shared Collection 2", hid=5
+    )
+    hdca2_id = persist(session, hdca2)
+
+    return (
+        session.get(model.HistoryDatasetCollectionAssociation, hdca1_id),
+        session.get(model.HistoryDatasetCollectionAssociation, hdca2_id),
+    )
+
+
+class TestTouchCollectionUpdateTime:
+    """Test cases for Dataset.touch_collection_update_time() method."""
+
+    def test_no_collections(self, session, dataset):
+        """Test that touch_collection_update_time works when dataset is not in any collections."""
+        # Should not raise any errors
+        dataset.touch_collection_update_time()
+        session.commit()
+
+    def test_simple_collection_hierarchy(self, session, hda):
+        """Test updating a simple collection hierarchy."""
+        hdca = create_simple_collection_hierarchy(session, hda)
+
+        # Record original update time
+        original_time = hdca.update_time
+
+        # Touch collection update time
+        hda.dataset.touch_collection_update_time()
+        session.commit()
+
+        # Refresh from database
+        session.refresh(hdca)
+
+        # Verify update time was changed
+        assert hdca.update_time > original_time
+
+    def test_nested_collection_hierarchy(self, session, hda):
+        """Test updating a nested collection hierarchy."""
+        hdca = create_nested_collection_hierarchy(session, hda)
+
+        # Record original update time
+        original_time = hdca.update_time
+
+        # Touch collection update time
+        hda.dataset.touch_collection_update_time()
+        session.commit()
+
+        # Refresh from database
+        session.refresh(hdca)
+
+        # Verify update time was changed
+        assert hdca.update_time > original_time
+
+    def test_multiple_hdcas_same_collection(self, session, hda):
+        """Test updating multiple HDCAs that reference the same collection."""
+        hdca1, hdca2 = create_multiple_hdca_hierarchy(session, hda)
+
+        # Record original update times
+        original_time1 = hdca1.update_time
+        original_time2 = hdca2.update_time
+
+        # Touch collection update time
+        hda.dataset.touch_collection_update_time()
+        session.commit()
+
+        # Refresh from database
+        session.refresh(hdca1)
+        session.refresh(hdca2)
+
+        # Verify both HDCAs were updated
+        assert hdca1.update_time > original_time1
+        assert hdca2.update_time > original_time2
+
+    def test_multiple_hdas_same_dataset(self, session, dataset, history, user):
+        """Test updating collections when dataset has multiple HDAs."""
+        # Create two HDAs for the same dataset
+        hda1 = model.HistoryDatasetAssociation(dataset=dataset, history=history, name="Test Dataset 1", hid=1)
+        persist(session, hda1)
+
+        hda2 = model.HistoryDatasetAssociation(dataset=dataset, history=history, name="Test Dataset 2", hid=2)
+        persist(session, hda2)
+
+        # Create collections for each HDA
+        hdca1 = create_simple_collection_hierarchy(session, hda1)
+        hdca2 = create_simple_collection_hierarchy(session, hda2)
+
+        # Record original update times
+        original_time1 = hdca1.update_time
+        original_time2 = hdca2.update_time
+
+        # Touch collection update time on the dataset
+        dataset.touch_collection_update_time()
+        session.commit()
+
+        # Refresh from database
+        session.refresh(hdca1)
+        session.refresh(hdca2)
+
+        # Verify both HDCAs were updated
+        assert hdca1.update_time > original_time1
+        assert hdca2.update_time > original_time2
+
+    def test_cte_approach(self, session, hda):
+        """Test CTE implementation."""
+        hdca = create_simple_collection_hierarchy(session, hda)
+        original_time = hdca.update_time
+        # This should use the SQLite CTE path
+        hda.dataset.touch_collection_update_time()
+        session.commit()
+        assert hdca.update_time > original_time
+
+    def test_fallback_approach(self, session, hda):
+        """Test Python fallback implementation updates HDCA properly."""
+        hdca = create_simple_collection_hierarchy(session, hda)
+
+        # Record original update time
+        original_time = hdca.update_time
+
+        # Create a second session to avoid the dialect mock interfering with refresh
+        with Session(session.bind) as fallback_session:
+            # Get the dataset and hda in the new session
+            dataset = fallback_session.get(model.Dataset, hda.dataset.id)
+            assert dataset
+
+            # Mock the dialect on the fallback session to force Python fallback
+            with patch.object(fallback_session.bind, "dialect") as mock_dialect:
+                mock_dialect.name = "unknown_db"
+
+                # This should use the Python fallback path
+                dataset.touch_collection_update_time()
+                fallback_session.commit()
+
+        # Refresh the HDCA in the original session to see the updates
+        session.refresh(hdca)
+
+        # Verify update time was changed
+        assert hdca.update_time > original_time
+
+    def test_depth_limiting(self, session, hda):
+        """Test that very deep hierarchies are handled with depth limiting."""
+        # Create a deep hierarchy (this tests the depth limit logic)
+        current_collection = None
+        collections = []
+
+        # Create 5 levels of nested collections
+        for i in range(5):
+            collection = model.DatasetCollection(collection_type=f"list{':list' * i}")
+            collection_id = persist(session, collection)
+            collection = session.get(model.DatasetCollection, collection_id)
+            collections.append(collection)
+
+            if i == 0:
+                # First level: connect to HDA
+                dce = model.DatasetCollectionElement(
+                    collection=collection, element=hda, element_identifier=f"item_{i}", element_index=0
+                )
+            else:
+                # Subsequent levels: connect to previous collection
+                dce = model.DatasetCollectionElement(
+                    collection=collection, element=current_collection, element_identifier=f"item_{i}", element_index=0
+                )
+            persist(session, dce)
+            current_collection = collection
+
+        # Create HDCA for the top-level collection
+        hdca = model.HistoryDatasetCollectionAssociation(
+            collection=current_collection, history=hda.history, name="Deep Collection", hid=10
+        )
+        hdca_id = persist(session, hdca)
+        hdca = session.get(model.HistoryDatasetCollectionAssociation, hdca_id)
+
+        original_time = hdca.update_time
+
+        # Should handle deep hierarchy without issues
+        hda.dataset.touch_collection_update_time()
+        session.commit()
+
+        session.refresh(hdca)
+        assert hdca.update_time > original_time
+
+    def test_empty_collections(self, session, history):
+        """Test behavior with empty collections (no elements)."""
+        # Create empty collection
+        collection = model.DatasetCollection(collection_type="list")
+        collection_id = persist(session, collection)
+        collection = session.get(model.DatasetCollection, collection_id)
+
+        # Create HDCA for empty collection
+        hdca = model.HistoryDatasetCollectionAssociation(
+            collection=collection, history=history, name="Empty Collection", hid=1
+        )
+        persist(session, hdca)
+
+        # Create a dataset not in any collection and get it back from session
+        dataset = model.Dataset(state="ok")
+        dataset_id = persist(session, dataset)
+        dataset = session.get(model.Dataset, dataset_id)
+
+        # Should not affect empty collections
+        dataset.touch_collection_update_time()
+        session.commit()

test/unit/data/test_galaxy_mapping.py

@@ -188,7 +188,11 @@ class TestMappings(BaseModelTestCase):
         assert c2.dataset_elements == [dce1, dce2]
         assert c2.dataset_action_tuples == []
         assert c2.populated_optimized
-        assert c2.dataset_states_and_extensions_summary == ({"new"}, {"txt", "bam"})
+        summary = c2.dataset_states_and_extensions_summary
+        extensions = summary.extensions
+        states = summary.states
+        assert states == {"new": 2}
+        assert extensions == ["bam", "txt"]
         assert c2.element_identifiers_extensions_paths_and_metadata_files == [
             [
                 ("inner_list", "forward"),
@@ -200,7 +204,11 @@ class TestMappings(BaseModelTestCase):
         ]
         assert c3.dataset_instances == []
         assert c3.dataset_elements == []
-        assert c3.dataset_states_and_extensions_summary == (set(), set())
+        summary_c3 = c3.dataset_states_and_extensions_summary
+        dbkeys_c3 = summary_c3.dbkeys
+        extensions_c3 = summary_c3.extensions
+        assert not dbkeys_c3
+        assert not extensions_c3
 
         stmt = c4._build_nested_collection_attributes_stmt(element_attributes=("element_identifier",))
         result = self.model.session.execute(stmt).all()