以下のクエリは、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で見た感じ、各タスクのスケジューリングにまだ相当改善できる余地が残っているので、この辺も追って改良を加えていきたい。
