perf(artifacts): use pageSize=1 when resolving prior artifacts (#2955)

This is a smaller-scoped attempt at
#2938, which relates to
spinnaker/spinnaker#4367.

`ArtifactResolver.resolveArtifacts(Map)` currently attempts to load
_all_ executions for a pipeline in order to find the most recent
execution in the course of resolving expected artifacts. This causes a
lot of unnecessary data to be loaded from Redis/SQL, only to be
discarded a few operations later.

This change makes use of the fact that the ExecutionRepository
implementations will respect the `executionCriteria.pageSize` argument
when retrieving Execitions for a pipelineId.

In the Redis-based implementation, the executions are stored in a sorted
set scored on `buildTime` (or `currentTimeMillis()` if `buildTime` is
null), so retrieving all of the executions for the pipelineId with a
pageSize=1 should load the Execution with the most recent `buildTime`.

In the SQL-based implementation,
`retrievePipelinesForPipelineConfigId(String, ExecutionCriteria)` sorts
the query results based on the `id` field.

For both implementations, this is a small change from the existing
behavior of ArtifactResolver, which retrieves all executions and then
uses the one with the most recent `startTime` (or `id`). This change
seems like it will lead to the same result in most cases though, since
buildTime is set when the Execution is stored, and the `id` field is
ascending based on the timestamp of when it is generated.

The `retrievePipelinesForPipelienConfigId` method in both
implementations currently ignores the `executionCriteria.sortType`
field, but I've added this in the call from ArtifactResolver to at
least document ArtifactResolver's desire.

orca-core/src/main/java/com/netflix/spinnaker/orca/pipeline/util/ArtifactResolver.java

@@ -252,8 +252,7 @@ public void resolveArtifacts(@Nonnull Map<String, Object> pipeline) {
       return;
     }
 
-    List<Artifact> priorArtifacts =
-        getArtifactsForPipelineId((String) pipeline.get("id"), new ExecutionCriteria());
+    List<Artifact> priorArtifacts = getPriorArtifacts(pipeline);
     LinkedHashSet<Artifact> resolvedArtifacts =
         resolveExpectedArtifacts(expectedArtifacts, receivedArtifacts, priorArtifacts, true);
     LinkedHashSet<Artifact> allArtifacts = new LinkedHashSet<>(receivedArtifacts);
@@ -277,6 +276,16 @@ public void resolveArtifacts(@Nonnull Map<String, Object> pipeline) {
     }
   }
 
+  private List<Artifact> getPriorArtifacts(final Map<String, Object> pipeline) {
+    // set pageSize to a single record to avoid hydrating all of the stored Executions for
+    // the pipeline, since getArtifactsForPipelineId only uses the most recent Execution from the
+    // returned Observable<Execution>
+    ExecutionCriteria criteria = new ExecutionCriteria();
+    criteria.setPageSize(1);
+    criteria.setSortType(ExecutionRepository.ExecutionComparator.START_TIME_OR_ID);
+    return getArtifactsForPipelineId((String) pipeline.get("id"), criteria);
+  }
+
   public Artifact resolveSingleArtifact(
       ExpectedArtifact expectedArtifact,
       List<Artifact> possibleMatches,

orca-core/src/test/groovy/com/netflix/spinnaker/orca/pipeline/util/ArtifactResolverSpec.groovy

@@ -32,8 +32,20 @@ import static com.netflix.spinnaker.orca.test.model.ExecutionBuilder.stage
 
 class ArtifactResolverSpec extends Specification {
   ObjectMapper objectMapper = new ObjectMapper()
+
+  def pipelineId = "abc"
+
+  def expectedExecutionCriteria = {
+    def criteria = new ExecutionRepository.ExecutionCriteria()
+    criteria.setPageSize(1)
+    return criteria
+  }()
+
   def executionRepository = Stub(ExecutionRepository) {
-    retrievePipelinesForPipelineConfigId(*_) >> Observable.empty();
+    // only a call to retrievePipelinesForPipelineConfigId() with these argument values is expected
+    retrievePipelinesForPipelineConfigId(pipelineId, expectedExecutionCriteria) >> Observable.empty();
+    // any other interaction is unexpected
+    0 * _
   }
 
   def makeArtifactResolver() {