How a BigQuery query runs

BigQuery turns a query into stages that exchange data through shuffle. What the stages and steps in a query plan mean, how to read the plan and the slot-time of a job, what the plan looked like for real joins, sorts and window functions, and what spills and performance insights tell you.

BigQuery splits a query into stages. A stage is one kind of work, such as reading a table, joining two inputs or sorting, and many workers run its pieces in parallel. Stages pass data to each other through shuffle, a distributed memory layer, not through disk. Each stage is made of steps, the individual operations: READ, FILTER, AGGREGATE, JOIN, SORT, WRITE. The list of stages with their numbers is the query plan, and it is recorded on every job that actually ran. This page uses plans we recorded on real tables of up to 10 million rows.

There is no plan before the query runs. A dry run has none, and neither does a job served from the query cache. We checked a cached job: cacheHit was true, the plan was empty and slot-time was missing.

A plan for a join

We joined two tables of 10 million rows on an INT64 key and aggregated the result. The plan had four stages:

The + in Join+ means the stage does more than the join. Its steps, in order, were:

READ   $1, $2 FROM __stage01_output
READ   $10, $11 FROM __stage00_output
AGGREGATE  $30 := COUNT_STAR(), $31 := SUM($40)
COMPUTE    $40 := multiply($50, $51)
JOIN       INNER HASH JOIN EACH WITH EACH ON $1 = $10
WRITE      $30, $31 TO __stage02_output

The variables ($1, $40) stand for columns and are local to a stage. __stage01_output is the shuffle written by S01. The aggregate runs before the stage writes, so only 100 rows leave S02, and S03 combines them. BigQuery does this partial aggregation without being asked.

The job totaled 16,384 slot-ms and took about 2 seconds, so it held about 8 slots on average. Run again, the same query reported 13.0 to 24.4 seconds of slot-time over eight runs. Use medians when you compare two versions of a query. The slot-time of the job is the number to watch on a reservation. With on-demand pricing you pay for the 320 MB read and slot-time isn’t billed. See how queries are billed.

Small queries have one stage

The same kind of join between a 10-million-row table and a 100,000-row table came out as one stage with no shuffle at all. So did a GROUP BY over 2 million rows. This is Google’s short query optimization, part of what it calls the advanced runtime: BigQuery decides while planning that a query can run as a single stage, based on the expected scan size, the data movement, the filter selectivity and statistics from earlier runs. The job record says so. query_info.optimization_details lists what was applied. Of the jobs we ran for this page, 119 had short_query_optimization in it, and they averaged 1.0 stages. The others, mostly joins, sorts and window functions over 10 million rows, averaged 4.5 to 4.9 stages.

The consequence is that a plan is not a fixed property of a query. The same SQL can have one stage on a small table and four on a big one. A query that filters out most of its input can drop from four stages to one: our 10M by 10M join became a single stage when we added a filter that kept about a tenth of one side.

When the plan grows

Other operations add stages. A window function over 2 million rows had an Input stage, a Sort+ stage for the window, a Coalesce stage and an Output stage. A full ORDER BY over the 10 million rows, with no LIMIT, had six:

The last stage sorts and writes the final order with 3 parallel inputs, and it took 31 of the 68 slot-seconds. It is the stage with the least parallelism, which is why a sort with no LIMIT is expensive, and why Google says to put ORDER BY only in the outermost query. We measured the effect in performance best practices.

Repartition and Coalesce stages are added by BigQuery while the query runs, to even out data between workers or to merge many small pieces. They are not in your SQL, and the console hides them unless you turn on Show shuffle redistribution stages. One of them in a plan is normal. Many of them can mean skewed data.

Shuffle and spills

Every stage reports shuffleOutputBytes and shuffleOutputBytesSpilled. When a query moves more data between stages than its shuffle quota allows, the extra goes to disk and the query slows down sharply. Google ties the shuffle quota to the slots a project has. The INFORMATION_SCHEMA.JOBS_TIMELINE view has a period_shuffle_ram_usage_ratio column that shows which jobs use the most.

We could not make a query spill. The largest we ran moved 1.8 GB through 13 stages, a DENSE_RANK() over all 10 million rows, and reported 0 spilled bytes. Spills show up on much larger inputs than a probe table, so this page has no measured example of one. The mitigation Google lists is the one that works for most plan problems: reduce the data before the stage that shuffles it, with filters, partition pruning and earlier aggregation.

Where to read a plan

• Querylab.io. After the query has run, click the bytes-and-time text in the status bar, or open the Execution Plan view of the Insights panel. Each stage shows its records read and written, shuffle bytes and slot-ms, the slowest stage is tagged as the bottleneck, and parallel branches are drawn as a graph (execution plan).
• Console. The Execution graph tab of a job draws stages as nodes, colored by slot-time, and shuffle as the edges. Clicking a stage shows its steps and timings.
• jobs.get. statistics.query.queryPlan has every field below, plus totalSlotMs for the job. bq show --format=prettyjson -j JOB_ID prints it.
• SQL. INFORMATION_SCHEMA.JOBS has a job_stages column with the same fields in snake_case, and total_slot_ms. It is empty for queries on tables with row-level security.

SELECT
  job_id,
  total_slot_ms,
  s.name,
  s.records_read,
  s.records_written,
  s.shuffle_output_bytes,
  s.shuffle_output_bytes_spilled,
  s.parallel_inputs,
  s.slot_ms,
  s.compute_ms_avg,
  s.compute_ms_max
FROM `region-us`.INFORMATION_SCHEMA.JOBS_BY_USER,
  UNNEST(job_stages) AS s
WHERE job_id = 'your_job_id'
ORDER BY s.name;

Each stage has wait, read, compute and write times, as an average and as a maximum over its workers, in milliseconds and as ratios of the longest. The average tells you where the time goes. The maximum tells you whether one worker held everything up.

What to look for

• Compute max far above compute avg. One worker got far more data than the rest. In one run of a join of 10 million rows with a hot key (70 percent of one side had the same key), the Join+ stage had computeMsMax of 1,019 ms and computeMsAvg of 56. For a join on evenly spread keys it was 144 and 58. See optimizing joins.
• Records written much larger than records read. The stage multiplies rows. This is a join on keys that repeat on both sides.
• A stage with few parallel inputs and a lot of compute. Work that doesn’t spread out, such as the final sort above.
• Huge shuffle where you expected little. An early aggregate or filter would shrink it.
• Wait time much larger than compute time. The stage was waiting to be scheduled. That is slot contention. See slots.

Performance insights

BigQuery also checks completed jobs for known problems and records them in query_info.performance_insights: slot contention, insufficient shuffle quota, high-cardinality joins, partition skew, and an input table that is at least 50 percent bigger than at the previous run. The skewed join above got one. Its insight said the join read 10,000,000 rows on each side and produced 99,999,910, which is the fan-out in numbers. To list every job in the last 30 days that has one, Google gives a query over JOBS_BY_PROJECT that tests those fields.

The insight record also has avg_previous_slot_ms and avg_previous_execution_ms, the averages for the same query text. That is where to look when a query “got slower”. Querylab.io’s execution plan flags the same problems it can see in the plan, such as Join Explosion Detected, Data Skew Detected, Slot Starvation Detected and Query Performance Regression, each with fixes (execution plan).

Repeated runs get optimized

Google applies history-based optimizations to queries it has seen before. It uses what earlier runs learned to change the plan on the next run, and it undoes an optimization that didn’t help. Google’s example is slot-time falling from 60 to 30 to 20 over three runs and then holding. The applied ones are in query_info.optimization_details. Something like it showed up for us: the first run of our 10-million-row join had only enhanced_vectorization there, and the later runs of the same text also listed parallelism_adjustment: applied. Google doesn’t say that this one is history-based, so treat the link as our guess. This matters when you measure: run the query several times and compare the later runs, or the first run will look slower than the system really is. We took the median of 4 to 8 runs for every number we quote.