Optimizing joins in BigQuery

How BigQuery runs a join, measured on tables of up to 10 million rows. Broadcast and hash joins, why table order and filter placement made no difference, join fan-out, skew, INT64 vs STRING keys, range joins that become cross joins, and self-joins vs window functions.

BigQuery runs a join in one of two ways. If one side is small, it sends that side to every worker and joins locally, which is a broadcast join. If both sides are large, it splits both by the join key so that matching keys land on the same worker, which is a hash join, and that moves both tables through shuffle. You don’t pick one. The optimizer does, from the sizes it expects. What you can change is how much data reaches the join and whether the join multiplies rows. Most of the advice on this page is about those two things. Some of Google’s advice, like putting the largest table first, did not change anything in our tests.

All timings below are from jobs on tables we generated, repeated 3 to 8 times, with medians reported. Slot-time is the work the job did, and it varies by 30 percent or more between runs of the same query. See how a query runs for how to read the numbers.

What the plan says about a join

The JOIN step in a plan names the type, and the label shows how the inputs were distributed. We saw three:

Google says EACH WITH ALL is a broadcast join and EACH WITH EACH is a hash join. ALL WITH ALL appeared only in plans with a single stage and no shuffle, which BigQuery picks for queries it expects to finish quickly on a small amount of data. The join of 10 million with 10 million rows moved 360 MB through shuffle and used about 18 slot-seconds. The same 10-million-row table joined to a 100,000-row table used 1.2 slot-seconds. A big-to-small join is cheap, and the sensible design is to keep one side of most joins small.

Table order did not matter

Google’s guide says to put the largest table first, then the smallest, and that this makes a broadcast join. We joined a 10-million-row table with a 100,000-row one, each way round:

The optimizer chose the sides itself every time. Writing the big table first is harmless, and we found no case in which it helped. We didn’t test inputs whose sizes are hard to predict.

Filters go where they help, not where you write them

Advice says to filter each table before the join. We wrote the same filter three ways: in WHERE, in the ON clause, and in a subquery in FROM. All three produced the same plan and read the same 320 MB. BigQuery moves the filter below the join itself.

What matters is whether the filter shrinks a side. With no filter our 10-million-row join was a hash join in four stages. A filter that kept about a tenth of one side turned it into the single-stage broadcast plan, and filters on both sides cut slot-time from about 17 seconds to 4. Bytes read didn’t change in any of them, because there was no partition or cluster to prune, so on-demand cost stayed at 320 MB. Filtering before a join reduces compute. It reduces bytes only when the filter prunes partitions or clusters.

Watch for row multiplication

A join on a key that repeats on both sides returns every pairing. Three shipments for one order make three joined rows for that order, and any SUM over the order’s columns counts it three times:

WITH orders AS (
  SELECT 1 AS order_id, 100 AS amount UNION ALL
  SELECT 2, 50 UNION ALL
  SELECT 3, 25
),
shipments AS (
  SELECT 1 AS order_id, 'box-a' AS parcel UNION ALL
  SELECT 1, 'box-b' UNION ALL
  SELECT 1, 'box-c' UNION ALL
  SELECT 2, 'box-d'
)
SELECT
  COUNT(*) AS joined_rows,
  SUM(o.amount) AS summed_amount
FROM orders AS o
JOIN shipments AS s USING (order_id)

The query says 350 where the shipped orders total 150. BigQuery reports this kind of join as a high-cardinality join in the job’s performance insights, when the output has far more rows than the two inputs. In our skewed test the insight recorded 10,000,000 rows on each side and 99,999,910 out. Querylab.io’s execution plan flags a join stage that wrote many times more rows than it read as Join Explosion Detected (execution plan).

The fix is to join at the grain you need. Aggregate the many-side first, so each key appears once:

WITH orders AS (
  SELECT 1 AS order_id, 100 AS amount UNION ALL
  SELECT 2, 50 UNION ALL
  SELECT 3, 25
),
shipments AS (
  SELECT 1 AS order_id, 'box-a' AS parcel UNION ALL
  SELECT 1, 'box-b' UNION ALL
  SELECT 1, 'box-c' UNION ALL
  SELECT 2, 'box-d'
)
SELECT
  o.order_id,
  o.amount,
  s.parcels
FROM orders AS o
LEFT JOIN (
  SELECT order_id, COUNT(*) AS parcels
  FROM shipments
  GROUP BY order_id
) AS s USING (order_id)
ORDER BY o.order_id

Google’s guide gives the same advice for queries that have both a join and a GROUP BY: aggregate before the join, so less data is joined. The numbers are in the next section.

Skew

If one key holds most of the rows, the worker that gets that key does most of the work while the others finish and wait. The plan shows it as a stage whose computeMsMax is much larger than computeMsAvg. We built a table where 70 percent of the rows had the same key, and joined it to a table in which that key appeared 10 times. In one run the Join+ stage had a maximum compute time of 1,019 ms and an average of 56, a ratio of 18. For a join on evenly spread keys the same stage had 144 and 58, a ratio of 2.5.

Google’s performance insights include a partition skew insight for the case where BigQuery can’t rebalance the partitions. Google’s advice for skewed joins is to pre-filter as early as possible or to split the query into two or more queries. The execution plan reports skew as Data Skew Detected, with the stages involved.

Aggregating the other side also worked for us, because it collapses the hot key to one row. We summed price per key first, then joined, instead of joining and then summing amount * price. The two give the same answer: for each row of the big side, the sum of amount * price over its matches equals amount times the sum of their prices. The plan went from four stages to one, and the median slot-time from 14.8 seconds to 5.4, with the elapsed time falling from 3.1 seconds to 0.8.

INT64 keys over STRING keys

Google says to join on INT64, not STRING, because comparing strings costs more. We joined the same two 10-million-row tables on an integer and on its string form, alternating the two for 8 runs each:

The string key cost about 19 percent more slot-time and 10 percent more shuffle, and the ranges overlap. It is a real but small effect on short strings. Longer keys would cost more. Google also advises against calling functions on join columns. We tried ON CAST(a.k AS STRING) = b.ks: it shuffled 398 MB, the string-sized amount, but its median slot-time over 4 runs was 18.6 seconds, the same as the integer join at 18.1. We could not measure a penalty for the cast.

Range joins become cross joins

A join with no equality in its condition has nothing to hash. We joined 20,000 timestamps to 2,000 intervals with ON ts BETWEEN start_ts AND end_ts. The plan was CROSS ALL WITH ALL followed by a FILTER, in a single worker, and it took 1.6 slot-seconds. The same join with an extra equality on a bucket, ON r.bucket = w.bucket AND r.ts BETWEEN ..., took 0.07, 23 times less, and ran as a hash join. A smaller version of both, 2,000 readings and 200 windows, returns the same count either way:

WITH readings AS (
  SELECT x AS ts FROM UNNEST(GENERATE_ARRAY(1, 2000)) AS x
),
windows AS (
  SELECT y AS start_ts, y + 9 AS end_ts FROM UNNEST(GENERATE_ARRAY(1, 2000, 10)) AS y
)
SELECT COUNT(*) AS matched_pairs
FROM readings AS r
JOIN windows AS w ON r.ts BETWEEN w.start_ts AND w.end_ts

WITH readings AS (
  SELECT x AS ts, DIV(x - 1, 10) AS bucket FROM UNNEST(GENERATE_ARRAY(1, 2000)) AS x
),
windows AS (
  SELECT y AS start_ts, y + 9 AS end_ts, DIV(y - 1, 10) AS bucket FROM UNNEST(GENERATE_ARRAY(1, 2000, 10)) AS y
)
SELECT COUNT(*) AS matched_pairs
FROM readings AS r
JOIN windows AS w ON r.bucket = w.bucket AND r.ts BETWEEN w.start_ts AND w.end_ts

Compute a coarse bucket on both sides (a day, a 10-unit range) so the equality cuts the candidate pairs, and keep the range test as the second condition. This works when each interval falls inside one bucket. An interval that spans buckets needs one row per bucket it touches, or it will miss matches.

Self-joins and window functions

Google says to avoid self-joins and use window functions. The common case is “the latest row per key”. A self-join finds the maximum and joins back. A window function ranks the rows:

WITH readings AS (
  SELECT 'a' AS sensor, 1 AS seq, 20.5 AS temp UNION ALL
  SELECT 'a', 2, 21.0 UNION ALL
  SELECT 'a', 3, 21.4 UNION ALL
  SELECT 'b', 1, 18.0 UNION ALL
  SELECT 'b', 2, 17.2
)
SELECT r.sensor, r.seq, r.temp
FROM readings AS r
JOIN (
  SELECT sensor, MAX(seq) AS last_seq
  FROM readings
  GROUP BY sensor
) AS latest
  ON r.sensor = latest.sensor AND r.seq = latest.last_seq
ORDER BY r.sensor

WITH readings AS (
  SELECT 'a' AS sensor, 1 AS seq, 20.5 AS temp UNION ALL
  SELECT 'a', 2, 21.0 UNION ALL
  SELECT 'a', 3, 21.4 UNION ALL
  SELECT 'b', 1, 18.0 UNION ALL
  SELECT 'b', 2, 17.2
)
SELECT sensor, seq, temp
FROM readings
WHERE TRUE
QUALIFY ROW_NUMBER() OVER (PARTITION BY sensor ORDER BY seq DESC) = 1
ORDER BY sensor

On our 10-million-row table, both read 240 MB. The self-join plan had 6 stages and the window plan 3, and the median slot-time was the same, 14.9 seconds. In our dry runs a self-join was billed for each column of the table once, so a self-join of the public Shakespeare table costs exactly what a plain read of the same columns does:

SELECT a.word, b.corpus
FROM `bigquery-public-data.samples.shakespeare` AS a
JOIN `bigquery-public-data.samples.shakespeare` AS b USING (word)

SELECT word, corpus
FROM `bigquery-public-data.samples.shakespeare`

Both are 3,797,568 bytes. So we can’t call self-joins slower in our test. The window version is shorter and has fewer ways to go wrong. Where self-joins do hurt is when the key repeats, which is the fan-out above. See QUALIFY and window functions.

Constraints let BigQuery drop joins

PRIMARY KEY and FOREIGN KEY constraints can be declared as NOT ENFORCED, and the optimizer uses them. A LEFT JOIN from a 10-million-row table to a 100,000-row one, selecting only columns of the big table, was dry-run at 160.8 MB with no constraints and 80 MB once the small table had a primary key and the big one a matching foreign key. The join was removed. See primary and foreign keys. The cheapest join is one you don’t need, and nested and repeated columns avoid joins in the same way. See nested and repeated data.