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.

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.