After a redistributing step finishes and reports its actual output sizes, which changes can a runtime still make to the unrun grouping or join?

## The evidence, and the window in which it is usable A **redistributing step** is a step that cannot be computed from what one worker already holds, so every record is first sent to the worker that owns its key. When one completes, the runtime knows something it could only guess at before: how many bytes and records each destination actually received. **Runtime replanning** - the engine re-deciding a part of the plan it has not run yet from the sizes a finished step actually produced - spends that evidence on the step about to consume those destinations. The window is narrow and one-directional: the producing side is done, the consuming side has not started, and only the consuming side is open. ## Move one - fewer, larger units for the consumer A step is planned with some number of destinations. If the data turns out much smaller than assumed, most of them hold very little. Each unit costs something regardless of how much data it holds: it is scheduled onto a worker, it opens and closes its inputs and outputs, it reports when it is done. Thousands of units holding a few megabytes each spend more of the run on that overhead than on the work. The correction is to give a group of neighbouring destinations to one worker, so the consumer runs with fewer, larger units. Two things are worth being precise about: - it is the **consumer** that is reshaped. The producing side already ran with the count it ran with, and that is not undone; - it is **not** the same decision as choosing how many pieces to cut the input into before the job starts. That choice is made from guesses at submission time; this one is made from a measurement. Engines that do this generally merge neighbouring destinations rather than arbitrary ones, and aim at a target size rather than a target count. The target exists in order to keep a unit large enough to amortise its own overhead and small enough to fit a worker's memory; its value differs between engines and is not worth memorising. ## Move two - dividing one that came out far too large The mirror case: one destination measures many times its peers, so the consumer would run for as long as that one unit takes while everything else idles. Where the division is legal, the runtime cuts that destination into sub-pieces that run in parallel. For a join, the sub-pieces only produce the right answer if the other side's matching rows are copied to each of them, so the move buys parallelism with extra reads and extra network. The limits on when this is legal are a subject of their own. ## Move three - copying the small side The plan assumed two large inputs and therefore planned to redistribute both, sending records from each across the network so equal keys meet. The finished branch then reports that one side is genuinely small. The unrun join can be switched to **copying the small side**: the whole of the smaller input goes to every worker, and the larger input is neither grouped by key nor moved. This is usually the largest single win available, because it deletes an entire redistribution of the big input rather than reshaping one. It can only be decided after a measurement when the small side is itself computed. A branch that filters, joins and aggregates before reaching this point has no trustworthy declared size; only its finished output does. ## The three moves side by side | move | what triggers it | what changes | what it costs | |---|---|---|---| | merging destinations | most destinations far below a useful unit size | the consumer runs fewer, larger units | less parallelism if overdone | | dividing a destination | one destination far above its peers | that unit becomes several running in parallel | the other side of a join is read and sent more than once | | copying the small side | one unrun join input measures small | the large input is never grouped by key and never moves | the small input is held in every worker's memory | ## What it reaches, and what it does not The corrections land where the next step's shape depends on the sizes just produced, which in practice means the redistribution boundaries of groupings and joins. A chain of per-record steps has nothing to re-decide. Which operators are actually covered differs between engines, and so does whether a given correction is applied without being asked for. And there is a whole class of job that gets almost none of it: a job whose input never ends has no step that finishes, so there is no completed output to measure and nothing on which to base a correction. ## What never changes All three moves are meant to be result-preserving - the rows the job emits are the same rows. What changes is the number of units, the volume crossing the network, and the runtime. If a proposed correction would change the answer, it is not this mechanism.