🥇PostgreSQL 中的并行查询

现代 CPU 有很多核心。多年来，应用程序一直向数据库并行发送查询。如果是对表中多行的报表查询，使用多个CPU时运行速度会更快，PostgreSQL从9.6版本开始就可以做到这一点。

并行查询功能的实现花了三年时间——我们必须在查询执行的不同阶段重写代码。 PostgreSQL 3 引入了基础设施来进一步改进代码。在后续版本中，其他类型的查询是并行执行的。

限制

如果所有核心都已繁忙，请勿启用并行执行，否则其他请求会减慢。
最重要的是，具有高 WORK_MEM 值的并行处理会使用大量内存 - 每个哈希连接或排序都会占用 work_mem 内存。
低延迟 OLTP 查询无法通过并行执行来加速。如果查询返回一行，并行处理只会减慢速度。
开发人员喜欢使用 TPC-H 基准测试。也许您对完美并行执行有类似的查询。
仅并行执行没有谓词锁定的 SELECT 查询。
有时，正确的索引比并行模式下的顺序表扫描更好。
不支持暂停查询和游标。
窗口函数和有序集聚合函数不并行。
您不会从 I/O 工作负载中获得任何好处。
没有并行排序算法。但在某些方面，带有排序的查询可以并行执行。
将 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 GB 的数据库。这足以看出并行和非并行查询的性能差异。
转换文件 tbl в csv с for и sed.
克隆存储库 pg_tpch 并复制文件 csv в pg_tpch/dss/data.
使用命令创建查询 qgen.
使用命令将数据加载到数据库中 ./tpch.sh.

并行顺序扫描

它可能更快不是因为并行读取，而是因为数据分布在许多 CPU 核心上。在现代操作系统中，PostgreSQL 数据文件被很好地缓存。通过预读，可以从存储中获取比 PG 守护进程请求更大的块。因此，查询性能不受磁盘 I/O 的限制。它消耗 CPU 周期来：

从表页中一次读取一行；
比较字符串值和条件 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

顺序扫描产生太多行而没有聚合，因此查询由单个 CPU 核心执行。

如果添加 SUM()，您可以看到两个工作流程将有助于加快查询速度：

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 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

并行聚合

并行顺序扫描节点生成用于部分聚合的行。 “部分聚合”节点使用以下方式修剪这些行 SUM()。最后，“Gather”节点收集每个工作进程的 SUM 计数器。

最终结果由“Finalize Aggregate”节点计算。如果您有自己的聚合函数，请不要忘记将它们标记为“并行安全”。

工作进程数

可以在不重新启动服务器的情况下增加工作进程的数量：

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 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

这里发生了什么？工作进程增加了 2 倍，而请求速度仅提高了 1,6599 倍。计算很有趣。我们有 2 个工作进程和 1 个领导进程。改后变成了4+1。

并行处理的最大加速比：5/3 = 1,66(6) 倍。

它是如何工作的呢？

流程

请求执行始终从主导进程开始。领导者执行所有非并行处理和一些并行处理。执行相同请求的其他进程称为工作进程。并行处理使用基础设施动态后台工作进程（从版本 9.4 开始）。由于 PostgreSQL 的其他部分使用进程而不是线程，因此具有 3 个工作进程的查询可能比传统处理快 4 倍。

相互作用

Worker进程通过消息队列（基于共享内存）与Leader进程通信。每个进程有 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 table => 1 worker
24MB table => 2 workers
72MB table => 3 workers
x => log(x / min_parallel_table_scan_size) / log(3) + 1 worker

每次桌子都比原来大3倍 min_parallel_(index|table)_scan_size，Postgres添加了一个工作进程。工作流程的数量并不基于成本。循环依赖使得复杂的实现变得困难。相反，规划者使用简单的规则。

实际上，这些规则并不总是适合生产，因此您可以更改特定表的工作进程数量： ALTER TABLE ... SET (parallel_workers = N).

为什么不使用并行处理？

除了一长串的限制之外，还有成本检查：

parallel_setup_cost - 避免并行处理短请求。该参数估计准备内存、启动进程和初始数据交换的时间。

parallel_tuple_cost：领导者和工作人员之间的通信可能会根据工作流程中的元组数量按比例延迟。该参数计算数据交换的成本。

嵌套循环连接

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 '%special%deposits%'
                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 !~~ '%special%deposits%'::text)

收集发生在最后阶段，因此 Nested Loop Left Join 是一个并行操作。仅并行索引扫描仅在版本 10 中引入。它的工作原理与并行串行扫描类似。健康）状况 c_custkey = o_custkey 每个客户字符串读取一个订单。所以它不是平行的。

哈希连接

每个工作进程都会创建自己的哈希表，直到 PostgreSQL 11。如果这些进程超过四个，性能将不会提高。在新版本中，哈希表是共享的。每个工作进程都可以使用 WORK_MEM 创建哈希表。

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 < l_commitdate
        and l_receiptdate >= 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 < 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 ms

TPC-H 的查询 12 清楚地显示了并行哈希连接。每个工作进程都有助于创建一个公共哈希表。

合并连接

合并联接本质上是非并行的。如果这是查询的最后一步，请不要担心 - 它仍然可以并行运行。

-- Query 2 from TPC-H
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
                     ->  Parallel Index Scan using <strong>part_pkey</strong> on part
                           Filter: (((p_type)::text ~~ '%BRASS'::text) AND (p_size = 36))
               ->  Materialize
                     ->  Sort
                           Sort Key: partsupp.ps_partkey
                           ->  Nested Loop
                                 ->  Nested Loop
                                       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
                 ->  Aggregate
                       ->  Nested Loop
                             Join Filter: (nation_1.n_regionkey = region_1.r_regionkey)
                             ->  Seq Scan on region region_1
                                   Filter: (r_name = 'AMERICA'::bpchar)
                             ->  Nested Loop
                                   ->  Nested Loop
                                         ->  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)

“Merge Join”节点位于“Gather Merge”之上。所以合并不使用并行处理。但“并行索引扫描”节点仍然有助于该段 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)

最重要的是，只有当这些部分足够大时，这些部分的连接才是并行的。

并行追加

并行追加可以在不同的工作流程中使用不同的块来代替。这通常发生在 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 <= date '2000-12-01' - interval '105' day;
                                           QUERY PLAN
------------------------------------------------------------------------------------------------
 Gather
   Workers Planned: 2
   ->  Parallel Append
         ->  Aggregate
               ->  Seq Scan on lineitem
                     Filter: (l_shipdate <= '2000-08-18 00:00:00'::timestamp without time zone)
         ->  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 — 将工作进程总数调整为服务器上的 CPU 核心数。
max_parallel_workers - 相同，但适用于并行工作流程。

结果

从版本 9.6 开始，并行处理可以极大地提高扫描许多行或索引的复杂查询的性能。在 PostgreSQL 10 中，默认启用并行处理。请记住在具有大量 OLTP 工作负载的服务器上禁用它。顺序扫描或索引扫描会消耗大量资源。如果您没有对整个数据集运行报告，则可以通过简单地添加缺失的索引或使用适当的分区来提高查询性能。

引用

来源： habr.com