DWS的MPP分布式架构，数据重分布动作导致NestLoop+IndexScan的使用场景受限，特别是LEFT JOIN场景下。针对LEFT JOIN特定场景，本文提供一种改写方案实现业务SQL性能提升。

1 预置条件

DROP TABLE tab1;
DROP TABLE tab2;
CREATE TABLE tab1(a int, b int, sysid text) DISTRIBUTE BY HASH(a);
CREATE TABLE tab2(a int, b int, sysid text) DISTRIBUTE BY HASH(a);
INSERT INTO tab1 VALUES (generate_series(1, 2^10), generate_series(1, 2^10), sys_guid());
INSERT INTO tab2 SELECT a, b, sysid FROM tab1;
INSERT INTO tab2 SELECT a+2^10, b+2^10, sys_guid() FROM tab2;
INSERT INTO tab2 SELECT a+2^11, b+2^11, sys_guid() FROM tab2;
INSERT INTO tab2 SELECT a+2^12, b+2^12, sys_guid() FROM tab2;
INSERT INTO tab2 SELECT a+2^13, b+2^13, sys_guid() FROM tab2;
INSERT INTO tab2 SELECT a+2^14, b+2^14, sys_guid() FROM tab2;
INSERT INTO tab2 SELECT a+2^15, b+2^15, sys_guid() FROM tab2;
INSERT INTO tab2 SELECT a+2^16, b+2^16, sys_guid() FROM tab2;
INSERT INTO tab2 SELECT a+2^17, b+2^17, sys_guid() FROM tab2;
INSERT INTO tab2 SELECT a+2^18, b+2^18, sys_guid() FROM tab2;
CREATE INDEX idx1 ON tab2 USING btree(sysid);
ANALYZE tab1;
ANALYZE tab2;

2 原始查询

2.1 原始查询语句

SELECT 
t1.*
FROM tab1 t1
LEFT JOIN tab2 t2 ON t1.sysid = t2.sysid
WHERE t1.a IN(1, 1001)
ORDER BY t1.a;

2.2 执行分析

原始语句performance信息(完整信息见附件《LEFT JOIN原始语句performance信息》)概要如下

从执行信息上看，主要是因为tab2表的全表扫描（Seq Scan）+重分布（Streaming(type: REDISTRIBUTE)）+随后的HashJoin，这三个步骤较长。占整体耗时的99%以上

仔细分析，可以发现上述执行有如下特征

1）tab1的结果集较小 ，tab2的数据较大， 且tab1和tab2的关联结果集较小

2） tab2的关联列不是分布列，因此LEFT JOIN的时候tab2必须做重分布

3）tab2表的关联字段sysid上存在索引

4)  tab1和 tab2关联的结果集不膨胀，即tab1的关联字段是惟一的，对于上述场景来说就是tab1的sysid具有唯一性

— 当tab1的关联字段存在非NULL的重复值时，tab1和tab2关联之后的条数会出现膨胀，即tab1和tab2关联结果集中tab2的同一条记录会出现多次，针对此场景的改写方案见4.2

通过上述分析，可以确定如果tab2走了sysid字段的索引扫描，执行的性能会导入提升。但是因为如上特征tab2上层算子必须为Streaming，导致tab2无法走索引。

3 语句改写

我们尝试通过SQL改写实现对tab2表在关联字段sysid上执行Index Scan的方式提升性能。

从LEFT JOIN的语义上讲，tab2中满足匹配条件t1.sysid = t2.sysid的记录才会对关联结果产生影响，所以我们的改写实际上分为两步

1)）把tab1和tab2做一个INNER JOIN，把tab2中满足等值关联条件的记录筛选出来。

tab1 INNER JOIN tab2执行时，tab1可以现在broadcast，然后tab2扫描的时候就可以走Index Scan。预期此步骤输出的结果集较小

2）把上一步的结果再和tab1做LEFT JOIN

3.1 改写语句

SELECT 
t1.*
FROM tab1 t1
LEFT JOIN (SELECT t2.*  /* 先把tab2跟tab1做INNER JOIN，从tab2中刷选出可以跟tab1匹配上的记录 */
FROM tab2 t2
INNER JOIN tab1 t1 ON t1.sysid = t2.sysid 
WHERE t1.a IN(1, 1001)
) t2 ON t1.sysid = t2.sysid
WHERE t1.a IN(1, 1001)
ORDER BY t1.a;

3.2 执行分析

改写后语句performance信息(完整信息见附件《LEFT JOIN改写语句performance信息》)概要如下

对比执行耗时分析，可以发现优化后的SQL语句执行耗时在7.5ms左右，比优化前(513.5ms)提升近百倍。

4. 场景扩展

4.1 tab1为复杂查询

当tab1为一个子查询，或者原始语句可以提炼出一个类似tab1的子查询时，上述思路同样适用。比如原始查询语句如下

SELECT
*
FROM f1_stg.odst_hst o
INNER JOIN f1_stg.fwwiptransaction a ON o.txnid = a.sysid
INNER JOIN f1_stg.fwwipstephistory c ON a.wipstepdef = c.sysid
LEFT JOIN f1_stg.fabwiptransactionext g2 ON c.trackintxn = g2.parent
WHERE o.activity IN ('Job', 'Rule', 'Smart');

当表f1_stg.fabwiptransactionext很大，且在字段parent有索引，同时o a c三个表的关联结果集比较小，可以把o a c三个表提炼成一个类似tab1的一个子查询，然后套用如上的逻辑把SQL语句改写为

SELECT
*
FROM f1_stg.odst_hst o
INNER JOIN f1_stg.fwwiptransaction a ON a.txnid = a.sysid
INNER JOIN f1_stg.fwwipstephistory c ON a.wipstepdef = c.sysid
LEFT JOIN (SELECT g2.* 
FROM f1_stg.fabwiptransactionext g2
INNER f1_stg.fwwipstephistory c ON c.trackintxn = g2.parent
INNER f1_stg.fwwiptransaction a ON a.wipstepdef = c.sysid
INNER f1_stg.odst_hst o ON o.txnid = a.sysid
WHERE o.activity IN ('Job', 'Rule', 'Smart')
) g2 ON c.trackintxn = g2.parent
WHERE o.activity IN ('Job', 'Rule', 'Smart');

或者

WITH t AS(
SELECT
*
FROM f1_stg.odst_hst o
INNER JOIN f1_stg.fwwiptransaction a ON a.txnid = a.sysid
INNER JOIN f1_stg.fwwipstephistory c ON a.wipstepdef = c.sysid
WHERE o.activity IN ('Job', 'Rule', 'Smart')
)
SELECT
*
FROM t
LEFT JOIN (SELECT g2.* 
FROM f1_stg.fabwiptransactionext g2
INNER JOIN t ON t.trackintxn = g2.parent
) g2 ON t.trackintxn = g2.parent;

4.1 tab1在关联字段上存在重复数据

a) 构造tab1的重复数据

INSERT INTO tab2 SELECT * FROM tab1 t1 WHERE t1.a IN(1, 1001);
INSERT INTO tab1 SELECT * FROM tab1 t1 WHERE t1.a IN(1, 1001);

b) 原始查询语句

SELECT 
t1.*
FROM tab1 t1
LEFT JOIN tab2 t2 ON t1.sysid = t2.sysid
WHERE t1.a IN(1, 1001)
ORDER BY t1.a;

原始查询结果集

postgres=# SELECT
postgres-#     t1.*
postgres-# FROM tab1 t1
postgres-# LEFT JOIN tab2 t2 ON t1.sysid = t2.sysid
postgres-# WHERE t1.a IN(1, 1001)
postgres-# ORDER BY t1.a;
a   |  b   |              sysid
------+------+----------------------------------
1 |    1 | 670DE19CA5E00300F6E618CE91A0FFFE
1 |    1 | 670DE19CA5E00300F6E618CE91A0FFFE
1 |    1 | 670DE19CA5E00300F6E618CE91A0FFFE
1 |    1 | 670DE19CA5E00300F6E618CE91A0FFFE
1001 | 1001 | 670DE19D10650F00FACE18CE91A0FFFE
1001 | 1001 | 670DE19D10650F00FACE18CE91A0FFFE
1001 | 1001 | 670DE19D10650F00FACE18CE91A0FFFE
1001 | 1001 | 670DE19D10650F00FACE18CE91A0FFFE
(8 rows)

c) 改写语句

SELECT
t1.*
FROM tab1 t1
LEFT JOIN (SELECT
*
FROM(SELECT t2.*,  /* 先把tab2跟tab1做INNER JOIN，从tab2中刷选出可以跟tab1匹配上的记录 */
rank() OVER (partition by t2.xc_node_id, t2.tableoid, t2.ctid::text order by t2.ctid::text) AS rk -- 去重标记位
FROM tab2 t2
INNER JOIN tab1 t1 ON t1.sysid = t2.sysid
WHERE t1.a IN(1, 1001)
) WHERE rk = 1
) t2 ON t1.sysid = t2.sysid
WHERE t1.a IN(1, 1001)
ORDER BY t1.a;

改写语句结果集

postgres=# SELECT
postgres-#     t1.*
postgres-# FROM tab1 t1
postgres-# LEFT JOIN (SELECT
postgres(#         *
postgres(#         FROM(SELECT t2.*,  /* 先把tab2跟tab1做INNER JOIN，从tab2中刷选出可以跟tab1匹配上的记录 */
postgres(#             row_number() OVER (partition by t2.xc_node_id, t2.tableoid, t2.ctid::text order by t2.ctid::text) AS rk -- 给t2的结果编号
postgres(#         FROM tab2 t2
postgres(#         INNER JOIN tab1 t1 ON t1.sysid = t2.sysid
postgres(#         WHERE t1.a IN(1, 1001)
postgres(#     ) WHERE rk = 1--排除重复结果
postgres(# ) t2 ON t1.sysid = t2.sysid
postgres-# WHERE t1.a IN(1, 1001)
postgres-# ORDER BY t1.a;
a   |  b   |              sysid
------+------+----------------------------------
1 |    1 | 670DE19CA5E00300F6E618CE91A0FFFE
1 |    1 | 670DE19CA5E00300F6E618CE91A0FFFE
1 |    1 | 670DE19CA5E00300F6E618CE91A0FFFE
1 |    1 | 670DE19CA5E00300F6E618CE91A0FFFE
1001 | 1001 | 670DE19D10650F00FACE18CE91A0FFFE
1001 | 1001 | 670DE19D10650F00FACE18CE91A0FFFE
1001 | 1001 | 670DE19D10650F00FACE18CE91A0FFFE
1001 | 1001 | 670DE19D10650F00FACE18CE91A0FFFE
(8 rows)