Skip to content

The join that should broadcast

Read a Spark query plan: why a dimension that fits in memory arrives in a shuffle hash join, what a broadcast hint, fresh statistics and a smaller input each change in the plan, and how the same join quietly multiplied the revenue total.

1 solved

What you will learn

  • How to read which join strategy a query got, from explain formatted alone.
  • Why a dimension small enough to fit in memory still arrives in a shuffle hash join.
  • What a BROADCAST hint, ANALYZE TABLE statistics and a smaller input each change, and which of the three is a fix rather than a patch.
  • How to prove a rewrite returned the same answer, because the usual rewrite does not.

The problem

A revenue report joins 7.5 million orders to a customer dimension that keeps history, so every customer is in it eight times and one of those rows is current. The report is slow, which is what got it reported. The total it prints has been wrong the whole time, which is what nobody noticed.

the dimension keeps history orders 7.5M rows customer dim 8 versions each shuffled, re-counted both sides move, and every order pays its price once per version where is_current, then join current rows one per customer orders in memory, one row per order

Set up

The starter reads samples.tpch.orders in place and builds the dimension in a schema of your own: one row per customer per version, eight versions each, is_current true on exactly one of them. Nothing about that dimension is wrong, and nothing in it is skewed.

Paste your salt into the first cell. One of the five answers is yours alone, so a notebook borrowed from somebody else fails that check and passes the rest.

The work

  1. Read the plan. The starter ships the report and a helper that pulls the join operator out of explain formatted. Say which operator the report gets today.
  2. Count the damage. Rows out of the join, over rows in the fact table. A join that keeps every order once comes out at 1.
  3. Change the plan. A hint, statistics, or less data. All three are reachable from here and one of them is also the answer to step 2.
  4. Prove the answer held. The total revenue from your rewrite, and the slice your salt picks. A rewrite that moves either number is not a fix.

Steps 1 to 4 are yours to write.

The questions

key what to print
plan_before the join operator the report gets today
fanout rows the join produces per row of samples.tpch.orders
plan_after the join operator your fixed report gets
total_price total revenue across every segment, from the fixed report
probe revenue for the customer keys your salt picks out

Notes

Serverless runs Photon, so the plan names its operators PhotonShuffledHashJoin and PhotonBroadcastHashJoin where a classic runtime would say ShuffleHashJoin and BroadcastHashJoin. The join_node helper in the starter strips the prefix and normalises the spelling, and the checks accept either, so the operator is what is being asked for and never the prefix. Photon prefers a shuffle hash join to a sort merge join, which is why a plan here says ShuffledHashJoin where most of the writing on this subject says SortMergeJoin.

explain formatted prints the plan the optimiser chose before the query ran. Adaptive query execution can still switch a shuffle to a broadcast once it has seen the real sizes, so the plan you read and the plan that ran are not always the same statement about the same query.

Setting spark.sql.autoBroadcastJoinThreshold is not the route here. Serverless owns its session configuration, and a fix that depends on a conf you cannot set is not a fix you can ship.

samples.tpch is scale factor 5, so orders holds 7,500,000 rows rather than the 1,500,000 the TPC-H spec would suggest.

We measure how the site is used. PostHog and Google Analytics, set to measurement only with advertising features off. We do not sell your data or share it for advertising, and you can turn this off. What this stores.