Moderne CPUs verfügen über viele Kerne. Über Jahre hinweg haben Anwendungen parallel Datenbankanfragen gesendet. Bei einer Berichtsanfrage an viele Zeilen in einer Tabelle wird die Ausführung schneller, wenn mehrere CPUs genutzt werden, und in PostgreSQL ist dies ab Version 9.6 möglich.
Es dauerte 3 Jahre, die Funktion für parallele Abfragen zu implementieren — der Code musste in verschiedenen Phasen der Abfrageausführung neu geschrieben werden. In PostgreSQL 9.6 wurde eine Infrastruktur zur weiteren Verbesserung des Codes eingeführt. In späteren Versionen können auch andere Abfragetypen parallel ausgeführt werden.
Einschränkungen
- Aktivieren Sie die parallele Ausführung nicht, wenn alle Kerne bereits belegt sind, da ansonsten andere Abfragen langsamer 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 Speicher in Höhe von work_mem.
- OLTP-Anfragen mit niedriger Latenz können durch parallele Ausführung nicht beschleunigt werden. Wenn eine Anfrage nur eine Zeile zurückgibt, wird die parallele Verarbeitung sie sogar verlangsamen.
- Entwickler nutzen gerne den TPC-H-Benchmark. Vielleicht haben Sie ähnliche Abfragen für eine optimale parallele Ausführung.
- Nur SELECT-Anfragen ohne prädikative Sperrung werden parallel ausgeführt.
- Manchmal ist die richtige Indizierung besser als das sequenzielle Scannen der Tabelle im Parallelmodus.
- Anfragen und Cursor werden nicht unterstützt.
- Fensterfunktionen und aggregierte Funktionen von geordneten Mengen sind nicht parallel.
- Sie gewinnen nichts bei der Eingabe-/Ausgabebelastung.
- Es gibt keine parallelen Sortieralgorithmen. Aber Anfragen mit Sortierungen können in bestimmten Aspekten parallel ausgeführt werden.
- Ersetzen Sie CTE (WITH …) durch ein geschachteltes SELECT, um die parallele Verarbeitung zu ermöglichen.
- Datenwrapper von Drittanbietern unterstützen bisher keine parallele Verarbeitung (könnten es aber!).
- FULL OUTER JOIN wird nicht unterstützt.
- max_rows deaktiviert die parallele Verarbeitung.
- Wenn eine Funktion im Abfrage nicht als PARALLEL SAFE gekennzeichnet ist, wird sie einsträngig.
- Das Isolationsniveau der Transaktion SERIALIZABLE deaktiviert die parallele Verarbeitung.
Testumgebung
Die Entwickler von PostgreSQL haben versucht, die Reaktionszeit der TPC-H-Benchmarkabfragen zu verkürzen. Laden Sie den Benchmark herunter und . Dies ist die inoffizielle Verwendung des TPC-H Benchmarks — nicht für den 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 die Daten:
./dbgen -s 10erstellt eine Datenbank von 23 GB. Das reicht aus, um den Unterschied in der Leistung zwischen parallelen und nicht-parallelen Abfragen zu sehen. - Konvertieren Sie die Dateien
tblincsv mit forundsed. - Klone das Repository
pg_tpchund kopieren Sie die Dateiencsvinpg_tpch/dss/data. - Erstellen Sie die Abfragen mit dem Befehl
qgen. - Laden Sie die Daten in die Datenbank mit dem Befehl hoch
./tpch.sh.
Paralleles sequentielles Scannen
Es könnte schneller sein, nicht aufgrund des parallelen Lesens, sondern weil die Daten über viele CPU-Kerne verteilt sind. In modernen Betriebssystemen werden PostgreSQL-Datendateien gut zwischengespeichert. Mit vorab gelesenen Daten kann mehr aus dem Speicher abgerufen werden, als der PG-Dämon anfordert. Daher wird die Abfrageleistung nicht durch die Festplatten-I/O limitiert. Sie benötigt CPU-Zyklen, um:
- Zeilen einzeln von den Tabellen-Seiten zu lesen;
- Werte der Zeilen und Bedingungen zu vergleichen
WHERE.
Führen wir eine einfache Abfrage durch 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 liefert zu viele Zeilen ohne Aggregation, weshalb die Abfrage nur von einem CPU-Kern ausgeführt wird.
Wenn Sie hinzufügen SUM(), wird deutlich, dass zwei Arbeitsprozesse helfen können, die Anfrage 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 „Paralleler Seq Scan“ erzeugt Zeilen für die partielle Aggregation. Der Knoten „Partielle Aggregation“ reduziert diese Zeilen mit Hilfe von SUM(). Am Ende werden die SUM-Zähler jeder Arbeitsinstanz vom Knoten „Gather“ gesammelt.
Das Endergebnis wird vom Knoten „Finalize Aggregate“ berechnet. Wenn Sie eigene Aggregationsfunktionen haben, vergessen Sie nicht, sie als „parallel safe“ zu kennzeichnen.
Anzahl der Arbeitsprozesse
Die Anzahl der Arbeitsprozesse kann ohne einen 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 Anfrage wurde nur um das 1,6599-fache schneller. Die Berechnungen sind interessant. Wir hatten 2 Arbeitsprozesse und 1 Führungsprozess. Nach der Änderung wurde es 4+1.
Unsere maximale Beschleunigung durch parallele Verarbeitung: 5/3 = 1,66(6) mal.
Wie funktioniert das?
Prozesse
Die Ausführung einer Anfrage beginnt immer mit dem führenden Prozess. Der Führungsprozess erledigt alles Nicht-Parallele und einen Teil der parallelen Verarbeitung. Die anderen Prozesse, die dieselben Anfragen bearbeiten, werden Arbeitsprozesse genannt. Die parallele Verarbeitung nutzt die Infrastruktur von (ab Version 9.4). Da andere Teile von PostgreSQL Prozesse anstelle von Threads verwenden, kann eine Anfrage mit 3 Arbeitern bis zu 4 Mal schneller sein als die traditionelle Verarbeitung.
Interaktion
Die Arbeiter kommunizieren mit dem Leader über eine Nachrichtenwarteschlange (auf Basis gemeinsamer Speicher). Jeder Prozess hat 2 Warteschlangen: eine für Fehler und eine für Tupel.
Wie viele Arbeiter werden benötigt?
Die Mindestgrenze wird durch den Parameter festgelegt. Dann nimmt der Abfrageausführer Arbeiter aus dem Pool, der durch den Parameter begrenzt ist. Die letzte Grenze ist , also die Gesamtzahl der Hintergrundprozesse.
Wenn kein Arbeiter zugeordnet werden kann, erfolgt die Verarbeitung mit einem einzelnen Prozess.
Der Abfrageplaner kann die Arbeiter 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 Arbeiter
24MB Tabelle => 2 Arbeiter
72MB Tabelle => 3 Arbeiter
x => log(x / min_parallel_table_scan_size) / log(3) + 1 ArbeiterJedes Mal, wenn die Tabelle 3 Mal größer ist als min_parallel_(index|table)_scan_size, Postgres fügt einen Arbeitsprozess hinzu. Die Anzahl der Arbeitsprozesse basiert nicht auf den Kosten. Zirkuläre Abhängigkeiten erschweren komplexe Implementierungen. Stattdessen verwendet der Planner einfache Regeln.
In der Praxis sind diese Regeln nicht immer für die Produktion geeignet, sodass Sie die Anzahl der Arbeitsprozesse für eine bestimmte Tabelle ändern können: ALTER TABLE … SET (parallel_workers = N).
Warum wird die parallele Verarbeitung nicht verwendet?
Neben einer langen Liste von Einschränkungen gibt es auch Kostenüberprüfungen:
— um auf parallele Verarbeitung kurzer Abfragen zu verzichten. Dieser Parameter schätzt die Zeit zum Vorbereiten des Speichers, zum Starten des Prozesses und zum ersten Datenaustausch.
: Die Kommunikation zwischen dem Leader und den Arbeitsprozessen kann proportional zur Anzahl der Tupel von den Arbeitsprozessen verzögert 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
Gruppen Schlüssel: customer.c_custkey
-> Gather Merge
Geplante Arbeiter: 4
-> Teilweise GroupAggregate
Gruppen Schlüssel: 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 Bedingung: (customer.c_custkey = o_custkey)
Filter: ((o_comment)::text !~~ '%specialposits%'::text)Die Zusammenführung 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 der parallele sequentielle Scan. Bedingung c_custkey = o_custkey liest eine Bestellung für jede Kundenzeile. Daher ist es nicht parallel.
Hash Join
Jeder Arbeitsprozess erstellt seine eigene Hash-Tabelle bis PostgreSQL 11. Wenn es mehr als vier dieser Prozesse gibt, wird die Leistung nicht gesteigert. In der neuen Version ist die Hash-Tabelle gemeinsam. Jeder Arbeitsprozess kann WORK_MEM nutzen, um eine Hash-Tabelle zu erstellen.
wählen
l_shipmode,
sum(case
when o_orderpriority = '1-URGENT'
or o_orderpriority = '2-HIGH'
dann 1
sonst 0
end) as high_line_count,
sum(case
when o_orderpriority <> '1-URGENT'
und o_orderpriority <> '2-HIGH'
dann 1
sonst 0
end) as low_line_count
from
orders,
lineitem
where
o_orderkey = l_orderkey
und l_shipmode in ('MAIL', 'AIR')
und l_commitdate < l_receiptdate
und l_shipdate < l_commitdate
und l_receiptdate >= date '1996-01-01'
und 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 < l_commitdate) AND (l_receiptdate >= '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 msDie Anfrage 12 aus TPC-H zeigt anschaulich eine parallele Hash-Verknüpfung. Jeder Workprozess trägt zur Erstellung einer gemeinsamen Hash-Tabelle bei.
Merge Join
Ein Merge Join ist grundsätzlich nicht parallel. Machen Sie sich keine Sorgen, wenn dies die letzte Phase der Anfrage ist – sie kann dennoch parallel ausgeführt werden.
-- Abfrage 2 aus TPC-H
Erklärung (Kosten aus)
Wählen Sie s_acctbal, s_name, n_name, p_partkey, p_mfgr, s_address, s_phone, s_comment
von Teil, Lieferant, 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 = 'AMERIKA'
und ps_supplycost = (
wählen
min(ps_supplycost)
von partsupp, Lieferant, 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 = 'AMERIKA'
)
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 unter Verwendung von <strong>part_pkey</strong> auf Teil
Filter: (((p_type)::text ~~ '%BRASS'::text) UND (p_size = 36))
-> Materialisiere
-> 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)
Teilplan 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“ ist über dem „Gather Merge“. Daher nutzt das Merging keine parallele Verarbeitung. Aber der Knoten „Parallel Index Scan“ hilft dennoch mit dem Segment. part_pkey.
Partition Join
in PostgreSQL 11 ist standardmäßig deaktiviert, da es sehr kostenintensive Planungen erfordert. Tabellen mit ähnlicher Partitionierung können Abschnitt für Abschnitt verknüpft werden. So verwendet Postgres kleinere Hash-Tabellen. Jede Verknüpfung der Partitionen 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 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 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 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 Parallel Hash
-> Parallel Seq Scan on prt1_p1 t1
Filter: (b = 0)Das wichtigste ist, dass die Partitionierungsverbindungen nur dann parallel sind, wenn diese Partitionen ausreichend groß sind.
Paralleler Anhang — Parallel Append
kann anstelle verschiedener Blöcke in unterschiedlichen Arbeitsabläufen verwendet werden. Dies geschieht häufig bei UNION ALL-Anfragen. Nachteil ist weniger Parallelität, da jeder Arbeitsablauf nur eine Anfrage bearbeitet.
Hier sind 2 Arbeitsabläufe gestartet, obwohl 4 aktiviert sind.
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 Paralleler Anhang
-> 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)Die wichtigsten Variablen
- WORK_MEM begrenzt den Speicher für jeden Prozess, nicht nur für Anfragen: work_mem Prozesse Verbindungen = sehr viel Speicher.
- — wie viele Arbeitsabläufe das ausführende Programm für die parallele Verarbeitung gemäß dem Plan verwenden wird.
- — passt die Gesamtlänge der Arbeitsabläufe an die Anzahl der CPU-Kerne im Server an.
- — das Gleiche, aber für parallele Arbeitsprozesse.
Ergebnisse
Seit Version 9.6 kann die parallele Verarbeitung die Leistung komplexer Abfragen deutlich verbessern, die viele Zeilen oder Indizes scannen. In PostgreSQL 10 ist die parallele Verarbeitung standardmäßig aktiviert. Denken Sie daran, sie auf Servern mit hoher OLTP-Last zu deaktivieren. Sequentielle Scans oder das Scannen von Indizes verbrauchen sehr viele Ressourcen. Wenn Sie keinen Bericht über den gesamten Datensatz erstellen, können Abfragen produktiver gestaltet werden, indem einfach fehlende Indizes hinzugefügt oder die richtige Partitionierung verwendet wird.
Links
Quelle: habr.com
