Reading the BigQuery Query Plan: Skew, Shuffle Spill, and the Six-Line Checklist

Every BigQuery performance complaint eventually lands in the same place: a query that used to take 20 seconds now takes six minutes, nobody changed the SQL, and the bill went up. Partitioning and clustering advice only gets you so far. To fix the hard cases you have to read the query execution plan — the stage-by-stage record BigQuery keeps of how it actually ran your query.

This tutorial walks through the plan end to end: where to find it, what each timing number means, and the four failure patterns (skew, shuffle spill, exploding joins, and pointless repartitioning) that account for most slow queries we see on client engagements.

Where the plan lives

There are three views of the same data, and you will use all three.

1. The Execution Graph / Execution Details tab in BigQuery Studio, next to the results of any completed query. Fastest way to eyeball one query.
2. INFORMATION_SCHEMA.JOBS, which carries a job_stages array with one row per stage. This is the programmatic view — it lets you diff a query against its own history.
3. The jobs.get REST response, field statistics.query.queryPlan. Same content, useful from CI or a monitoring job.

Start with the SQL view, because it scales:

SELECT
  job_id,
  TIMESTAMP_DIFF(end_time, creation_time, SECOND) AS wall_seconds,
  total_slot_ms / 1000 AS slot_seconds,
  total_bytes_processed / POW(1024, 4) AS tib_processed,
  ARRAY_LENGTH(job_stages) AS stage_count
FROM `region-us`.INFORMATION_SCHEMA.JOBS
WHERE creation_time > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 2 DAY)
  AND statement_type != 'SCRIPT'
  AND job_type = 'QUERY'
  AND state = 'DONE'
ORDER BY total_slot_ms DESC
LIMIT 20;

Pick one offending job_id, then unnest its stages.

Anatomy of a stage

SELECT
  stage.id,
  stage.name,
  stage.records_read,
  stage.records_written,
  stage.parallel_inputs        AS input_units,
  stage.shuffle_output_bytes / POW(1024, 3) AS shuffle_gib,
  stage.shuffle_output_bytes_spilled / POW(1024, 3) AS spilled_gib,
  stage.wait_ms_avg,  stage.wait_ms_max,
  stage.read_ms_avg,  stage.read_ms_max,
  stage.compute_ms_avg, stage.compute_ms_max,
  stage.write_ms_avg, stage.write_ms_max,
  stage.slot_ms / 1000 AS slot_seconds
FROM `region-us`.INFORMATION_SCHEMA.JOBS,
  UNNEST(job_stages) AS stage
WHERE job_id = 'bquxjob_xxxxxxxx_xxxxxxxxxxx'
  AND creation_time > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 2 DAY)
ORDER BY stage.slot_ms DESC;

A stage is a set of identical workers that each process one input unit and write their output to shuffle. The four timing phases are:

Phase	What the worker was doing	Typical cause when it dominates
wait	Queued, waiting for slots or for the previous stage to finish	Slot starvation, or a serial dependency chain
read	Pulling bytes from storage or from the previous stage's shuffle	Too many bytes scanned; no partition pruning
compute	Evaluating expressions, joining, aggregating	Expensive UDFs, regex, ORDER BY, huge joins
write	Pushing results into shuffle or the destination table	Shuffle volume; spill to disk

Each phase is reported as both _avg and _max across the workers in the stage. That pair is the single most informative thing in the plan.

Rule one: compare avg to max

If compute_ms_max is roughly equal to compute_ms_avg, the stage is balanced — every worker did the same amount of work, and the stage is as fast as it can be for that slot allocation. If compute_ms_max is 10x, 50x, 500x the average, you have data skew: one worker got a disproportionate share of the rows, and the whole query waits on it while the rest of your slots sit idle.

A quick skew score across a job:

SELECT
  stage.id,
  stage.name,
  ROUND(SAFE_DIVIDE(stage.compute_ms_max, NULLIF(stage.compute_ms_avg, 0)), 1) AS compute_skew,
  ROUND(SAFE_DIVIDE(stage.wait_ms_max,    NULLIF(stage.wait_ms_avg, 0)), 1)    AS wait_skew,
  stage.records_read,
  stage.parallel_inputs
FROM `region-us`.INFORMATION_SCHEMA.JOBS, UNNEST(job_stages) AS stage
WHERE job_id = 'bquxjob_xxxxxxxx_xxxxxxxxxxx'
  AND creation_time > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 2 DAY)
HAVING compute_skew > 4
ORDER BY compute_skew DESC;

Anything above about 4 deserves a look; above 20 is almost always a real bug in the data or the join key.

Pattern 1: key skew on a join or GROUP BY

Symptom. One JOIN or AGGREGATE stage with compute_skew in the tens or hundreds, and records_read on that stage far larger than you expected.

Diagnosis. Find the hot key:

SELECT customer_id, COUNT(*) AS n
FROM `proj.ds.events`
GROUP BY customer_id
ORDER BY n DESC
LIMIT 10;

Nine times out of ten the top row is NULL, '', 0, or a sentinel like unknown / -1 that arrived from an upstream system. Every one of those rows hashes to the same worker.

Fixes, in order of preference:

1. Filter the sentinel out before the join. WHERE customer_id IS NOT NULL costs nothing and often ends the investigation.
2. Salt the key when the skew is legitimate (a genuine whale customer). Add a random bucket to the join key, join, then re-aggregate:

WITH salted_events AS (
  SELECT *, MOD(ABS(FARM_FINGERPRINT(CAST(event_id AS STRING))), 8) AS salt
  FROM `proj.ds.events`
),
salted_dim AS (
  SELECT d.*, salt
  FROM `proj.ds.customers` d, UNNEST(GENERATE_ARRAY(0, 7)) AS salt
)
SELECT e.customer_id, ANY_VALUE(d.segment) AS segment, COUNT(*) AS events
FROM salted_events e
JOIN salted_dim d
  ON e.customer_id = d.customer_id AND e.salt = d.salt
GROUP BY e.customer_id;

The dimension is fanned out 8x — fine when it is small, wasteful when it is not. 3. Broadcast instead of shuffle. If one side is small (roughly tens of MB), BigQuery will usually pick a broadcast join by itself; the plan shows it as a JOIN stage with no preceding repartition. Help it by filtering and projecting the small side down to only the columns you need, inside a CTE, before the join. 4. Pre-aggregate the fact table so the join happens on grouped rows rather than raw events. Skew that survives aggregation is much rarer.

Pattern 2: shuffle spill

Symptom. shuffle_output_bytes_spilled is greater than zero, write_ms_max is large, and wall time is far worse than slot time suggests.

BigQuery's shuffle layer keeps intermediate results in memory and spills to disk when a worker exceeds its memory allotment. Spilling is not fatal — but it is typically an order of magnitude slower than in-memory shuffle, and it usually travels with skew (the one fat worker is the one that spills).

Fixes:

• Reduce the row width entering the shuffle. SELECT * across a 300-column table before a join is the classic cause; project only the columns you need, before the join, not after.
• Push WHERE predicates below the join rather than in the outer query.
• Avoid ORDER BY on a full result set. A global sort funnels everything through a single worker. If you need a top-N, use ORDER BY ... LIMIT n, which BigQuery can execute as a partial sort per worker.
• Replace ROW_NUMBER() OVER (PARTITION BY k ORDER BY ts DESC) = 1 deduplication with ARRAY_AGG(t ORDER BY ts DESC LIMIT 1)[OFFSET(0)], which keeps only one row per key in memory instead of the whole window.

-- Memory-friendly "latest row per key"
SELECT
  key,
  ARRAY_AGG(t ORDER BY t.updated_at DESC LIMIT 1)[OFFSET(0)].*
FROM `proj.ds.records` t
GROUP BY key;

• If spill persists on a genuinely large sort or join, give the query more capacity: a reservation with a higher max, or Enterprise/Enterprise Plus edition, raises per-worker resources.

Pattern 3: the exploding join

Symptom. A stage whose records_written is many multiples of its records_read, followed by stages that read absurd row counts.

This is a fan-out: duplicate keys on both sides of a join produce a cartesian product per key. It is easy to miss because the SQL looks innocent and the result is plausible — just with inflated metrics downstream.

Diagnosis. Check uniqueness of the join key on the side you assumed was a dimension:

SELECT COUNT(*) AS rows, COUNT(DISTINCT customer_id) AS keys
FROM `proj.ds.customers`;

If those two numbers differ, your "dimension" is a slowly-changing history table and you need to pick one version — an AS OF predicate on the validity range, or a QUALIFY ROW_NUMBER() ... = 1 inside a CTE:

WITH current_customers AS (
  SELECT *
  FROM `proj.ds.customers`
  QUALIFY ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY valid_from DESC) = 1
)
SELECT ...

Pattern 4: repartition churn

Symptom. Multiple consecutive REPARTITION or COALESCE stages, each with large shuffle bytes and small compute time.

BigQuery inserts repartition stages when consecutive operations need data hashed on different keys. Three joins on three different keys means three shuffles of the whole dataset. You cannot remove them all, but you can often reduce the number:

• Cluster the base tables on the join key. Clustering does not eliminate shuffle, but it makes the read side cheaper and the distribution more even.
• Order the joins so the most selective one runs first, shrinking the data before subsequent shuffles. BigQuery's optimizer reorders joins, but it relies on statistics, and it can be wrong about heavily filtered or freshly written tables.
• Materialize a wide, pre-joined table in Dataform or dbt if the same multi-key join runs dozens of times a day. Shuffle you never perform is the cheapest shuffle.

Reading the graph in BigQuery Studio

The visual Execution Graph is the fastest triage tool once you know what the numbers mean:

• Stages are boxes; the width of the timing bar shows the avg/max split, so skew is visible as a long thin tail.
• Hover any stage for the substeps — READ, COMPUTE, WRITE, JOIN, AGGREGATE, SORT, plus the actual column expressions. This tells you which part of your SQL the stage corresponds to, which is otherwise the hardest thing to work out.
• Look for $1-style internal column names in substeps; they trace back to the CTE that produced them, in order.

Making this a habit, not a fire drill

The plan is most useful as a baseline, not a forensic tool. Two things worth automating:

1. A regression watch. Store a daily snapshot of per-query-hash slot time and flag jumps:

SELECT
  query_info.query_hashes.normalized_literals AS query_hash,
  DATE(creation_time) AS day,
  COUNT(*) AS runs,
  SUM(total_slot_ms) / 1000 AS slot_seconds,
  SUM(total_bytes_processed) / POW(1024, 4) AS tib
FROM `region-us`.INFORMATION_SCHEMA.JOBS
WHERE creation_time > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)
  AND job_type = 'QUERY' AND state = 'DONE' AND error_result IS NULL
GROUP BY query_hash, day
ORDER BY slot_seconds DESC;

The normalized query hash is stable across changing literals, so the same dashboard query is one series even as its date filters move.

2. A skew alert. Run the skew-score query over yesterday's top 50 jobs by slot time and write the results to a small table. When a compute_skew above 20 appears on a stage that was balanced last week, something changed in the data — usually before anyone notices the cost.

A short checklist

When a query is slow, in this order:

1. Is wait_ms the dominant phase? → It is a capacity problem, not a SQL problem. Check reservations and concurrency.
2. Is read_ms dominant with huge total_bytes_processed? → Pruning problem. Check partition filters and SELECT *.
3. Is compute_ms_max >> compute_ms_avg? → Skew. Find the hot key.
4. Is shuffle_output_bytes_spilled > 0? → Memory. Narrow the rows, drop the global sort, rewrite window dedup.
5. Does records_written >> records_read anywhere? → Exploding join. Check key uniqueness.
6. Multiple repartitions in a row? → Rethink join order, clustering, or materialize.

Most "BigQuery is slow" tickets resolve to one of those six lines, and the plan tells you which one within a couple of minutes.

---

Stuck with a query that resists all six? Get in touch — BigQuery performance optimization is one of the things we do most, and a plan review is usually a short, fixed-scope engagement.