并行查询的执行 · k8s原理与实践

[TOC] # 并行查询的执行并行执行（Parallel Execution）是将一个较大的任务切分为多个较小的任务，启动多个线程或者进程来并行处理这些小任务，这样可以利用更多的 CPU 与 IO 资源来缩短操作的响应时间。并行执行分为并行查询（Parallel Query）、并行 DDL（Parallel DDL）和并行 DML（Parallel DML）。目前 OceanBase 数据库仅支持并行查询，并行 DDL 与并行 DML 还未支持。启动并行查询的方式有以下两种： * 通过 PARALLEL HINT 指定并行度（dop）的方式启动并行查询。 * 针对查询分区数大于 1 的分区表会自动启动并行查询。 ## 启用分区表并行查询针对分区表的查询，如果查询的目标分区数大于 1，系统会自动启用并行查询，dop 的值由系统默认指定为 1。如下例所示，创建一个分区表 ptable，对 ptable 进行全表数据的扫描操作，通过 EXPLAIN 命令查看生成的执行计划。 ~~~ obclient>CREATE TABLE PTABLE(c1 INT , c2 INT) PARTITION BY HASH(c1) PARTITIONS 16; Query OK, 0 rows affected (0.12 sec) obclient>EXPLAIN SELECT * FROM ptable\G; *************************** 1. row *************************** Query Plan: ======================================================= |ID|OPERATOR |NAME |EST. ROWS|COST | ------------------------------------------------------- |0 |EXCHANGE IN DISTR | |1600000 |1246946| |1 | EXCHANGE OUT DISTR |:EX10000|1600000 |1095490| |2 | PX PARTITION ITERATOR| |1600000 |1095490| |3 | TABLE SCAN |ptable |1600000 |1095490| ======================================================= Outputs & filters: ------------------------------------- 0 - output([ptable.c1], [ptable.c2]), filter(nil) 1 - output([ptable.c1], [ptable.c2]), filter(nil), dop=1 2 - output([ptable.c1], [ptable.c2]), filter(nil) 3 - output([ptable.c1], [ptable.c2]), filter(nil), access([ptable.c1], [ptable.c2]), partitions(p[0-15]) ~~~ 通过执行计划可以看出，分区表默认的并行查询的 dop 为 1。如果 OceanBase 集群一共有 3 个 OBServer，表 ptable 的 16 个分区分散在 3 个 OBServer 中，那么每一个 OBServer 都会启动一个工作线程（Worker Thread）来执行分区数据的扫描工作，一共需要启动 3 个工作线程来执行表的扫描工作。针对分区表，添加 PARALLEL HINT 启动并行查询，并指定 dop，通过 EXPLAIN 命令查看生成的执行计划。 ~~~ obclient>EXPLAIN SELECT /*+ PARALLEL(8) */ * FROM ptable\G; *************************** 1. row *************************** Query Plan: ======================================================= |ID|OPERATOR |NAME |EST. ROWS|COST | ------------------------------------------------------- |0 |EXCHANGE IN DISTR | |1600000 |1246946| |1 | EXCHANGE OUT DISTR |:EX10000|1600000 |1095490| |2 | PX PARTITION ITERATOR| |1600000 |1095490| |3 | TABLE SCAN |ptable |1600000 |1095490| ======================================================= Outputs & filters: ------------------------------------- 0 - output([ptable.c1], [ptable.c2]), filter(nil) 1 - output([ptable.c1], [ptable.c2]), filter(nil), dop=8 2 - output([ptable.c1], [ptable.c2]), filter(nil) 3 - output([ptable.c1], [ptable.c2]), filter(nil), access([ptable.c1], [ptable.c2]), partitions(p[0-15]) ~~~ 通过执行计划可以看出，并行查询的 dop 为 8。如果查询分区所在的 OBServer 的个数小于等于 dop，那么工作线程（总个数等于 dop）会按照一定的策略分配到涉及的 OBServer 上；如果查询分区所在的 OBServer 的个数大于 dop，那么每一个 OBServer 都会至少启动一个工作线程，一共需要启动的工作线程的数目会大于 dop。例如，当`dop=8`，如果 16 个分区均匀的分布在 4 台 OBServer 节点上，那么每一个 OBServer 上都会启动 2 个工作线程来扫描其对应的分区（一共启动 8 个工作线程）；如果 16 个分区分布在 16 台 OBServer 节点上（每一个节点一个分区），那么每一台 OBServer 上都会启动 1 个工作线程来扫描其对应的分区（一共启动 16 个工作线程）。如果针对分区表的查询，查询分区数目小于等于 1，系统不会启动并行查询。如下例所示，对 ptable 的查询添加一个过滤条件`c1=1`。 ~~~ obclient>EXPLAIN SELECT * FROM ptable WHERE c1 = 1\G; *************************** 1. row *************************** Query Plan: ====================================== |ID|OPERATOR |NAME |EST. ROWS|COST | -------------------------------------- |0 |TABLE SCAN|ptable|990 |85222| ====================================== Outputs & filters: ------------------------------------- 0 - output([ptable.c1], [ptable.c2]), filter([ptable.c1 = 1]), access([ptable.c1], [ptable.c2]), partitions(p1) ~~~ 通过计划可以看出，查询的目标分区个数为 1，系统没有启动并行查询。如果希望针对一个分区的查询也能够进行并行执行，就只能通过添加 PARALLEL HINT 的方式进行分区内并行查询，通过 EXPLAIN 命令查看生成的执行计划。 ~~~ obclient>EXPLAIN SELECT /*+ PARALLEL(8) */ * FROM ptable WHERE c1 = 1\G; *************************** 1. row *************************** Query Plan: ================================================= |ID|OPERATOR |NAME |EST. ROWS|COST | ------------------------------------------------- |0 |EXCHANGE IN DISTR | |990 |85316| |1 | EXCHANGE OUT DISTR|:EX10000|990 |85222| |2 | PX BLOCK ITERATOR| |990 |85222| |3 | TABLE SCAN |ptable |990 |85222| ================================================= Outputs & filters: ------------------------------------- 0 - output([ptable.c1], [ptable.c2]), filter(nil) 1 - output([ptable.c1], [ptable.c2]), filter(nil), dop=8 2 - output([ptable.c1], [ptable.c2]), filter(nil) 3 - output([ptable.c1], [ptable.c2]), filter([ptable.c1 = 1]), access([ptable.c1], [ptable.c2]), partitions(p1) ~~~ **注意** * 如果希望在查询分区数等于 1 的情况下，能够采用 HINT 的方式进行分区内并行查询，需要对应的 dop 的值大于等于 2。 * 如果 dop 的值为空或者小于 2 将不启动并行查询。 ## 启用非分区表并行查询非分区表本质上是只有 1 个分区的分区表，因此针对非分区表的查询，只能通过添加 PARALLEL HINT 的方式启动分区内并行查询，否则不会启动并行查询。如下例所示，创建一个非分区表 stable，对 stable 进行全表数据的扫描操作，通过 EXPLAIN 命令查看生成的执行计划。 ~~~ obclient>CREATE TABLE stable(c1 INT, c2 INT); Query OK, 0 rows affected (0.12 sec) obclient>EXPLAIN SELECT * FROM stable\G; *************************** 1. row *************************** Query Plan: ====================================== |ID|OPERATOR |NAME |EST. ROWS|COST | -------------------------------------- |0 |TABLE SCAN|stable|100000 |68478| ====================================== Outputs & filters: ------------------------------------- 0 - output([stable.c1], [stable.c2]), filter(nil), access([stable.c1], [stable.c2]), partitions(p0) ~~~ 通过执行计划可以看出，非分区表不使用 HINT 的情况下，不会启动并行查询。针对非分区表，添加 PARALLEL HINT 启动分区内并行查询，并指定 dop（大于等于 2），通过 EXPLAIN 命令查看生成的执行计划。 ~~~ obclient>EXPLAIN SELECT /*+ PARALLEL(4)*/ * FROM stable\G; *************************** 1. row *************************** Query Plan: ================================================= |ID|OPERATOR |NAME |EST. ROWS|COST | ------------------------------------------------- |0 |EXCHANGE IN DISTR | |100000 |77944| |1 | EXCHANGE OUT DISTR|:EX10000|100000 |68478| |2 | PX BLOCK ITERATOR| |100000 |68478| |3 | TABLE SCAN |stable |100000 |68478| ================================================= Outputs & filters: ------------------------------------- 0 - output([stable.c1], [stable.c2]), filter(nil) 1 - output([stable.c1], [stable.c2]), filter(nil), dop=4 2 - output([stable.c1], [stable.c2]), filter(nil) 3 - output([stable.c1], [stable.c2]), filter(nil), access([stable.c1], [stable.c2]), partitions(p0) ~~~ ## 启用多表并行查询在查询中，多表 JOIN 查询最为常见。如下例所示，首先创建两张分区表 p1table 和 p2table： ~~~ obclient>CREATE TABLE p1table(c1 INT ,c2 INT) PARTITION BY HASH(c1) PARTITIONS 2; Query OK, 0 rows affected (0.02 sec) obclient>CREATE TABLE p2table(c1 INT ,c2 INT) PARTITION BY HASH(c1) PARTITIONS 4; Query OK, 0 rows affected (0.02 sec) ~~~ 查询 p1table 与 p2table 的 JOIN 结果，JOIN 条件是`p1table.c1=p2table.c2`，执行计划如下： ~~~ obclient>EXPLAIN SELECT * FROM p1table p1 JOIN p2table p2 ON p1.c1=p2.c2\G; *************************** 1. row *************************** Query Plan: ==================================================================== |ID|OPERATOR |NAME |EST. ROWS|COST | -------------------------------------------------------------------- |0 |EXCHANGE IN DISTR | |784080000|614282633| |1 | EXCHANGE OUT DISTR |:EX10001|784080000|465840503| |2 | HASH JOIN | |784080000|465840503| |3 | EXCHANGE IN DISTR | |200000 |155887 | |4 | EXCHANGE OUT DISTR (BROADCAST)|:EX10000|200000 |136955 | |5 | PX PARTITION ITERATOR | |200000 |136955 | |6 | TABLE SCAN |p1 |200000 |136955 | |7 | PX PARTITION ITERATOR | |400000 |273873 | |8 | TABLE SCAN |p2 |400000 |273873 | ==================================================================== Outputs & filters: ------------------------------------- 0 - output([p1.c1], [p1.c2], [p2.c1], [p2.c2]), filter(nil) 1 - output([p1.c1], [p1.c2], [p2.c1], [p2.c2]), filter(nil), dop=1 2 - output([p1.c1], [p1.c2], [p2.c1], [p2.c2]), filter(nil), equal_conds([p1.c1 = p2.c2]), other_conds(nil) 3 - output([p1.c1], [p1.c2]), filter(nil) 4 - output([p1.c1], [p1.c2]), filter(nil), dop=1 5 - output([p1.c1], [p1.c2]), filter(nil) 6 - output([p1.c1], [p1.c2]), filter(nil), access([p1.c1], [p1.c2]), partitions(p[0-1]) 7 - output([p2.c1], [p2.c2]), filter(nil) 8 - output([p2.c1], [p2.c2]), filter(nil), access([p2.c1], [p2.c2]), partitions(p[0-3]) ~~~ 默认情况下针对 p1table 与 p2table（两张表需要查询的分区数都大于 1）都会采用并行查询，默认的 dop 为 1。同样，也可以通过使用 PARALLEL HINT 的方式来改变并行度。如下例所示，改变 JOIN 的条件为`p1table.c1=p2table.c2`和`p2table.c1=1`，这样针对 p2table 仅仅会选择单个分区，执行计划如下所示： ~~~ obclient>EXPLAIN SELECT * FROM p1table p1 JOIN p2table p2 ON p1.c1=p2.c2 AND p2.c1=1\G; *************************** 1. row *************************** Query Plan: ============================================================= |ID|OPERATOR |NAME |EST. ROWS|COST | ------------------------------------------------------------- |0 |EXCHANGE IN DISTR | |1940598 |1807515| |1 | EXCHANGE OUT DISTR |:EX10001|1940598 |1440121| |2 | HASH JOIN | |1940598 |1440121| |3 | EXCHANGE IN DISTR | |990 |85316 | |4 | EXCHANGE OUT DISTR (PKEY)|:EX10000|990 |85222 | |5 | TABLE SCAN |p2 |990 |85222 | |6 | PX PARTITION ITERATOR | |200000 |136955 | |7 | TABLE SCAN |p1 |200000 |136955 | ============================================================= Outputs & filters: ------------------------------------- 0 - output([p1.c1], [p1.c2], [p2.c1], [p2.c2]), filter(nil) 1 - output([p1.c1], [p1.c2], [p2.c1], [p2.c2]), filter(nil), dop=1 2 - output([p1.c1], [p1.c2], [p2.c1], [p2.c2]), filter(nil), equal_conds([p1.c1 = p2.c2]), other_conds(nil) 3 - output([p2.c1], [p2.c2]), filter(nil) 4 - (#keys=1, [p2.c2]), output([p2.c1], [p2.c2]), filter(nil), dop=1 5 - output([p2.c1], [p2.c2]), filter([p2.c1 = 1]), access([p2.c1], [p2.c2]), partitions(p1) 6 - output([p1.c1], [p1.c2]), filter(nil) 7 - output([p1.c1], [p1.c2]), filter(nil), access([p1.c1], [p1.c2]), partitions(p[0-1]) ~~~ 通过计划可以看出，p2table 仅需要扫描一个分区，在默认情况下不进行并行查询；p1table 需要扫描两个分区，默认情况下进行并行查询。同样，也可以通过添加 PARALLEL HINT 的方式改变并行度，使 p2table 针对一个分区的查询变为分区内并行查询。 ## 关闭并行查询分区表在查询的时候会自动启动并行查询（查询分区个数大于 1），如果不想启动并行查询，可以使用添加 HINT`/*+ NO_USE_PX */`来关闭并行查询。例如，针对分区表 ptable，添加 HINT`/*+ NO_USE_PX */`来关闭并行查询，通过生成的执行计划可以看出对 ptable 表的扫描没有进行并行查询。 ~~~ obclient>EXPLAIN SELECT /*+ NO_USE_PX */ * FROM ptable\G; *************************** 1. row *************************** Query Plan: ================================================= |ID|OPERATOR |NAME |EST. ROWS|COST | ------------------------------------------------- |0 |EXCHANGE IN DISTR | |1600000 |1246946| |1 | EXCHANGE OUT DISTR| |1600000 |1095490| |2 | TABLE SCAN |ptable|1600000 |1095490| ================================================= Outputs & filters: ------------------------------------- 0 - output([ptable.c1], [ptable.c2]), filter(nil) 1 - output([ptable.c1], [ptable.c2]), filter(nil) 2 - output([ptable.c1], [ptable.c2]), filter(nil), access([ptable.c1], [ptable.c2]), partitions(p[0-15]) ~~~ ## 并行执行相关的系统视图 OceanBase 数据库提供了系统视图`gv$sql_audit/v$sql_audit`来查看并行执行的运行状态以及一些统计信息。 `gv$sql_audit/v$sql_audit`包含字段较多，其中与并行执行相关的字段为：qc\_id、dfo\_id、sqc\_id 和 worker\_id。详细信息请参考[(g)v$sql\_audit 介绍](https://open.oceanbase.com/docs/community/oceanbase-database/V3.1.0/g-v-sql_audit-introduction-1)。