Why does a step with key-bound state require a redistribution by key upstream, and what does that routing buy?

## The guarantee the redistribution makes A **redistribution by key** is the step that sends every record to the worker that owns its key: the key is reduced to a routing decision, and every record carrying that key follows the same route. Its output guarantee is narrow and very strong — **single ownership**. For any key, one worker sees all of its records, and no other worker sees any of them. Everything else about key-bound state follows from that one guarantee. **Key-bound state** is state a step may read or write only under the grouping key of the record it is currently handling, held by the worker that owns that key, so the read needs no coordination with any other worker. That last clause is a consequence, not an independent design choice. ## Why single ownership removes the lock Concurrency control exists to arbitrate between writers. Single ownership removes the second writer, so there is nothing left to arbitrate: - **No lock.** One worker owns the entry, and within it the step processes that key's records one at a time, so a read-then-modify-then-write sequence cannot interleave with another update to the same entry. - **No remote call.** The entry sits in the same process as the record, whether that process keeps it as live objects or in a worker-local store on the same machine. - **No consistency protocol.** There is no second copy to agree with, so questions of read-your-writes or last-writer-wins never come up for a state access. - **No lost update.** The classic failure — two workers each reading a counter as 41 and each writing 42 — is impossible by construction rather than prevented by a mechanism. Contrast this with a job that keeps its per-key counters in an external database: every access is a network round trip, concurrent updaters must be serialised somehow, and the throughput ceiling of the job becomes the throughput ceiling of that database. Key-bound state is what you get instead when you are willing to pay a redistribution once. ## What the arrangement costs the program's shape | Property | What you get | What it demands | |---|---|---| | State access | Local, lock-free, per-record | All records of a key routed to one worker | | Job structure | A stateful step that scales with width | A boundary in the step graph before it | | Grouping changes | A fresh set of entries per grouping | A new redistribution whenever the key changes | | Reuse | Consecutive steps on the same key add nothing | The arrangement must not have been disturbed | The last row is worth stating out loud in an interview: the redistribution is a property of the data's current arrangement, not a tax charged per stateful step. Two aggregations in a row on the same grouping key can share one arrangement; inserting a step that regroups the records between them destroys it. ## A step that never grouped If a program applies a stateful step without ever grouping, the runtime has no key from which to derive ownership, so it cannot offer key-bound state. What it can offer is **per-worker state** — state attached to one parallel instance of a step rather than to a key, typically a source's read position or a small lookup copy handed to every worker, and redistributed by its own rule when the worker count changes. The two are not interchangeable: a per-entity running total kept as per-worker state would be split across however many instances happened to receive that entity's records, and the parts would never combine. ## Where engines differ 1. **How the exchange is carried.** In the batch lineage the redistribution's output is written to the worker's local disk and fetched afterwards; a **record-at-a-time** runtime pushes records across the network as they are produced, with no materialisation. The guarantee delivered to the stateful step is the same; the failure behaviour and the cost profile are not, and the cost itself belongs to the movement node rather than here. 2. **How visible the step is.** Some surfaces make the grouping an explicit call the author writes before the stateful step; in others it is implied by an aggregation or a join and appears only in the generated plan. The requirement is identical either way. 3. **How often the state is touched.** Under **repeated small finite jobs**, in which the runtime cuts an endless input into short finite chunks and runs a complete job over each one, the entries for a key are loaded and written once per chunk; under record-at-a-time processing the access happens per record. Same locality, very different per-access budget. 4. **Whether the width can change afterwards.** Once the arrangement exists, the number of workers that may own state is constrained by how the key space was bucketed at the job's first start — which is why a candidate who understands this rule usually gets asked about raising parallelism next. ## What an interviewer is listening for The weak answer is "you have to group before you can use state, that is just the rule." The strong answer names the guarantee — one owner per key — and derives the local, lock-free read from it, then notes that the arrangement is reusable across consecutive steps on the same key and destroyed by a regrouping.