diff --git a/orca-sql/src/main/kotlin/com/netflix/spinnaker/config/SqlConfiguration.kt b/orca-sql/src/main/kotlin/com/netflix/spinnaker/config/SqlConfiguration.kt
index 741b905cbe..83b5dfb59e 100644
--- a/orca-sql/src/main/kotlin/com/netflix/spinnaker/config/SqlConfiguration.kt
+++ b/orca-sql/src/main/kotlin/com/netflix/spinnaker/config/SqlConfiguration.kt
@@ -17,8 +17,10 @@ package com.netflix.spinnaker.config
 
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.netflix.spectator.api.Registry
+import com.netflix.spinnaker.kork.dynamicconfig.DynamicConfigService
 import com.netflix.spinnaker.orca.sql.JooqSqlCommentAppender
 import com.netflix.spinnaker.orca.sql.JooqToSpringExceptionTransformer
+import com.netflix.spinnaker.orca.sql.QueryLogger
 import com.netflix.spinnaker.orca.sql.SpringLiquibaseProxy
 import com.netflix.spinnaker.orca.sql.SqlHealthIndicator
 import com.netflix.spinnaker.orca.sql.SqlHealthcheckActivator
@@ -80,13 +82,15 @@ class SqlConfiguration {
 
   @Bean fun jooqConfiguration(
     connectionProvider: DataSourceConnectionProvider,
-    properties: SqlProperties
+    properties: SqlProperties,
+    dynamicConfigService: DynamicConfigService
   ): DefaultConfiguration =
     DefaultConfiguration().apply {
       set(*DefaultExecuteListenerProvider.providers(
         JooqToSpringExceptionTransformer(),
         JooqSqlCommentAppender(),
-        SlowQueryLogger()
+        SlowQueryLogger(),
+        QueryLogger(dynamicConfigService)
       ))
       set(connectionProvider)
       setSQLDialect(properties.connectionPool.dialect)
diff --git a/orca-sql/src/main/kotlin/com/netflix/spinnaker/orca/sql/QueryLogger.kt b/orca-sql/src/main/kotlin/com/netflix/spinnaker/orca/sql/QueryLogger.kt
new file mode 100644
index 0000000000..4e45fd5c82
--- /dev/null
+++ b/orca-sql/src/main/kotlin/com/netflix/spinnaker/orca/sql/QueryLogger.kt
@@ -0,0 +1,58 @@
+/*
+ * Copyright 2019 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.netflix.spinnaker.orca.sql
+
+import com.netflix.spinnaker.kork.dynamicconfig.DynamicConfigService
+import org.jooq.ExecuteContext
+import org.jooq.impl.DefaultExecuteListener
+import org.slf4j.LoggerFactory
+import java.util.regex.Pattern
+import java.util.regex.PatternSyntaxException
+
+/**
+ * Simple query logger, for when you don't/can't run general query logging in prod (looking at you, Aurora).
+ *
+ * When enabled, healthcheck queries will not be reported by default.
+ */
+class QueryLogger(
+  private val dynamicConfigService: DynamicConfigService
+) : DefaultExecuteListener() {
+
+  private val log = LoggerFactory.getLogger(javaClass)
+
+  override fun end(ctx: ExecuteContext) {
+    if (dynamicConfigService.isEnabled("sql.query-logger", false)) {
+      val statements = ctx.batchSQL().joinToString("\n")
+      val exclude = dynamicConfigService.getConfig(
+        String::class.java,
+        "sql.query-logger.exclude-pattern",
+        ".*healthcheck.*"
+      )
+      if (exclude.isNotEmpty()) {
+        val pattern: Pattern? = try {
+          Pattern.compile(exclude)
+        } catch (e: PatternSyntaxException) {
+          log.error("Exclusion pattern invalid: '$exclude'", e)
+          null
+        }
+        if (pattern?.matcher(statements)?.matches() == true) {
+          return
+        }
+      }
+      log.debug(statements)
+    }
+  }
+}
diff --git a/orca-sql/src/main/kotlin/com/netflix/spinnaker/orca/sql/pipeline/persistence/SqlExecutionRepository.kt b/orca-sql/src/main/kotlin/com/netflix/spinnaker/orca/sql/pipeline/persistence/SqlExecutionRepository.kt
index 88baa9f3d1..c68be6256d 100644
--- a/orca-sql/src/main/kotlin/com/netflix/spinnaker/orca/sql/pipeline/persistence/SqlExecutionRepository.kt
+++ b/orca-sql/src/main/kotlin/com/netflix/spinnaker/orca/sql/pipeline/persistence/SqlExecutionRepository.kt
@@ -192,11 +192,26 @@ class SqlExecutionRepository(
   }
 
   override fun delete(type: ExecutionType, id: String) {
+    val correlationField = if (type == PIPELINE) "pipeline_id" else "orchestration_id"
+    val (ulid, _) = mapLegacyId(jooq, type.tableName, id)
+
+    val correlationId = jooq.select(field("id")).from("correlation_ids")
+      .where(field(correlationField).eq(ulid))
+      .limit(1)
+      .fetchOne()
+      ?.into(String::class.java)
+    val stageIds = jooq.select(field("id")).from(type.stagesTableName)
+      .where(field("execution_id").eq(ulid))
+      .fetch()
+      ?.into(String::class.java)?.toTypedArray()
+
     jooq.transactional {
-      val correlationField = if (type == PIPELINE) "pipeline_id" else "orchestration_id"
-      val (ulid, _) = mapLegacyId(it, type.tableName, id)
-      it.delete(table("correlation_ids")).where(field(correlationField).eq(ulid)).execute()
-      it.delete(type.stagesTableName).where(field("execution_id").eq(ulid)).execute()
+      if (correlationId != null) {
+        it.delete(table("correlation_ids")).where(field("id").eq(correlationId)).execute()
+      }
+      if (stageIds != null) {
+        it.delete(type.stagesTableName).where(field("id").`in`(*stageIds)).execute()
+      }
       it.delete(type.tableName).where(field("id").eq(ulid)).execute()
     }
   }
diff --git a/orca-sql/src/main/resources/db/changelog-master.yml b/orca-sql/src/main/resources/db/changelog-master.yml
index 7beeba325e..9828414bd2 100644
--- a/orca-sql/src/main/resources/db/changelog-master.yml
+++ b/orca-sql/src/main/resources/db/changelog-master.yml
@@ -29,3 +29,6 @@ databaseChangeLog:
 - include:
     file: changelog/20181016-add-start-time.yml
     relativeToChangelogFile: true
+- include:
+    file: changelog/20190516-index-correlated-execution-ids.yml
+    relativeToChangelogFile: true
diff --git a/orca-sql/src/main/resources/db/changelog/20190516-index-correlated-execution-ids.yml b/orca-sql/src/main/resources/db/changelog/20190516-index-correlated-execution-ids.yml
new file mode 100644
index 0000000000..8f3b13b300
--- /dev/null
+++ b/orca-sql/src/main/resources/db/changelog/20190516-index-correlated-execution-ids.yml
@@ -0,0 +1,24 @@
+databaseChangeLog:
+  - changeSet:
+      id: 20190516-index-correlated-execution-ids
+      author: robzienert
+      changes:
+        - createIndex:
+            indexName: pipeline_id_idx
+            tableName: correlation_ids
+            columns:
+              - column:
+                  name: pipeline_id
+        - createIndex:
+            indexName: orchestration_id_idx
+            tableName: correlation_ids
+            columns:
+              - column:
+                  name: orchestration_id
+      rollback:
+        - dropIndex:
+            indexName: orchestration_id_idx
+            tableName: correlation_ids
+        - dropIndex:
+            indexName: pipeline_id_idx
+            tableName: correlation_ids