🥇Paralelní dotazy v PostgreSQL

V moderních procesorech je mnoho jader. Aplikace posílaly žádosti do databází paralelně po celá léta. Pokud jde o dotaz na mnoho řádků v tabulce, je rychlejší, když využívá více procesorů, a v PostgreSQL je to možné od verze 9.6.

Trvalo tři roky, než byla implementována funkce paralelních dotazů – bylo nutné přepsat kód v různých fázích provádění dotazů. V PostgreSQL 9.6 byla vytvořena infrastruktura pro další zlepšení kódu. V následujících verzích se zároveň provádí i jiné typy dotazů.

Omezení

Nezapínejte paralelní provádění, pokud jsou všechna jádra již obsazena, jinak budou ostatní dotazy zpomalovat.
Nejdůležitější je, že paralelní zpracování s vysokými hodnotami WORK_MEM využívá hodně paměti – každé hash spojení nebo třídění zabírá paměť ve velikosti work_mem.
OLTP dotazy s nízkou latencí nelze urychlit paralelním prováděním. A pokud dotaz vrací jeden řádek, paralelní zpracování ho pouze zpomalí.
Vývojáři rádi používají benchmark TPC-H. Možná máte podobné dotazy pro ideální paralelní provedení.
Pouze dotazy SELECT bez predikátní blokace se vykonávají paralelně.
Někdy je správná indexace lepší než sekvenční skenování tabulky v paralelním režimu.
Pozastavení dotazů a kurzory nejsou podporovány.
Okna a agregátní funkce uspořádaných sad nejsou paralelní.
V pracovní zátěži s vstupem a výstupem nezískáte nic.
Paralelní algoritmy třídění neexistují. Nicméně, dotazy s tříděním mohou být prováděny paralelně v některých aspektech.
Nahraďte CTE (WITH …) vnořeným SELECT, abyste zahrnuli paralelní zpracování.
Obaly externích dat zatím nepodporují paralelní zpracování (a mohly by!).
FULL OUTER JOIN není podporován.
max_rows deaktivuje paralelní zpracování.
Pokud je v dotazu funkce, která není označena jako PARALLEL SAFE, bude jednovláknový.
Úroveň izolace transakce SERIALIZABLE deaktivuje paralelní zpracování.

Testovací prostředí

Vývojáři PostgreSQL se pokusili zkrátit dobu odezvy dotazů benchmarku TPC-H. Stáhněte benchmark a přizpůsobte ho PostgreSQL. Toto je neoficiální použití benchmarku TPC-H — ne pro porovnání databází nebo hardwaru.

Stáhněte si TPC-H_Tools_v2.17.3.zip (nebo novější verzi) z oficiálních stránek TPC.
Přejmenujte makefile.suite na Makefile a změňte ho, jak je popsáno zde: https://github.com/tvondra/pg_tpch . Zkompilujte kód příkazem make.
Vygenerujte data: .\/dbgen -s 10 vytvoří databázi o velikosti 23 GB. To je dostatečné pro pozorování rozdílu ve výkonu paralelních a nepř Paralelní dotazů.
Převeďte soubory tbl v csv s for a sed.
Naklonujte repozitář pg_tpch a zkopírujte soubory csv v pg_tpch\/dss\/data.
Vytvořte dotazy příkazem qgen.
Načtěte data do databáze příkazem .\/tpch.sh.

Paralelní sekvenční skenování

Může být rychlejší ne díky paralelnímu čtení, ale protože jsou data rozložena na více jádrech CPU. V moderních OS jsou soubory dat PostgreSQL dobře ukládány do mezipaměti. Předběžným čtením lze získat z úložiště blok větší, než kolik požaduje démon PG. Proto výkon dotazu není omezen vstupy/výstupy disku. Spotřebovává cykly CPU, aby:

četl řádky po jedné ze stránek tabulky;
porovnával hodnoty řádků a podmínky není spolehlivý:.

Proveďme jednoduchý dotaz 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

Sekvenční skenování dává příliš mnoho řádků bez agregace, takže dotaz je prováděn jedním jádrem CPU.

Pokud přidáme SUM(), je vidět, že dva pracovní procesy pomohou urychlit dotaz:

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

Paralelní agregace

Uzlová 'Parallel Seq Scan' produkuje řádky pro částečnou agregaci. Uzlová 'Partial Aggregate' zkracuje tyto řádky pomocí SUM(). Na konci je čítač SUM z každého pracovního procesu shromážděn uzlem 'Gather'.

Konečný výsledek se počítá pomocí uzlu „Finalize Aggregate“. Pokud máte vlastní agregační funkce, nezapomeňte je označit jako „parallel safe“.

Počet pracovních procesů

Počet pracovních procesů lze zvýšit bez restartování serveru:

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

Co se zde děje? Počet pracovních procesů se zdvojnásobil a dotaz se zrychlil pouze 1,6599krát. Výpočty jsou zajímavé. Měli jsme 2 pracovní procesy a 1 vedoucí. Po změně máme 4+1.

Naše maximální zrychlení díky paralelnímu zpracování: 5/3 = 1,66(6)krát.

Jak to funguje?

Procesy

Zpracování dotazu vždy začíná vedoucím procesem. Vedoucí provádí veškeré nepohodlné a část paralelního zpracování. Ostatní procesy, které vykonávají stejné dotazy, se nazývají pracovními procesy. Paralelní zpracování využívá infrastrukturu dynamických pozadí pracovních procesů (od verze 9.4). Jelikož jiné části PostgreSQL používají procesy, a nikoli vlákna, dotaz se 3 pracovními procesy mohl být 4krát rychlejší než tradiční zpracování.

Interakce

Pracovní procesy komunikují s vedoucím pomocí fronty zpráv (na základě sdílené paměti). Každý proces má 2 fronty: na chyby a na tuplíky.

Kolik pracovních procesů je potřeba?

Minimální limit určuje parametr max_parallel_workers_per_gather. Pak prováděcí proces dotazů bere pracovní procesy z bazénu, který je omezen parametrem max_parallel_workers size. Poslední limit je max_worker_processes, tedy celkový počet pozadí procesů.

Pokud se pracovní proces nepodaří přidělit, zpracování bude jednoprocesní.

Plánovač dotazů může snížit pracovní procesy na základě velikosti tabulky nebo indexu. K tomu existují parametry min_parallel_table_scan_size a min_parallel_index_scan_size.

set min_parallel_table_scan_size='8MB'
8MB tabulka => 1 pracovník
24MB tabulka => 2 pracovníci
72MB tabulka => 3 pracovníci
x => log(x / min_parallel_table_scan_size) / log(3) + 1 pracovník

Každýkrát, když je tabulka 3krát větší než min_parallel_(index|table)_scan_size, Postgres přidává pracovní proces. Počet pracovních procesů není založen na nákladech. Cyklová závislost ztěžuje složité implementace. Místo toho plánovač používá jednoduchá pravidla.

V praxi tato pravidla nevždy vyhovují produkčnímu prostředí, takže je možné změnit počet pracovních procesů pro konkrétní tabulku: ALTER TABLE … SET (parallel_workers = N).

Proč se paralelní zpracování nepoužívá?

Kromě dlouhého seznamu omezení existují také náklady na kontrolu:

parallel_setup_cost — aby se předešlo paralelnímu zpracování krátkých dotazů. Tento parametr odhaduje čas potřebný na přípravu paměti, spuštění procesu a počáteční výměnu dat.

parallel_tuple_cost: komunikace vedoucího s pracovními procesy se může protáhnout v poměru k počtu tuple od pracovních procesů. Tento parametr počítá náklady na výměnu dat.

Spojení vnořených cyklů — Nested Loop Join

PostgreSQL 9.6+ může vykonávat vnořené cykly paralelně — je to jednoduchá operace.

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)

Sběr probíhá na posledním kroku, takže Nested Loop Left Join — to je paralelní operace. Paralelní Index Only Scan byl zaveden až ve verzi 10. Funguje podobně jako paralelní sekvenční skenování. Podmínka c_custkey = o_custkey čte jeden příkaz pro každý zákaznický řádek. Takže není paralelní.

Hash Join — Hash Join

Každý pracovní proces vytváří svou vlastní hash tabulku do PostgreSQL 11. A pokud je těchto procesů více než čtyři, výkon se nezvýší. V nové verzi je hash tabulka společná. Každý pracovní proces může použít WORK_MEM k vytvoření hash tabulky.

vyberte
        l_shipmode,
        součet(případ
                když o_orderpriority = '1-URGENT'
                        nebo o_orderpriority = '2-HIGH'
                        pak 1
                jinak 0
        end) jako high_line_count,
        součet(případ
                když o_orderpriority  '1-URGENT'
                        a o_orderpriority  '2-HIGH'
                        pak 1
                jinak 0
        end) jako low_line_count
z
        orders,
        lineitem
kde
        o_orderkey = l_orderkey
        a l_shipmode v ('MAIL', 'AIR')
        a l_commitdate < l_receiptdate
        a l_shipdate = datum '1996-01-01'
        a l_receiptdate < datum '1996-01-01' + interval '1' rok
grupovat podle
        l_shipmode
řadit podle
        l_shipmode
OMEZENÍ 1;
                                                                                                                                    PLÁN DOTAZU
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 OMEZENÍ  (cena=1964755.66..1964961.44 řádků=1 šířka=27) (skutečný čas=7579.592..7922.997 řádků=1 smyčky=1)
   ->  Závěrečná skupinová agregace  (cena=1964755.66..1966196.11 řádků=7 šířka=27) (skutečný čas=7579.590..7579.591 řádků=1 smyčky=1)
         Klíč skupiny: lineitem.l_shipmode
         ->  Shromáždění spojení  (cena=1964755.66..1966195.83 řádků=28 šířka=27) (skutečný čas=7559.593..7922.319 řádků=6 smyčky=1)
               Plánované pracovníky: 4
               Spuštěné pracovníky: 4
               ->  Částečná skupinová agregace  (cena=1963755.61..1965192.44 řádků=7 šířka=27) (skutečný čas=7548.103..7564.592 řádků=2 smyčky=5)
                     Klíč skupiny: lineitem.l_shipmode
                     ->  Řazení  (cena=1963755.61..1963935.20 řádků=71838 šířka=27) (skutečný čas=7530.280..7539.688 řádků=62519 smyčky=5)
                           Klíč pro řazení: lineitem.l_shipmode
                           Metoda řazení: externí sloučení  Disk: 2304kB
                           Pracovník 0:  Metoda řazení: externí sloučení  Disk: 2064kB
                           Pracovník 1:  Metoda řazení: externí sloučení  Disk: 2384kB
                           Pracovník 2:  Metoda řazení: externí sloučení  Disk: 2264kB
                           Pracovník 3:  Metoda řazení: externí sloučení  Disk: 2336kB
                           ->  Paralelní hash spojení  (cena=382571.01..1957960.99 řádků=71838 šířka=27) (skutečný čas=7036.917..7499.692 řádků=62519 smyčky=5)
                                 Podmínka hash: (lineitem.l_orderkey = orders.o_orderkey)
                                 ->  Paralelní sekvenční skenování na lineitem  (cena=0.00..1552386.40 řádků=71838 šířka=19) (skutečný čas=0.583..4901.063 řádků=62519 smyčky=5)
                                       Filtr: ((l_shipmode = JAKÝKOLI ('{MAIL,AIR}'::bpchar[])) A (l_commitdate < l_receiptdate) A (l_shipdate = '1996-01-01'::datum) A (l_receiptdate < '1997-01-01 00:00:00'::čas bez časového pásma))
                                       Odebrané řádky filtrem: 11934691
                                 ->  Paralelní hash  (cena=313722.45..313722.45 řádků=3750045 šířka=20) (skutečný čas=2011.518..2011.518 řádků=3000000 smyčky=5)
                                       Koše: 65536  Dávky: 256  Použití paměti: 3840kB
                                       ->  Paralelní sekvenční skenování na orders  (cena=0.00..313722.45 řádků=3750045 šířka=20) (skutečný čas=0.029..995.948 řádků=3000000 smyčky=5)
 Čas plánování: 0.977 ms
 Čas provádění: 7923.770 ms

Dotaz 12 z TPC-H názorně ukazuje paralelní hash spojení. Každý pracovní proces se podílí na vytváření společné hash tabulky.

Sloučení — Merge Join

Sloučení je z pohledu paralelnosti nepárné. Nebojte se, pokud je to poslední fáze dotazu, stále může být vykonávána paralelně.

-- Dotaz 2 z TPC-H
vysvětlit (náklady vypnuto) vybrat s_acctbal, s_name, n_name, p_partkey, p_mfgr, s_address, s_phone, s_comment
z     part, supplier, partsupp, nation, region
kde
        p_partkey = ps_partkey
        a s_suppkey = ps_suppkey
        a p_size = 36
        a p_type jako '%BRASS'
        a s_nationkey = n_nationkey
        a n_regionkey = r_regionkey
        a r_name = 'AMERIKA'
        a ps_supplycost = (
                vybrat
                        min(ps_supplycost)
                z     partsupp, supplier, nation, region
                kde
                        p_partkey = ps_partkey
                        a s_suppkey = ps_suppkey
                        a s_nationkey = n_nationkey
                        a n_regionkey = r_regionkey
                        a r_name = 'AMERIKA'
        )
řadit podle s_acctbal sestupně, n_name, s_name, p_partkey
OMEZIT 100;
                                                PLÁN DOTAZU
----------------------------------------------------------------------------------------------------------
 OMEZIT
   -&gt;  Řadit
         Klíč pro řazení: supplier.s_acctbal DESC, nation.n_name, supplier.s_name, part.p_partkey
         -&gt;  Sloučit spojení
               Podmínka sloučení: (part.p_partkey = partsupp.ps_partkey)
               Filtr spojení: (partsupp.ps_supplycost = (Subplán 1))
               -&gt;  Sbrat s pomocí shromáždění
                     Plánovaní pracovníci: 4
                     -&gt;  Paralelní indexové skenování pomocí <strong>part_pkey</strong> na part
                           Filtr: (((p_type)::text ~~ '%BRASS'::text) A (p_size = 36))
               -&gt;  Materializovat
                     -&gt;  Řadit
                           Klíč pro řazení: partsupp.ps_partkey
                           -&gt;  Vnořená smyčka
                                 -&gt;  Vnořená smyčka
                                       Filtr spojení: (nation.n_regionkey = region.r_regionkey)
                                       -&gt;  Sekvenční skenování na region
                                             Filtr: (r_name = 'AMERIKA'::bpchar)
                                       -&gt;  Hashové spojení
                                             Podmínka hashe: (supplier.s_nationkey = nation.n_nationkey)
                                             -&gt;  Sekvenční skenování na supplier
                                             -&gt;  Hashe
                                                   -&gt;  Sekvenční skenování na nation
                                 -&gt;  Indexové skenování pomocí idx_partsupp_suppkey na partsupp
                                       Podmínka indexu: (ps_suppkey = supplier.s_suppkey)
               Subplán 1
                 -&gt;  Agregace
                       -&gt;  Vnořená smyčka
                             Filtr spojení: (nation_1.n_regionkey = region_1.r_regionkey)
                             -&gt;  Sekvenční skenování na region region_1
                                   Filtr: (r_name = 'AMERIKA'::bpchar)
                             -&gt;  Vnořená smyčka
                                   -&gt;  Vnořená smyčka
                                         -&gt;  Indexové skenování pomocí idx_partsupp_partkey na partsupp partsupp_1
                                               Podmínka indexu: (part.p_partkey = ps_partkey)
                                         -&gt;  Indexové skenování pomocí supplier_pkey na supplier supplier_1
                                               Podmínka indexu: (s_suppkey = partsupp_1.ps_suppkey)
                                   -&gt;  Indexové skenování pomocí nation_pkey na nation nation_1
                                         Podmínka indexu: (n_nationkey = supplier_1.s_nationkey)

Uzlová struktura „Merge Join“ se nachází nad „Gather Merge“. Proto sloučení nepoužívá paralelní zpracování. Nicméně uzel „Parallel Index Scan“ stále pomáhá s segmentem part_pkey.

Sloučení podle oddílů

V PostgreSQL 11 sloučení podle oddílů je ve výchozím nastavení vypnuto: má velmi nákladné plánování. Tabulky se podobným dělením lze sloučit oddíl po oddílu. Takže Postgres využije menší hash tabulky. Každé sloučení oddílů může být paralelní.

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;
                    PLÁN DOTAZU
---------------------------------------------------
 Připojit
   ->  Hash Join
         Hash Podmínka: (t2.b = t1.a)
         ->  Seq Scan na prt2_p1 t2
               Filtrovat: ((b >= 0) AND (b <= 10000))
         ->  Hash
               ->  Seq Scan na prt1_p1 t1
                     Filtrovat: (b = 0)
   ->  Hash Join
         Hash Podmínka: (t2_1.b = t1_1.a)
         ->  Seq Scan na prt2_p2 t2_1
               Filtrovat: ((b >= 0) AND (b <= 10000))
         ->  Hash
               ->  Seq Scan na prt1_p2 t1_1
                     Filtrovat: (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;
                        PLÁN DOTAZU
-----------------------------------------------------------
 Shromáždit
   Plánovaní pracovníci: 4
   ->  Paralelní Připojit
         ->  Paralelní Hash Join
               Hash Podmínka: (t2_1.b = t1_1.a)
               ->  Paralelní Seq Scan na prt2_p2 t2_1
                     Filtrovat: ((b >= 0) AND (b <= 10000))
               ->  Paralelní Hash
                     ->  Paralelní Seq Scan na prt1_p2 t1_1
                           Filtrovat: (b = 0)
         ->  Paralelní Hash Join
               Hash Podmínka: (t2.b = t1.a)
               ->  Paralelní Seq Scan na prt2_p1 t2
                     Filtrovat: ((b >= 0) AND (b <= 10000))
               ->  Paralelní Hash
                     ->  Paralelní Seq Scan na prt1_p1 t1
                           Filtrovat: (b = 0)

Hlavní věc je, že sloučení podle oddílů je možné provádět paralelně, pouze pokud jsou tyto oddíly dostatečně velké.

Paralelní přidání — Parallel Append

Parallel Append může být použito místo různých bloků v různých pracovních procesech. Obvykle se vyskytuje u dotazů UNION ALL. Nevýhodou je méně paralelismu, protože každý pracovní proces zpracovává pouze 1 dotaz.

Zde byly spuštěny 2 pracovní procesy, přestože bylo zapnuto 4.

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)

Nejdůležitější proměnné

WORK_MEM omezuje množství paměti pro každý proces, nejen pro dotazy: work_mem procesy připojení = velmi hodně paměti.
max_parallel_workers_per_gather — kolik pracovních procesů bude program používat pro paralelní zpracování podle plánu.
max_worker_processes — přizpůsobuje celkový počet pracovních procesů počtu jader CPU na serveru.
max_parallel_workers — totéž, ale pro paralelní pracovní procesy.

Závěry

Od verze 9.6 může paralelní zpracování výrazně zlepšit výkon složitých dotazů, které skenují mnoho řádků nebo indexů. V PostgreSQL 10 je paralelní zpracování výchozí. Nezapomeňte jej vypnout na serverech s vysokou zátěží OLTP. Sekvenční skeny nebo skeny indexů spotřebovávají velmi mnoho prostředků. Pokud nevytváříte zprávu pro celý datový soubor, dotazy lze učinit efektivnějšími jednoduše přidáním chybějících indexů nebo použitím správné partitionace.