Exchange Operators: Partitioning, Broadcasting, and Parallel Pipelines

A serial plan is a tree of operators in which each parent pulls rows from its children. Parallelism requires that some operator break the pull relationship and move rows between processes or threads. That operator is the exchange, and every parallel plan is a serial plan with exchanges inserted at the boundaries between fragments.

The exchange is not a relational operator in the algebraic sense: it does not change the result set. It changes where rows are computed and in what order they arrive. Everything else about parallel query execution follows from the properties the exchange gives up.

Fragments and the shape of a parallel plan

A fragment is a maximal subtree of the plan that runs within one worker. An exchange sits between fragments, consuming rows from its child fragment and producing rows into its parent fragment. A plan with three exchanges has four fragments, and the fragments can be assigned to workers in several patterns: one worker per fragment, many workers per fragment, or many fragments per worker.

The assignment matters because fragments have different cardinalities. A fragment that scans a partitioned table can run on as many workers as there are partitions, while a fragment that performs a final aggregation runs on one worker. The boundary between them is an exchange whose fan-in is many and fan-out is one, and it becomes a serialization point: all workers must finish before the final fragment can produce its result.

The standard parallel plan for SELECT count(*) FROM t has a scan fragment on each worker, a partial aggregate in each, and an exchange that gathers partial counts into a single final aggregate. The scan is embarrassingly parallel; the gather is the bottleneck, but it moves one row per worker, so its cost is negligible.

Repartition, broadcast, and gather

Three exchange flavours cover nearly every case, distinguished by how the routing function maps an input row to an output worker.

A repartition exchange applies a hash function to a set of columns and sends each row to the worker owning that hash value. It preserves the property that rows with equal values on the partitioning columns land on the same worker, which is exactly what a hash join or a hash aggregate needs. Its cost is a network hop for every row, and its failure mode is skew: a dominant key value sends most rows to one worker. The join algorithms on either side of it are described in hash joins and hash aggregation.

A broadcast exchange sends every row to every worker. It is used when one input of a join is small enough to replicate, so that each worker can perform the join locally without a repartition of the other side. The cost is rows × workers in transferred bytes, which is why it is chosen only when the replicated side is genuinely small; a broadcast of a misestimated input is one of the most expensive mistakes a parallel plan can make.

A gather exchange collects rows from many workers into one. It is the terminal exchange of most plans and the one that limits the parallelism of the final result.

A fourth variant, a range repartition, sorts by range boundaries and routes each row to the worker owning its range. It is used when the consumer needs both partitioning and ordering, such as a parallel sort whose merge step can then merge per-worker ranges in order rather than sorting globally.

def repartition(rows, key_of, n_workers):
    out = [[] for _ in range(n_workers)]
    for r in rows:
        out[hash(key_of(r)) % n_workers].append(r)   # stable for equal keys
    return out

The hash % n_workers routing has a property worth stating: the number of workers is fixed for the duration of the query, so the mapping is stable. Resizing workers mid-query would move rows between workers and invalidate any local state built on the assumption.

What the exchange does to order and to blocking

An exchange destroys tuple order unless it is designed not to. A repartition hash exchange scrambles the input order because rows go to different workers by hash; even within one worker, the arrival order interleaves rows from all input workers. A merge exchange that reads all input streams in key order and emits the merged stream preserves order, but requires every input to be sorted on the same key, which in turn requires each worker to sort its share.

This is the reason parallel plans are built around ordering requirements. A parallel sort is usually a per-worker sort followed by an order-preserving merge exchange. A parallel hash join is a repartition of both inputs followed by local joins. A parallel nested loop with an inner index scan is a repartition of the outer input, so that each worker performs lookups against the inner side locally.

The exchange also blocks. A repartition exchange cannot emit a row until it has read enough of its input to know which worker the row belongs to, but it does not need to read everything; it can stream rows through as they arrive. A merge exchange must see the head of every input stream, so it blocks until each input has produced at least one row, and it stalls if one input is slow. A gather exchange with an aggregation above it must consume everything.

Blocking points are where a pipeline breaks. In the Volcano model, a plan is a set of pipelines separated by blocking operators: sorts, hash table builds, and exchanges that must drain an input. The number of concurrently open pipelines is bounded, and each pipeline’s degree of parallelism is set by the exchanges at its boundaries.

Skew and the countermeasures

Repartition by hash concentrates rows with equal keys. If a query joins on a column with one value occurring in half the rows, half the rows go to one worker and the join’s parallel speedup vanishes for that input. The plan is still correct, and the elapsed time is set by the slowest worker.

Countermeasures fall into three groups. The first is to detect skew from statistics at plan time and choose a different distribution, such as broadcasting the skewed side or using a range repartition that splits the hot value across workers. The second is to handle skew at run time: a repartition exchange can observe the distribution as it routes rows and, once a worker’s queue exceeds a threshold, divert further rows to other workers for a local join followed by a union. The third is to avoid the exchange entirely by exploiting an existing partition scheme: if both inputs are already partitioned on the join key in a compatible way, the join is local to each partition pair and needs no exchange at all. That property is what makes partition-wise join a design goal for large tables, and its cost is that the partitioning must be declared and maintained rather than chosen per query.

Skew interacts badly with memory. The worker that receives the hot partition also builds the largest hash table, so it is the one most likely to spill, and its spill makes it slower still. A plan that would fit comfortably with even distribution can fail to fit on one worker.

Degree of parallelism and the cost of starting workers

The degree of parallelism is the number of workers assigned to each fragment. Higher degrees reduce per-worker work but increase coordination: more exchanges, more messages, more contention on shared structures, and more memory granted in total. The total memory for a hash join is the grant per worker times the number of workers, which is why increasing parallelism can turn a fitting plan into a spilling one.

The optimum is usually bounded by the input size. A fragment scanning a table with a thousand pages gains little from thirty-two workers, because each worker’s share is small and the per-worker startup cost dominates. A fragment scanning a table with a million pages saturates the device with far fewer workers than the maximum. Systems therefore choose the degree per fragment, not per query, and cap it by the estimated input size.

Startup cost is real and often underestimated. Each worker needs a process or thread, a memory context, and a snapshot; each exchange needs shared memory queues or network connections. For a query that runs for seconds the startup is negligible, and for one that runs for a few milliseconds it can dominate. This is why short queries often run with parallelism disabled by a cost threshold, and why the threshold’s setting has a visible effect on the latency of small queries.

Reading a parallel plan

A parallel plan’s text usually shows the exchange as an explicit node, sometimes named for its routing (Redistribute, Broadcast, Gather) and sometimes implicit in an operator named for its parallelism. The quickest way to understand a plan is to locate the exchanges and ask, for each, what property the consumer needs: partitioning on which columns, ordering on which columns, or a single stream. A repartition that does not match the join key above it is a wasted hop; a broadcast that should have been a repartition is a bandwidth problem; a gather below a large aggregation is a serialization point.

The other thing to check is whether the exchanges form a chain of blocking points that prevents pipelining. A plan whose every operator sits above a blocking exchange cannot overlap work between fragments, so its elapsed time is the sum of its fragments’ times rather than the maximum. That is a plan that has parallelism in its text and none in its behaviour. Within each fragment, the operators themselves may still be row-at-a-time or batch-at-a-time, which changes the per-worker cost without changing the exchange’s, as described in vectorized execution.