GpuJoin + GpuPreAgg combined kernel

以下のクエリは、t0とt1の2つのテーブルをJOINし、その結果をGROUP BYして出力するものである。
しかし、EXPLAIN ANALYZEの出力には奇妙な点がある。

postgres=# explain analyze
           select cat,count(*),avg(ax) from t0 natural join t1 group by cat;
                                        QUERY PLAN
--------------------------------------------------------------------------------
 GroupAggregate  (cost=955519.94..955545.74 rows=26 width=20)
                 (actual time=5964.955..5964.972 rows=26 loops=1)
   Group Key: t0.cat
   ->  Sort  (cost=955519.94..955523.12 rows=1274 width=44)
             (actual time=5964.943..5964.947 rows=26 loops=1)
         Sort Key: t0.cat
         Sort Method: quicksort  Memory: 28kB
         ->  Gather  (cost=955323.19..955454.23 rows=1274 width=44)
                     (actual time=5964.756..5964.914 rows=26 loops=1)
               Workers Planned: 7
               Workers Launched: 7
               ->  Parallel Custom Scan (GpuPreAgg)  (cost=954323.19..954326.83 rows=182 width=44)
                                                     (actual time=5596.730..5596.735 rows=3 loops=8)
                     Reduction: Local
                     GPU Projection: cat, pgstrom.nrows(), pgstrom.nrows((ax IS NOT NULL)), pgstrom.psum(ax)
                     Unified GpuJoin: enabled
                     ->  Parallel Custom Scan (GpuJoin) on t0  (cost=45544.82..840948.19 rows=100000000 width=12)
                                                               (never executed)
                           GPU Projection: t0.cat, t1.ax
                           Outer Scan: t0  (cost=0.00..976191.14 rows=14285714 width=8)
                                           (actual time=50.762..891.266 rows=100000000 loops=1)
                           Depth 1: GpuHashJoin  (plan nrows: 14285714...100000000, actual nrows: 100000000...0)
                                    HashKeys: t0.aid
                                    JoinQuals: (t0.aid = t1.aid)
                                    KDS-Hash (size plan: 10.78MB, exec: 16.00MB)
                           ->  Seq Scan on t1  (cost=0.00..1935.00 rows=100000 width=12)
                                               (actual time=0.010..26.491 rows=100000 loops=1)
 Planning time: 0.501 ms
 Execution time: 6008.393 ms
(22 rows)

GpuScanとGpuPreAggの間に挟まれたGpuJoinが(never executed)となっているのである。
これは、PostgreSQLのエグゼキュータから見た時に、GpuJoinが一度も呼び出されていないことを意味する。
ただ、その割には、OUTER側の t0 テーブルのスキャン、INNER側の t1 テーブルのスキャン(Seq Scan)はしっかり実行されている。

これは、GpuPreAggの直下にGpuJoinが存在している場合の特別な最適化で、GpuPreAggのGPUカーネルがJOINとGROUP BYの両方のタスクを一気に実行しており、GpuJoinを実行する必要がなかった事を意味している。
(但し、GpuPreAggへの入力として、t0およびt1テーブルを読み出す必要はある)


複数のテーブルを(場合によってはWHERE句付きで)スキャンし、それを何らかのキーで結合して、最終的にGROUP BYを使って集約するというのは非常によくある処理である。
PostgreSQL*1これを内部的にいくつかのステップに分解し、順を追って処理していく。

例えば、以下のような非常に単純なクエリの場合

SELECT cat, count(*), avg(x)
  FROM t0 JOIN t1 ON t0.id = t1.id
 WHERE y like ‘%abc%’
 GROUP BY cat;

先ず最初にSCANのロジックが動作する。幸い、WHERE句の条件は複数のテーブルに跨るものではないので、条件に合致しないレコードはこの時点で捨てられる。これによって、JOINすべき行数を減らすことができる。
次に、SCANの出力はJOINの入力となり、t0.id = t1.id という条件によって t0 テーブルと t1 テーブルを結合する。JOINの結果生成されたレコードは次のAggregation/GROUP BYの入力となり、ここでcat列の値ごとに集計され、行数とx値の平均が出力される。

WHERE句の評価、JOIN、GROUP BY/AggregationはそれぞれPG-StromがGPUで実行可能なワークロードであるが、PostgreSQLの実行計画におけるScan, Join, Aggを単純に置き換えるとどうなるか。

実はCPUとGPUの間でデータ転送のピンポンが発生してしまう。
CPUはまずストレージからデータを読み出し、これをDMAバッファにロードする*2
GpuScan kernel関数はWHERE句の評価を並列実行し、結果をDMAバッファに書き戻す。これは次いでGpuJoin kernel関数の入力になり、t0, t1テーブルの結合処理を並列実行し、同じく結果をDMAバッファに書き戻す。さらにこれは次のGpuPreAgg kernel関数の入力になり、GROUP BYと集計演算の結果をDMAバッファに書き戻す。最後に、CPUが最終ステージの集約を実行して、結果をユーザへ返却する。

いくら非同期処理とはいえ、これだけ何度もCPU⇔GPU間でデータ転送を行うと色々と辛い。

なので、PG-Stromには以前から OUTER SCAN Pull-up という仕組みを実装していた。

これは、比較的単純なWHERE句の評価を GpuJoin 側に取り込んで実行するもので、WHERE句の評価をくぐり抜けたレコードのみをJOIN処理の対象とする事で実現している。

スキャン処理(WHERE句評価)の仕組みは単純なので、例えば、GROUP BYの直下にスキャンがあるようなケース。
例えば以下のクエリのような場合、

SELECT cat, count(*), sum(x) FROM big_table;

同様にGpuPreAggもWHERE句の評価を取り込んで実行するという芸当が可能であった。


今回、新たに実装し、冒頭のEXPLAIN ANALYZE文で実行してみたのは、元々の(GpuScan + GpuJoin)をGpuPreAgg側に取り込み、SCAN→JOINを実行し、さらに集約演算を実行した結果を書き戻すという機能。

一般に、GROUP BYや集約演算の実行によって大規模なデータから非常に小さな集計結果を出力する事が期待されているため、元々のSCAN+JOINやSCAN+PreAggといったシンプルなワークロードの結合に比べると、転送すべきデータ量を削減する効果は非常に大きい事が期待できる。

実際、CUDAのプロファイラでタイムラインを採取してみると

MemCpy(HtoD)(CPU側からGPU側へのデータ転送)と、それを処理するGPU kernel関数の実行は非常に多く発生しているにも関わらず、MemCpy(DtoH)(GPU側からCPU側へのデータ転送)は、最後に一回だけ、しかも僅か 1.2kB が書き戻されているだけである。

このようにシンプルな集計処理であれば、SCAN→JOIN→GROUP BYを1回のGPU kernel呼び出しで済ませてしまう事で、PCIeバス上のデータ転送量を大幅に減らし、必要最小限のデータ転送で済ませる事ができるようになる。

なお、Visual Profilerで見た感じ、各タスクのスケジューリングにまだ相当改善できる余地が残っているので、この辺も追って改良を加えていきたい。

*1:別にPostgreSQLに限った事ではないと思うが

*2:SSD-to-GPUダイレクトSQL実行の事はここでは忘れてほしい