In modernen CPUs gibt es sehr viele Kerne. Über Jahre hinweg haben Anwendungen parallel Anfragen an Datenbanken gesendet. Wenn es sich um eine Berichtsanfrage für viele Zeilen in einer Tabelle handelt, wird sie schneller ausgeführt, wenn mehrere CPUs beteiligt sind, und das ist in PostgreSQL seit Version 9.6 möglich.
Es hat 3 Jahre gedauert, um die Funktion für parallele Anfragen zu implementieren — der Code musste in verschiedenen Phasen der Anfrageausführung überarbeitet werden. In PostgreSQL 9.6 wurde eine Infrastruktur zur weiteren Verbesserung des Codes eingeführt. In den folgenden Versionen werden auch andere Arten von Anfragen parallel ausgeführt.
Einschränkungen
- Aktivieren Sie die parallele Ausführung nicht, wenn alle Kerne bereits ausgelastet sind, da sonst andere Anfragen verzögert werden.
- Am wichtigsten ist, dass die parallele Verarbeitung mit hohen Werten für WORK_MEM viel Speicher beansprucht – jede Hash-Verbindung oder Sortierung benötigt Arbeitsspeicher in der Größe von work_mem.
- Anfragen im OLTP mit geringer Latenz können nicht durch parallele Ausführung beschleunigt werden. Und wenn die Anfrage eine Zeile zurückgibt, verzögert die parallele Verarbeitung sie nur.
- Entwickler verwenden gerne den TPC-H-Benchmark. Vielleicht haben Sie ähnliche Anfragen für eine ideale parallele Ausführung.
- Nur SELECT-Anfragen ohne prädikatenbasierte Sperren werden parallel ausgeführt.
- Manchmal ist die richtige Indizierung besser als eine sequenzielle Durchsuchung der Tabelle im parallelen Modus.
- Anfragen und Cursor werden nicht unterstützt.
- Fensterfunktionen und Aggregatfunktionen für geordnete Mengen sind nicht parallel.
- In der Arbeitslast von Eingabe-Ausgabe profitieren Sie nicht von Parallelität.
- Es gibt keine parallelen Sortieralgorithmen. Aber Abfragen mit Sortierungen können in einigen Aspekten parallel ausgeführt werden.
- Ersetzen Sie CTE (WITH …) durch eine verschachtelte SELECT-Abfrage, um die parallele Verarbeitung zu aktivieren.
- Wrapper für externe Daten unterstützen bislang keine parallele Verarbeitung (während sie könnten!)
- FULL OUTER JOIN wird nicht unterstützt.
- max_rows schaltet die parallele Verarbeitung aus.
- Wenn in der Anfrage eine Funktion enthalten ist, die nicht als PARALLEL SAFE gekennzeichnet ist, wird sie wieder in einem Thread ausgeführt.
- Das Isolationslevel der Transaktion SERIALIZABLE schaltet die parallele Verarbeitung aus.
Testumgebung
Die Entwickler von PostgreSQL haben versucht, die Antwortzeiten von Anfragen des TPC-H-Benchmarks zu verkürzen. Laden Sie den Benchmark herunter und . Diese inoffizielle Verwendung des TPC-H-Benchmarks dient nicht dem Vergleich von Datenbanken oder Hardware.
- Laden Sie TPC-H_Tools_v2.17.3.zip (oder eine neuere Version) herunter .
- Benennen Sie makefile.suite in Makefile um und ändern Sie es wie hier beschrieben: . Kompilieren Sie den Code mit dem Befehl make.
- Generieren Sie Daten:
./dbgen -s 10generiert eine Datenbank von 23 GB. Das ist ausreichend, um den Unterschied im Leistungsvermögen von parallelen und nicht-parallelen Abfragen zu erkennen. - Konvertieren Sie die Dateien
tblincsv mit forundsed. - Klonen Sie das Repository
pg_tpchund kopieren Sie die Dateiencsvinpg_tpch/dss/data. - Erstellen Sie Abfragen mit dem Befehl
qgen. - Laden Sie die Daten in die Datenbank mit dem Befehl
./tpch.sh.
Paralleles sequenzielles Scannen
Es kann schneller sein, nicht wegen des parallelen Lesens, sondern weil die Daten über viele CPU-Kerne verteilt sind. In modernen Betriebssystemen werden PostgreSQL-Daten Dateien gut zwischengespeichert. Durch vorabrufende Lesevorgänge können größere Blöcke als vom PG-Dämon angefordert aus dem Speicher geladen werden. Daher wird die Abfragegeschwindigkeit nicht durch die I/O-Leistung der Festplatte begrenzt. Sie verbraucht CPU-Zyklen für:
- das Zeilenweise Lesen von Seiten der Tabelle;
- den Vergleich von Zeilenwerten und Bedingungen.
WHERE.
Führen wir eine einfache Abfrage aus: 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 msEin sequentieller Scan gibt zu viele Zeilen ohne Aggregation zurück, sodass die Abfrage von einem einzigen CPU-Kern ausgeführt wird.
Wenn man SUM()hinzufügt, sieht man, dass zwei Arbeitsprozesse dazu beitragen, die Abfrage zu beschleunigen:
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 msParallele Aggregation
Der Knoten „Parallel Seq Scan“ erzeugt Zeilen für die partielle Aggregation. Der Knoten „Partial Aggregate“ kürzt diese Zeilen mithilfe von SUM(). Am Ende wird der SUM-Zähler aus jedem Arbeitsprozess vom Knoten „Gather“ gesammelt.
Das Endergebnis wird vom Knoten „Finalize Aggregate“ berechnet. Wenn Sie eigene Aggregatfunktionen haben, vergessen Sie nicht, diese als „parallel safe“ zu kennzeichnen.
Anzahl der Arbeitsprozesse
Die Anzahl der Arbeitsprozesse kann ohne Neustart des Servers erhöht werden:
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 msWas passiert hier? Die Anzahl der Arbeitsprozesse hat sich verdoppelt, und die Abfrage wurde nur um 1,6599-mal schneller. Die Berechnungen sind interessant. Wir hatten 2 Arbeitsprozesse und 1 Leader. Nach der Änderung wurden es 4+1.
Unser maximaler Geschwindigkeitszuwachs durch parallele Verarbeitung: 5/3 = 1,66(6)-fach.
Wie funktioniert das?
Prozesse
Die Ausführung einer Abfrage beginnt immer mit dem führenden Prozess. Der Leader führt alles nicht-parallele und einen Teil der parallelen Verarbeitung durch. Andere Prozesse, die dieselben Abfragen ausführen, werden Arbeitsprozesse genannt. Die parallele Verarbeitung nutzt die Infrastruktur (seit Version 9.4). Da andere Teile von PostgreSQL Prozesse anstelle von Threads verwenden, könnte eine Abfrage mit 3 Arbeitsprozessen bis zu 4-mal schneller sein als die traditionelle Verarbeitung.
Interaktion
Die Arbeitsprozesse kommunizieren über eine Nachrichtenwarteschlange (basierend auf gemeinsamem Speicher) mit dem Leader. Jeder Prozess hat 2 Warteschlangen: für Fehler und für Tupel.
Wie viele Arbeitsprozesse werden benötigt?
Die minimale Begrenzung wird durch den Parameter gesetzt. Dann zieht der Abfrageausführer Arbeitsprozesse aus dem Pool, der durch den Parameter begrenzt wird. Das letzte Limit ist , also die Gesamtzahl der Hintergrundprozesse.
Wenn es nicht möglich ist, einen Arbeitsprozess zuzuweisen, erfolgt die Verarbeitung in einem Einzelprozess.
Der Abfrageplaner kann die Arbeitsprozesse je nach Größe der Tabelle oder des Indexes reduzieren. Dafür gibt es die Parameter und .
set min_parallel_table_scan_size='8MB'
8MB Tabelle => 1 Arbeitsprozess
24MB Tabelle => 2 Arbeitsprozesse
72MB Tabelle => 3 Arbeitsprozesse
x => log(x / min_parallel_table_scan_size) / log(3) + 1 ArbeitsprozessJedes Mal, wenn die Tabelle 3-mal größer ist als min_parallel_(index|table)_scan_size, fügt Postgres einen Arbeitsprozess hinzu. Die Anzahl der Arbeitsprozesse basiert nicht auf den Kosten. Zirkuläre Abhängigkeiten erschweren komplexe Implementierungen. Stattdessen verwendet der Planer einfache Regeln.
In der Praxis sind diese Regeln nicht immer für die Produktion geeignet, sodass die Anzahl der Arbeitsprozesse für eine bestimmte Tabelle geändert werden kann: ALTER TABLE … SET (parallel_workers = N).
Warum wird die parallele Verarbeitung nicht verwendet?
Neben einer langen Liste von Einschränkungen gibt es auch Prüfungen der Kosten:
— um ohne parallele Verarbeitung kurzer Anfragen auszukommen. Dieser Parameter schätzt die Zeit für die Vorbereitung des Speichers, den Start des Prozesses und den anfänglichen Datenaustausch.
: Die Kommunikation zwischen dem Leader und den Worker-Prozessen kann proportional zur Anzahl der Tupel von den Worker-Prozessen in die Länge gezogen werden. Dieser Parameter berechnet die Kosten für den Datenaustausch.
Verschachtelte Schleifenverknüpfungen — Nested Loop Join
PostgreSQL 9.6+ kann verschachtelte Schleifen parallel ausführen – das ist eine einfache 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)Die Sammlung erfolgt in der letzten Phase, sodass der Nested Loop Left Join eine parallele Operation ist. Der Parallel Index Only Scan wurde erst in Version 10 eingeführt. Er funktioniert ähnlich wie ein paralleles sequenzielles Scannen. Die Bedingung c_custkey = o_custkey liest einen Auftrag für jede Kundenzeile. Daher ist er nicht parallel.
Hash-Verknüpfungen — Hash Join
Jeder Worker-Prozess erstellt seine eigene Hash-Tabelle bis PostgreSQL 11. Und wenn es mehr als vier dieser Prozesse gibt, wird die Leistung nicht gesteigert. In der neuen Version ist die Hash-Tabelle gemeinsam. Jeder Worker-Prozess kann WORK_MEM verwenden, um eine Hash-Tabelle zu erstellen.
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 < date '1996-01-01' + interval '1' year
group by
l_shipmode
order by
l_shipmode
LIMIT 1;
QUERY PLAN
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Limit (cost=1964755.66..1964961.44 rows=1 width=27) (actual time=7579.592..7922.997 rows=1 loops=1)
-> 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 < '1997-01-01 00:00:00'::timestamp without time zone))
Rows Removed by Filter: 11934691
-> 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 msAnfrage 12 aus TPC-H zeigt anschaulich das parallele Hash-Joint. Jeder Arbeitsprozess trägt zur Erstellung einer gemeinsamen Hash-Tabelle bei.
Verschmelzungs-Verknüpfung — Merge Join
Die Verschmelzungs-Verknüpfung ist von Natur aus nicht parallel. Machen Sie sich keine Sorgen, wenn dies die letzte Phase der Anfrage ist — sie kann trotzdem parallel ausgeführt werden.
-- Abfrage 2 von TPC-H
Erklären (Kosten aus) wähle s_acctbal, s_name, n_name, p_partkey, p_mfgr, s_address, s_phone, s_comment
von part, supplier, partsupp, nation, region
wo
p_partkey = ps_partkey
und s_suppkey = ps_suppkey
und p_size = 36
und p_type wie '%BRASS'
und s_nationkey = n_nationkey
und n_regionkey = r_regionkey
und r_name = 'AMERICA'
und ps_supplycost = (
wähle
min(ps_supplycost)
von partsupp, supplier, nation, region
wo
p_partkey = ps_partkey
und s_suppkey = ps_suppkey
und s_nationkey = n_nationkey
und n_regionkey = r_regionkey
und r_name = 'AMERICA'
)
bestellen nach s_acctbal absteigend, n_name, s_name, p_partkey
LIMIT 100;
ABFRAGEPLAN
----------------------------------------------------------------------------------------------------------
Limit
-> Sortieren
Sortierschlüssel: supplier.s_acctbal DESC, nation.n_name, supplier.s_name, part.p_partkey
-> Merge Join
Merge-Bedingung: (part.p_partkey = partsupp.ps_partkey)
Join-Filter: (partsupp.ps_supplycost = (SubPlan 1))
-> Gather Merge
Geplante Arbeiter: 4
-> Paralleler Index-Scan mit <strong>part_pkey</strong> auf Teil
Filter: (((p_type)::Text ~~ '%BRASS'::Text) UND (p_size = 36))
-> Materialisieren
-> Sortieren
Sortierschlüssel: partsupp.ps_partkey
-> Verschachtelter Loop
-> Verschachtelter Loop
Join-Filter: (nation.n_regionkey = region.r_regionkey)
-> Seq Scan auf region
Filter: (r_name = 'AMERICA'::bpchar)
-> Hash Join
Hash-Bedingung: (supplier.s_nationkey = nation.n_nationkey)
-> Seq Scan auf supplier
-> Hash
-> Seq Scan auf nation
-> Index-Scan unter Verwendung von idx_partsupp_suppkey auf partsupp
Index-Bedingung: (ps_suppkey = supplier.s_suppkey)
SubPlan 1
-> Aggregat
-> Verschachtelter Loop
Join-Filter: (nation_1.n_regionkey = region_1.r_regionkey)
-> Seq Scan auf region region_1
Filter: (r_name = 'AMERICA'::bpchar)
-> Verschachtelter Loop
-> Verschachtelter Loop
-> Index-Scan unter Verwendung von idx_partsupp_partkey auf partsupp partsupp_1
Index-Bedingung: (part.p_partkey = ps_partkey)
-> Index-Scan unter Verwendung von supplier_pkey auf supplier supplier_1
Index-Bedingung: (s_suppkey = partsupp_1.ps_suppkey)
-> Index-Scan unter Verwendung von nation_pkey auf nation nation_1
Index-Bedingung: (n_nationkey = supplier_1.s_nationkey)Der Knoten „Merge Join“ steht über „Gather Merge“. Daher nutzt die Verschmelzung keine parallele Verarbeitung. Aber der Knoten „Parallel Index Scan“ hilft immer noch mit dem Segment. part_pkey.
Teilungs-Verknüpfung
In PostgreSQL 11 ist standardmäßig deaktiviert: Sie hat eine sehr aufwendige Planung. Tabellen mit ähnlicher Partitionierung können Abschnitt für Abschnitt verknüpft werden. So wird Postgres kleinere Hash-Tabellen verwenden. Jede Abschnittsverknüpfung kann parallel sein.
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)Wichtig ist, dass die Teilungs-Verknüpfung nur dann parallel sein kann, wenn diese Abschnitte groß genug sind.
Paralleles Anhängen — Parallel Append
kann anstelle verschiedener Blöcke in verschiedenen Arbeitsprozessen verwendet werden. Dies kommt normalerweise bei UNION ALL-Abfragen vor. Nachteil — weniger Parallelität, da jeder Arbeitsprozess nur 1 Anfrage bearbeitet.
Hier sind 2 Arbeitsprozesse gestartet, obwohl 4 aktiviert sind.
tpch=# erkläre (Kosten aus) wähle sum(l_quantity) als sum_qty aus lineitem wo l_shipdate <= datum '1998-12-01' - intervall '105' tag union all wähle sum(l_quantity) als sum_qty aus lineitem wo l_shipdate <= datum '2000-12-01' - intervall '105' tag;
QUERY PLAN
------------------------------------------------------------------------------------------------
Sammeln
Geplante Arbeiter: 2
-> Paralleles Hinzufügen
-> Aggregat
-> Seq Scan auf lineitem
Filter: (l_shipdate <= '2000-08-18 00:00:00'::timestamp ohne Zeitzone)
-> Aggregat
-> Seq Scan auf lineitem lineitem_1
Filter: (l_shipdate <= '1998-08-18 00:00:00'::timestamp ohne Zeitzone)Die wichtigsten Variablen
- WORK_MEM begrenzt den Speicherumfang für jeden Prozess, nicht nur für Abfragen: work_mem Prozesse Verbindungen = sehr viel Speicher.
- — wie viele Arbeitsprozesse das ausführende Programm für die parallele Verarbeitung aus dem Plan verwenden wird.
- — passt die Gesamtanzahl der Arbeitsprozesse an die Anzahl der CPU-Kerne auf dem Server an.
- — das Gleiche, aber für parallele Arbeitsprozesse.
Ergebnisse
Seit Version 9.6 kann die parallele Verarbeitung die Leistung komplexer Abfragen, die viele Zeilen oder Indizes scannen, erheblich verbessern. In PostgreSQL 10 ist die parallele Verarbeitung standardmäßig aktiviert. Denken Sie daran, sie auf Servern mit hoher OLTP-Last zu deaktivieren. Sequenzielle Scans oder Indizes verbrauchen sehr viele Ressourcen. Wenn Sie keinen Bericht über den gesamten Datensatz erstellen, können Anfragen effizienter gestaltet werden, indem einfach die fehlenden Indizes hinzugefügt oder das richtige Partitioning verwendet wird.
Links
Quelle: habr.com
