citus对join的支持
前言
citus对支持的SQL有一定的限制，其中包括最常见的join，具体如下
inner join
无限制。根据情况会以下面几种方式之一支持
- 亲和表
即2个表的分片规则完全相同，且join列即为分片列值 - 参考表
join的2个表中，其中有一个表不分片且每个worker上都存一份副本的表，即"参考表" - 小表广播
分片数小于citus.large_table_shard_count的表被认为是小表(默认值为4）,citus会将小表的分片广播到所有worker上并缓存。小表机制容易产生bug不建议使用，比如之后更新小表时不会更新缓存从而导致数据不一致，建议改用参考表代替。 - 数据重分布
使用task-tracker执行器可支持数据重分布，然后以MapMerge的方式支持。
outer join
与inner join的主要不同是不支持数据重分布，因此无法支持两个分片规则不一致的大表的outer join。 另外，参考表只有出现在left join的右边或right join的左边才被支持。
实验
下面演示2个大表的join
环境
- CentOS release 6.5 x64物理机(16C/128G/3TB SSD)
- PostgreSQL 9.6.2
- citus 6.1.0
- pgbouncer 1.7.2
master和worker都在1台机器上，端口号不同
- master :60001
- worker1:60002
- worker2:60003
worker设置
在master上添加worker节点
SELECT * from master_add_node('/tmp', 60002);
SELECT * from master_add_node('/tmp', 60003);
数据定义
postgres=# create table tb1(id int,k int);
CREATE TABLE
postgres=# create table tb2(id int,k int);
CREATE TABLE
postgres=# select create_distributed_table('tb1','id');
create_distributed_table
--------------------------
(1 行记录)
postgres=# select create_distributed_table('tb2','id');
create_distributed_table
--------------------------
(1 行记录)
数据导入
postgres=# create unlogged table tbx as select id,id ss from ( select generate_series(1,1000000) id) a; SELECT 1000000 时间：414.776 ms postgres=# copy tb1 from PROGRAM 'psql -p60001 -Atc "copy tbx to STDOUT"'; COPY 1000000 时间：748.383 ms postgres=# copy tb2 from PROGRAM 'psql -p60001 -Atc "copy tbx to STDOUT"'; COPY 1000000 时间：757.981 ms
执行查询
inner join(分片列相同)
postgres=# select count(1) from tb1 join tb2 on(tb1.id=tb2.id); count --------- 1000000 (1 行记录) 时间：1889.941 ms
执行计划
postgres=# explain select count(1) from tb1 join tb2 on(tb1.id=tb2.id);
QUERY PLAN
--------------------------------------------------------------------------------------------------
Distributed Query into pg_merge_job_0033
Executor: Task-Tracker
Task Count: 32
Tasks Shown: One of 32
-> Task
Node: host=/tmp port=60002 dbname=postgres
-> Aggregate (cost=1818.06..1818.07 rows=1 width=8)
-> Hash Join (cost=849.88..1739.19 rows=31550 width=0)
Hash Cond: (tb1.id = tb2.id)
-> Seq Scan on tb1_102008 tb1 (cost=0.00..455.50 rows=31550 width=4)
-> Hash (cost=455.50..455.50 rows=31550 width=4)
-> Seq Scan on tb2_102040 tb2 (cost=0.00..455.50 rows=31550 width=4)
Master Query
-> Aggregate (cost=0.00..0.00 rows=0 width=0)
-> Seq Scan on pg_merge_job_0033 (cost=0.00..0.00 rows=0 width=0)
(15 行记录)
时间：14.952 ms
inner join(分片列不同)
postgres=# select count(1) from tb1 join tb2 on(tb1.id=tb2.k); ERROR: cannot use real time executor with repartition jobs 提示: Set citus.task_executor_type to "task-tracker". 时间：16.238 ms
默认的real-time执行器不支持这种join，先设置执行器为'task-tracker'
set citus.task_executor_type='task-tracker'
再执行SQL
postgres=# select count(1) from tb1 join tb2 on(tb1.id=tb2.k); count --------- 1000000 (1 行记录) 时间：16339.376 ms postgres=# select count(1) from tb1 join tb2 on(tb1.k=tb2.k); count --------- 1000000 (1 行记录) 时间：16263.971 ms
16秒完成2个100w大表的join效率也不低了。
执行计划如下：
postgres=# explain select count(1) from tb1 join tb2 on(tb1.k=tb2.k);
QUERY PLAN
-----------------------------------------------------------------------------
Distributed Query into pg_merge_job_0036
Executor: Task-Tracker
Task Count: 8
Tasks Shown: None, not supported for re-partition queries
-> MapMergeJob
Map Task Count: 32
Merge Task Count: 8
-> MapMergeJob
Map Task Count: 32
Merge Task Count: 8
Master Query
-> Aggregate (cost=0.00..0.00 rows=0 width=0)
-> Seq Scan on pg_merge_job_0036 (cost=0.00..0.00 rows=0 width=0)
(13 行记录)
时间：22.865 ms
postgres=# explain select count(1) from tb1 join tb2 on(tb1.k=tb2.k);
QUERY PLAN
-----------------------------------------------------------------------------
Distributed Query into pg_merge_job_0039
Executor: Task-Tracker
Task Count: 8
Tasks Shown: None, not supported for re-partition queries
-> MapMergeJob
Map Task Count: 32
Merge Task Count: 8
-> MapMergeJob
Map Task Count: 32
Merge Task Count: 8
Master Query
-> Aggregate (cost=0.00..0.00 rows=0 width=0)
-> Seq Scan on pg_merge_job_0039 (cost=0.00..0.00 rows=0 width=0)
(13 行记录)
时间：21.905 ms
left join
join列和分片列一致时可以支持
postgres=# select count(1) from tb1 left join tb2 on(tb1.id=tb2.id); count --------- 1000000 (1 行记录) 时间：1929.182 ms
join列和分片列不一致时不支持
postgres=# select count(1) from tb1 left join tb2 on(tb1.id=tb2.k); ERROR: cannot run outer join query if join is not on the partition column 描述: Outer joins requiring repartitioning are not supported. 时间：0.268 ms
和参考表的outer join
创建参考表tb3
postgres=# create table tb3(id int,k int);
CREATE TABLE
时间：0.758 ms
postgres=# select create_reference_table('tb3');
create_reference_table
------------------------
(1 行记录)
时间：28.051 ms
参考表在left join右边时可以支持
postgres=# select count(1) from tb1 left join tb3 on(tb1.k=tb3.k); count --------- 1000000 (1 行记录) 时间：1942.156 ms
参考表在left join左边时不支持
postgres=# select count(1) from tb3 left join tb1 on(tb1.k=tb3.k); ERROR: cannot run outer join query if join is not on the partition column 描述: Outer joins requiring repartitioning are not supported. 时间：0.183 ms
right join正好相反
postgres=# select count(1) from tb3 right join tb1 on(tb1.k=tb3.k); count --------- 1000000 (1 行记录) 时间：2155.268 ms postgres=# select count(1) from tb1 right join tb3 on(tb1.k=tb3.k); ERROR: cannot run outer join query if join is not on the partition column 描述: Outer joins requiring repartitioning are not supported. 时间：0.348 ms
full join不支持
postgres=# select count(1) from tb1 full join tb3 on(tb1.k=tb3.k); ERROR: cannot run outer join query if join is not on the partition column 描述: Outer joins requiring repartitioning are not supported. 时间：0.180 ms postgres=# select count(1) from tb3 full join tb1 on(tb1.k=tb3.k); ERROR: cannot run outer join query if join is not on the partition column 描述: Outer joins requiring repartitioning are not supported. 时间：0.163 ms
