Inside an MPP Warehouse: Nodes, Slices, and the Slowest Slice
A leader node plans, compute nodes execute, and every distributed table is sharded across every slice. The query runs on all of them at once and finishes at the speed of the slowest one, which is why skew in slice-level row counts is a performance bug.
What MPP means
A cloud warehouse is not one big database server. It is a cluster running massively parallel processing (MPP): many machines each holding a slice of the data, each running the same query over its own slice, with the results combined at the end.
The cluster has two kinds of machine, and the split matters:
- The leader node takes your SQL, parses it, builds the plan, and hands compiled steps to the compute nodes. It also collects their partial results and returns the final answer. It stores no user table data.
- Compute nodes hold the data and do the work. Each compute node is divided into slices, and a slice is the real unit of parallelism: it owns a portion of every distributed table and its own share of the node's memory and disk.
So a four-slice cluster running one query is really running four copies of the same scan at once, each over roughly a quarter of the rows.
Stage 1 of 3: The leader parses the SQL and builds one plan for the whole cluster.
Placement is a design decision here, not a runtime one
This is the sharpest difference from the file engine in Level 6. In sql-l6-distributed-execution the engine looked at a folder of Parquet files at read time and handed each task a split. Nothing about the data's arrangement was fixed in advance, so the engine paid a shuffle whenever a query needed rows to co-locate.
A warehouse decides earlier. When you load a table you choose how its rows are spread across the slices, and they stay there. That buys you joins with no data movement at all, and it costs you the chance to be wrong: a bad choice is baked into the table until you reload it. Lesson two is about making that choice well.
Storage inside a warehouse is columnar for the same reason Parquet is: a query that names three columns reads three columns' worth of blocks and skips the rest. Slices and columns are orthogonal. Slices divide the rows, columnar blocks divide the columns.
Why one crowded slice ruins a query
A parallel query is only as fast as its slowest participant. If four slices hold 1,000,000 rows each, all four finish together. If one holds 4,200,000 and the others hold about 1,000,000, three slices sit idle while the fourth grinds through more than four times the work of any neighbor. Spread evenly, those same 7,240,000 rows would be 1,810,000 per slice, so the skew costs you roughly 2.3 times the wall clock you would otherwise have had. This is data skew, and it is the warehouse version of the straggler task from sql-l6-skew-and-joins.
Skew comes from the distribution choice. Spreading fact_trips by driver_id sounds sensible until one fleet account is responsible for a third of all trips: every one of its rows hashes to the same slice.
You detect it by reading slice-level row counts, which is exactly what the exercises do.
Common mistake: reading the cluster's average CPU and concluding it is idle. During a skewed query the average looks fine because most slices genuinely are idle. The number that matters is the spread between the busiest and the quietest slice, not the mean.
Interview nuance: "an MPP query finishes at the slowest slice" is the one-sentence answer that shows systems understanding. Say it, then name the cause (a skewed distribution key) and the check (compare rows per slice). That progression from symptom to cause to diagnostic is what separates a memorized answer from an operational one.
On a real platform this differs. Redshift exposes this through
STV_SLICES(which slice lives on which node) andSVV_TABLE_INFO(per-table size and skew), and you would join them rather than read one flat table. Snowflake hides placement entirely behind automatically managed micro-partitions, so you tune clustering keys instead of distribution keys, but the failure mode is the same: uneven work makes the slowest unit set your runtime.
CREATE TABLE warehouse_slices (
node_id INTEGER,
slice_id INTEGER, -- each compute node owns a fixed set of slices
table_name TEXT,
rows_on_slice INTEGER, -- rows of that table physically stored on this slice
mb_on_slice REAL -- megabytes on this slice; a distributed table's slices sum to its size, an ALL table keeps a full copy per slice
);
INSERT INTO warehouse_slices (node_id, slice_id, table_name, rows_on_slice, mb_on_slice) VALUES
(1, 0, 'fact_trips', 4200000, 22276.0), -- one very busy driver_id hashed to this slice
(1, 1, 'fact_trips', 980000, 5198.0),
(2, 2, 'fact_trips', 1010000, 5357.0),
(2, 3, 'fact_trips', 1050000, 5569.0),
(1, 0, 'fact_payments', 1500000, 3200.0), -- EVEN: round-robin, so the slices match
(1, 1, 'fact_payments', 1498000, 3196.0),
(2, 2, 'fact_payments', 1502000, 3204.0),
(2, 3, 'fact_payments', 1500000, 3200.0),
(1, 0, 'fact_sessions', 2400000, 4266.0),
(1, 1, 'fact_sessions', 900000, 1600.0),
(2, 2, 'fact_sessions', 1100000, 1956.0),
(2, 3, 'fact_sessions', 1000000, 1778.0),
(1, 0, 'fact_promos', 1850000, 1850.0), -- one heavily redeemed promo, skewed but still inside the 2x rule
(1, 1, 'fact_promos', 1000000, 1000.0),
(2, 2, 'fact_promos', 1050000, 1050.0),
(2, 3, 'fact_promos', 1100000, 1100.0),
(1, 0, 'dim_driver', 42000, 120.0), -- ALL: a full copy of the table on every slice
(1, 1, 'dim_driver', 42000, 120.0),
(2, 2, 'dim_driver', 42000, 120.0),
(2, 3, 'dim_driver', 42000, 120.0);
CREATE TABLE warehouse_tables (
table_name TEXT,
diststyle TEXT, -- how the loader placed the rows across the slices
tbl_rows INTEGER -- logical rows in the table, not counting ALL-style copies
);
INSERT INTO warehouse_tables (table_name, diststyle, tbl_rows) VALUES
('fact_trips', 'KEY(driver_id)', 7240000),
('fact_payments', 'EVEN', 6000000),
('fact_sessions', 'KEY(driver_id)', 5400000),
('fact_promos', 'KEY(promo_id)', 5000000),
('dim_driver', 'ALL', 42000);-- Slice-level row counts. fact_trips is spread badly; fact_payments is spread evenly.
SELECT s.node_id, s.slice_id, s.table_name, s.rows_on_slice, t.diststyle
FROM warehouse_slices s
JOIN warehouse_tables t ON t.table_name = s.table_name
ORDER BY s.table_name, s.slice_id;Apply
Your turn
The task this lesson builds to.
Write a query that returns each table's slice spread as (table_name, min_rows, max_rows), widest spread first, over warehouse_slices(node_id, slice_id, table_name, rows_on_slice, mb_on_slice).
Group the slices by table_name, then report the smallest and largest rows_on_slice for that table. Alias the columns exactly min_rows and max_rows, ordered by the spread (max_rows minus min_rows) 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 returns the tables where the busiest slice holds more than twice the rows of the quietest, as (table_name, skew_ratio) rounded to 2 decimals, most skewed first, over the same warehouse_slices table.
skew_ratio is the largest rows_on_slice divided by the smallest. Keep only tables whose ratio is strictly greater than 2, and order by skew_ratio descending.
3 hints and 1 automated check are waiting in the workspace.