Nei moderni CP ci sono moltissimi core. Per anni, le applicazioni hanno inviato richieste ai database in parallelo. Se si tratta di una richiesta di report su molte righe nella tabella, viene eseguita più rapidamente quando utilizza più CP, e in PostgreSQL ciò è possibile a partire dalla versione 9.6.
Ci sono voluti 3 anni per implementare la funzione delle richieste parallele — è stato necessario riscrivere il codice in diverse fasi dell'esecuzione delle richieste. In PostgreSQL 9.6 è stata introdotta un'infrastruttura per un ulteriore miglioramento del codice. Nelle versioni successive, anche altri tipi di richieste vengono eseguiti in parallelo.
Limitazioni
- Non attivare l'esecuzione parallela se tutti i core sono già occupati, altrimenti altre richieste saranno rallentate.
- La cosa più importante è che l'elaborazione parallela con valori elevati di WORK_MEM utilizza molta memoria: ogni connessione hash o ordinamento occupa memoria per un ammontare di work_mem.
- Le richieste OLTP con bassa latenza non possono essere accelerate mediante l'esecuzione parallela. Se una richiesta restituisce una sola riga, l'elaborazione parallela non farà altro che rallentarla.
- Gli sviluppatori amano utilizzare il benchmark TPC-H. Forse hai richieste simili per un'esecuzione parallela ottimale.
- Solo le richieste SELECT senza blocchi predicati vengono eseguite in parallelo.
- A volte una corretta indicizzazione è migliore della scansione sequenziale della tabella in modalità parallela.
- La sospensione delle richieste e i cursori non sono supportati.
- Le funzioni finestra e le funzioni aggregate dei set ordinati non sono parallele.
- Non guadagnate nulla in carico di I/O.
- Non esistono algoritmi di ordinamento paralleli. Tuttavia, le richieste con ordinamenti possono essere eseguite in parallelo in alcuni aspetti.
- Sostituisci CTE (WITH …) con un SELECT annidato per includere l'elaborazione parallela.
- I wrapper di dati di terze parti non supportano ancora l'elaborazione parallela (anche se potrebbero!).
- FULL OUTER JOIN non è supportato.
- max_rows disabilita l'elaborazione parallela.
- Se nella richiesta è presente una funzione non contrassegnata come PARALLEL SAFE, essa sarà monothread.
- Il livello di isolamento della transazione SERIALIZABLE disabilita l'elaborazione parallela.
Ambiente di test
Gli sviluppatori di PostgreSQL hanno cercato di ridurre i tempi di risposta delle richieste del benchmark TPC-H. Scarica il benchmark e . Questo utilizzo non ufficiale del benchmark TPC-H non è per il confronto di database o hardware.
- Carica TPC-H_Tools_v2.17.3.zip (o una versione più recente) .
- Rinomina makefile.suite in Makefile e modifica come descritto qui: . Compila il codice con il comando make.
- Genera i dati:
.\/dbgen -s 10crea un database di 23 GB. Questo è sufficiente per vedere la differenza nelle prestazioni delle query parallele e non parallele. - Converti i file
tblincsv con foresed. - Clona il repository
pg_tpche copia i filecsvinpg_tpch\/dss\/data. - Crea le query con il comando
qgen. - Carica i dati nel database con il comando
.\/tpch.sh.
Scansione sequenziale parallela
Può risultare più veloce non a causa della lettura parallela, ma perché i dati sono distribuiti su molti core della CPU. Nei sistemi operativi moderni, i file di dati PostgreSQL vengono ben memorizzati nella cache. Con la lettura anticipata è possibile ottenere dal magazzino un blocco più grande rispetto a quello richiesto dal demone PG. Pertanto, le prestazioni della query non sono limitate dall'I/O del disco. Consuma cicli della CPU per:
- leggere le righe una ad una dalle pagine della tabella;
- confrontare i valori delle righe e le condizioni
DOVE.
Eseguiamo una semplice query 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 senza fuso orario)
Righe rimosse dal filtro: 1146337
Tempo di pianificazione: 0.203 ms
Tempo di esecuzione: 19035.100 msLa scansione sequenziale restituisce troppe righe senza aggregazione, quindi la query viene eseguita da un solo core della CPU.
Se aggiungiamo SUM(), si nota che due processi di lavoro aiutano ad accelerare la query:
explain analyze select sum(l_quantity) as sum_qty from lineitem where l_shipdate <= date '1998-12-01' - interval '105' day;
QUERY PLAN
----------------------------------------------------------------------------------------------------------------------------------------------------
Finalize Aggregate (cost=1589702.14..1589702.15 rows=1 width=32) (actual time=8553.365..8553.365 rows=1 loops=1)
-> Gather (cost=1589701.91..1589702.12 rows=2 width=32) (actual time=8553.241..8555.067 rows=3 loops=1)
Workers pianificati: 2
Workers lanciati: 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 senza fuso orario)
Righe rimosse dal filtro: 382112
Tempo di pianificazione: 0.241 ms
Tempo di esecuzione: 8555.131 msAggregazione parallela
Il nodo «Parallel Seq Scan» produce righe per l'aggregazione parziale. Il nodo «Partial Aggregate» riduce queste righe usando SUM(). Alla fine, il contatore SUM di ciascun processo di lavoro viene raccolto dal nodo «Gather».
Il risultato finale è calcolato dal nodo «Finalize Aggregate». Se hai le tue funzioni di aggregazione, assicurati di contrassegnarle come «parallel safe».
Numero di processi di lavoro
Il numero di processi di lavoro può essere aumentato senza riavviare il server:
explain analyze select sum(l_quantity) as sum_qty from lineitem where l_shipdate <= date '1998-12-01' - interval '105' day;
QUERY PLAN
----------------------------------------------------------------------------------------------------------------------------------------------------
Finalize Aggregate (cost=1589702.14..1589702.15 rows=1 width=32) (actual time=8553.365..8553.365 rows=1 loops=1)
-> Gather (cost=1589701.91..1589702.12 rows=2 width=32) (actual time=8553.241..8555.067 rows=3 loops=1)
Workers pianificati: 2
Workers lanciati: 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 senza fuso orario)
Righe rimosse dal filtro: 382112
Tempo di pianificazione: 0.241 ms
Tempo di esecuzione: 8555.131 msCosa sta succedendo qui? I processi di lavoro sono raddoppiati, mentre la richiesta è diventata solo 1,6599 volte più veloce. I calcoli sono interessanti. Avevamo 2 processi di lavoro e 1 leader. Dopo la modifica, sono diventati 4+1.
La nostra massima accelerazione dalla lavorazione parallela: 5/3 = 1,66(6) volte.
Come funziona?
Processi
L'esecuzione della richiesta inizia sempre con il processo guida. Il leader gestisce tutto ciò che non è paralellizzato e parte del trattamento parallelo. Gli altri processi, che eseguono le stesse richieste, vengono chiamati processi di lavoro. L'elaborazione parallela utilizza l'infrastruttura (dalla versione 9.4). Poiché altre parti di PostgreSQL utilizzano processi e non thread, una richiesta con 3 processi di lavoro potrebbe essere 4 volte più veloce rispetto all'elaborazione tradizionale.
Interazione
I processi di lavoro comunicano con il leader tramite una coda di messaggi (basata su memoria condivisa). Ogni processo ha 2 code: per gli errori e per le tuple.
Quanti processi di lavoro sono necessari?
Il limite minimo è impostato dal parametro . Poi l'esecutore delle query prende i processi di lavoro dal pool, limitato dal parametro . L'ultimo limite è dato da , cioè il numero totale di processi in background.
Se non è stato possibile allocare un processo di lavoro, l'elaborazione sarà monoprocessuale.
Il pianificatore delle query può ridurre i processi di lavoro a seconda delle dimensioni della tabella o dell'indice. Ci sono parametri per questo e .
set min_parallel_table_scan_size='8MB'
8MB tabella => 1 lavoratore
24MB tabella => 2 lavoratori
72MB tabella => 3 lavoratori
x => log(x / min_parallel_table_scan_size) / log(3) + 1 lavoratoreOgni volta che la tabella è 3 volte più grande di min_parallel_(index|table)_scan_size, Postgres aggiunge un processo di lavoro. Il numero di processi di lavoro non si basa sui costi. La dipendenza circolare rende difficili le implementazioni complesse. Invece, il pianificatore utilizza regole semplici.
Nella pratica queste regole non sono sempre adatte per la produzione, quindi è possibile modificare il numero di processi di lavoro per una tabella specifica: ALTER TABLE … SET (parallel_workers = N).
Perché non viene utilizzata l'elaborazione parallela?
Oltre a un lungo elenco di limitazioni, ci sono anche controlli sui costi:
— per evitar di gestire in parallelo richieste brevi. Questo parametro stima il tempo per la preparazione della memoria, l'avvio del processo e lo scambio iniziale di dati.
: la comunicazione tra il leader e i lavoratori può allungarsi proporzionalmente al numero di tuple dai processi di lavoro. Questo parametro calcola i costi per lo scambio di dati.
Join a cicli annidati — Nested Loop Join
PostgreSQL 9.6+ può eseguire cicli annidati in parallelo: si tratta di un'operazione semplice.
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)Il raccolto avviene nell'ultima fase, quindi il Nested Loop Left Join è un'operazione in parallelo. Il Parallel Index Only Scan è apparso solo nella versione 10. Funziona in modo analogo alla scansione sequenziale parallela. La condizione c_custkey = o_custkey legge un ordine per ciascuna riga del cliente. Quindi non è parallela.
Hash Join — Hash Join
Ogni processo di lavoro crea la propria tabella hash fino a PostgreSQL 11. E se ci sono più di quattro di questi processi, le prestazioni non aumenteranno. Nella nuova versione, la tabella hash è condivisa. Ogni processo di lavoro può utilizzare WORK_MEM per creare una tabella hash.
seleziona
l_shipmode,
somma(caso
quando o_orderpriority = '1-URGENTE'
o o_orderpriority = '2-ALTA'
allora 1
altro 0
fine) come high_line_count,
somma(caso
quando o_orderpriority '1-URGENTE'
e o_orderpriority '2-ALTA'
allora 1
altro 0
fine) come low_line_count
da
ordini,
lineitem
dove
o_orderkey = l_orderkey
e l_shipmode in ('MAIL', 'ARIA')
e l_commitdate < l_receiptdate
e l_shipdate = data '1996-01-01'
e l_receiptdate < data '1996-01-01' + intervallo '1' anno
gruppo per
l_shipmode
ordine per
l_shipmode
LIMIT 1;
PIANO QUERY
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Limit (costo=1964755.66..1964961.44 righe=1 larghezza=27) (tempo effettivo=7579.592..7922.997 righe=1 cicli=1)
-> Finalizza GroupAggregate (costo=1964755.66..1966196.11 righe=7 larghezza=27) (tempo effettivo=7579.590..7579.591 righe=1 cicli=1)
Chiave di gruppo: lineitem.l_shipmode
-> Raggruppa Merge (costo=1964755.66..1966195.83 righe=28 larghezza=27) (tempo effettivo=7559.593..7922.319 righe=6 cicli=1)
Lavoratori pianificati: 4
Lavoratori lanciati: 4
-> Gruppo Parziale Aggregato (costo=1963755.61..1965192.44 righe=7 larghezza=27) (tempo effettivo=7548.103..7564.592 righe=2 cicli=5)
Chiave di gruppo: lineitem.l_shipmode
-> Ordina (costo=1963755.61..1963935.20 righe=71838 larghezza=27) (tempo effettivo=7530.280..7539.688 righe=62519 cicli=5)
Chiave di ordinamento: lineitem.l_shipmode
Metodo di ordinamento: fusione esterna Disco: 2304kB
Lavoratore 0: Metodo di ordinamento: fusione esterna Disco: 2064kB
Lavoratore 1: Metodo di ordinamento: fusione esterna Disco: 2384kB
Lavoratore 2: Metodo di ordinamento: fusione esterna Disco: 2264kB
Lavoratore 3: Metodo di ordinamento: fusione esterna Disco: 2336kB
-> Unione Hash parallela (costo=382571.01..1957960.99 righe=71838 larghezza=27) (tempo effettivo=7036.917..7499.692 righe=62519 cicli=5)
Condizione Hash: (lineitem.l_orderkey = orders.o_orderkey)
-> Scansione Seq parallela su lineitem (costo=0.00..1552386.40 righe=71838 larghezza=19) (tempo effettivo=0.583..4901.063 righe=62519 cicli=5)
Filtro: ((l_shipmode = QUALSIASI ('{MAIL,ARIA}'::bpchar[])) E (l_commitdate < l_receiptdate) E (l_shipdate = '1996-01-01'::data) E (l_receiptdate < '1997-01-01 00:00:00'::timestamp senza fuso orario))
Righe rimosse dal filtro: 11934691
-> Hash parallelo (costo=313722.45..313722.45 righe=3750045 larghezza=20) (tempo effettivo=2011.518..2011.518 righe=3000000 cicli=5)
Focolari: 65536 Batches: 256 Utilizzo memoria: 3840kB
-> Scansione Seq parallela su ordini (costo=0.00..313722.45 righe=3750045 larghezza=20) (tempo effettivo=0.029..995.948 righe=3000000 cicli=5)
Tempo di pianificazione: 0.977 ms
Tempo di esecuzione: 7923.770 msLa Query 12 di TPC-H mostra chiaramente la connessione hash parallela. Ogni processo di lavoro partecipa alla creazione di una tabella hash comune.
Unione per fusione — Merge Join
L'unione per fusione è intrinsecamente non parallela. Non preoccupatevi se questo è l'ultimo passaggio della query, può comunque essere eseguito in parallelo.
-- Query 2 da TPC-H
spiega (costi disattivati) seleziona s_acctbal, s_name, n_name, p_partkey, p_mfgr, s_address, s_phone, s_comment
da part, supplier, partsupp, nation, region
dove
p_partkey = ps_partkey
e s_suppkey = ps_suppkey
e p_size = 36
e p_type like '%BRASS'
e s_nationkey = n_nationkey
e n_regionkey = r_regionkey
e r_name = 'AMERICA'
e ps_supplycost = (
seleziona
min(ps_supplycost)
da partsupp, supplier, nation, region
dove
p_partkey = ps_partkey
e s_suppkey = ps_suppkey
e s_nationkey = n_nationkey
e n_regionkey = r_regionkey
e r_name = 'AMERICA'
)
ordina per s_acctbal desc, n_name, s_name, p_partkey
LIMIT 100;
PIANTA DELLA QUERY
----------------------------------------------------------------------------------------------------------
Limit
-> Ordina
Chiave di ordinamento: supplier.s_acctbal DESC, nation.n_name, supplier.s_name, part.p_partkey
-> Merge Join
Condizione di unione: (part.p_partkey = partsupp.ps_partkey)
Filtro di unione: (partsupp.ps_supplycost = (SubPiano 1))
-> Raccogli Merge
Lavoratori pianificati: 4
-> Scansione Indice Parallela usando <strong>part_pkey</strong> su parte
Filtro: (((p_type)::text ~~ '%BRASS'::text) E (p_size = 36))
-> Materializza
-> Ordina
Chiave di ordinamento: partsupp.ps_partkey
-> Ciclo annidato
-> Ciclo annidato
Filtro di unione: (nation.n_regionkey = region.r_regionkey)
-> Scansione sequenziale su region
Filtro: (r_name = 'AMERICA'::bpchar)
-> Hash Join
Condizione Hash: (supplier.s_nationkey = nation.n_nationkey)
-> Scansione sequenziale su supplier
-> Hash
-> Scansione sequenziale su nation
-> Scansione Indice usando idx_partsupp_suppkey su partsupp
Condizione indice: (ps_suppkey = supplier.s_suppkey)
SubPiano 1
-> Aggregato
-> Ciclo annidato
Filtro di unione: (nation_1.n_regionkey = region_1.r_regionkey)
-> Scansione sequenziale su region region_1
Filtro: (r_name = 'AMERICA'::bpchar)
-> Ciclo annidato
-> Ciclo annidato
-> Scansione Indice usando idx_partsupp_partkey su partsupp partsupp_1
Condizione indice: (part.p_partkey = ps_partkey)
-> Scansione Indice usando supplier_pkey su supplier supplier_1
Condizione indice: (s_suppkey = partsupp_1.ps_suppkey)
-> Scansione Indice usando nation_pkey su nation nation_1
Condizione indice: (n_nationkey = supplier_1.s_nationkey)Il nodo «Merge Join» è posizionato sopra «Gather Merge». Quindi la fusione non utilizza l'elaborazione parallela. Tuttavia, il nodo «Parallel Index Scan» aiuta comunque nel segmento part_pkey.
Unione per sezioni
In PostgreSQL 11 è disattivata per impostazione predefinita: ha una pianificazione molto costosa. Le tabelle con sezioni simili possono essere unite sezione per sezione. In questo modo Postgres utilizzerà tabelle hash più piccole. Ogni unione di sezioni può essere parallela.
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)Soprattutto, l'unione per sezioni può essere parallela solo se queste sezioni sono abbastanza grandi.
Appendice parallela — Parallel Append
può essere utilizzata al posto di diversi blocchi in diversi processi di lavoro. Di solito questo avviene con query UNION ALL. Il difetto è che c'è meno parallelismo, poiché ogni processo di lavoro gestisce solo 1 query.
Qui sono in esecuzione 2 processi di lavoro, nonostante siano previsti 4.
tpch=# spiega (costi disattivati) seleziona somma(l_quantity) come sum_qty da lineitem dove l_shipdate <= data '1998-12-01' - intervallo '105' giorni unione tutto seleziona somma(l_quantity) come sum_qty da lineitem dove l_shipdate <= data '2000-12-01' - intervallo '105' giorni;
PIANO DI QUERY
------------------------------------------------------------------------------------------------
Raccolta
Lavoratori Pianificati: 2
-> Appendice Parallela
-> Aggregato
-> Scansione Sequenziale su lineitem
Filtro: (l_shipdate <= '2000-08-18 00:00:00'::timestamp senza fuso orario)
-> Aggregato
-> Scansione Sequenziale su lineitem lineitem_1
Filtro: (l_shipdate <= '1998-08-18 00:00:00'::timestamp senza fuso orario)Le variabili più importanti
- WORK_MEM limita la quantità di memoria per ogni processo, non solo per le query: work_mem processi connessioni = molta memoria.
- — quanti processi di lavoro utilizzerà il programma in esecuzione per l'elaborazione parallela dal piano.
- — adatta il numero totale di processi di lavoro al numero di core della CPU sul server.
- — lo stesso, ma per i processi di lavoro paralleli.
Conclusioni
A partire dalla versione 9.6, l'elaborazione parallela può migliorare significativamente le prestazioni delle query complesse che esaminano molte righe o indici. In PostgreSQL 10, l'elaborazione parallela è abilitata per impostazione predefinita. Ricorda di disabilitarla sui server con un carico di lavoro OLTP elevato. Le scansioni sequenziali o le scansioni degli indici consumano molte risorse. Se non stai eseguendo un report su tutto il set di dati, le query possono diventare più efficienti semplicemente aggiungendo gli indici mancanti o utilizzando una corretta partizionamento.
Link
Fonte: habr.com
