diff --git a/client/src/api/index.ts b/client/src/api/index.ts
index 6f9ce9d0213d..839c4308da8d 100644
--- a/client/src/api/index.ts
+++ b/client/src/api/index.ts
@@ -312,4 +312,7 @@ 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 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/FormInputCollection.vue b/client/src/components/Workflow/Editor/Forms/FormInputCollection.vue
index 352f74996b0f..0f3d9fc7d7cd 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,6 +68,13 @@ 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 | null;
     if (formatStr) {
@@ -97,5 +108,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/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..0a518d8b6701
--- /dev/null
+++ b/lib/galaxy/model/dataset_collections/types/sample_sheet_util.py
@@ -0,0 +1,101 @@
+from typing import (
+    List,
+    Optional,
+    Union,
+)
+
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    RootModel,
+    TypeAdapter,
+)
+
+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")
+    type: SampleSheetColumnType
+    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
+    return SampleSheetColumnDefinitionModel(**column_definition)
+
+
+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/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..6e73e06ae588 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,25 @@ 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):
+    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 +1018,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 +1167,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 +1684,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..d303519f2da1 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,144 @@ 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"}],
+            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"}]
+        create_response = self._post("dataset_collections", payload, json=True)
+        assert_status_code_is(create_response, 400)
+        payload["column_definitions"] = [{"type": "int", "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"}],
+            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"}]
+        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", "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"}],
+            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/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..bf0bffc91274
--- /dev/null
+++ b/test/unit/data/dataset_collections/test_sample_sheet_util.py
@@ -0,0 +1,96 @@
+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"}])
+
+
+def test_sample_sheet_validation_int_type():
+    validate_row([1], [{"type": "int"}])
+
+    with pytest.raises(RequestParameterInvalidException):
+        validate_row(["sample1"], [{"type": "int"}])
+
+
+def test_sample_sheet_validation_float_type():
+    validate_row([1.0], [{"type": "float"}])
+
+    with pytest.raises(RequestParameterInvalidException):
+        validate_row(["sample1"], [{"type": "float"}])
+
+
+def test_sample_sheet_validation_string_type():
+    validate_row(["sample1"], [{"type": "string"}])
+
+    with pytest.raises(RequestParameterInvalidException):
+        validate_row([1], [{"type": "string"}])
+
+
+def test_sample_sheet_validation_boolean_type():
+    validate_row([True], [{"type": "boolean"}])
+
+    with pytest.raises(RequestParameterInvalidException):
+        validate_row([1], [{"type": "boolean"}])
+
+
+def test_sample_sheet_validation_restrictions():
+    validate_row(["control"], [{"type": "string", "restrictions": ["treatment", "control"]}])
+
+    with pytest.raises(RequestParameterInvalidException):
+        validate_row(["controlx"], [{"type": "string", "restrictions": ["treatment", "control"]}])
+
+
+def test_sample_sheet_validation_length():
+    column_definitions = [{"type": "string", "validators": [{"type": "length", "min": 6}]}]
+    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}]}]
+    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"]}]
+    validate_column_definitions(column_definitions)
+
+
+def test_column_definitions_validators_invalid_length():
+    column_definitions = [{"type": "string", "validators": [{"type": "length", "min": 6}]}]
+    validate_column_definitions(column_definitions)
+
+
+def test_column_definitions_do_not_allow_unsafe_validators():
+    column_definitions = [
+        {
+            "type": "string",
+            "validators": [
+                {"type": "expression", "expression": "False"},
+            ],
+        }
+    ]
+
+    with pytest.raises(ValidationError):
+        validate_column_definitions(column_definitions)