Parallel Hash Join and Shared Hash Tables #
A join between two 200-million-row tables runs with four workers. The plan shows Parallel Hash Join above a Parallel Hash with Batches: 64 and a Memory Usage figure that is larger than your work_mem would suggest — yet still spilling. Another plan for a similar query shows a plain Hash Join under Gather with a Hash whose child is a non-parallel Seq Scan, and its total memory is five times what the node reports. These are the two ways a hash join runs in parallel, and they account for memory differently.
The worker model underneath is covered in parallel query execution. This page covers the join itself.
The Algorithmic Condition #
Parallel-aware: Parallel Hash Join + Parallel Hash (PostgreSQL 11+, enable_parallel_hash). Every participating process — workers and the leader — scans a portion of the inner relation with a Parallel Seq Scan or parallel index scan and inserts into one hash table in dynamic shared memory. Once built, every process probes that same table with its portion of the outer relation. The inner side is read once in total.
Parallel-oblivious: Hash Join + Hash under Gather. Each process builds its own private hash table from a full, non-parallel scan of the inner relation, then probes it with its share of the outer rows. The inner side is read once per process. For a small dimension table this is cheap and avoids coordination; for a large one it multiplies I/O and memory.
Memory budget. For Parallel Hash the budget is shared and scales with participants:
work_mem × hash_mem_multiplier × (workers launched + 1)
A parallel-oblivious hash gives each process its own work_mem × hash_mem_multiplier, and the node’s Memory Usage shows one copy — the true total is that figure times the number of processes.
Batching. When the shared table exceeds its budget, Parallel Hash increases the batch count for all processes together, and batch files are shared. Skewed keys that cannot be split behave as they do in a serial join — see hash join skew and hot-key batches — and the serial skew optimisation is not used in the parallel-aware case.
Annotated EXPLAIN Evidence #
EXPLAIN (ANALYZE, BUFFERS)
SELECT o.region, sum(li.amount)
FROM line_items li JOIN orders o ON o.id = li.order_id
GROUP BY o.region;
Finalize HashAggregate (actual time=41022.6..41022.7 rows=12 loops=1)
-> Gather (actual rows=60 loops=1)
Workers Planned: 4
Workers Launched: 4
-> Partial HashAggregate (actual rows=12 loops=5)
-> Parallel Hash Join (actual time=2210.4..38804.1 rows=40010220 loops=5)
Hash Cond: (li.order_id = o.id)
-> Parallel Seq Scan on line_items li (actual rows=40010220 loops=5)
-> Parallel Hash (actual time=2180.6..2180.6 rows=12004110 loops=5)
Buckets: 67108864 Batches: 64 Memory Usage: 131456kB
-- Batches 64 → the shared table spilled
-- budget: 64MB × 2.0 × (4 + 1) = 640 MB for the whole table
-- Memory Usage is per process; 131 MB × 5 ≈ the 640 MB budget
-> Parallel Seq Scan on orders o (actual rows=12004110 loops=5)
Execution Time: 41104.9 ms
The rows values on parallel nodes are per loop averages: rows=12004110 loops=5 means about 60 million orders in total. The loops=5 confirms four workers plus the leader participated — the check described in why parallel workers are not launched.
With VERBOSE, each worker’s share is listed:
Worker 0: actual time=2179.9..38790.2 rows=40120430 loops=1
Worker 1: actual time=2181.2..38811.6 rows=39904120 loops=1
-- balanced rows per worker; a large imbalance would point at skewed batches
Step-by-Step Resolution #
-
Identify the shape.
Parallel Hash Joinwith aParallel Hashchild is shared;Hash Joinwith aHashover a non-parallel scan underGatheris private. -
Compute the real memory.
- Shared:
work_mem × hash_mem_multiplier × (Workers Launched + 1). - Private: node
Memory Usage × (Workers Launched + 1).
- Shared:
-
For shared spills, raise memory for the statement and check the batch count:
BEGIN; SET LOCAL work_mem = '256MB'; EXPLAIN (ANALYZE, BUFFERS) <query>; ROLLBACK; -- 256MB × 2.0 × 5 = 2.56 GB budget → Batches: 1Budget the total against concurrent sessions, as in pool size and the total work_mem budget.
-
For a large private hash, test the shared version:
SHOW enable_parallel_hash; -- must be onIf it is on and the planner still chose private tables, the inner relation may be too small to justify shared memory — which is usually correct — or the inner scan is not parallel-safe.
-
Check worker balance with
VERBOSE. A single worker with far more rows or time than others suggests skewed batches.
Before and After #
-- BEFORE: work_mem 64MB, 4 workers
Parallel Hash Batches: 64 Memory Usage: 131456kB Execution Time: 41104.9 ms
-- AFTER: SET LOCAL work_mem = '256MB' for the reporting transaction
Parallel Hash Batches: 1 Memory Usage: 498204kB Execution Time: 17208.3 ms
The larger budget removed batch files entirely, so every probe became an in-memory lookup rather than a read of the matching batch file, and the build phase stopped writing temporary files. Before accepting a statement-level work_mem of 256 MB, remember that a single query with four workers can now hold 2.5 GB for one hash table alone, and more if the plan has several memory-hungry nodes.
Common Pitfalls #
Reading Memory Usage as a total. For private hashes it is one copy; for shared hashes it is a per-process view of one table. Diagnostic signal: server memory well above what nodes report. Fix: multiply by participants.
Budgeting for workers that did not launch. The shared budget uses participants that actually joined. Diagnostic signal: Workers Launched below planned and more batches than a quiet-system test showed. Fix: account for pool exhaustion when sizing memory.
Disabling enable_parallel_hash to reduce memory. The private alternative scans the inner side once per process and can use more memory in total. Diagnostic signal: higher I/O and memory after the change. Fix: reduce max_parallel_workers_per_gather or work_mem for the statement instead.
Misreading per-loop row counts. rows=12004110 loops=5 is about 60 million rows. Diagnostic signal: estimates that look five times too high. Fix: multiply by loops before comparing with table sizes.
Frequently Asked Questions #
What is the difference between Parallel Hash Join and a Hash Join under Gather?
Parallel Hash Join builds one shared hash table that all workers insert into and probe. A Hash Join under Gather builds a private copy of the inner hash table in every process, scanning the inner relation once per process.
How much memory can a Parallel Hash use?
Up to work_mem × hash_mem_multiplier × (workers + 1) for the one shared table.
Why is my query using a parallel-oblivious hash join?
enable_parallel_hash is off, the inner side cannot be scanned in parallel, or the planner judged a small private table cheaper — often a reasonable choice.
Related #
- Parallel Query Execution — parent guide: Gather, workers and per-loop counts
- Parallel Worker Allocation Strategies — sibling: sizing workers against memory
- Diagnosing Hash Join Batch Spills — the serial version of the batch arithmetic