More flexible SubPipeline API

op/applied/type.go

@@ -147,10 +147,10 @@ func (p Pipeline) Clone() Pipeline {
 	return Pipeline{operations: clone}
 }
 
-// SubPipeline returns a sub-pipeline starting from step `idx`
-// of the current pipeline.
-func (p Pipeline) SubPipeline(idx int) Pipeline {
-	return Pipeline{operations: p.operations[idx:]}
+// SubPipeline returns a sub-pipeline containing operations between step `startInclusive`
+// and step `endExclusive` of the current pipeline.
+func (p Pipeline) SubPipeline(startInclusive int, endExclusive int) Pipeline {
+	return Pipeline{operations: p.operations[startInclusive:endExclusive]}
 }
 
 func (p Pipeline) String() string {

op/applied/type_test.go

@@ -353,11 +353,30 @@ func TestPipelineSubPipeline(t *testing.T) {
 		},
 	}
 	p := NewPipeline(operations)
+	inputs := []struct {
+		startInclusive int
+		endExclusive   int
+		expected       Pipeline
+	}{
+		{
+			startInclusive: 0,
+			endExclusive:   0,
+			expected:       NewPipeline([]Union{}),
+		},
+		{
+			startInclusive: 0,
+			endExclusive:   4,
+			expected:       NewPipeline(operations),
+		},
+		{
+			startInclusive: 1,
+			endExclusive:   3,
+			expected:       NewPipeline(operations[1:3]),
+		},
+	}
 
-	for i := 0; i < 4; i++ {
-		subp := p.SubPipeline(i)
-		expected := NewPipeline(operations[i:])
-		require.Equal(t, expected, subp)
+	for _, input := range inputs {
+		require.Equal(t, input.expected, p.SubPipeline(input.startInclusive, input.endExclusive))
 	}
 }