feat(backend): support for optional input parameters in nested pipelines

[alyssacgoins]

Description of your changes: Resolves optional input parameters for components of a nested pipeline. fixes #11957

Checklist:

• [X] You have signed off your commits
• [X] The title for your pull request (PR) should follow our title convention. Learn more about the pull request title convention used in this repository.

[mprahl]

@alyssacgoins I need to review your PR more indepth when I have a bit more time but I was originally thinking the solution may be something simple like this. To help me review your PR, could you please explain why this is naive? :sweat_smile:

diff --git a/backend/src/v2/driver/resolve.go b/backend/src/v2/driver/resolve.go
index 98469d26b..6866ec201 100644
--- a/backend/src/v2/driver/resolve.go
+++ b/backend/src/v2/driver/resolve.go
@@ -270,6 +270,9 @@ func resolveInputs(
 		return inputs, nil
 	}
 
+	// TODO: Verify if this is always accurate.
+	isDagDriver := opts.Container == nil && opts.Task != nil
+
 	// Handle parameters.
 	for name, paramSpec := range task.GetInputs().GetParameters() {
 		v, err := resolveInputParameter(ctx, dag, pipeline, opts, mlmd, paramSpec, inputParams)
@@ -280,6 +283,17 @@ func resolveInputs(
 
 			componentParam, ok := opts.Component.GetInputDefinitions().GetParameters()[name]
 			if ok && componentParam != nil && componentParam.IsOptional {
+				if isDagDriver {
+					if componentParam.GetDefaultValue() != nil {
+						// If this is a nested pipeline, pass along the default value to the component.
+						inputs.ParameterValues[name] = componentParam.GetDefaultValue()
+					} else {
+						// If this is a nested pipeline, pass along the null value to the component.
+						inputs.ParameterValues[name] = structpb.NewNullValue()
+					}
+
+					continue
+				}
 				// If the resolved paramter was null and the component input parameter is optional, just skip setting
 				// it and the launcher will handle defaults.
 				continue

[alyssacgoins]

	// If this is a nested pipeline, pass along the default value to the component.
+						inputs.ParameterValues[name] = componentParam.GetDefaultValue()
+					} else {
+						// If this is a nested pipeline, pass along the null value to the component.
+						inputs.ParameterValues[name] = structpb.NewNullValue()
+					}

In my understanding, the block of code added in this diff is called within the if err != nil and statement, and only if the error is ErrResolvedParameterNull. However, when an input parameter is not located within a task, the error thrown is "parent DAG does not have input parameter". I think the diff you've included is definitely part of the overall solution. The block of code added to backend/src/v2/compiler/argocompilerargo.task() is necessary because it adds the necessary nested dag parameters to component task inputs

[alyssacgoins]

/retest

[alyssacgoins]

/retest

[alyssacgoins]

/retest

[mprahl]

/approve

[google-oss-prow[bot]]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: mprahl

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files:

• ~~backend/OWNERS~~ [mprahl]
• ~~samples/OWNERS~~ [mprahl]

Approvers can indicate their approval by writing /approve in a comment Approvers can cancel approval by writing /approve cancel in a comment