Repository: kestra-io/kestra
Issue: Improve PostgreSQL JDBC queue poll performance

### Issue description

There has been poll performance optimization done already (#1012), although the query still tends to execute rather slow (hundreds of milliseconds). The query plan seems to use _Seq Scan_ instead of using the indexes.

Query plan even after running `VACUUM ANALYZE queues;` on the table with 3+M rows:
```Limit  (cost=353189.18..353195.43 rows=500 width=662) (actual time=828.565..828.566 rows=0 loops=1)
  Output: value, "offset", ctid
  ->  LockRows  (cost=353189.18..357966.07 rows=382151 width=662) (actual time=828.563..828.564 rows=0 loops=1)
        Output: value, "offset", ctid
        ->  Sort  (cost=353189.18..354144.56 rows=382151 width=662) (actual time=828.563..828.563 rows=0 loops=1)
              Output: value, "offset", ctid
              Sort Key: queues."offset"
              Sort Method: quicksort  Memory: 25kB
              ->  Seq Scan on public.queues  (cost=0.00..334147.01 rows=382151 width=662) (actual time=828.550..828.550 rows=0 loops=1)
                    Output: value, "offset", ctid
                    Filter: ((NOT queues.consumer_executor) AND (queues.type = 'io.kestra.core.models.executions.Execution'::queue_type))
                    Rows Removed by Filter: 3094525
Query Identifier: 5205766377754712210
Planning Time: 0.092 ms
Execution Time: 828.587 ms
```

Tuning the cost optimizer parameter `random_page_cost=1.1` leads to correct usage of indices:

```Limit  (cost=271995.12..272001.37 rows=500 width=662) (actual time=0.015..0.016 rows=0 loops=1)
  Output: value, "offset", ctid
  ->  LockRows  (cost=271995.12..276772.00 rows=382151 width=662) (actual time=0.015..0.015 rows=0 loops=1)
        Output: value, "offset", ctid
        ->  Sort  (cost=271995.12..272950.49 rows=382151 width=662) (actual time=0.014..0.015 rows=0 loops=1)
              Output: value, "offset", ctid
              Sort Key: queues."offset"
              Sort Method: quicksort  Memory: 25kB
              ->  Index Scan using queues_type__consumer_executor on public.queues  (cost=0.43..252952.94 rows=382151 width=662) (actual time=0.012..0.012 rows=0 loops=1)
                    Output: value, "offset", ctid
                    Index Cond: ((queues.type = 'io.kestra.core.models.executions.Execution'::queue_type) AND (queues.consumer_executor = false))
Query Identifier: 5205766377754712210
Planning Time: 0.091 ms
Execution Time: 0.032 ms
```

Issue comments:
I notice that when there is a lot of new entries in the queues table it uses a SeqScan.

Then, if you recompute the statistics via `analyze queues` it correctly uses the right index (Bitmap Index Scan of `queues_type__consumer_scheduler` for scheduler related queues).

Is the performance issue has been found on newly executed Kestra instance after created a lot of executions (for ex during a benchmark) ? Does it disappear after time or after recomputed the statistics ?
It was observed during performance benchmarking - the number of executions was high. Running `VACUUM ANALYZE queues;` did not have any effect on the query plan.

Sadly, no long term effects were observed since we hit this weird issue just during benchmarking.

Pull request review comments:

Patch that resolved the issue:
diff --git a/jdbc-h2/src/main/resources/migrations/h2/V18__index_logs.sql b/jdbc-h2/src/main/resources/migrations/h2/V18__index_logs.sql
new file mode 100644
index 0000000000..03e23cefe5
--- /dev/null
+++ b/jdbc-h2/src/main/resources/migrations/h2/V18__index_logs.sql
@@ -0,0 +1,3 @@
+DROP INDEX logs_namespace;
+DROP INDEX logs_timestamp;
+CREATE INDEX logs_namespace_flow ON logs ("deleted", "timestamp", "level", "namespace", "flow_id");
\ No newline at end of file
diff --git a/jdbc-postgres/src/main/resources/migrations/postgres/V13__queue_index_optim.sql b/jdbc-postgres/src/main/resources/migrations/postgres/V13__queue_index_optim.sql
deleted file mode 100644
index 6769520e51..0000000000
--- a/jdbc-postgres/src/main/resources/migrations/postgres/V13__queue_index_optim.sql
+++ /dev/null
@@ -1,8 +0,0 @@
--- We drop the PK and the queues_type__offset, otherwise they are used by the poll query which is sub-optimal.
--- We create an hash index on offset that will be used instead when filtering on offset.
-
-ALTER TABLE queues DROP CONSTRAINT IF EXISTS queues_pkey;
-
-DROP INDEX IF EXISTS queues_type__offset;
-
-CREATE INDEX IF NOT EXISTS queues_offset ON queues USING hash ("offset");
diff --git a/jdbc-postgres/src/main/resources/migrations/postgres/V21_index_logs.sql b/jdbc-postgres/src/main/resources/migrations/postgres/V21_index_logs.sql
deleted file mode 100644
index 670d66af69..0000000000
--- a/jdbc-postgres/src/main/resources/migrations/postgres/V21_index_logs.sql
+++ /dev/null
@@ -1,3 +0,0 @@
-DROP INDEX logs_namespace;
-DROP INDEX logs_timestamp;
-CREATE INDEX logs_namespace_flow ON logs (deleted, timestamp, level, namespace, flow_id);
\ No newline at end of file
diff --git a/jdbc-postgres/src/main/resources/migrations/postgres/V22__index_queues.sql b/jdbc-postgres/src/main/resources/migrations/postgres/V22__index_queues.sql
new file mode 100644
index 0000000000..4fe5ac87d6
--- /dev/null
+++ b/jdbc-postgres/src/main/resources/migrations/postgres/V22__index_queues.sql
@@ -0,0 +1,24 @@
+-- Recreate the queues_type__* indexes by adding the offset column otherwise the index is not used as we order on offset.
+-- Also make them partial to lower the index size.
+DROP INDEX queues_type__consumer_flow_topology;
+DROP INDEX queues_type__consumer_indexer;
+DROP INDEX queues_type__consumer_executor;
+DROP INDEX queues_type__consumer_worker;
+DROP INDEX queues_type__consumer_scheduler;
+
+CREATE INDEX queues_type__consumer_flow_topology ON queues (type, consumer_flow_topology, "offset") WHERE consumer_flow_topology = false;
+CREATE INDEX queues_type__consumer_indexer ON queues (type, consumer_indexer, "offset") WHERE consumer_indexer = false;
+CREATE INDEX queues_type__consumer_executor ON queues (type, consumer_executor, "offset") WHERE consumer_executor = false;
+CREATE INDEX queues_type__consumer_worker ON queues (type, consumer_worker, "offset") WHERE consumer_worker = false;
+CREATE INDEX queues_type__consumer_scheduler ON queues (type, consumer_scheduler, "offset") WHERE consumer_scheduler = false;
+
+-- Go back to the original PK and queues_offset__type as they are useful for offset based poll and updates
+DO $$
+    BEGIN
+        IF NOT exists (select constraint_name from information_schema.table_constraints where table_name = 'queues' and constraint_type = 'PRIMARY KEY') then
+            ALTER TABLE queues ADD PRIMARY KEY("offset");
+        END IF;
+    END;
+$$;
+DROP INDEX IF EXISTS queues_offset;
+CREATE INDEX IF NOT EXISTS queues_type__offset ON queues (type, "offset");
\ No newline at end of file


Test patch: