openGauss向量化引擎－－hash join

传统的行执行器采用一次一个元组的执行模式，执行过程中CPU大部分时间没有用了处理数据，都用在了遍历执行树等操作，导致CPU的有效利用率较低。面向OLAP场景大量函数调用次数，需要巨大开销，为解决次问题，openGauss中开发了向量化引擎。采用一次一批元组的执行模式，可大幅减少遍历执行节点及调用函数的开销。

本文主要介绍hash join如何进行向量化的。

算子之间数据传递结构

算子之间数据组织及传递结构是VectorBatch：

主要由3个数组在表示：m＿vals为列值数组，m＿flag为对应列的行值是否为NULL，m＿sel为该行是否满足过滤条件。

VecHashJoin

向量化hash join的算子是VecHashJoin。其执行函数是ExecVecHashJoin，分为2个阶段：HASH＿BUILD和HASH＿PROBE。

VectorBatch＊ ExecVecHashJoin（VecHashJoinState＊ node）

｛    

       int64 rows ＝ 0；    

       for （；；） ｛        

             switch （node－＞joinState） ｛            

                     case HASH＿BUILD： ｛       

             if （node－＞hashTbl ＝＝ NULL）          

                    node－＞hashTbl ＝ New（CurrentMemoryContext） HashJoinTbl（node）；        

           （（HashJoinTbl＊）（node－＞hashTbl））－＞Build（）；／／构建hash表        

             rows ＝ （（HashJoinTbl＊）（node－＞hashTbl））－＞getRows（）；             

                     ｝ break；            

                     case HASH＿PROBE： ｛                

                               result ＝ （（HashJoinTbl＊）（node－＞hashTbl））－＞Probe（）；／／进行hash探测并构建join结果                

                               return result；            

                     ｝            

                    default：                

                               break；        

                     ｝    

         ｝

｝HASH＿BUILD

其中build的阶段又分为2个小阶段：1）获取内表的batch，然后通过m＿funBuild：申请hashCell ＊cell＿arr连续内存，每个节点是一个hashCell，大小是m＿cellSize，共有batch中记录的行数个。然后将其接入m＿cache链表。然后将batch中列值依次存入cell＿arr中。2）通过PrepareProbe函数构建Hash表，并将cell＿arr中值放到hash表中。

第1阶段：

第2阶段：

第2阶段，通过m＿keyIdx数组得到哪一列是join key，将cell＿arr中该列值拿出来通过m＿innerHashFuncs函数计算hash值，将其保存到m＿cacheLoc［］数组中，作为m＿data数组下标，通过这种方式将内表列值放到hash表中。

HASH＿PROBE

通过probeHashTable进行探测，并join。也分为2个小阶段：1）外表hash阶段：首先获取外表的batch，通过m＿outerHashFuncs hash函数将外表的join key列hash出的值放到m＿cacheLoc数组中，作为hash表数组的下标：m＿hashTbl－＞m＿data［m＿cacheLoc［i］］。2）join阶段：定位到的m＿hashTbl－＞m＿data［m＿cacheLoc［i］］中列值和外表中列值是否相同，若相等则通过m＿keyMatch［］数组标记。最后将m＿keyMatch［］数组标记为1的列值构建成向量batch，并返回。

VectorBatch＊ HashJoinTbl：：probeHashTable（hashSource＊ probSource）

｛    

        VectorBatch＊ res＿batch ＝ NULL；    

        while （true） ｛   

              switch （m＿probeStatus） ｛          

                     case PROBE＿FETCH：        

             ／／获取外表batch                

                            m＿outRawBatch ＝ probSource－＞getBatch（）；               

                            if （BatchIsNull（m＿outRawBatch）） ｛                    

                            ｝ else ｛      

                                  int row ＝ m＿outRawBatch－＞m＿rows；        

                                  int mask ＝ m＿hashTbl－＞m＿size － 1；

hashBatch（m＿outRawBatch， m＿outKeyIdx， m＿cacheLoc， m＿outerHashFuncs）；          

              for （int i ＝ 0； i ＜ row； i＋＋） ｛            

                      m＿cacheLoc［i］ ＝ m＿outRawBatch－＞m＿arr［icol］．m＿vals［i］ ＆ mask；       

                      m＿cellCache［i］ ＝ m＿hashTbl－＞m＿data［m＿cacheLoc［i］］；            

                      m＿match［i］ ＝ false；  flag all the row no match   

                      m＿keyMatch［i］ ＝ true；       

                ｝          

                ．．．               

                            ｝              

                            break；            

                      case PROBE＿DATA：              

                            res＿batch ＝ （this－＞＊m＿joinFun）（m＿outRawBatch）；              

                            if （！BatchIsNull（res＿batch））                    

                                     return res＿batch；

                            break；

                      case PROBE＿FINAL：              

                            return endJoin（）；            

                      default：  

                            break；        

          ｝    

      ｝

｝

／／

VectorBatch＊ HashJoinTbl：：innerJoinT（VectorBatch＊ batch）／／外部batch

｛ 

         while （m＿doProbeData） ｛        

                     last＿build＿idx ＝ 0；

               RuntimeBinding（m＿matchKeyFunction， i）（＆batch－＞m＿arr［m＿outKeyIdx［i］］， row， m＿keyIdx［i］， i）；    

               for （row＿idx ＝ last＿build＿idx； row＿idx ＜ row； row＿idx＋＋） ｛      

                   if （m＿keyMatch［row＿idx］） ｛        

                      val ＝ m＿cellCache［row＿idx］－＞m＿val；        

                      for （i ＝ 0； i ＜ m＿innerBatch－＞m＿cols； i＋＋） ｛    

                                           p＿vector ＝ ＆m＿innerBatch－＞m＿arr［i］；
                  

                                           p＿vector－＞m＿vals［result＿row］ ＝ val［i］．val；

                                           p＿vector－＞m＿flag［result＿row］ ＝ val［i］．flag；              

                                   ｝

                                         for （i ＝ 0； i ＜ m＿outerBatch－＞m＿cols； i＋＋） ｛       

                                          p＿vector ＝ ＆m＿outerBatch－＞m＿arr［i］；   

                                          p＿vector－＞m＿vals［result＿row］ ＝ batch－＞m＿arr［i］．m＿vals［row＿idx］；    

                                          p＿vector－＞m＿flag［result＿row］ ＝ batch－＞m＿arr［i］．m＿flag［row＿idx］；             

                                   ｝             

                                   result＿row＋＋；             

                    ｝  

             ｝

     ｝

     return buildResult（m＿innerBatch， m＿outerBatch， true）；

｝