Citus 报错（四）HINT: Set canopy.enable_repartition_joins to on to enable repartitioning

报错和解决办法如下：

create table customers (customers_id bigint,product_name varchar(100));
alter table customers add primary key (customers_id);

create table orders (order_id bigint,customers_id bigint,order_detail varchar(100));
alter table orders add primary key(order_id,customers_id);

create table products (product_id bigint,customers_id bigint,product_name varchar(100));
alter table products add primary key (product_id);

select create_distributed_table('customers','customers_id');
select create_distributed_table('orders','order_id');
--这里很显然分区的时候，如果还按照customer_id来关联，数据是分布在不同的work节点上的

insert into customers 
select n,md5(random()::text)
from generate_series(1,10000) n;

insert into orders 
select n*3,n,md5(random()::text)
from generate_series(1,200000) n;

insert into products 
select 10+n*2,n,md5(random()::text)
from generate_series(1,10000) n;

lightdb@test=# explain analyze select c.* , o.* 
 from customers c , orders o 
 where o.customers_id = c.customers_id;
ERROR:  the query contains a join that requires repartitioning
HINT:  Set canopy.enable_repartition_joins to on to enable repartitioning
lightdb@test=# 
lightdb@test=# select c.* , o.* 
lightdb@test-#  from customers c , orders o 
lightdb@test-#  where o.order_detail = c.product_name;
ERROR:  the query contains a join that requires repartitioning
HINT:  Set canopy.enable_repartition_joins to on to enable repartitioning
lightdb@test=# set canopy.enable_repartition_joins = on;
SET
select c.* , o.* 
 from customers c , orders o 
 where o.order_detail = c.product_name;
 customers_id | product_name | order_id | customers_id | order_detail 
--------------+--------------+----------+--------------+--------------
(0 rows)

那么这个错误的原因是什么呢？
首先我们看下enable_repartition_joins参数
canopy.enable_repartition_joins (boolean)
Ordinarily, attempting to perform Repartition joins with the adaptive executor will fail with an error message. However, setting canopy.enable_repartition_joins to true allows Canopy to perform the join. The default value is false.
单独看这段其实也不能完全理解，那么什么是repartition join呢？
http://www.light-pg.com/docs/canopy/13.8-22.4/develop/reference_sql.html#repartition-joins

Repartition joins

In some cases, you may need to join two tables on columns other than the distribution column. For such cases, Canopy also allows joining on non-distribution key columns by dynamically repartitioning the tables for the query.

In such cases the table(s) to be partitioned are determined by the query optimizer on the basis of the distribution columns, join keys and sizes of the tables. With repartitioned tables, it can be ensured that only relevant shard pairs are joined with each other reducing the amount of data transferred across network drastically.

In general, co-located joins are more efficient than repartition joins as repartition joins require shuffling of data. So, you should try to distribute your tables by the common join keys whenever possible.

Canopy的它的Repartition Join是一种优化技术，用于优化联接操作的性能。在Repartition Join中，Citus会将联接操作的输入数据重新分区，以确保输入数据在联接之前被分布到同一节点上。

具体来说，Repartition Join分为两个步骤：

1、Repartition:Canopy会将需要联接的表按照联接键进行分区，并将分区后的数据分发到各个节点上。这样，同一联接键的数据就可以分布在同一节点上。

2、Join Canopy会在各个节点上执行本地联接操作，然后将结果返回给协调节点，最终在协调节点上完成全局联接操作。

下面是一个使用Repartition Join的示例：

假设有两个表，一个是orders表，包含订单号（order_id）和订单总金额（total_amount）两个字段；另一个是customers表，包含订单号（order_id）和客户姓名（customer_name）两个字段。现在需要联接这两个表，以获取每个订单的总金额和客户姓名。

使用Repartition Join，Citus会先将orders表和customers表按照订单号进行分区，然后将分区后的数据分发到各个节点上。接下来，各个节点会在本地执行联接操作，并返回结果给协调节点。最后，协调节点将所有结果汇总，并返回给用户。

使用Repartition Join可以大大减少数据传输和网络开销，提高联接操作的性能。

lightdb@test=# set canopy.enable_repartition_joins = on;
SET
explain analyze select c.* , o.* 
 from customers c , orders o 
 where o.customers_id = c.customers_id;
                                                        QUERY PLAN                                                        
--------------------------------------------------------------------------------------------------------------------------
 Custom Scan (Canopy Adaptive)  (cost=0.00..0.00 rows=100000 width=460) (actual time=530.725..531.246 rows=10000 loops=1)
   Task Count: 8
   Tuple data received from nodes: 746 kB
   Tasks Shown: None, not supported for re-partition queries
   ->  MapMergeJob
         Map Task Count: 4
         Merge Task Count: 8
   ->  MapMergeJob
         Map Task Count: 4
         Merge Task Count: 8
 Planning Time: 1.085 ms
 Execution Time: 532.342 ms
(12 rows)

可以看到执行计划的不同

explain analyze select c.* , o.* 
 from customers c , orders o 
 where o.order_id = c.customers_id;
                                                                             QUERY PLAN                                                                      
       
-------------------------------------------------------------------------------------------------------------------------------------------------------------
-------
 Custom Scan (Canopy Adaptive)  (cost=0.00..0.00 rows=100000 width=460) (actual time=49.760..49.963 rows=3333 loops=1)
   Task Count: 4
   Tuple data received from nodes: 246 kB
   Tasks Shown: One of 4
   ->  Task
         Tuple data received from node: 60 kB
         Node: host=10.20.30.12 port=5432 dbname=test
         ->  Merge Join  (cost=0.63..125.36 rows=2445 width=90) (actual time=0.030..2.687 rows=811 loops=1)
               Merge Cond: (c.customers_id = o.order_id)
               ->  Index Scan using customers_pkey_102115 on customers_102115 c  (cost=0.28..68.95 rows=2445 width=41) (actual time=0.013..1.076 rows=2445 lo
ops=1)
               ->  Index Scan using orders_pkey_102119 on orders_102119 o  (cost=0.29..1453.86 rows=49771 width=49) (actual time=0.011..0.738 rows=812 loops=
1)
             Planning Time: 0.919 ms
             Execution Time: 3.121 ms
 Planning Time: 0.713 ms
 Execution Time: 50.224 ms
(15 rows)