A modern CPU has many cores. For years, applications have been sending requests to databases in parallel. If it's a reporting request for many rows in a table, it executes faster when multiple CPUs are involved, and in PostgreSQL, this has been possible since version 9.6.
It took 3 years to implement the parallel query feature — the code had to be rewritten at various stages of query execution. PostgreSQL 9.6 introduced the infrastructure for further code enhancements. In later versions, other types of queries can also be executed in parallel.
Korlátozások
- Do not enable parallel execution if all cores are already busy, otherwise, other queries will slow down.
- Most importantly, parallel processing with high WORK_MEM values consumes a lot of memory — each hash connection or sorting operation takes up memory equivalent to work_mem.
- OLTP queries with low latency cannot be accelerated through parallel execution. If a query returns a single row, parallel processing will only slow it down.
- Developers love using the TPC-H benchmark. Maybe you have similar queries for optimal parallel execution.
- Only SELECT queries without predicate locking execute in parallel.
- Sometimes proper indexing is better than sequential table scanning in parallel mode.
- Query suspensions and cursors are not supported.
- Window functions and aggregate functions of ordered sets are not parallel.
- You gain nothing in I/O workload.
- There are no parallel sorting algorithms. However, queries with sorts can execute in parallel in some aspects.
- Replace CTE (WITH …) with a nested SELECT to enable parallel processing.
- Third-party data wrappers do not currently support parallel processing (but they could!).
- FULL OUTER JOIN is not supported.
- max_rows disables parallel processing.
- If a function in the query is not marked as PARALLEL SAFE, it will be single-threaded.
- The SERIALIZABLE transaction isolation level disables parallel processing.
Tesztkörnyezet
PostgreSQL developers tried to reduce the response time of TPC-H benchmark queries. Download the benchmark and . This unofficial use of the TPC-H benchmark is not for comparing databases or hardware.
- Töltse le a TPC-H_Tools_v2.17.3.zip fájlt (vagy egy újabb verziót) .
- Renommozza a makefile.suite fájlt Makefile névre, és módosítsa az alábbiak szerint: . A kódot a make parancs végrehajtásával fordítsa le.
- Generáljon adatokat:
./dbgen -s 10ez egy 23 GB-os adatbázist hoz létre. Ez elegendő ahhoz, hogy lássa a párhuzamos és nem párhuzamos lekérdezések teljesítménykülönbségét. - Konvertálja a fájlokat
tblbe.csv formátumba a foréssed. - Klónozza a tárolót
pg_tpchés másolja a fájlokatcsvbe.pg_tpch/dss/data. - Hozza létre a lekérdezéseket a parancs segítségével
qgen. - Töltse fel az adatokat a következő paranccsal:
./tpch.sh.
Párhuzamos soros beolvasás
Ez nem feltétlenül a párhuzamos olvasás miatt lehet gyorsabb, hanem mert az adatok sok processzormag között oszlanak el. A modern operációs rendszerek jól gyorsítják a PostgreSQL adatfájlokat. Előrelátó olvasással egy blokkot is kaphatunk a tárolóból, ami nagyobb lehet, mint amit a PG démon kér. Így a lekérdezés teljesítménye nem korlátozódik a lemez I/O-ra. CPU ciklusokat használ, hogy:
- egyesével olvassa a sorokat a táblázat lapjain;
- összehasonlítsa a sorok értékeit és a feltételeket
AHOL.
Futtassunk egy egyszerű lekérdezést 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 msA soros beolvasás túl sok sort ad vissza aggregálás nélkül, ezért a lekérdezést egyetlen CPU mag hajtja végre.
Ha hozzáadunk egyet SUM(), láthatóan két munkafolyamat segít a lekérdezés gyorsabbá tételében:
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)
-> Párhuzamos 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 msPárhuzamos aggregáció
A „Párhuzamos Seq Scan” csomópont sorokat generál a részleges aggregáláshoz. A „Részleges Aggregátum” csomópont ezeket a sorokat csökkenti a következővel SUM(). A végén a SUM számláló minden munkafolyamatból összegyűjthető a „Gather” csomópont által.
A végső eredmény a „Finalize Aggregate” node által kerül kiszámításra. Ha saját aggregáló funkciókat használ, ne felejtse el megjelölni őket mint „párhuzamosan biztonságos”.
A munkafolyamatok száma
A munkafolyamatok száma újraindítás nélkül növelhető a szerveren:
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)
-> Párhuzamos 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 msMi történik itt? A munkafolyamatok száma megduplázódott, miközben a kérés csupán 1,6599-szer gyorsabb lett. Érdekes számítások. Eleinte 2 munkafolyamatunk volt és 1 vezető. A változtatás után 4+1 lett.
A párhuzamos feldolgozásunk maximális gyorsulása: 5/3 = 1,66(6) szoros.
Hogyan működik ez?
Folyamatok
A kérés végrehajtása mindig a vezető folyamatból indul. A vezető kezeli a teljes nem párhuzamos és a részben párhuzamos feldolgozást. A többi folyamat, amely ugyanazokat a kéréseket dolgozza fel, munkafolyamatoknak nevezzük. A párhuzamos feldolgozás az infrastruktúrát használja (a 9.4-es verziótól). Mivel a PostgreSQL más részei folyamatokat használnak, nem szálakat, a 3 munkafolyamattal rendelkező kérés akár 4-szer is gyorsabb lehet a hagyományos feldolgozáshoz képest.
Interakció
A munkafolyamatok üzenetsoron keresztül kommunikálnak a vezetővel (megosztott memória alapú). Minden folyamatnak 2 sora van: hibaüzenetekre és tuple-ökre.
Hány munkafolyamatra van szükség?
A minimális korlátot a következő paraméter adja meg . Aztán a lekérdező végrehajtó a munkafolyamatokat a korlátozott paraméterből veszi . Az utolsó korlát a , azaz a teljes háttér folyamataid száma.
Ha nem lehet munkafolyamatot kiosztani, a feldolgozás egy folyamatú lesz.
A lekérdezés ütemezője csökkentheti a munkafolyamatok számát a táblázat vagy index mérete alapján. Ehhez a következő paraméterek tartoznak és .
set min_parallel_table_scan_size='8MB'
8MB tábla => 1 munkafolyamat
24MB tábla => 2 munkafolyamat
72MB tábla => 3 munkafolyamat
x => log(x / min_parallel_table_scan_size) / log(3) + 1 munkafolyamatMinden alkalommal, amikor a tábla 3-szor nagyobb, mint min_parallel_(index|table)_scan_size, a Postgres hozzáad egy munkafolyamatot. A munkafolyamatok száma nem költségen alapul. A körkörös függőség bonyolult megvalósítást nehezít. Ehelyett az ütemező egyszerű szabályokat használ.
A gyakorlatban ezek a szabályok nem mindig alkalmazhatók a termelésre, így a munkafolyamatok számát meg lehet változtatni egy adott táblához: ALTER TABLE … SET (parallel_workers = N).
Miért nem használják a párhuzamos feldolgozást?
A hosszú korlátozások listáján kívül vannak költségellenőrzések is:
— hogy elkerülhessük a párhuzamosan feldolgozott rövid lekérdezéseket. Ez a paraméter megbecsüli a memória előkészítésének, a folyamat elindításának és az első adats Exchange időt.
: a vezető és a munkások közötti kommunikáció arányos lehet a munkafolyamatok tuple-jeinek számával. Ez a paraméter számolja az adatok cseréjének költségeit.
Belső ciklusok összekapcsolása — Nested Loop Join
A PostgreSQL 9.6+ párhuzamosan végrehajthat beágyazott ciklusokat — ez egy egyszerű művelet.
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)A gyűjtés az utolsó lépésben történik, így a Nested Loop Left Join - párhuzamos művelet. A Parallel Index Only Scan csak a 10-es verzióban jelent meg. Ez hasonlóan működik a párhuzamos szekvenciális vizsgálathoz. A feltétel c_custkey = o_custkey minden ügyfélsorhoz egy rendelést olvas be. Így nem párhuzamos.
Hash Join — Hash Join
Minden munkafolyamat saját hash-táblát készít a PostgreSQL 11 előtt. És ha ezekből a folyamatokból több mint négy van, a teljesítmény nem növekszik. Az új verzióban a hash-tábla közös. Minden munkafolyamat használhat WORK_MEM-t a hash-tábla létrehozásához.
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 msA 12. kérdés a TPC-H-ból világosan bemutatja a párhuzamos hash-jelentkezést. Minden munkafolyamat részt vesz egy közös hash-táblázat létrehozásában.
Összekapcsolás egyesítéssel — Merge Join
Az összekapcsolás egyesítéssel természeténél fogva nem párhuzamos. Ne aggódjon, ha ez a lekérdezés utolsó lépése, — akkor is végrehajtható párhuzamosan.
-- Lekérdezés 2 a TPC-H-ból
explain (costs off) 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;
QUERY PLAN
----------------------------------------------------------------------------------------------------------
Limit
-> Sort
Sort Key: supplier.s_acctbal DESC, nation.n_name, supplier.s_name, part.p_partkey
-> Merge Join
Merge Cond: (part.p_partkey = partsupp.ps_partkey)
Join Filter: (partsupp.ps_supplycost = (SubPlan 1))
-> Gather Merge
Workers Planned: 4
-> Párhuzamos Index Ellenőrzés használatával <strong>part_pkey</strong> a part
Filter: (((p_type)::text ~~ '%BRASS'::text) AND (p_size = 36))
-> Materializálás
-> Sort
Sort Key: partsupp.ps_partkey
-> Beágyazott Ciklus
-> Beágyazott Ciklus
Join Filter: (nation.n_regionkey = region.r_regionkey)
-> Seq Scan on region
Filter: (r_name = 'AMERICA'::bpchar)
-> Hash Join
Hash Cond: (supplier.s_nationkey = nation.n_nationkey)
-> Seq Scan on supplier
-> Hash
-> Seq Scan on nation
-> Index Scan using idx_partsupp_suppkey on partsupp
Index Cond: (ps_suppkey = supplier.s_suppkey)
SubPlan 1
-> Aggregerés
-> Beágyazott Ciklus
Join Filter: (nation_1.n_regionkey = region_1.r_regionkey)
-> Seq Scan on region region_1
Filter: (r_name = 'AMERICA'::bpchar)
-> Beágyazott Ciklus
-> Beágyazott Ciklus
-> Index Scan using idx_partsupp_partkey on partsupp partsupp_1
Index Cond: (part.p_partkey = ps_partkey)
-> Index Scan using supplier_pkey on supplier supplier_1
Index Cond: (s_suppkey = partsupp_1.ps_suppkey)
-> Index Scan using nation_pkey on nation nation_1
Index Cond: (n_nationkey = supplier_1.s_nationkey)A »Merge Join« csomópont a »Gather Merge« felett található. Így az egyesítés nem használ párhuzamos feldolgozást. De a »Parallel Index Scan« csomópont mégis segít a szegmensen. part_pkey.
Szekciók szerinti összekapcsolás
A PostgreSQL 11-ben alapértelmezés szerint letiltva: nagyon költséges a tervezése. Olyan táblákat lehet összekapcsolni, amelyek hasonló szekciózással rendelkeznek, szekcióról szekcióra. Így a Postgres kisebb hash-táblákat fog használni. Minden szekció összekapcsolása párhuzamos lehet.
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)A legfontosabb, hogy a szekciók szerinti összekapcsolás párhuzamos csak akkor, ha ezek a szekciók elegendő nagyok.
Párhuzamos kiegészítés — Parallel Append
különböző blokkok helyett használható különböző munkafolyamatokban. Ezt általában az UNION ALL lekérdezéseknél tapasztalhatjuk. Hátrány — kevesebb párhuzamosság, mivel minden munkafolyamat csak 1 lekérdezést dolgoz fel.
Itt 2 munkafolyamat indult el, bár 4 volt bekapcsolva.
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)A legfontosabb változók
- A WORK_MEM korlátozza a memóriát, amely minden folyamat számára elérhető, nemcsak a lekérdezésekhez: work_mem folyamatok kapcsolatok = nagyon sok memória.
- — hány munkafolyamatot használ a végrehajtó program párhuzamos feldolgozáshoz az ütemezésből.
- — a munkafolyamatok összlétszámát a szerveren lévő CPU magok számához igazítja.
- — ugyanaz, de a párhuzamos munkafolyamatokra vonatkozik.
Összegzés
A 9.6 verziótól kezdődően a párhuzamos feldolgozás jelentősen javíthatja a nagy sorok vagy indexek beolvasását végző összetett lekérdezések teljesítményét. A PostgreSQL 10-ben a párhuzamos feldolgozás alapértelmezés szerint be van kapcsolva. Ne feledje el kikapcsolni a nagy OLTP terhelésű szervereken. A szekvenciális beolvasások vagy indexbeolvasások nagyon sok erőforrást igényelnek. Ha nem készít jelentést az összes adathalmazon, a lekérdezések hatékonyabbá tehetők, ha egyszerűen hozzáadja a hiányzó indexeket vagy a megfelelő partícionálást használja.
Hivatkozások
Forrás: habr.com
