atc: resolve pinned versions earlier

By resolving the pinned version to IDs earlier, we can run those queries
only once before the algorithm rather than having to run the query
multiple times if we put it in the algorithm.

[#3602]

Signed-off-by: Clara Fu <cfu@pivotal.io>
Co-authored-by: Rui Yang <ryang@pivotal.io>

atc/db/migration/migrations/1559071366_add_to_build_id_index_to_build_pipes.down.sql

@@ -0,0 +1,5 @@
+BEGIN;
+  DROP INDEX build_pipes_to_build_id_idx;
+
+  DROP INDEX build_resource_config_version_inputs_version_md5_idx;
+COMMIT;

atc/db/migration/migrations/1559071366_add_to_build_id_index_to_build_pipes.up.sql

@@ -0,0 +1,5 @@
+BEGIN;
+  CREATE INDEX build_pipes_to_build_id_idx ON build_pipes (to_build_id);
+
+  CREATE INDEX build_resource_config_version_inputs_version_md5_idx ON build_resource_config_version_inputs (version_md5);
+COMMIT;

atc/scheduler/algorithm/algorithm_test.go

@@ -1,6 +1,8 @@
 package algorithm_test
 
 import (
+	"github.com/concourse/concourse/atc"
+	"github.com/concourse/concourse/atc/scheduler/algorithm"
 	. "github.com/onsi/ginkgo/extensions/table"
 )
 
@@ -119,13 +121,7 @@ var _ = DescribeTable("Input resolving",
 			},
 		},
 
-		Result: Result{
-			OK:     false,
-			Values: map[string]string{},
-			Errors: map[string]string{
-				"resource-x": "pinned version ver:rxv2 not found",
-			},
-		},
+		Error: algorithm.PinnedVersionNotFoundError{atc.Version{"ver": "rxv2"}},
 	}),
 
 	Entry("finds only versions that passed through together", Example{
@@ -1685,13 +1681,7 @@ var _ = DescribeTable("Input resolving",
 			},
 		},
 
-		Result: Result{
-			OK:     false,
-			Values: map[string]string{},
-			Errors: map[string]string{
-				"resource-x": "pinned version ver:rxv2 not found",
-			},
-		},
+		Error: algorithm.PinnedVersionNotFoundError{atc.Version{"ver": "rxv2"}},
 	}),
 
 	Entry("resolves the version that is pinned with passed", Example{

atc/scheduler/algorithm/input_configs.go

@@ -3,6 +3,8 @@ package algorithm
 import (
 	"errors"
 	"fmt"
+	"log"
+	"strings"
 
 	"github.com/concourse/concourse/atc"
 	"github.com/concourse/concourse/atc/db"
@@ -14,7 +16,7 @@ type InputConfig struct {
 	Name            string
 	Passed          db.JobSet
 	UseEveryVersion bool
-	PinnedVersion   atc.Version
+	PinnedVersionID int
 	ResourceID      int
 	JobID           int
 }
@@ -92,7 +94,6 @@ func (im *inputMapper) MapInputs(
 	job db.Job,
 	resources db.Resources,
 ) (db.InputMapping, bool, error) {
-
 	inputConfigs := InputConfigs{}
 
 	for _, input := range job.Config().Inputs() {
@@ -113,16 +114,30 @@ func (im *inputMapper) MapInputs(
 			JobID:      job.ID(),
 		}
 
+		var pinnedVersion atc.Version
 		if resource.CurrentPinnedVersion() != nil {
-			inputConfig.PinnedVersion = resource.CurrentPinnedVersion()
+			pinnedVersion = resource.CurrentPinnedVersion()
 		}
 
 		if input.Version != nil {
 			inputConfig.UseEveryVersion = input.Version.Every
 
 			if input.Version.Pinned != nil {
-				inputConfig.PinnedVersion = input.Version.Pinned
+				pinnedVersion = input.Version.Pinned
+			}
+		}
+
+		if pinnedVersion != nil {
+			id, found, err := versions.FindVersionOfResource(inputConfig.ResourceID, pinnedVersion)
+			if err != nil {
+				return nil, false, err
+			}
+
+			if !found {
+				return nil, false, PinnedVersionNotFoundError{pinnedVersion}
 			}
+
+			inputConfig.PinnedVersionID = id
 		}
 
 		inputConfigs = append(inputConfigs, inputConfig)
@@ -220,19 +235,9 @@ func (im *inputMapper) tryResolve(depth int, db *db.VersionsDB, inputConfigs Inp
 			}
 
 			var versionID int
-			if inputConfig.PinnedVersion != nil {
+			if inputConfig.PinnedVersionID != 0 {
 				// pinned
-				id, found, err := db.FindVersionOfResource(inputConfig.ResourceID, inputConfig.PinnedVersion)
-				if err != nil {
-					return false, err
-				}
-
-				if !found {
-					unresolvedCandidates[i] = newCandidateError(PinnedVersionNotFoundError{inputConfig.PinnedVersion})
-					return false, nil
-				}
-
-				versionID = id
+				versionID = inputConfig.PinnedVersionID
 				debug("setting candidate", i, "to unconstrained version", versionID)
 			} else if inputConfig.UseEveryVersion {
 				buildID, found, err := db.LatestBuildID(inputConfig.JobID)
@@ -378,22 +383,10 @@ func (im *inputMapper) tryResolve(depth int, db *db.VersionsDB, inputConfigs Inp
 							break outputs
 						}
 
-						if inputConfigs[c].PinnedVersion != nil {
-							id, found, err := db.FindVersionOfResource(inputConfigs[c].ResourceID, inputConfigs[c].PinnedVersion)
-							if err != nil {
-								return false, err
-							}
-
-							if !found {
-								unresolvedCandidates[i] = newCandidateError(PinnedVersionNotFoundError{inputConfigs[c].PinnedVersion})
-								return false, nil
-							}
-
-							if id != output.VersionID {
-								debug("mismatch pinned version", output.VersionID, jobID)
-								mismatch = true
-								break outputs
-							}
+						if inputConfigs[c].PinnedVersionID != 0 && inputConfigs[c].PinnedVersionID != output.VersionID {
+							debug("mismatch pinned version", output.VersionID, jobID)
+							mismatch = true
+							break outputs
 						}
 
 						if candidate != nil && candidate.ID != output.VersionID {

atc/scheduler/algorithm/table_helpers_test.go

@@ -42,6 +42,7 @@ type Example struct {
 	DB     DB
 	Inputs Inputs
 	Result Result
+	Error  error
 }
 
 type Inputs []Input
@@ -369,17 +370,43 @@ func (example Example) Run() {
 		}
 
 		if len(input.Version.Pinned) != 0 {
-			inputConfigs[i].PinnedVersion = atc.Version{"ver": input.Version.Pinned}
+			versionJSON, err := json.Marshal(atc.Version{"ver": input.Version.Pinned})
+			Expect(err).ToNot(HaveOccurred())
+
+			var pinnedID int
+			rows, err := setup.psql.Select("rcv.id").
+				From("resource_config_versions rcv").
+				Join("resources r ON r.resource_config_scope_id = rcv.resource_config_scope_id").
+				Where(sq.Eq{
+					"rcv.version": versionJSON,
+					"r.id":        inputConfigs[i].ResourceID,
+				}).
+				Query()
+			Expect(err).ToNot(HaveOccurred())
+
+			if rows.Next() {
+				err = rows.Scan(&pinnedID)
+				Expect(err).ToNot(HaveOccurred())
+			}
+
+			rows.Close()
+
+			if pinnedID != 0 {
+				inputConfigs[i].PinnedVersionID = pinnedID
+			} else {
+				// Pinning to an inexistant version id
+				inputConfigs[i].PinnedVersionID = 123
+			}
 		}
 	}
 
 	inputs := atc.PlanSequence{}
-	for _, input := range inputConfigs {
+	for i, input := range inputConfigs {
 		var version *atc.VersionConfig
 		if input.UseEveryVersion {
 			version = &atc.VersionConfig{Every: true}
-		} else if input.PinnedVersion != nil {
-			version = &atc.VersionConfig{Pinned: input.PinnedVersion}
+		} else if input.PinnedVersionID != 0 {
+			version = &atc.VersionConfig{Pinned: atc.Version{"ver": example.Inputs[i].Version.Pinned}}
 		} else {
 			version = &atc.VersionConfig{Latest: true}
 		}
@@ -455,33 +482,37 @@ func (example Example) Run() {
 
 	inputMapper := algorithm.NewInputMapper()
 	resolved, ok, err := inputMapper.MapInputs(versionsDB, job, dbResources)
-	Expect(err).ToNot(HaveOccurred())
+	if example.Error != nil {
+		Expect(err).To(Equal(example.Error))
+	} else {
+		Expect(err).ToNot(HaveOccurred())
 
-	prettyValues := map[string]string{}
-	erroredValues := map[string]string{}
-	skippedValues := map[string]bool{}
-	for name, inputSource := range resolved {
-		if inputSource.ResolveSkipped == true {
-			skippedValues[name] = true
-		} else if inputSource.ResolveError != nil {
-			erroredValues[name] = inputSource.ResolveError.Error()
-		} else {
-			prettyValues[name] = setup.versionIDs.Name(inputSource.Input.AlgorithmVersion.VersionID)
+		prettyValues := map[string]string{}
+		erroredValues := map[string]string{}
+		skippedValues := map[string]bool{}
+		for name, inputSource := range resolved {
+			if inputSource.ResolveSkipped == true {
+				skippedValues[name] = true
+			} else if inputSource.ResolveError != nil {
+				erroredValues[name] = inputSource.ResolveError.Error()
+			} else {
+				prettyValues[name] = setup.versionIDs.Name(inputSource.Input.AlgorithmVersion.VersionID)
+			}
 		}
-	}
 
-	actualResult := Result{OK: ok}
-	if len(erroredValues) != 0 {
-		actualResult.Errors = erroredValues
-	}
+		actualResult := Result{OK: ok}
+		if len(erroredValues) != 0 {
+			actualResult.Errors = erroredValues
+		}
 
-	if len(skippedValues) != 0 {
-		actualResult.Skipped = skippedValues
-	}
+		if len(skippedValues) != 0 {
+			actualResult.Skipped = skippedValues
+		}
 
-	actualResult.Values = prettyValues
+		actualResult.Values = prettyValues
 
-	Expect(actualResult).To(Equal(example.Result))
+		Expect(actualResult).To(Equal(example.Result))
+	}
 }
 
 type setupDB struct {

ci/pipelines/branch.yml

@@ -10,6 +10,10 @@ resource_types:
   type: registry-image
   source: {repository: frodenas/gcs-resource}
 
+- name: bosh-release
+  type: registry-image
+  source: {repository: dpb587/bosh-release-resource}
+
 groups:
 - name: develop
   jobs:
@@ -317,13 +321,13 @@ jobs:
   plan:
   - in_parallel:
     - get: concourse
-      passed: [testflight, watsjs, upgrade, downgrade]
+      passed: [testflight, watsjs]
       trigger: true
     - get: dev-image
       trigger: true
-      passed: [testflight, watsjs, upgrade, downgrade]
+      passed: [testflight, watsjs]
     - get: unit-image
-      passed: [testflight, watsjs, upgrade, downgrade]
+      passed: [testflight, watsjs]
       trigger: true
 
 - name: build-rc
@@ -452,13 +456,13 @@ jobs:
   plan:
   - in_parallel:
     - get: concourse
-      passed: [bosh-check-props]
+      passed: [build-rc]
       trigger: true
     - get: unit-image
-      passed: [bosh-check-props]
+      passed: [build-rc]
       trigger: true
     - get: linux-rc
-      passed: [build-rc, bosh-check-props]
+      passed: [build-rc]
       trigger: true
     - get: windows-rc
       passed: [build-rc]