FQS：一种神奇的数仓查询优化技术
本文分享自华为云社区《根据执行计划优化SQL【绽放吧！GaussDB(DWS)云原生数仓】》，作者：西岭雪山。
引言
如果您刚接触DWS那一定会好奇想要知道"REMOTE_FQS_QUERY" 到底代表什么意思？我们看官网的描述是代表这执行计划已经CN直接将原语句下发到DN，各DN单独执行，并将执行结果在CN上进行汇总。且不需要做过多的调整了，真的是这样吗？
FQS计划，完全下推
两表JOIN，且其连接条件为各表的分布列，在关闭stream算子的情况下，CN会直接将该语句发送至各DN执行，最后结果在CN汇总。
SET enable_stream_operator=off; SET explain_perf_mode=normal; EXPLAIN (VERBOSE on,COSTS off) SELECT * FROM tt01,tt02 WHERE tt01.c1=tt02.c2; QUERY PLAN ------------------------------------------------------------------------------------------------------------------- Data Node Scan on "__REMOTE_FQS_QUERY__" Output: tt01.c1, tt01.c2, tt02.c1, tt02.c2 Node/s: All datanodes Remote query: SELECT tt01.c1, tt01.c2, tt02.c1, tt02.c2 FROM dbadmin.tt01, dbadmin.tt02 WHERE tt01.c1 = tt02.c2 (4 rows)
像上面的执行计划只显示了Data Node Scan on "__REMOTE_FQS_QUERY__"，这样的执行计划太过粗糙，不知道内部是如何执行的，是否走了索引等更为详细的信息。
下面我们建表进行验证
create table t5 (bh varchar(300),bh2 varchar(300),c_name varchar(300),c_info varchar(300))distribute by hash(bh); insert into t4 select uuid_generate_v1(), uuid_generate_v1(),'测试','sdfffffffffffffffsdf' from generate_series(1,50000); insert into t4 select * from t4; --1、没有索引的情况下： postgres=# explain analyze select * from t4 where bh2 = '652e4e0e-ba60-0400-25b5-4ee5e490fffe'; QUERY PLAN ----------------------------------------------------------------------------------------------------------------------------- id | operation | A-time | A-rows | E-rows | Peak Memory | A-width | E-width | E-costs ----+----------------------------------------------+---------+--------+--------+-------------+---------+---------+--------- 1 | -> Data Node Scan on "__REMOTE_FQS_QUERY__" | 256.364 | 32 | 0 | 56KB | | 0 | 0.00 ====== Query Summary ===== ----------------------------------------- Coordinator executor start time: 0.055 ms Coordinator executor run time: 256.410 ms Coordinator executor end time: 0.010 ms Planner runtime: 0.145 ms Query Id: 73746443917091633 Total runtime: 256.557 ms (12 rows) Time: 259.051 ms --2、添加索引，并添加hint indexscan postgres=# create index i_t4 on t4(bh2); CREATE INDEX Time: 3328.258 ms postgres=# explain analyze select /*+ indexscan(t4 i_t4) */ * from t4 where bh2 = '652e4e0e-ba60-0400-25b5-4ee5e490fffe'; QUERY PLAN ---------------------------------------------------------------------------------------------------------------------------- id | operation | A-time | A-rows | E-rows | Peak Memory | A-width | E-width | E-costs ----+----------------------------------------------+--------+--------+--------+-------------+---------+---------+--------- 1 | -> Data Node Scan on "__REMOTE_FQS_QUERY__" | 2.269 | 32 | 0 | 56KB | | 0 | 0.00 ====== Query Summary ===== ----------------------------------------- Coordinator executor start time: 0.027 ms Coordinator executor run time: 2.298 ms Coordinator executor end time: 0.009 ms Planner runtime: 0.074 ms Query Id: 73746443917091930 Total runtime: 2.401 ms (12 rows)
可以看到没有创建索引的时候执行计划和创建索引的执行计划完全一样，但是执行的时间是259.051ms和2.401ms，相差非常明显，很可能第二个执行计划已经走索引了，但是执行计划一样，这对于优化人员不够直观。
即使在执行计划中加入了 /*+ indexscan(t4 i_t4) */，但并没有打印出是否走了索引，执行计划过于简洁，并且pg_stat_all_indexes中业务表的所有统计信息都是0，也没发判断。
CPUTime
对于上面的时间区别也可以用CPU耗时对比，在执行计划中加入CPU的耗时：
--没有索引的执行计划 postgres=# explain (analyze,buffers,verbose,cpu,nodes )select * from t4 where bh2 = '652e4e0e-ba60-0400-25b5-4ee5e490fffe'; QUERY PLAN --------------------------------------------------------------------------------------------------------------------------- Data Node Scan on "__REMOTE_FQS_QUERY__" (cost=0.00..0.00 rows=0 width=0) (actual time=244.096..244.108 rows=32 loops=1) Output: t4.bh, t4.bh2, t4.c_name, t4.c_info Node/s: All datanodes Remote query: SELECT bh, bh2, c_name, c_info FROM sa.t4 WHERE bh2::text = '652e4e0e-ba60-0400-25b5-4ee5e490fffe'::text (CPU: ex c/r=762829, ex row=32, ex cyc=24410534, inc cyc=24410534) Total runtime: 244.306 ms (6 rows) --创建索引后的执行计划 postgres=# explain (analyze,buffers,verbose,cpu,nodes )select * from t4 where bh2 = '652e4e0e-ba60-0400-25b5-4ee5e490fffe'; QUERY PLAN -------------------------------------------------------------------------------------------------------------------------- Data Node Scan on "__REMOTE_FQS_QUERY__" (cost=0.00..0.00 rows=0 width=0) (actual time=1.035..2.148 rows=32 loops=1) Output: t4.bh, t4.bh2, t4.c_name, t4.c_info Node/s: All datanodes Remote query: SELECT bh, bh2, c_name, c_info FROM sa.t4 WHERE bh2::text = '652e4e0e-ba60-0400-25b5-4ee5e490fffe'::text (CPU: ex c/r=6698, ex row=32, ex cyc=214354, inc cyc=214354) Total runtime: 2.242 ms (6 rows)
对比执行计划可以看到是一样的。
其中cyc代表的是CPU的周期数，ex cyc表示的是当前算子的周期数，不包含其子节点；inc cyc是包含子节点的周期数；ex row是当前算子输出的数据行数；ex c/r则是ex cyc/ex row得到的每条数据所用的平均周期数。
cpu平均周期对比：没索引：762829，创建索引后：6698，大约是一百多倍。
查看详细计划
__REMOTE_FQS_QUERY__是直接将语句发送给了nodedata，所以cn节点不生成执行计划，所以没法看到是否走索引，如果我们将enable_fast_query_shipping关闭，就能在cn上面生成执行计划，可以看到是否走了索引。
--关闭fast_query postgres=# set enable_fast_query_shipping to off; postgres=# set explain_perf_mode=normal; --走索引的执行计划 postgres=# explain analyze select * from t4 where bh2 = '652e4e0e-ba60-0400-25b5-4ee5e490fffe'; QUERY PLAN ------------------------------------------------------------------------------------------------------------------------------ Streaming (type: GATHER) (cost=4.95..51.75 rows=31 width=102) (actual time=1.695..2.263 rows=32 loops=1) Node/s: All datanodes -> Bitmap Heap Scan on t4 (cost=4.33..43.75 rows=31 width=102) (actual time=[0.040,0.040]..[0.057,0.153], rows=32) Recheck Cond: ((bh2)::text = '652e4e0e-ba60-0400-25b5-4ee5e490fffe'::text) -> Bitmap Index Scan on i_t4 (cost=0.00..4.33 rows=31 width=0) (actual time=[0.035,0.035]..[0.042,0.042], rows=32) Index Cond: ((bh2)::text = '652e4e0e-ba60-0400-25b5-4ee5e490fffe'::text) Total runtime: 2.569 ms (7 rows) Time: 5.226 ms --删除索引后的全表扫描 postgres=# explain analyze select * from t4 where bh2 = '652e4e0e-ba60-0400-25b5-4ee5e490fffe'; QUERY PLAN ------------------------------------------------------------------------------------------------------------------------- Streaming (type: GATHER) (cost=0.62..31755.34 rows=31 width=102) (actual time=294.661..294.814 rows=32 loops=1) Node/s: All datanodes -> Seq Scan on t4 (cost=0.00..31747.34 rows=31 width=102) (actual time=[0.084,258.294]..[280.141,293.190], rows=32) Filter: ((bh2)::text = '652e4e0e-ba60-0400-25b5-4ee5e490fffe'::text) Rows Removed by Filter: 3199968 Total runtime: 295.154 ms (6 rows) Time: 297.348 ms
使用enable_fast_query_shipping控制是否使用分布式框架，以此来查看具体的执行计划，针对优化SQL有帮助。
仅凭 "REMOTE_FQS_QUERY"是没法判断有没有走索引，还需要进一步验证。
小小的缺陷：即使SQL走了索引，统计信息表pg_stat_all_indexes和pg_stat_all_table中的index_scan索引扫描次数都是0。
分布键类型影响
常见的fqs一般单表简单查询，以及多表连接且关联键是同类型分布键。
当查询中有函数，多表关联关联键字段类型不同，分布键类型不同，以及非等值情况都可能造成不下推。
下面举例分布键类型不一样
--t1和t2表结构完全一样，分布键都是hash(id) postgres=# \d+ t1 Table "sa.t1" Column | Type | Modifiers | Storage | Stats target | Description --------+------------------------+-----------+----------+--------------+------------- id | character varying(300) | | extended | | c_name | character varying(300) | | extended | | c_info | character varying(300) | | extended | | Indexes: "i_t1" btree (id) TABLESPACE pg_default "i_t1_id" btree (id) TABLESPACE pg_default Has OIDs: no Distribute By: HASH(id) Location Nodes: ALL DATANODES Options: orientation=row, compression=no --可以下推，执行计划显示FQS postgres=# explain select * from t1,t2 where t1.id=t2.id; QUERY PLAN ---------------------------------------------------------------------------------- id | operation | E-rows | E-width | E-costs ----+----------------------------------------------+--------+---------+--------- 1 | -> Data Node Scan on "__REMOTE_FQS_QUERY__" | 0 | 0 | 0.00 (3 rows) --修改其中一个表的分布键为随机分布roundrobin postgres=# alter table t1 distribute by roundrobin; ALTER TABLE postgres=# explain select * from t1,t2 where t1.id=t2.id; QUERY PLAN ------------------------------------------------------------------------------------------------ id | operation | E-rows | E-memory | E-width | E-costs ----+-----------------------------------------+----------+--------------+---------+----------- 1 | -> Streaming (type: GATHER) | 13021186 | | 60 | 159866.51 2 | -> Hash Join (3,5) | 13021186 | 1MB | 60 | 159449.88 3 | -> Streaming(type: REDISTRIBUTE) | 1600000 | 2MB | 30 | 53357.30 4 | -> Seq Scan on t1 | 1600000 | 1MB | 30 | 9357.33 5 | -> Hash | 1599999 | 48MB(4435MB) | 30 | 9355.33 6 | -> Seq Scan on t2 | 1600000 | 1MB | 30 | 9355.33 RunTime Analyze Information ---------------------------------- "sa.t1" runtime: 219.368ms "sa.t2" runtime: 184.141ms Predicate Information (identified by plan id) -------------------------------------------------- 2 --Hash Join (3,5) Hash Cond: ((t1.id)::text = (t2.id)::text) ====== Query Summary ===== ------------------------------- System available mem: 4546560KB Query Max mem: 4546560KB Query estimated mem: 131072KB (24 rows) --将t2表修改为随机分布，结果是查询时两个表都需要重分布 postgres=# alter table t2 distribute by roundrobin; ALTER TABLE postgres=# explain select * from t1,t2 where t1.id=t2.id; QUERY PLAN --------------------------------------------------------------------------------------------------- id | operation | E-rows | E-memory | E-width | E-costs ----+--------------------------------------------+----------+--------------+---------+----------- 1 | -> Streaming (type: GATHER) | 12804286 | | 60 | 203041.85 2 | -> Hash Join (3,5) | 12804286 | 1MB | 60 | 202625.22 3 | -> Streaming(type: REDISTRIBUTE) | 1600000 | 2MB | 30 | 53357.30 4 | -> Seq Scan on t2 | 1600000 | 1MB | 30 | 9357.33 5 | -> Hash | 1599999 | 68MB(4433MB) | 30 | 53357.30 6 | -> Streaming(type: REDISTRIBUTE) | 1600000 | 2MB | 30 | 53357.30 7 | -> Seq Scan on t1 | 1600000 | 1MB | 30 | 9357.33 RunTime Analyze Information ---------------------------------- "sa.t2" runtime: 203.933ms Predicate Information (identified by plan id) -------------------------------------------------- 2 --Hash Join (3,5) Hash Cond: ((t2.id)::text = (t1.id)::text) ====== Query Summary ===== ------------------------------- System available mem: 4546560KB Query Max mem: 4546560KB Query estimated mem: 131072KB (24 rows)
当t1表是随机分布的时候连表查询，t1表会要做重分布，t2也是随机分布的时候，连表查询也需要做重分布。随机分布的情况下是没法完全下推的。
replication模式就不演示了，因为replication是所有dn都有一份数据，所以数据量是dn数量*表数据量，每个节点都有一份完整的数据，肯定是可以下推的。
将t1和t2都改成hash分布，然后关联建选择一个非分布列，这很明显的是没法直接完全下推的：
postgres=# alter table t1 distribute by hash(id); ALTER TABLE postgres=# alter table t2 distribute by hash(id); ALTER TABLE --关联建加入c_name postgres=# explain select * from t1,t2 where t1.id=t2.c_name; QUERY PLAN --------------------------------------------------------------------------------------------------------------------- id | operation | E-rows | E-memory | E-width | E-costs ----+--------------------------------------------------------------+----------+--------------+---------+----------- 1 | -> Streaming (type: GATHER) | 12621020 | | 61 | 182863.95 2 | -> Hash Join (3,5) | 12621020 | 1MB | 61 | 182447.32 3 | -> Streaming(type: PART REDISTRIBUTE PART ROUNDROBIN) | 1600000 | 2MB | 30 | 54688.64 4 | -> Seq Scan on t2 | 1600000 | 1MB | 30 | 9355.33 5 | -> Hash | 1599999 | 48MB(4433MB) | 31 | 32355.32 6 | -> Streaming(type: PART LOCAL PART BROADCAST) | 1600000 | 2MB | 31 | 32355.32 7 | -> Seq Scan on t1 | 1600000 | 1MB | 31 | 9355.33 -- 如果将t1改成replication postgres=# alter table t1 distribute by replication ; ALTER TABLE postgres=# explain select * from t1,t2 where t1.id=t2.id; QUERY PLAN ---------------------------------------------------------------------------------- id | operation | E-rows | E-width | E-costs ----+----------------------------------------------+--------+---------+--------- 1 | -> Data Node Scan on "__REMOTE_FQS_QUERY__" | 0 | 0 | 0.00 (3 rows) --可以看到t1是复制表，t2是hash表也可以完全下推 --再将t2改为随机分布，关联查询会是怎样呢？ postgres=# alter table t2 distribute by replication; ALTER TABLE postgres=# explain select * from t1,t2 where t1.id=t2.id; QUERY PLAN ---------------------------------------------------------------------------------- id | operation | E-rows | E-width | E-costs ----+----------------------------------------------+--------+---------+--------- 1 | -> Data Node Scan on "__REMOTE_FQS_QUERY__" | 0 | 0 | 0.00 (3 rows)
当关联建中有非分布键的时候是没法完全下推的，如果将其中一个表改成复制表（每个dn都有数据），无论另外一张表是如何分布都是可以完全下推。但是复制表只适合小表
常见非FQS
- 聚合和排序操作：当查询需要进行复杂的聚合操作或排序时，通常需要在协调节点上进行。FQS不适合这些情况，因为在数据节点上执行这些操作可能会导致性能下降。
- 跨多个分布键的连接：如果查询需要连接多个表，并且这些表的连接条件涉及不同的分布键，FQS可能不是最佳选择。这样的查询可能需要在协调节点上执行，以便正确处理跨多个数据节点的连接。
- 子查询和复杂逻辑：包含复杂子查询或逻辑的查询通常需要在协调节点上进行，因为这些查询需要协调多个步骤以生成结果。
- 涉及外部数据源或函数：如果查询涉及与外部数据源通信或需要使用数据库之外的函数，FQS可能无法应用，因为这些操作通常需要在协调节点上执行，函数分三种形态，要分具体情况
总的来说，FQS是一种性能优化工具，适用于许多查询，但并非所有查询都适合。数据库查询优化通常涉及权衡，需要根据具体查询和性能需求来选择合适的执行方式。可以通过观察执行计划和性能测试来确定是否应使用FQS。
总结
1、在DWS中，FQS（Fast Query Shipping）是一种查询优化技术，允许将查询转发到数据节点以在数据节点上执行，从而减少数据传输和提高查询性能。
2、DWS中当前主要存在三类计划：
- FQS：是cn直接将原语句下发到dn，各dn单独执行，并将执行结果在cn上进行汇总
- Stream：计划是CN根据原语句生成计划并将计划下发给DN进行执行，各DN执行过程中使用Stream算子进行数据交互。
- Remote-Query：CN生成计划后，将部分原语句下发到DN，各DN单独执行，执行后将结果发送给CN，CN执行剩余计划。
3、仅凭 "REMOTE_FQS_QUERY"是没法判断有没有走索引，还需要进一步验证，使用enable_fast_query_shipping控制是否使用分布式框架，以此来查看具体的执行计划，针对优化SQL有帮助。
4、当使用随机分布的时候由于数据是随机分布的所以在进行关联查询的时候该表基本都需要进行重分布，代价较高。
5、replication模式由于各个节点都有一份数据，所以都可以完全下推，使用replication模式适合查询频繁的小表。
6、分布键和非分布键关联也不能完全下推，这是比较常见的情况，所以在进行表设计的时候分布键字段类型一致，join的列最好。
7、小小的缺陷：即使SQL走了索引，统计信息表pg_stat_all_indexes和pg_stat_all_table中的index_scan索引扫描次数都是0。
8、应该尽量保证执行计划是fqs，在fqs的基础上如果还能继续优化就可以使用enable_fast_query_shipping关闭完全下推，查看执行计划针对性的优化。
