Where parallelism comes from and what bounds it

Why does adding cores stop helping, and what is the ceiling for data, pipeline and task parallelism?

How is work divided, to whom, and when?

The first three articles asked what a boundary between operators costs. Parallelism asks a different question, with a different currency:

The costs are no longer CPU cycles. They are idle cores and wasted work.

The new constraint is about information, and it is what makes this topic hard: you cannot know how long a piece of work will take before you run it. The only forecast available is the planner's estimate of how many rows each step will produce. That estimate is wrong by some factor, and its error multiplies with every join in the plan (a 2× error at each of five joins can become a 32× error at the top). Every mechanism in this article is one of three things:

• a bet placed on the forecast (static assignment),
• a correction applied after observing reality (dynamic assignment), or
• a hedge that pays for insurance with duplicated work (speculative execution).

There is no fourth option, because there is no fourth source of information.

The constraint that stops mattering is the cost of one boundary. At cluster scale, handing a task to a worker costs about a millisecond. The question is no longer "is the handoff cheap?" but "was anyone waiting?"

---

The one bound, in three costumes

There are three well-known kinds of parallelism: data, pipeline and task parallelism. They are three ways of finding parallelism, and each one runs into the same ceiling with a different denominator. If we state the ceiling once and then apply it three times, three vague warnings become three numbers you can compute.

The work-span bound. Let T₁ be the total work (the time one worker would need to do everything). Let T∞ be the span, also called the critical path: the time needed along the longest chain of steps that depend on each other, even if you had unlimited workers. With p workers, no schedule can beat T₁/p (the work must be shared), and none can beat T∞ (the chain must be run in order). So:

Example. A job has T₁ = 1,000 core-seconds and its longest chain takes T∞ = 50 seconds. With 10 workers the bound is max(100, 50) = 100 s. With 100 workers it is max(10, 50) = 50 s. With 1,000 workers it is still 50 s. Past 20 workers (1,000 ÷ 50) more cores buy nothing.

Amdahl's law (which says that the part of a program that must run serially limits the speedup of the whole) is this bound in a special case: the "serial fraction" is a critical path that happens to be one unsplittable block of code. The span form is more useful, because in a real query the unbreakable chain is usually not a section of serial code. It is a big partition, a slow stage, or a long path through the plan.

---

Data parallelism, and the number that should govern your intuition

What it is. Split the input into partitions, run a copy of the operator on each partition, then combine the results. Example: sum kwh per tariff_zone by giving each of 640 cores a slice of readings. The mechanism is simple. Its ceiling is not.

Elapsed time equals the time of the largest partition. Ideal time is total work divided by p. So speedup is capped at (total work) ÷ (largest partition) = 1/f, no matter how many cores you buy.

Real distributions. Key frequencies in real data are roughly Zipfian: a few keys are very common and most keys are rare. For a Zipf distribution with exponent 1, the top key's share is 1/H_n, where H_n is the n-th harmonic number, approximately ln n + 0.577 for n distinct keys.

• With a million distinct keys: H_n ≈ ln(1,000,000) + 0.577 = 13.8 + 0.577 ≈ 14.4. The top key holds 1/14.4 ≈ 7% of the rows, and the maximum data-parallel speedup is about 14.
• With a billion distinct keys: H_n ≈ 20.7 + 0.577 ≈ 21.3. The ceiling rises to about 21.

The ceiling grows as the logarithm of the number of distinct keys. A thousand times more keys buys only 1.5× more speedup. In other words, it barely grows at all.

This is the sharpest case of the uniformity assumption across the platform. Every partitioner assumes that keys are spread evenly. That assumption is false in a specific, measurable way, and the falsehood turns directly into a closed-form ceiling on the very thing you bought the cluster for.

---

Pipeline parallelism

What it is. Overlap stages. While stage 2 works on item i, stage 1 already produces item i + 1. Like a car wash: while car 1 is being dried, car 2 is being rinsed and car 3 is being soaped. This is the only form of parallelism available when the operators cannot be partitioned. It is also the only one that reduces time to the first result rather than time to the last result.

Two ceilings, both tight

• Depth. With d stages you get at most d× speedup. That is a small number, because real plans have only single-digit stages per pipeline.
• Balance. Steady-state throughput is 1/max(stage time). Example: three stages take 1 ms, 2 ms and 1 ms per item. Throughput is 1 item per 2 ms, not 1 per 1 ms. The pipeline wastes half of the fast stages' time, whatever its depth. Unlike data parallelism, you cannot fix this by adding workers to the slow stage unless that stage is also data-parallel. That is why real engines always combine pipeline parallelism with data parallelism rather than choosing one.

The cost is a buffer between every pair of stages. Its required depth comes from Little's Law: to keep a consumer busy at rate λ across a producer latency W, you need λ × W items in flight. Example: a consumer takes 50,000 batches per second and the producer's latency is 2 ms: 50,000 × 0.002 = 100 batches must be in flight. Memory for buffers is therefore (buffer depth) × (item size) × (number of edges). If the buffer is too small, the pipeline serializes. If it is too large, you have spent memory to hide variance that you could have reduced.

Stages come from pipeline breakers. Pipeline parallelism requires the upstream operators to be non-blocking. A pipeline breaker (a sort, a hash-join build, an aggregation without pre-aggregation) consumes its whole input before it produces anything, so it ends the pipeline by definition. Stages are not a design choice. They are the intervals between pipeline breakers. This is the prerequisite for the whole discussion: you cannot say how many stages overlap until you know where one stage ends.

---

Task parallelism

What it is. Independent subtrees of the plan run at the same time: two scans feeding a join, two branches of a UNION, or unrelated pipelines in one query. The ceiling is T₁/T∞, total work divided by the critical path. So task parallelism is a property of the shape of the plan, not of the data.

• A left-deep join tree (each join takes the previous result plus one new table, so the joins form a chain) has a critical path almost equal to its total work. It has almost no task parallelism.
• A bushy tree (several joins can run independently and meet later) has much more.

This is a real reason to prefer bushy plans. The planner's cost model minimizes total work, so it will not tell you about this. It is the one place where a plan choice is made for a purely runtime reason.

Memory, not cores, is the real constraint. Running k subtrees at once means k operators' worth of state is in memory at the same time. So concurrency is bounded by:

Example: a worker has 48 GB for query state and each join build side needs 6 GB. At most 8 such operators can be active together, even if there are 16 cores. (How memory is granted, reserved and spilled is the job of memory management and is not covered here. We use only the bound.) The bound also explains a difference between engine designs. Pipelined MPP engines (massively parallel processing engines that stream every edge between stages) have a concurrency limit set by the widest point of their plans, because streaming every edge needs every stage resident at once. Materialized-DAG engines (engines that write each stage's output out before the next stage starts) do not have this limit.

---

Intra- versus inter-operator parallelism, and the algebraic gate

The two terms are a re-cut of the three kinds above:

• Intra-operator parallelism is data parallelism inside one operator.
• Inter-operator parallelism is pipeline plus task parallelism across operators.

We keep the distinction because they have different preconditions, and the precondition for intra-operator parallelism is not physical. It is algebraic.

An operator can be parallelized over partitions if and only if it can be written as:

That is a statement about the operator's algebra, so whether an operator can be parallelized is decided by mathematics (the principle algebra gates the topology: only associative reductions can be combined in parts or in trees), not by engineering effort.

The operator's algebra determines which parallelism is legal. The runtime only determines how well it is used. So the question "how do I parallelize this operator?" is answered by the operator's design, before anything here applies.

Parallelism as a plan node. The second insight is structural. Parallelism can be expressed as a plan node. An exchange operator sits between two stages. Every other operator stays unaware of parallelism: it sees a stream of tuples and does not know whether it is one instance of one or one of a thousand. That separation is what makes parallel query execution manageable at all. It is also why the shuffle gets its own article: the exchange is the only place where parallelism lives, so all its cost concentrates there. The shuffle covers the mechanics.