[release-v0.36.x] de-dupe order and resource dependencies

pkg/apis/pipeline/v1beta1/pipeline_types.go

@@ -469,51 +469,38 @@ func (pt PipelineTask) Validate(ctx context.Context) (errs *apis.FieldError) {
 
 // Deps returns all other PipelineTask dependencies of this PipelineTask, based on resource usage or ordering
 func (pt PipelineTask) Deps() []string {
-	deps := []string{}
+	// hold the list of dependencies in a set to avoid duplicates
+	deps := sets.NewString()
 
-	deps = append(deps, pt.resourceDeps()...)
-	deps = append(deps, pt.orderingDeps()...)
-
-	uniqueDeps := sets.NewString()
-	for _, w := range deps {
-		if uniqueDeps.Has(w) {
-			continue
-		}
-		uniqueDeps.Insert(w)
-	}
-
-	return uniqueDeps.List()
-}
-
-func (pt PipelineTask) resourceDeps() []string {
-	resourceDeps := []string{}
+	// add any new dependents from a resource/workspace
 	if pt.Resources != nil {
 		for _, rd := range pt.Resources.Inputs {
-			resourceDeps = append(resourceDeps, rd.From...)
+			for _, f := range rd.From {
+				deps.Insert(f)
+			}
 		}
 	}
 
 	// Add any dependents from conditional resources.
 	for _, cond := range pt.Conditions {
 		for _, rd := range cond.Resources {
-			resourceDeps = append(resourceDeps, rd.From...)
+			for _, f := range rd.From {
+				deps.Insert(f)
+			}
 		}
 	}
 
-	// Add any dependents from result references.
+	// add any new dependents from result references - resource dependency
 	for _, ref := range PipelineTaskResultRefs(&pt) {
-		resourceDeps = append(resourceDeps, ref.PipelineTask)
+		deps.Insert(ref.PipelineTask)
 	}
 
-	return resourceDeps
-}
-
-func (pt PipelineTask) orderingDeps() []string {
-	orderingDeps := []string{}
+	// add any new dependents from runAfter - order dependency
 	for _, runAfter := range pt.RunAfter {
-		orderingDeps = append(orderingDeps, runAfter)
+		deps.Insert(runAfter)
 	}
-	return orderingDeps
+
+	return deps.List()
 }
 
 // PipelineTaskList is a list of PipelineTasks

pkg/apis/pipeline/v1beta1/pipeline_types_test.go

@@ -590,12 +590,58 @@ func TestPipelineTaskList_Deps(t *testing.T) {
 				Operator: "in",
 				Values:   []string{"foo"},
 			}},
+			Matrix: []Param{{
+				Value: ArrayOrString{
+					Type: ParamTypeArray,
+					ArrayVal: []string{
+						"$(tasks.task-2.results.result)",
+						"$(tasks.task-5.results.result)",
+					},
+				}},
+			},
+		}, {
+			Name: "task-6",
+			Resources: &PipelineTaskResources{
+				Inputs: []PipelineTaskInputResource{{
+					From: []string{"task-1", "task-1"},
+				}},
+			},
+		}, {
+			Name: "task-7",
+			WhenExpressions: WhenExpressions{{
+				Input:    "$(tasks.task-3.results.result1)",
+				Operator: "in",
+				Values:   []string{"foo"},
+			}, {
+				Input:    "$(tasks.task-3.results.result2)",
+				Operator: "in",
+				Values:   []string{"foo"},
+			}},
+		}, {
+			Name: "task-8",
+			Params: []Param{{
+				Value: ArrayOrString{
+					Type:      "string",
+					StringVal: "$(tasks.task-4.results.result1)",
+				}}, {
+				Value: ArrayOrString{
+					Type:      "string",
+					StringVal: "$(tasks.task-4.results.result2)",
+				}},
+			},
+		}, {
+			Name:     "task-9",
+			RunAfter: []string{"task-1", "task-1", "task-1", "task-1"},
 		}},
 		expectedDeps: map[string][]string{
 			"task-2": {"task-1"},
 			"task-3": {"task-1", "task-2"},
 			"task-4": {"task-1", "task-2", "task-3"},
 			"task-5": {"task-1", "task-2", "task-3", "task-4"},
+			"task-6": {"task-1"},
+			"task-7": {"task-3"},
+			"task-8": {"task-4"},
+			"task-9": {"task-1"},
 		},
 	}}
 	for _, tc := range pipelines {

pkg/reconciler/pipelinerun/resources/pipelinerunstate_test.go

@@ -969,7 +969,7 @@ func TestDAGExecutionQueueSequentialRuns(t *testing.T) {
 }
 
 func TestPipelineRunState_CompletedOrSkippedDAGTasks(t *testing.T) {
-	largePipelineState := buildPipelineStateWithLargeDepencyGraph(t)
+	largePipelineState := buildPipelineStateWithLargeDependencyGraph(t)
 	tcs := []struct {
 		name          string
 		state         PipelineRunState
@@ -1036,6 +1036,14 @@ func TestPipelineRunState_CompletedOrSkippedDAGTasks(t *testing.T) {
 		name:          "large deps, not started",
 		state:         largePipelineState,
 		expectedNames: []string{},
+	}, {
+		name:          "large deps through params, not started",
+		state:         buildPipelineStateWithMultipleTaskResults(t, false),
+		expectedNames: []string{},
+	}, {
+		name:          "large deps through params and when expressions, not started",
+		state:         buildPipelineStateWithMultipleTaskResults(t, true),
+		expectedNames: []string{},
 	}, {
 		name:          "one-run-started",
 		state:         oneRunStartedState,
@@ -1069,7 +1077,7 @@ func TestPipelineRunState_CompletedOrSkippedDAGTasks(t *testing.T) {
 	}
 }
 
-func buildPipelineStateWithLargeDepencyGraph(t *testing.T) PipelineRunState {
+func buildPipelineStateWithLargeDependencyGraph(t *testing.T) PipelineRunState {
 	t.Helper()
 	var task = &v1beta1.Task{
 		ObjectMeta: metav1.ObjectMeta{
@@ -1128,6 +1136,74 @@ func buildPipelineStateWithLargeDepencyGraph(t *testing.T) PipelineRunState {
 	return pipelineRunState
 }
 
+func buildPipelineStateWithMultipleTaskResults(t *testing.T, includeWhen bool) PipelineRunState {
+	t.Helper()
+	var task = &v1beta1.Task{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "task",
+		},
+		Spec: v1beta1.TaskSpec{
+			Steps: []v1beta1.Step{{
+				Name: "step1",
+			}},
+		},
+	}
+	var pipelineRunState PipelineRunState
+	pipelineRunState = []*ResolvedPipelineRunTask{{
+		PipelineTask: &v1beta1.PipelineTask{
+			Name:    "t1",
+			TaskRef: &v1beta1.TaskRef{Name: "task"},
+		},
+		TaskRun: nil,
+		ResolvedTaskResources: &resources.ResolvedTaskResources{
+			TaskSpec: &task.Spec,
+		},
+	}}
+	for i := 2; i < 400; i++ {
+		var params []v1beta1.Param
+		whenExpressions := v1beta1.WhenExpressions{}
+		var alpha byte
+		// the task has a reference to multiple task results (a through j) from each parent task - causing a redundant references
+		// the task dependents on all predecessors in a graph through params and/or whenExpressions
+		for j := 1; j < i; j++ {
+			for alpha = 'a'; alpha <= 'j'; alpha++ {
+				// include param with task results
+				params = append(params, v1beta1.Param{
+					Name: fmt.Sprintf("%c", alpha),
+					Value: v1beta1.ArrayOrString{
+						Type:      v1beta1.ParamTypeString,
+						StringVal: fmt.Sprintf("$(tasks.t%d.results.%c)", j, alpha),
+					},
+				})
+			}
+			if includeWhen {
+				for alpha = 'a'; alpha <= 'j'; alpha++ {
+					// include when expressions with task results
+					whenExpressions = append(whenExpressions, v1beta1.WhenExpression{
+						Input:    fmt.Sprintf("$(tasks.t%d.results.%c)", j, alpha),
+						Operator: selection.In,
+						Values:   []string{"true"},
+					})
+				}
+			}
+		}
+		pipelineRunState = append(pipelineRunState, &ResolvedPipelineRunTask{
+			PipelineTask: &v1beta1.PipelineTask{
+				Name:            fmt.Sprintf("t%d", i),
+				Params:          params,
+				TaskRef:         &v1beta1.TaskRef{Name: "task"},
+				WhenExpressions: whenExpressions,
+			},
+			TaskRun: nil,
+			ResolvedTaskResources: &resources.ResolvedTaskResources{
+				TaskSpec: &task.Spec,
+			},
+		},
+		)
+	}
+	return pipelineRunState
+}
+
 func TestPipelineRunState_GetFinalTasks(t *testing.T) {
 	tcs := []struct {
 		name               string