Windows and Partitioning

Concepts covered: Window Partitioning

The engine runs a window function in 3 physical steps: it hashes every row by the PARTITION BY columns so rows sharing a key land on the same worker, sorts each partition by the ORDER BY columns, and streams the sorted rows through the function. The first step is a shuffle, the second a sort, and on a large table both cost more than the function itself, so a fast window query is one that runs those 2 steps as few times as possible over as few rows as possible.

One sort can serve many windows

The planner groups window functions by their specification. Any 2 functions with the same PARTITION BY and ORDER BY share a single shuffle and sort, whatever their frames, because the frame is applied as rows stream past; a function with a different partition key or a different order needs a pass of its own. Count the distinct PARTITION BY and ORDER BY combinations in a query and you have counted its sorts.

Two keys

Partitioning by user_id and by country are different keys, so the table is shuffled and sorted twice.

SELECT
  session_id,
  user_id,
  country,
  duration_s,
  ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY started_at) AS session_seq,
  SUM(duration_s) OVER (PARTITION BY country) AS country_total
FROM sessions
ORDER BY session_id

One key

Both windows share PARTITION BY user_id ORDER BY started_at, so one shuffle and one sort feed both functions.

SELECT
  session_id,
  user_id,
  country,
  duration_s,
  ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY started_at) AS session_seq,
  SUM(duration_s) OVER (PARTITION BY user_id ORDER BY started_at) AS user_running_s
FROM sessions
ORDER BY session_id

Filter first

WHERE runs before the window, so only the 4 US rows are shuffled and sorted, and the sequence counts US sessions only.

SELECT
  session_id,
  user_id,
  country,
  duration_s,
  ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY started_at) AS session_seq
FROM sessions
WHERE country = 'US'
ORDER BY session_id

The first tab pays twice: one hash of the table on user_id, a second on country. If the per-country figure can come from a GROUP BY, its few rows broadcast-join back to the sessions at a fraction of the cost of a second shuffle of the full table, and the window pass on user_id stays the only big one.

Filter the input or filter the output

A WHERE clause before the window shrinks the rows that get shuffled and sorted, which is the cheapest optimisation there is, and it also changes what the window sees. A filter on the window's own output has to wait until the window has run, in an outer query or a following CTE, so the 2 placements cost differently and answer different questions.

Session 4 is u2's first long session on the left and u2's second session overall on the right, because the 45-second session 3 was removed before the numbering on the left and after it on the right. Both are correct answers to different questions, so choose the placement by the question and take the saving when it allows. A filter on a column that is constant within each partition, such as country for a window partitioned by user_id, removes whole partitions and changes no surviving value, so it always belongs before the window.

Skewed and single partitions

Partitions are the unit of parallelism, so their sizes set the wall-clock time. If PARTITION BY country sends 60% of a billion rows to one worker, that worker sorts 600M rows while the others finish and wait. OVER () with no partition and OVER (ORDER BY ts) with a global order put every row in 1 partition on 1 worker, turning a distributed sort into a single-machine one, and a global ROW_NUMBER over a billion rows is the most common way a window query runs out of memory.

How many shuffle-and-sort passes does this query need at minimum?

SELECT
  user_id,
  country,
  ts,
  LAG(ts) OVER (PARTITION BY user_id ORDER BY ts) AS prev_ts,
  ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY ts) AS seq,
  COUNT(*) OVER (PARTITION BY country) AS country_rows
FROM events

• Two: the first 2 windows share one specification, and the third needs its own partition key.
• Three: every window function sorts the table on its own, whatever its OVER clause says.
• One: all 3 windows read the same table, so one sort of the table serves them all.
• Two: LAG and COUNT share a pass because neither ranks rows, and ROW_NUMBER sorts alone.

About this lesson

This section is part of the Window Functions: Advanced lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.