În procesoarele moderne există foarte multe nuclee. De ani de zile, aplicațiile trimiteau cereri în baze de date în paralel. Dacă este o cerere de raportare pentru un număr mare de rânduri dintr-o tabelă, aceasta se execută mai repede atunci când utilizează mai multe procesoare, iar în PostgreSQL acest lucru este posibil începând cu versiunea 9.6.
Au fost necesari 3 ani pentru a implementa funcția cererilor paralele — a fost necesar să rescriem codul în diferite etape ale execuției cererilor. În PostgreSQL 9.6 a fost introdusă infrastructura pentru îmbunătățirea ulterioară a codului. În versiunile ulterioare, și alte tipuri de cereri sunt executate în paralel.
Limitări
- Nu activați execuția paralelă dacă toate nucleele sunt deja ocupate, altfel alte cereri vor fi încetinite.
- Cel mai important, procesarea paralelă cu valori mari WORK_MEM folosește multă memorie — fiecare conexiune hash sau sortare ocupă memorie în volum de work_mem.
- Cererile OLTP cu latență mică nu pot fi accelerate prin execuția paralelă. Iar dacă cererea returnează un singur rând, procesarea paralelă o va încetini.
- Dezvoltatorii iubesc să folosească benchmark-ul TPC-H. Poate că aveți cereri asemănătoare pentru o execuție paralelă ideală.
- Numai cererile SELECT fără blocare prin predicat sunt executate în paralel.
- Uneori, indexarea corectă este mai bună decât scanarea secvențială a tabelului în modul paralel.
- Suspendarea cererilor și cursurile nu sunt suportate.
- Funcțiile fereastră și funcțiile agregate ale seturilor ordonate nu sunt paralele.
- Nu obțineți nimic în sarcina de lucru de intrare-ieșire.
- Nu există algoritmi de sortare paralele. Dar cererile cu sortări pot fi executate în paralel în anumite aspecte.
- Înlocuiți CTE (WITH …) cu un SELECT încorporat pentru a activa procesarea paralelă.
- Înfășurăturile de date externe nu suportă încă procesarea paralelă (dar ar putea!)
- FULL OUTER JOIN nu este suportat.
- max_rows dezactivează procesarea paralelă.
- Dacă în cerere există o funcție care nu este marcată ca PARALLEL SAFE, aceasta va fi unică.
- Nivelul de izolare al tranzacției SERIALIZABLE dezactivează procesarea paralelă.
Mediu de testare
Dezvoltatorii PostgreSQL au încercat să reducă timpul de răspuns al cererilor benchmark-ului TPC-H. Descărcați benchmark-ul și . Aceasta este o utilizare neoficială a benchmark-ului TPC-H — nu pentru compararea bazelor de date sau a echipamentelor.
- Descărcați TPC-H_Tools_v2.17.3.zip (sau o versiune mai recentă) .
- Renumiți makefile.suite în Makefile și modificați-l conform instrucțiunilor de aici: . Compilați codul cu comanda make.
- Generați datele:
.\/dbgen -s 10creează o bază de date de 23 GB. Acest lucru este suficient pentru a observa diferența de performanță între interogările paralele și cele neparalele. - Conversați fișierele
tblîncsv cu forșised. - Clonare repository
pg_tpchși copiați fișierelecsvînpg_tpch\/dss\/data. - Creați interogările cu comanda
qgen. - Încărcați datele în bază cu comanda
.\/tpch.sh.
Scanare secvențială paralelă
Aceasta poate fi mai rapidă nu din cauza citirii paralele, ci pentru că datele sunt dispersate pe multe nuclee ale CPU-ului. În sistemele de operare moderne, fișierele de date PostgreSQL sunt bine cache-uie. Cu citirea anticipată, se pot obține blocuri mai mari din magazin decât cererea demonului PG. Prin urmare, performanța interogării nu este limitată de intrările și ieșirile discului. Consuma cicluri CPU pentru a:
- citi rânduri câte unul de pe paginile tabelului;
- compara valorile rândurilor și condițiile
WHERE.
Să executăm o interogare simplă select:
tpch=# explain analyze select l_quantity as sum_qty from lineitem where l_shipdate <= date '1998-12-01' - interval '105' day;
PLANUL INTEROGĂRII
--------------------------------------------------------------------------------------------------------------------------
Scanare Secvențială pe lineitem (cost=0.00..1964772.00 rows=58856235 width=5) (timp efectiv=0.014..16951.669 rows=58839715 loops=1)
Filtru: (l_shipdate <= '1998-08-18 00:00:00'::timestamp fără fus orar)
Rânduri eliminate de Filtru: 1146337
Timp de planificare: 0.203 ms
Timp de execuție: 19035.100 msScanarea secvențială oferă prea multe rânduri fără agregare, astfel încât interogarea este executată pe un singur nucleu CPU.
Dacă adăugăm SUM(), se observă că două procese de lucru ajută la accelerarea interogării:
explain analyze select sum(l_quantity) as sum_qty from lineitem where l_shipdate <= date '1998-12-01' - interval '105' day;
PLANUL INTEROGĂRII
----------------------------------------------------------------------------------------------------------------------------------------------------
Finalizează Agregarea (cost=1589702.14..1589702.15 rows=1 width=32) (timp efectiv=8553.365..8553.365 rows=1 loops=1)
-> Adună (cost=1589701.91..1589702.12 rows=2 width=32) (timp efectiv=8553.241..8555.067 rows=3 loops=1)
Lucrători Planificați: 2
Lucrători Lansati: 2
-> Agregare Parțială (cost=1588701.91..1588701.92 rows=1 width=32) (timp efectiv=8547.546..8547.546 rows=1 loops=3)
-> Scanare Secvențială Paralelă pe lineitem (cost=0.00..1527393.33 rows=24523431 width=5) (timp efectiv=0.038..5998.417 rows=19613238 loops=3)
Filtru: (l_shipdate <= '1998-08-18 00:00:00'::timestamp fără fus orar)
Rânduri eliminate de Filtru: 382112
Timp de planificare: 0.241 ms
Timp de execuție: 8555.131 msAgregare paralelă
Nodul „Scanare Secvențială Paralelă” generează rânduri pentru agregarea parțială. Nodul „Agregare Parțială” taie aceste rânduri folosind SUM(). La final, contorul SUM din fiecare proces de lucru este adunat de nodul „Adună”.
Rezultatul final este calculat de nodul „Finalize Aggregate”. Dacă aveți funcții de agregare proprii, nu uitați să le marcați ca „parallel safe”.
Numărul de procese de lucru
Numărul de procese de lucru poate fi crescut fără a reporni serverul:
explain analyze select sum(l_quantity) as sum_qty from lineitem where l_shipdate <= date '1998-12-01' - interval '105' day;
PLANUL INTEROGĂRII
----------------------------------------------------------------------------------------------------------------------------------------------------
Finalizează Agregarea (cost=1589702.14..1589702.15 rows=1 width=32) (timp efectiv=8553.365..8553.365 rows=1 loops=1)
-> Adună (cost=1589701.91..1589702.12 rows=2 width=32) (timp efectiv=8553.241..8555.067 rows=3 loops=1)
Lucrători Planificați: 2
Lucrători Lansati: 2
-> Agregare Parțială (cost=1588701.91..1588701.92 rows=1 width=32) (timp efectiv=8547.546..8547.546 rows=1 loops=3)
-> Scanare Secvențială Paralelă pe lineitem (cost=0.00..1527393.33 rows=24523431 width=5) (timp efectiv=0.038..5998.417 rows=19613238 loops=3)
Filtru: (l_shipdate <= '1998-08-18 00:00:00'::timestamp fără fus orar)
Rânduri eliminate de Filtru: 382112
Timp de planificare: 0.241 ms
Timp de execuție: 8555.131 msCe se întâmplă aici? Numărul de procese de lucru s-a dublat, iar cererea a devenit de doar 1,6599 ori mai rapidă. Calculul este interesant. Aveam 2 procese de lucru și 1 lider. După modificare, avem 4+1.
Accelerația maximă obținută prin procesare paralelă este: 5/3 = 1,66(6) ori.
Cum funcționează?
Procese
Executarea cererii începe întotdeauna cu procesul de conducere. Liderul se ocupă de tot ce este nepătruns și de o parte din procesarea paralelă. Altele procese care execută aceleași cereri sunt numite procese de lucru. Procesarea paralelă folosește infrastructura (începând cu versiunea 9.4). Deoarece alte părți ale PostgreSQL folosesc procese, nu fire, o cerere cu 3 procese de lucru ar putea fi cu 4 ori mai rapidă decât procesarea tradițională.
Interacțiune
Procesele de lucru comunică cu liderul printr-o coadă de mesaje (bazată pe memorie comună). Fiecare proces are 2 cozi: pentru erori și pentru tupluri.
Câte procese de lucru sunt necesare?
Limitarea minimă este stabilită de parametrul . Apoi, executorul cererii ia procese de lucru din pool-ul limitat de parametrul . Ultima limitare este , adică numărul total de procese de fundal.
Dacă nu s-a reușit alocarea unui proces de lucru, procesarea va fi uniprotocarea.
Planificatorul de cereri poate reduce procesele de lucru în funcție de dimensiunea tabelei sau a indexului. Pentru aceasta, există parametrii și .
set min_parallel_table_scan_size='8MB'
8MB tabel => 1 muncitor
24MB tabel => 2 muncitori
72MB tabel => 3 muncitori
x => log(x / min_parallel_table_scan_size) / log(3) + 1 muncitorDe fiecare dată când tabela este de 3 ori mai mare decât min_parallel_(index|table)_scan_size, Postgres adaugă un proces de lucru. Numărul de procese de lucru nu se bazează pe costuri. Dependența circulară îngreunează implementările complexe. În schimb, planificatorul folosește reguli simple.
În practică, aceste reguli nu sunt întotdeauna adecvate pentru producție, așa că se poate modifica numărul de procese de lucru pentru o tabelă specifică: ALTER TABLE … SET (parallel_workers = N).
De ce nu se utilizează procesarea paralelă?
Pe lângă lunga listă de restricții, există și verificări ale costurilor:
— pentru a evita procesarea paralelă a cererilor scurte. Acest parametru estimează timpul pentru pregătirea memoriei, pornirea procesului și schimbul inițial de date.
: comunicarea liderului cu lucrătorii poate dura proporțional cu numărul de tuple-uri din procesele de lucru. Acest parametru calculează costurile schimbului de date.
Îmbinările ciclice înfășurate — Nested Loop Join
PostgreSQL 9.6+ poate executa bucle înlănțuite în paralel — aceasta este o operațiune simplă.
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)Colectarea se face în ultima etapă, astfel că Nested Loop Left Join — este o operație paralelă. Parallel Index Only Scan a apărut doar în versiunea 10. Funcționează similar cu scanarea secvențială paralelă. Condiția c_custkey = o_custkey citește o comandă pentru fiecare linie a clientului. Așadar, nu este paralel.
Îmbinarea prin hashtable — Hash Join
Fiecare proces de lucru își creează propria tabelă hash până la PostgreSQL 11. Și dacă procesele sunt mai mult de patru, performanța nu va crește. În noua versiune, tabela hash este comună. Fiecare proces de lucru poate utiliza WORK_MEM pentru a crea o tabelă hash.
select
l_shipmode,
sum(case
when o_orderpriority = '1-URGENT'
or o_orderpriority = '2-HIGH'
then 1
else 0
end) as high_line_count,
sum(case
when o_orderpriority '1-URGENT'
and o_orderpriority '2-HIGH'
then 1
else 0
end) as low_line_count
from
orders,
lineitem
where
o_orderkey = l_orderkey
and l_shipmode in ('MAIL', 'AIR')
and l_commitdate < l_receiptdate
and l_shipdate = date '1996-01-01'
and l_receiptdate Finalize GroupAggregate (cost=1964755.66..1966196.11 rows=7 width=27) (actual time=7579.590..7579.591 rows=1 loops=1)
Group Key: lineitem.l_shipmode
-> Gather Merge (cost=1964755.66..1966195.83 rows=28 width=27) (actual time=7559.593..7922.319 rows=6 loops=1)
Workers Planned: 4
Workers Launched: 4
-> Partial GroupAggregate (cost=1963755.61..1965192.44 rows=7 width=27) (actual time=7548.103..7564.592 rows=2 loops=5)
Group Key: lineitem.l_shipmode
-> Sort (cost=1963755.61..1963935.20 rows=71838 width=27) (actual time=7530.280..7539.688 rows=62519 loops=5)
Sort Key: lineitem.l_shipmode
Sort Method: external merge Disk: 2304kB
Worker 0: Sort Method: external merge Disk: 2064kB
Worker 1: Sort Method: external merge Disk: 2384kB
Worker 2: Sort Method: external merge Disk: 2264kB
Worker 3: Sort Method: external merge Disk: 2336kB
-> Parallel Hash Join (cost=382571.01..1957960.99 rows=71838 width=27) (actual time=7036.917..7499.692 rows=62519 loops=5)
Hash Cond: (lineitem.l_orderkey = orders.o_orderkey)
-> Parallel Seq Scan on lineitem (cost=0.00..1552386.40 rows=71838 width=19) (actual time=0.583..4901.063 rows=62519 loops=5)
Filter: ((l_shipmode = ANY ('{MAIL,AIR}'::bpchar[])) AND (l_commitdate < l_receiptdate) AND (l_shipdate = '1996-01-01'::date) AND (l_receiptdate Parallel Hash (cost=313722.45..313722.45 rows=3750045 width=20) (actual time=2011.518..2011.518 rows=3000000 loops=5)
Buckets: 65536 Batches: 256 Memory Usage: 3840kB
-> Parallel Seq Scan on orders (cost=0.00..313722.45 rows=3750045 width=20) (actual time=0.029..995.948 rows=3000000 loops=5)
Planning Time: 0.977 ms
Execution Time: 7923.770 msInterogarea 12 din TPC-H arată clar un join hash paralel. Fiecare proces lucrează la crearea unei tabele hash comune.
Îmbinarea — Merge Join
Îmbinarea este prin natura sa non-paralelă. Nu vă faceți griji dacă aceasta este ultima etapă a interogării — poate totuși fi executată în paralel.
-- Interogare 2 din TPC-H
explain (costuri dezactivate) select s_acctbal, s_name, n_name, p_partkey, p_mfgr, s_address, s_phone, s_comment
from part, supplier, partsupp, nation, region
where
p_partkey = ps_partkey
and s_suppkey = ps_suppkey
and p_size = 36
and p_type like '%BRASS'
and s_nationkey = n_nationkey
and n_regionkey = r_regionkey
and r_name = 'AMERICA'
and ps_supplycost = (
select
min(ps_supplycost)
from partsupp, supplier, nation, region
where
p_partkey = ps_partkey
and s_suppkey = ps_suppkey
and s_nationkey = n_nationkey
and n_regionkey = r_regionkey
and r_name = 'AMERICA'
)
order by s_acctbal desc, n_name, s_name, p_partkey
LIMIT 100;
PLAN DE INTEROGARE
----------------------------------------------------------------------------------------------------------
Limit
-> Sort
Cheie Sortare: supplier.s_acctbal DESC, nation.n_name, supplier.s_name, part.p_partkey
-> Merge Join
Condiție de îmbinare: (part.p_partkey = partsupp.ps_partkey)
Filtru de îmbinare: (partsupp.ps_supplycost = (SubPlan 1))
-> Gather Merge
Lucrători planificați: 4
-> Scan de Indice Paralel folosind <strong>part_pkey</strong> on part
Filtru: (((p_type)::text ~~ '%BRASS'::text) ȘI (p_size = 36))
-> Materialize
-> Sort
Cheie Sortare: partsupp.ps_partkey
-> Nested Loop
-> Nested Loop
Filtru de îmbinare: (nation.n_regionkey = region.r_regionkey)
-> Scan Secvențial pe region
Filtru: (r_name = 'AMERICA'::bpchar)
-> Hash Join
Condiție Hash: (supplier.s_nationkey = nation.n_nationkey)
-> Scan Secvențial pe supplier
-> Hash
-> Scan Secvențial pe nation
-> Scan de Indice folosind idx_partsupp_suppkey pe partsupp
Condiție de Indice: (ps_suppkey = supplier.s_suppkey)
SubPlan 1
-> Aggregate
-> Nested Loop
Filtru de îmbinare: (nation_1.n_regionkey = region_1.r_regionkey)
-> Scan Secvențial pe region region_1
Filtru: (r_name = 'AMERICA'::bpchar)
-> Nested Loop
-> Nested Loop
-> Scan de Indice folosind idx_partsupp_partkey pe partsupp partsupp_1
Condiție de Indice: (part.p_partkey = ps_partkey)
-> Scan de Indice folosind supplier_pkey pe supplier supplier_1
Condiție de Indice: (s_suppkey = partsupp_1.ps_suppkey)
-> Scan de Indice folosind nation_pkey pe nation nation_1
Condiție de Indice: (n_nationkey = supplier_1.s_nationkey)Nodul „Merge Join” se află deasupra „Gather Merge”. Astfel, îmbinarea nu folosește procesare paralelă. Totuși, nodul „Parallel Index Scan” ajută în continuare cu segmentul part_pkey.
Îmbinarea pe secțiuni
În PostgreSQL 11 dezactivat implicit: are un plan foarte costisitor. Tabele cu secționare similare pot fi îmbinate secțiune cu secțiune. Astfel, Postgres va utiliza table hash mai mici. Fiecare îmbinare de secțiuni poate fi paralelă.
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;
PLAN INTEROGARE
---------------------------------------------------
Append
-> Hash Join
Condiție Hash: (t2.b = t1.a)
-> Scanare Secvențială pe prt2_p1 t2
Filtru: ((b >= 0) ȘI (b <= 10000))
-> Hash
-> Scanare Secvențială pe prt1_p1 t1
Filtru: (b = 0)
-> Hash Join
Condiție Hash: (t2_1.b = t1_1.a)
-> Scanare Secvențială pe prt2_p2 t2_1
Filtru: ((b >= 0) ȘI (b <= 10000))
-> Hash
-> Scanare Secvențială pe prt1_p2 t1_1
Filtru: (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;
PLAN INTEROGARE
-----------------------------------------------------------
Gather
Lucrători Planificați: 4
-> Îmbinare paralelă
-> Îmbinare Hash Paralelă
Condiție Hash: (t2_1.b = t1_1.a)
-> Scanare Secvențială Paralelă pe prt2_p2 t2_1
Filtru: ((b >= 0) ȘI (b <= 10000))
-> Hash
-> Scanare Secvențială Paralelă pe prt1_p2 t1_1
Filtru: (b = 0)
-> Îmbinare Hash Paralelă
Condiție Hash: (t2.b = t1.a)
-> Scanare Secvențială Paralelă pe prt2_p1 t2
Filtru: ((b >= 0) ȘI (b <= 10000))
-> Hash
-> Scanare Secvențială Paralelă pe prt1_p1 t1
Filtru: (b = 0)Principalul aspect, îmbinarea pe secțiuni este paralelă, doar dacă acele secțiuni sunt suficient de mari.
Adăugare paralelă — Parallel Append
poate fi utilizat în locul diferitelor blocuri în diferite procese de lucru. De obicei, acest lucru se întâmplă cu interogări UNION ALL. Dezavantajul este că există mai puțin paralelism, deoarece fiecare proces de lucru procesează doar o interogare.
Aici sunt active 2 procese de lucru, deși sunt activate 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 <= date '2000-12-01' - interval '105' day;
QUERY PLAN
------------------------------------------------------------------------------------------------
Gather
Workers Planned: 2
-> Parallel Append
-> Aggregate
-> Seq Scan on lineitem
Filter: (l_shipdate <= '2000-08-18 00:00:00'::timestamp without time zone)
-> Aggregate
-> Seq Scan on lineitem lineitem_1
Filter: (l_shipdate <= '1998-08-18 00:00:00'::timestamp without time zone)Cele mai importante variabile
- WORK_MEM limitează cantitatea de memorie pentru fiecare proces, nu doar pentru interogări: work_mem procese conexiuni = foarte multă memorie.
- — câte procese de lucru va utiliza programul pentru procesarea paralelă din plan.
- — ajustează numărul total de procese de lucru la numărul de nuclee CPU de pe server.
- — același lucru, dar pentru procese de lucru paralele.
Concluzii
Începând cu versiunea 9.6, procesarea paralelă poate îmbunătăți semnificativ performanța interogărilor complexe care scanează multe rânduri sau indexuri. În PostgreSQL 10, procesarea paralelă este activată implicit. Nu uitați să o dezactivați pe serverele cu o sarcină de lucru OLTP mare. Scanările secvențiale sau scanările indexurilor consumă foarte multe resurse. Dacă nu efectuați un raport pe întregul set de date, interogările pot fi mai eficiente, adăugând pur și simplu indexurile lipsă sau folosind o partiționare corectă.
Linkuri
Sursa: habr.com
