sparksql中的broadcast join和prestodb中的dynamic filter比较

今天在prestodb的qq群里看到有人提到说一个子查询在presto中非常慢：

SELECT *
FROM his_data_opt
WHERE act_no IN (
        SELECT act_no
        FROM id_act_map
        WHERE id_number = '726067685144725'
        );
           

可以看出，这是一个普通的非相关子查询，如果内部子查询经过过滤条件只剩几条，那么整个查询应该非常完美的在几秒中出结果，结果却卡着不动。原来是presto join处理的问题，对于普通的where条件，外部查询会把这个where条件下推，进行表过滤，但现在外表这个过滤条件是一个动态生成的条件，presto在进行上层的逻辑计划优化时，不知道这个动态生成的条件到底会产生多少条结果，于是presto把外部表进行了全表扫描，这在presto中成为dynamic filter，目前有人提了PR，还没有合并到主版本中。来看presto中的执行计划，查询语句如下：

explain

SELECT *
FROM nation
WHERE regionkey IN (
        SELECT regionkey
        FROM region
        WHERE regionkey = 
        );
           

执行计划：

从图中可以看出，presto对region表进行了谓词下推，而对nation表则没有生成Filter这样的物理操作符。同时可以看出，即使对于region和nation这样只有几十行的表，presto依然会对这两个表进行repartition，即shuffle操作，物理操作符是RemoteExchange，这代价也忒大了点吧，难怪会慢的不行。另外可以看出，对于这样的子查询，presto把它解析成了一个semi join来处理，即子查询中如果有重复的regionkey被查出来，只返回一个即可，这满足semi join的语义。

那么来看看sparksql中怎么处理这个问题的吧，同样的表，同样的sql语句，来看sparksql中的执行计划：

从图中可以看出，通过Broadcast实现了连接查询，是不是表本身太小了导致的呢？换大表试试：

explain

SELECT *
FROM orders
WHERE o_orderkey IN (
        SELECT l_orderkey
        FROM lineitem
        WHERE l_orderkey = 
        );
           

orders表150万行，lineitem表600万行，执行计划如下：

发现还是broadcast实现的join，于是直接执行了一下，然后去yarn上看看DAG图，如下：

可以看出，对lineitem首先进行过滤，过滤完后只剩5条，然后把这5条数据进行broadcast，和presto进行比较的话，相当于在这里把一个dynamic filter广播到了集群的所有节点上。然后用这个条件去查询外表，就会快很多了。可见，presto和sparksql在join的实现上，还有有一定差距的，presto实现了类似hadoop里的map-side join，才有可能使用dynamic filter。没有对比就没有伤害啊！

还没完，prestodb的社区里，来看看上边提到的这个dynamic filter相关的PR：

https://github.com/prestodb/presto/issues/7428

SELECT 
    count(*)
FROM 
    fact_table a 
JOIN d_date b 
ON a.date_key = b.date_key 
WHERE 
    b.year = ;
           

pr中有这个查询语句，虽然不像上边那样的子查询，但是却可以像上边子查询那样来分析，比如，b.year = 2017这个查询条件下推到b表后，b表过滤后的数据条数非常少，比如只有一条，那么如果用这一条再和a表join，那么查询将会非常快，可惜，presto中还是因为没有实现broadcast，导致其不能完成这样的优化。