perf(sql): Optimise searchForPipelinesByTrigger LIMIT and OFFSET SQL …

…query (#4698)

* fix(jooq): Refactor query for execution history retrieval

* fix(sql): Selecting only necessary fiels

(cherry picked from commit a27e88a)

orca-sql/src/main/kotlin/com/netflix/spinnaker/orca/sql/pipeline/persistence/SqlExecutionRepository.kt

@@ -654,36 +654,50 @@ class SqlExecutionRepository(
     executionCriteria: ExecutionCriteria
   ): List<PipelineExecution> {
     withPool(poolName) {
-      val select = jooq.selectExecutions(
-        PIPELINE,
-        conditions = {
-          var conditions = it.where(
+      val select = jooq.select(selectFields())
+        .from(PIPELINE.tableName)
+        .join(
+          jooq.selectExecutions(
+          PIPELINE,
+          listOf(field("id")),
+          conditions = {
+            var conditions = it.where(
             field("config_id").`in`(*pipelineConfigIds.toTypedArray())
               .and(field("build_time").gt(buildTimeStartBoundary))
               .and(field("build_time").lt(buildTimeEndBoundary))
-          )
+            )
 
-          if (executionCriteria.statuses.isNotEmpty()) {
-            val statusStrings = executionCriteria.statuses.map { it.toString() }
-            conditions = conditions.and(field("status").`in`(*statusStrings.toTypedArray()))
-          }
+            if (executionCriteria.statuses.isNotEmpty()) {
+              val statusStrings = executionCriteria.statuses.map { it.toString() }
+              conditions = conditions.and(field("status").`in`(*statusStrings.toTypedArray()))
+            }
 
-          conditions
-        },
-        seek = {
-          val seek = when (executionCriteria.sortType) {
-            ExecutionComparator.BUILD_TIME_ASC -> it.orderBy(field("build_time").asc())
-            ExecutionComparator.BUILD_TIME_DESC -> it.orderBy(field("build_time").desc())
-            ExecutionComparator.START_TIME_OR_ID -> it.orderBy(field("start_time").desc())
-            ExecutionComparator.NATURAL_ASC -> it.orderBy(field("id").desc())
-            else -> it.orderBy(field("id").asc())
+            conditions
+                       },
+          seek = {
+            val seek = when (executionCriteria.sortType) {
+              ExecutionComparator.BUILD_TIME_ASC -> it.orderBy(field("build_time").asc())
+              ExecutionComparator.BUILD_TIME_DESC -> it.orderBy(field("build_time").desc())
+              ExecutionComparator.START_TIME_OR_ID -> it.orderBy(field("start_time").desc())
+              ExecutionComparator.NATURAL_ASC -> it.orderBy(field("id").desc())
+              else -> it.orderBy(field("id").asc())
+            }
+            seek
+              .limit(executionCriteria.pageSize)
+              .offset((executionCriteria.page - 1) * executionCriteria.pageSize)
           }
-          seek
-            .limit(executionCriteria.pageSize)
-            .offset((executionCriteria.page - 1) * executionCriteria.pageSize)
-        }
-      )
-
+          )
+        )
+        .using(field("id"))
+        .orderBy(
+          when (executionCriteria.sortType) {
+            ExecutionComparator.BUILD_TIME_ASC -> field("build_time").asc()
+            ExecutionComparator.BUILD_TIME_DESC -> field("build_time").desc()
+            ExecutionComparator.START_TIME_OR_ID -> field("start_time").desc()
+            ExecutionComparator.NATURAL_ASC -> field("id").desc()
+            else -> field("id").asc()
+          }
+        )
       return select.fetchExecutions().toList()
     }
   }