Distribution Styles, Sort Keys, and Reading the Query Plan

Level 7: Warehouses, Lakehouse & Dimensional Modelinghard30 minDISTSTYLE KEY, ALL, and EVENsort keyszone mapsskew detectionquery-plan readingjoinsCASE classification

KEY co-locates joins, ALL copies small dimensions everywhere, EVEN round-robins facts nobody joins on. The plan tells you when you chose wrong, and DS_DIST_BOTH is the step interviewers ask you to spot.

Distribution decides where rows live before any query runs

When you create a warehouse table you pick a distribution style, and the loader honors it forever after. There are three you need cold:

KEY: hash one column and send each row to the slice that column maps to. Two tables distributed by the same key have their matching rows on the same slice, so joining them moves nothing.
ALL: put a complete copy of the table on every slice. Every join against it is local by construction. You pay for it in storage, once per slice, so it is only sane for small dimensions.
EVEN: round-robin the rows across slices with no regard for content. Perfectly balanced, and useless for co-location, which is right for a staging table nobody joins on.

The tradeoffs line up cleanly:

Table

Three distribution styles. KEY buys co-location and risks skew; ALL buys co-location and costs storage; EVEN buys balance and costs network.
diststyle	where rows go	join against it	the cost
KEY	hash of the chosen column	free when both sides share the key	skew if the key has hot values
ALL	a full copy on every slice	always free	storage multiplied by the slice count
EVEN	round-robin, ignores content	always redistributed	network traffic on every join

Three distribution styles. KEY buys co-location and risks skew; ALL buys co-location and costs storage; EVEN buys balance and costs network.

The plan tells you when you chose wrong

Run EXPLAIN on a join and the planner labels each join step with what it has to move. Four labels carry the whole story:

DS_DIST_NONE: nothing moves, because both sides are already distributed on the join column. This is the outcome you designed for.
DS_DIST_ALL_NONE: nothing moves either, but for the other reason. The inner table is DISTSTYLE ALL, so every slice already holds a full copy of it and there is nothing to redistribute. Redshift gives this case its own label, so do not call it DS_DIST_NONE in an interview.
DS_BCAST_INNER: the inner (smaller) table is broadcast, meaning a copy is sent to every node. Fine when that table is genuinely small. It is the same broadcast-join decision you met in sql-l6-skew-and-joins, made by the warehouse planner instead of Spark.
DS_DIST_BOTH: both sides are redistributed across the cluster before the join. This is the expensive one, and it is the red flag interviewers name by name. It happens when neither side is distributed on the join column, so the planner has no co-located option at all.

Order of evaluation

Scan both tableseach slice reads its own rows
DS_DIST_BOTHboth sides reshuffled across the network
Hash joinonly now can matching rows meet
Aggregatepartial results per slice
Leader combinesfinal answer

A join between two facts distributed on different keys. Everything before the join is cheap; the redistribution is the bill.

Compare that with the co-located version: same scan, same hash join, same aggregate, and the redistribution step simply is not there.

Sort keys are the warehouse's zone maps

Distribution decides which slice a row lives on. The sort key decides what order the rows sit in inside that slice, and the warehouse records the minimum and maximum value of every column per block. A query filtering on completed_at reads the block headers, sees that a block's range cannot contain the requested dates, and skips it without reading it.

That is the same min/max skipping you learned as Parquet row-group statistics in sql-l6-row-groups-pushdown, now built into the warehouse's own storage. The sort key is what makes those ranges narrow enough to be useful: on an unsorted column the values are scattered, so nearly every block's range overlaps your filter and nothing gets skipped.

The practical rule: sort on the column you filter by, which for a fact table is almost always the date column, and distribute on the column you join by.

Common mistake: choosing the primary key as the distribution key because it is unique and therefore perfectly balanced. It is balanced, and it is also useless, because nobody joins on it. Distribute on the column that appears in your join predicates, then check the skew.

Interview nuance: "join on the dist key, or broadcast the small table" is the expected sentence. If you can add "and if I see DS_DIST_BOTH in the plan, neither side is distributed on the join column," you have answered the Redshift question the way a practitioner answers it.

On a real platform this differs. Redshift's SVV_TABLE_INFO carries these columns for real (diststyle, sortkey1, skew_rows, unsorted, size) and the plan comes from EXPLAIN as indented text rather than a table you can join to. Redshift also offers DISTSTYLE AUTO, which starts a small table as ALL and converts it to EVEN or KEY as it grows. Snowflake and BigQuery hide distribution altogether, so the same reasoning shows up as clustering keys and partition pruning instead.

Sample data for this example

CREATE TABLE warehouse_table_stats (
  table_name TEXT,
  diststyle  TEXT,      -- KEY | ALL | EVEN
  dist_key   TEXT,      -- the distribution column, '' when the style is not KEY
  sortkey1   TEXT,      -- the first sort-key column, '' when the table has no sort key
  tbl_rows   INTEGER,
  skew_rows  REAL,      -- rows on the fullest slice divided by rows on the emptiest slice
  size_mb    INTEGER    -- logical size of the table, not multiplied by the copies an ALL table keeps
);
INSERT INTO warehouse_table_stats (table_name, diststyle, dist_key, sortkey1, tbl_rows, skew_rows, size_mb) VALUES
  ('fact_trips',        'KEY',  'driver_id',  'completed_at', 7240000, 4.29, 38400),
  ('fact_payments',     'EVEN', '',           'paid_at',      6000000, 1.00, 12800),
  ('fact_sessions',     'KEY',  'driver_id',  '',             5400000, 2.67,  9600),
  ('fact_promos',       'KEY',  'promo_id',   'issued_at',    5000000, 1.85,  5000),
  ('fact_ratings',      'KEY',  'trip_id',    'rated_at',     4800000, 1.04, 15200),
  ('dim_driver',        'ALL',  '',           'driver_id',      42000, 1.00,   120),
  ('dim_vehicle',       'ALL',  '',           '',              180000, 1.01,    60),
  ('dim_city',          'EVEN', '',           'city_id',          320, 1.00,     4),
  ('stg_trip_events',   'EVEN', '',           '',             9100000, 1.03, 44000),
  ('stg_payment_events','EVEN', '',           '',              700000, 1.02,  3100);
CREATE TABLE query_plan_steps (
  query_id    INTEGER,
  step_no     INTEGER,
  operation   TEXT,
  outer_table TEXT,   -- the streamed side of a join, '' on non-join steps
  inner_table TEXT,   -- the built side of a join, '' on non-join steps
  dist_type   TEXT,   -- DS_DIST_NONE | DS_DIST_ALL_NONE | DS_BCAST_INNER | DS_DIST_BOTH, '' on non-join steps
  rows_out    INTEGER,
  mb_moved    INTEGER -- megabytes the step pushed across the network
);
INSERT INTO query_plan_steps (query_id, step_no, operation, outer_table, inner_table, dist_type, rows_out, mb_moved) VALUES
  (101, 1, 'seq scan',       'fact_trips',    '',              '',                   5200000,     0),
  (101, 2, 'hash join',      'fact_trips',    'dim_driver',    'DS_DIST_ALL_NONE',   5200000,     0),
  (101, 3, 'hash aggregate', '',              '',              '',                       120,     0),
  (102, 1, 'seq scan',       'fact_payments', '',              '',                   3100000,     0),
  (102, 2, 'hash join',      'fact_payments', 'dim_city',      'DS_BCAST_INNER',     3100000,     4),
  (102, 3, 'hash aggregate', '',              '',              '',                        40,     0),
  (103, 1, 'seq scan',       'fact_trips',    '',              '',                   5200000,     0),
  (103, 2, 'hash join',      'fact_trips',    'fact_payments', 'DS_DIST_BOTH',       5200000, 41200),
  (103, 3, 'hash aggregate', '',              '',              '',                       900,     0),
  (104, 1, 'seq scan',       'fact_sessions', '',              '',                   2600000,     0),
  (104, 2, 'hash join',      'fact_sessions', 'fact_ratings',  'DS_DIST_BOTH',       2600000,  9800),
  (104, 3, 'sort',           '',              '',              '',                   2600000,     0),
  (105, 1, 'seq scan',       'fact_trips',    '',              '',                   4800000,     0),
  (105, 2, 'hash join',      'fact_trips',    'fact_sessions', 'DS_DIST_NONE',       3900000,     0),   -- both sides KEY on driver_id, so nothing moves
  (105, 3, 'hash aggregate', '',              '',              '',                      1400,     0);

Worked example (SQL)

-- Three joins, three verdicts. 101 is local because dim_driver is ALL, 105 because both sides
-- share a dist key, and 103 pays for a full redistribution.
SELECT query_id, step_no, operation, outer_table, inner_table, dist_type, mb_moved
FROM query_plan_steps
WHERE query_id IN (101, 103, 105)
ORDER BY query_id, step_no;

Apply

Your turn

The task this lesson builds to.

Write a query that returns every query whose plan contains a DS_DIST_BOTH step, as (query_id, mb_moved), most data moved first, over query_plan_steps(query_id, step_no, operation, outer_table, inner_table, dist_type, rows_out, mb_moved).

Keep only the steps whose dist_type is exactly 'DS_DIST_BOTH' and report that step's mb_moved, ordered descending.

3 hints and 1 automated check are waiting in the workspace.

Practice

Make it stick

A second problem on the same idea, plus 3 bonus drills.

Write a query that recommends a distribution style for each table as (table_name, recommended_diststyle), in table-name order, over warehouse_table_stats(table_name, diststyle, dist_key, sortkey1, tbl_rows, skew_rows, size_mb).

Apply these rules in order, first match wins: 'ALL' when tbl_rows is under 500000, otherwise 'KEY' when dist_key is not the empty string, otherwise 'EVEN'.

3 hints and 1 automated check are waiting in the workspace.