列存索引中GroupJoin運算元的實現 - PolarDB

本文介紹了PolarDB IMCI中GroupJoin運算元的一些限制條件和實現方式，以及其他資料的一些相關實現。閱讀本文前請先瞭解基礎的HASH JOIN與HASH GROUP BY演算法。

背景

SELECT
  key1,
  SUM(sales) as total_sales
FROM
  fact_table LEFT JOIN dimension_table ON fact_table.key1 = dimension_table.key1
GROUP BY
  fact_table.key1
ORDER BY
  total_sales
LIMIT 100;

在PolarDB IMCI中，類似以上查詢語句的執行計畫是先執行一遍HASH JOIN，再執行HASH GROUP BY key1。在這兩個操作中，都會使用key1建立雜湊表（注意fact_table.key1 = dimension_table.key1），執行計畫說明如下：

HASH JOIN：使用dimension_table.key1建雜湊表，使用fact_table.key1查雜湊表並輸出資料；
HASH GROUP BY：使用fact_table.key1建雜湊表，寫入雜湊表的過程中做彙總運算。

從執行效率的角度來看，這兩個操作可以合并成一個，使用dimension_table.key1建雜湊表以及做彙總運算，使用fact_table.key1查雜湊表以及做彙總運算，因此節省了使用fact_table.key1建雜湊表的時間。這種將HASH JOIN與HASH GROUP BY兩個運算元合并成一個的操作，就是GroupJoin。

從執行效率的角度來看，將這兩個操作合并成一個操作，不僅可以減少一次建雜湊表的操作，還可以減小中間結果大小。因為JOIN是一個可能使“結果集膨脹”的運算，一張表的一行可能會匹配上另一張表的多行，最壞情況下便是笛卡兒積：N行的表與M行的表JOIN的結果最大可能是N×M的結果集。因此在HASH JOIN+HASH GROUP BY的執行方式中，一張N行的雜湊表可能會輸出N×M×S行結果（S代表selectivity，0≤S≤1），然後在HASH GROUP BY的grouping操作中再被彙總成一張新的雜湊表，這會造成資源浪費。即使是上面例子中“事實表”（大表，大小為M）與“維度資料表”（小表，大小為N）的LEFT OUTER JOIN，且key1都是unique key，也是從一張N行的雜湊表，經過HASH JOIN輸出M行結果，然後彙總成M行的雜湊表。相對而言，GroupJoin只需要在N行的雜湊表中完成join&aggr運算，不僅中間結果變少了，同時記憶體佔用也變小了。

基於以上考慮，PolarDB MySQL版在PolarDB IMCI中增加了GroupJoin運算元。

演算法設計

概述

IMCI裡的GroupJoin實現，是HashJoin與HashGroupby兩個運算元的融合：

先使用左表（小表）建立雜湊表，涉及左表的aggr函數會在建雜湊表的時候直接運算掉。這個過程與對左表彙總（i.e., HashGroupby left_table）的操作是相同的。
使用右表（大表）查雜湊表，查詢命中則在hash table entry上運算涉及右表的aggr函數，否則丟棄或者直接輸出。

以上介紹了IMCI GroupJoin演算法的基本思路，下文會對演算法進行詳細的描述以及介紹簡化的方法。

限制條件

出於實現的複雜度考慮，相對於理論上最完備的GroupJoin實現，PolarDB MySQL版做了如下幾點限制：

group by key要完全符合某一邊，且只能是某一邊的join key，雖然某些情況下join key的一部分，也能唯一定義這個key（i.e., functional dependency）；
RIGHT JOIN、GROUP BY RIGHT的情境，要求right keys是unique keys。否則可能會轉成LEFT JOIN、GROUP BY LEFT的方式，或者不使用GroupJoin；
任意一個aggr函數只能單獨引用左表，或者單獨引用右表；如果原GROUP BY運算元中的aggr函數同時引用了左右兩個表（e.g., SUN(t1.a+t2.a)），則不適用GroupJoin。

演算法

INNER JOIN/GROUP BY LEFT

此情境如下SQL所示：

l_table INNER JOIN r_table
ON l_table.key1 = r_table.key1
GROUP BY l_table.key1

說明

假設實際執行順序與SQL描述一樣，且Join過程中不會動態換邊。

使用左表建雜湊表，並且建立雜湊表的過程中直接運算涉及左表的aggr函數；涉及右表的aggr函數，對應設一個“repeat count”，這等同於一個hash table entry對應的payload的數量；
在join過程中，使用右表查雜湊表，如果不匹配，則右表的行直接被丟棄；如果匹配，左表的aggr context的“repeat count”會增加1，右表的aggr函數直接進行運算；
join完成後，只輸出曾經被匹配上的hash table entry的aggr結果，沒有被匹配上的hash table entry全部忽略；
輸出aggr結果時，要考慮“repeat count”，例如如果一個SUM(expr)的結果是200，“repeat count”是5，則最終結果是1000。

INNER JOIN/GROUP BY RIGHT

此情境如下SQL所示：

l_table INNER JOIN r_table
ON l_table.key1 = r_table.key1
GROUP BY r_table.key1

考慮到l_table.key1=r_table.key1，這種情況被歸到“INNER JOIN, GROUP BY LEFT”裡。

LEFT OUTER JOIN/GROUP BY LEFT

此情境如下SQL所示：

l_table LEFT OUTER JOIN r_table
ON l_table.key1 = r_table.key1
GROUP BY l_table.key1

使用左表建雜湊表，建雜湊表的過程中運算左表的aggr函數；涉及右表的aggr函數，對應設一個“repeat count”；
在join過程中，使用右表查雜湊表，如果不匹配，則右表的行直接被丟棄；如果匹配，左表的aggr context的“repeat count”會增加1，右表的aggr函數直接進行運算；
與INNER JOIN不同，此情境中join完成後，被匹配上的hash table entry的aggr結果直接輸出，沒有被匹配上的每個hash table entry單獨成為一個GROUP，對應的右表的aggr函數的輸入都是NULL。

LEFT OUTER JOIN/GROUP BY RIGHT

此情境如下SQL所示：

l_table LEFT OUTER JOIN r_table
ON l_table.key1 = r_table.key1
GROUP BY r_table.key1

使用左表建雜湊表，建雜湊表的過程中運算左表的aggr函數；涉及右表的aggr函數，對應設一個 “repeat count”；
在join過程中，使用右表查雜湊表，如果不匹配，則右表的行直接被丟棄；如果匹配，左表的aggr context的“repeat count”會增加1，右表的aggr函數直接進行運算；
與其他情境不同，此情境中join完成後，被匹配上的hash table entry的aggr結果直接輸出，沒有被匹配上的所有hash table entry成為一個GROUP，對應的右表的aggr函數的輸入都是NULL。

RIGHT OUTER JOIN/GROUP BY LEFT

此情境如下SQL所示：

l_table RIGHT OUTER JOIN r_table
ON l_table.key1 = r_table.key1
GROUP BY l_table.key1

使用左表建雜湊表，建立雜湊表的過程中運算左表的aggr函數；涉及右表的aggr函數，對應設一個“repeat count”；
與其他情境不同，此情境在join過程中，使用右表查雜湊表，如果匹配，左表的aggr context的“repeat count”會增加1，右表的aggr函數直接進行運算；如果不匹配，則右表的所有不匹配的行成為一個GROUP，對應的左表的aggr函數結果都是NULL；
與其他情境不同，此情境在join完成後，被匹配上的hash table entry的aggr結果直接輸出，沒有被匹配上的所有hash table entry全都忽略。

RIGHT OUTER JOIN/GROUP BY RIGHT

此情境如下SQL所示：

l_table RIGHT OUTER JOIN r_table
ON l_table.key1 = r_table.key1
GROUP BY r_table.key1

限制條件

要求r_table.key1必須是distinct的，否則這種join是不合法的；如果不能確定r_table.key1是distinct的，則需要在最佳化器裡將這種join+groupby轉成LEFT OUTER JOIN、GROUP BY LEFT。

執行步驟

使用左表建雜湊表，建雜湊表的過程中運算左表的aggr函數；涉及右表的aggr函數，對應設一個“repeat count”；
與其他情境不同，此情境在join過程中，使用右表查雜湊表，如果匹配，直接輸出左右表的aggr結果；如果不匹配，也輸出aggr結果，此時左表的aggr結果都是NULL；
與其他情境不同，此情境在join完成後，GroupJoin即完成，不需要處理任何hash table entry。

運行時落盤（spilling）處理

GroupJoin的落盤處理，類似於partition-style的HashJoin&HashGroupby的落盤處理，方法如下：

GroupJoin的整體演算法採用分區（partition）的方式；
使用左表構建雜湊表時，記憶體中的partition，構建hash table的演算法與演算法一節描述一致；
使用左表構建雜湊表時，不在記憶體中的partition，刷到磁碟中對應的臨時檔案，後續新寫入這個partition 的資料也會直接刷到磁碟中對應的臨時檔案；落盤的partition會建立一個bloomfilter，方便後續尋找的時候快速過濾掉不可能匹配的右表資料；
完成左表的雜湊表構建後，使用右表資料查雜湊表：
1. 在查雜湊表的過程中，如果對應partition在記憶體中，則如演算法中的處理方式一樣；如果對應的partition不在記憶體中，則先查bloomfilter，如果不命中bloomfilter則直接丟棄或者直接輸出，否則刷入該partition對應的臨時檔案中。
2. 記憶體中的partition完全處理完畢後，逐個處理磁碟中的partition，此時假設至少一個partition時能放入磁碟中的，不需要再切分一個partition；處理演算法與演算法中的處理方式一樣。

GroupJoin在TPCH中的應用

TPCH是一個常用的測試一個AP系統的分析查詢能力的benchmark。在TPCH的22條查詢中，有不少都是適用GroupJoin運算元的。不過，除了TPCH Q13，其他的查詢語句都需要經過一定改造才能適用GroupJoin運算元。

Q13

TPCH Q13，可以直接適用GroupJoin運算元：

select
    c_count,
    count(*) as custdist
from
    (
        select
            c_custkey,
            count(o_orderkey) as c_count
        from
            customer
            left outer join orders on c_custkey = o_custkey
            and o_comment not like '%pending%deposits%'
        group by
            c_custkey
    ) c_orders
group by
    c_count
order by
    custdist desc,
    c_count desc;

在IMCI中，如果不使用GroupJoin，則執行計畫如下：
如果使用GroupJoin，執行計畫如下：

Q3

對TPCH的Q3而言，GroupJoin的最佳化需要經過一系列等價變換：

select
    l_orderkey,
    sum(l_extendedprice * (1 - l_discount)) as revenue,
    o_orderdate,
    o_shippriority
from
    customer,
    orders,
    lineitem
where
    c_mktsegment = 'BUILDING'
    and c_custkey = o_custkey
    and l_orderkey = o_orderkey
    and o_orderdate < date '1995-03-15'
    and l_shipdate > date '1995-03-15'
group by
    l_orderkey,
    o_orderdate,
    o_shippriority
order by
    revenue desc,
    o_orderdate
limit
    10;

Q3的一種可行的執行計畫如下（IMCI中的執行計畫）： DERKEY,TEMPTABLE

由於此SQL的grouping keys是l_orderkey、o_orderdate、o_shippriority，與任何一個join keys都不相同，因此並不能直接適用GroupJoin。通過一些等價推導，可得出以下結論：

由於lineitem與orders表的join predicate是l_orderkey=o_orderkey，而且是INNER JOIN，因此可以判斷出，這個join的結果集裡面，l_orderkey=o_orderkey；
由於l_orderkey=o_orderkey，因此 GROUP BY l_orderkey、o_orderdate、o_shippriority 等價於 GROUP BY o_orderkey、o_orderdate、o_shippriority；
由於o_orderkey是orders表的PRIMARY KEY，因此每一個o_orderkey都能直接確定唯一的o_orderdate和o_shippriority （i.e.，o_orderdate and o_shippriority functionally depend on o_orderkey）；
由於o_orderkey能唯一確定o_orderdate和o_shippriority，因此GROUP BY o_orderkey、o_orderdate、o_shippriority等價於GROUP BY o_orderkey；

由上面的推導，可以將Q3的group by clause等價變換成GROUP BY o_orderkey，如此可適用於GroupJoin了：KEY,TEMPTABLE3.SUM(LINETTEM.EXTENDEDPRTCE*1.00-LUNETEM._DLSCOL

這種“functional dependency”的推導，對最佳化器有一定要求。目前MySQL最佳化器中，實現了部分functional dependency的推導，但是依然無法推匯出上面的GROUP BY o_orderkey變換。經過嘗試，發現SQL SERVER是可以推匯出GROUP BY o_orderkey變換的，這方面有比較完備的理論，但是IMCI目前在這方面還沒有完全實現。在TPCH裡面，Q3/Q4/Q10/Q13/Q18/Q20/Q21都有這種特徵，如果能做這種等價推導，將可以縮短GROUP BY的grouping keys，提高彙總操作的速度。

Q10

TPCH的Q10也不能直接適用GroupJoin：

select
    c_custkey,
    c_name,
    sum(l_extendedprice * (1 - l_discount)) as revenue,
    c_acctbal,
    n_name,
    c_address,
    c_phone,
    c_comment
from
    customer,
    orders,
    lineitem,
    nation
where
    c_custkey = o_custkey
    and l_orderkey = o_orderkey
    and o_orderdate >= date '1993-10-01'
    and o_orderdate < date '1993-10-01' + interval '3' month
    and l_returnflag = 'R'
    and c_nationkey = n_nationkey
group by
    c_custkey,
    c_name,
    c_acctbal,
    c_phone,
    n_name,
    c_address,
    c_comment
order by
    revenue desc
limit
    20;

如果要使用GroupJoin，需要做以下兩個變換：

通過等價變換把grouping keys變成c_custkey（customer表的PRIMARY KEY），這個變換與上文的Q3類似；
Join order要調整，使得customer表的JOIN在最外層。

其中1總是有益的，但是2中join order的調整，不一定是有益的。

Q17

TPCH的Q17包含一條關聯子查詢：

select
    sum(l_extendedprice) / 7.0 as avg_yearly
from
    lineitem,
    part
where
    p_partkey = l_partkey
    and p_brand = 'Brand#44'
    and p_container = 'WRAP PKG'
    and l_quantity < (
        select
            0.2 * avg(l_quantity)
        from
            lineitem
        where
            l_partkey = p_partkey
    );

其去關聯的方式有幾種，目前IMCI針對scalar aggr實現的兩種去關聯演算法得到的執行計畫分別是：

這些執行計畫都不適用GroupJoin運算元。如果採用MagicSet運算元的去關聯方式，在移除MagicSet運算元之前，會得到一個適合GroupJoin的中間態：

也就是paper_2中所描述的過程： NERALNESTING:DEEORRELATIONOFDEPENDENTSUB-

因此可以適用GroupJoin。目前IMCI部分實現了採用MagicSet運算元的去關聯方式，但是不會產生hared children的執行計畫，因此IMCI裡面無法對TPCH Q17適用GroupJoin。

Q18

TPCH Q18也是可以適用GroupJoin的，不過依然要利用等價變換轉換執行計畫，才能得到適用GroupJoin的執行計畫。為了方便描述，不失一般性，此處把Q18裡的IN子查詢以及最後的ORDER BY去掉：

select
    c_name,
    c_custkey,
    o_orderkey,
    o_orderdate,
    o_totalprice,
    sum(l_quantity)
from
    customer,
    orders,
    lineitem
where
    c_custkey = o_custkey
    and o_orderkey = l_orderkey
group by
    c_name,
    c_custkey,
    o_orderkey,
    o_orderdate,
    o_totalprice

對於這個查詢，做如下等價推導：

因為c_custkey是customer表的PRIMARY KEY，因此c_name可以由c_custkey唯一確定（functional dependency）；同理o_orderkey是orders表的PRIMARY KEY，o_orderdate與o_totalprice都可以由o_orderkey唯一確定。因此，group by clause可以被等價轉換為GROUP BY c_custkey, o_orderkey；
由於customer表與orders表的join predicate是c_custkey=o_custkey，因此可以斷言，join的結果集中，c_custkey=o_custkey；
由於c_custkey=o_custkey，因此group by clause可以被等價轉換為GROUP BY o_custkey, o_orderkey；
由於o_orderkey唯一確定o_custkey （o_orderkey是orders表的主鍵），因此group by clause可以被等價改寫為GROUP BY o_orderkey。

經過以上等價推導，整個查詢可以被等價改成類似如下一個SQL：

select
    ANY_VALUE(c_name),
    ANY_VALUE(c_custkey),
    o_orderkey,
    ANY_VALUE(o_orderdate),
    ANY_VALUE(o_totalprice),
    sum(l_quantity)
from
    customer,
    orders,
    lineitem
where
    c_custkey = o_custkey
    and o_orderkey = l_orderkey
group by
    o_orderkey

不帶GroupJoin的執行計畫
帶GroupJoin的執行計畫

上面的等價推導，因為能減少GROUP BY的grouping keys的長度，因此針對常規的執行計畫，也是有用的。

Q20

TPCH Q20的關聯子查詢的pattern與Q17是類似的：採用MagicSet運算元的去關聯方式，在移除MagicSet運算元之前，會得到一個適合GroupJoin的中間態。

select
...
and ps_availqty > (
    select
        0.5 * sum(l_quantity) < ! --- scalar aggr --->
    from
        lineitem
    where
        l_partkey = ps_partkey         < ! --- 關聯項 1 --->
        and l_suppkey = ps_suppkey     < ! --- 關聯項 2 --->
        and l_shipdate >= '1993-01-01'
        and l_shipdate < date_add('1993-01-01', interval '1' year)
)

其他

按論文paper_1和paper_2所述，Q5/Q9/Q16/Q21這4條SQL都適用GroupJoin運算元，但是暫時還沒找到合適的轉換路徑；通過查詢hyper資料庫的執行計畫（https://hyper-db.de/interface.html#），它的最佳化器也沒有為這幾條SQL產生帶有GroupJoin的執行計畫。

總結

從效果來講，因為GroupJoin在運行時能避免的重複的工作，因此在某些情境能得到比較大的效能提升。這個效果已經在實際應用中得到驗證。因此從結果的角度，GroupJoin是值得實現的。

從通用性來講，GroupJoin並不通用。GroupJoin只適用於equal join+group by且要求grouping keys與任意一邊join keys相同，而且對aggr函數、實現方式等有諸多限制；這是一種特化，而隨之而來的是比較大的實現和維護代價。從開發的角度來說，應該花更大力氣去最佳化“通用路徑”，利用通用路徑的效能提升來達到最佳化SQL查詢效率的目的，而不是通過為某個情境尋求定製性的解法。因此從這個角度來說，GroupJoin不是一個好方法。

因此在實現的時候，應該做一定的裁剪或簡化，不追求在一個特化實現裡面實現最完備的功能，但是追求最常見情境的效用（效能）最大化。

Query	HashJoin+HashGroupby	GroupJoin
Q3	130 MB	152 MB
Q13	11 MB	33 MB
Q18	315 MB	1 GB