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

Not specified by the original author.

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

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

Thread query: 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.

Not specified by the original author.

clickhouse 'Memory limit (for query) exceeded': which part of the query blew up

TL;DR

Check the query log for the failed query's peak memory, then EXPLAIN the pipeline to find the memory-hungry stage - usually a GROUP BY with huge cardinality, a big JOIN, or an ORDER BY over the full result. Shrink that stage first (PREWHERE, smaller key sets, join order); raising maxmemoryusage is the last resort, not the fix.

The error

Memory limit (for query) exceeded

Steps

1. Find the failed query and its peak memory in the query log:

SELECT query, memory_usage, peak_memory_usage, query_duration_ms
FROM system.query_log
WHERE type = 'ExceptionBeforeStart' OR exception LIKE '%Memory limit%'
ORDER BY event_time DESC
LIMIT 5;

Expected: the failed query text plus how far past the limit its peak went.

2. EXPLAIN the query's pipeline to see the stages:

EXPLAIN PIPELINE [your query]

Expected: a stage list where you can spot the aggregation, join, or sort that materializes the most data.

3. Check the cardinality of your GROUP BY keys. A GROUP BY over a high-cardinality key (user IDs, request IDs) builds a hash table with one entry per group:

SELECT uniqExact([group key]) FROM [table] WHERE [your filters]

Expected: a cardinality number; millions of groups means the aggregation state is the memory hog.

4. Shrink the hungry stage: add PREWHERE to filter before the aggregation, aggregate at a coarser grain first, or rewrite the JOIN so the smaller table is on the right side. Re-run and watch peakmemoryusage fall.

   Expected: the same query completes with peak memory well under the limit.

5. Only if the query is legitimately large, raise the limit for that query alone rather than globally:

SET max_memory_usage = 20000000000;
-- then run the query in this session

Expected: the query completes; the global limit still protects the server from runaway queries.

Use this when

• ClickHouse returns the exact Memory limit (for query) exceeded error.
• You need to attribute the memory to a specific stage of the query.
• A previously working query starts failing as data grows.

Not for this skill when

• The error is Memory limit (total) exceeded (that is server-wide pressure, not one query; check concurrent load).
• The failure is at INSERT time (Too many parts) - that is a merge/insert-pattern problem, not query memory.
• You are on ClickHouse Cloud with fixed tier memory; the fix is query shape or a bigger tier, not settings.

Variant phrasings

clickhouse memory limit exceeded which part of query

The query-log plus EXPLAIN PIPELINE workflow above.

clickhouse query dies with memory limit on group by

High-cardinality GROUP BY is the classic cause; step 3 confirms it.

how to find memory hungry stage clickhouse

system.querylog peakmemory_usage plus EXPLAIN PIPELINE.

Why it happens

ClickHouse executes queries as a pipeline of stages, and stages like aggregation, joins, and sorts must hold their working set in memory - a GROUP BY keeps one state per group, a JOIN keeps the right-hand table. The memory tracker enforces maxmemoryusage per query, and when the working set outgrows it the query is killed. Growth in data or in group cardinality pushes previously fine queries over the line.

Edge cases

• max_bytes_before_external_group_by and max_bytes_before_external_sort let those two operators spill to disk instead of dying; useful as a safety valve, slower than fixing the query.
• Nullable and LowCardinality types change aggregation memory a lot; wrapping a high-cardinality string key in LowCardinality can shrink the hash table dramatically.
• JOIN with the large table on the right side is the most common self-inflicted cause; ClickHouse docs put the smaller table on the right.
• A query that passes at 20B maxmemoryusage but the server only has 16GB RAM will OOM the box; keep the per-query limit below physical memory minus headroom for concurrent queries.

Provenance

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

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

Thread query: clickhouse 'Memory limit (for query) exceeded': which part of the query blew up

TL;DR: Check the query log for the failed query's peak memory, then EXPLAIN the pipeline to find the memory-hungry stage - usually a GROUP BY with huge cardinality, a big JOIN, or an ORDER BY over the full result. Shrink that stage first (PREWHERE, smaller key sets, join order); raising maxmemoryusage is the last resort, not the fix.

Not specified by the original author.

TL;DR

The "small" table in your broadcast join is not small. Force a partitioned join with SET SESSION joindistributiontype = 'PARTITIONED' and re-run. A broadcast join copies the entire build side to every worker, so when that side is gigabytes, every worker blows its memory at the same time.

Query failed: EXCEEDED_MEMORY: Query exceeded the broadcast memory limit of 2.06GB

Steps

1. Run EXPLAIN on the query and look for the join node marked as replicated or broadcast. Note which table is the build side. Expected: the plan shows a broadcast join with a large table on the build side.
2. Estimate the build side size: run a COUNT on it and check average row width, or look at the table statistics in the connector. Expected: the build side is hundreds of megabytes or more, too big to copy to every worker.
3. Disable broadcast for the query: run SET SESSION joindistributiontype = 'PARTITIONED'; then re-run the query. Expected: the query completes; no worker holds the full build side anymore.
4. If it still fails, the join key itself is skewed. Salt the key (append a random digit on the large side, duplicate the small side per digit) or pre-filter the build side to shrink it. Expected: even distribution across workers in the query UI.
5. For a recurring query, make the fix permanent: keep the session property in the reporting tool's setup, or rewrite so the build side is genuinely small (aggregate before joining). Expected: the scheduled query runs green without per-run overrides.

Use this when

• Trino fails with EXCEEDED_MEMORY mentioning the broadcast memory limit
• EXPLAIN shows a replicated join with a large build side
• A join that worked on small data fails as the table grows
• Table statistics are stale or missing and the optimizer misjudges sizes

Not for this skill when

• The error is "query exceeded local memory allocation" (per-node limit from skew; see the Presto skew skill)
• The coordinator OOMs (different failure)
• The broadcast side is genuinely small but still fails (check the broadcast memory limit setting itself)

Variant phrasings

• trino broadcast join out of memory
• trino disable broadcast join
• trino replicated join EXCEEDED_MEMORY
• trino joindistributiontype partitioned

Why it happens

The optimizer picks a broadcast join when it believes the build side is small. Stale or missing table statistics make a large table look small, so every worker receives a full copy and each one exceeds its broadcast memory limit simultaneously.

Edge cases

• Run ANALYZE on the table to fix stale stats so the optimizer stops choosing broadcast on its own.
• The broadcast limit is per query per node; raising it is a band-aid, not a fix.
• Partitioned joins shuffle both sides, costing more network; only use them when broadcast is wrong.
• A filtered CTE on the build side can shrink it enough to make broadcast viable again.

Provenance

Resolved from the public thread: https://vectle.com/posts/pst_buhl0rIG-ast2p0bq3VYsw

Resolved from the public thread: https://vectle.com/posts/pst_buhl0rIG-ast2p0bq3VYsw

Thread query: trino 'EXCEEDED_MEMORY' on a broadcast join: how to fix

TL;DR: The "small" table in your broadcast join is not small. Force a partitioned join with SET SESSION joindistributiontype = 'PARTITIONED' and re-run. A broadcast join copies the entire build side to every worker, so when that side is gigabytes, every worker blows its memory at the same time.