diff --git a/client/src/api/index.ts b/client/src/api/index.ts
index 6f9ce9d0213d..27a4d572c797 100644
--- a/client/src/api/index.ts
+++ b/client/src/api/index.ts
@@ -312,4 +312,8 @@ export type ObjectExportTaskResponse = components["schemas"]["ObjectExportTaskRe
 export type ExportObjectRequestMetadata = components["schemas"]["ExportObjectRequestMetadata"];
 export type ExportObjectResultMetadata = components["schemas"]["ExportObjectResultMetadata"];
 
+export type SampleSheetColumnDefinition = components["schemas"]["SampleSheetColumnDefinition"];
+export type SampleSheetColumnDefinitionType = SampleSheetColumnDefinition["type"];
+export type SampleSheetColumnDefinitions = SampleSheetColumnDefinition[] | null;
+
 export type AsyncTaskResultSummary = components["schemas"]["AsyncTaskResultSummary"];
diff --git a/client/src/components/Workflow/Editor/Forms/FormCollectionType.vue b/client/src/components/Workflow/Editor/Forms/FormCollectionType.vue
index a93cb5ded6f1..77b5aa8161f2 100644
--- a/client/src/components/Workflow/Editor/Forms/FormCollectionType.vue
+++ b/client/src/components/Workflow/Editor/Forms/FormCollectionType.vue
@@ -23,6 +23,7 @@ const collectionTypeOptions = [
     { value: "list", label: "List of Datasets" },
     { value: "paired", label: "Dataset Pair" },
     { value: "list:paired", label: "List of Dataset Pairs" },
+    { value: "sample_sheet", label: "Sample Sheet of Datasets" },
 ];
 
 function updateValue(newValue: string | undefined) {
diff --git a/client/src/components/Workflow/Editor/Forms/FormColumnDefinition.vue b/client/src/components/Workflow/Editor/Forms/FormColumnDefinition.vue
new file mode 100644
index 000000000000..1e74cb43b04f
--- /dev/null
+++ b/client/src/components/Workflow/Editor/Forms/FormColumnDefinition.vue
@@ -0,0 +1,56 @@
+<script setup lang="ts">
+import { type SampleSheetColumnDefinition, type SampleSheetColumnDefinitionType } from "@/api";
+
+import FormColumnDefinitionType from "./FormColumnDefinitionType.vue";
+import FormElement from "@/components/Form/FormElement.vue";
+
+interface Props {
+    value: SampleSheetColumnDefinition;
+    index: number;
+    prefix: string; // prefix for ID objects
+}
+
+const props = defineProps<Props>();
+
+const emit = defineEmits(["onChange"]);
+
+function stateCopy(): SampleSheetColumnDefinition {
+    return JSON.parse(JSON.stringify(props.value));
+}
+
+function onName(name: string) {
+    const state = stateCopy();
+    state.name = name;
+    emit("onChange", state, props.index);
+}
+
+function onType(newType: SampleSheetColumnDefinitionType) {
+    const state = stateCopy();
+
+    state.type = newType;
+    emit("onChange", state, props.index);
+}
+</script>
+
+<template>
+    <div>
+        <FormElement
+            :id="prefix + '_name'"
+            :value="value.name"
+            title="Name"
+            type="text"
+            help="Provide a short, unique name to describe this column."
+            @input="onName" />
+        <FormColumnDefinitionType :value="value.type" :prefix="prefix" @onChange="onType" />
+        <FormElement
+            :id="prefix + '_description'"
+            :value="value.description"
+            title="Description"
+            type="text"
+            help="Provide a longer description to help people running this workflow under what is expected to be entered in this column."
+            @input="onName" />
+        TODO: There are more fields to enter here including restrictions and validations that vary based on the type
+        chosen. There will be a lot of overlap with the same validation options for workflow parameters so it might be
+        best to wait until those components can be developed in parallel.
+    </div>
+</template>
diff --git a/client/src/components/Workflow/Editor/Forms/FormColumnDefinitions.vue b/client/src/components/Workflow/Editor/Forms/FormColumnDefinitions.vue
new file mode 100644
index 000000000000..220c1be3df88
--- /dev/null
+++ b/client/src/components/Workflow/Editor/Forms/FormColumnDefinitions.vue
@@ -0,0 +1,148 @@
+<script setup lang="ts">
+import { library } from "@fortawesome/fontawesome-svg-core";
+import { faCaretDown, faCaretUp, faPlus, faTrashAlt } from "@fortawesome/free-solid-svg-icons";
+import { FontAwesomeIcon } from "@fortawesome/vue-fontawesome";
+import { BLink } from "bootstrap-vue";
+import { computed } from "vue";
+
+import { type SampleSheetColumnDefinition, type SampleSheetColumnDefinitions } from "@/api";
+import localize from "@/utils/localization";
+
+import FormColumnDefinition from "./FormColumnDefinition.vue";
+import FormCard from "@/components/Form/FormCard.vue";
+
+library.add(faPlus, faTrashAlt, faCaretUp, faCaretDown);
+
+interface Props {
+    value: SampleSheetColumnDefinitions;
+}
+
+const props = defineProps<Props>();
+
+function addColumn() {
+    const state = stateCopy();
+    state.push({ type: "int", name: "column" });
+    emit("onChange", state);
+}
+
+function stateCopy(): SampleSheetColumnDefinition[] {
+    return JSON.parse(JSON.stringify(props.value || []));
+}
+
+function onRemove(index: number) {
+    const state = stateCopy();
+    state.splice(index, 1);
+    emit("onChange", state);
+}
+
+function titleForColumnDefinition(index: number) {
+    return `Column ${index + 1}`;
+}
+
+function getPrefix(index: number) {
+    const name = `column_definition_${index}`;
+    return name;
+}
+
+function getButtonId(index: number, direction: "up" | "down") {
+    const prefix = getPrefix(index);
+    return `${prefix}_${direction}`;
+}
+
+function swap(index: number, swapWith: number, direction: "up" | "down") {
+    // the FormRepeat version does cool highlighting - probably worth implementing
+    // on next pass
+    const state = stateCopy();
+    if (swapWith >= 0 && swapWith < state.length && index >= 0 && index < state.length) {
+        const wasSwapped = state[swapWith] as SampleSheetColumnDefinition;
+        state[swapWith] = state[index] as SampleSheetColumnDefinition;
+        state[index] = wasSwapped;
+    }
+    emit("onChange", state);
+}
+
+function onChildUpdate(childState: SampleSheetColumnDefinition, index: number) {
+    const state = stateCopy();
+    state[index] = childState;
+    emit("onChange", state);
+}
+
+const deleteTooltip = computed(() => {
+    return localize(`Click to delete column definition`);
+});
+
+const emit = defineEmits(["onChange"]);
+</script>
+
+<template>
+    <div class="ui-form-element section-row">
+        <div class="ui-form-title">
+            <span class="ui-form-title-text">Column definitions</span>
+        </div>
+        <FormCard
+            v-for="(columnDefinition, index) in value"
+            v-bind:key="index"
+            data-description="column definition block"
+            class="card"
+            :title="titleForColumnDefinition(index)">
+            <template v-slot:operations>
+                <!-- code modelled after FormRepeat -->
+                <span class="float-right">
+                    <b-button-group>
+                        <b-button
+                            :id="getButtonId(index, 'up')"
+                            v-b-tooltip.hover.bottom
+                            title="move up"
+                            role="button"
+                            variant="link"
+                            size="sm"
+                            class="ml-0"
+                            @click="() => swap(index, index - 1, 'up')">
+                            <FontAwesomeIcon icon="caret-up" />
+                        </b-button>
+                        <b-button
+                            :id="getButtonId(index, 'down')"
+                            v-b-tooltip.hover.bottom
+                            title="move down"
+                            role="button"
+                            variant="link"
+                            size="sm"
+                            class="ml-0"
+                            @click="() => swap(index, index + 1, 'down')">
+                            <FontAwesomeIcon icon="caret-down" />
+                        </b-button>
+                    </b-button-group>
+
+                    <span v-b-tooltip.hover.bottom :title="deleteTooltip">
+                        <b-button
+                            title="delete"
+                            role="button"
+                            variant="link"
+                            size="sm"
+                            class="ml-0"
+                            @click="() => onRemove(index)">
+                            <FontAwesomeIcon icon="trash-alt" />
+                        </b-button>
+                    </span>
+                </span>
+            </template>
+            <template v-slot:body>
+                <FormColumnDefinition
+                    :index="index"
+                    :value="columnDefinition"
+                    :prefix="getPrefix(index)"
+                    @onChange="onChildUpdate" />
+            </template>
+        </FormCard>
+        <BLink @click="addColumn">Add column.</BLink>
+    </div>
+</template>
+
+<style lang="scss" scoped>
+@import "../../../Form/_form-elements.scss";
+
+.column-definition-list {
+    padding: 0px;
+    list-style-type: none;
+}
+</style>
diff --git a/client/src/components/Workflow/Editor/Forms/FormInputCollection.vue b/client/src/components/Workflow/Editor/Forms/FormInputCollection.vue
index 46899793dca4..ef5379b0e0d1 100644
--- a/client/src/components/Workflow/Editor/Forms/FormInputCollection.vue
+++ b/client/src/components/Workflow/Editor/Forms/FormInputCollection.vue
@@ -1,6 +1,7 @@
 <script setup lang="ts">
 import { computed, toRef } from "vue";
 
+import { type SampleSheetColumnDefinitions } from "@/api";
 import type { DatatypesMapperModel } from "@/components/Datatypes/model";
 import type { Step } from "@/stores/workflowStepStore";
 
@@ -8,6 +9,7 @@ import { useToolState } from "../composables/useToolState";
 
 import FormElement from "@/components/Form/FormElement.vue";
 import FormCollectionType from "@/components/Workflow/Editor/Forms/FormCollectionType.vue";
+import FormColumnDefinitions from "@/components/Workflow/Editor/Forms/FormColumnDefinitions.vue";
 import FormDatatype from "@/components/Workflow/Editor/Forms/FormDatatype.vue";
 
 interface ToolState {
@@ -15,6 +17,7 @@ interface ToolState {
     optional: boolean;
     format: string | null;
     tag: string | null;
+    column_definitions: SampleSheetColumnDefinitions;
 }
 
 const props = defineProps<{
@@ -34,6 +37,7 @@ function cleanToolState(): ToolState {
             optional: false,
             tag: null,
             format: null,
+            column_definitions: null,
         };
     }
 }
@@ -64,9 +68,16 @@ function onCollectionType(newCollectionType: string | null) {
     emit("onChange", state);
 }
 
+function onColumnDefinitions(newColumnDefinitions: SampleSheetColumnDefinitions) {
+    const state = cleanToolState();
+    console.log(newColumnDefinitions);
+    state.column_definitions = newColumnDefinitions;
+    emit("onChange", state);
+}
+
 const formatsAsList = computed(() => {
-    const formatStr = toolState.value?.format as (string | string[] | null);
-    if (formatStr && (typeof formatStr === "string")) {
+    const formatStr = toolState.value?.format as string | string[] | null;
+    if (formatStr && typeof formatStr === "string") {
         return formatStr.split(/\s*,\s*/);
     } else if (formatStr) {
         return formatStr;
@@ -100,5 +111,9 @@ emit("onChange", cleanToolState());
             type="text"
             help="Tags to automatically filter inputs"
             @input="onTags" />
+        <FormColumnDefinitions
+            v-if="toolState?.collection_type == 'sample_sheet'"
+            :value="toolState?.column_definitions"
+            @onChange="onColumnDefinitions" />
     </div>
 </template>
diff --git a/client/src/components/Workflow/Editor/modules/collectionTypeDescription.ts b/client/src/components/Workflow/Editor/modules/collectionTypeDescription.ts
index cc48c2a69117..8012d2f2aefa 100644
--- a/client/src/components/Workflow/Editor/modules/collectionTypeDescription.ts
+++ b/client/src/components/Workflow/Editor/modules/collectionTypeDescription.ts
@@ -129,7 +129,7 @@ export class CollectionTypeDescription implements CollectionTypeDescriptor {
     }
 }
 
-const collectionTypeRegex = /^(list|paired)(:(list|paired))*$/;
+const collectionTypeRegex = /^((list|paired)(:(list|paired))*|sample_sheet)$/;
 
 export function isValidCollectionTypeStr(collectionType: string | undefined) {
     if (collectionType) {
diff --git a/client/src/stores/workflowStepStore.ts b/client/src/stores/workflowStepStore.ts
index dcb0547c9d7b..f0ff7109352f 100644
--- a/client/src/stores/workflowStepStore.ts
+++ b/client/src/stores/workflowStepStore.ts
@@ -1,5 +1,6 @@
 import { computed, del, ref, set } from "vue";
 
+import { type SampleSheetColumnDefinitions } from "@/api";
 import { type CollectionTypeDescriptor } from "@/components/Workflow/Editor/modules/collectionTypeDescription";
 import { type Connection, getConnectionId, useConnectionStore } from "@/stores/workflowConnectionStore";
 import { assertDefined } from "@/utils/assertions";
@@ -71,6 +72,7 @@ export interface DataStepInput extends BaseStepInput {
 export interface DataCollectionStepInput extends BaseStepInput {
     input_type: "dataset_collection";
     collection_types: string[];
+    column_definitions: SampleSheetColumnDefinitions;
 }
 
 export interface ParameterStepInput extends Omit<BaseStepInput, "input_type"> {
diff --git a/lib/galaxy/managers/collections.py b/lib/galaxy/managers/collections.py
index e9042e3aac55..dfe6863d9b38 100644
--- a/lib/galaxy/managers/collections.py
+++ b/lib/galaxy/managers/collections.py
@@ -176,6 +176,8 @@ def create(
         completed_job=None,
         output_name=None,
         fields=None,
+        column_definitions=None,
+        rows=None,
     ):
         """
         PRECONDITION: security checks on ability to add to parent
@@ -201,6 +203,8 @@ def create(
                 copy_elements=copy_elements,
                 history=history,
                 fields=fields,
+                column_definitions=column_definitions,
+                rows=rows,
             )
 
         implicit_inputs = []
@@ -288,6 +292,8 @@ def create_dataset_collection(
         copy_elements=False,
         history=None,
         fields=None,
+        column_definitions=None,
+        rows=None,
     ):
         # Make sure at least one of these is None.
         assert element_identifiers is None or elements is None
@@ -324,9 +330,12 @@ def create_dataset_collection(
 
         if elements is not self.ELEMENTS_UNINITIALIZED:
             type_plugin = collection_type_description.rank_type_plugin()
-            dataset_collection = builder.build_collection(type_plugin, elements, fields=fields)
+            dataset_collection = builder.build_collection(
+                type_plugin, elements, fields=fields, column_definitions=column_definitions, rows=rows
+            )
         else:
             # TODO: Pass fields here - need test case first.
+            # TODO: same with column definitions I think.
             dataset_collection = model.DatasetCollection(populated=False)
         dataset_collection.collection_type = collection_type
         return dataset_collection
@@ -783,10 +792,16 @@ def __init_rule_data(self, elements, collection_type_description, parent_identif
             identifiers = parent_identifiers + [element.element_identifier]
             if not element.is_collection:
                 data.append([])
+                columns = None
+                collection_type_str = collection_type_description.collection_type
+                if collection_type_str == "sample_sheet":
+                    columns = element.columns
+                    assert isinstance(columns, list)
                 source = {
                     "identifiers": identifiers,
                     "dataset": element_object,
                     "tags": element_object.make_tag_string_list(),
+                    "columns": columns,
                 }
                 sources.append(source)
             else:
diff --git a/lib/galaxy/managers/collections_util.py b/lib/galaxy/managers/collections_util.py
index 7f129992c754..8fbef8ed3a20 100644
--- a/lib/galaxy/managers/collections_util.py
+++ b/lib/galaxy/managers/collections_util.py
@@ -9,6 +9,7 @@
     exceptions,
     model,
 )
+from galaxy.model.dataset_collections.types.sample_sheet_util import validate_column_definitions
 from galaxy.util import string_as_bool
 
 log = logging.getLogger(__name__)
@@ -33,6 +34,9 @@ def api_payload_to_create_params(payload):
         message = f"Missing required parameters {missing_parameters}"
         raise exceptions.ObjectAttributeMissingException(message)
 
+    column_definitions = payload.get("column_definitions", None)
+    validate_column_definitions(column_definitions)
+
     params = dict(
         collection_type=payload.get("collection_type"),
         element_identifiers=payload.get("element_identifiers"),
@@ -40,6 +44,8 @@ def api_payload_to_create_params(payload):
         hide_source_items=string_as_bool(payload.get("hide_source_items", False)),
         copy_elements=string_as_bool(payload.get("copy_elements", False)),
         fields=payload.get("fields", None),
+        column_definitions=column_definitions,
+        rows=payload.get("rows", None),
     )
     return params
 
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 7123cace4990..e3c1657e8426 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -181,6 +181,8 @@
     DatasetValidatedState,
     InvocationsStateCounts,
     JobState,
+    SampleSheetColumnDefinitions,
+    SampleSheetRow,
     ToolRequestState,
 )
 from galaxy.schema.workflow.comments import WorkflowCommentModel
@@ -260,6 +262,7 @@ class ConfigurationTemplateEnvironmentVariable(TypedDict):
 CONFIGURATION_TEMPLATE_CONFIGURATION_VARIABLES_TYPE = Dict[str, CONFIGURATION_TEMPLATE_CONFIGURATION_VALUE_TYPE]
 CONFIGURATION_TEMPLATE_CONFIGURATION_SECRET_NAMES_TYPE = List[str]
 CONFIGURATION_TEMPLATE_DEFINITION_TYPE = Dict[str, Any]
+DATA_COLLECTION_FIELDS = List[Dict[str, Any]]
 
 
 class TransformAction(TypedDict):
@@ -6521,6 +6524,10 @@ class DatasetCollection(Base, Dictifiable, UsesAnnotations, Serializable):
     element_count: Mapped[Optional[int]]
     create_time: Mapped[datetime] = mapped_column(default=now, nullable=True)
     update_time: Mapped[datetime] = mapped_column(default=now, onupdate=now, nullable=True)
+    # if collection_type is 'record' (heterogenous collection)
+    fields: Mapped[Optional[DATA_COLLECTION_FIELDS]] = mapped_column(JSONType)
+    # if collection_type is 'sample_sheet' (collection of rows that datasets with extra column metadata)
+    column_definitions: Mapped[Optional[SampleSheetColumnDefinitions]] = mapped_column(JSONType)
 
     elements: Mapped[List["DatasetCollectionElement"]] = relationship(
         primaryjoin=(lambda: DatasetCollection.id == DatasetCollectionElement.dataset_collection_id),
@@ -6540,14 +6547,15 @@ def __init__(
         populated=True,
         element_count=None,
         fields=None,
+        column_definitions=None,
     ):
         self.id = id
         self.collection_type = collection_type
         if not populated:
             self.populated_state = DatasetCollection.populated_states.NEW
         self.element_count = element_count
-        # TODO: persist fields...
         self.fields = fields
+        self.column_definitions = column_definitions
 
     def _build_nested_collection_attributes_stmt(
         self,
@@ -6956,6 +6964,7 @@ def _base_to_dict(self, view):
             name=self.name,
             collection_id=self.collection_id,
             collection_type=self.collection.collection_type,
+            column_definitions=self.collection.column_definitions,
             populated=self.populated,
             populated_state=self.collection.populated_state,
             populated_state_message=self.collection.populated_state_message,
@@ -7443,6 +7452,7 @@ class DatasetCollectionElement(Base, Dictifiable, Serializable):
     # Element index and identifier to define this parent-child relationship.
     element_index: Mapped[Optional[int]]
     element_identifier: Mapped[Optional[str]] = mapped_column(Unicode(255))
+    columns: Mapped[Optional[SampleSheetRow]] = mapped_column(JSONType)
 
     hda = relationship(
         "HistoryDatasetAssociation",
@@ -7463,7 +7473,7 @@ class DatasetCollectionElement(Base, Dictifiable, Serializable):
 
     # actionable dataset id needs to be available via API...
     dict_collection_visible_keys = ["id", "element_type", "element_index", "element_identifier"]
-    dict_element_visible_keys = ["id", "element_type", "element_index", "element_identifier"]
+    dict_element_visible_keys = ["id", "element_type", "element_index", "element_identifier", "columns"]
 
     UNINITIALIZED_ELEMENT = object()
 
@@ -7474,6 +7484,7 @@ def __init__(
         element=None,
         element_index=None,
         element_identifier=None,
+        columns: Optional[SampleSheetRow] = None,
     ):
         if isinstance(element, HistoryDatasetAssociation):
             self.hda = element
@@ -7489,6 +7500,7 @@ def __init__(
         self.collection = collection
         self.element_index = element_index
         self.element_identifier = element_identifier or str(element_index)
+        self.columns = columns
 
     def __strict_check_before_flush__(self):
         if self.collection.populated_optimized:
diff --git a/lib/galaxy/model/dataset_collections/builder.py b/lib/galaxy/model/dataset_collections/builder.py
index 73af774904fe..f7358e501bc7 100644
--- a/lib/galaxy/model/dataset_collections/builder.py
+++ b/lib/galaxy/model/dataset_collections/builder.py
@@ -4,19 +4,31 @@
 from .type_description import COLLECTION_TYPE_DESCRIPTION_FACTORY
 
 
-def build_collection(type, dataset_instances, collection=None, associated_identifiers=None, fields=None):
+def build_collection(
+    type,
+    dataset_instances,
+    collection=None,
+    associated_identifiers=None,
+    fields=None,
+    column_definitions=None,
+    rows=None,
+):
     """
     Build DatasetCollection with populated DatasetcollectionElement objects
     corresponding to the supplied dataset instances or throw exception if
     this is not a valid collection of the specified type.
     """
-    dataset_collection = collection or model.DatasetCollection(fields=fields)
+    dataset_collection = collection or model.DatasetCollection(fields=fields, column_definitions=column_definitions)
     associated_identifiers = associated_identifiers or set()
-    set_collection_elements(dataset_collection, type, dataset_instances, associated_identifiers, fields=fields)
+    set_collection_elements(
+        dataset_collection, type, dataset_instances, associated_identifiers, fields=fields, rows=rows
+    )
     return dataset_collection
 
 
-def set_collection_elements(dataset_collection, type, dataset_instances, associated_identifiers, fields=None):
+def set_collection_elements(
+    dataset_collection, type, dataset_instances, associated_identifiers, fields=None, rows=None
+):
     new_element_keys = OrderedSet(dataset_instances.keys()) - associated_identifiers
     new_dataset_instances = {k: dataset_instances[k] for k in new_element_keys}
     dataset_collection.element_count = dataset_collection.element_count or 0
@@ -24,7 +36,10 @@ def set_collection_elements(dataset_collection, type, dataset_instances, associa
     elements = []
     if fields == "auto":
         fields = guess_fields(dataset_instances)
-    for element in type.generate_elements(new_dataset_instances, fields=fields):
+    column_definitions = dataset_collection.column_definitions
+    for element in type.generate_elements(
+        new_dataset_instances, fields=fields, rows=rows, column_definitions=column_definitions
+    ):
         element.element_index = element_index
         add_object_to_object_session(element, dataset_collection)
         element.collection = dataset_collection
diff --git a/lib/galaxy/model/dataset_collections/registry.py b/lib/galaxy/model/dataset_collections/registry.py
index bd148edafd2d..3ba42faed8d8 100644
--- a/lib/galaxy/model/dataset_collections/registry.py
+++ b/lib/galaxy/model/dataset_collections/registry.py
@@ -3,12 +3,14 @@
     list,
     paired,
     record,
+    sample_sheet,
 )
 
 PLUGIN_CLASSES = [
     list.ListDatasetCollectionType,
     paired.PairedDatasetCollectionType,
     record.RecordDatasetCollectionType,
+    sample_sheet.SampleSheetDatasetCollectionType,
 ]
 
 
diff --git a/lib/galaxy/model/dataset_collections/types/sample_sheet.py b/lib/galaxy/model/dataset_collections/types/sample_sheet.py
new file mode 100644
index 000000000000..f47d79861cf5
--- /dev/null
+++ b/lib/galaxy/model/dataset_collections/types/sample_sheet.py
@@ -0,0 +1,30 @@
+from galaxy.exceptions import RequestParameterMissingException
+from galaxy.model import DatasetCollectionElement
+from . import BaseDatasetCollectionType
+from .sample_sheet_util import validate_row
+
+
+class SampleSheetDatasetCollectionType(BaseDatasetCollectionType):
+    """A flat list of named elements starting rows with column metadata."""
+
+    collection_type = "sample_sheet"
+
+    def generate_elements(self, dataset_instances, **kwds):
+        rows = kwds.get("rows", None)
+        column_definitions = kwds.get("column_definitions", None)
+        if rows is None:
+            raise RequestParameterMissingException(
+                "Missing or null parameter 'rows' required for 'sample_sheet' collection types."
+            )
+        if len(dataset_instances) != len(rows):
+            self._validation_failed("Supplied element do not match 'rows'.")
+
+        for identifier, element in dataset_instances.items():
+            columns = rows[identifier]
+            validate_row(columns, column_definitions)
+            association = DatasetCollectionElement(
+                element=element,
+                element_identifier=identifier,
+                columns=columns,
+            )
+            yield association
diff --git a/lib/galaxy/model/dataset_collections/types/sample_sheet_util.py b/lib/galaxy/model/dataset_collections/types/sample_sheet_util.py
new file mode 100644
index 000000000000..25a970ffc518
--- /dev/null
+++ b/lib/galaxy/model/dataset_collections/types/sample_sheet_util.py
@@ -0,0 +1,109 @@
+from typing import (
+    List,
+    Optional,
+    Union,
+)
+
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    RootModel,
+    TypeAdapter,
+    ValidationError,
+)
+
+from galaxy.exceptions import RequestParameterInvalidException
+from galaxy.schema.schema import (
+    SampleSheetColumnDefinition,
+    SampleSheetColumnDefinitions,
+    SampleSheetColumnType,
+    SampleSheetColumnValueT,
+    SampleSheetRow,
+)
+from galaxy.tool_util.parser.parameter_validators import (
+    AnySafeValidatorModel,
+    DiscriminatedAnySafeValidatorModel,
+    parse_dict_validators,
+    UnsafeValidatorConfiguredInUntrustedContext,
+)
+
+
+class SampleSheetColumnDefinitionModel(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    name: str
+    type: SampleSheetColumnType
+    description: Optional[str] = None
+    validators: Optional[List[AnySafeValidatorModel]] = None
+    restrictions: Optional[List[SampleSheetColumnValueT]] = None
+    suggestions: Optional[List[SampleSheetColumnValueT]] = None
+
+
+SampleSheetColumnDefinitionsModel = RootModel[List[SampleSheetColumnDefinitionModel]]
+SampleSheetColumnDefinitionDictOrModel = Union[SampleSheetColumnDefinition, SampleSheetColumnDefinitionModel]
+
+
+def sample_sheet_column_definition_to_model(
+    column_definition: SampleSheetColumnDefinitionDictOrModel,
+) -> SampleSheetColumnDefinitionModel:
+    if isinstance(column_definition, SampleSheetColumnDefinitionModel):
+        return column_definition
+    else:
+        return SampleSheetColumnDefinitionModel.model_validate(column_definition)
+
+
+def validate_column_definitions(column_definitions: Optional[SampleSheetColumnDefinitions]):
+    for column_definition in column_definitions or []:
+        _validate_column_definition(column_definition)
+
+
+def _validate_column_definition(column_definition: SampleSheetColumnDefinition):
+    # we should do most of this with pydantic but I just wanted to especially make sure
+    # we were only using safe validators
+    try:
+        return SampleSheetColumnDefinitionModel(**column_definition)
+    except ValidationError as e:
+        # reuse code to convert this until we have ported the API endpoint to expect this
+        # and then just pass through the ValidationError as-is
+        raise RequestParameterInvalidException(str(e))
+
+
+def validate_row(row: SampleSheetRow, column_definitions: Optional[SampleSheetColumnDefinitions]):
+    if column_definitions is None:
+        return
+    if len(row) != len(column_definitions):
+        raise RequestParameterInvalidException(
+            "Sample sheet row validation failed, incorrect number of columns specified."
+        )
+    for column_value, column_definition in zip(row, column_definitions):
+        validate_column_value(column_value, column_definition)
+
+
+def validate_column_value(
+    column_value: SampleSheetColumnValueT, column_definition: SampleSheetColumnDefinitionDictOrModel
+):
+    column_definition_model = sample_sheet_column_definition_to_model(column_definition)
+    column_type = column_definition_model.type
+    if column_type == "int":
+        if not isinstance(column_value, int):
+            raise RequestParameterInvalidException(f"{column_value} was not an integer as expected")
+    elif column_type == "float":
+        if not isinstance(column_value, (float, int)):
+            raise RequestParameterInvalidException(f"{column_value} was not a number as expected")
+    elif column_type == "string":
+        if not isinstance(column_value, (str,)):
+            raise RequestParameterInvalidException(f"{column_value} was not a string as expected")
+    elif column_type == "boolean":
+        if not isinstance(column_value, (bool,)):
+            raise RequestParameterInvalidException(f"{column_value} was not a boolean as expected")
+    restrictions = column_definition_model.restrictions
+    if restrictions is not None:
+        if column_value not in restrictions:
+            raise RequestParameterInvalidException(
+                f"{column_value} was not in specified list of valid values as expected"
+            )
+    validators = column_definition_model.validators or []
+    for validator in validators:
+        try:
+            validator.statically_validate(column_value)
+        except ValueError as e:
+            raise RequestParameterInvalidException(str(e))
diff --git a/lib/galaxy/model/dataset_collections/types/sample_sheet_workbook.py b/lib/galaxy/model/dataset_collections/types/sample_sheet_workbook.py
new file mode 100644
index 000000000000..553ba6885c8f
--- /dev/null
+++ b/lib/galaxy/model/dataset_collections/types/sample_sheet_workbook.py
@@ -0,0 +1,157 @@
+from typing import List
+
+from openpyxl import Workbook
+from openpyxl.styles import Font
+from openpyxl.worksheet.datavalidation import DataValidation
+from openpyxl.worksheet.worksheet import Worksheet
+from pydantic import BaseModel
+from typing_extensions import Literal
+
+from .sample_sheet_util import SampleSheetColumnDefinitionModel
+
+DEFAULT_TITLE = "Sample Sheet for Galaxy"
+
+
+class CreateWorkbook(BaseModel):
+    title: str = DEFAULT_TITLE
+    first_column_type: Literal["URI"] = "URI"
+    column_definitions: List[SampleSheetColumnDefinitionModel]
+
+
+INSTRUCTIONS = [
+    "Use this spreadsheet to describe your samples. For each sample (i.e. each file), ensure all the labeled columns are specified and correct.",
+    "If you're using Google Sheets, data validation will be applied automatically - just make sure no cell values have a red mark indicating they are invalid.",
+    "If you're using Microsft Excel, it is best to run data validation after you've completed filling out this sheet. This can be done by clicking on 'Data' > 'Data Validation' > 'Circle Invalid Data'.",
+    "Once data entry is complete, drop this file back into Galaxy to finish creating a sample sheet collection for your inputs.",
+]
+
+
+def generate_workbook(payload: CreateWorkbook):
+    # Create a workbook and select the active worksheet
+    workbook = Workbook()
+    worksheet = workbook.active
+    worksheet.title = payload.title
+
+    column_definitions = payload.column_definitions
+    num_initial_columns = 1
+    headers = [_first_column_header(payload)] + [cd.name for cd in column_definitions]
+    worksheet.append(headers)
+    _make_bold(worksheet, 1, 0)
+    for index in range(len(headers)):
+        _make_bold(worksheet, 1, index + 1)
+
+    for index, header_name in enumerate(headers):
+        if index == 0:
+            width = 80
+        else:
+            width = 20
+        _set_width(worksheet, index, width)
+
+    _add_first_column_validation(payload, worksheet)
+    worksheet.freeze_panes = f"B2"
+
+    for index, column_definition in enumerate(column_definitions):
+        validation: Optional[DataValidation] = None
+        if column_definition.type == "int":
+            validation = DataValidation(type="whole", allow_blank=True)
+            # TODO: operator="between", formula1="1", formula2="1000"
+        elif column_definition.type == "float":
+            validation = DataValidation(type="decimal", allow_blank=True)
+            # TODO: operator="between", formula1="1", formula2="1000"
+        elif column_definition.type == "boolean":
+            column_str = _index_to_excel_column(index + num_initial_columns)
+            validation = DataValidation(
+                type="custom",
+                formula1=f"OR({column_str}2=TRUE, {column_str}2=FALSE)",
+                showDropDown=False,
+                allow_blank=True,
+            )
+
+            dropdown_validation = DataValidation(
+                type="list", formula1=f'"TRUE,FALSE"', showDropDown=False, allow_blank=True
+            )
+            _add_validation(_index_to_excel_column(index + num_initial_columns), dropdown_validation, worksheet)
+        elif column_definition.restrictions:
+            list_as_formula = ",".join([str(r) for r in column_definition.restrictions])
+            validation = DataValidation(
+                type="list", formula1=f'"{list_as_formula}"', showDropDown=False, allow_blank=True
+            )
+            validation.prompt = "Please select from the list"
+
+        if validation:
+            validation.error = "Invalid input"
+            validation.errorTitle = "Error"
+            validation.promptTitle = column_definition.name
+            _add_validation(_index_to_excel_column(index + num_initial_columns), validation, worksheet)
+
+    help_label_index = num_initial_columns + len(column_definitions) + 1
+    _set_width(worksheet, help_label_index, 10)
+
+    help_start_row = 3
+
+    worksheet.cell(row=help_start_row, column=help_label_index + 1, value="Instructions")
+    bold_font = Font(bold=True)
+    worksheet[f"{_index_to_excel_column(help_label_index)}3"].font = bold_font
+
+    for instruction_index, instruction in enumerate(INSTRUCTIONS):
+        worksheet.cell(
+            row=help_start_row + instruction_index + 1, column=help_label_index + 2, value=f"> {instruction_index + 1}."
+        )
+        worksheet.cell(row=help_start_row + instruction_index + 1, column=help_label_index + 3, value=instruction)
+
+    column_help_start_row = help_start_row + len(INSTRUCTIONS) + 2
+
+    worksheet.cell(row=column_help_start_row, column=help_label_index + 1, value="Columns")
+    _make_bold(worksheet, column_help_start_row, help_label_index)
+    # bold_font = Font(bold=True)
+    # worksheet[f"{_index_to_excel_column(help_label_index)}{column_help_start_row}"].font = bold_font
+    _set_width(worksheet, help_label_index + 1, 15)
+    _set_width(worksheet, help_label_index + 2, 150)
+
+    for column_index, column in enumerate(column_definitions):
+        worksheet.cell(row=column_help_start_row + column_index + 1, column=help_label_index + 2, value=column.name)
+        worksheet.cell(
+            row=column_help_start_row + column_index + 1, column=help_label_index + 3, value=column.description
+        )
+
+    return workbook
+
+
+def _first_column_header(payload: CreateWorkbook):
+    assert payload.first_column_type == "URI"
+    return "URI"
+
+
+def _set_width(worksheet: Worksheet, index: int, width: int):
+    worksheet.column_dimensions[_index_to_excel_column(index)].width = width
+
+
+def _make_bold(worksheet: Worksheet, row: int, column: int):
+    bold_font = Font(bold=True)
+    worksheet[f"{_index_to_excel_column(column)}{row}"].font = bold_font
+
+
+def _add_first_column_validation(payload: CreateWorkbook, worksheet: Worksheet):
+    assert payload.first_column_type == "URI"
+    # Add data validation for "URI" column
+    # We cannot assume http/https since drs, gxfiles, etc... are all fine
+    uri_validation = DataValidation(type="custom", formula1='=ISNUMBER(FIND("://", E2))', allow_blank=True)
+    _add_validation("A", uri_validation, worksheet)
+
+
+def _add_validation(column: str, data_validation: DataValidation, worksheet: Worksheet):
+    worksheet.add_data_validation(data_validation)
+    data_validation.add(f"{column}2:{column}1048576")
+
+
+def _index_to_excel_column(index: int) -> str:
+    """Converts a numeric index (0-based) into an Excel column label."""
+    if index < 0:
+        raise ValueError("Index must be 0 or greater")
+
+    column_label = ""
+    while index >= 0:
+        column_label = chr(index % 26 + 65) + column_label
+        index = index // 26 - 1  # Move to the next "digit" in base-26, adjusting for zero-based indexing
+
+    return column_label
diff --git a/lib/galaxy/model/migrations/alembic/versions_gxy/ec25b23b08e2_implement_sample_sheets.py b/lib/galaxy/model/migrations/alembic/versions_gxy/ec25b23b08e2_implement_sample_sheets.py
index 558a485020cc..435bdce0ab6e 100644
--- a/lib/galaxy/model/migrations/alembic/versions_gxy/ec25b23b08e2_implement_sample_sheets.py
+++ b/lib/galaxy/model/migrations/alembic/versions_gxy/ec25b23b08e2_implement_sample_sheets.py
@@ -28,10 +28,12 @@
 def upgrade():
     with transaction():
         add_column(dataset_collection_table, Column("column_definitions", JSONType(), default=None))
+        add_column(dataset_collection_table, Column("fields", JSONType(), default=None))
         add_column(dataset_collection_element_table, Column("columns", JSONType(), default=None))
 
 
 def downgrade():
     with transaction():
         drop_column(dataset_collection_table, "column_definitions")
+        drop_column(dataset_collection_table, "fields")
         drop_column(dataset_collection_element_table, "columns")
diff --git a/lib/galaxy/schema/schema.py b/lib/galaxy/schema/schema.py
index fe8d280f7515..f7257d922ba0 100644
--- a/lib/galaxy/schema/schema.py
+++ b/lib/galaxy/schema/schema.py
@@ -33,6 +33,8 @@
 from typing_extensions import (
     Annotated,
     Literal,
+    NotRequired,
+    TypedDict,
 )
 
 from galaxy.schema import partial_model
@@ -346,6 +348,27 @@ class LimitedUserModel(Model):
 
 MaybeLimitedUserModel = Union[UserModel, LimitedUserModel]
 
+# named in compatiblity with CWL - trying to keep CWL fields in mind with
+# this implementation. https://www.commonwl.org/user_guide/topics/inputs.html#inputs
+SampleSheetColumnType = Literal[
+    "string", "int", "float", "boolean"
+]  # excluding "long" and "double" and composite types from CWL for now - we don't think at this level of abstraction in Galaxy generally
+
+
+class SampleSheetColumnDefinition(TypedDict):
+    name: str
+    description: NotRequired[Optional[str]] = None
+    type: SampleSheetColumnType
+    validators: NotRequired[Optional[List[Dict[str, Any]]]] = None
+    restrictions: NotRequired[Optional[List[str]]] = None
+    suggestions: NotRequired[Optional[List[str]]] = None
+
+
+SampleSheetColumnValueT = Union[str, int, float, bool]
+SampleSheetColumnDefinitions = List[SampleSheetColumnDefinition]
+SampleSheetRow = List[SampleSheetColumnValueT]
+SampleSheetRows = Dict[str, SampleSheetRow]
+
 
 class DiskUsageUserModel(Model):
     total_disk_usage: float = TotalDiskUsageField
@@ -997,6 +1020,11 @@ class DCESummary(Model, WithModelClass):
         title="Object",
         description="The element's specific data depending on the value of `element_type`.",
     )
+    columns: Optional[SampleSheetRow] = Field(
+        None,
+        title="Columns",
+        description="A row (or list of columns) of data associated with this element",
+    )
 
 
 DCObject.model_rebuild()
@@ -1141,6 +1169,10 @@ class HDCADetailed(HDCASummary):
         None,
         description="Encoded ID for the ICJ object describing the collection of jobs corresponding to this collection",
     )
+    column_definitions: Optional[SampleSheetColumnDefinitions] = Field(
+        None,
+        description="Column data associated with each element of this collection.",
+    )
 
 
 class HistoryContentItemBase(Model):
@@ -1654,6 +1686,16 @@ class CreateNewCollectionPayload(Model):
         title="Element Identifiers",
         description="List of elements that should be in the new collection.",
     )
+    column_definitions: Optional[SampleSheetColumnDefinitions] = Field(
+        default=None,
+        title="Column Definitions",
+        description="Specify definitions for row data if collection_type if sample_sheet",
+    )
+    rows: Optional[SampleSheetRows] = Field(
+        default=None,
+        title="Row data",
+        description="Specify rows of metadata data corresponding to an indentifier if collection_type is sample_sheet",
+    )
     name: Optional[str] = Field(
         default=None,
         title="Name",
diff --git a/lib/galaxy/tool_util/client/staging.py b/lib/galaxy/tool_util/client/staging.py
index 3d4d6ca73080..136b2498f6b8 100644
--- a/lib/galaxy/tool_util/client/staging.py
+++ b/lib/galaxy/tool_util/client/staging.py
@@ -233,7 +233,9 @@ def _attach_file(upload_payload: Dict[str, Any], uri: str, index: int = 0) -> No
             else:
                 raise ValueError(f"Unsupported type for upload_target: {type(upload_target)}")
 
-        def create_collection_func(element_identifiers: List[Dict[str, Any]], collection_type: str) -> Dict[str, Any]:
+        def create_collection_func(
+            element_identifiers: List[Dict[str, Any]], collection_type: str, rows: Optional[Dict[str, Any]] = None
+        ) -> Dict[str, Any]:
             payload = {
                 "name": "dataset collection",
                 "instance_type": "history",
@@ -241,6 +243,7 @@ def create_collection_func(element_identifiers: List[Dict[str, Any]], collection
                 "element_identifiers": element_identifiers,
                 "collection_type": collection_type,
                 "fields": None if collection_type != "record" else "auto",
+                "rows": rows,
             }
             return self._post("dataset_collections", payload)
 
diff --git a/lib/galaxy/tool_util/cwl/util.py b/lib/galaxy/tool_util/cwl/util.py
index 2f3edde16dba..c36fe364049a 100644
--- a/lib/galaxy/tool_util/cwl/util.py
+++ b/lib/galaxy/tool_util/cwl/util.py
@@ -25,6 +25,7 @@
 import yaml
 from typing_extensions import (
     Literal,
+    Protocol,
     TypedDict,
 )
 
@@ -130,11 +131,19 @@ def path_or_uri_to_uri(path_or_uri: str) -> str:
         return path_or_uri
 
 
+class CollectionCreateFunc(Protocol):
+
+    def __call__(
+        self, element_identifiers: List[Dict[str, Any]], collection_type: str, rows: Optional[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
+        """Create a collection from these identifiers."""
+
+
 def galactic_job_json(
     job: Dict[str, Any],
     test_data_directory: str,
     upload_func: Callable[["UploadTarget"], Dict[str, Any]],
-    collection_create_func: Callable[[List[Dict[str, Any]], str], Dict[str, Any]],
+    collection_create_func: CollectionCreateFunc,
     tool_or_workflow: Literal["tool", "workflow"] = "workflow",
 ) -> Tuple[Dict[str, Any], List[Dict[str, Any]]]:
     """Adapt a CWL job object to the Galaxy API.
@@ -340,8 +349,10 @@ def replacement_collection(value: Dict[str, Any]) -> Dict[str, str]:
         assert "collection_type" in value
         collection_type = value["collection_type"]
         elements = to_elements(value, collection_type)
-
-        collection = collection_create_func(elements, collection_type)
+        kwds = {}
+        if collection_type == "sample_sheet":
+            kwds["rows"] = value["rows"]
+        collection = collection_create_func(elements, collection_type, **kwds)
         dataset_collections.append(collection)
         hdca_id = collection["id"]
         return {"src": "hdca", "id": hdca_id}
diff --git a/lib/galaxy/tool_util/parser/parameter_validators.py b/lib/galaxy/tool_util/parser/parameter_validators.py
index 29af9494b0bd..f52d36f93545 100644
--- a/lib/galaxy/tool_util/parser/parameter_validators.py
+++ b/lib/galaxy/tool_util/parser/parameter_validators.py
@@ -467,8 +467,22 @@ def default_message(self) -> str:
     Field(discriminator="type"),
 ]
 
+AnySafeValidatorModel = Annotated[
+    Union[
+        RegexParameterValidatorModel,
+        InRangeParameterValidatorModel,
+        LengthParameterValidatorModel,
+    ],
+    Field(discriminator="type"),
+]
+
 
 DiscriminatedAnyValidatorModel = TypeAdapter(AnyValidatorModel)  # type:ignore[var-annotated]
+DiscriminatedAnySafeValidatorModel = TypeAdapter(AnySafeValidatorModel)  # type:ignore[var-annotated]
+
+
+class UnsafeValidatorConfiguredInUntrustedContext(AssertionError):
+    pass
 
 
 def parse_dict_validators(validator_dicts: List[Dict[str, Any]], trusted: bool) -> List[AnyValidatorModel]:
@@ -477,7 +491,8 @@ def parse_dict_validators(validator_dicts: List[Dict[str, Any]], trusted: bool)
         validator = DiscriminatedAnyValidatorModel.validate_python(validator_dict)
         if not trusted:
             # Don't risk instantiating unsafe validators for user-defined code
-            assert validator._safe
+            if not validator._safe:
+                raise UnsafeValidatorConfiguredInUntrustedContext()
         validator_models.append(validator)
     return validator_models
 
diff --git a/lib/galaxy/util/rules_dsl.py b/lib/galaxy/util/rules_dsl.py
index ebea78be82a1..cfd3ffdf0fd5 100644
--- a/lib/galaxy/util/rules_dsl.py
+++ b/lib/galaxy/util/rules_dsl.py
@@ -250,6 +250,29 @@ def new_row(row):
         return list(map(new_row, data)), sources
 
 
+class AddColumnFromSampleSheetByIndex(BaseRuleDefinition):
+    rule_type = "add_sample_sheet_column_by_index"
+
+    def validate_rule(self, rule):
+        _ensure_rule_contains_keys(
+            rule,
+            {
+                "value": int,
+            },
+        )
+
+    def apply(self, rule, data, sources):
+        sample_sheet_column_index = rule["value"]
+
+        new_rows = []
+        for index, row in enumerate(data):
+            source = sources[index]
+            columns = source["columns"]
+            new_rows.append(row + [columns[sample_sheet_column_index]])
+
+        return new_rows, sources
+
+
 class RemoveColumnsRuleDefinition(BaseRuleDefinition):
     rule_type = "remove_columns"
 
@@ -604,6 +627,7 @@ def display(self):
     AddColumnRownumRuleDefinition,
     AddColumnValueRuleDefinition,
     AddColumnSubstrRuleDefinition,
+    AddColumnFromSampleSheetByIndex,
     RemoveColumnsRuleDefinition,
     AddFilterRegexRuleDefinition,
     AddFilterCountRuleDefinition,
diff --git a/lib/galaxy/util/rules_dsl_spec.yml b/lib/galaxy/util/rules_dsl_spec.yml
index 1514d05c8020..6b708c5280b0 100644
--- a/lib/galaxy/util/rules_dsl_spec.yml
+++ b/lib/galaxy/util/rules_dsl_spec.yml
@@ -451,6 +451,28 @@
   final:
     data: [["moo", "barn"], ["meow", "house"], ["bark", "firestation"]]
 
+- doc: add column from a sample sheet by index
+  rules:
+    - type: add_sample_sheet_column_by_index
+      value: 0
+  initial:
+    data: [["moo"], ["cow"]]
+    sources: [{"columns": [0, 1]}, {"columns": [2, 3]}]
+  final:
+    data: [["moo", 0], ["cow", 2]]
+
+- doc: add multiple columns from a sample sheet by index
+  rules:
+    - type: add_sample_sheet_column_by_index
+      value: 0
+    - type: add_sample_sheet_column_by_index
+      value: 1
+  initial:
+    data: [["moo"], ["cow"]]
+    sources: [{"columns": [0, 1]}, {"columns": [2, 3]}]
+  final:
+    data: [["moo", 0, 1], ["cow", 2, 3]]
+
 - rules:
     - type: invalid_rule_type
   error: true
diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py
index f3ea538b37c4..ccabb8b74ad5 100644
--- a/lib/galaxy/workflow/modules.py
+++ b/lib/galaxy/workflow/modules.py
@@ -1161,7 +1161,12 @@ def _parse_state_into_dict(self):
             collection_type = inputs["collection_type"]
         else:
             collection_type = self.default_collection_type
+        if "column_definitions" in inputs:
+            column_definitions = inputs["column_definitions"]
+        else:
+            column_definitions = None
         state_as_dict["collection_type"] = collection_type
+        state_as_dict["column_definitions"] = column_definitions
         return state_as_dict
 
 
diff --git a/lib/galaxy_test/api/test_dataset_collections.py b/lib/galaxy_test/api/test_dataset_collections.py
index a4ba01877e5e..c4c63e3ef924 100644
--- a/lib/galaxy_test/api/test_dataset_collections.py
+++ b/lib/galaxy_test/api/test_dataset_collections.py
@@ -5,7 +5,10 @@
 from urllib.parse import quote
 
 from galaxy.util.unittest_utils import skip_if_github_down
-from galaxy_test.base.api_asserts import assert_object_id_error
+from galaxy_test.base.api_asserts import (
+    assert_object_id_error,
+    assert_status_code_is,
+)
 from galaxy_test.base.decorators import requires_new_user
 from galaxy_test.base.populators import (
     DatasetCollectionPopulator,
@@ -201,6 +204,148 @@ def test_record_field_validation(self, history_id):
             create_response = self._post("dataset_collections", payload)
             self._assert_status_code_is(create_response, 400)
 
+    def test_sample_sheet_column_definition_problems(self, history_id):
+        contents = [
+            ("sample1", "1\t2\t3"),
+            ("sample2", "4\t5\t6"),
+        ]
+        sample_sheet_identifiers = self.dataset_collection_populator.list_identifiers(history_id, contents)
+        payload = dict(
+            name="my cool sample sheet",
+            instance_type="history",
+            history_id=history_id,
+            element_identifiers=sample_sheet_identifiers,
+            collection_type="sample_sheet",
+            column_definitions=[{"type": "int", "name": "replicate"}],
+            rows={"sample1": [42], "sample2": [45]},
+        )
+        create_response = self._post("dataset_collections", payload, json=True)
+        self._check_create_response(create_response)
+        payload["column_definitions"] = [{"type": "intx"}]
+        create_response = self._post("dataset_collections", payload, json=True)
+        assert_status_code_is(create_response, 400)
+        payload["column_definitions"] = [{"typex": "int"}]
+        create_response = self._post("dataset_collections", payload, json=True)
+        assert_status_code_is(create_response, 400)
+        payload["column_definitions"] = [{"type": "int", "restrictions": "wrongtype", "name": "replicate"}]
+        create_response = self._post("dataset_collections", payload, json=True)
+        assert_status_code_is(create_response, 400)
+        payload["column_definitions"] = [
+            {"type": "int", "name": "replicate", "validators": [{"type": "expression", "expression": "False"}]}
+        ]
+        create_response = self._post("dataset_collections", payload, json=True)
+        assert_status_code_is(create_response, 400)
+
+    def test_sample_sheet_validating_against_column_definition(self, history_id):
+        contents = [
+            ("sample1", "1\t2\t3"),
+            ("sample2", "4\t5\t6"),
+        ]
+        sample_sheet_identifiers = self.dataset_collection_populator.list_identifiers(history_id, contents)
+        payload = dict(
+            name="my cool sample sheet",
+            instance_type="history",
+            history_id=history_id,
+            element_identifiers=sample_sheet_identifiers,
+            collection_type="sample_sheet",
+            column_definitions=[{"type": "int", "name": "replicate"}],
+            rows={"sample1": [42], "sample2": [45]},
+        )
+        create_response = self._post("dataset_collections", payload, json=True)
+        self._check_create_response(create_response)
+        # now the datatype of the row data is wrong....
+        payload["column_definitions"] = [{"type": "string", "name": "replicate"}]
+        create_response = self._post("dataset_collections", payload, json=True)
+        assert_status_code_is(create_response, 400)
+
+        # now the row values are too small for the supplied validator
+        payload["column_definitions"] = [
+            {"type": "int", "name": "replicate", "validators": [{"type": "in_range", "min": 60}]}
+        ]
+        create_response = self._post("dataset_collections", payload, json=True)
+        assert_status_code_is(create_response, 400)
+
+    def test_sample_sheet_requires_columns(self, history_id):
+        contents = [
+            ("sample1", "1\t2\t3"),
+            ("sample2", "4\t5\t6"),
+        ]
+        sample_sheet_identifiers = self.dataset_collection_populator.list_identifiers(history_id, contents)
+        payload = dict(
+            name="my cool sample sheet",
+            instance_type="history",
+            history_id=history_id,
+            element_identifiers=sample_sheet_identifiers,
+            collection_type="sample_sheet",
+            column_definitions=[{"type": "int", "name": "replicate"}],
+            rows={"sample1": [42], "sample2": [45]},
+        )
+        create_response = self._post("dataset_collections", payload, json=True)
+        dataset_collection = self._check_create_response(create_response)
+
+        self._assert_has_keys(dataset_collection, "collection_type", "column_definitions")
+        column_definitions = dataset_collection["column_definitions"]
+        assert len(column_definitions) == 1
+        self._assert_has_keys(column_definitions[0], "type")
+        assert column_definitions[0]["type"] == "int"
+
+        # TODO: restore assertion and test before merging...
+        # assert something about column definition here....
+        assert dataset_collection["collection_type"] == "sample_sheet"
+        assert dataset_collection["name"] == "my cool sample sheet"
+        returned_collections = dataset_collection["elements"]
+        assert len(returned_collections) == 2, dataset_collection
+        sheet_row_0_element = returned_collections[0]
+        self._assert_has_keys(sheet_row_0_element, "element_index", "columns")
+        record_pos_0_object = sheet_row_0_element["object"]
+        self._assert_has_keys(record_pos_0_object, "name", "history_content_type")
+        row_0 = sheet_row_0_element["columns"]
+        assert row_0[0] == 42
+
+        sheet_row_1_element = returned_collections[1]
+        self._assert_has_keys(sheet_row_1_element, "element_index", "columns")
+        row_1 = sheet_row_1_element["columns"]
+        assert row_1[0] == 45
+        # TODO: test case where column definition does not match supplied data
+        # TODO: test case without column definition, implement definition inference based on supplied datatypes
+
+    def test_sample_sheet_column_definition_inference(self, history_id):
+        contents = [
+            ("sample1", "1\t2\t3"),
+            ("sample2", "4\t5\t6"),
+        ]
+        sample_sheet_identifiers = self.dataset_collection_populator.list_identifiers(history_id, contents)
+        payload = dict(
+            name="my cool sample sheet",
+            instance_type="history",
+            history_id=history_id,
+            element_identifiers=sample_sheet_identifiers,
+            collection_type="sample_sheet",
+            rows={"sample1": [42], "sample2": [45]},
+        )
+        create_response = self._post("dataset_collections", payload, json=True)
+        dataset_collection = self._check_create_response(create_response)
+
+        self._assert_has_keys(dataset_collection, "collection_type", "column_definitions")
+        column_definitions = dataset_collection["column_definitions"]
+        # TODO: restore assertion and test before merging...
+        # assert column_definitions, column_definitions
+        assert dataset_collection["collection_type"] == "sample_sheet"
+        assert dataset_collection["name"] == "my cool sample sheet"
+        returned_collections = dataset_collection["elements"]
+        assert len(returned_collections) == 2, dataset_collection
+        sheet_row_0_element = returned_collections[0]
+        self._assert_has_keys(sheet_row_0_element, "element_index", "columns")
+        record_pos_0_object = sheet_row_0_element["object"]
+        self._assert_has_keys(record_pos_0_object, "name", "history_content_type")
+        row_0 = sheet_row_0_element["columns"]
+        assert row_0[0] == 42
+
+        sheet_row_1_element = returned_collections[1]
+        self._assert_has_keys(sheet_row_1_element, "element_index", "columns")
+        row_1 = sheet_row_1_element["columns"]
+        assert row_1[0] == 45
+
     def test_list_download(self):
         with self.dataset_populator.test_history(require_new=False) as history_id:
             fetch_response = self.dataset_collection_populator.create_list_in_history(
diff --git a/lib/galaxy_test/api/test_tools.py b/lib/galaxy_test/api/test_tools.py
index 41d982d246c9..fcf29f928a99 100644
--- a/lib/galaxy_test/api/test_tools.py
+++ b/lib/galaxy_test/api/test_tools.py
@@ -848,6 +848,9 @@ def test_apply_rules_5(self):
     def test_apply_rules_6(self):
         self._apply_rules_and_check(rules_test_data.EXAMPLE_6)
 
+    def test_apply_rules_nested_list_from_sample_sheet(self):
+        self._apply_rules_and_check(rules_test_data.EXAMPLE_SAMPLE_SHEET_SIMPLE_TO_NESTED_LIST)
+
     @skip_without_tool("galaxy_json_sleep")
     def test_dataset_hidden_after_job_finish(self):
         with self.dataset_populator.test_history() as history_id:
diff --git a/lib/galaxy_test/base/rules_test_data.py b/lib/galaxy_test/base/rules_test_data.py
index 1530c0a05984..6009125b58d2 100644
--- a/lib/galaxy_test/base/rules_test_data.py
+++ b/lib/galaxy_test/base/rules_test_data.py
@@ -287,3 +287,54 @@ def check_example_6(hdca, dataset_populator):
     "check": check_example_6,
     "output_hid": 8,
 }
+
+
+def check_example_sample_sheet_simple_to_nested_list(hdca, dataset_populator):
+    assert hdca["collection_type"] == "list:list"
+    assert hdca["element_count"] == 2
+    treat1_el = hdca["elements"][0]
+    assert "object" in treat1_el, hdca
+    assert "element_identifier" in treat1_el
+    assert treat1_el["element_identifier"] == "treat1", hdca
+
+    treat2_el = hdca["elements"][1]
+    assert "object" in treat2_el, hdca
+    assert "element_identifier" in treat2_el
+    assert treat2_el["element_identifier"] == "treat2", hdca
+
+
+EXAMPLE_SAMPLE_SHEET_SIMPLE_TO_NESTED_LIST = {
+    "rules": {
+        "rules": [
+            {
+                "type": "add_sample_sheet_column_by_index",
+                "value": 0,
+            },
+            {
+                "type": "add_column_metadata",
+                "value": "identifier0",
+            },
+        ],
+        "mapping": [
+            {
+                "type": "list_identifiers",
+                "columns": [0, 1],
+            },
+        ],
+    },
+    "test_data": {
+        "type": "sample_sheet",
+        "elements": [
+            {"identifier": "i1", "contents": "0", "class": "File"},
+            {"identifier": "i2", "contents": "1", "class": "File"},
+            {"identifier": "i3", "contents": "2", "class": "File"},
+        ],
+        "rows": {
+            "i1": ["treat1"],
+            "i2": ["treat2"],
+            "i3": ["treat1"],
+        },
+    },
+    "check": check_example_sample_sheet_simple_to_nested_list,
+    "output_hid": 8,
+}
diff --git a/packages/data/setup.cfg b/packages/data/setup.cfg
index 2dd4e7e16b07..cb3d0dd99e8e 100644
--- a/packages/data/setup.cfg
+++ b/packages/data/setup.cfg
@@ -51,6 +51,7 @@ install_requires =
     msal
     mrcfile
     numpy
+    openpyxl>=3.1.5
     parsley
     pycryptodome
     pydantic[email]>=2.7.4
diff --git a/pyproject.toml b/pyproject.toml
index 08ba68afcd48..5d98b8e81d29 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -66,6 +66,7 @@ dependencies = [
     "numpy",
     "numpy>=1.26.0 ; python_version=='3.9'",  # Python 3.12 support
     "numpy>=2.1.0 ; python_version>='3.10'",  # Python 3.13 support
+    "openpyxl>=3.1.5",  # Minimum version ever tested with
     "packaging",
     "paramiko!=2.9.0,!=2.9.1",  # https://github.com/paramiko/paramiko/issues/1961
     "Parsley",
diff --git a/test/unit/data/dataset_collections/test_sample_sheet_util.py b/test/unit/data/dataset_collections/test_sample_sheet_util.py
new file mode 100644
index 000000000000..28b37cae9b92
--- /dev/null
+++ b/test/unit/data/dataset_collections/test_sample_sheet_util.py
@@ -0,0 +1,97 @@
+import pytest
+from pydantic import ValidationError
+
+from galaxy.exceptions import RequestParameterInvalidException
+from galaxy.model.dataset_collections.types.sample_sheet_util import (
+    validate_column_definitions,
+    validate_row,
+)
+from galaxy.schema.schema import (
+    SampleSheetColumnDefinitions,
+    SampleSheetRow,
+)
+
+
+def test_sample_sheet_validation_skipped_on_empty_definitions():
+    validate_row([0, 1], None)  # just ensure no execption is thrown
+
+
+def test_sample_sheet_validation_number_columns():
+    with pytest.raises(RequestParameterInvalidException):
+        validate_row([0, 1], [{"type": "int", "name": "replicate number"}])
+
+
+def test_sample_sheet_validation_int_type():
+    validate_row([1], [{"type": "int", "name": "replicate number"}])
+
+    with pytest.raises(RequestParameterInvalidException):
+        validate_row(["sample1"], [{"type": "int", "name": "replicate number"}])
+
+
+def test_sample_sheet_validation_float_type():
+    validate_row([1.0], [{"type": "float", "name": "seconds"}])
+
+    with pytest.raises(RequestParameterInvalidException):
+        validate_row(["sample1"], [{"type": "float", "name": "seconds"}])
+
+
+def test_sample_sheet_validation_string_type():
+    validate_row(["sample1"], [{"type": "string", "name": "condition"}])
+
+    with pytest.raises(RequestParameterInvalidException):
+        validate_row([1], [{"type": "string", "name": "condition"}])
+
+
+def test_sample_sheet_validation_boolean_type():
+    validate_row([True], [{"type": "boolean", "name": "control?"}])
+
+    with pytest.raises(RequestParameterInvalidException):
+        validate_row([1], [{"type": "boolean", "name": "control?"}])
+
+
+def test_sample_sheet_validation_restrictions():
+    validate_row(["control"], [{"type": "string", "restrictions": ["treatment", "control"], "name": "condition"}])
+
+    with pytest.raises(RequestParameterInvalidException):
+        validate_row(["controlx"], [{"type": "string", "restrictions": ["treatment", "control"], "name": "condition"}])
+
+
+def test_sample_sheet_validation_length():
+    column_definitions = [{"type": "string", "validators": [{"type": "length", "min": 6}], "name": "condition"}]
+    validate_row(["treatment"], column_definitions)
+
+    with pytest.raises(RequestParameterInvalidException):
+        validate_row(["treat"], column_definitions)
+
+
+def test_sample_sheet_validation_min_max():
+    column_definitions = [{"type": "int", "validators": [{"type": "in_range", "min": 6}], "name": "replicate number"}]
+    validate_row([7], column_definitions)
+
+    with pytest.raises(RequestParameterInvalidException):
+        validate_row([5], column_definitions)
+
+
+def test_column_definitions_validators_on_valid_defs():
+    column_definitions = [{"type": "string", "restrictions": ["treatment", "control"], "name": "condition"}]
+    validate_column_definitions(column_definitions)
+
+
+def test_column_definitions_validators_invalid_length():
+    column_definitions = [{"type": "string", "validators": [{"type": "length", "min": 6}], "name": "condition"}]
+    validate_column_definitions(column_definitions)
+
+
+def test_column_definitions_do_not_allow_unsafe_validators():
+    column_definitions = [
+        {
+            "type": "string",
+            "name": "condition",
+            "validators": [
+                {"type": "expression", "expression": "False"},
+            ],
+        }
+    ]
+
+    with pytest.raises(ValidationError):
+        validate_column_definitions(column_definitions)
diff --git a/test/unit/data/dataset_collections/test_sample_sheet_workbook.py b/test/unit/data/dataset_collections/test_sample_sheet_workbook.py
new file mode 100644
index 000000000000..4f13f3e22b23
--- /dev/null
+++ b/test/unit/data/dataset_collections/test_sample_sheet_workbook.py
@@ -0,0 +1,58 @@
+import os
+
+from galaxy.model.dataset_collections.types.sample_sheet_workbook import (
+    _index_to_excel_column,
+    CreateWorkbook,
+    generate_workbook,
+)
+
+WRITE_TEST_WORKBOOKS = False
+TEST_DATA = [
+    ["https://zenodo.org/records/3263975/files/DRR000770.fastqsanger.gz", 1, "treatment1", False],
+    ["https://zenodo.org/records/3263975/files/DRR000771.fastqsanger.gz", 2, "treatment1", False],
+    ["https://zenodo.org/records/3263975/files/DRR000772.fastqsanger.gz", 1, "none", True],
+    ["https://zenodo.org/records/3263975/files/DRR000773.fastqsanger.gz", 1, "treatment2", False],
+    ["https://zenodo.org/records/3263975/files/DRR000774.fastqsanger.gz", 2, "treatment3", False],
+    ["https://zenodo.org/records/3263975/files/DRR000775.fastqsanger.gz", "badnumber", "treatment2", False],
+    ["https://zenodo.org/records/3263975/files/DRR000776.fastqsanger.gz", 2, "wrongtreament", False],
+    ["https://zenodo.org/records/3263975/files/DRR000777.fastqsanger.gz", 3, "treatment2", "badbool"],
+]
+
+
+def test_index():
+    create = CreateWorkbook.model_validate(
+        {
+            "column_definitions": [
+                {"name": "replicate number", "type": "int", "description": "The replicate number of this sample."},
+                {
+                    "name": "treatment",
+                    "type": "string",
+                    "restrictions": ["treatment1", "treatment2", "none"],
+                    "description": "The treatment code for this sample.",
+                },
+                {
+                    "name": "is control?",
+                    "type": "boolean",
+                    "description": "Was this sample a control? If TRUE, please ensure treatment is set to none.",
+                },
+            ],
+        }
+    )
+    workbook = generate_workbook(create)
+    if WRITE_TEST_WORKBOOKS:
+        for index, row in enumerate(TEST_DATA):
+            for col, column in enumerate(row):
+                workbook.active.cell(row=index + 2, column=col + 1, value=column)
+
+        path = "~/test_workbook.xlsx"
+        expanded_path = os.path.expanduser(path)
+        workbook.save(expanded_path)
+
+
+def test_index_to_excel_column():
+    assert _index_to_excel_column(0) == "A"
+    assert _index_to_excel_column(25) == "Z"
+    assert _index_to_excel_column(26) == "AA"
+    assert _index_to_excel_column(700) == "ZY"
+    assert _index_to_excel_column(701) == "ZZ"
+    assert _index_to_excel_column(702) == "AAA"