Parallella förfrågningar i PostgreSQL

I moderna CPU:er finns det många kärnor. Under flera år har applikationer skickat förfrågningar till databaser parallellt. Om det är en rapportförfrågan om många rader i en tabell utförs den snabbare när flera CPU:er är involverade, och i PostgreSQL är detta möjligt sedan version 9.6.

Det tog tre år att implementera funktionen för parallella förfrågningar – koden behövde skrivas om vid olika steg i förfrågningarnas utförande. PostgreSQL 9.6 introducerade infrastrukturen för att ytterligare förbättra koden. I senare versioner utförs även andra typer av förfrågningar parallellt.

Begränsningar

• Aktivera inte parallell exekvering om alla kärnor redan är upptagna, annars kommer andra förfrågningar att bromsa.
• Det viktigaste är att parallell behandling med höga values WORK_MEM använder mycket minne – varje hash-förbindelse eller sortering tar minne motsvarande work_mem.
• OLTP-förfrågningar med låg latens kan inte öka hastigheten med parallell exekvering. Om en förfrågan returnerar en rad kommer den parallella behandlingen bara att sakta ner den.
• Utvecklare älskar att använda TPC-H-benchmarken. Kanske har du liknande förfrågningar för perfekt parallell exekvering.
• Endast SELECT-förfrågningar utan predikantlåsning utförs parallellt.
• Ibland kan rätt indexering vara bättre än sekventiell skanning av tabellen i parallell läge.
• Stopp av förfrågningar och markörer stöds inte.
• Fönsterfunktioner och aggregerade funktioner av ordnade uppsättningar är inte parallella.
• Du vinner inget i in- och utdataarbetsbelastning.
• Det finns inga parallella sorteringsalgoritmer. Men förfrågningar med sorteringar kan exekveras parallellt i vissa aspekter.
• Ersätt CTE (WITH …) med en inbäddad SELECT för att inkludera parallell bearbetning.
• Wraps för tredjepartsdata stöder fortfarande inte parallell bearbetning (men det skulle kunna!)
• FULL OUTER JOIN stöds inte.
• max_rows inaktiverar parallell bearbetning.
• Om en förfrågan innehåller en funktion som inte är markerad som PARALLEL SAFE kommer den att vara en-trådad.
• Transaktionsisoleringsnivån SERIALIZABLE inaktiverar parallell bearbetning.

Testmiljö

PostgreSQL-utvecklarna försökte minska svarstiden för TPC-H-benchmarkfågor. Ladda ner benchmarken och anpassa den till PostgreSQL. Detta är ett icke-officiellt användande av TPC-H-benchmarken – inte för att jämföra databaser eller hårdvara.

1. Ladda ner TPC-H_Tools_v2.17.3.zip (eller nyare version) från TPC:s officiella webbplats.
2. Ändra namnet på makefile.suite till Makefile och som beskrivs här: https://github.com/tvondra/pg_tpch . Kompilera koden med kommandot make.
3. Generera data: ./dbgen -s 10 skapar en databas på 23 GB. Det räcker för att se skillnaden i prestanda mellan parallella och icke-parallella förfrågningar.
4. Konvertera filer tbl på csv till för och sed.
5. Klona repositoriet pg_tpch och kopiera filerna csv på pg_tpch/dss/data.
6. Skapa förfrågningar med kommandot qgen.
7. Ladda upp data i databasen med kommandot ./tpch.sh.

Parallell sekventiell skanning

Det kan vara snabbare, inte för att data läses parallellt, utan eftersom data sprids över många CPU-kärnor. I moderna operativsystem cachas PostgreSQL-datafiler väl. Med förhandsläsning kan man hämta fler block från lagringen än vad PG-processen begär. Därför begränsas förfrågningsprestandan inte av diskens I/O. Den använder CPU-cykler för att:

• läsa rader en och en från tabellens sidor;
• jämföra värden och villkor för rader. DÄR.

Låt oss utföra en enkel förfrågning select:

tpch=# explain analyze select l_quantity as sum_qty from lineitem where l_shipdate <= date '1998-12-01' - interval '105' day;
QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------
Seq Scan on lineitem (cost=0.00..1964772.00 rows=58856235 width=5) (actual time=0.014..16951.669 rows=58839715 loops=1)
Filter: (l_shipdate <= '1998-08-18 00:00:00'::timestamp without time zone)
Rows Removed by Filter: 1146337
Planning Time: 0.203 ms
Execution Time: 19035.100 ms

Sekventiell skanning ger för många rader utan aggregering, så förfrågan körs med en enda CPU-kärna.

Om man lägger till SUM(), syns det att två arbetstrådar hjälper till att påskynda förfrågan:

explain analyze select sum(l_quantity) as sum_qty from lineitem where l_shipdate  Gather (cost=1589701.91..1589702.12 rows=2 width=32) (actual time=8553.241..8555.067 rows=3 loops=1)
Workers Planned: 2
Workers Launched: 2
-> Partial Aggregate (cost=1588701.91..1588701.92 rows=1 width=32) (actual time=8547.546..8547.546 rows=1 loops=3)
-> Parallel Seq Scan on lineitem (cost=0.00..1527393.33 rows=24523431 width=5) (actual time=0.038..5998.417 rows=19613238 loops=3)
Filter: (l_shipdate <= '1998-08-18 00:00:00'::timestamp without time zone)
Rows Removed by Filter: 382112
Planning Time: 0.241 ms
Execution Time: 8555.131 ms

Parallell aggregering

Noden «Parallel Seq Scan» producerar rader för partiell aggregering. Noden «Partial Aggregate» filtrerar dessa rader med hjälp av SUM(). I slutet sammanställs SUM-räknaren från varje arbetstråd av noden «Gather».

Det slutgiltiga resultatet beräknas av noden "Finalize Aggregate". Om du har egna aggregeringsfunktioner, glöm inte att märka dem som "parallel safe".

Antal arbetsprocesser

Antalet arbetsprocesser kan ökas utan att starta om servern:

explain analyze select sum(l_quantity) as sum_qty from lineitem where l_shipdate  Gather (cost=1589701.91..1589702.12 rows=2 width=32) (actual time=8553.241..8555.067 rows=3 loops=1)
Workers Planned: 2
Workers Launched: 2
-> Partial Aggregate (cost=1588701.91..1588701.92 rows=1 width=32) (actual time=8547.546..8547.546 rows=1 loops=3)
-> Parallel Seq Scan on lineitem (cost=0.00..1527393.33 rows=24523431 width=5) (actual time=0.038..5998.417 rows=19613238 loops=3)
Filter: (l_shipdate <= '1998-08-18 00:00:00'::timestamp without time zone)
Rows Removed by Filter: 382112
Planning Time: 0.241 ms
Execution Time: 8555.131 ms

Vad händer här? Antalet arbetsprocesser har blivit dubbelt så många, men förfrågan blev bara 1,6599 gånger snabbare. Beräkningarna är intressanta. Vi hade 2 arbetsprocesser och 1 ledare. Efter förändringen blev det 4+1.

Vår maximala hastighetsökning från parallell bearbetning: 5/3 = 1,66(6) gånger.

Hur fungerar det?

Processer

Utförandet av en förfrågan börjar alltid med ledarprocessen. Ledaren gör allt icke-parallellt och en del av den parallella bearbetningen. Andra processer som utför samma förfrågningar kallas arbetsprocesser. Parallell bearbetning använder infrastrukturen av dynamiska bakgrundsarbetare (från version 9.4). Eftersom andra delar av PostgreSQL använder processer istället för trådar, kunde en förfrågan med 3 arbetsprocesser vara 4 gånger snabbare än traditionell bearbetning.

Interaktion

Arbetsprocesserna kommunicerar med ledaren via ett meddelandekö (baserat på delad minne). Varje process har 2 köer: för fel och för tupler.

Hur många arbetsprocesser behövs?

Det minimi begränsning som ställs in av parametern max_parallel_workers_per_gather. Sedan hämtar frågeexecutorn arbetsprocesser från poolen, begränsad av parametern max_parallel_workers size. Den sista begränsningen är max_worker_processes, det vill säga det totala antalet bakgrundsprocesser.

Om det inte går att tilldela en arbetsprocess, kommer bearbetningen att vara enprocessig.

Frågeplaneraren kan minska arbetsprocesser beroende på storleken av tabellen eller indexet. Det finns parametrar för detta: min_parallel_table_scan_size och min_parallel_index_scan_size.

set min_parallel_table_scan_size='8MB'
8MB-tabell => 1 arbetare
24MB-tabell => 2 arbetare
72MB-tabell => 3 arbetare
x => log(x / min_parallel_table_scan_size) / log(3) + 1 arbetare

Varje gång tabellen är tre gånger större än min_parallel_(index|table)_scan_size, lägger Postgres till en arbetsprocess. Antalet arbetsprocesser baseras inte på kostnader. Cykelberoendet gör komplexa implementationer svåra. Istället använder planeraren enkla regler.

I praktiken är dessa regler inte alltid lämpliga för produktion, så man kan ändra antalet arbetsprocesser för en specifik tabell: ALTER TABLE … SET (parallel_workers = N).

Varför används inte parallell bearbetning?

Förutom den långa listan över begränsningar finns det även kostnadskontroller:

parallel_setup_cost — för att undvika parallell bearbetning av korta förfrågningar. Denna parameter uppskattar tiden för att förbereda minnet, starta processen och initialt utbyta data.

parallel_tuple_cost: ledarens kommunikation med arbetarna kan förlängas proportionellt mot antalet tuples från arbetande processer. Denna parameter beräknar kostnaderna för datautbyte.

Nästlad loopsammanfogning — Nested Loop Join

PostgreSQL 9.6+ kan utföra nästlade loopar parallellt — det är en enkel operation.

explain (costs off) select c_custkey, count(o_orderkey)
 from customer left outer join orders on
 c_custkey = o_custkey and o_comment not like '%specialposits%'
 group by c_custkey;
 QUERY PLAN
--------------------------------------------------------------------------------------
 Finalize GroupAggregate
 Group Key: customer.c_custkey
 -> Gather Merge
 Workers Planned: 4
 -> Partial GroupAggregate
 Group Key: customer.c_custkey
 -> Nested Loop Left Join
 -> Parallel Index Only Scan using customer_pkey on customer
 -> Index Scan using idx_orders_custkey on orders
 Index Cond: (customer.c_custkey = o_custkey)
 Filter: ((o_comment)::text !~~ '%specialposits%'::text)

Samlingsprocessen sker i det sista steget, så Nested Loop Left Join är en parallell operation. Parallel Index Only Scan introducerades först i version 10. Det fungerar på samma sätt som ett parallellt sekventiellt scan. Villkoret c_custkey = o_custkey läser en order för varje kundrad. Så det är inte parallellt.

Hash-sammanfogning — Hash Join

Varje arbetsprocess skapar sin egen hash-tabell fram till PostgreSQL 11. Och om det finns fler än fyra sådana processer ökar inte prestandan. I den nya versionen är hash-tabellen gemensam. Varje arbetsprocess kan använda WORK_MEM för att skapa en hash-tabell.

välj
        l_shipmode,
        summa(fall
                när o_orderpriority = '1-URGENT'
                        eller o_orderpriority = '2-HIGH'
                        då 1
                annars 0
        slut) som high_line_count,
        summa(fall
                när o_orderpriority <> '1-URGENT'
                        och o_orderpriority <> '2-HIGH'
                        då 1
                annars 0
        slut) som low_line_count
från
        orders,
        lineitem
var
        o_orderkey = l_orderkey
        och l_shipmode i ('MAIL', 'AIR')
        och l_commitdate < l_receiptdate
        och l_shipdate < l_commitdate
        och l_receiptdate >= datum '1996-01-01'
        och l_receiptdate < datum '1996-01-01' + intervall '1' år
grupp efter
        l_shipmode
ordna efter
        l_shipmode
BEGRÄNSA 1;
                                                                                                                                    QUERY PLAN
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 Begränsning  (kostnad=1964755.66..1964961.44 rader=1 bred=27) (aktuell tid=7579.592..7922.997 rader=1 loop=1)
   ->  Avsluta GroupAggregate  (kostnad=1964755.66..1966196.11 rader=7 bred=27) (aktuell tid=7579.590..7579.591 rader=1 loop=1)
         Gruppnyckel: lineitem.l_shipmode
         ->  Samla Samman  (kostnad=1964755.66..1966195.83 rader=28 bred=27) (aktuell tid=7559.593..7922.319 rader=6 loop=1)
               Arbetare Planerade: 4
               Arbetare Startade: 4
               ->  Delad GroupAggregate  (kostnad=1963755.61..1965192.44 rader=7 bred=27) (aktuell tid=7548.103..7564.592 rader=2 loop=5)
                     Gruppnyckel: lineitem.l_shipmode
                     ->  Sortera  (kostnad=1963755.61..1963935.20 rader=71838 bred=27) (aktuell tid=7530.280..7539.688 rader=62519 loop=5)
                           Sorteringsnyckel: lineitem.l_shipmode
                           Sorteringsmetod: extern sammanfogning  Disk: 2304kB
                           Arbetare 0:  Sorteringsmetod: extern sammanfogning  Disk: 2064kB
                           Arbetare 1:  Sorteringsmetod: extern sammanfogning  Disk: 2384kB
                           Arbetare 2:  Sorteringsmetod: extern sammanfogning  Disk: 2264kB
                           Arbetare 3:  Sorteringsmetod: extern sammanfogning  Disk: 2336kB
                           ->  Parallell Hash Join  (kostnad=382571.01..1957960.99 rader=71838 bred=27) (aktuell tid=7036.917..7499.692 rader=62519 loop=5)
                                 Hash Villkor: (lineitem.l_orderkey = orders.o_orderkey)
                                 ->  Parallell Sekventiell Skanning på lineitem  (kostnad=0.00..1552386.40 rader=71838 bred=19) (aktuell tid=0.583..4901.063 rader=62519 loop=5)
                                       Filter: ((l_shipmode = VALFRI ('{MAIL,AIR}'::bpchar[])) OCH (l_commitdate < l_receiptdate) OCH (l_shipdate < l_commitdate) OCH (l_receiptdate >= '1996-01-01'::datum) OCH (l_receiptdate < '1997-01-01 00:00:00'::tidsstämpel utan tidszon))
                                       Rader Borttagna av Filter: 11934691
                                 ->  Parallell Hash  (kostnad=313722.45..313722.45 rader=3750045 bred=20) (aktuell tid=2011.518..2011.518 rader=3000000 loop=5)
                                       Bucket: 65536  Batch: 256  Minnesbruk: 3840kB
                                       ->  Parallell Sekventiell Skanning på orders  (kostnad=0.00..313722.45 rader=3750045 bred=20) (aktuell tid=0.029..995.948 rader=3000000 loop=5)
 Planeringstid: 0.977 ms
 Utförandeavdelningstid: 7923.770 ms

Fråga 12 från TPC-H visar tydligt en parallell hash-förbindelse. Varje arbetare deltar i att skapa en gemensam hash-tabell.

Sammanläggning — Merge Join

Sammanläggning som inte är parallell till sin natur. Oroa dig inte om detta är det sista steget i förfrågan, den kan fortfarande utföras parallellt.

-- Förfrågan 2 från TPC-H
förklara (kostnader av) välj s_acctbal, s_name, n_name, p_partkey, p_mfgr, s_address, s_phone, s_comment
från part, supplier, partsupp, nation, region
var
		p_partkey = ps_partkey
		och s_suppkey = ps_suppkey
		och p_size = 36
		och p_type som '%BRASS'
		och s_nationkey = n_nationkey
		och n_regionkey = r_regionkey
		och r_name = 'AMERIKA'
		och ps_supplycost = (
			välj
			min(ps_supplycost)
		från partsupp, supplier, nation, region
			var
				p_partkey = ps_partkey
				och s_suppkey = ps_suppkey
				och s_nationkey = n_nationkey
				och n_regionkey = r_regionkey
				och r_name = 'AMERIKA'
		)
ordna efter s_acctbal desc, n_name, s_name, p_partkey
BEGRÄNSA 100;
                                                FRÅGEPLAN
----------------------------------------------------------------------------------------------------------
 Begränsa
   ->  Sortera
         Sorteringsnyckel: supplier.s_acctbal DESC, nation.n_name, supplier.s_name, part.p_partkey
         ->  Sammanfoga
               Sammanfoga villkor: (part.p_partkey = partsupp.ps_partkey)
               Join Filter: (partsupp.ps_supplycost = (SubPlan 1))
               ->  Samla samman
                     Planerade arbetare: 4
                     -&gt;  Parallell indexskanning med hjälp av <strong>part_pkey</strong> på del
                           Filter: (((p_type)::text ~~ '%BRASS'::text) OCH (p_size = 36))
               -&gt;  Materialisera
                     -&gt;  Sortera
                           Sorteringsnyckel: partsupp.ps_partkey
                           -&gt;  Nästlad slinga
                                 -&gt;  Nästlad slinga
                                       Join Filter: (nation.n_regionkey = region.r_regionkey)
                                       -&gt;  Sekventiell skanning på region
                                             Filter: (r_name = 'AMERIKA'::bpchar)
                                       -&gt;  Hash-sammanfogning
                                             Hash-villkor: (supplier.s_nationkey = nation.n_nationkey)
                                             -&gt;  Sekventiell skanning på leverantör
                                             -&gt;  Hash
                                                   -&gt;  Sekventiell skanning på nation
                                 -&gt;  Indexskanning med hjälp av idx_partsupp_suppkey på partsupp
                                       Indexvillkor: (ps_suppkey = supplier.s_suppkey)
               SubPlan 1
                 -&gt;  Aggregat
                       -&gt;  Nästlad slinga
                             Join Filter: (nation_1.n_regionkey = region_1.r_regionkey)
                             -&gt;  Sekventiell skanning på region region_1
                                   Filter: (r_name = 'AMERIKA'::bpchar)
                             -&gt;  Nästlad slinga
                                   -&gt;  Nästlad slinga
                                         -&gt;  Indexskanning med hjälp av idx_partsupp_partkey på partsupp partsupp_1
                                               Indexvillkor: (part.p_partkey = ps_partkey)
                                         -&gt;  Indexskanning med hjälp av supplier_pkey på supplier supplier_1
                                               Indexvillkor: (s_suppkey = partsupp_1.ps_suppkey)
                                   -&gt;  Indexskanning med hjälp av nation_pkey på nation nation_1
                                         Indexvillkor: (n_nationkey = supplier_1.s_nationkey)

Noden «Merge Join» ligger ovanför «Gather Merge». Så sammanläggningen använder inte parallell bearbetning. Men noden «Parallel Index Scan» hjälper ändå till med segmentet. part_pkey.

Sammanläggning efter sektioner

I PostgreSQL 11 sammanläggning efter sektioner är stängd som standard: den har mycket kostsam planering. Tabeller med liknande sektionering kan sammanläggas sektion för sektion. Så kommer Postgres att använda mindre hash-tabeller. Varje sektion kan vara parallell.

tpch=# set enable_partitionwise_join=t;
tpch=# explain (costs off) select * from prt1 t1, prt2 t2
where t1.a = t2.b and t1.b = 0 and t2.b between 0 and 10000;
                    QUERY PLAN
---------------------------------------------------
 Append
   ->  Hash Join
         Hash Cond: (t2.b = t1.a)
         ->  Seq Scan on prt2_p1 t2
               Filter: ((b >= 0) AND (b <= 10000))
         ->  Hash
               ->  Seq Scan on prt1_p1 t1
                     Filter: (b = 0)
   ->  Hash Join
         Hash Cond: (t2_1.b = t1_1.a)
         ->  Seq Scan on prt2_p2 t2_1
               Filter: ((b >= 0) AND (b <= 10000))
         ->  Hash
               ->  Seq Scan on prt1_p2 t1_1
                     Filter: (b = 0)
tpch=# set parallel_setup_cost = 1;
tpch=# set parallel_tuple_cost = 0.01;
tpch=# explain (costs off) select * from prt1 t1, prt2 t2
where t1.a = t2.b and t1.b = 0 and t2.b between 0 and 10000;
                        QUERY PLAN
-----------------------------------------------------------
 Gather
   Workers Planned: 4
   ->  Parallel Append
         ->  Parallel Hash Join
               Hash Cond: (t2_1.b = t1_1.a)
               ->  Parallel Seq Scan on prt2_p2 t2_1
                     Filter: ((b >= 0) AND (b <= 10000))
               ->  Parallel Hash
                     ->  Parallel Seq Scan on prt1_p2 t1_1
                           Filter: (b = 0)
         ->  Parallel Hash Join
               Hash Cond: (t2.b = t1.a)
               ->  Parallel Seq Scan on prt2_p1 t2
                     Filter: ((b >= 0) AND (b <= 10000))
               ->  Parallel Hash
                     ->  Parallel Seq Scan on prt1_p1 t1
                           Filter: (b = 0)

Huvudsaken är att sammanläggning efter sektioner endast kan vara parallell om dessa sektioner är tillräckligt stora.

Parallell bilaga — Parallel Append

Parallel Append kan användas istället för olika block i olika arbetsprocesser. Detta inträffar vanligtvis med UNION ALL-förfrågningar. Nackdelen är mindre parallellism, eftersom varje arbetsprocess bara behandlar 1 förfrågan.

Här körs 2 arbetsprocesser, även om 4 är aktiverade.

tpch=# explain (costs off) select sum(l_quantity) as sum_qty from lineitem where l_shipdate <= date '1998-12-01' - interval '105' day union all select sum(l_quantity) as sum_qty from lineitem where l_shipdate   Parallel Append
         ->  Aggregate
               ->  Seq Scan on lineitem
                     Filter: (l_shipdate   Aggregate
               ->  Seq Scan on lineitem lineitem_1
                     Filter: (l_shipdate <= '1998-08-18 00:00:00'::timestamp without time zone)

De viktigaste variablerna

• WORK_MEM begränsar mängden minne för varje process, inte bara för frågor: work_mem processer anslutningar = väldigt mycket minne.
• max_parallel_workers_per_gather — hur många arbetsprocesser programmet kommer att använda för parallell bearbetning från planen.
• max_worker_processes — justerar det totala antalet arbetsprocesser baserat på antalet CPU-kärnor på servern.
• max_parallel_workers —samma, men för parallella arbetsprocesser.

Sammanfattning

Från och med version 9.6 kan parallell bearbetning avsevärt förbättra prestandan för komplexa frågor som skannar många rader eller index. I PostgreSQL 10 är parallell bearbetning aktiverad som standard. Glöm inte att stänga av den på servrar med hög OLTP-belastning. Sekventiella skanningar eller indexskanningar förbrukar mycket resurser. Om du inte kör en rapport på hela datamängden kan frågor göras mer effektiva genom att bara lägga till saknade index eller använda korrekt partitionering.

Länkar

• https://www.postgresql.org/docs/11/how-parallel-query-works.html
• https://www.postgresql.org/docs/11/parallel-plans.html
• http://ashutoshpg.blogspot.com/2017/12/partition-wise-joins-divide-and-conquer.html
• http://rhaas.blogspot.com/2016/04/postgresql-96-with-parallel-query-vs.html
• http://amitkapila16.blogspot.com/2015/11/parallel-sequential-scans-in-play.html
• https://write-skew.blogspot.com/2018/01/parallel-hash-for-postgresql.html
• http://rhaas.blogspot.com/2017/03/parallel-query-v2.html
• https://blog.2ndquadrant.com/parallel-monster-benchmark/
• https://blog.2ndquadrant.com/parallel-aggregate/
• https://www.depesz.com/2018/02/12/waiting-for-postgresql-11-support-parallel-btree-index-builds/
• Parallellism i PostgreSQL 11

Källa: habr.com