Skew and elasticity

What do you do when one piece of work is simply too big, and when does adding machines mid-query pay off?

The first step: classify the long tail

All three failure modes (imbalance, straggler, skew) produce the same picture: a long tail at the end of the stage. The fix for one is useless against the others. So diagnose before you fix. Plot per-task duration (y axis) against per-task input bytes (x axis):

Signature	Cause	Fix	Fix that will not work
Bytes similar, durations vary randomly	Imbalance or environment noise	Finer granularity, dynamic assignment, work stealing	Splitting partitions
Bytes similar, one task far slower	Straggler: bad node, contention, cold cache	Speculative duplicate	Splitting partitions
Duration proportional to bytes, bytes very unequal	Skew	Split the unit; replicate the other side	Speculation: the duplicate is equally large

One long tail, three causesschematic

That scatter plot is the entire triage. It is cheap to produce from per-task counters, and it can be shown wrong by data (it is falsifiable): if the points do not line up as predicted, your diagnosis is wrong.

Salting

Salting adds a random suffix to a hot key so that it becomes k sub-keys spread over k tasks. For an aggregation it is legal when the reduction is associative, and it needs a second merge pass. For a join, every sub-key must still meet all the matching rows of the other side, so that side's rows are replicated k times. Either way it requires knowing in advance which keys are hot. How a join survives a hot key, and the Zipf arithmetic behind how big the hottest key gets, are query processing's; routing a key that outweighs a partition is data movement's.

Adaptive partition splitting

This does the same thing without foresight. After the producing stage finishes, the engine reads the actual per-partition output sizes and assigns several consumer tasks to byte ranges of one oversized partition.

It has a hard dependency on the shuffle file format. The partition must be range-readable: a consumer must be able to read "bytes 0 to 100 MB of partition 7". This is true with sorted, consolidated shuffle files (The shuffle) and not true with arbitrary per-pair streams. A runtime capability is determined by a storage-format decision made one article later. This is the tightest coupling here, no mechanism can correct an imbalance smaller than the unit it can address, and here that unit is set by the file format.

Heavy-hitter separation

This splits the plan rather than the data.

Detect the hot keys with a sketch (a small summary structure, here a count-min sketch or heavy-hitter counter, built for this job).
Route the hot keys down a broadcast or replicated path, where no shuffle is needed.
Send the long tail of ordinary keys down the normal hash path.

Two execution strategies for two distributions, in one query. It is the right response to a bimodal input (data that has two very different populations), and structurally the same idea as a hybrid join.

Why join skew is worse than it looks

For a key with L rows on one side and R on the other, the join produces L × R rows, so join skew is the product of the two sides' skew: a key 10× oversized on both sides is 100× the fair share of work (two-sided skew). Input skew measured on either side alone understates it.

Who owns what

Detecting skew from runtime statistics and rewriting the plan in response belongs to the planner's adaptive execution. The runtime machinery belongs to this mechanism: splitting a partition across consumers, replicating the other side, and arbitrating the commit. Choosing a partition key that does not skew is a data-design and planning decision. The bill for a bad key arrives here.

Elastic parallelism

Two facts motivate elasticity: the right degree of parallelism is not known until the data is seen, and capacity has a price that changes with time. Two requirements make it possible, and one piece of arithmetic makes it mostly useless inside a single query.

Requirement 1: work must be reassignable. This is trivial with a morsel queue, easy with partition-level tasks over materialized edges, and hard with stateful operators whose state must be moved to new workers. (Rescaling of stateful operators in a continuous stream is another topic; here we mean only queries with a beginning and an end.)

Requirement 2: partition count must be independent of worker count. This forces the design that everyone converged on: over-partition. Choose p partitions well above the number of workers, so adding a worker means handing out more partitions, not re-dividing anything.

The price of over-partitioning is set by the fixed cost per operation. If dispatching a task costs F ≈ 10 ms and you want at most 1% overhead, the minimum task duration is F ÷ 0.01 = 1 second. So the number of partitions cannot exceed (total work) ÷ (1 s). That gives a hard ceiling on elasticity:

The arithmetic that should temper expectations. Scaling out during a query pays only if the remaining work saved is more than the cost of getting new workers. Going from p to p′ workers saves remaining/p − remaining/p′, and costs a startup latency S (provisioning, cold caches, no data locality).

RelationshipWhen scaling out mid-query pays →

remaining / p - remaining / p^{'} vs S

$W$: core-seconds still to do
$p, p^{'}$: workers before and after
$S_{startup}$: provisioning, cold caches, no data locality

Trade-off: Isolation vs. utilization →

That makes it useless for the queries people most want to rescue (the ones already three-quarters done and running late). It is useful only for long queries where the need is detected early.

So in practice elasticity lives between queries, not inside one. A warehouse that scales up because a queue of queries formed applies the same mechanism where the remaining work is large and the acquisition latency is spread over many queries. That is autoscaling at the level of a whole workload. The honest conclusion for query runtimes is that intra-query elasticity is a narrow win, not a general one.

Case, resolvedConstructed case · Hana at Gridlark

One meter id with 7% of the rows

Mateo shows her the scatter plot. Duration is proportional to bytes and meter_id = 0 has 7% of the rows. It is skew, not a straggler, which is why speculation made things worse: the duplicate of the giant task was equally giant, and it consumed capacity that other queries needed (at high utilization the queue formula 1/(1 − ρ) turned that into a slowdown). He fixes it in two ways. He separates the placeholder id: meter_id = 0 rows are detected with a sketch and aggregated on a separate path (they are mostly junk readings and are summarized with a broadcast-style partial aggregate), while the remaining keys go down the normal hash path. For the remaining unknown hot keys, adaptive splitting divides any oversized partition into byte ranges, which works because the shuffle output is range-readable (The shuffle). Over-partitioning to 10 tasks per worker with 1-second minimum tasks gives balance without paying too much dispatch cost. The autoscaler is turned off inside queries (a 30-second machine start-up cannot help a stage that has 10 seconds left) and kept for queue build-up between queries. The 40× tail disappears.