Window Functions at Scale

## What a window function actually requires A window function computes a value for every input row using a set of *other* rows — its partition, ordered, optionally narrowed to a frame. The defining property is that the answer for one row depends on rows the engine may not have next to it. In a single-machine engine that is only a sorting problem. In a massively parallel (MPP) engine, where the table is spread across many independent nodes that share nothing but a network, it is first a *placement* problem: no worker can compute a correct value for a row until every peer row of that partition is on the same worker. ## The exchange operator The planner solves placement by inserting a repartitioning **exchange** (also called a shuffle) below the window operator. Each node walks the rows it already holds, applies a hash function to the partition-key columns, maps the hash to one of N buckets, and sends each row to the worker that owns that bucket. Every worker is simultaneously a sender and a receiver. When the exchange finishes, the invariant the window operator needs holds: all rows with a given partition key sit on exactly one worker, and no partition is split. This is the same physical mechanism a hash aggregation or a hash-redistribution join uses. The difference matters for cost: an aggregation can pre-reduce on the sending side (compute local partial sums, ship one row per group per node), while a plain window function generally cannot, because it must emit one output row per input row. A window shuffle therefore moves *all* the rows and *all* the columns the query still needs. ## The sort after the shuffle Once rows land, the worker must put each partition into the order the `ORDER BY` inside `OVER` specifies. Engines normally do one combined sort on (partition key, order key) so that partitions come out contiguous and internally ordered, then a single sequential pass computes the function, resetting state whenever the partition key changes. That sort — not the shuffle — is usually the memory-hungry part, because it materialises rows with their full payload. ## Where the cost lands Three separate costs hide behind one clean-looking clause: - **Network**: bytes moved is roughly the width of the projected row times the row count. Selecting columns you do not need inflates this directly. - **Sort**: CPU and memory per worker, proportional to that worker's share of the rows and to row width. If a partition exceeds the operator's memory budget, it spills to local disk. - **Pipeline breaking**: the window cannot emit its first row until its input for that partition is complete, so downstream operators wait, and one slow worker stalls the whole stage. ## When the exchange can be skipped If the table is already distributed on the same key the window partitions by, a good planner recognises that rows are already co-located and omits the exchange — a local sort is all that remains. The same applies to a chain of operators: a join that already redistributed on `customer_id`, followed by a window partitioned by `customer_id`, can keep one distribution for both. Aligning the join key, the group-by key and the window partition key across a pipeline is one of the highest-leverage rewrites in warehouse SQL, because it collapses several exchanges into one. Similarly, if the physical layout already sorts data within each partition on the order key, some engines can cheapen or skip the sort and stream the window instead. That is a property of the storage layout, not of the SQL. ## Multiple windows in one query A query with `sum(x) OVER (PARTITION BY a ORDER BY t)` and `rank() OVER (PARTITION BY a ORDER BY t)` needs one exchange and one sort; the two functions are evaluated in the same pass. Add `sum(x) OVER (PARTITION BY b)` and the plan grows a second exchange plus a second sort, because the placement invariant for `b` is different from the one for `a`. Reading a plan, the count of exchange operators is therefore roughly the count of *distinct* partition keys in the query, and that count is a good first estimate of the query's shuffle cost. ## Practical consequences Filter before the window, not after — rows removed after the window were still shuffled and sorted. Project only the columns you need, since row width multiplies both network and sort cost. Prefer one partition key across a query where the semantics allow. And treat a window over an unfiltered fact table in a dashboard query as an expensive operation by default, because it moves the whole table across the network on every run.