Performance issues with parallelism and LIMIT

Hi hackers,

While migrating from PostgreSQL 14 to 15, we encountered the following 
performance degradation caused by commit 46846433a03dff: "shm_mq: Update 
mq_bytes_written less often", discussion in [1].

The batching can make queries with a LIMIT clause run significantly 
slower compared to PostgreSQL 14, because neither the ring buffer write 
position is updated, nor the latch to inform the leader that there's 
data available is set, before a worker's queue is 1/4th full. This can 
be seen in the number of rows produced by a parallel worker. Worst-case, 
the data set is large and all rows to answer the query appear early, but 
are not big enough to fill the queue to 1/4th (e.g. when the LIMIT and 
the tuple sizes are small). Here is an example to reproduce the problem.

CREATE TABLE t(id1 INT, id2 INT, id3 INT, id4 INT, id5 INT);
INSERT INTO t(id1, id2, id3, id4, id5) SELECT i%1000, i, i, i, i FROM 
generate_series(1, 10000000) AS i;
ANALYZE t;
SET parallel_tuple_cost = 0;
SET parallel_setup_cost = 0;
SET min_parallel_table_scan_size = 0;
SET max_parallel_workers_per_gather = 8;
EXPLAIN ANALYZE VERBOSE SELECT id2 FROM t WHERE id1 = 100 LIMIT 100;

PostgreSQL 15:

  Limit  (cost=0.00..797.43 rows=100 width=4) (actual 
time=65.083..69.207 rows=100 loops=1)
    Output: id2
    ->  Gather  (cost=0.00..79320.18 rows=9947 width=4) (actual 
time=65.073..68.417 rows=100 loops=1)
          Output: id2
          Workers Planned: 8
          Workers Launched: 7
          ->  Parallel Seq Scan on public.t (cost=0.00..79320.18 
rows=1243 width=4) (actual time=0.204..33.049 rows=100 loops=7)
                Output: id2
                Filter: (t.id1 = 100)
                Rows Removed by Filter: 99345
                Worker 0:  actual time=0.334..32.284 rows=100 loops=1
                Worker 1:  actual time=0.060..32.680 rows=100 loops=1
                Worker 2:  actual time=0.637..33.954 rows=98 loops=1
                Worker 3:  actual time=0.136..33.301 rows=100 loops=1
                Worker 4:  actual time=0.140..31.942 rows=100 loops=1
                Worker 5:  actual time=0.062..33.673 rows=100 loops=1
                Worker 6:  actual time=0.062..33.512 rows=100 loops=1
  Planning Time: 0.113 ms
  Execution Time: 69.772 ms

PostgreSQL 14:

  Limit  (cost=0.00..797.75 rows=100 width=4) (actual 
time=30.602..38.459 rows=100 loops=1)
    Output: id2
    ->  Gather  (cost=0.00..79320.18 rows=9943 width=4) (actual 
time=30.592..37.669 rows=100 loops=1)
          Output: id2
          Workers Planned: 8
          Workers Launched: 7
          ->  Parallel Seq Scan on public.t (cost=0.00..79320.18 
rows=1243 width=4) (actual time=0.221..5.181 rows=15 loops=7)
                Output: id2
                Filter: (t.id1 = 100)
                Rows Removed by Filter: 15241
                Worker 0:  actual time=0.129..4.840 rows=15 loops=1
                Worker 1:  actual time=0.125..4.924 rows=15 loops=1
                Worker 2:  actual time=0.314..5.249 rows=17 loops=1
                Worker 3:  actual time=0.252..5.341 rows=15 loops=1
                Worker 4:  actual time=0.163..5.179 rows=15 loops=1
                Worker 5:  actual time=0.422..5.248 rows=15 loops=1
                Worker 6:  actual time=0.139..5.489 rows=16 loops=1
  Planning Time: 0.084 ms
  Execution Time: 38.880 ms

I had a quick look at the code and I started wondering if we can't 
achieve the same performance improvement without batching by e.g.:

- Only set the latch if new data is written to an empty queue. 
Otherwise, the leader should anyways keep try reading from the queues 
without waiting for the latch, so no need to set the latch again.

- Reorganize struct shm_mq. There seems to be false sharing happening 
between at least mq_ring_size and the atomics and potentially also 
between the atomics. I'm wondering if the that's not the root cause of 
the "slow atomics" observed in [1]? I'm happy to do some profiling.

Alternatively, we could always set the latch if numberTuples in 
ExecutePlan() is reasonably low. To do so, the DestReceiver's receive() 
method would only need an additional "force flush" argument.


A slightly different but related problem is when some workers have 
already produced enough rows to answer the LIMIT query, but other 
workers are still running without producing any new rows. In that case 
the "already done" workers will stop running even though they haven't 
reached 1/4th of the queue size, because the for-loop in execMain.c 
bails out in the following condition:

         if (numberTuples && numberTuples == current_tuple_count)
             break;

Subsequently, the leader will end the plan and then wait in the Gather 
node for all workers to shutdown. However, workers still running but not 
producing any new rows will never reach the following condition in 
execMain.c to check if they're supposed to stop (the shared memory queue 
dest receiver will return false on detached queues):

             /*
              * If we are not able to send the tuple, we assume the 
destination
              * has closed and no more tuples can be sent. If that's the 
case,
              * end the loop.
              */
             if (!dest->receiveSlot(slot, dest))
                 break;

Reproduction steps for this problem are below. Here the worker getting 
the first table page will be done right away, but the query takes as 
long as it takes to scan all pages of the entire table.

CREATE TABLE bar (col INT);
INSERT INTO bar SELECT generate_series(1, 5000000);
SET max_parallel_workers_per_gather = 8;
EXPLAIN ANALYZE VERBOSE SELECT col FROM bar WHERE col = 1 LIMIT 1;

  Limit  (cost=0.00..1.10 rows=1 width=4) (actual time=32.289..196.200 
rows=1 loops=1)
    Output: col
    ->  Gather  (cost=0.00..30939.03 rows=28208 width=4) (actual 
time=32.278..196.176 rows=1 loops=1)
          Output: col
          Workers Planned: 8
          Workers Launched: 7
          ->  Parallel Seq Scan on public.bar (cost=0.00..30939.03 
rows=3526 width=4) (actual time=137.251..137.255 rows=0 loops=7)
                Output: col
                Filter: (bar.col = 1)
                Rows Removed by Filter: 713769
                Worker 0:  actual time=160.177..160.181 rows=0 loops=1
                Worker 1:  actual time=160.111..160.115 rows=0 loops=1
                Worker 2:  actual time=0.043..0.047 rows=1 loops=1
                Worker 3:  actual time=160.040..160.044 rows=0 loops=1
                Worker 4:  actual time=160.167..160.171 rows=0 loops=1
                Worker 5:  actual time=160.018..160.022 rows=0 loops=1
                Worker 6:  actual time=160.201..160.205 rows=0 loops=1
  Planning Time: 0.087 ms
  Execution Time: 196.247 ms

We would need something similar to CHECK_FOR_INTERRUPTS() which returns 
a NULL slot if a parallel worker is supposed to stop execution (we could 
e.g. check if the queue got detached). Or could we amend 
CHECK_FOR_INTERRUPTS() to just stop the worker gracefully if the queue 
got detached?

Jasper Smit, Spiros Agathos and Dimos Stamatakis helped working on this.

[1] 
https://www.postgresql.org/message-id/flat/CAFiTN-tVXqn_OG7tHNeSkBbN%2BiiCZTiQ83uakax43y1sQb2OBA%40mail.gmail.com

-- 
David Geier
(ServiceNow)