presto 'query exceeded local memory allocation': how to find the skewed join

TL;DR

Find the stage where one task got far more rows than its siblings, then fix the join distribution. Open the query in the Presto web UI and compare input rows per task; the task with 10x the rows is your skewed join. Then force a partitioned join with SET SESSION joindistributiontype = 'PARTITIONED' and re-run.

Query exceeded local memory allocation of 4.00GB

Steps

1. Get the query ID from the error or your client, and open it in the Presto coordinator web UI (usually port 8080 on the coordinator host). Expected: the query details page loads with per-stage stats.
2. Expand the stages and compare input rows per task. One task with far more rows than its siblings means skew. Expected: a single stage shows one task far above the rest; that stage contains the join.
3. Run EXPLAIN on the query and find the join node. Note whether the build side is replicated (broadcast) or partitioned. Expected: the plan shows the join type and which side is the build side.
4. If the plan broadcasts a large table, switch to partitioned for the query: run SET SESSION joindistributiontype = 'PARTITIONED'; and re-run. Expected: the query completes, and stage stats show rows spread evenly across tasks.
5. If skew persists on a partitioned join, salt the skewed key: append a random digit 0-9 to the join key on the large side, duplicate the small side once per digit value, join on the salted key, then drop the salt column. Expected: no single task dominates; peak memory per node drops.

Use this when

• Presto fails with "query exceeded local memory allocation"
• One worker OOMs while the others are nearly idle
• The query joins on a low-cardinality or skewed key (status codes, NULLs, default values)
• A query that worked on small data fails as the table grows

Not for this skill when

• The error says "exceeded global memory limit" (that is total cluster memory, a different fix)
• EXCEEDED_MEMORY on a broadcast join in Trino (see the Trino broadcast-join skill)
• The coordinator itself OOMs (that is a coordinator failure, not worker skew)

Variant phrasings

• presto query exceeded per-node memory
• presto one node out of memory join
• presto skewed join out of memory
• how to find skewed join in presto

Why it happens

Presto allocates query memory per node. A skewed join sends most rows to one worker's task, so that node exceeds its local limit while the cluster as a whole has plenty free. Broadcast joins make it worse by copying the entire build side to every node on top of the skew.

Edge cases

• Raising the per-node limit just moves the failure to bigger data; fix the distribution instead.
• Skew can hide in the probe side too, not just the build side; check both.
• EXPLAIN shows the planned distribution, the UI shows the actual one; trust the UI.
• Salted joins change results if you forget to drop the salt column or duplicate incorrectly; validate row counts before and after.

Provenance

Resolved from the public thread: https://vectle.com/posts/pst_y6CUYTVlmvHiNh3iejw03Q