🥇Паралелни запитвания в PostgreSQL

В съвременните ЦП има много ядра. Дълго време приложенията изпращаха заявки към бази данни паралелно. Ако това е запитване за множество редове в таблица, то се изпълнява по-бързо, когато използва няколко ЦП, и в PostgreSQL това е възможно от версия 9.6.

Отне 3 години, за да се реализира функцията за паралелни заявки — кодът трябваше да бъде пренаписан на различни етапи от изпълнението на заявките. В PostgreSQL 9.6 се появи инфраструктура за допълнително подобряване на кода. В последващите версии и други типове заявки се изпълняват паралелно.

Ограничения

Не включвайте паралелното изпълнение, ако всички ядра вече са заети, в противен случай други заявки ще забавят.
Най-важното е, че паралелната обработка с висок WORK_MEM заема много памет — всяка хешсъединение или сортиране заема памет в обема на work_mem.
Запитванията OLTP с ниска латентност не могат да бъдат ускорени с паралелно изпълнение. А ако запитването връща един ред, паралелната обработка само ще го забави.
Разработчиците обичат да използват бенчмарк TPC-H. Може би имате подобни заявки за идеално паралелно изпълнение.
Само заявки SELECT без предикатно блокиране се изпълняват паралелно.
Понякога правилната индексация е по-добра от последователното сканиране на таблици в паралелен режим.
Спирането на заявки и курсори не се поддържа.
Окончателните функции и агрегатните функции на подредени набори не са паралелни.
Не печелите нищо в работната натовареност на вход-изход.
Не съществуват паралелни сортировъчни алгоритми. Но запитванията със сортировки могат да се изпълняват паралелно в определени аспекти.
Заменете CTE (WITH …) с вложен SELECT, за да включите паралелната обработка.
Обвивките на трети данни все още не поддържат паралелна обработка (въпреки че биха могли!)
FULL OUTER JOIN не се поддържа.
max_rows деактивира паралелната обработка.
Ако в запитването има функция, която не е маркирана като PARALLEL SAFE, тя ще бъде еднопоточна.
Нивото на изолация на транзакцията SERIALIZABLE деактивира паралелната обработка.

Тестова среда

Разработчиците на PostgreSQL се опитаха да намалят времето за отговор на запитванията от бенчмарка TPC-H. Изтеглете бенчмарка и адаптирайте го за PostgreSQL. Това е неформално използване на бенчмарка TPC-H — не за сравнение на бази данни или оборудване.

Изтеглете TPC-H_Tools_v2.17.3.zip (или по-нова версия) от сайта TPC.
Преименувайте makefile.suite в Makefile и променете, както е описано тук: https://github.com/tvondra/pg_tpch . Компилирайте кода с командата make.
Генерирайте данни: .\/dbgen -s 10 създава база данни от 23 ГБ. Това е достатъчно, за да видите разликата в производителността на паралелни и непаралелни заявки.
Конвертирайте файловете tbl в csv с for и sed.
Клонирайте репото pg_tpch и копирайте файловете csv в pg_tpch\/dss\/data.
Създайте заявки с командата qgen.
Заредете данните в базата с командата .\/tpch.sh.

Паралелно последователно сканиране

Може да е по-бързо не заради паралелното четене, а защото данните са разпръснати по много ядра на ЦП. В съвременните операционни системи файловете с данни на PostgreSQL се кешират добре. С предварителното четене можете да получите от хранилището блок, по-голям от този, който иска демонът PG. Следователно производителността на заявката не е ограничена от входно-изходните операции на диска. Тя консумира цикли на ЦП, за да:

чете редове по един от страниците на таблицата;
сравнява стойностите на редовете и условията WHERE.

Нека изпълним проста заявка 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 ms

Последователното сканиране предоставя твърде много редове без агрегация, така че заявката се изпълнява с едно ядро на ЦП.

Ако добавите SUM(), ще видите, че два работни процеса ще помогнат за ускоряване на заявката:

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 ms

Паралелна агрегация

Възелът «Parallel Seq Scan» произвежда редове за частична агрегация. Възелът «Partial Aggregate» съкращава тези редове с SUM(). В края, броячът SUM от всеки работен процес се събира от възела «Gather».

Крайната сума се изчислява от нодата „Finalize Aggregate“. Ако имате свои функции за агрегация, не забравяйте да ги маркирате като „паралелно безопасни“.

Брой работни процеси

Броят на работните процеси може да бъде увеличен без да е необходимо да се рестартира сървърът:

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 ms

Какво се случва тук? Работните процеси станаха два пъти повече, а запитването стана само 1,6599 пъти по-бързо. Изчисленията са интересни. Имахме 2 работни процеса и 1 лидер. След промяната станаха 4+1.

Нашето максимално ускорение от паралелна обработка: 5/3 = 1,66(6) пъти.

Как работи това?

Процеси

Изпълнението на запитването винаги започва с водещия процес. Лидерът извършва всичко непаралелно и част от паралелната обработка. Другите процеси, които извършват същите запитвания, се наричат работни процеси. Паралелната обработка използва инфраструктурата на динамични фонови работни процеси (от версия 9.4). Тъй като другите части на PostgreSQL използват процеси, а не нишки, запитването с 3 работни процеса може да бъде 4 пъти по-бързо в сравнение с традиционната обработка.

Взаимодействие

Работните процеси общуват с лидера чрез опашка от съобщения (базирана на споделена памет). Всеки процес разполага с 2 опашки: за грешки и за кортежи.

Колко работни процеси са необходими?

Минималното ограничение задава параметърът max_parallel_workers_per_gather. След това изпълнителят на запитвания взема работни процеси от пула, ограничен от параметъра max_parallel_workers size. Последното ограничение е max_worker_processes, тоест общият брой на фоновите процеси.

Ако не успее да спечели работен процес, обработката ще бъде еднопроцесна.

Планиращият запитвания може да намали работните процеси в зависимост от размера на таблицата или индекса. За целта има параметри min_parallel_table_scan_size и min_parallel_index_scan_size.

set min_parallel_table_scan_size='8MB'
8MB таблица => 1 работник
24MB таблица => 2 работници
72MB таблица => 3 работници
x => log(x / min_parallel_table_scan_size) / log(3) + 1 работник

Всеки път, когато таблицата е 3 пъти по-голяма от min_parallel_(index|table)_scan_size, Postgres добавя работен процес. Броят на работните процеси не се основава на разходите. Кръговата зависимост затруднява сложните реализации. Вместо това планиращият използва прости правила.

На практика тези правила не винаги са подходящи за продукцията, така че можете да промените броя на работните процеси за конкретна таблица: ALTER TABLE … SET (parallel_workers = N).

Защо не се използва паралелна обработка?

Освен дългия списък с ограничения, има и проверки за разходите:

parallel_setup_cost — за да се избегне паралелната обработка на кратки запитвания. Този параметър предполага времето за настройка на паметта, стартиране на процеса и начален обмен на данни.

parallel_tuple_cost: комуникацията между водача и работниците може да се забави пропорционално на броя на кортежите от работните процеси. Този параметър изчислява разходите за обмен на данни.

Свързвания на вложени цикли — Nested Loop Join

PostgreSQL 9.6+ може да изпълнява вложени цикли паралелно — това е проста операция.

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)

Събирането става на последния етап, така че Nested Loop Left Join — това е паралелна операция. Parallel Index Only Scan се появи само в версия 10. Той работи подобно на паралелното последователно сканиране. Условието c_custkey = o_custkey четe един ред за всяка клиентска поръчка. Така че то не е паралелно.

Хеш-съединение — Hash Join

Всеки работен процес създава своя хеш-таблица до PostgreSQL 11. И ако тези процеси са повече от четири, производителността не се увеличава. В новата версия хеш-таблицата е обща. Всеки работен процес може да използва WORK_MEM, за да създаде хеш-таблица.

изберете
        l_shipmode,
        сума(случай
                когато o_orderpriority = '1-URGENT'
                        или o_orderpriority = '2-HIGH'
                        тогава 1
                иначе 0
        край) като high_line_count,
        сума(случай
                когато o_orderpriority  '1-URGENT'
                        и o_orderpriority  '2-HIGH'
                        тогава 1
                иначе 0
        край) като low_line_count
от
        поръчки,
        ред
където
        o_orderkey = l_orderkey
        и l_shipmode в ('MAIL', 'AIR')
        и l_commitdate < l_receiptdate
        и l_shipdate = дата '1996-01-01'
        и l_receiptdate < дата '1996-01-01' + интервал '1' година
групиране по
        l_shipmode
поръчка по
        l_shipmode
LIMIT 1;
                                                                                                                                    ПЛАН ЗА ЗАПИТВАНЕ
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 Limit  (cost=1964755.66..1964961.44 rows=1 width=27) (actual time=7579.592..7922.997 rows=1 loops=1)
   ->  Финализиране на GroupAggregate  (cost=1964755.66..1966196.11 rows=7 width=27) (actual time=7579.590..7579.591 rows=1 loops=1)
         Групов Ключ: lineitem.l_shipmode
         ->  Събиране на Merge  (cost=1964755.66..1966195.83 rows=28 width=27) (actual time=7559.593..7922.319 rows=6 loops=1)
               Планирани Работници: 4
               Стартирани Работници: 4
               ->  Частично GroupAggregate  (cost=1963755.61..1965192.44 rows=7 width=27) (actual time=7548.103..7564.592 rows=2 loops=5)
                     Групов Ключ: lineitem.l_shipmode
                     ->  Сортиране  (cost=1963755.61..1963935.20 rows=71838 width=27) (actual time=7530.280..7539.688 rows=62519 loops=5)
                           Сортиращ Ключ: lineitem.l_shipmode
                           Сортиращ Метод: външно сливане  Диск: 2304kB
                           Работник 0:  Сортиращ Метод: външно сливане  Диск: 2064kB
                           Работник 1:  Сортиращ Метод: външно сливане  Диск: 2384kB
                           Работник 2:  Сортиращ Метод: външно сливане  Диск: 2264kB
                           Работник 3:  Сортиращ Метод: външно сливане  Диск: 2336kB
                           ->  Паралелно Хеш Съединение  (cost=382571.01..1957960.99 rows=71838 width=27) (actual time=7036.917..7499.692 rows=62519 loops=5)
                                 Хеш Условие: (lineitem.l_orderkey = orders.o_orderkey)
                                 ->  Паралелно Последователно Сканиране на lineitem  (cost=0.00..1552386.40 rows=71838 width=19) (actual time=0.583..4901.063 rows=62519 loops=5)
                                       Филтър: ((l_shipmode = ANY ('{MAIL,AIR}'::bpchar[])) И (l_commitdate < l_receiptdate) И (l_shipdate = '1996-01-01'::date) И (l_receiptdate < '1997-01-01 00:00:00'::timestamp без времева зона))
                                       Редове Премахнати от Филтър: 11934691
                                 ->  Паралелно Хеш  (cost=313722.45..313722.45 rows=3750045 width=20) (actual time=2011.518..2011.518 rows=3000000 loops=5)
                                       Броя: 65536  Пакети: 256  Използване на Памет: 3840kB
                                       ->  Паралелно Последователно Сканиране на orders  (cost=0.00..313722.45 rows=3750045 width=20) (actual time=0.029..995.948 rows=3000000 loops=5)
 Време за Планиране: 0.977 ms
 Време за Изпълнение: 7923.770 ms

Запитването 12 от TPC-H наглядно показва паралелно хеш-съединение. Всеки работник участва в създаването на обща хеш-таблица.

Сливане на свързване — Merge Join

Свързването по сливане е по природа непаралелно. Не се притеснявайте, ако това е последният етап на заявката, — той все пак може да се изпълнява паралелно.

-- Запит 2 от TPC-H
обясни (разходи изключени) избери s_acctbal, s_name, n_name, p_partkey, p_mfgr, s_address, s_phone, s_comment
от    part, supplier, partsupp, nation, region
където
        p_partkey = ps_partkey
        и s_suppkey = ps_suppkey
        и p_size = 36
        и p_type подобно '%BRASS'
        и s_nationkey = n_nationkey
        и n_regionkey = r_regionkey
        и r_name = 'AMERICA'
        и ps_supplycost = (
                избери
                        min(ps_supplycost)
                от    partsupp, supplier, nation, region
                където
                        p_partkey = ps_partkey
                        и s_suppkey = ps_suppkey
                        и s_nationkey = n_nationkey
                        и n_regionkey = r_regionkey
                        и r_name = 'AMERICA'
        )
редактиране по s_acctbal низходящо, n_name, s_name, p_partkey
LIMIT 100;
                                                ПЛАН ЗА ЗАПИТ
----------------------------------------------------------------------------------------------------------
 Ограничение
   -&gt;  Сортиране
         Ключ за сортиране: supplier.s_acctbal НИЗХОДЯЩО, nation.n_name, supplier.s_name, part.p_partkey
         -&gt;  Слияние на присъединяване
               Условие за сливане: (part.p_partkey = partsupp.ps_partkey)
               Филтър за присъединяване: (partsupp.ps_supplycost = (SubPlan 1))
               -&gt;  Събиране на слияние
                     Планирани работници: 4
                     -&gt;  Паралелно сканиране на индекс с <strong>part_pkey</strong> на part
                           Филтър: (((p_type)::text ~~ '%BRASS'::text) И (p_size = 36))
               -&gt;  Материализиране
                     -&gt;  Сортиране
                           Ключ за сортиране: partsupp.ps_partkey
                           -&gt;  Вложена циклична структура
                                 -&gt;  Вложена циклична структура
                                       Филтър за присъединяване: (nation.n_regionkey = region.r_regionkey)
                                       -&gt;  Последователно сканиране на region
                                             Филтър: (r_name = 'AMERICA'::bpchar)
                                       -&gt;  Хеш-присъединяване
                                             Условие на хеша: (supplier.s_nationkey = nation.n_nationkey)
                                             -&gt;  Последователно сканиране на supplier
                                             -&gt;  Хеш
                                                   -&gt;  Последователно сканиране на nation
                                 -&gt;  Индексно сканиране с idx_partsupp_suppkey на partsupp
                                       Условие на индекса: (ps_suppkey = supplier.s_suppkey)
               SubPlan 1
                 -&gt;  Агресор
                       -&gt;  Вложена циклична структура
                             Филтър за присъединяване: (nation_1.n_regionkey = region_1.r_regionkey)
                             -&gt;  Последователно сканиране на region region_1
                                   Филтър: (r_name = 'AMERICA'::bpchar)
                             -&gt;  Вложена циклична структура
                                   -&gt;  Вложена циклична структура
                                         -&gt;  Индексно сканиране с idx_partsupp_partkey на partsupp partsupp_1
                                               Условие на индекса: (part.p_partkey = ps_partkey)
                                         -&gt;  Индексно сканиране с supplier_pkey на supplier supplier_1
                                               Условие на индекса: (s_suppkey = partsupp_1.ps_suppkey)
                                   -&gt;  Индексно сканиране с nation_pkey на nation nation_1
                                         Условие на индекса: (n_nationkey = supplier_1.s_nationkey)

Възелът „Merge Join“ е над „Gather Merge“. Така че сливането не използва паралелна обработка. Но възелът „Parallel Index Scan“ все пак помага на сегмента. part_pkey.

Свързване по секции

В PostgreSQL 11 свързване по секции по подразбиране е отключено: то има много скъпо планиране. Таблици с подобно секциониране могат да бъдат свързвани секция по секция. По този начин Postgres ще използва по-малки хеш-таблици. Всяко свързване на секции може да бъде паралелно.

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)

Основното е, че свързването по секции може да бъде паралелно, само ако тези секции са достатъчно големи.

Паралелно допълнение — Parallel Append

Parallel Append може да се използва вместо различни блокове в различни работни процеси. Обикновено това се случва с заявки UNION ALL. Недостатъкът е по-малкото паралелизъм, тъй като всеки работен процес обработва само 1 заявка.

Тук са стартирани 2 работни процеса, въпреки че са включени 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   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)

Най-важните променливи

WORK_MEM ограничава обема памет за всеки процес, не само за заявки: work_mem процеси съединения = много памет.
max_parallel_workers_per_gather — колко работни процеса изпълняващата програма ще използва за паралелна обработка от плана.
max_worker_processes — коригира общия брой работни процеси спрямо броя на ядрата на ЦП в сървъра.
max_parallel_workers — същото, но за паралелни работни процеси.

Итог

Започвайки от версия 9.6, паралелната обработка може сериозно да подобри производителността на сложни запитвания, които сканират много редове или индекси. В PostgreSQL 10 паралелната обработка е включена по подразбиране. Не забравяйте да я изключите на сървъри с голяма натовареност OLTP. Последователните сканирания или сканирания на индекси консумират много ресурси. Ако не извършвате отчет по целия набор от данни, запитванията могат да станат по-продуктивни, просто като добавите липсващи индекси или използвате правилното секциониране.