はじめに

この記事は現在調査中であるPostgreSQLのhash joinについてのメモとなります。
あくまで調査中であることや、素人が見様見真似で調査していることから間違いが勘違いなどが多分に含まれている可能性があることにご注意ください。

基本的な仕組みについて

次の論文にあるHybrid Join(GRACE joinの効率を良くしたもの)であると思われます。

Join Processing in Database System with Large Main Memories

Hash Joins - PGCon

歴代のJoinのアルゴリズムやトレンドなどを把握できておらず恐縮ですが、データベースの教科書にも登場することから、鉄板の古典的なアルゴリズムであると考えられます。

チューニングのポイントについて

work_memを拡張することでbatchサイズ(前述の論文でいうところのpartition数)を削減し、結果として一時ファイルのI/Oの発生数を削減して処理時間を改善することが可能です。
以下、実際に動作させた例となります、temp read/written数が削減され、結果として処理時間が徐々に改善されていることがわかります。
※hybrid joinのアルゴリズムより、batch数が半分だから一時ファイルのI/Oが半減する「わけではない」ことに注意


postgres=# show work_mem;
-[ RECORD 1 ]--
work_mem | 64kB

postgres=# EXPLAIN (ANALYZE, BUFFERS) SELECT COUNT(*) FROM hoge INNER JOIN hage using(id);
                                                        QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------
 Aggregate  (cost=3075.00..3075.01 rows=1 width=8) (actual time=83.202..83.204 rows=1 loops=1)
   Buffers: shared hit=488, temp read=314 written=314
   ->  Hash Join  (cost=310.00..3050.00 rows=10000 width=0) (actual time=11.833..81.830 rows=10000 loops=1)
         Hash Cond: (hage.id = hoge.id)
         Buffers: shared hit=488, temp read=314 written=314
         ->  Seq Scan on hage  (cost=0.00..1443.00 rows=100000 width=4) (actual time=0.016..11.230 rows=100000 loops=1)
               Buffers: shared hit=443
         ->  Hash  (cost=145.00..145.00 rows=10000 width=4) (actual time=9.675..9.676 rows=10000 loops=1)
               Buckets: 2048  Batches: 16  Memory Usage: 36kB
               Buffers: shared hit=45, temp written=15
               ->  Seq Scan on hoge  (cost=0.00..145.00 rows=10000 width=4) (actual time=0.008..1.932 rows=10000 loops=1)
                     Buffers: shared hit=45
 Planning Time: 0.169 ms
 Execution Time: 83.293 ms
(14 rows)

postgres=# set work_mem='128kB';
SET
postgres=# EXPLAIN (ANALYZE, BUFFERS) SELECT COUNT(*) FROM hoge INNER JOIN hage using(id);
                                                        QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------
 Aggregate  (cost=3075.00..3075.01 rows=1 width=8) (actual time=78.063..78.067 rows=1 loops=1)
   Buffers: shared hit=488, temp read=288 written=288
   ->  Hash Join  (cost=310.00..3050.00 rows=10000 width=0) (actual time=15.242..76.824 rows=10000 loops=1)
         Hash Cond: (hage.id = hoge.id)
         Buffers: shared hit=488, temp read=288 written=288
         ->  Seq Scan on hage  (cost=0.00..1443.00 rows=100000 width=4) (actual time=0.012..10.184 rows=100000 loops=1)
               Buffers: shared hit=443
         ->  Hash  (cost=145.00..145.00 rows=10000 width=4) (actual time=13.091..13.092 rows=10000 loops=1)
               Buckets: 4096  Batches: 8  Memory Usage: 75kB
               Buffers: shared hit=45, temp written=21
               ->  Seq Scan on hoge  (cost=0.00..145.00 rows=10000 width=4) (actual time=0.005..0.998 rows=10000 loops=1)
                     Buffers: shared hit=45
 Planning Time: 0.124 ms
 Execution Time: 78.147 ms
(14 rows)

postgres=# set work_mem='256kB';
SET
postgres=# EXPLAIN (ANALYZE, BUFFERS) SELECT COUNT(*) FROM hoge INNER JOIN hage using(id);
                                                        QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------
 Aggregate  (cost=3075.00..3075.01 rows=1 width=8) (actual time=65.284..65.286 rows=1 loops=1)
   Buffers: shared hit=488, temp read=245 written=245
   ->  Hash Join  (cost=310.00..3050.00 rows=10000 width=0) (actual time=4.062..64.128 rows=10000 loops=1)
         Hash Cond: (hage.id = hoge.id)
         Buffers: shared hit=488, temp read=245 written=245
         ->  Seq Scan on hage  (cost=0.00..1443.00 rows=100000 width=4) (actual time=0.015..10.524 rows=100000 loops=1)
               Buffers: shared hit=443
         ->  Hash  (cost=145.00..145.00 rows=10000 width=4) (actual time=4.032..4.033 rows=10000 loops=1)
               Buckets: 8192  Batches: 4  Memory Usage: 152kB
               Buffers: shared hit=45, temp written=21
               ->  Seq Scan on hoge  (cost=0.00..145.00 rows=10000 width=4) (actual time=0.006..1.007 rows=10000 loops=1)
                     Buffers: shared hit=45
 Planning Time: 0.129 ms
 Execution Time: 65.360 ms
(14 rows)

注意点としては、batch=1に到達すると一時ファイルの利用がなくなることから、これ以上work_memを増やしても一時ファイルのI/Oの削減の効果がないことに注意です。
以下の例ではwork_memが1MBの時点でbatch=1でtemp read/writtenが存在せず、2MBのときと性能が同じ程度の処理時間であることがわかります。

postgres=# set work_mem='1MB';
SET

postgres=# EXPLAIN (ANALYZE, BUFFERS) SELECT COUNT(*) FROM hoge INNER JOIN hage using(id);
                                                        QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------
 Aggregate  (cost=2213.00..2213.01 rows=1 width=8) (actual time=39.288..39.293 rows=1 loops=1)
   Buffers: shared hit=488
   ->  Hash Join  (cost=270.00..2188.00 rows=10000 width=0) (actual time=3.403..38.136 rows=10000 loops=1)
         Hash Cond: (hage.id = hoge.id)
         Buffers: shared hit=488
         ->  Seq Scan on hage  (cost=0.00..1443.00 rows=100000 width=4) (actual time=0.011..10.452 rows=100000 loops=1)
               Buffers: shared hit=443
         ->  Hash  (cost=145.00..145.00 rows=10000 width=4) (actual time=3.380..3.383 rows=10000 loops=1)
               Buckets: 16384  Batches: 1  Memory Usage: 480kB
               Buffers: shared hit=45
               ->  Seq Scan on hoge  (cost=0.00..145.00 rows=10000 width=4) (actual time=0.005..1.104 rows=10000 loops=1)
                     Buffers: shared hit=45
 Planning Time: 0.167 ms
 Execution Time: 39.386 ms
(14 rows)

postgres=# set work_mem='2MB';
SET

postgres=# EXPLAIN (ANALYZE, BUFFERS) SELECT COUNT(*) FROM hoge INNER JOIN hage using(id);
                                                        QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------
 Aggregate  (cost=2213.00..2213.01 rows=1 width=8) (actual time=40.604..40.607 rows=1 loops=1)
   Buffers: shared hit=488
   ->  Hash Join  (cost=270.00..2188.00 rows=10000 width=0) (actual time=4.280..39.490 rows=10000 loops=1)
         Hash Cond: (hage.id = hoge.id)
         Buffers: shared hit=488
         ->  Seq Scan on hage  (cost=0.00..1443.00 rows=100000 width=4) (actual time=0.011..10.441 rows=100000 loops=1)
               Buffers: shared hit=443
         ->  Hash  (cost=145.00..145.00 rows=10000 width=4) (actual time=4.254..4.255 rows=10000 loops=1)
               Buckets: 16384  Batches: 1  Memory Usage: 480kB
               Buffers: shared hit=45
               ->  Seq Scan on hoge  (cost=0.00..145.00 rows=10000 width=4) (actual time=0.006..1.294 rows=10000 loops=1)
                     Buffers: shared hit=45
 Planning Time: 0.129 ms
 Execution Time: 40.736 ms
(14 rows)

この「一時ファイルが出力されるかどうか」のしきい値は以下のように「inner tableの行データ + bucket(hash table)の情報がwork_memにまるごと収まるかどうか」で判定されています。
以下、ソースコードの該当部分となります。

work_memとの比較

nodeHash.c

void
ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
...
/*
	 * If there's not enough space to store the projected number of tuples and
	 * the required bucket headers, we will need multiple batches.
	 */
	bucket_bytes = sizeof(HashJoinTuple) * nbuckets;
	if (inner_rel_bytes + bucket_bytes > hash_table_bytes)
	{

複数のbatchに分かれる場合の一時ファイル用配列の確保

nodeHash.c


HashJoinTable
ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, bool keepNulls)
...
	if (nbatch > 1 && hashtable->parallel_state == NULL)
	{
		/*
		 * allocate and initialize the file arrays in hashCxt (not needed for
		 * parallel case which uses shared tuplestores instead of raw files)
		 */
		hashtable->innerBatchFile = (BufFile **)
			palloc0(nbatch * sizeof(BufFile *));
		hashtable->outerBatchFile = (BufFile **)
			palloc0(nbatch * sizeof(BufFile *));
		/* The files will not be opened until needed... */
		/* ... but make sure we have temp tablespaces established for them */
		PrepareTempTablespaces();
	}

work_memに収まり切らない場合にはinner/outer tableを複数のbatchに細切れにして、batch単位での突き合わせを行うので、見かけ上work_memがある閾値を超えた瞬間にいきなり一時ファイルの利用がゼロになる分かりづらい振る舞いとなります。
以下、実際に実行した例となります。

postgres=# \d+
                                             List of relations
 Schema |          Name           | Type  |  Owner   | Persistence | Access Method |  Size   | Description
--------+-------------------------+-------+----------+-------------+---------------+---------+-------------
 public | hage                    | table | postgres | permanent   | heap          | 3576 kB |
 public | hoge                    | table | postgres | permanent   | heap          | 392 kB  |

postgres=# set work_mem='500kB';
SET
postgres=# EXPLAIN (ANALYZE, BUFFERS) SELECT COUNT(*) FROM hoge INNER JOIN hage using(id);
                                                        QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------
 Aggregate  (cost=3075.00..3075.01 rows=1 width=8) (actual time=61.161..61.164 rows=1 loops=1)
   Buffers: shared hit=488, temp read=162 written=162
   ->  Hash Join  (cost=310.00..3050.00 rows=10000 width=0) (actual time=4.676..59.848 rows=10000 loops=1)
         Hash Cond: (hage.id = hoge.id)
         Buffers: shared hit=488, temp read=162 written=162
         ->  Seq Scan on hage  (cost=0.00..1443.00 rows=100000 width=4) (actual time=0.013..11.603 rows=100000 loops=1)
               Buffers: shared hit=443
         ->  Hash  (cost=145.00..145.00 rows=10000 width=4) (actual time=4.645..4.646 rows=10000 loops=1)
               Buckets: 16384  Batches: 2  Memory Usage: 303kB
               Buffers: shared hit=45, temp written=14
               ->  Seq Scan on hoge  (cost=0.00..145.00 rows=10000 width=4) (actual time=0.006..1.266 rows=10000 loops=1)
                     Buffers: shared hit=45
 Planning Time: 0.243 ms
 Execution Time: 61.247 ms
(14 rows)

postgres=# set work_mem='600kB';
SET
postgres=# EXPLAIN (ANALYZE, BUFFERS) SELECT COUNT(*) FROM hoge INNER JOIN hage using(id);
                                                        QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------
 Aggregate  (cost=2213.00..2213.01 rows=1 width=8) (actual time=38.392..38.396 rows=1 loops=1)
   Buffers: shared hit=488
   ->  Hash Join  (cost=270.00..2188.00 rows=10000 width=0) (actual time=3.163..37.372 rows=10000 loops=1)
         Hash Cond: (hage.id = hoge.id)
         Buffers: shared hit=488
         ->  Seq Scan on hage  (cost=0.00..1443.00 rows=100000 width=4) (actual time=0.012..10.582 rows=100000 loops=1)
               Buffers: shared hit=443
         ->  Hash  (cost=145.00..145.00 rows=10000 width=4) (actual time=3.135..3.138 rows=10000 loops=1)
               Buckets: 16384  Batches: 1  Memory Usage: 480kB
               Buffers: shared hit=45
               ->  Seq Scan on hoge  (cost=0.00..145.00 rows=10000 width=4) (actual time=0.005..0.935 rows=10000 loops=1)
                     Buffers: shared hit=45
 Planning Time: 0.124 ms
 Execution Time: 38.456 ms
(14 rows)

work_mem=500kBのときにはhash joinの実行時にtemp read=162, written=162と大量の一時ファイルの入出力が行われ、batch=2となっていますが、work_mem=600kBのときには一時ファイルの入出力が存在しないことがわかります。
hogeテーブルのサイズが392kBであることから、テーブル+hash bucketでおおよそ5xxkB程度のメモリが必要となっていると考えられます。

現実的に考えると、性能が問題になるようなケースでセッション単位でhash join対象のinner tableをまるごとwork_memに収めることは大変厳しいです。
(work_memにGBのオーダーのメモリを割り当てて複数のセッションで並行してクエリを実行すると、それだけで物理メモリを大きく圧迫してしまう)

したがってまるごとwork_memに収めることを狙うよりも、処理時間と割当可能なメモリ容量のトレードオフを考えながら適宜work_memのチューニングを行う必要があると考えられます。

チューニングのポイントまとめ

EXPLAIN ANALYZEで確認可能なbatchが「1」になるまでは、work_memの増強より一時ファイルのI/Oを減らす事が可能です
- それ以上の増強は一時ファイルのI/Oを減らす効果はありません
hash joinのアルゴリズムにより、work_memの増強で線形に処理性能が改善されるわけではありません
- 現実的に割り当て可能なwork_memの容量と、短縮される処理時間のトレードオフをよく考える必要があります